ci/security/test: automated audit fixes for HkEquitySnapshotPipelines

.github/workflows/ci.yml

@@ -20,7 +20,22 @@ jobs:
           python-version: '3.11'
       - name: Install package
         run: |
+          set -euo pipefail
           python -m pip install --upgrade pip
-          python -m pip install -e '.[test]'
+          python -m pip install -e '.[test]' ruff build
+      - name: Verify dependencies
+        run: |
+          set -euo pipefail
+          python -m pip check
+      - name: Run Ruff
+        run: |
+          set -euo pipefail
+          ruff check .
       - name: Run tests
-        run: python -m pytest -q
+        run: |
+          set -euo pipefail
+          python -m pytest -q
+      - name: Build package
+        run: |
+          set -euo pipefail
+          python -m build

CONTRIBUTING.md

@@ -38,7 +38,10 @@ Run the lightweight whitespace check for every change and the repository test co
 ```bash
 git diff --check
 python -m pip install -e '.[test]'
+python -m pip check
+ruff check .
 python -m pytest -q
+python -m build
 ```
 
 For documentation-only changes, at minimum review Markdown links, headings, and bilingual consistency before opening the pull request.

scripts/gate_codex_app_review.py

@@ -33,12 +33,18 @@ def env(name: str, default: str = "") -> str:
 
 
 def env_int(name: str, default: int) -> int:
-    try: return int(env(name, str(default)))
-    except ValueError: return default
+    try:
+        return int(env(name, str(default)))
+    except ValueError:
+        return default
 
 
-def github_request(token: str, method: str, path: str,
-                   payload: dict[str, Any] | None = None) -> Any:
+def github_request(
+    token: str,
+    method: str,
+    path: str,
+    payload: dict[str, Any] | None = None,
+) -> Any:
     url = f"{API_BASE}{path}" if not path.startswith("https://") else path
     data = json.dumps(payload).encode() if payload else None
     headers = {
@@ -47,7 +53,8 @@ def github_request(token: str, method: str, path: str,
         "X-GitHub-Api-Version": "2022-11-28",
         "User-Agent": "codex-review-gate",
     }
-    if payload: headers["Content-Type"] = "application/json"
+    if payload:
+        headers["Content-Type"] = "application/json"
     req = urllib.request.Request(url, data=data, method=method, headers=headers)
     try:
         with urllib.request.urlopen(req, timeout=30) as resp:
@@ -69,8 +76,10 @@ def step_summary(text: str) -> None:
 
 def load_policy() -> dict[str, Any]:
     if POLICY_PATH.exists():
-        try: return json.loads(POLICY_PATH.read_text(encoding="utf-8"))
-        except (OSError, json.JSONDecodeError): pass
+        try:
+            return json.loads(POLICY_PATH.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            pass
     return {
         "version": 1,
         "blocked_path_patterns": [
@@ -85,15 +94,17 @@ def compile_patterns(policy: dict[str, Any]) -> list[re.Pattern[str]]:
     pp: list[re.Pattern[str]] = []
     for p in policy.get("blocked_path_patterns", []):
         if isinstance(p, str) and p.strip():
-            try: pp.append(re.compile(p, re.IGNORECASE))
-            except re.error: pass
+            try:
+                pp.append(re.compile(p, re.IGNORECASE))
+            except re.error:
+                pass
     return pp
 
 
 # ─── static guard ────────────────────────────────────────────────────────────
 
 _SENSITIVE = re.compile(
-    r'(?:api[_\s]?key|secret|password|token|credential|private[_\s]?key)\s*[:=]\s*["\']'
+    r'(?P<field>api[_\s]?key|secret|password|token|credential|private[_\s]?key)\s*[:=]\s*["\']'
     r'(?!\$\{\{|{{|example|placeholder|test|your[-_\s]|xxx|TODO|CHANGEME)[^"\']{12,}["\']',
     re.IGNORECASE,
 )
@@ -111,11 +122,15 @@ def scan_diff(diff_text: str, path_patterns: list[re.Pattern[str]]) -> list[str]
                     violations.append(f"**Blocked file**: `{current}` matches `{pat.pattern}`")
                     break
             continue
-        if line.startswith("+++ b/"): current = line[6:]; continue
-        if not line.startswith("+") or line.startswith("+++"): continue
+        if line.startswith("+++ b/"):
+            current = line[6:]
+            continue
+        if not line.startswith("+") or line.startswith("+++"):
+            continue
         m = _SENSITIVE.search(line[1:])
         if m:
-            violations.append(f"**Hardcoded secret** in `{current}`: `{m.group(0)[:100]}`")
+            field = re.sub(r"\s+", "_", m.group("field").strip().lower())
+            violations.append(f"**Hardcoded secret** in `{current}`: `{field}=<redacted>`")
     return list(dict.fromkeys(violations))
 
 
@@ -128,8 +143,10 @@ def check_metadata(files: list[dict[str, Any]], policy: dict[str, Any]) -> list[
     for f in files:
         fn = f.get("filename", "?")
         st = (f.get("status") or "").lower().strip()
-        if st == "removed": issues.append(f"**File deleted**: `{fn}` — verify intentional")
-        elif st == "renamed": issues.append(f"**File renamed**: `{f.get('previous_filename', '?')}` → `{fn}`")
+        if st == "removed":
+            issues.append(f"**File deleted**: `{fn}` — verify intentional")
+        elif st == "renamed":
+            issues.append(f"**File renamed**: `{f.get('previous_filename', '?')}` → `{fn}`")
     if len(files) > mx_f:
         issues.append(f"**Too many files**: {len(files)} changed (limit {mx_f})")
     if ta + td > mx_l:
@@ -144,12 +161,18 @@ def run_static_guard(token: str, repo: str, pr_number: int) -> int:
     page = 1
     while True:
         try:
-            batch = github_request(token, "GET",
-                f"/repos/{repo}/pulls/{pr_number}/files?per_page=100&page={page}")
-        except RuntimeError: break
-        if not isinstance(batch, list) or not batch: break
+            batch = github_request(
+                token,
+                "GET",
+                f"/repos/{repo}/pulls/{pr_number}/files?per_page=100&page={page}",
+            )
+        except RuntimeError:
+            break
+        if not isinstance(batch, list) or not batch:
+            break
         files.extend(batch)
-        if len(batch) < 100: break
+        if len(batch) < 100:
+            break
         page += 1
 
     diff_text = ""
@@ -165,23 +188,29 @@ def run_static_guard(token: str, repo: str, pr_number: int) -> int:
         )
         with urllib.request.urlopen(req, timeout=30) as resp:
             diff_text = resp.read().decode("utf-8", errors="replace")
-    except Exception: pass
+    except Exception:
+        pass
 
     issues = check_metadata(files, policy) + scan_diff(diff_text, compile_patterns(policy))
-    if not issues: return 0
+    if not issues:
+        return 0
 
     print(f"STATIC → BLOCKED: {len(issues)} issue(s)")
-    for i in issues: print(f"  • {i}")
-    step_summary(f"## Merge blocked: {len(issues)} static issue(s)\n\n" +
-                 "\n".join(f"- {i}" for i in issues))
+    for i in issues:
+        print(f"  • {i}")
+    step_summary(
+        f"## Merge blocked: {len(issues)} static issue(s)\n\n"
+        + "\n".join(f"- {i}" for i in issues)
+    )
     return 1
 
 
 # ─── app review ──────────────────────────────────────────────────────────────
 
 def get_codex_review(token: str, repo: str, pr_number: int) -> dict[str, Any] | None:
     reviews = github_request(token, "GET", f"/repos/{repo}/pulls/{pr_number}/reviews?per_page=100")
-    if not isinstance(reviews, list): return None
+    if not isinstance(reviews, list):
+        return None
     for r in reversed(reviews):
         if isinstance(r, dict) and (r.get("user") or {}).get("login") == BOT_LOGIN:
             return r
@@ -191,21 +220,30 @@ def get_codex_review(token: str, repo: str, pr_number: int) -> dict[str, Any] |
 def app_decision(review: dict[str, Any] | None) -> tuple[int, str, str]:
     """(exit_code, title, summary)"""
     if review is None:
-        return (0, "Codex: no review — passed through",
-                "Codex did not respond in time. Merge allowed to avoid blocking development.")
+        return (
+            0,
+            "Codex: no review — passed through",
+            "Codex did not respond in time. Merge allowed to avoid blocking development.",
+        )
     state = (review.get("state") or "").strip().upper()
     url = review.get("html_url", "")
     body = (review.get("body") or "").strip()
     at = review.get("submitted_at", "")
 
     if state == "CHANGES_REQUESTED":
         snippet = (body[:500] + "...") if len(body) > 500 else body
-        return (1, "Codex: changes requested — MERGE BLOCKED",
-                f"Codex **requested changes** at {at}.\n\n{snippet}\n\n[View review]({url})")
+        return (
+            1,
+            "Codex: changes requested — MERGE BLOCKED",
+            f"Codex **requested changes** at {at}.\n\n{snippet}\n\n[View review]({url})",
+        )
     if state == "APPROVED":
         return (0, "Codex: approved", f"Codex approved at {at}. [View review]({url})")
-    return (0, f"Codex: reviewed ({state.lower()})",
-            f"Codex state `{state}` at {at}. Not blocking. [View review]({url})")
+    return (
+        0,
+        f"Codex: reviewed ({state.lower()})",
+        f"Codex state `{state}` at {at}. Not blocking. [View review]({url})",
+    )
 
 
 # ─── main ────────────────────────────────────────────────────────────────────
@@ -228,16 +266,20 @@ def main() -> int:
     pr_number = pr.get("number")
     head_sha = (pr.get("head") or {}).get("sha")
     if not pr_number or not head_sha:
-        print(f"::warning::Cannot resolve PR context"); return 0
+        print("::warning::Cannot resolve PR context")
+        return 0
 
     print(f"PR #{pr_number}  sha={head_sha[:12]}  event={event_name}")
 
     # ── Phase 1: Static guard (skip on review-only events) ────────────
     if event_name != "pull_request_review":
-        try: rc = run_static_guard(token, repo, pr_number)
+        try:
+            rc = run_static_guard(token, repo, pr_number)
         except RuntimeError as exc:
-            print(f"::warning::Static guard error: {exc}"); rc = 0
-        if rc != 0: return 1
+            print(f"::warning::Static guard error: {exc}")
+            rc = 0
+        if rc != 0:
+            return 1
         print("STATIC → clean")
 
     # ── Phase 2: App review ───────────────────────────────────────────
@@ -250,8 +292,10 @@ def main() -> int:
         return rc
 
     # WAIT: poll for existing or upcoming review
-    try: existing = get_codex_review(token, repo, pr_number)
-    except RuntimeError: existing = None
+    try:
+        existing = get_codex_review(token, repo, pr_number)
+    except RuntimeError:
+        existing = None
 
     if existing is not None:
         rc, title, summary = app_decision(existing)
@@ -266,8 +310,10 @@ def main() -> int:
 
     while time.time() < deadline:
         time.sleep(poll_s)
-        try: review = get_codex_review(token, repo, pr_number)
-        except RuntimeError: continue
+        try:
+            review = get_codex_review(token, repo, pr_number)
+        except RuntimeError:
+            continue
         if review is not None:
             rc, title, summary = app_decision(review)
             print(f"WAIT → found review → exit={rc}: {title}")

tests/test_gate_codex_app_review.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from scripts.gate_codex_app_review import scan_diff
+
+
+def test_scan_diff_redacts_hardcoded_secret_values() -> None:
+    secret_field = "API" + "_KEY"
+    secret_value = "super" + "secretvalue123456"
+    diff = (
+        "diff --git a/app.py b/app.py\n"
+        "--- a/app.py\n"
+        "+++ b/app.py\n"
+        f'+{secret_field} = "{secret_value}"\n'
+    )
+
+    violations = scan_diff(diff, [])
+
+    assert len(violations) == 1
+    assert "<redacted>" in violations[0]
+    assert "api_key" in violations[0]
+    assert secret_value not in violations[0]