#!/usr/bin/env python3
"""Find Docker Content Trust / Notary v1 dependencies before notary.docker.io shuts down.

Read-only. Standard library only. Never prints file contents, only path:line, a
category and, for DOCKER_CONTENT_TRUST values, the short value itself. It never opens
anything under a Docker trust directory except to count file names.

  python3 -I check-docker-content-trust.py [PATH ...]    scan files under PATH (default: .)
  python3 -I check-docker-content-trust.py --host        inspect this account's environment,
                                                         shell start-up files and ~/.docker/trust

Exit status: 0 nothing found (disabled-only settings do not count), 1 at least one
dependency or lead found, 2 usage or read error.
CyberNative AI LLC. Checker version 1, written 2026-11-14 for the 2026-12-08 shutdown.
"""
import os
import re
import shutil
import subprocess
import sys
from pathlib import Path
from urllib.parse import urlparse

FALSE_VALUES = {"0", "f", "F", "FALSE", "false", "False"}   # Go strconv.ParseBool
TRUE_VALUES = {"1", "t", "T", "TRUE", "true", "True"}       # same function
SKIP_DIRS = {".git", ".hg", ".svn", "node_modules", "vendor", ".venv", "venv", "__pycache__", ".terraform"}
MAX_BYTES = 1_000_000
SELF_MARK = b"Find Docker Content Trust / Notary v1 dependencies before notary.docker.io shuts down."   # first line of this file's docstring
SKIP_SUFFIXES = (".md", ".markdown", ".rst", ".adoc", ".txt", ".lock", ".sum", ".map", ".min.js", ".svg", ".html", ".htm")
SKIP_NAMES = {"package-lock.json", "yarn.lock", "pnpm-lock.yaml", "CHANGELOG", "LICENSE", "NOTICE"}

ENV_ASSIGN = re.compile(r"""\bDOCKER_CONTENT_TRUST\b["']?\s*[:=]\s*(?P<v>"[^"]*"|'[^']*'|\S*)""")
ENV_DOCKERFILE = re.compile(r"^\s*(?:ENV|ARG)\s+DOCKER_CONTENT_TRUST\s+(?P<v>\S+)", re.I)
ENV_UNSET = re.compile(r"\bunset\s+DOCKER_CONTENT_TRUST\b")
ENV_ANY = re.compile(r"\bDOCKER_CONTENT_TRUST\b")
SERVER = re.compile(r"""\bDOCKER_CONTENT_TRUST_SERVER\b["']?\s*[:=]\s*(?P<v>"[^"]*"|'[^']*'|\S*)""")
PASSPHRASE = re.compile(r"\bDOCKER_CONTENT_TRUST_(ROOT|REPOSITORY)_PASSPHRASE\b")
NOTARY_HOST = re.compile(r"notary\.docker\.io", re.I)
TRUST_CMD = re.compile(r"\bdocker\b[^\n#|;&]{0,40}?\s+trust\s+(sign|inspect|revoke|key|signer)\b")
FLAG = re.compile(r"--disable-content-trust(?:=(?P<v>\w+))?")
NOTARY_CLI = re.compile(r"(?:^|[\s;&|(`])notary\s+(?:-\S+\s+\S+\s+)*(list|add|publish|key|delegation|init|lookup|verify|witness|status|reset|sync|cert|remove|-s|--server)\b")
ADMISSION = re.compile(r"\b(trustServer|notaryv1)\b")

# category -> (counts as a finding?, label)
CATS = {
    "ENABLED": (True, "Content Trust switched on"),
    "SIGNING": (True, "Content Trust signing or Notary client"),
    "SERVER": (True, "names Docker's Notary service or a trust server"),
    "UNRESOLVED": (True, "mentions the switch, value not visible here"),
    "DISABLED": (False, "Content Trust switched off (informational)"),
}


def classify_value(raw):
    v = raw.strip().strip("\"'").strip()
    if v == "" or v in FALSE_VALUES:
        return "DISABLED", v or "(empty)"
    if re.fullmatch(r"[A-Za-z0-9]{1,12}", v):
        return "ENABLED", v          # 1, true, but also yes/on: Docker treats any non-false value as on
    return "UNRESOLVED", "expression or long value"


