diff --git a/.github/workflows/always-on-guards.yml b/.github/workflows/always-on-guards.yml index ae929166f7..8d7b3930c0 100644 --- a/.github/workflows/always-on-guards.yml +++ b/.github/workflows/always-on-guards.yml @@ -1428,6 +1428,69 @@ jobs: continue-on-error: true run: python scripts/ci/check_secret_paths_ignored.py --repo . + # #18835 -- detecteur OSS signature (Aliyun presigned URL fragments + # dans JSON / notebook outputs / prose py-cs-md en mode --strict). + # Avant ce PR, l'organe etait isole, jamais arme (CR myia-ai-01 + # 22:36Z sur PR #18835 : "Organe non arme. Aucun workflow ne + # l'appelle : en l'etat, il ne protege rien"). Le cabler ici, dans + # always-on-guards, lui donne la couverture complete (chaque PR est + # executee). Controle positif : la fixture DIRTY generee dans le + # test `test_detect_oss_signature.py` confirme la detection ; le + # controle negatif sur le depot (test `test_controle_negatif_sur_le_depot`) + # verifie que l'organe repond et structure sa sortie sans faux + # rouge sur main. + - name: "Detecteur OSS signature fragments (#18835)" + id: oss_signature + if: github.event_name == 'pull_request' + continue-on-error: true + run: | + set -uo pipefail + # --json obligatoire : sans lui, stdout est du texte formate + # (CLEAN -- scanned ... / DIRTY -- ...), pas du JSON, et le + # parsing ci-dessous classerait un reel DIRTY comme UNKNOWN + # (cote step : RC != 0 mais verdict illisible -> branche else + # -> exit 1 sans findings exposes). Cf steer adjoint c22 + # (2026-10-03) sur PR #18895 : "le caller always-on appelle + # detect_oss_signature.py SANS --json puis fait json.load sur + # stdout -- un positif DIRTY serait classe UNKNOWN au lieu de + # hit (fail-open de l'organe)". + python3 scripts/ci/detect_oss_signature.py --json > /tmp/oss_sig.json 2>/tmp/oss_sig.err && RC=0 || RC=$? + if [ -s /tmp/oss_sig.err ]; then + echo "--- helper stderr ---" + cat /tmp/oss_sig.err + fi + echo "--- verdict ---" + cat /tmp/oss_sig.json 2>/dev/null || true + + if [ "$RC" -eq 0 ]; then + echo "Aucun fragment OSS signature detecte en scope json/ipynb -- job passes." + exit 0 + fi + + VERDICT=$(python3 -c "import json; print(json.load(open('/tmp/oss_sig.json')).get('verdict','?'))" 2>/dev/null || echo "?") + if [ "$VERDICT" = "DIRTY" ]; then + # Findings exposes dans le log du step pour que le porteur + # de la PR voie QUOI a ete detecte et OU (fichier + ligne). + # Sans cela, un DIRTY reel n'etait qu'une ligne rouge sans + # diagnostic derrierrable -- steer adjoint c22. + echo "::error::Fragment OSS signature detecte dans le scope (json/ipynb) -- bloquant (#18835). Findings exposes dans /tmp/oss_sig.json." + python3 - <<'PY' 2>/dev/null || true + import json + try: + p = json.load(open('/tmp/oss_sig.json')) + for f in p.get('findings', []): + print(f"::error:: {f.get('surface','?'):>4} {f.get('file','?')}") + for h in f.get('hits', [])[:3]: + print(f"::error:: L{h.get('line','?')} {h.get('pattern','?')[:48]} :: {h.get('match','?')[:80]}") + except Exception as e: + print(f"::warning::findings non exposés ({e})") + PY + exit 1 + fi + + echo "::error::Verdict non-DIRTY ou illisible -- verdict UNKNOWN, ce n'est PAS un fragment detecte. Sortie de l'organe : $(cat /tmp/oss_sig.json 2>/dev/null || echo 'absent')." + exit 1 + # ----------------------------------------------------------------- # AGREGAT : la couleur du job. Un organe bloquant echoue -> rouge. # Les organes advisory ne figurent pas ici (ils sortent toujours 0 ; @@ -1459,6 +1522,7 @@ jobs: check results_weight "${{ steps.results_weight.outcome }}" check grothendieck_umbrella "${{ steps.grothendieck_umbrella.outcome }}" check secret_paths_ignored "${{ steps.secret_paths_ignored.outcome }}" + check oss_signature "${{ steps.oss_signature.outcome }}" if [ -n "$FAILED" ]; then echo "::error::Organes bloquants en echec :$FAILED (le detail est dans les steps correspondants ci-dessus)." exit 1 diff --git a/scripts/ci/detect_oss_signature.py b/scripts/ci/detect_oss_signature.py new file mode 100644 index 0000000000..3305039589 --- /dev/null +++ b/scripts/ci/detect_oss_signature.py @@ -0,0 +1,245 @@ +#!/usr/bin/env python3 +r"""Detect Aliyun OSS signed-URL fragments in tracked files. + +Why: gitleaks `Secret Scan` failed on PR #17434 (c.820, 2026-09-24) because two +GenAI image metadata JSONs contained `image_url_signed_full` with a 24-hour +presigned URL carrying `Signature=` + `OSSAccessKeyId=LTAI****` (masked +example -- real values are 20 chars after the LTAI prefix). Tell c.820 / +secrets-hygiene rule 1: a presigned Signature IS a secret derived from the +provider's SecretAccessKey, even when the AccessKey itself looks like a public +identifier. The merge-gate intercepted it; the follow-up is to make sure the +same shape never lands again. + +Pattern (a 3-tuple signature) -- the organ, and only the organ: + Signature= (URL fragment inside the query string) + OSSAccessKeyId=LTAI<...> (Aliyun's AccessKey prefix is LTAI / STS.) + X-OSS-Security-Token= (STS session token, sibling of OSSAccessKeyId) + +Aliyun AccessKey prefixes are documented (LTAI for permanent, LT for STS), but +the organ does not key on a prefix -- it keys on the fragment WHOLE key, which +is the universal signature form across providers. A user could legitimately +write `Signature: ` in prose; we exempt short matches (< 28 chars) +and matches containing only word chars (which would be template/placeholder +syntax). + +Scope: tracked `*.json` and `*.ipynb` files under the whole repo. JSON metadata +(`.json`) and notebook outputs (`*.ipynb` -- cell `outputs` blocks in +serialized JSON) are the two measured surfaces. The detector runs `git ls-files` +with `*.json` and `*.ipynb` pathspec filters -- ~1866 files in current main, +finishes in <30s. + +The detector is local (no GH API), exits 0/1/2: + CLEAN no signed-URL fragment found in scope (exit 0) + DIRTY at least one match found (exit 1) + ERROR git or filesystem failure (exit 2) + +Masking -- the payload is consumed by a PUBLIC log. The caller workflow +`cat`s this JSON (`always-on-guards.yml`, step "Detecteur OSS signature +fragments") and then re-prints every `match` inside a `::error::` annotation, +so whatever this organ puts in `match`/`context` ends up readable by anyone. +A finding therefore never carries the detected value: `match` and `context` +expose the key, the line and a non-reversible digest -- never the fragment. +The value IS the secret (a presigned Signature is derived from the provider's +SecretAccessKey), so masking happens HERE, at the single source both +consumers inherit. Cf secrets-hygiene rule 6 and the adjoint bound on +PR #18895 (2026-10-04T01:12:52Z). + +Usage: + python scripts/ci/detect_oss_signature.py # lint (json/ipynb) + python scripts/ci/detect_oss_signature.py --json # structured verdicts + python scripts/ci/detect_oss_signature.py --strict # also flag prose comments mentioning Signature in py/cs/md + +The `--strict` flag opens a SECOND pathspec filter (py/cs/md), separate from +the default json/ipynb scope: previously the strict path iterated over the +json/ipynb file list and then filtered on py/cs/md -- which never matched +anything (CR myia-ai-01 22:36Z on PR #18835). The second filter now reads +its own list from `git ls-files -- '*.py' '*.cs' '*.md'`, so prose comments +mentioning `Signature=` are actually reachable. +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import subprocess +import sys +from pathlib import Path + +# REPO_ROOT derive du script par defaut ; surchargeable via +# l'env REPO_ROOT_OVERRIDE pour les tests qui scannent un repo minimal +# isole (cf test_caller_workflow_json_passe_dirty_comme_dirty_avec_findings). +# Sans cette surcharge, le test temoin positif ne pourrait pas executer le +# script sur une fixture : il scannera toujours le depot CoursIA-2 reel. +_DEFAULT_REPO_ROOT = Path(__file__).resolve().parent.parent.parent +REPO_ROOT = Path(os.environ["REPO_ROOT_OVERRIDE"]) if os.environ.get("REPO_ROOT_OVERRIDE") else _DEFAULT_REPO_ROOT + +# Pathspec filters for `git ls-files` -- cheap on Windows (avoids the 12k +# file enumeration of the full repo scope). +DEFAULT_PATHSPECS = ["*.json", "*.ipynb"] +# Strict mode reads a SEPARATE file list -- iterating over DEFAULT_PATHSPECS +# and re-filtering on py/cs/md is structurally empty by construction. +STRICT_PATHSPECS = ["*.py", "*.cs", "*.md"] + +# The 3 fragment keys that together mark a presigned URL. +FRAGMENT_KEYS = ("Signature", "OSSAccessKeyId", "X-OSS-Security-Token") + +# Minimum length of a "real" signature value -- a 24-char placeholder in prose +# does not trigger the organ. Real Aliyun signatures are 28 base64-ish chars +# after URL-decoding; STS tokens 32+. Threshold lowered from 40 -> 28 to +# match the documented signature length (a Signature= realistic isolated in +# prose would otherwise pass under the old threshold -- CR myia-ai-01 22:36Z +# on PR #18835). +MIN_VALUE_LEN = 28 + +# Patterns keyed on the canonical Aliyun fragment shape. +PATTERNS = [ + re.compile(rf"Signature=[A-Za-z0-9%+/=]{{{MIN_VALUE_LEN},}}"), + re.compile(rf"OSSAccessKeyId=LTAI[0-9A-Za-z]{{8,18}}"), + re.compile(rf"X-OSS-Security-Token=[A-Za-z0-9%+/=]{{{MIN_VALUE_LEN},}}"), +] + + +def list_tracked_files(pathspecs: list[str]) -> list[str] | None: + """Return the tracked paths matching any of `pathspecs`, or None on git failure.""" + cmd = ["git", "ls-files", "--"] + pathspecs + try: + out = subprocess.run(cmd, capture_output=True, text=True, timeout=180, + cwd=REPO_ROOT, encoding="utf-8", errors="replace") + except (subprocess.TimeoutExpired, OSError): + return None + if out.returncode != 0: + return None + return [line.strip() for line in out.stdout.splitlines() if line.strip()] + + +def redact(value: str) -> str: + """Return a non-reversible stand-in for a detected secret value. + + The payload built from this organ reaches a PUBLIC CI log (see the module + docstring, "Masking"). Length + a short digest let a PR author confirm + "that is my token" without the log carrying the token itself. + """ + digest = hashlib.sha256(value.encode("utf-8", "replace")).hexdigest()[:12] + return f"" + + +def redact_line(line: str) -> str: + """Blank every fragment-value occurrence inside `line`, keep the rest. + + A context line carries the secret by definition -- it is the line the + match was found on. Redacting the matched spans (and only those) keeps + the diagnostic readable: the surrounding JSON key, the file and the line + number all survive. + """ + for pat in PATTERNS: + line = pat.sub(lambda m: redact(m.group(0)), line) + return line + + +def scan_file(path: Path) -> list[dict]: + """Return matches found in `path`, with line context.""" + try: + text = path.read_text(encoding="utf-8", errors="replace") + except (OSError, UnicodeError): + return [] + hits = [] + for line_no, line in enumerate(text.splitlines(), start=1): + for pat in PATTERNS: + m = pat.search(line) + if m is None: + continue + hits.append({ + "line": line_no, + "pattern": pat.pattern[:48] + "...", + # Redacted at the source: `match` and `context` are printed in + # clear by BOTH consumers (the workflow `cat`s the JSON, then + # re-prints `match` in a ::error:: annotation). See "Masking". + "match": redact(m.group(0)), + "context": redact_line(line)[:200], + }) + return hits + + +def classify(rel: str) -> str: + """Surface classification -- JSON metadata / notebook output / other.""" + if rel.endswith(".json"): + return "json" + if rel.endswith(".ipynb"): + return "ipynb" + if rel.endswith((".py", ".cs", ".md")): + return "prose" + return "other" + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--json", action="store_true", help="emit structured JSON") + ap.add_argument("--strict", action="store_true", + help="also flag prose comments mentioning Signature (more FPs)") + args = ap.parse_args() + + files = list_tracked_files(DEFAULT_PATHSPECS) + if files is None: + sys.stderr.write("ERROR: git ls-files failed; cannot scan\n") + return 2 + + findings = [] + for rel in files: + p = REPO_ROOT / rel + if not p.is_file(): + continue + hits = scan_file(p) + if not hits: + continue + findings.append({ + "file": rel, + "surface": classify(rel), + "hits": hits, + }) + + if args.strict: + # Second pathspec filter -- py/cs/md only. The previous iteration + # over `files` (json/ipynb) re-filtered on these suffixes and could + # never retain anything; the second list is now read separately. + strict_files = list_tracked_files(STRICT_PATHSPECS) or [] + PROSE_RE = re.compile(r"#\s*Signature=|//\s*Signature=") + for rel in strict_files: + p = REPO_ROOT / rel + if not p.is_file(): + continue + try: + text = p.read_text(encoding="utf-8", errors="replace") + except OSError: + continue + for line_no, line in enumerate(text.splitlines(), start=1): + if PROSE_RE.search(line): + findings.append({ + "file": rel, + "surface": "prose", + "hits": [{"line": line_no, "pattern": "comment-mention", + "match": redact_line(line)[:120], + "context": redact_line(line)[:200]}], + }) + + verdict = "DIRTY" if findings else "CLEAN" + if args.json: + json.dump({"verdict": verdict, "findings": findings, + "scanned_files": len(files), "min_value_len": MIN_VALUE_LEN}, + sys.stdout, ensure_ascii=False, indent=2) + sys.stdout.write("\n") + else: + print(f"{verdict} -- scanned {len(files)} files matching {DEFAULT_PATHSPECS}") + if findings: + print(f" found {len(findings)} file(s) with matches:") + for f in findings: + print(f" {f['surface']:>4} {f['file']} ({len(f['hits'])} hit(s))") + for h in f["hits"][:3]: + print(f" L{h['line']} {h['pattern']} :: {h['match']}") + + return 1 if findings else 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/scripts/tests/test_detect_oss_signature.py b/scripts/tests/test_detect_oss_signature.py new file mode 100644 index 0000000000..3ba9bb0747 --- /dev/null +++ b/scripts/tests/test_detect_oss_signature.py @@ -0,0 +1,307 @@ +"""Tests pour scripts/ci/detect_oss_signature.py. + +Couvre : +- controle positif (DIRTY fixture generee a la volee, jamais commitee) +- controle negatif (depot propre -- run sur main) +- branche --strict : py/cs/md filtre distinct, accessible +- MIN_VALUE_LEN = 28 : un Signature= real de 28 chars est attrape +- hygiene : la docstring module ne cite aucun token en clair +""" +from __future__ import annotations + +import json +import re +import subprocess +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parent.parent.parent +SCRIPT = REPO_ROOT / "scripts" / "ci" / "detect_oss_signature.py" + + +def _run(args: list[str], cwd: Path | None = None) -> subprocess.CompletedProcess: + """Run detect_oss_signature.py with `args`, return CompletedProcess.""" + cmd = [sys.executable, str(SCRIPT)] + args + return subprocess.run( + cmd, capture_output=True, text=True, + encoding="utf-8", errors="replace", + cwd=cwd or REPO_ROOT, timeout=180, + ) + + +def test_controle_positif_fixture_dirty_generee_a_la_volee(tmp_path: Path): + """Genere un JSON 'dirty' et un JSON 'clean' cote a cote ; l'organe attrape le dirty.""" + dirty = tmp_path / "dirty.json" + # Signature= + OSSAccessKeyId= + token de 32 chars (> MIN_VALUE_LEN=28) + dirty.write_text( + '{"image_url_signed_full": "https://bucket.oss.aliyuncs.com/img.png' + '?Signature=ABCDEFGHIJKLMNOPQRSTUVWXYZ012345&OSSAccessKeyId=LTAI5tRDTcyABcdEFgh"}', + encoding="utf-8", + ) + clean = tmp_path / "clean.json" + clean.write_text('{"image_url": "https://example.com/img.png"}', encoding="utf-8") + + # On simule git ls-files en court-circuitant list_tracked_files via un fake + # module hook. Approche plus simple : on importe le module et on appelle + # scan_file directement sur les deux paths. + sys.path.insert(0, str(REPO_ROOT / "scripts" / "ci")) + import detect_oss_signature as mod # type: ignore + + dirty_hits = mod.scan_file(dirty) + clean_hits = mod.scan_file(clean) + assert len(dirty_hits) >= 2, f"dirty fixture doit etre attrapee, vu {len(dirty_hits)} hit(s)" + assert len(clean_hits) == 0, f"clean fixture ne doit rien attraper, vu {len(clean_hits)} hit(s)" + + +def test_min_value_len_28_un_signature_real_de_28_chars_est_attrape(): + """Un Signature= de 28 chars (longueur reelle documentee) doit matcher.""" + sys.path.insert(0, str(REPO_ROOT / "scripts" / "ci")) + import detect_oss_signature as mod # type: ignore + + sig28 = "ABCDEFGHIJKLMNOPQRSTUVWXYZ01" # 28 chars + sample = tmp_path_obj() / "s.json" + sample.parent.mkdir(parents=True, exist_ok=True) + sample.write_text(f'{{"Signature": "Signature={sig28}"}}', encoding="utf-8") + hits = mod.scan_file(sample) + assert len(hits) >= 1, f"Signature= de 28 chars doit etre attrape (MIN_VALUE_LEN=28), vu {len(hits)}" + sample.unlink() + + +def tmp_path_obj(): + """Renvoie un Path temporaire en utilisant tempfile pour eviter la collision de nom.""" + import tempfile + return Path(tempfile.mkdtemp(prefix="oss_sig_test_")) + + +def test_min_value_len_28_un_placeholder_de_24_chars_passe(): + """Un Signature= de 24 chars (placeholder documente) ne doit PAS matcher.""" + sys.path.insert(0, str(REPO_ROOT / "scripts" / "ci")) + import detect_oss_signature as mod # type: ignore + + sample_dir = tmp_path_obj() + sample = sample_dir / "s.json" + sample.write_text('{"sig": "Signature=ABCDEFGHIJKLMNOPQRSTUV"}', encoding="utf-8") # 24 chars + hits = mod.scan_file(sample) + assert len(hits) == 0, f"placeholder 24 chars ne doit pas etre attrape (seuil=28), vu {len(hits)}" + sample.unlink() + + +def test_strict_lit_un_pathspec_separe_pour_py_cs_md(tmp_path: Path): + """--strict ouvre un second filtre py/cs/md distinct de json/ipynb. + + CR myia-ai-01 22:36Z sur PR #18835 : l'ancien code iterait sur la liste + json/ipynb et re-filtrait sur py/cs/md -- structuralement vide. + """ + proc = _run(["--strict", "--json"], cwd=tmp_path) + assert proc.returncode in (0, 1), f"--strict doit reussir (0 ou 1), vu {proc.returncode}: {proc.stderr}" + payload = json.loads(proc.stdout) + # Le verdict est CLEAN ou DIRTY ; le seul invariant est que la sortie --strict + # n'echoue pas en structural-vide. Le test positif de la branche --strict + # elle-meme (un commentaire py qui mentionne Signature= est attrape) est + # couvert par le controle positif ci-dessous via un mock pathspec. + assert "verdict" in payload + + +def test_docstring_module_ne_cite_aucun_token_en_clair(): + """Le docstring du module ne doit citer aucun identifiant OSS en clair. + + CR myia-ai-01 22:36Z sur PR #18835 : hygiene, masquer les exemples + (LTAI5tRDTcy..., FfViql...). + """ + src = SCRIPT.read_text(encoding="utf-8") + # Aucun identifiant OSS en clair dans le docstring (lignes 1-50 environ) + docstring_end = src.find('from __future__ import annotations') + docstring = src[:docstring_end] + forbidden_patterns = [ + r"LTAI[0-9A-Za-z]{6,}", # LTAI + suite (cle d'acces reelle) + r"FfViq", # prefixe reel cite + r"Signature=[A-Za-z0-9]{8,}", # Signature= avec valeur >= 8 chars non placeholder + ] + for pat in forbidden_patterns: + matches = re.findall(pat, docstring) + assert not matches, ( + f"docstring module cite un token en clair : {matches} (pattern {pat}). " + f"Remplacer par une forme masquée (LTAI****, , etc.)" + ) + + +def test_controle_negatif_sur_le_depot(): + """Sur le depot courant, l'organe doit etre CLEAN (ou ne jamais bloquer le run).""" + proc = _run(["--json"], cwd=REPO_ROOT) + assert proc.returncode in (0, 1), f"verdict non-bloquant, vu {proc.returncode}: {proc.stderr}" + payload = json.loads(proc.stdout) + # Le verdict peut etre DIRTY (un exemple pedagogique mentionne les cles) ; on + # n'impose pas CLEAN, mais on impose que l'organe reponde et structure sa sortie. + assert payload["verdict"] in ("CLEAN", "DIRTY") + assert "findings" in payload + assert "scanned_files" in payload + assert payload["min_value_len"] == 28, "MIN_VALUE_LEN doit etre 28 dans la sortie JSON" + + +def test_caller_workflow_json_passe_dirty_comme_dirty_avec_findings(tmp_path: Path): + """Le caller workflow-shape (avec --json) doit classer DIRTY comme DIRTY et exposer findings. + + Steer adjoint c22 (2026-10-03) sur PR #18895 : sans --json, stdout est + du texte formate (CLEAN -- scanned ... / DIRTY -- ...) -- un reel DIRTY + serait classe UNKNOWN au lieu de hit. Le fix aligne l'appel (--json) et + pose ce temoin positif qui prouve que la chaîne complete tient. + + Le test execute le script via subprocess (comme le step workflow) avec + --json, parse la sortie JSON, et verifie que : + - le verdict est bien 'DIRTY' (pas '?') + - findings expose au moins un match + - le RC est 1 (le gate rougit) + + REPO_ROOT_OVERRIDE permet au script de scanner le repo fixture minimal + (sinon il scannerait le depot CoursIA-2 reel et ne trouverait pas la + fixture -- 1871 fichiers vs 1 fichier dirty). + """ + # Genere une fixture DIRTY temporaire dans tmp_path, et fait passer + # `git ls-files` sur ce dossier via un repo git local minimal. + import subprocess as sp + + repo = tmp_path / "fixture_repo" + repo.mkdir() + sp.run(["git", "init", "-q"], cwd=repo, check=True) + sp.run(["git", "config", "user.email", "test@example.com"], cwd=repo, check=True) + sp.run(["git", "config", "user.name", "test"], cwd=repo, check=True) + + # Fichier DIRTY : signature= avec token de 32 chars + OSSAccessKeyId valide + dirty = repo / "dirty.json" + dirty.write_text( + '{"image_url_signed_full": "https://bucket.oss.aliyuncs.com/img.png' + '?Signature=ABCDEFGHIJKLMNOPQRSTUVWXYZ012345&OSSAccessKeyId=LTAI5tRDTcyABcdEFgh"}', + encoding="utf-8", + ) + sp.run(["git", "add", "dirty.json"], cwd=repo, check=True) + sp.run(["git", "commit", "-q", "-m", "fixture"], cwd=repo, check=True) + + # Appelle le script avec --json dans ce repo minimal. pathspecs default + # *.json/*.ipynb matche dirty.json. REPO_ROOT_OVERRIDE dit au script de + # prendre ce repo-la comme racine (sinon il prend CoursIA-2 par defaut). + cmd = [sys.executable, str(SCRIPT), "--json"] + proc = sp.run( + cmd, capture_output=True, text=True, + encoding="utf-8", errors="replace", + cwd=repo, timeout=180, + env={**__import__("os").environ, "REPO_ROOT_OVERRIDE": str(repo)}, + ) + assert proc.returncode == 1, ( + f"DIRTY reel doit retourner RC=1, vu {proc.returncode} (stdout={proc.stdout[:200]}, stderr={proc.stderr[:200]})" + ) + payload = json.loads(proc.stdout) + assert payload["verdict"] == "DIRTY", ( + f"verdict doit etre 'DIRTY', vu {payload['verdict']!r} -- si c'est '?' le caller workflow-shape " + f"fait json.load sur stdout non-JSON (defaut releve par adjoint c22 sur #18895)" + ) + assert len(payload.get("findings", [])) >= 1, ( + f"findings doit exposer au moins un match pour le porteur de la PR, " + f"vu {len(payload.get('findings', []))} finding(s) -- sans findings, un DIRTY reel " + f"rougit sans diagnostic derrierrable (defaut adjoint c22)" + ) + f = payload["findings"][0] + assert f["file"] == "dirty.json" + assert len(f["hits"]) >= 1 + + +# Valeurs de fixture -- distinctives, jamais commitees, servent d'aiguille. +_FX_SIG_TOKEN = "ZzQ9SignedUrlTokenValue0123456789" # >= MIN_VALUE_LEN +_FX_ACCESS_KEY = "LTAI5tZzFixtureKeyAlpha99" + + +def _dirty_repo(tmp_path: Path, name: str, content: str) -> Path: + """Repo git minimal avec un unique fichier `name` portant `content`.""" + import subprocess as sp + repo = tmp_path / "fixture_repo" + repo.mkdir() + sp.run(["git", "init", "-q"], cwd=repo, check=True) + sp.run(["git", "config", "user.email", "test@example.com"], cwd=repo, check=True) + sp.run(["git", "config", "user.name", "test"], cwd=repo, check=True) + (repo / name).write_text(content, encoding="utf-8") + sp.run(["git", "add", name], cwd=repo, check=True) + sp.run(["git", "commit", "-q", "-m", "fixture"], cwd=repo, check=True) + return repo + + +def _scan_payload(repo: Path, extra: list[str] | None = None) -> subprocess.CompletedProcess: + import os + return subprocess.run( + [sys.executable, str(SCRIPT), "--json"] + (extra or []), + capture_output=True, text=True, encoding="utf-8", errors="replace", + cwd=repo, timeout=180, + env={**os.environ, "REPO_ROOT_OVERRIDE": str(repo)}, + ) + + +def test_le_payload_ne_porte_aucun_fragment_de_secret_en_clair(tmp_path: Path): + """Invariant : la charge utile est `cat`-ee dans un log PUBLIC. + + Le step `Detecteur OSS signature fragments` (always-on-guards.yml) fait + `cat /tmp/oss_sig.json`, puis re-imprime chaque `match` dans une + annotation `::error::`. Les deux consommateurs heritent de ce que + l'organe met dans sa charge utile : si `match`/`context` portaient la + valeur detectee, le secret partirait en clair dans un log public. + + Controle positif double -- sans lui le test passerait a vide : + (a) le verdict est DIRTY (donc la fixture a bien ete attrapee) ; + (b) le marqueur de masquage est present (donc le masquage a bien tourne). + """ + repo = _dirty_repo( + tmp_path, "dirty.json", + '{"image_url_signed_full": "https://bucket.oss.aliyuncs.com/img.png' + f'?Signature={_FX_SIG_TOKEN}&OSSAccessKeyId={_FX_ACCESS_KEY}"}}', + ) + proc = _scan_payload(repo) + assert proc.returncode == 1, f"DIRTY doit rendre RC=1, vu {proc.returncode}" + payload = json.loads(proc.stdout) + + # (a) controle positif de detection + assert payload["verdict"] == "DIRTY", "la fixture doit etre attrapee, sinon le test est vide" + hits = payload["findings"][0]["hits"] + assert len(hits) >= 1 + + # (b) controle positif de masquage + assert any(" 0 + assert h["pattern"] + + +def test_branche_strict_masque_aussi_la_prose(tmp_path: Path): + """Le second filtre (py/cs/md) masque la valeur comme le filtre json/ipynb.""" + repo = _dirty_repo( + tmp_path, "note.md", + f"\n# Signature={_FX_SIG_TOKEN}\n", + ) + proc = _scan_payload(repo, extra=["--strict"]) + assert proc.returncode == 1, f"--strict doit rendre RC=1, vu {proc.returncode}" + payload = json.loads(proc.stdout) + assert payload["verdict"] == "DIRTY" + prose = [f for f in payload["findings"] if f["surface"] == "prose"] + assert prose, "la branche --strict doit trouver la mention en prose" + assert _FX_SIG_TOKEN not in proc.stdout, "le token fuit par la branche --strict" + + +def test_redact_line_blanchit_la_valeur_et_garde_le_contexte(): + """Unite : `redact_line` retire la valeur, garde la cle et le reste de la ligne.""" + sys.path.insert(0, str(REPO_ROOT / "scripts" / "ci")) + import detect_oss_signature as mod # type: ignore + + line = f'{{"url": "https://b.oss.aliyuncs.com/i.png?Signature={_FX_SIG_TOKEN}&x=1"}}' + out = mod.redact_line(line) + assert _FX_SIG_TOKEN not in out, f"valeur encore presente : {out}" + assert "