diff --git a/.github/workflows/trivy.yml b/.github/workflows/trivy.yml index d79ec32e1..20a309174 100644 --- a/.github/workflows/trivy.yml +++ b/.github/workflows/trivy.yml @@ -5,10 +5,18 @@ on: branches: - develop - main + pull_request: + branches: + - develop + - main permissions: contents: read +concurrency: + group: trivy-${{ github.repository }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + env: GIT_CONFIG_COUNT: "1" GIT_CONFIG_KEY_0: init.defaultBranch @@ -23,6 +31,8 @@ jobs: security-events: write steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + persist-credentials: false - name: Run Trivy filesystem scan summary uses: aquasecurity/trivy-action@ed142fd0673e97e23eac54620cfb913e5ce36c25 # v0.36.0; SHA pinning retained as supply-chain attack mitigation, do not replace with tag. with: diff --git a/CHANGELOG.md b/CHANGELOG.md index 0b6f7e784..9636b6a9c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,9 @@ ### Fixed +- Run the pinned Trivy filesystem/SARIF scan on pull-request heads targeting `develop` or `main`, so GitHub code scanning can compare the PR configuration instead of reporting the protected-branch Trivy configuration as missing. +- Hardened the Trivy policy regression so ordinary `pull_request` coverage for `develop` and `main` is structurally required and `pull_request_target` is rejected, while CodeQL and OSSF Scorecard remain push-only repository-local signals. +- Reject restrictive Trivy `pull_request.types` filters unless they preserve `opened`, `synchronize`, and `reopened`, so every newly opened, updated, or reopened PR head remains eligible for current-head SARIF evidence. - Upgraded the local score PDF parser to `pdfjs-dist` 6.2.108, pinned Undici 7.29.0 across the workspace, and constrained PDF loading to copied in-memory bytes with a same-origin bundled worker and npm-generated lock provenance. ## [0.1.3] - 2026-04-29 diff --git a/scripts/checks/test_verify_trivy_pr_scan.py b/scripts/checks/test_verify_trivy_pr_scan.py new file mode 100644 index 000000000..ba3da67e3 --- /dev/null +++ b/scripts/checks/test_verify_trivy_pr_scan.py @@ -0,0 +1,594 @@ +"""Regression checks for the Trivy pull-request workflow contract.""" + +from __future__ import annotations + +import subprocess +import sys +import tempfile +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +TRIVY_CONTRACT_CHECKER = REPO_ROOT / "scripts" / "checks" / "verify_trivy_pr_scan.py" + +MISSING_PR_TARGETS = """name: trivy + +on: + push: + branches: + - develop + - main + pull_request: + types: [opened] + +jobs: + trivy-fs-scan: + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +RESTRICTED_PR_ACTIVITY = """name: trivy + +on: + push: + branches: + - develop + - main + pull_request: + branches: + - develop + - main + types: [opened] + +jobs: + trivy-fs-scan: + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +TARGET_ONLY = """name: trivy + +on: + push: + branches: + - develop + - main + pull_request_target: + branches: + - develop + - main + +jobs: + trivy-fs-scan: + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +MIXED_PR_TARGET = """name: trivy + +on: + push: + branches: + - develop + - main + pull_request: + branches: + - develop + - main + pull_request_target: + branches: + - develop + - main + +jobs: + trivy-fs-scan: + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +COMMENTED_MIXED_PR_TARGET = """name: trivy + +on: + push: + branches: + - develop + - main + pull_request: + branches: + - develop + - main + pull_request_target: # privileged event must not be hidden by a comment + branches: + - develop + - main + +jobs: + trivy-fs-scan: + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +WRONG_PR_TARGETS = """name: trivy + +on: + push: + branches: + - develop + - main + pull_request: + branches: + - develop + - release + +jobs: + trivy-fs-scan: + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +DISCONNECTED_SARIF = """name: trivy + +on: + pull_request: + branches: + - develop + - main + +jobs: + trivy-fs-scan: + steps: + - name: Unrelated formatter + run: echo harmless + with: + format: sarif + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: table + output: trivy-results.txt + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: unrelated.sarif +""" + +MISMATCHED_SARIF = """name: trivy + +on: + pull_request: + branches: + - develop + - main + +jobs: + trivy-fs-scan: + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: different-results.sarif +""" + +UPLOAD_BEFORE_PRODUCER = """name: trivy + +on: + pull_request: + branches: + - develop + - main + +jobs: + trivy-fs-scan: + steps: + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif +""" + +JOB_PUSH_ONLY = """name: trivy + +on: + pull_request: + branches: + - develop + - main + +jobs: + trivy-fs-scan: + if: github.event_name == 'push' + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +TRIVY_STEP_PUSH_ONLY = """name: trivy + +on: + pull_request: + branches: + - develop + - main + +jobs: + trivy-fs-scan: + steps: + - name: Run Trivy filesystem scan + if: github.event_name == 'push' + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +UPLOAD_STEP_PUSH_ONLY = """name: trivy + +on: + pull_request: + branches: + - develop + - main + +jobs: + trivy-fs-scan: + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + if: github.event_name == 'push' + with: + sarif_file: trivy-results.sarif +""" + +PUSH_ONLY_NEEDS_SCALAR = """name: trivy + +on: + pull_request: + branches: + - develop + - main + +jobs: + prepare-scan: + if: github.event_name == 'push' + steps: + - run: echo prepare + trivy-fs-scan: + needs: prepare-scan + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +PUSH_ONLY_NEEDS_INLINE = """name: trivy + +on: + pull_request: + branches: + - develop + - main + +jobs: + prepare-scan: + if: github.event_name == 'push' + steps: + - run: echo prepare + trivy-fs-scan: + needs: [prepare-scan] + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +PUSH_ONLY_NEEDS_BLOCK = """name: trivy + +on: + pull_request: + branches: + - develop + - main + +jobs: + prepare-scan: + if: github.event_name == 'push' + steps: + - run: echo prepare + trivy-fs-scan: + needs: + - prepare-scan + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +INLINE_COMMENTED_SARIF = """name: trivy + +on: + pull_request: + branches: + - develop + - main + +jobs: + trivy-fs-scan: + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif # GitHub code scanning format + output: trivy-results.sarif # produced by Trivy + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif # upload the same result +""" + +QUOTED_HASH_SARIF = """name: trivy + +on: + pull_request: + branches: + - develop + - main + +jobs: + trivy-fs-scan: + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: "sarif" # GitHub code scanning format + output: "trivy#results.sarif" # # inside quotes is data + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: 'trivy#results.sarif' # same path, different YAML quoting +""" + +EXPLICIT_COMPLETE_PR_ACTIVITY = """name: trivy + +on: + pull_request: + branches: + - develop + - main + types: [opened, synchronize, reopened] + +jobs: + trivy-fs-scan: + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +QUOTED_BLOCK_PR_ACTIVITY = """name: trivy + +on: + pull_request: + branches: + - "develop" # protected development branch + - 'main' + types: + - "opened" # newly opened pull request + - 'synchronize' # updated pull-request head + - reopened # restored pull request + +jobs: + trivy-fs-scan: + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +EXPLICIT_PR_ELIGIBILITY = """name: trivy + +on: + pull_request: + branches: + - develop + - main + +jobs: + trivy-fs-scan: + if: ${{ github.event_name == 'pull_request' }} + steps: + - name: Run Trivy filesystem scan + if: github.event_name == 'pull_request' + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + if: always() + with: + sarif_file: trivy-results.sarif +""" + +PR_ELIGIBLE_NEEDS_CHAIN = """name: trivy + +on: + pull_request: + branches: + - develop + - main + +jobs: + establish-context: + if: github.event_name == 'pull_request' + steps: + - run: echo context + prepare-scan: + needs: [establish-context] + steps: + - run: echo prepare + trivy-fs-scan: + needs: + - prepare-scan + steps: + - name: Run Trivy filesystem scan + uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +INVALID_CASES = { + "missing protected PR targets": MISSING_PR_TARGETS, + "activity filter drops synchronized PR heads": RESTRICTED_PR_ACTIVITY, + "target-only privileged PR event": TARGET_ONLY, + "mixed pull_request and pull_request_target events": MIXED_PR_TARGET, + "commented privileged PR event": COMMENTED_MIXED_PR_TARGET, + "wrong pull_request branch set": WRONG_PR_TARGETS, + "SARIF format detached from the Trivy action": DISCONNECTED_SARIF, + "Trivy output and upload paths disagree": MISMATCHED_SARIF, + "SARIF upload precedes its producer": UPLOAD_BEFORE_PRODUCER, + "push-only Trivy job condition": JOB_PUSH_ONLY, + "push-only Trivy action condition": TRIVY_STEP_PUSH_ONLY, + "push-only SARIF upload condition": UPLOAD_STEP_PUSH_ONLY, + "scalar push-only prerequisite": PUSH_ONLY_NEEDS_SCALAR, + "inline-list push-only prerequisite": PUSH_ONLY_NEEDS_INLINE, + "block-list push-only prerequisite": PUSH_ONLY_NEEDS_BLOCK, +} + +VALID_CASES = { + "equivalent SARIF paths with inline comments": INLINE_COMMENTED_SARIF, + "quoted SARIF path containing a literal hash": QUOTED_HASH_SARIF, + "explicit complete PR-head activity filter": EXPLICIT_COMPLETE_PR_ACTIVITY, + "quoted block PR-head activity filter": QUOTED_BLOCK_PR_ACTIVITY, + "explicit pull-request eligibility conditions": EXPLICIT_PR_ELIGIBILITY, + "recursive PR-eligible needs chain": PR_ELIGIBLE_NEEDS_CHAIN, +} + + +def _run_checker(workflow_text: str) -> subprocess.CompletedProcess[str]: + """Run the production checker against one isolated workflow fixture.""" + with tempfile.TemporaryDirectory() as temp_dir: + workflow_path = Path(temp_dir) / ".github" / "workflows" / "trivy.yml" + workflow_path.parent.mkdir(parents=True) + workflow_path.write_text(workflow_text, encoding="utf-8") + return subprocess.run( + [sys.executable, str(TRIVY_CONTRACT_CHECKER)], + cwd=temp_dir, + capture_output=True, + check=False, + text=True, + ) + + +def main() -> int: + """Reject unsafe wiring without rejecting valid YAML scalar comments.""" + accepted_invalid_case_names = [ + case_name + for case_name, workflow_text in INVALID_CASES.items() + if _run_checker(workflow_text).returncode == 0 + ] + rejected_valid_case_names = [ + case_name + for case_name, workflow_text in VALID_CASES.items() + if _run_checker(workflow_text).returncode != 0 + ] + + if accepted_invalid_case_names or rejected_valid_case_names: + print("Trivy PR contract regression:") + for case_name in accepted_invalid_case_names: + print(f"- accepted malformed workflow: {case_name}") + for case_name in rejected_valid_case_names: + print(f"- rejected valid workflow: {case_name}") + return 1 + print("Trivy PR contract regressions passed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/checks/test_verify_trivy_pr_scan_branch_patterns.py b/scripts/checks/test_verify_trivy_pr_scan_branch_patterns.py new file mode 100644 index 000000000..bd6f90b33 --- /dev/null +++ b/scripts/checks/test_verify_trivy_pr_scan_branch_patterns.py @@ -0,0 +1,85 @@ +"""Regression coverage for ordered Trivy pull-request branch filters.""" + +from __future__ import annotations + +import importlib.util +import tempfile +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +CHECKER_PATH = REPO_ROOT / "scripts" / "checks" / "verify_trivy_pr_scan.py" + +WORKFLOW_TEMPLATE = """name: trivy + +on: + pull_request: + branches: +{branch_items} + types: [opened, synchronize, reopened] + +jobs: + trivy-fs-scan: + steps: + - uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + + +def _load_checker(): + spec = importlib.util.spec_from_file_location("verify_trivy_pr_scan", CHECKER_PATH) + if spec is None or spec.loader is None: + raise RuntimeError("unable to load Trivy contract checker") + checker = importlib.util.module_from_spec(spec) + spec.loader.exec_module(checker) + return checker + + +def _workflow(branch_patterns: list[str]) -> str: + branch_items = "\n".join(f" - '{branch_pattern}'" for branch_pattern in branch_patterns) + return WORKFLOW_TEMPLATE.format(branch_items=branch_items) + + +def _check(checker, workflow_text: str) -> int: + with tempfile.TemporaryDirectory() as temp_dir: + workflow_path = Path(temp_dir) / "trivy.yml" + workflow_path.write_text(workflow_text, encoding="utf-8") + checker.TRIVY_WORKFLOW = workflow_path + return checker.main() + + +def main() -> int: + """Prove ordered negative filters cannot hide protected PR heads.""" + checker = _load_checker() + invalid_cases = { + "late exact exclusions": ["develop", "main", "!develop", "!main"], + "late glob exclusion": ["develop", "main", "!m*"], + } + valid_cases = { + "later positive re-inclusion": ["!develop", "!main", "develop", "main"], + "unrelated exclusion": ["develop", "main", "!release/**"], + } + + failures: list[str] = [] + for case_name, branch_patterns in invalid_cases.items(): + if _check(checker, _workflow(branch_patterns)) == 0: + failures.append(f"accepted invalid case: {case_name}") + for case_name, branch_patterns in valid_cases.items(): + if _check(checker, _workflow(branch_patterns)) != 0: + failures.append(f"rejected valid case: {case_name}") + + if failures: + print("Trivy ordered branch-pattern regressions failed:") + for failure in failures: + print(f"- {failure}") + return 1 + print("Trivy ordered branch-pattern regressions passed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/checks/test_verify_trivy_pr_scan_commented_needs.py b/scripts/checks/test_verify_trivy_pr_scan_commented_needs.py new file mode 100644 index 000000000..b2540590c --- /dev/null +++ b/scripts/checks/test_verify_trivy_pr_scan_commented_needs.py @@ -0,0 +1,92 @@ +"""Regressions for inline-commented Trivy prerequisite job headers.""" + +from __future__ import annotations + +import subprocess +import sys +import tempfile +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +TRIVY_CONTRACT_CHECKER = REPO_ROOT / "scripts" / "checks" / "verify_trivy_pr_scan.py" + +COMMENTED_PREREQUISITE = """name: trivy + +on: + pull_request: + branches: [develop, main] + +jobs: + prepare-scan: # shared setup + if: github.event_name == 'pull_request' + steps: + - run: echo prepare + trivy-fs-scan: + needs: prepare-scan + steps: + - uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + +PREFIXED_PREREQUISITE_ONLY = """name: trivy + +on: + pull_request: + branches: [develop, main] + +jobs: + prepare-scan-extra: # must not satisfy needs: prepare-scan + if: github.event_name == 'pull_request' + steps: + - run: echo prepare + trivy-fs-scan: + needs: [prepare-scan] + steps: + - uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +""" + + +def _run_checker(workflow_text: str) -> subprocess.CompletedProcess[str]: + """Run the production contract checker against one isolated workflow.""" + with tempfile.TemporaryDirectory() as temporary_directory: + workflow_path = Path(temporary_directory) / ".github" / "workflows" / "trivy.yml" + workflow_path.parent.mkdir(parents=True) + workflow_path.write_text(workflow_text, encoding="utf-8") + return subprocess.run( + [sys.executable, str(TRIVY_CONTRACT_CHECKER)], + cwd=temporary_directory, + capture_output=True, + check=False, + text=True, + ) + + +def main() -> int: + """Accept an exact commented job key and reject a similarly prefixed key.""" + commented_result = _run_checker(COMMENTED_PREREQUISITE) + prefixed_result = _run_checker(PREFIXED_PREREQUISITE_ONLY) + if commented_result.returncode != 0: + print("Commented prerequisite job header was rejected:") + print(commented_result.stdout) + print(commented_result.stderr) + return 1 + if prefixed_result.returncode == 0: + print("Similarly prefixed prerequisite job incorrectly satisfied needs") + return 1 + print("Trivy commented-needs regressions passed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/checks/test_verify_trivy_pr_scan_upload_guards.py b/scripts/checks/test_verify_trivy_pr_scan_upload_guards.py new file mode 100644 index 000000000..b80ae265f --- /dev/null +++ b/scripts/checks/test_verify_trivy_pr_scan_upload_guards.py @@ -0,0 +1,33 @@ +"""Regression coverage for failure-safe Trivy SARIF upload conditions.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +CHECKER_PATH = Path(__file__).with_name("verify_trivy_pr_scan.py") +CHECKER_SPEC = importlib.util.spec_from_file_location("verify_trivy_pr_scan", CHECKER_PATH) +if CHECKER_SPEC is None or CHECKER_SPEC.loader is None: + raise RuntimeError("could not load Trivy PR contract checker") +CHECKER_MODULE = importlib.util.module_from_spec(CHECKER_SPEC) +CHECKER_SPEC.loader.exec_module(CHECKER_MODULE) + +FAILURE_SAFE_PR_CONDITIONS = ( + "always() && github.event_name == 'pull_request'", + 'always() && github.event_name == "pull_request"', + "github.event_name == 'pull_request' && always()", + 'github.event_name == "pull_request" && always()', +) + +for upload_condition in FAILURE_SAFE_PR_CONDITIONS: + assert CHECKER_MODULE._condition_preserves_pull_request_eligibility(upload_condition), upload_condition + assert CHECKER_MODULE._condition_runs_after_prior_failure(upload_condition), upload_condition + +for rejected_condition in ( + "always() && github.event_name == 'push'", + "github.event_name == 'push' && always()", + "success() && github.event_name == 'pull_request'", +): + assert not CHECKER_MODULE._condition_preserves_pull_request_eligibility(rejected_condition), rejected_condition + +print("Trivy failure-safe PR upload-condition regressions passed") diff --git a/scripts/checks/test_verify_trivy_pr_scan_yaml_scalars.py b/scripts/checks/test_verify_trivy_pr_scan_yaml_scalars.py new file mode 100644 index 000000000..cbf6b5773 --- /dev/null +++ b/scripts/checks/test_verify_trivy_pr_scan_yaml_scalars.py @@ -0,0 +1,89 @@ +"""Focused regressions for YAML-equivalent Trivy trigger scalars.""" + +from __future__ import annotations + +import importlib.util +import tempfile +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +CHECKER_PATH = REPO_ROOT / "scripts" / "checks" / "verify_trivy_pr_scan.py" + +ESCAPED_BLOCK_LISTS = r'''name: trivy + +on: + pull_request: + branches: + - "\u0064evelop" + - "\u006dain" + types: + - "\u006fpened" + - "\u0073ynchronize" + - "\u0072eopened" + +jobs: + trivy-fs-scan: + steps: + - uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +''' + +ESCAPED_INLINE_LISTS = r'''name: trivy + +on: + pull_request: + branches: ["\u0064evelop", "\u006dain"] + types: ["\u006fpened", "\u0073ynchronize", "\u0072eopened"] + +jobs: + trivy-fs-scan: + steps: + - uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + with: + sarif_file: trivy-results.sarif +''' + + +def _load_checker(): + spec = importlib.util.spec_from_file_location("verify_trivy_pr_scan", CHECKER_PATH) + if spec is None or spec.loader is None: + raise RuntimeError("unable to load Trivy contract checker") + checker = importlib.util.module_from_spec(spec) + spec.loader.exec_module(checker) + return checker + + +def main() -> int: + """Require escaped block and inline YAML lists to preserve contract semantics.""" + checker = _load_checker() + rejected_cases: list[str] = [] + for case_name, workflow_text in { + "escaped block lists": ESCAPED_BLOCK_LISTS, + "escaped inline lists": ESCAPED_INLINE_LISTS, + }.items(): + with tempfile.TemporaryDirectory() as temp_dir: + workflow_path = Path(temp_dir) / "trivy.yml" + workflow_path.write_text(workflow_text, encoding="utf-8") + checker.TRIVY_WORKFLOW = workflow_path + if checker.main() != 0: + rejected_cases.append(case_name) + if rejected_cases: + print("Trivy YAML scalar regression:") + for case_name in rejected_cases: + print(f"- rejected valid workflow: {case_name}") + return 1 + print("Trivy YAML scalar regressions passed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/checks/verify_trivy_pr_scan.py b/scripts/checks/verify_trivy_pr_scan.py new file mode 100644 index 000000000..14b6ead52 --- /dev/null +++ b/scripts/checks/verify_trivy_pr_scan.py @@ -0,0 +1,469 @@ +"""Fail closed when Trivy code scanning cannot run on pull-request heads.""" + +from fnmatch import fnmatchcase +from pathlib import Path + +TRIVY_WORKFLOW = Path(".github/workflows/trivy.yml") + + +def _indented_block(workflow_lines: list[str], mapping_header: str, mapping_indent: int) -> list[str]: + """Return the YAML-like block nested under an exact-indentation mapping key.""" + indent_prefix = " " * mapping_indent + mapping_target = f"{indent_prefix}{mapping_header}:" + for line_index, workflow_line in enumerate(workflow_lines): + mapping_suffix = ( + workflow_line[len(mapping_target) :] + if workflow_line.startswith(mapping_target) + else "" + ) + has_inline_comment = ( + bool(mapping_suffix) + and mapping_suffix[0].isspace() + and mapping_suffix.lstrip().startswith("#") + ) + if workflow_line != mapping_target and not has_inline_comment: + continue + nested_block: list[str] = [] + for candidate_line in workflow_lines[line_index + 1 :]: + candidate_text = candidate_line.strip() + if not candidate_text or candidate_text.startswith("#"): + nested_block.append(candidate_line) + continue + candidate_indent = len(candidate_line) - len(candidate_line.lstrip(" ")) + if candidate_indent <= mapping_indent: + break + nested_block.append(candidate_line) + return nested_block + return [] + + +def _has_mapping_key(workflow_lines: list[str], mapping_header: str, mapping_indent: int) -> bool: + """Return whether ``mapping_header`` is a key at exactly ``mapping_indent``.""" + mapping_prefix = f"{' ' * mapping_indent}{mapping_header}:" + return any(workflow_line.startswith(mapping_prefix) for workflow_line in workflow_lines) + + +def _list_sequence(workflow_lines: list[str], mapping_header: str, mapping_indent: int) -> list[str]: + """Return normalized YAML scalar list items in source order.""" + nested_block = _indented_block(workflow_lines, mapping_header, mapping_indent) + item_prefix = " " * (mapping_indent + 2) + "- " + normalized_items: list[str] = [] + for workflow_line in nested_block: + if not workflow_line.startswith(item_prefix): + continue + raw_list_item = workflow_line[len(item_prefix) :].strip() + if raw_list_item and (normalized_item := _yaml_scalar(raw_list_item)): + normalized_items.append(normalized_item) + return normalized_items + + +def _list_values(workflow_lines: list[str], mapping_header: str, mapping_indent: int) -> set[str]: + """Return normalized YAML scalar list items under the requested mapping key.""" + return set(_list_sequence(workflow_lines, mapping_header, mapping_indent)) + + +def _mapping_list_sequence(workflow_lines: list[str], mapping_header: str, mapping_indent: int) -> list[str]: + """Return block- or inline-list scalar values in YAML source order.""" + mapping_prefix = f"{' ' * mapping_indent}{mapping_header}:" + for workflow_line in workflow_lines: + if not workflow_line.startswith(mapping_prefix): + continue + scalar_value = _yaml_scalar(workflow_line[len(mapping_prefix) :].strip()) + if not scalar_value: + return _list_sequence(workflow_lines, mapping_header, mapping_indent) + if scalar_value.startswith("[") and scalar_value.endswith("]"): + return [ + normalized_item + for item in scalar_value[1:-1].split(",") + if (normalized_item := _yaml_scalar(item.strip())) + ] + return [scalar_value] + return [] + + +def _mapping_list_values(workflow_lines: list[str], mapping_header: str, mapping_indent: int) -> set[str]: + """Return block- or inline-list scalar values for one mapping key.""" + return set(_mapping_list_sequence(workflow_lines, mapping_header, mapping_indent)) + + +def _branch_patterns_allow(branch_patterns: list[str], protected_branch: str) -> bool: + """Evaluate ordered GitHub branch include/exclude patterns for one branch. + + GitHub evaluates ``branches`` patterns in order: a matching ``!`` pattern + excludes a previously included ref, while a later positive pattern can + re-include it. Preserve that ordering so a contract checker cannot be + fooled by merely seeing ``develop``/``main`` somewhere in the list. + ``fnmatchcase`` covers the ordinary glob forms relevant to these literal + protected branch names; patterns that do not match simply leave the prior + decision unchanged. + """ + included = False + for branch_pattern in branch_patterns: + is_negative = branch_pattern.startswith("!") + effective_pattern = branch_pattern[1:] if is_negative else branch_pattern + if not effective_pattern: + continue + if fnmatchcase(protected_branch, effective_pattern): + included = not is_negative + return included + + +def _list_item_blocks(workflow_lines: list[str], mapping_header: str, mapping_indent: int) -> list[list[str]]: + """Split one YAML-like sequence block into its top-level item blocks.""" + nested_block = _indented_block(workflow_lines, mapping_header, mapping_indent) + item_prefix = " " * (mapping_indent + 2) + "- " + item_blocks: list[list[str]] = [] + current_item_block: list[str] = [] + for workflow_line in nested_block: + if workflow_line.startswith(item_prefix): + if current_item_block: + item_blocks.append(current_item_block) + current_item_block = [workflow_line] + elif current_item_block: + current_item_block.append(workflow_line) + if current_item_block: + item_blocks.append(current_item_block) + return item_blocks + + +def _step_action(workflow_step: list[str]) -> str | None: + """Return the action reference from a workflow step, if the step uses one.""" + for workflow_line in workflow_step: + line_text = workflow_line.strip() + if line_text.startswith("- uses:"): + return line_text.removeprefix("- uses:").strip() + if line_text.startswith("uses:"): + return line_text.removeprefix("uses:").strip() + return None + + +def _step_mapping_value(workflow_step: list[str], mapping_key: str) -> str | None: + """Return a direct scalar mapping value from one workflow step.""" + for workflow_line in workflow_step: + line_text = workflow_line.strip() + for mapping_prefix in (f"- {mapping_key}:", f"{mapping_key}:"): + if line_text.startswith(mapping_prefix): + return _yaml_scalar(line_text[len(mapping_prefix) :].strip()) + return None + + +def _direct_mapping_value(workflow_lines: list[str], mapping_key: str, mapping_indent: int) -> str | None: + """Return a scalar value from a mapping key at one exact indentation.""" + mapping_prefix = f"{' ' * mapping_indent}{mapping_key}:" + for workflow_line in workflow_lines: + if workflow_line.startswith(mapping_prefix): + return _yaml_scalar(workflow_line[len(mapping_prefix) :].strip()) + return None + + +def _normalized_condition(condition_text: str | None) -> str | None: + """Return a whitespace-free GitHub condition expression, if one exists.""" + if condition_text is None: + return None + normalized_condition = condition_text.strip() + if normalized_condition.startswith("${{") and normalized_condition.endswith("}}"): + normalized_condition = normalized_condition[3:-2].strip() + return "".join(normalized_condition.split()) + + +def _condition_preserves_pull_request_eligibility(condition_text: str | None) -> bool: + """Accept only conditions proven not to exclude ordinary pull-request runs. + + Missing conditions inherit GitHub's normal job/step eligibility. Explicit + conditions are intentionally fail-closed: only unconditional forms and + direct pull-request gates, including failure-safe ``always()`` conjunctions, + are accepted. More complex expressions must be made structurally auditable + before this admission checker can rely on them. + """ + compact_condition = _normalized_condition(condition_text) + if compact_condition is None: + return True + if compact_condition.lower() == "true": + return True + if compact_condition in {"always()", "success()"}: + return True + return compact_condition in { + "github.event_name=='pull_request'", + 'github.event_name=="pull_request"', + "'pull_request'==github.event_name", + '"pull_request"==github.event_name', + "always()&&github.event_name=='pull_request'", + 'always()&&github.event_name=="pull_request"', + "github.event_name=='pull_request'&&always()", + 'github.event_name=="pull_request"&&always()', + } + + +def _condition_runs_after_prior_failure(condition_text: str | None) -> bool: + """Require an upload condition that survives a preceding Trivy exit code 1. + + GitHub implicitly applies ``success()`` to a step without a status-check + function, so an absent condition, ``true``, or explicit ``success()`` is + insufficient after Trivy deliberately exits non-zero for findings. Keep + this fail-closed and accept only ``always()`` or a direct PR gate conjoined + with ``always()`` until a broader expression parser is justified. + """ + compact_condition = _normalized_condition(condition_text) + if compact_condition == "always()": + return True + return compact_condition in { + "always()&&github.event_name=='pull_request'", + 'always()&&github.event_name=="pull_request"', + "github.event_name=='pull_request'&&always()", + 'github.event_name=="pull_request"&&always()', + } + + +def _job_preserves_pull_request_eligibility( + jobs_block: list[str], + job_name: str, + visiting_job_names: set[str] | None = None, +) -> bool: + """Require a job and its complete ``needs`` chain to remain PR-eligible. + + ``needs`` participates in GitHub's admission semantics: a Trivy job can + have a harmless-looking condition yet still be skipped when a prerequisite + is push-only. Resolve scalar, inline-list, and block-list dependencies + recursively. Missing jobs, cycles, or conditions whose pull-request + eligibility cannot be established fail closed. + """ + active_job_names = set(visiting_job_names or set()) + if job_name in active_job_names: + return False + active_job_names.add(job_name) + job_block = _indented_block(jobs_block, job_name, 2) + if not job_block: + return False + job_condition = _direct_mapping_value(job_block, "if", 4) + if not _condition_preserves_pull_request_eligibility(job_condition): + return False + dependency_names = _mapping_list_sequence(job_block, "needs", 4) + return all( + _job_preserves_pull_request_eligibility(jobs_block, dependency_name, active_job_names) + for dependency_name in dependency_names + ) + + +def _decode_yaml_double_quoted_scalar(quoted_scalar: str) -> str | None: + """Decode a one-line YAML double-quoted scalar without external dependencies.""" + yaml_simple_escapes = { + "0": "\0", + "a": "\a", + "b": "\b", + "t": "\t", + "n": "\n", + "v": "\v", + "f": "\f", + "r": "\r", + "e": "\x1b", + " ": " ", + '"': '"', + "/": "/", + "\\": "\\", + "N": "\u0085", + "_": "\u00a0", + "L": "\u2028", + "P": "\u2029", + } + decoded_characters: list[str] = [] + scalar_index = 1 + scalar_end = len(quoted_scalar) - 1 + while scalar_index < scalar_end: + scalar_character = quoted_scalar[scalar_index] + if scalar_character != "\\": + decoded_characters.append(scalar_character) + scalar_index += 1 + continue + scalar_index += 1 + if scalar_index >= scalar_end: + return None + escape_character = quoted_scalar[scalar_index] + if escape_character in yaml_simple_escapes: + decoded_characters.append(yaml_simple_escapes[escape_character]) + scalar_index += 1 + continue + hexadecimal_lengths = {"x": 2, "u": 4, "U": 8} + hexadecimal_length = hexadecimal_lengths.get(escape_character) + if hexadecimal_length is None: + return None + hexadecimal_start = scalar_index + 1 + hexadecimal_end = hexadecimal_start + hexadecimal_length + hexadecimal_text = quoted_scalar[hexadecimal_start:hexadecimal_end] + if len(hexadecimal_text) != hexadecimal_length: + return None + try: + decoded_characters.append(chr(int(hexadecimal_text, 16))) + except (ValueError, OverflowError): + return None + scalar_index = hexadecimal_end + return "".join(decoded_characters) + + +def _yaml_scalar(scalar_text: str) -> str | None: + """Normalize the simple YAML scalars used by workflow mappings and lists.""" + quote_delimiter: str | None = None + escape_pending = False + comment_index: int | None = None + for character_index, text_character in enumerate(scalar_text): + if quote_delimiter == '"': + if escape_pending: + escape_pending = False + continue + if text_character == "\\": + escape_pending = True + continue + if text_character == '"': + quote_delimiter = None + continue + if quote_delimiter == "'": + if text_character == "'": + quote_delimiter = None + continue + if text_character in {"'", '"'}: + quote_delimiter = text_character + continue + if text_character == "#" and (character_index == 0 or scalar_text[character_index - 1].isspace()): + comment_index = character_index + break + normalized_scalar = scalar_text[:comment_index].strip() if comment_index is not None else scalar_text.strip() + if not normalized_scalar: + return None + if ( + len(normalized_scalar) >= 2 + and normalized_scalar[0] == normalized_scalar[-1] + and normalized_scalar[0] in {"'", '"'} + ): + if normalized_scalar[0] == '"': + return _decode_yaml_double_quoted_scalar(normalized_scalar) + return normalized_scalar[1:-1].replace("''", "'") + return normalized_scalar + + +def _mapping_value(workflow_lines: list[str], mapping_header: str, mapping_key: str) -> str | None: + """Return a scalar from a nested mapping without borrowing sibling evidence.""" + mapping_target = f"{mapping_header}:" + for line_index, workflow_line in enumerate(workflow_lines): + if workflow_line.strip() != mapping_target: + continue + header_indent = len(workflow_line) - len(workflow_line.lstrip(" ")) + for candidate_line in workflow_lines[line_index + 1 :]: + candidate_text = candidate_line.strip() + if not candidate_text or candidate_text.startswith("#"): + continue + candidate_indent = len(candidate_line) - len(candidate_line.lstrip(" ")) + if candidate_indent <= header_indent: + break + mapping_key_prefix = f"{mapping_key}:" + if candidate_text.startswith(mapping_key_prefix): + return _yaml_scalar(candidate_text[len(mapping_key_prefix) :].strip()) + return None + return None + + +def main() -> int: + """Require the Trivy workflow to cover every protected-branch PR head.""" + workflow_lines = TRIVY_WORKFLOW.read_text(encoding="utf-8").splitlines() + pull_request_block = _indented_block(workflow_lines, "pull_request", 2) + pull_request_branch_patterns = _mapping_list_sequence(pull_request_block, "branches", 4) + pull_request_activity_types = _mapping_list_values(pull_request_block, "types", 4) + jobs_block = _indented_block(workflow_lines, "jobs", 0) + trivy_job = _indented_block(jobs_block, "trivy-fs-scan", 2) + workflow_steps = _list_item_blocks(trivy_job, "steps", 4) + trivy_job_pull_request_eligible = _job_preserves_pull_request_eligibility( + jobs_block, + "trivy-fs-scan", + ) + + trivy_action_steps = [ + workflow_step + for workflow_step in workflow_steps + if (_step_action(workflow_step) or "").startswith("aquasecurity/trivy-action@") + ] + eligible_trivy_steps = [ + workflow_step + for workflow_step in trivy_action_steps + if _condition_preserves_pull_request_eligibility(_step_mapping_value(workflow_step, "if")) + ] + upload_action_steps = [ + workflow_step + for workflow_step in workflow_steps + if (_step_action(workflow_step) or "").startswith("github/codeql-action/upload-sarif@") + ] + eligible_upload_steps = [ + workflow_step + for workflow_step in upload_action_steps + if _condition_preserves_pull_request_eligibility(_step_mapping_value(workflow_step, "if")) + ] + + trivy_sarif_outputs = [ + ( + step_index, + output_path, + _mapping_value(workflow_step, "with", "exit-code") not in {None, "0"}, + ) + for step_index, workflow_step in enumerate(workflow_steps) + if workflow_step in eligible_trivy_steps + if _mapping_value(workflow_step, "with", "format") == "sarif" + if (output_path := _mapping_value(workflow_step, "with", "output")) + ] + uploaded_sarif_paths = [ + ( + step_index, + sarif_file_path, + _step_mapping_value(workflow_step, "if"), + ) + for step_index, workflow_step in enumerate(workflow_steps) + if workflow_step in eligible_upload_steps + if (sarif_file_path := _mapping_value(workflow_step, "with", "sarif_file")) + ] + ordered_matching_sarif_pair = any( + producer_path == upload_path + and producer_index < upload_index + and ( + not producer_may_fail + or _condition_runs_after_prior_failure(upload_condition) + ) + for producer_index, producer_path, producer_may_fail in trivy_sarif_outputs + for upload_index, upload_path, upload_condition in uploaded_sarif_paths + ) + + missing_contract_items: list[str] = [] + if not _has_mapping_key(workflow_lines, "pull_request", 2): + missing_contract_items.append("pull_request event") + if _has_mapping_key(workflow_lines, "pull_request_target", 2): + missing_contract_items.append("forbidden pull_request_target event") + for protected_branch in ("develop", "main"): + if not _branch_patterns_allow(pull_request_branch_patterns, protected_branch): + missing_contract_items.append(f"pull_request branch {protected_branch!r}") + if _has_mapping_key(pull_request_block, "types", 4): + for required_activity in ("opened", "synchronize", "reopened"): + if required_activity not in pull_request_activity_types: + missing_contract_items.append(f"pull_request activity {required_activity!r}") + if not trivy_job: + missing_contract_items.append("jobs.trivy-fs-scan") + elif not trivy_job_pull_request_eligible: + missing_contract_items.append("trivy-fs-scan job and needs chain eligible on pull_request") + if trivy_action_steps and not eligible_trivy_steps: + missing_contract_items.append("Trivy action step eligible on pull_request") + if upload_action_steps and not eligible_upload_steps: + missing_contract_items.append("CodeQL SARIF upload step eligible on pull_request") + if not trivy_sarif_outputs: + missing_contract_items.append("Trivy SARIF-producing action step with an output file") + if not uploaded_sarif_paths: + missing_contract_items.append("CodeQL SARIF upload step with sarif_file") + if trivy_sarif_outputs and uploaded_sarif_paths and not ordered_matching_sarif_pair: + missing_contract_items.append( + "matching ordered Trivy output and CodeQL sarif_file that uploads after findings" + ) + + if missing_contract_items: + print("Trivy PR code-scanning contract is incomplete:") + for missing_contract_item in missing_contract_items: + print(f"- missing {missing_contract_item}") + return 1 + print("Trivy PR code-scanning contract passed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) \ No newline at end of file diff --git a/scripts/harness/quickcheck.sh b/scripts/harness/quickcheck.sh index f2b87e4e8..d15ceb73b 100755 --- a/scripts/harness/quickcheck.sh +++ b/scripts/harness/quickcheck.sh @@ -9,6 +9,12 @@ python3 scripts/checks/verify_security_notes.py python3 scripts/checks/security_gates.py python3 scripts/checks/verify_supply_chain.py python3 scripts/checks/verify_github_bootstrap_policy.py +python3 scripts/checks/test_verify_trivy_pr_scan.py +python3 scripts/checks/test_verify_trivy_pr_scan_yaml_scalars.py +python3 scripts/checks/test_verify_trivy_pr_scan_branch_patterns.py +python3 scripts/checks/test_verify_trivy_pr_scan_commented_needs.py +python3 scripts/checks/test_verify_trivy_pr_scan_upload_guards.py +python3 scripts/checks/verify_trivy_pr_scan.py npm run lint npm run typecheck npm run test diff --git a/services/analysis-engine/tests/test_supply_chain_policy.py b/services/analysis-engine/tests/test_supply_chain_policy.py index ab43df89f..907265499 100644 --- a/services/analysis-engine/tests/test_supply_chain_policy.py +++ b/services/analysis-engine/tests/test_supply_chain_policy.py @@ -1235,20 +1235,40 @@ def test_supply_chain_check_accepts_repo_ossf_publish_restrictions( assert not any("ossf scorecard" in violation for violation in violations) -def test_central_governance_workflows_are_push_only_where_local_signals_remain() -> None: - """Ensure central PR governance keeps only repo-local push security signals.""" +def test_central_governance_workflows_preserve_local_security_signal_boundaries() -> None: + """Ensure local signals keep their intended push and PR trigger boundaries.""" repo_root = Path(__file__).resolve().parents[3] workflows_dir = repo_root / ".github" / "workflows" assert not (workflows_dir / "dependency-review.yml").exists() - for local_signal in ("codeql.yml", "ossf-scorecard.yml", "trivy.yml"): + for local_signal in ("codeql.yml", "ossf-scorecard.yml"): workflow = workflows_dir / local_signal assert workflow.exists(), ( f"{local_signal} keeps repository-local security-tab/SAST signal " "while central required workflows handle PR enforcement" ) - assert "pull_request:" not in workflow.read_text(encoding="utf-8") + content = workflow.read_text(encoding="utf-8") + assert "push:" in content, f"{local_signal} must retain push-based reporting" + assert "pull_request:" not in content, ( + f"{local_signal} must not become a duplicate PR gate" + ) + + trivy_workflow = workflows_dir / "trivy.yml" + assert trivy_workflow.exists(), ( + "trivy.yml keeps repository-local SARIF reporting while providing " + "the repository's per-PR vulnerability scan" + ) + trivy_content = trivy_workflow.read_text(encoding="utf-8") + assert "push:" in trivy_content, "trivy.yml must retain push-based SARIF reporting" + assert "pull_request:" in trivy_content, ( + "trivy.yml must scan pull-request heads for current-head SARIF evidence" + ) + pull_request_section = trivy_content.split("pull_request:", 1)[1].split( + "permissions:", 1 + )[0] + assert "- develop" in pull_request_section + assert "- main" in pull_request_section supply_chain = load_module( "scripts/checks/verify_supply_chain.py", "verify_supply_chain_central" @@ -1258,7 +1278,6 @@ def test_central_governance_workflows_are_push_only_where_local_signals_remain() assert ".github/workflows/codeql.yml" in required assert ".github/workflows/ossf-scorecard.yml" in required - def test_opencode_review_declares_top_level_token_permissions() -> None: """Ensure OpenCode token posture is delegated to the central required workflow.""" policy = central_required_workflow_policy_text() diff --git a/services/analysis-engine/tests/test_trivy_failure_upload_contract.py b/services/analysis-engine/tests/test_trivy_failure_upload_contract.py new file mode 100644 index 000000000..5b094d24d --- /dev/null +++ b/services/analysis-engine/tests/test_trivy_failure_upload_contract.py @@ -0,0 +1,68 @@ +"""Regression tests for Trivy SARIF publication after security findings.""" + +from __future__ import annotations + +import subprocess +import sys +import tempfile +from pathlib import Path + +_REPOSITORY_ROOT = Path(__file__).resolve().parents[3] +_TRIVY_CONTRACT_CHECKER = _REPOSITORY_ROOT / "scripts" / "checks" / "verify_trivy_pr_scan.py" + + +def _run_contract_checker(workflow_text: str) -> subprocess.CompletedProcess[str]: + """Run the repository checker against one isolated workflow fixture.""" + with tempfile.TemporaryDirectory() as temporary_directory: + temporary_root = Path(temporary_directory) + workflow_path = temporary_root / ".github" / "workflows" / "trivy.yml" + workflow_path.parent.mkdir(parents=True) + workflow_path.write_text(workflow_text, encoding="utf-8") + return subprocess.run( + [sys.executable, str(_TRIVY_CONTRACT_CHECKER)], + cwd=temporary_root, + check=False, + capture_output=True, + text=True, + ) + + +def _workflow_fixture(upload_condition: str, pull_request_comment_spacing: str = "") -> str: + """Build the minimal fail-on-findings Trivy workflow needed by the checker.""" + comment_suffix = ( + f"{pull_request_comment_spacing}# ordinary PR heads" + if pull_request_comment_spacing + else "" + ) + return f"""name: trivy +on: + push: + branches: [develop, main] + pull_request:{comment_suffix} + branches: [develop, main] +jobs: + trivy-fs-scan: + steps: + - uses: aquasecurity/trivy-action@0123456789abcdef + with: + format: sarif + output: trivy-results.sarif + exit-code: '1' + - uses: github/codeql-action/upload-sarif@fedcba9876543210 + if: {upload_condition} + with: + sarif_file: trivy-results.sarif +""" + + +def test_trivy_checker_accepts_multi_space_inline_mapping_comments() -> None: + """Valid mapping comments may have more than one separating whitespace character.""" + result = _run_contract_checker(_workflow_fixture("always()", " ")) + assert result.returncode == 0, result.stdout + result.stderr + + +def test_trivy_checker_rejects_success_only_upload_after_finding() -> None: + """A producer that exits 1 for findings must not pair with a success-only uploader.""" + result = _run_contract_checker(_workflow_fixture("success()")) + assert result.returncode == 1 + assert "uploads after findings" in result.stdout diff --git a/services/analysis-engine/tests/test_trivy_workflow_trigger_policy.py b/services/analysis-engine/tests/test_trivy_workflow_trigger_policy.py new file mode 100644 index 000000000..74cde4d42 --- /dev/null +++ b/services/analysis-engine/tests/test_trivy_workflow_trigger_policy.py @@ -0,0 +1,159 @@ +"""Structural security-policy tests for the repository-local Trivy workflow triggers.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml + +_REPOSITORY_ROOT = Path(__file__).resolve().parents[3] +_TRIVY_WORKFLOW_PATH = _REPOSITORY_ROOT / ".github" / "workflows" / "trivy.yml" +_EXPECTED_PULL_REQUEST_BRANCHES = frozenset({"develop", "main"}) + + +def _workflow_document(workflow_text: str) -> dict[str, object]: + """Return the structurally parsed GitHub Actions workflow document.""" + workflow_document = yaml.safe_load(workflow_text) + assert isinstance(workflow_document, dict), "workflow document must be a mapping" + return workflow_document + + +def _workflow_trigger_mapping(workflow_text: str) -> dict[str, object]: + """Return the structurally parsed GitHub Actions trigger mapping.""" + workflow_document = _workflow_document(workflow_text) + + workflow_triggers = workflow_document.get("on") + if workflow_triggers is None: + # PyYAML's YAML 1.1 resolver interprets the unquoted GitHub key `on` as True. + workflow_triggers = workflow_document.get(True) + assert isinstance(workflow_triggers, dict), "workflow on block must be a mapping" + return workflow_triggers + + +def _assert_trivy_pull_request_policy(workflow_text: str) -> None: + """Require ordinary PR coverage and reject privileged target-context execution.""" + workflow_triggers = _workflow_trigger_mapping(workflow_text) + assert "push" in workflow_triggers, "Trivy must retain push-based SARIF reporting" + assert "pull_request_target" not in workflow_triggers, ( + "Trivy must not execute PR-controlled code through pull_request_target" + ) + + pull_request_settings = workflow_triggers.get("pull_request") + assert isinstance(pull_request_settings, dict), ( + "Trivy must scan pull-request heads through the ordinary pull_request event" + ) + configured_branches = pull_request_settings.get("branches") + assert isinstance(configured_branches, list), "pull_request.branches must be a list" + configured_branch_names = {str(branch_name) for branch_name in configured_branches} + assert _EXPECTED_PULL_REQUEST_BRANCHES.issubset(configured_branch_names), ( + "Trivy pull_request coverage must include develop and main" + ) + + +def _assert_trivy_concurrency_policy(workflow_text: str) -> None: + """Require PR-stable cancellation so predecessor scans cannot saturate runners.""" + workflow_document = _workflow_document(workflow_text) + concurrency_settings = workflow_document.get("concurrency") + assert isinstance( + concurrency_settings, dict + ), "Trivy must declare workflow-level concurrency" + concurrency_group = concurrency_settings.get("group") + assert isinstance( + concurrency_group, str + ), "Trivy concurrency.group must be a string" + assert "github.repository" in concurrency_group, ( + "Trivy concurrency must be repository-scoped" + ) + assert "github.event.pull_request.number" in concurrency_group, ( + "Trivy PR concurrency must be stable across head-SHA changes" + ) + assert "github.sha" not in concurrency_group and "head.sha" not in concurrency_group, ( + "Trivy concurrency must not preserve stale runs by keying on the head SHA" + ) + assert concurrency_settings.get("cancel-in-progress") is True, ( + "Trivy must cancel superseded predecessor scans" + ) + + +def test_repository_trivy_workflow_uses_safe_pull_request_triggers() -> None: + """Ensure the checked-in Trivy workflow satisfies the structural trigger contract.""" + workflow_text = _TRIVY_WORKFLOW_PATH.read_text(encoding="utf-8") + _assert_trivy_pull_request_policy(workflow_text) + _assert_trivy_concurrency_policy(workflow_text) + + +@pytest.mark.parametrize( + ("fixture_name", "workflow_fixture"), + [ + ( + "target-only", + """ +name: trivy +on: + push: + branches: [develop, main] + pull_request_target: + branches: [develop, main] +""".strip(), + ), + ( + "mixed-event", + """ +name: trivy +on: + push: + branches: [develop, main] + pull_request: + branches: [develop, main] + pull_request_target: + branches: [develop, main] +""".strip(), + ), + ( + "wrong-branch", + """ +name: trivy +on: + push: + branches: [develop, main] + pull_request: + branches: [develop, release] +""".strip(), + ), + ], +) +def test_trivy_workflow_rejects_unsafe_pull_request_trigger_fixtures( + fixture_name: str, + workflow_fixture: str, +) -> None: + """Reject target-context and wrong-branch workflows by parsed YAML structure.""" + with pytest.raises(AssertionError), pytest.MonkeyPatch.context() as patch_context: + # Keep the fixture name visible in pytest failure context without changing policy behavior. + patch_context.setenv("BANDSCOPE_TRIVY_POLICY_FIXTURE", fixture_name) + _assert_trivy_pull_request_policy(workflow_fixture) + + +@pytest.mark.parametrize( + "workflow_fixture", + [ + """ +name: trivy +concurrency: + group: trivy-${{ github.repository }}-${{ github.sha }} + cancel-in-progress: true +""".strip(), + """ +name: trivy +concurrency: + group: trivy-${{ github.repository }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: false +""".strip(), + ], +) +def test_trivy_workflow_rejects_stale_run_concurrency_fixtures( + workflow_fixture: str, +) -> None: + """Reject SHA-keyed or non-cancelling concurrency that preserves obsolete runs.""" + with pytest.raises(AssertionError): + _assert_trivy_concurrency_policy(workflow_fixture)