def scan_line(line):
    """Yield (category, message) for one line of text."""
    m = ENV_DOCKERFILE.match(line) or ENV_ASSIGN.search(line)
    if m:
        cat, v = classify_value(m.group("v"))
        if cat == "ENABLED":
            note = "" if v in TRUE_VALUES else " (not a recognised boolean: Docker still turns Content Trust on)"
            yield cat, f"DOCKER_CONTENT_TRUST={v}{note}"
        elif cat == "DISABLED":
            yield cat, f"DOCKER_CONTENT_TRUST={v}"
        else:
            yield cat, f"DOCKER_CONTENT_TRUST set from an {v}; read the line"
    elif ENV_UNSET.search(line):
        yield "DISABLED", "unset DOCKER_CONTENT_TRUST"
    elif ENV_ANY.search(line):
        yield "UNRESOLVED", "DOCKER_CONTENT_TRUST referenced; its value is set elsewhere"
    s = SERVER.search(line)
    if s:
        v = s.group("v").strip().strip("\"'")
        host = (urlparse(v).hostname or "") if "://" in v else ""
        if host.lower() == "notary.docker.io":
            yield "SERVER", "DOCKER_CONTENT_TRUST_SERVER is notary.docker.io"
        elif host:
            yield "SERVER", f"DOCKER_CONTENT_TRUST_SERVER is {host}, not Docker's service; the shutdown does not remove it"
        else:
            yield "SERVER", "DOCKER_CONTENT_TRUST_SERVER set; host not visible here"
    elif NOTARY_HOST.search(line):
        yield "SERVER", "notary.docker.io named"
    if PASSPHRASE.search(line):
        yield "SIGNING", "Content Trust key passphrase variable (value not printed)"
    t = TRUST_CMD.search(line)
    if t:
        yield "SIGNING", f"docker trust {t.group(1)}"
    f = FLAG.search(line)
    if f:
        if (f.group("v") or "true").lower() in ("false", "0"):
            yield "ENABLED", "--disable-content-trust=false turns verification on for that command"
        else:
            yield "DISABLED", "--disable-content-trust skips verification for that command"
    if NOTARY_CLI.search(line):
        yield "SIGNING", "notary client command"
    a = ADMISSION.search(line)
    if a:
        yield "SERVER", f"{a.group(1)} (admission controller or policy setting for a Notary v1 trust server)"


def wanted(rel):
    """Every text file except documentation, lock files and generated assets. Code is scanned too."""
    name = rel.rsplit("/", 1)[-1]
    return name not in SKIP_NAMES and not name.lower().endswith(SKIP_SUFFIXES)


def scan_file(path, show):
    hits, commented = [], 0
    try:
        if path.stat().st_size > MAX_BYTES:
            return None, "skipped: larger than 1 MB"
        data = path.read_bytes()
    except OSError as e:
        return None, f"unreadable: {e.strerror}"
    if b"\x00" in data[:4096]:
        return None, "skipped: binary"
    if SELF_MARK in data[:400]:
        return None, "skipped: a copy of this checker (it mentions the variable it looks for)"
    for n, line in enumerate(data.decode("utf-8", "replace").splitlines(), 1):
        stripped = line.lstrip()
        for cat, msg in scan_line(line):
            if stripped.startswith(("#", "//")):
                commented += 1
                continue
            hits.append((n, cat, msg))
    return (hits, commented), None


def scan_tree(roots):
    findings, files, notes = [], 0, []
    for root in roots:
        root = Path(root)
        if not root.exists():
            print(f"error: {root} does not exist", file=sys.stderr)
            return None
        if root.is_file():
            candidates = [root]
        else:
            candidates = []
            for dirpath, dirs, names in os.walk(root, followlinks=False):
                dirs[:] = [d for d in dirs if d not in SKIP_DIRS]
                for name in names:
                    p = Path(dirpath, name)
                    if p.is_symlink() or not wanted(p.as_posix()):
                        continue
                    candidates.append(p)
        for p in sorted(candidates):
            res, note = scan_file(p, True)
            if note:
                notes.append(f"{p}: {note}")
                continue
            files += 1
            for n, cat, msg in res[0]:
                findings.append((str(p), n, cat, msg))
    return findings, files, notes


