#!/usr/bin/env python3
# Finds credential-shaped values in the host files dejima copies into a
# container, and prints what it found WITHOUT printing any value.
#
# This exists because the alternative does not scale. Two leaks were found by
# hand in one week - a console API key in ~/.claude.json, credential helpers
# in ~/.gitconfig - and both fixes were denylists: they encode exactly the
# knowledge dejima did not have the week before. A harness's authentication is
# not a stable interface, so the next field added leaks silently unless
# something is looking (#291).
#
# NOTHING HERE DECIDES ANYTHING. It classifies and reports; bin/dejima asks
# the operator and applies the answer. Keeping the judgement out of here is
# what lets the same classifier run in a test without a terminal.
#
# THE ONE INVARIANT: a credential value never leaves this program. Not in a
# finding, not in an error, not in a traceback. What a caller gets is a
# redacted path and a shape name - enough to report a bug to whoever ships the
# harness, and useless to anyone who intercepts it.
#
# Which is why the caller never sees a real path either. In a gitconfig the
# userinfo sits in the SECTION HEADER, so `url.https://me:PAT@github.com/` is
# a path that IS a credential. Findings carry a redacted path for a human and
# an opaque id for matching, and --scrub takes ids back - so real paths stay
# inside this program, which is the only place that has to hold them.

import argparse
import hashlib
import json
import math
import re
import sys

# Anything that looks like credentials in a URL, wherever it appears.
USERINFO = re.compile(r"(?P<scheme>[a-z][a-z0-9+.-]*://)(?P<user>[^/@\s:]*):"
                      r"[^/@\s]+@")


def redact(text):
    """A path an operator can recognise, with any embedded secret removed."""
    return USERINFO.sub(lambda m: "%s%s:***@" % (m.group("scheme"),
                                                 m.group("user")), text)


def finding_id(label, path):
    """Stable across runs, and reveals nothing about what it identifies.

    Keyed on the real path, so a decision survives the credential itself
    rotating - the operator answered about a field, not about a value."""
    h = hashlib.sha256(("%s\t%s" % (label, path)).encode("utf-8"))
    return h.hexdigest()[:16]

# Vendor prefixes, which are worth naming separately from the heuristics: a
# match here is not a guess, and the shape name tells an operator exactly what
# they are looking at without showing it to them.
KNOWN_SHAPES = (
    ("anthropic-api-key", re.compile(r"sk-ant-")),
    ("openrouter-key", re.compile(r"sk-or-")),
    ("openai-key", re.compile(r"^sk-[A-Za-z0-9]{20,}$")),
    ("github-token", re.compile(r"(ghp|gho|ghu|ghs|ghr)_[A-Za-z0-9]{8,}|github_pat_")),
    ("slack-token", re.compile(r"xox[baprs]-")),
    ("aws-access-key-id", re.compile(r"^AKIA[0-9A-Z]{16}$")),
    ("google-api-key", re.compile(r"^AIza[0-9A-Za-z_-]{35}$")),
    ("jwt", re.compile(r"^eyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.")),
    ("private-key-pem", re.compile(r"^-----BEGIN [A-Z ]*PRIVATE KEY-----")),
    ("auth-header-value", re.compile(r"(?i)(^|:\s*)(basic|bearer)\s+\S{8,}")),
    ("url-userinfo", re.compile(r"^[a-z][a-z0-9+.-]*://[^/@\s]+:[^/@\s]+@")),
)

# Names that say "this is a secret", matched as WHOLE WORDS rather than as
# substrings. Substring matching put `auth` inside `oauthAccount` and flagged
# every field of the signed-in account's description, and `tengu_mcp_local_
# oauth_blocked_hosts` along with them.
SUSPECT_WORDS = frozenset((
    "key", "apikey", "token", "secret", "secrets", "credential", "credentials",
    "password", "passwd", "auth", "authorization", "bearer", "cookie",
    "signature", "privatekey",
))

# Secrets are drawn from a restricted alphabet and contain no whitespace:
# base64, hex, base58, or a vendor's own. Prose is not, which is the whole
# discriminator - English text scores 4.0-4.5 bits per character by this
# measure, so entropy ALONE flags every sentence in a config file.
SECRETISH = re.compile(r"^[A-Za-z0-9+/=_.~-]+$")


def words(name):
    """Splits an identifier into lowercase words: camelCase, snake_case and
    kebab-case all reduce to the same thing."""
    name = re.sub(r"\[\d+\]$", "", name)
    parts = re.split(r"[^A-Za-z0-9]+", name)
    out = []
    for part in parts:
        out.extend(w.lower() for w in
                   re.findall(r"[A-Z]+(?![a-z])|[A-Z][a-z0-9]*|[a-z0-9]+", part))
    return out