def host_findings():
    out = []
    v = os.environ.get("DOCKER_CONTENT_TRUST")
    if v is not None:
        cat, val = classify_value(v)
        out.append(("environment of this process", 0, cat, f"DOCKER_CONTENT_TRUST={val}" if cat != "UNRESOLVED" else "DOCKER_CONTENT_TRUST set to a long value"))
    s = os.environ.get("DOCKER_CONTENT_TRUST_SERVER")
    if s:
        out.append(("environment of this process", 0, "SERVER", f"DOCKER_CONTENT_TRUST_SERVER host {urlparse(s).hostname or '(invalid)'}"))
    home = Path(os.path.expanduser("~"))
    cfg = Path(os.environ.get("DOCKER_CONFIG") or home / ".docker")
    trust = cfg / "trust"
    if trust.is_dir():
        private = trust / "private"
        keys = sum(1 for _ in private.glob("*")) if private.is_dir() else 0
        tuf = trust / "tuf"
        repos = sum(1 for _ in tuf.glob("*/*")) if tuf.is_dir() else 0
        out.append((str(trust), 0, "SIGNING", f"trust directory present: {keys} key file(s), {repos} cached trust repositor(ies) (names counted, contents not read)"))
    files = [home / n for n in (".bashrc", ".bash_profile", ".profile", ".zshrc", ".zshenv", ".zprofile")]
    files += [Path("/etc/environment"), Path("/etc/profile")]
    pd = Path("/etc/profile.d")
    if pd.is_dir():
        files += sorted(pd.glob("*"))
    scanned = 0
    for f in files:
        if f.is_file():
            res, note = scan_file(f, True)
            if note:
                continue
            scanned += 1
            out += [(str(f), n, c, m) for n, c, m in res[0]]
    docker = shutil.which("docker")
    note = "docker client not found on PATH"
    if docker:
        try:
            r = subprocess.run([docker, "version", "--format", "{{.Client.Version}}"], capture_output=True, text=True, timeout=15)
            ver = r.stdout.strip()
            note = f"docker client version: {ver or '(not reported)'}"
            m = re.match(r"(\d+)\.", ver)
            if m and int(m.group(1)) >= 29:
                # 'docker trust --help' exits 0 even when the command is missing (it prints the root help),
                # so look for the trust command's own usage line instead of the exit status.
                try:
                    t = subprocess.run([docker, "trust", "--help"], capture_output=True, text=True, timeout=15)
                    has_trust = bool(re.search(r"^Usage:\s+docker trust\b", t.stdout, re.M))
                except (OSError, subprocess.SubprocessError):
                    has_trust = None
                if has_trust:
                    note += (" - Docker CLI 29 and later has no built-in Content Trust, but a separately built docker-trust"
                             " plugin is installed here: its 'docker trust' commands still use the Notary service and will fail on December 8")
                elif has_trust is False:
                    note += (" - Docker CLI 29 and later has no built-in Content Trust and no docker-trust plugin was found:"
                             " DOCKER_CONTENT_TRUST is ignored, so this client does not fail on December 8 but also does not verify anything"
                             " (tested: 29.0.0, 29.1.3, 29.9.3)")
                else:
                    note += (" - Docker CLI 29 and later has no built-in Content Trust; 'docker trust --help' did not run, so check"
                             " by hand whether a separately built docker-trust plugin is installed")
        except (OSError, subprocess.SubprocessError):
            note = "docker client found but 'docker version' did not run"
    return out, scanned, [note]


def main(argv):
    if argv and argv[0] in ("-h", "--help"):
        print(__doc__)
        return 0
    if argv and argv[0] == "--host":
        res = host_findings()
        scope = f"host mode: this process's environment, {res[1]} shell start-up file(s), the Docker trust directory"
    else:
        res = scan_tree(argv or ["."])
        scope = f"{res[1] if res else 0} text file(s), skipping documentation (.md .rst .txt .html), lock files, binaries, files over 1 MB and {len(SKIP_DIRS)} dependency or VCS directory names"
    if res is None:
        return 2
    findings, _, notes = res
    findings = sorted(findings)
    for path, n, cat, msg in findings:
        where = f"{path}:{n}" if n else path
        print(f"{where}: [{cat}] {msg}")
    for note in notes:
        print(f"note: {note}")
    counted = [f for f in findings if CATS[f[2]][0]]
    off = len(findings) - len(counted)
    print(f"{len(counted)} finding(s), {off} informational 'switched off' line(s); scanned {scope}.")
    print("Not seen: settings kept in your CI provider's UI or secrets store, the build agent's own environment,"
          " other user accounts, running processes, images that call docker themselves, and registry or cluster policies.")
    return 1 if counted else 0


if __name__ == "__main__":
    try:
        sys.exit(main(sys.argv[1:]))
    except BrokenPipeError:
        sys.exit(2)