def suspect_name(path):
    """The leaf, plus its immediate parent.

    The parent matters for a gitconfig, where `credential.helper` is a
    credential because of the section it sits in. Anything further up is too
    broad: a whole subtree named after a feature flag about oauth is not a
    subtree of credentials."""
    segments = path.split(".")
    for seg in segments[-2:]:
        if SUSPECT_WORDS & set(words(seg)):
            return True
    return False

# Shapes that look like secrets to an entropy test and are not. Checked before
# entropy, so a UUID machine id does not cost an operator a prompt.
NOT_SECRET = (
    re.compile(r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-"
               r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$"),          # uuid
    re.compile(r"^[a-z]+://"),                                # url
    re.compile(r"^/"),                                        # absolute path
    re.compile(r"^~?/"),                                      # home path
    re.compile(r"^\d{4}-\d{2}-\d{2}"),                        # timestamp
    re.compile(r"^[0-9.]+$"),                                 # version/number
)

MIN_ENTROPY_LEN = 32
MIN_ENTROPY_BITS = 4.0


def shannon(value):
    if not value:
        return 0.0
    counts = {}
    for ch in value:
        counts[ch] = counts.get(ch, 0) + 1
    n = float(len(value))
    return -sum((c / n) * math.log(c / n, 2) for c in counts.values())


def classify(path, value):
    """Returns (verdict, shape) with verdict in credential|suspect|None.

    Never returns any part of `value`.
    """
    if not isinstance(value, str) or not value.strip():
        return (None, None)
    v = value.strip()
    for shape, pattern in KNOWN_SHAPES:
        if pattern.search(v):
            return ("credential", shape)
    if any(p.match(v) for p in NOT_SECRET):
        return (None, None)
    # Both remaining rules require a value that could BE a secret: one token,
    # restricted alphabet, no whitespace. Without that, a field called
    # "authDescription" holding a sentence is a finding.
    if not SECRETISH.match(v):
        return (None, None)
    if suspect_name(path) and len(v) >= 8:
        return ("suspect", "named-field")
    if len(v) >= MIN_ENTROPY_LEN and shannon(v) >= MIN_ENTROPY_BITS:
        return ("suspect", "high-entropy")
    return (None, None)


def walk_json(doc, prefix="", parts=()):
    """Yields (display path, value, components).

    The display path is what an operator reads and what finding_id hashes, so
    its spelling is fixed - it is recorded in the decisions file and a change
    would orphan every standing answer. It is also ambiguous: a dict key of
    `x[0]` and element 0 of a list `x` produce the same string, and so do a key
    containing a dot and the traversal a dot denotes.

    The components are the same walk as a tuple, which is not ambiguous. Anything
    that has to FIND the value again uses those; re-parsing the string deleted
    the wrong element, and for a dotted key raised KeyError out of the scrub,
    leaving the file to be copied unscrubbed.
    """
    if isinstance(doc, dict):
        for k, v in doc.items():
            yield from walk_json(v, "%s.%s" % (prefix, k) if prefix else str(k),
                                 parts + (("key", k),))
    elif isinstance(doc, list):
        # Indexed, so a decision recorded against one element does not silently
        # cover a different one later.
        for i, v in enumerate(doc):
            yield from walk_json(v, "%s[%d]" % (prefix, i), parts + (("idx", i),))
    else:
        yield (prefix, doc, parts)


def ini_logical_lines(text):
    """Yields (raw_lines, logical_text) with backslash continuations joined.

    A gitconfig value may run across lines, and both reading and removing have
    to treat the whole run as one thing. Reading it line by line finds a
    credential under a key that does not exist - `password=...` on a
    continuation of `helper = !echo \\` parses as its own key - and removing
    one line of it leaves the rest behind."""
    lines = text.splitlines(True)
    i = 0
    while i < len(lines):
        raws = [lines[i]]
        logical = lines[i].rstrip("\n")
        while logical.rstrip().endswith("\\") and i + 1 < len(lines):
            i += 1
            raws.append(lines[i])
            logical = logical.rstrip()[:-1] + lines[i].rstrip("\n")
        yield (raws, logical.strip())
        i += 1


def ini_entries(text):
    """Yields (raw_lines, kind, key, value) where kind is section|pair|other."""
    section = ""
    for raws, line in ini_logical_lines(text):
        if not line or line.startswith("#") or line.startswith(";"):
            yield (raws, "other", None, None)
            continue
        if line.startswith("[") and line.endswith("]"):
            inner = line[1:-1].strip()
            m = re.match(r'^(\S+)\s+"(.*)"$', inner)
            if m:
                section = "%s.%s" % (m.group(1), m.group(2))
                # The subsection text is part of the key, and a URL with
                # userinfo in it is a credential sitting in a section header.
                yield (raws, "section", section, m.group(2))
            else:
                section = inner
                yield (raws, "section", section, None)
            continue
        if "=" not in line:
            yield (raws, "other", None, None)
            continue
        k, _, v = line.partition("=")
        key = "%s.%s" % (section, k.strip()) if section else k.strip()
        yield (raws, "pair", key, v.strip())


def walk_ini(text):
    for _raws, kind, key, value in ini_entries(text):
        if kind in ("section", "pair") and value is not None:
            yield (key, value)


def components_sort_key(parts):
    """Orders paths so a list element is deleted after any later sibling.

    `del cur[i]` shifts every later index down, so applying findings in document
    order deletes the wrong elements from the second onwards: two secrets in one
    array left the second one in the file and removed an innocent value instead,
    and the file is then mounted into the container - the operator answers "keep
    it out" and it goes in. Deepest and highest index first makes each deletion
    independent of the ones after it.
    """
    return [(1, p[1]) if p[0] == "idx" else (0, p[1]) for p in parts]


def delete_json_components(doc, drop):
    """Every component path, in an order under which no deletion invalidates
    another: deepest and highest index first, so a list deletion cannot shift an
    element that is still to be removed."""
    for parts in sorted(drop, key=components_sort_key, reverse=True):
        cur = doc
        for kind, item in parts[:-1]:
            cur = cur[item]
        del cur[parts[-1][1]]


def drop_ini_paths(text, paths):
    """Removes whole sections for a section-header finding, whole logical
    lines otherwise. Line-based rather than via a parser, so a file this
    cannot fully model is changed only where it matched - but a match takes
    every line of the value with it."""
    drop = set(paths)
    out = []
    in_dropped_section = False
    for raws, kind, key, _value in ini_entries(text):
        if kind == "section":
            # A finding on the header itself takes the section with it: the
            # secret is the header, and the keys under it belong to it.
            in_dropped_section = key in drop
            if in_dropped_section:
                continue
            out.extend(raws)
            continue
        if in_dropped_section:
            continue
        if kind == "pair" and key in drop:
            continue
        out.extend(raws)
    return "".join(out)


def load_list(path):
    if not path:
        return []
    out = []
    try:
        with open(path) as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith("#"):
                    out.append(line)
    except OSError:
        pass
    return out


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--file", required=True)
    p.add_argument("--format", required=True, choices=("json", "ini"))
    # What this file is called where an operator would recognise it, rather
    # than the temp path of the copy being scanned.
    p.add_argument("--label", required=True)
    # Paths judged safe, and paths already decided. Both are plain lists of
    # "<label>\t<path>" so bin/dejima can build them without a parser.
    p.add_argument("--known-safe", default=None)
    p.add_argument("--decided", default=None)
    # Ids to remove from the file, in place. The caller never had the real
    # paths, so this is the only way it can ask for one to go.
    p.add_argument("--scrub", default=None)
    args = p.parse_args()

    try:
        with open(args.file) as f:
            text = f.read()
    except OSError as e:
        # The filename, never the contents.
        print("cannot read %s: %s" % (args.label, e.strerror), file=sys.stderr)
        return 2

    if args.format == "json":
        try:
            pairs = list(walk_json(json.loads(text)))
        except ValueError:
            print("%s is not readable as JSON" % args.label, file=sys.stderr)
            return 2
    else:
        pairs = [(p, v, ()) for p, v in walk_ini(text)]

    if args.scrub:
        wanted = set(load_list(args.scrub))
        drop = [(path, parts) for path, value, parts in pairs
                if finding_id(args.label, path) in wanted]
        if not drop:
            return 0
        if args.format == "json":
            doc = json.loads(text)
            delete_json_components(doc, [parts for _path, parts in drop])
            out = json.dumps(doc, indent=2) + "\n"
        else:
            out = drop_ini_paths(text, [p for p, _parts in drop])
        with open(args.file, "w") as f:
            f.write(out)
        return 0

    # Two different keys, deliberately.
    #
    # A judged-safe entry is a "<label>\t<path>" written by hand, and it
    # covers the subtree beneath it - `customApiKeyResponses.approved` has to
    # answer for `...approved[0]` and every sibling, or an array of the same
    # judged-safe thing asks once per element.
    #
    # A decision is an opaque id and matches ONE path exactly. Broadening
    # those the same way would let one answer about one field quietly cover
    # fields the operator never saw.
    safe = [e for e in load_list(args.known_safe) if "\t" in e]
    decided = set(load_list(args.decided))

    def judged_safe(entry):
        for s_entry in safe:
            if entry == s_entry or entry.startswith(s_entry + ".") \
                    or entry.startswith(s_entry + "["):
                return True
        return False

    for path, value, _parts in pairs:
        fid = finding_id(args.label, path)
        if fid in decided or judged_safe("%s\t%s" % (args.label, path)):
            continue
        verdict, shape = classify(path, value)
        if verdict:
            # verdict, id, label, redacted path, shape. No value, and no
            # length either - a length is a hint about a secret.
            print("%s\t%s\t%s\t%s\t%s"
                  % (verdict, fid, args.label, redact(path), shape))
    return 0


if __name__ == "__main__":
    sys.exit(main())
