diff --git a/.github/workflows/osv-direct-source-quality-ci.yml b/.github/workflows/osv-direct-source-quality-ci.yml new file mode 100644 index 000000000..a4dab7690 --- /dev/null +++ b/.github/workflows/osv-direct-source-quality-ci.yml @@ -0,0 +1,63 @@ +name: OSV Direct Source Quality CI + +on: + pull_request: + branches: [main] + paths: + - '.github/workflows/security-scan.yml' + - '.github/workflows/osv-direct-source-quality-ci.yml' + - 'scripts/ci/osv_direct_source_reconcile.py' + - 'tests/test_osv_direct_source_reconcile.py' + - 'docs/doctoring/osv-direct-source-provenance.md' +permissions: + contents: read + +concurrency: + group: osv-direct-source-quality-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + provenance-contract: + runs-on: ubuntu-24.04 + timeout-minutes: 15 + steps: + - name: Checkout exact source revision + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + - name: Set up Python + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: '3.14' + - name: Install exact hash-verified quality dependencies + env: + PIP_DISABLE_PIP_VERSION_CHECK: '1' + PIP_NO_INPUT: '1' + shell: bash --noprofile --norc -e -o pipefail {0} + run: | + cat >"${RUNNER_TEMP}/osv-provenance-quality-requirements.txt" <<'EOF' + coverage==7.15.2 --hash=sha256:b9a6367e4aff723e8ee8190836836124284e8fcd4265e307c844010cfa074f3f + iniconfig==2.1.0 --hash=sha256:9deba5723312380e77435581c6bf4935c94cbfab9b1ed33ef8d238ea168eb760 + packaging==26.2 --hash=sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e + pluggy==1.6.0 --hash=sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746 + pygments==2.20.0 --hash=sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176 + pytest==9.1.1 --hash=sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c + EOF + python -m pip install \ + --only-binary=:all: \ + --require-hashes \ + -r "${RUNNER_TEMP}/osv-provenance-quality-requirements.txt" + - name: Verify full central suite and provenance coverage + shell: bash --noprofile --norc -e -o pipefail {0} + run: | + test "$(git rev-parse HEAD)" = "${{ github.event.pull_request.head.sha || github.sha }}" + python -m coverage run --branch -m pytest --import-mode=importlib tests -q + python -m coverage report \ + --include='scripts/ci/osv_direct_source_reconcile.py' \ + --show-missing \ + --fail-under=100 + python -m compileall -q \ + scripts/ci/osv_direct_source_reconcile.py \ + tests/test_osv_direct_source_reconcile.py + git diff --exit-code diff --git a/.github/workflows/security-scan.yml b/.github/workflows/security-scan.yml index 435179a8d..73a2ce51e 100644 --- a/.github/workflows/security-scan.yml +++ b/.github/workflows/security-scan.yml @@ -75,6 +75,81 @@ jobs: path: source fetch-depth: 0 persist-credentials: false + - name: Install trusted OSV result evidence classifier + shell: bash + run: | + set -euo pipefail + cat >"$RUNNER_TEMP/classify-osv-result.py" <<'PY' + import json + import os + from pathlib import Path + + result = Path(os.environ["RESULT_FILE"]) + outcome = os.environ.get("SCAN_OUTCOME", "") + complete = False + message = "" + try: + if result.is_symlink() or (result.exists() and not result.is_file()): + raise ValueError("result document must be a regular file") + if not result.exists() or result.stat().st_size == 0: + if outcome != "success": + raise ValueError("failed without an OSV result document") + result.write_text('{"results":[]}\n', encoding="utf-8") + complete = True + message = "completed successfully without findings output" + else: + document = json.loads(result.read_text(encoding="utf-8")) + if not isinstance(document, dict): + raise ValueError("result document must be an object") + results = document.get("results") + if not isinstance(results, list): + raise ValueError("results must be a list") + finding_count = 0 + for scan_result in results: + if not isinstance(scan_result, dict): + raise ValueError("each result must be an object") + packages = scan_result.get("packages", []) + if not isinstance(packages, list): + raise ValueError("packages must be a list") + for package in packages: + if not isinstance(package, dict): + raise ValueError("each package must be an object") + vulnerabilities = package.get("vulnerabilities", []) + if not isinstance(vulnerabilities, list): + raise ValueError("vulnerabilities must be a list") + if not all(isinstance(item, dict) for item in vulnerabilities): + raise ValueError("each vulnerability must be an object") + finding_count += len(vulnerabilities) + if outcome == "success": + complete = True + message = f"completed successfully with {finding_count} finding(s)" + elif outcome == "failure" and finding_count > 0: + complete = True + message = ( + "failure with authoritative vulnerability evidence " + f"({finding_count} finding(s)) is a completed scan, not infrastructure failure" + ) + else: + raise ValueError("failed without authoritative vulnerability evidence") + except (OSError, UnicodeError, ValueError, json.JSONDecodeError) as error: + message = str(error) + + with Path(os.environ["GITHUB_OUTPUT"]).open("a", encoding="utf-8") as output: + output.write(f"complete={'true' if complete else 'false'}\n") + level = "notice" if complete else "warning" + print(f"::{level}::OSV result evidence: {message}") + PY + chmod 500 "$RUNNER_TEMP/classify-osv-result.py" + - name: Prepare base OSV output boundary + shell: python3 {0} + run: | + from pathlib import Path + + result = Path("old-results.json") + if result.exists() or result.is_symlink(): + if result.is_dir() and not result.is_symlink(): + raise SystemExit("old-results.json is a directory; refusing an untrusted scan output boundary") + result.unlink() - name: Scan base with OSV id: osv_base continue-on-error: true @@ -83,29 +158,81 @@ jobs: with: scan-args: | --format=json - --output=old-results.json + --output-file=old-results.json --maven-registry=https://maven-central.storage-download.googleapis.com/maven2 --no-resolve --allow-no-lockfiles -r source/ + - name: Classify base OSV result evidence + id: osv_base_evidence + if: always() + shell: python3 {0} + env: + RESULT_FILE: old-results.json + SCAN_OUTCOME: ${{ steps.osv_base.outcome }} + run: | + import os + import runpy + + runpy.run_path(os.path.join(os.environ["RUNNER_TEMP"], "classify-osv-result.py")) - name: Explain base OSV resolver fallback - if: steps.osv_base.outcome == 'failure' + if: steps.osv_base_evidence.outputs.complete != 'true' run: | - echo "::warning::OSV base scan failed or timed out before reporter output was trusted; retrying the --no-resolve direct manifest/lockfile scan. Direct manifest and lockfile vulnerability evidence remains enforced while external transitive registry resolution is intentionally avoided." + echo "::warning::OSV base scan did not produce authoritative result evidence; retrying the --no-resolve direct manifest/lockfile scan. Direct manifest and lockfile vulnerability evidence remains enforced while external transitive registry resolution is intentionally avoided." + - name: Clear incomplete base OSV output before retry + if: steps.osv_base_evidence.outputs.complete != 'true' + shell: python3 {0} + run: | + from pathlib import Path + + result = Path("old-results.json") + if result.exists() or result.is_symlink(): + if result.is_dir() and not result.is_symlink(): + raise SystemExit("old-results.json is a directory; refusing an untrusted retry boundary") + result.unlink() - name: Retry base OSV without transitive resolution - if: steps.osv_base.outcome == 'failure' + if: steps.osv_base_evidence.outputs.complete != 'true' + id: osv_base_retry continue-on-error: true timeout-minutes: 4 uses: google/osv-scanner-action/osv-scanner-action@a82132c0bd6c7261ffcb78e754c46c70ab57ad9a # v2.3.8 with: scan-args: | --format=json - --output=old-results.json + --output-file=old-results.json --no-resolve --allow-no-lockfiles -r source/ + - name: Classify retried base OSV result evidence + id: osv_base_retry_evidence + if: steps.osv_base_evidence.outputs.complete != 'true' + shell: python3 {0} + env: + RESULT_FILE: old-results.json + SCAN_OUTCOME: ${{ steps.osv_base_retry.outcome }} + run: | + import os + import runpy + + runpy.run_path(os.path.join(os.environ["RUNNER_TEMP"], "classify-osv-result.py")) + - name: Preserve base OSV evidence and direct-source provenance + run: | + set -euo pipefail + mkdir -p "$RUNNER_TEMP/osv-base-provenance" + if [ -L old-results.json ] || [ ! -f old-results.json ]; then + echo "::error::Base OSV result document is not an authoritative regular file." + exit 1 + fi + cp -- old-results.json "$RUNNER_TEMP/osv-base-provenance/old-results.json" + if [ -L source/pnpm-lock.yaml ]; then + echo "::error::Base pnpm-lock.yaml is a symlink; direct-source provenance is not authoritative." + exit 1 + fi + if [ -f source/pnpm-lock.yaml ]; then + cp -- source/pnpm-lock.yaml "$RUNNER_TEMP/osv-base-provenance/pnpm-lock.yaml" + fi - name: Checkout head uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 with: @@ -114,6 +241,16 @@ jobs: path: source fetch-depth: 0 persist-credentials: false + - name: Prepare head OSV output boundary + shell: python3 {0} + run: | + from pathlib import Path + + result = Path("new-results.json") + if result.exists() or result.is_symlink(): + if result.is_dir() and not result.is_symlink(): + raise SystemExit("new-results.json is a directory; refusing an untrusted scan output boundary") + result.unlink() - name: Scan head with OSV id: osv_head continue-on-error: true @@ -122,29 +259,133 @@ jobs: with: scan-args: | --format=json - --output=new-results.json + --output-file=new-results.json --maven-registry=https://maven-central.storage-download.googleapis.com/maven2 --no-resolve --allow-no-lockfiles -r source/ + - name: Classify head OSV result evidence + id: osv_head_evidence + if: always() + shell: python3 {0} + env: + RESULT_FILE: new-results.json + SCAN_OUTCOME: ${{ steps.osv_head.outcome }} + run: | + import os + import runpy + + runpy.run_path(os.path.join(os.environ["RUNNER_TEMP"], "classify-osv-result.py")) - name: Explain head OSV resolver fallback - if: steps.osv_head.outcome == 'failure' + if: steps.osv_head_evidence.outputs.complete != 'true' run: | - echo "::warning::OSV head scan failed or timed out before reporter output was trusted; retrying the --no-resolve direct manifest/lockfile scan. Direct manifest and lockfile vulnerability evidence remains enforced while external transitive registry resolution is intentionally avoided." + echo "::warning::OSV head scan did not produce authoritative result evidence; retrying the --no-resolve direct manifest/lockfile scan. Direct manifest and lockfile vulnerability evidence remains enforced while external transitive registry resolution is intentionally avoided." + - name: Clear incomplete head OSV output before retry + if: steps.osv_head_evidence.outputs.complete != 'true' + shell: python3 {0} + run: | + from pathlib import Path + + result = Path("new-results.json") + if result.exists() or result.is_symlink(): + if result.is_dir() and not result.is_symlink(): + raise SystemExit("new-results.json is a directory; refusing an untrusted retry boundary") + result.unlink() - name: Retry head OSV without transitive resolution - if: steps.osv_head.outcome == 'failure' + if: steps.osv_head_evidence.outputs.complete != 'true' + id: osv_head_retry continue-on-error: true timeout-minutes: 4 uses: google/osv-scanner-action/osv-scanner-action@a82132c0bd6c7261ffcb78e754c46c70ab57ad9a # v2.3.8 with: scan-args: | --format=json - --output=new-results.json + --output-file=new-results.json --no-resolve --allow-no-lockfiles -r source/ + - name: Classify retried head OSV result evidence + id: osv_head_retry_evidence + if: steps.osv_head_evidence.outputs.complete != 'true' + shell: python3 {0} + env: + RESULT_FILE: new-results.json + SCAN_OUTCOME: ${{ steps.osv_head_retry.outcome }} + run: | + import os + import runpy + + runpy.run_path(os.path.join(os.environ["RUNNER_TEMP"], "classify-osv-result.py")) + - name: Restore authoritative base OSV evidence + shell: python3 {0} + env: + RUNNER_TEMP: ${{ runner.temp }} + run: | + import os + import shutil + from pathlib import Path + + preserved = Path(os.environ["RUNNER_TEMP"]) / "osv-base-provenance" / "old-results.json" + destination = Path("old-results.json") + if preserved.is_symlink() or not preserved.is_file(): + raise SystemExit("preserved base OSV result document is not an authoritative regular file") + if destination.exists() or destination.is_symlink(): + if destination.is_dir() and not destination.is_symlink(): + raise SystemExit("old-results.json is a directory; refusing an untrusted restore boundary") + destination.unlink() + shutil.copyfile(preserved, destination) + - name: Require authoritative base and head OSV evidence + run: | + set -euo pipefail + base_complete="${{ steps.osv_base_evidence.outputs.complete == 'true' || steps.osv_base_retry_evidence.outputs.complete == 'true' }}" + head_complete="${{ steps.osv_head_evidence.outputs.complete == 'true' || steps.osv_head_retry_evidence.outputs.complete == 'true' }}" + if [ "$base_complete" != true ] || [ "$head_complete" != true ]; then + echo "::error::OSV base/head scan did not produce authoritative complete result evidence." + exit 1 + fi + - name: Checkout exact central provenance policy + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + repository: ContextualWisdomLab/.github + ref: ${{ github.event.pull_request.base.repo.full_name == 'ContextualWisdomLab/.github' && github.event.pull_request.base.sha || github.workflow_sha }} + path: .cwl-trusted-security-policy + persist-credentials: false + - name: Reconcile immutable direct-source provenance + run: | + set -euo pipefail + reconciler=".cwl-trusted-security-policy/scripts/ci/osv_direct_source_reconcile.py" + audit="osv-provenance-audit.json" + if [ -L "$reconciler" ]; then + echo "::error::Trusted OSV provenance policy is a symlink; refusing an untrusted policy boundary." + exit 1 + elif [ ! -f "$reconciler" ]; then + echo "::notice::trusted provenance policy is not yet present; retaining raw OSV evidence without reconciliation." + else + base_lock="$RUNNER_TEMP/osv-base-provenance/pnpm-lock.yaml" + if [ -f "$base_lock" ]; then + python3 "$reconciler" \ + --results old-results.json \ + --lockfile "$base_lock" \ + --source-path source/pnpm-lock.yaml \ + --audit "$audit" \ + --label base + fi + + if [ -L source/pnpm-lock.yaml ]; then + echo "::error::Head pnpm-lock.yaml is a symlink; direct-source provenance is not authoritative." + exit 1 + fi + if [ -f source/pnpm-lock.yaml ]; then + python3 "$reconciler" \ + --results new-results.json \ + --lockfile source/pnpm-lock.yaml \ + --source-path source/pnpm-lock.yaml \ + --audit "$audit" \ + --label head + fi + fi - name: Require OSV scan output run: | set -euo pipefail @@ -193,7 +434,7 @@ jobs: uses: google/osv-scanner-action/osv-reporter-action@8dc09193bb540e09b23da07ad7e30bd33bf87018 # v2.3.8 with: scan-args: | - --output=results.sarif + --output-files=sarif:results.sarif --old=old-results.json --new=new-results.json --gh-annotations=true @@ -249,6 +490,7 @@ jobs: old-results.json new-results.json results.sarif + osv-provenance-audit.json if-no-files-found: ignore retention-days: 5 diff --git a/CHANGELOG.md b/CHANGELOG.md index 8639823d5..d580ed7c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -138,6 +138,9 @@ Semantic Versioning where the repository publishes a release. ### Fixed +- Normalized a successful OSV scan with no findings output to a valid empty + result document after both base and head scan outcomes are verified, while + keeping failed scans, failed retries, and symlinked result paths fail-closed. - Resolve Strix visibility from the trusted GitHub event for ordinary push, schedule, and pull-request runs, reserving API retries for cross-repository dispatches whose workflow token may not see the target repository. diff --git a/docs/doctoring/osv-direct-source-provenance.md b/docs/doctoring/osv-direct-source-provenance.md new file mode 100644 index 000000000..97bd5d2d4 --- /dev/null +++ b/docs/doctoring/osv-direct-source-provenance.md @@ -0,0 +1,57 @@ +# OSV direct-source provenance reconciliation + +## Decision + +The central Security Scan preserves OSV Scanner as the hard base/head +vulnerability gate. It adds a narrow evidence reconciliation step for pnpm +dependencies whose lock entry proves an exact immutable non-registry source. +The first governed source is the official SheetJS CE HTTPS release artifact. + +A registry advisory is removed from the reporter input only when all of these +facts agree: + +1. package name and strict three-component version; +2. the pnpm package key and `resolution.tarball` canonical HTTPS URL; +3. the official `cdn.sheetjs.com` origin and versioned release path; +4. one valid SHA-512 integrity receipt; and +5. an advisory-provided, machine-checkable affected upper bound (`<` or `<=`) + that the exact artifact version is outside. Equality at a `<=` boundary + remains affected. + +An affected version remains a finding. Missing integrity, an unknown host, +version disagreement, malformed JSON or an absent/ambiguous affected range is +retained and recorded as `SCANNER_METADATA_CONFLICT`. No advisory identifier, +package name, or severity is blanket-ignored. The audit artifact records every +retained or reconciled direct-source decision without copying untrusted +advisory prose. + +## Empty result documents + +The pinned OSV action can complete successfully without creating its requested +JSON file when a repository has no findings or no supported lockfile. For each +base or head scan independently, the central workflow first requires that +scan's outcome to be success before writing the valid empty document +`{"results":[]}` for its missing or empty result file. A failed first scan and failed retry never enter this path, and +symlinked result paths remain a hard failure. This preserves the distinction +between a verified clean scan and an unavailable scan without weakening the +base/head reporter gate. + +## Rollback and operations + +Rollback removes the reconciliation step and helper together, restoring raw +OSV reporter inputs. Operators should inspect `osv-provenance-audit.json` +alongside `old-results.json`, `new-results.json`, and `results.sarif`. A metadata +conflict is non-passing when the underlying OSV finding is new because the +finding remains in the reporter input. + +## References + +Google. (2026). *OSV-Scanner documentation*. Open Source Vulnerabilities. +https://google.github.io/osv-scanner/ + +OpenSSF. (2025). *Open source vulnerability format specification*. Open Source +Security Foundation. https://ossf.github.io/osv-schema/ + +pnpm. (2026). *Settings: Lockfile*. https://pnpm.io/settings#lockfile + +SheetJS LLC. (2026). *SheetJS Community Edition*. https://docs.sheetjs.com/ diff --git a/scripts/ci/osv_direct_source_reconcile.py b/scripts/ci/osv_direct_source_reconcile.py new file mode 100644 index 000000000..24cf0e76f --- /dev/null +++ b/scripts/ci/osv_direct_source_reconcile.py @@ -0,0 +1,599 @@ +"""Reconcile OSV registry findings with exact immutable direct-source evidence.""" + +from __future__ import annotations + +import argparse +import base64 +import binascii +import json +import os +import re +import stat +import sys +import tempfile +from collections.abc import Iterable +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import Any +from urllib.parse import urlsplit + +SEMVER_RE = re.compile(r"^(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)$") +DIRECT_HEADER_RE = re.compile( + r"(?m)^ (?P(?:@[^/\s]+/)?[^@\s]+)@(?Phttps://[^\s]+):[ \t]*$" +) +PACKAGES_SECTION_RE = re.compile(r"(?m)^packages:[ \t]*\n") +TOP_LEVEL_SECTION_RE = re.compile(r"(?m)^[^ \t\r\n#][^:\r\n]*:[^\r\n]*$") +ENTRY_BOUNDARY_RE = re.compile(r"(?m)^ \S") +VERSION_LINE_RE = re.compile(r"(?m)^ version:[ \t]*['\"]?([^'\"\s]+)['\"]?[ \t]*$") +TARBALL_RE = re.compile(r"(?:^|[, {])[ \t]*tarball:[ \t]*([^, }]+)") +INTEGRITY_RE = re.compile(r"(?:^|[, {])[ \t]*integrity:[ \t]*(sha512-[A-Za-z0-9+/=]+)") +AFFECTED_RANGE_RE = re.compile( + r"^[ \t]*(?P<=|<)[ \t]*(?P\d+\.\d+\.\d+)[ \t]*$" +) +SHEETJS_EXCEPTION_VERSION = "0.20.3" +SHEETJS_EXCEPTION_INTEGRITY = ( + "sha512-oLDq3jw7AcLqKWH2AhCpVTZl8mf6X2YReP+Neh0SJUzV/" + "BdZYjth94tG5toiMB1PPrYtxOCfaoUCkvtuH+3AJA==" +) +SHEETJS_URL_RE = re.compile( + r"^/xlsx-(?P\d+\.\d+\.\d+)/xlsx-(?P=version)\.tgz$" +) + + +@dataclass(frozen=True) +class DirectSource: + """Evidence parsed from one pnpm direct-tarball package record.""" + + package_name: str + version: str + source_url: str + integrity: str + valid: bool + reason: str + + +def parse_semver(value: str) -> tuple[int, int, int] | None: + """Parse a strict three-component SemVer core.""" + + match = SEMVER_RE.fullmatch(value) + if not match: + return None + return tuple(int(part) for part in match.groups()) # type: ignore[return-value] + + +def valid_sha512_integrity(value: str) -> bool: + """Return whether an integrity string contains one exact SHA-512 digest.""" + + if not value.startswith("sha512-"): + return False + try: + decoded = base64.b64decode(value.removeprefix("sha512-"), validate=True) + except (ValueError, binascii.Error): + return False + return len(decoded) == 64 + + +def validate_sheetjs_source( + package_name: str, + header_url: str, + tarball_url: str, + version: str, + integrity: str, +) -> tuple[bool, str]: + """Validate the exact official immutable SheetJS release identity.""" + + if package_name != "xlsx": + return False, "package is not governed by the SheetJS direct-source contract" + try: + parsed = urlsplit(header_url) + parsed_port = parsed.port + except ValueError: + return False, "direct source URL is malformed" + if ( + parsed.scheme != "https" + or parsed.hostname != "cdn.sheetjs.com" + or parsed_port is not None + or parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + ): + return False, "direct source is not the canonical SheetJS HTTPS origin" + path_match = SHEETJS_URL_RE.fullmatch(parsed.path) + if not path_match or path_match.group("version") != version: + return False, "package version does not match the immutable SheetJS URL" + if header_url != tarball_url: + return False, "pnpm package key and resolution tarball disagree" + if not valid_sha512_integrity(integrity): + return False, "direct source lacks one valid SHA-512 integrity receipt" + if integrity != SHEETJS_EXCEPTION_INTEGRITY: + return False, "direct source integrity does not match the governed SheetJS artifact" + return True, "exact official immutable SheetJS release" + + +def parse_direct_sources(lock_text: str) -> list[DirectSource]: + """Parse direct HTTPS package records from pnpm's ``packages`` map. + + pnpm v9 repeats direct-tarball keys in ``snapshots``. Snapshot entries + describe dependency edges and do not carry resolution provenance, so + parsing only the ``packages`` map prevents a false duplicate-source + conflict while retaining conflicting package metadata as a finding. + """ + + sources: list[DirectSource] = [] + for packages_section in PACKAGES_SECTION_RE.finditer(lock_text): + section_start = packages_section.end() + next_section = TOP_LEVEL_SECTION_RE.search(lock_text, section_start) + section_end = next_section.start() if next_section else len(lock_text) + package_text = lock_text[section_start:section_end] + matches = list(DIRECT_HEADER_RE.finditer(package_text)) + for match in matches: + start = match.end() + boundary = ENTRY_BOUNDARY_RE.search(package_text, start) + end = boundary.start() if boundary else len(package_text) + block = package_text[start:end] + version_match = VERSION_LINE_RE.search(block) + tarball_match = TARBALL_RE.search(block) + integrity_match = INTEGRITY_RE.search(block) + version = version_match.group(1) if version_match else "" + tarball = tarball_match.group(1) if tarball_match else "" + integrity = integrity_match.group(1) if integrity_match else "" + valid, reason = validate_sheetjs_source( + match.group("name"), match.group("url"), tarball, version, integrity + ) + sources.append( + DirectSource( + package_name=match.group("name"), + version=version, + source_url=match.group("url"), + integrity=integrity, + valid=valid, + reason=reason, + ) + ) + return sources + + +def iter_result_packages( + payload: dict[str, Any], +) -> Iterable[tuple[str | None, dict[str, Any]]]: + """Yield each scanner source path with its package findings.""" + + results = payload.get("results") + if not isinstance(results, list): + raise TypeError("OSV results must contain a results array") + for result in results: + if not isinstance(result, dict): + raise TypeError("OSV result entries must be objects") + source = result.get("source") + observed_source_path = ( + source.get("path") + if isinstance(source, dict) and isinstance(source.get("path"), str) + else None + ) + packages = result.get("packages") or [] + if not isinstance(packages, list): + raise TypeError("OSV result packages must be an array") + for package in packages: + if not isinstance(package, dict): + raise TypeError("OSV package entries must be objects") + yield observed_source_path, package + + +def iter_packages(payload: dict[str, Any]) -> Iterable[dict[str, Any]]: + """Yield package findings from an OSV Scanner result document.""" + + for _, package in iter_result_packages(payload): + yield package + + +def validate_source_path(source_path: str) -> str: + """Return one normalized repository-relative governed lockfile path.""" + + candidate = PurePosixPath(source_path) + if ( + not source_path + or candidate.is_absolute() + or "\\" in source_path + or any(part in {"", ".", ".."} for part in candidate.parts) + ): + raise ValueError("governed OSV source path must be one normalized relative path") + return candidate.as_posix() + + +def source_matches_governed_lockfile( + observed_source_path: str | None, governed_source_path: str +) -> bool: + """Bind a scanner result to the exact governed workspace lockfile.""" + + governed = validate_source_path(governed_source_path) + return observed_source_path in { + governed, + f"/github/workspace/{governed}", + } + + +def authoritative_affected_range( + vulnerability: dict[str, Any], package_name: str +) -> str | None: + """Return one GitHub-reviewed npm affected bound with live OSV shape.""" + + vulnerability_id = vulnerability.get("id") + affected = vulnerability.get("affected") + if not isinstance(vulnerability_id, str) or not isinstance(affected, list): + return None + candidates: list[str] = [] + for item in affected: + if not isinstance(item, dict): + continue + package = item.get("package") + database_specific = item.get("database_specific") + ranges = item.get("ranges") + if ( + not isinstance(package, dict) + or package.get("ecosystem") != "npm" + or package.get("name") != package_name + or not isinstance(database_specific, dict) + or not isinstance(ranges, list) + ): + continue + affected_range = database_specific.get("last_known_affected_version_range") + source = database_specific.get("source") + expected_source_fragment = f"/{vulnerability_id}/{vulnerability_id}.json" + if ( + not isinstance(affected_range, str) + or not isinstance(source, str) + or not source.startswith( + "https://github.com/github/advisory-database/blob/" + ) + or not source.endswith(expected_source_fragment) + ): + continue + semver_ranges = [ + range_item + for range_item in ranges + if isinstance(range_item, dict) and range_item.get("type") == "SEMVER" + ] + if len(semver_ranges) != 1: + continue + events = semver_ranges[0].get("events") + if events != [{"introduced": "0"}]: + continue + candidates.append(affected_range) + return candidates[0] if len(set(candidates)) == 1 and candidates else None + + +def audit_entry( + *, + label: str, + source: DirectSource, + package_name: str, + package_version: str, + vulnerability: dict[str, Any], + affected_range: str | None, + status: str, + reason: str, +) -> dict[str, str]: + """Build one stable audit record without copying advisory prose.""" + + return { + "scan": label, + "status": status, + "reason": reason, + "package": package_name, + "version": package_version, + "vulnerability_id": str(vulnerability.get("id") or "unknown"), + "affected_range": affected_range or "unknown", + "source_url": source.source_url, + "integrity": source.integrity or "missing", + } + + +def reconcile_payload( + payload: dict[str, Any], + lock_text: str, + *, + label: str, + source_path: str = "pnpm-lock.yaml", +) -> tuple[dict[str, Any], list[dict[str, str]]]: + """Remove only findings disproven by exact source and affected-range evidence.""" + + direct_sources = parse_direct_sources(lock_text) + audit: list[dict[str, str]] = [] + governed_source_path = validate_source_path(source_path) + for observed_source_path, package in iter_result_packages(payload): + package_info = package.get("package") + vulnerabilities = package.get("vulnerabilities") or [] + if not isinstance(package_info, dict) or not isinstance(vulnerabilities, list): + raise TypeError("OSV package evidence is malformed") + package_name = str(package_info.get("name") or "") + package_version = str(package_info.get("version") or "") + if not source_matches_governed_lockfile( + observed_source_path, governed_source_path + ): + if package_name != "xlsx": + continue + source = DirectSource( + package_name=package_name, + version=package_version, + source_url="", + integrity="", + valid=False, + reason="OSV finding source does not match governed lockfile", + ) + for vulnerability in vulnerabilities: + if not isinstance(vulnerability, dict): + raise TypeError("OSV vulnerability entries must be objects") + audit.append( + audit_entry( + label=label, + source=source, + package_name=package_name, + package_version=package_version, + vulnerability=vulnerability, + affected_range=authoritative_affected_range( + vulnerability, package_name + ), + status="SCANNER_METADATA_CONFLICT", + reason=source.reason, + ) + ) + continue + candidates = [ + source + for source in direct_sources + if source.package_name == package_name + and (source.version == package_version or not source.version) + ] + if not candidates and package_name != "xlsx": + continue + if len(candidates) != 1: + reason = ( + "no unique direct-source provenance matches package and version" + if not candidates + else "multiple direct-source records match package and version" + ) + source = candidates[0] if candidates else DirectSource( + package_name=package_name, + version=package_version, + source_url="", + integrity="", + valid=False, + reason=reason, + ) + retained = [] + for vulnerability in vulnerabilities: + if not isinstance(vulnerability, dict): + raise TypeError("OSV vulnerability entries must be objects") + retained.append(vulnerability) + audit.append( + audit_entry( + label=label, + source=source, + package_name=package_name, + package_version=package_version, + vulnerability=vulnerability, + affected_range=authoritative_affected_range( + vulnerability, package_name + ), + status="SCANNER_METADATA_CONFLICT", + reason=reason, + ) + ) + package["vulnerabilities"] = retained + continue + source = candidates[0] + retained: list[dict[str, Any]] = [] + for vulnerability in vulnerabilities: + if not isinstance(vulnerability, dict): + raise TypeError("OSV vulnerability entries must be objects") + affected_range = authoritative_affected_range( + vulnerability, package_name + ) + if not source.valid: + retained.append(vulnerability) + audit.append( + audit_entry( + label=label, + source=source, + package_name=package_name, + package_version=package_version, + vulnerability=vulnerability, + affected_range=affected_range, + status="SCANNER_METADATA_CONFLICT", + reason=source.reason, + ) + ) + continue + range_match = AFFECTED_RANGE_RE.fullmatch(affected_range or "") + version_key = parse_semver(package_version) + upper_key = ( + parse_semver(range_match.group("version")) if range_match else None + ) + if version_key is None or upper_key is None: + retained.append(vulnerability) + audit.append( + audit_entry( + label=label, + source=source, + package_name=package_name, + package_version=package_version, + vulnerability=vulnerability, + affected_range=affected_range, + status="SCANNER_METADATA_CONFLICT", + reason="advisory lacks one machine-checkable affected upper bound", + ) + ) + continue + inside_affected_range = ( + version_key <= upper_key + if range_match.group("operator") == "<=" + else version_key < upper_key + ) + if package_version != SHEETJS_EXCEPTION_VERSION: + retained.append(vulnerability) + if inside_affected_range: + status = "AFFECTED" + reason = "exact direct-source version remains inside the affected range" + else: + status = "SCANNER_METADATA_CONFLICT" + reason = ( + "direct-source reconciliation is limited to immutable " + "SheetJS xlsx@0.20.3" + ) + audit.append( + audit_entry( + label=label, + source=source, + package_name=package_name, + package_version=package_version, + vulnerability=vulnerability, + affected_range=affected_range, + status=status, + reason=reason, + ) + ) + continue + if inside_affected_range: + retained.append(vulnerability) + status = "AFFECTED" + reason = "exact direct-source version remains inside the affected range" + else: + status = "RECONCILED" + reason = "exact immutable direct-source version is outside the affected range" + audit.append( + audit_entry( + label=label, + source=source, + package_name=package_name, + package_version=package_version, + vulnerability=vulnerability, + affected_range=affected_range, + status=status, + reason=reason, + ) + ) + package["vulnerabilities"] = retained + return payload, audit + + +def atomic_json_write(path: Path, value: object) -> None: + """Write JSON through a same-directory temporary regular file.""" + + path.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + temporary = Path(temporary_name) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + json.dump(value, handle, indent=2, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + finally: + temporary.unlink(missing_ok=True) + + +def load_json_object(path: Path) -> dict[str, Any]: + """Load one JSON object from a regular non-symlink file.""" + + value = json.loads(read_utf8_text(path, "required JSON input")) + if not isinstance(value, dict): + raise TypeError(f"required JSON input is not an object: {path}") + return value + + +def _read_regular_utf8_text(path: Path, description: str) -> str: + """Read one regular file through a descriptor that cannot follow symlinks. + + Lockfiles and scanner receipts are attacker-controlled repository inputs. + Opening with ``O_NOFOLLOW`` and checking the descriptor's type closes the + check/use gap between path validation and the actual read. + """ + + no_follow = getattr(os, "O_NOFOLLOW", None) + if no_follow is None: + raise ValueError("secure regular-file reads require O_NOFOLLOW support") + descriptor: int | None = None + try: + descriptor = os.open(os.fspath(path), os.O_RDONLY | no_follow) + if not stat.S_ISREG(os.fstat(descriptor).st_mode): + raise ValueError(f"{description} is not a regular file: {path}") + with os.fdopen(descriptor, "r", encoding="utf-8") as handle: + descriptor = None + return handle.read() + except FileNotFoundError as error: + raise ValueError(f"{description} is not a regular file: {path}") from error + except UnicodeDecodeError as error: + raise ValueError(f"{description} is not valid UTF-8: {path}") from error + except OSError as error: + raise ValueError(f"{description} is not a regular file: {path}") from error + finally: + if descriptor is not None: + os.close(descriptor) + + +def read_utf8_text(path: Path, description: str) -> str: + """Read trusted text input and turn malformed UTF-8 into an explicit error.""" + + return _read_regular_utf8_text(path, description) + + +def read_optional_utf8_text(path: Path, description: str) -> str | None: + """Read an existing regular file, returning ``None`` only when absent.""" + + try: + return read_utf8_text(path, description) + except ValueError as error: + if isinstance(error.__cause__, FileNotFoundError): + return None + raise + + +def parse_args() -> argparse.Namespace: + """Parse command-line arguments.""" + + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--results", required=True, type=Path) + parser.add_argument("--lockfile", required=True, type=Path) + parser.add_argument("--audit", required=True, type=Path) + parser.add_argument("--label", default="scan") + parser.add_argument("--source-path", default="pnpm-lock.yaml") + return parser.parse_args() + + +def main() -> int: + """Reconcile one OSV document and publish append-only audit evidence.""" + + try: + args = parse_args() + payload = load_json_object(args.results) + reconciled, new_audit = reconcile_payload( + payload, + read_utf8_text(args.lockfile, "pnpm lock provenance input"), + label=args.label, + source_path=args.source_path, + ) + existing_audit: list[dict[str, str]] = [] + audit_text = read_optional_utf8_text(args.audit, "audit input") + if audit_text is not None: + loaded_audit = json.loads(audit_text) + if not isinstance(loaded_audit, list): + raise ValueError("audit output must contain an array") + existing_audit = loaded_audit + atomic_json_write(args.results, reconciled) + atomic_json_write(args.audit, [*existing_audit, *new_audit]) + for entry in new_audit: + print( + "OSV provenance " + f"{entry['status']}: {entry['package']}@{entry['version']} " + f"{entry['vulnerability_id']} ({entry['reason']})" + ) + return 0 + except (OSError, TypeError, ValueError) as error: + print(f"::error::{error}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_osv_direct_source_reconcile.py b/tests/test_osv_direct_source_reconcile.py new file mode 100644 index 000000000..0ccd9ee11 --- /dev/null +++ b/tests/test_osv_direct_source_reconcile.py @@ -0,0 +1,647 @@ +"""Regression tests for provenance-aware OSV direct-source reconciliation.""" + +from __future__ import annotations + +import copy +import importlib.util +import io +import json +import runpy +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +REPO_ROOT = Path(__file__).resolve().parents[1] +SCRIPT = REPO_ROOT / "scripts" / "ci" / "osv_direct_source_reconcile.py" +SECURITY_WORKFLOW = REPO_ROOT / ".github" / "workflows" / "security-scan.yml" +OFFICIAL_URL = "https://cdn.sheetjs.com/xlsx-{version}/xlsx-{version}.tgz" +INTEGRITY = "sha512-oLDq3jw7AcLqKWH2AhCpVTZl8mf6X2YReP+Neh0SJUzV/BdZYjth94tG5toiMB1PPrYtxOCfaoUCkvtuH+3AJA==" + +SPEC = importlib.util.spec_from_file_location("osv_direct_source_reconcile", SCRIPT) +assert SPEC is not None and SPEC.loader is not None +OSV = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = OSV +SPEC.loader.exec_module(OSV) + + +def vulnerability(vuln_id: str, affected_range: str | None) -> dict[str, object]: + """Return a minimal OSV vulnerability record.""" + + record: dict[str, object] = {"id": vuln_id, "aliases": []} + if affected_range is not None: + record["affected"] = [ + { + "database_specific": { + "last_known_affected_version_range": affected_range, + "source": ( + "https://github.com/github/advisory-database/blob/main/" + f"advisories/github-reviewed/2026/08/{vuln_id}/{vuln_id}.json" + ), + }, + "package": {"ecosystem": "npm", "name": "xlsx"}, + "ranges": [ + { + "events": [{"introduced": "0"}], + "type": "SEMVER", + } + ], + } + ] + return record + + +def results(version: str, vulnerabilities: list[dict[str, object]]) -> dict[str, object]: + """Return a minimal OSV Scanner JSON document.""" + + return { + "results": [ + { + "source": {"path": "pnpm-lock.yaml", "type": "lockfile"}, + "packages": [ + { + "package": { + "name": "xlsx", + "version": version, + "ecosystem": "npm", + }, + "vulnerabilities": vulnerabilities, + } + ], + } + ] + } + + +def direct_lock(version: str, *, integrity: str | None = INTEGRITY, host: str = "cdn.sheetjs.com") -> str: + """Return the exact pnpm v9 direct-tarball evidence shape used by Inkspan.""" + + url = OFFICIAL_URL.format(version=version).replace("cdn.sheetjs.com", host) + resolution_parts = [] + if integrity is not None: + resolution_parts.append(f"integrity: {integrity}") + resolution_parts.append(f"tarball: {url}") + resolution = ", ".join(resolution_parts) + return ( + "lockfileVersion: '9.0'\n\n" + "packages:\n\n" + f" xlsx@{url}:\n" + f" resolution: {{{resolution}}}\n" + f" version: {version}\n" + ) + + +def registry_lock(version: str) -> str: + """Return an ordinary registry-backed pnpm package record.""" + + return ( + "lockfileVersion: '9.0'\n\n" + "packages:\n\n" + f" xlsx@{version}:\n" + f" resolution: {{integrity: {INTEGRITY}}}\n" + ) + + +def reconcile(tmp_path: Path, payload: dict[str, object], lock_text: str) -> tuple[dict[str, object], list[dict[str, object]]]: + """Run the production reconciler and return its rewritten result and audit.""" + + assert SCRIPT.is_file(), "production OSV provenance reconciler is missing" + result_path = tmp_path / "results.json" + lock_path = tmp_path / "pnpm-lock.yaml" + audit_path = tmp_path / "audit.json" + result_path.write_text(json.dumps(payload), encoding="utf-8") + lock_path.write_text(lock_text, encoding="utf-8") + with mock.patch.object( + sys, + "argv", + [ + str(SCRIPT), + "--results", + str(result_path), + "--lockfile", + str(lock_path), + "--audit", + str(audit_path), + ], + ): + with unittest.TestCase().assertRaises(SystemExit) as exit_context: + runpy.run_path(str(SCRIPT), run_name="__main__") + assert exit_context.exception.code == 0 + return ( + json.loads(result_path.read_text(encoding="utf-8")), + json.loads(audit_path.read_text(encoding="utf-8")), + ) + + +def remaining_ids(payload: dict[str, object]) -> list[str]: + """Return vulnerability IDs retained in a reconciled document.""" + + package = payload["results"][0]["packages"][0] # type: ignore[index] + return [item["id"] for item in package["vulnerabilities"]] # type: ignore[index] + + +class DirectSourceReconcileTests(unittest.TestCase): + """Exercise exact provenance, affected-range, and fail-closed boundaries.""" + + def run_case( + self, payload: dict[str, object], lock_text: str + ) -> tuple[dict[str, object], list[dict[str, object]]]: + """Run one isolated production reconciliation case.""" + + with tempfile.TemporaryDirectory() as directory: + return reconcile(Path(directory), payload, lock_text) + + def test_official_immutable_xlsx_0203_drops_only_metadata_disproven_findings(self) -> None: + """Drop only advisories disproven by the exact immutable release evidence.""" + payload = results( + "0.20.3", + [ + vulnerability("GHSA-4r6h-8v6p-xvw6", "< 0.19.3"), + vulnerability("GHSA-5pgg-2g8v-p4x9", "< 0.20.2"), + ], + ) + + reconciled, audit = self.run_case(payload, direct_lock("0.20.3")) + + self.assertEqual(remaining_ids(reconciled), []) + self.assertEqual( + [entry["status"] for entry in audit], ["RECONCILED", "RECONCILED"] + ) + self.assertTrue(all(entry["integrity"] == INTEGRITY for entry in audit)) + + def test_registry_finding_from_another_lockfile_cannot_borrow_direct_source_provenance( + self, + ) -> None: + """Bind reconciliation to the exact scanner source that supplied the lock evidence.""" + payload = results( + "0.20.3", [vulnerability("GHSA-4r6h-8v6p-xvw6", "< 0.19.3")] + ) + payload["results"][0]["source"]["path"] = ( # type: ignore[index] + "/github/workspace/packages/registry/pnpm-lock.yaml" + ) + + reconciled, audit = OSV.reconcile_payload( + payload, + direct_lock("0.20.3"), + label="head", + source_path="pnpm-lock.yaml", + ) + + self.assertEqual(remaining_ids(reconciled), ["GHSA-4r6h-8v6p-xvw6"]) + self.assertEqual(audit[0]["status"], "SCANNER_METADATA_CONFLICT") + self.assertIn("does not match governed lockfile", audit[0]["reason"]) + + def test_container_workspace_path_matches_governed_root_lockfile(self) -> None: + """Accept OSV's pinned container mount path for the exact root lockfile.""" + payload = results( + "0.20.3", [vulnerability("GHSA-4r6h-8v6p-xvw6", "< 0.19.3")] + ) + payload["results"][0]["source"]["path"] = ( # type: ignore[index] + "/github/workspace/pnpm-lock.yaml" + ) + + reconciled, audit = OSV.reconcile_payload( + payload, + direct_lock("0.20.3"), + label="head", + source_path="pnpm-lock.yaml", + ) + + self.assertEqual(remaining_ids(reconciled), []) + self.assertEqual(audit[0]["status"], "RECONCILED") + + def test_source_binding_helpers_and_malformed_cross_source_evidence_fail_closed( + self, + ) -> None: + """Cover normalized-path validation and cross-source malformed evidence.""" + payload = results("0.20.3", []) + self.assertEqual(len(list(OSV.iter_packages(payload))), 1) + for invalid in ("", "/pnpm-lock.yaml", r"nested\pnpm-lock.yaml", "../pnpm-lock.yaml"): + with self.subTest(source_path=invalid), self.assertRaisesRegex( + ValueError, "normalized relative path" + ): + OSV.validate_source_path(invalid) + + unrelated = results("1.0.0", []) + unrelated["results"][0]["source"]["path"] = "other-lock.json" # type: ignore[index] + unrelated["results"][0]["packages"][0]["package"]["name"] = "react" # type: ignore[index] + reconciled, audit = OSV.reconcile_payload( + unrelated, direct_lock("0.20.3"), label="head" + ) + self.assertEqual(reconciled, unrelated) + self.assertEqual(audit, []) + + malformed = results("0.20.3", []) + malformed["results"][0]["source"]["path"] = "other-lock.json" # type: ignore[index] + malformed["results"][0]["packages"][0]["vulnerabilities"] = ["bad"] # type: ignore[index] + with self.assertRaisesRegex(TypeError, "vulnerability entries"): + OSV.reconcile_payload( + malformed, direct_lock("0.20.3"), label="head" + ) + + def test_official_but_affected_versions_remain_findings(self) -> None: + """Keep versions inside their authoritative affected range.""" + for version, affected_range in ( + ("0.18.5", "< 0.19.3"), + ("0.19.2", "< 0.19.3"), + ("0.20.1", "< 0.20.2"), + ): + with self.subTest(version=version): + payload = results( + version, [vulnerability("GHSA-control", affected_range)] + ) + reconciled, audit = self.run_case(payload, direct_lock(version)) + self.assertEqual(remaining_ids(reconciled), ["GHSA-control"]) + self.assertEqual(audit[0]["status"], "AFFECTED") + + def test_registry_package_never_borrows_direct_source_exception(self) -> None: + """Never apply a direct-source exception to a registry-backed package.""" + payload = results("0.18.5", [vulnerability("GHSA-registry", "< 0.19.3")]) + reconciled, audit = self.run_case(payload, registry_lock("0.18.5")) + self.assertEqual(remaining_ids(reconciled), ["GHSA-registry"]) + self.assertEqual(audit[0]["status"], "SCANNER_METADATA_CONFLICT") + + payload["results"][0]["packages"][0]["package"]["name"] = "other" + reconciled, audit = OSV.reconcile_payload( + payload, registry_lock("0.18.5"), label="other" + ) + self.assertEqual(audit, []) + self.assertEqual( + reconciled["results"][0]["packages"][0]["vulnerabilities"], + [vulnerability("GHSA-registry", "< 0.19.3")], + ) + + def test_unverifiable_direct_provenance_fails_closed(self) -> None: + """Retain findings when direct URL or integrity provenance is unverifiable.""" + for lock_text in ( + direct_lock("0.20.3", integrity=None), + direct_lock("0.20.3", host="example.invalid"), + direct_lock("0.20.3").replace( + "tarball: https://cdn.sheetjs.com/xlsx-0.20.3/xlsx-0.20.3.tgz", + "tarball: https://cdn.sheetjs.com/xlsx-0.20.2/xlsx-0.20.2.tgz", + ), + ): + with self.subTest(lock_text=lock_text): + payload = results( + "0.20.3", + [vulnerability("GHSA-unknown-source", "< 0.20.2")], + ) + reconciled, audit = self.run_case(payload, lock_text) + self.assertEqual(remaining_ids(reconciled), ["GHSA-unknown-source"]) + self.assertEqual(audit[0]["status"], "SCANNER_METADATA_CONFLICT") + + def test_only_exact_sheetjs_exception_version_can_reconcile(self) -> None: + """Restrict reconciliation to the one explicitly governed SheetJS version.""" + for version in ("0.20.2", "0.20.4", "1.0.0"): + with self.subTest(version=version): + payload = results( + version, [vulnerability("GHSA-version-control", "< 0.20.2")] + ) + reconciled, audit = self.run_case( + payload, direct_lock(version) + ) + self.assertEqual( + remaining_ids(reconciled), ["GHSA-version-control"] + ) + self.assertEqual(audit[0]["status"], "SCANNER_METADATA_CONFLICT") + + def test_conflicting_duplicate_direct_sources_fail_closed(self) -> None: + """Reject duplicate direct-source records and malformed vulnerability entries.""" + payload = results( + "0.20.3", [vulnerability("GHSA-duplicate-source", "< 0.20.2")] + ) + lock_text = direct_lock("0.20.3") + direct_lock( + "0.20.3", host="example.invalid" + ) + reconciled, audit = self.run_case(payload, lock_text) + self.assertEqual(remaining_ids(reconciled), ["GHSA-duplicate-source"]) + self.assertEqual(audit[0]["status"], "SCANNER_METADATA_CONFLICT") + self.assertIn("multiple direct-source", audit[0]["reason"]) + + payload["results"][0]["packages"][0]["vulnerabilities"] = ["bad"] + with self.assertRaises(TypeError): + OSV.reconcile_payload(payload, lock_text, label="bad") + + def test_exact_exception_version_inside_range_remains_affected(self) -> None: + """Retain the exception version when the advisory range still includes it.""" + payload = results( + "0.20.3", [vulnerability("GHSA-affected-exception", "< 0.20.4")] + ) + reconciled, audit = self.run_case(payload, direct_lock("0.20.3")) + self.assertEqual( + remaining_ids(reconciled), ["GHSA-affected-exception"] + ) + self.assertEqual(audit[0]["status"], "AFFECTED") + + def test_inclusive_affected_bound_is_respected(self) -> None: + """Treat an OSV ``<=`` upper bound as affected at the boundary.""" + for affected_range, expected_ids, expected_status in ( + ("<= 0.20.3", ["GHSA-inclusive"], "AFFECTED"), + ("<= 0.20.2", [], "RECONCILED"), + ): + with self.subTest(affected_range=affected_range): + payload = results( + "0.20.3", [vulnerability("GHSA-inclusive", affected_range)] + ) + reconciled, audit = self.run_case(payload, direct_lock("0.20.3")) + self.assertEqual(remaining_ids(reconciled), expected_ids) + self.assertEqual(audit[0]["status"], expected_status) + + def test_low_level_semver_integrity_and_source_validation_boundaries(self) -> None: + """Exercise malformed SemVer, digest, URL, and package-source boundaries.""" + self.assertEqual(OSV.parse_semver("0.20.3"), (0, 20, 3)) + self.assertIsNone(OSV.parse_semver("01.20.3")) + self.assertFalse(OSV.valid_sha512_integrity("sha256-deadbeef")) + self.assertFalse(OSV.valid_sha512_integrity("sha512-%%%")) + self.assertFalse(OSV.valid_sha512_integrity("sha512-YQ==")) + valid_url = OFFICIAL_URL.format(version="0.20.3") + well_formed_but_untrusted_integrity = "sha512-" + ("A" * 86) + "==" + self.assertTrue(OSV.valid_sha512_integrity(well_formed_but_untrusted_integrity)) + cases = ( + ("other", valid_url, valid_url, "0.20.3", INTEGRITY), + ("xlsx", "https://cdn.sheetjs.com:bad/x.xlsx", "", "0.20.3", INTEGRITY), + ("xlsx", valid_url.replace("https://", "http://"), valid_url, "0.20.3", INTEGRITY), + ("xlsx", valid_url.replace("cdn.sheetjs.com", "example.invalid"), valid_url, "0.20.3", INTEGRITY), + ("xlsx", valid_url.replace("cdn.sheetjs.com", "cdn.sheetjs.com:443"), valid_url, "0.20.3", INTEGRITY), + ("xlsx", valid_url.replace("cdn.sheetjs.com", "user@cdn.sheetjs.com"), valid_url, "0.20.3", INTEGRITY), + ("xlsx", valid_url.replace("cdn.sheetjs.com", "user:pass@cdn.sheetjs.com"), valid_url, "0.20.3", INTEGRITY), + ("xlsx", valid_url + "?download=1", valid_url, "0.20.3", INTEGRITY), + ("xlsx", valid_url + "#fragment", valid_url, "0.20.3", INTEGRITY), + ("xlsx", "https://cdn.sheetjs.com/not-xlsx.tgz", "", "0.20.3", INTEGRITY), + ("xlsx", valid_url, valid_url, "0.20.2", INTEGRITY), + ("xlsx", valid_url, valid_url.replace("0.20.3", "0.20.2"), "0.20.3", INTEGRITY), + ("xlsx", valid_url, valid_url, "0.20.3", "missing"), + ( + "xlsx", + valid_url, + valid_url, + "0.20.3", + well_formed_but_untrusted_integrity, + ), + ) + for arguments in cases: + with self.subTest(arguments=arguments): + self.assertFalse(OSV.validate_sheetjs_source(*arguments)[0]) + self.assertTrue( + OSV.validate_sheetjs_source("xlsx", valid_url, valid_url, "0.20.3", INTEGRITY)[0] + ) + + def test_direct_source_parser_handles_boundaries_and_missing_fields(self) -> None: + """Parse adjacent lockfile records without trusting incomplete resolutions.""" + valid_url = OFFICIAL_URL.format(version="0.20.3") + lock = ( + "packages:\n\n" + f" xlsx@{valid_url}:\n" + " resolution: {}\n" + " other@https://example.invalid/other.tgz:\n" + " resolution: {tarball: https://example.invalid/other.tgz}\n" + ) + sources = OSV.parse_direct_sources(lock) + self.assertEqual(len(sources), 2) + self.assertFalse(sources[0].valid) + self.assertFalse(sources[1].valid) + + def test_direct_source_parser_ignores_pnpm_snapshot_duplicates(self) -> None: + """Read one package provenance record when pnpm repeats it in snapshots.""" + valid_url = OFFICIAL_URL.format(version="0.20.3") + lock = direct_lock("0.20.3") + ( + "\nsnapshots:\n\n" + f" xlsx@{valid_url}:\n" + " dependencies: {}\n" + ) + sources = OSV.parse_direct_sources(lock) + self.assertEqual(len(sources), 1) + self.assertTrue(sources[0].valid) + payload = results("0.20.3", [vulnerability("GHSA-snapshot", "< 0.20.2")]) + reconciled, audit = OSV.reconcile_payload(payload, lock, label="snapshot") + self.assertEqual(remaining_ids(reconciled), []) + self.assertEqual(audit[0]["status"], "RECONCILED") + + def test_malformed_osv_container_evidence_fails_closed(self) -> None: + """Reject malformed OSV containers while accepting an empty result set.""" + malformed = ( + {}, + {"results": ["bad"]}, + {"results": [{"packages": "bad"}]}, + {"results": [{"packages": ["bad"]}]}, + ) + for payload in malformed: + with self.subTest(payload=payload), self.assertRaises(TypeError): + list(OSV.iter_packages(payload)) + self.assertEqual(list(OSV.iter_packages({"results": [{"packages": []}]})), []) + + def test_authoritative_range_rejects_every_ambiguous_shape(self) -> None: + """Reject advisory shapes that cannot prove one authoritative affected range.""" + valid = vulnerability("GHSA-shape", "< 0.20.2") + self.assertEqual(OSV.authoritative_affected_range(valid, "xlsx"), "< 0.20.2") + malformed = [ + {}, + {"id": 1, "affected": []}, + {"id": "GHSA-shape", "affected": "bad"}, + {"id": "GHSA-shape", "affected": [None]}, + ] + item_mutations = ( + ("package", None), + ("package", {"ecosystem": "PyPI", "name": "xlsx"}), + ("package", {"ecosystem": "npm", "name": "other"}), + ("database_specific", None), + ("ranges", None), + ) + for key, value in item_mutations: + candidate = copy.deepcopy(valid) + candidate["affected"][0][key] = value + malformed.append(candidate) + database_mutations = ( + ("last_known_affected_version_range", None), + ("source", None), + ("source", "https://example.invalid/advisory.json"), + ("source", "https://github.com/github/advisory-database/blob/main/wrong.json"), + ) + for key, value in database_mutations: + candidate = copy.deepcopy(valid) + candidate["affected"][0]["database_specific"][key] = value + malformed.append(candidate) + for ranges in ( + [], + [{"type": "ECOSYSTEM", "events": [{"introduced": "0"}]}], + [ + {"type": "SEMVER", "events": [{"introduced": "0"}]}, + {"type": "SEMVER", "events": [{"introduced": "0"}]}, + ], + [{"type": "SEMVER", "events": [{"introduced": "1.0.0"}]}], + [None], + ): + candidate = copy.deepcopy(valid) + candidate["affected"][0]["ranges"] = ranges + malformed.append(candidate) + conflicting = copy.deepcopy(valid) + second = copy.deepcopy(conflicting["affected"][0]) + second["database_specific"]["last_known_affected_version_range"] = "< 0.19.3" + conflicting["affected"].append(second) + malformed.append(conflicting) + duplicate = copy.deepcopy(valid) + duplicate["affected"].append(copy.deepcopy(duplicate["affected"][0])) + self.assertEqual(OSV.authoritative_affected_range(duplicate, "xlsx"), "< 0.20.2") + for candidate in malformed: + with self.subTest(candidate=candidate): + self.assertIsNone(OSV.authoritative_affected_range(candidate, "xlsx")) + + def test_reconcile_rejects_malformed_packages_and_vulnerabilities(self) -> None: + """Reject malformed package and vulnerability records before rewriting results.""" + bad_packages = ( + {"results": [{"packages": [{"package": None, "vulnerabilities": []}]}]}, + {"results": [{"packages": [{"package": {}, "vulnerabilities": "bad"}]}]}, + ) + for payload in bad_packages: + with self.subTest(payload=payload), self.assertRaises(TypeError): + OSV.reconcile_payload(payload, direct_lock("0.20.3"), label="bad") + bad_vulnerability = results("0.20.3", []) + bad_vulnerability["results"][0]["packages"][0]["vulnerabilities"] = ["bad"] + with self.assertRaises(TypeError): + OSV.reconcile_payload(bad_vulnerability, direct_lock("0.20.3"), label="bad") + untouched, audit = OSV.reconcile_payload( + results("0.20.3", [vulnerability("GHSA-registry", "< 0.20.2")]), + registry_lock("0.20.3"), + label="registry", + ) + self.assertEqual(remaining_ids(untouched), ["GHSA-registry"]) + self.assertEqual(audit[0]["status"], "SCANNER_METADATA_CONFLICT") + + def test_io_and_existing_audit_boundaries_fail_closed(self) -> None: + """Exercise regular-file, UTF-8, audit, and atomic-write failure boundaries.""" + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + missing = root / "missing.json" + with self.assertRaises(ValueError): + OSV.load_json_object(missing) + target = root / "target.json" + target.write_text("{}", encoding="utf-8") + linked = root / "linked.json" + linked.symlink_to(target) + with self.assertRaises(ValueError): + OSV.load_json_object(linked) + with mock.patch.object(OSV.os, "O_NOFOLLOW", None), self.assertRaises(ValueError): + OSV.read_utf8_text(target, "required JSON input") + directory = root / "directory" + directory.mkdir() + with self.assertRaises(ValueError): + OSV.read_utf8_text(directory, "required JSON input") + array_path = root / "array.json" + array_path.write_text("[]", encoding="utf-8") + with self.assertRaises(TypeError): + OSV.load_json_object(array_path) + destination = root / "atomic.json" + with mock.patch.object( + OSV.os, "replace", side_effect=OSError("replace failed") + ), self.assertRaises(OSError): + OSV.atomic_json_write(destination, {}) + + malformed = root / "malformed.json" + malformed.write_text('{"results": "bad"}', encoding="utf-8") + malformed_argv = [ + str(SCRIPT), "--results", str(malformed), "--lockfile", str(target), + "--audit", str(root / "malformed-audit.json"), + ] + with ( + mock.patch.object(OSV.sys, "argv", malformed_argv), + mock.patch.object(OSV.sys, "stderr", new_callable=io.StringIO) as stderr, + ): + self.assertEqual(OSV.main(), 1) + self.assertIn("OSV results must contain a results array", stderr.getvalue()) + + results_path = root / "results.json" + lock_path = root / "pnpm-lock.yaml" + audit_path = root / "audit.json" + results_path.write_text(json.dumps({"results": []}), encoding="utf-8") + lock_path.write_text("lockfileVersion: '9.0'\n", encoding="utf-8") + audit_path.write_text('[{"status":"prior"}]', encoding="utf-8") + argv = [ + str(SCRIPT), "--results", str(results_path), "--lockfile", str(lock_path), + "--audit", str(audit_path), "--label", "head", + ] + with ( + mock.patch.object( + OSV, "atomic_json_write", side_effect=OSError("write failed") + ), + mock.patch.object(OSV.sys, "argv", argv), + mock.patch.object( + OSV.sys, "stderr", new_callable=io.StringIO + ) as stderr, + ): + self.assertEqual(OSV.main(), 1) + self.assertIn("::error::write failed", stderr.getvalue()) + + with mock.patch.object(sys, "argv", argv): + self.assertEqual(OSV.main(), 0) + self.assertEqual(json.loads(audit_path.read_text(encoding="utf-8")), [{"status": "prior"}]) + + lock_path.write_bytes(b"lockfileVersion: '9.0'\n\xc0\xc0") + with ( + mock.patch.object(sys, "argv", argv), + mock.patch.object(OSV.sys, "stderr", new_callable=io.StringIO) as stderr, + ): + self.assertEqual(OSV.main(), 1) + self.assertIn("pnpm lock provenance input is not valid UTF-8", stderr.getvalue()) + + lock_path.write_text("lockfileVersion: '9.0'\n", encoding="utf-8") + + audit_path.write_text("{}", encoding="utf-8") + with mock.patch.object(sys, "argv", argv): + self.assertEqual(OSV.main(), 1) + audit_path.unlink() + audit_target = root / "audit-target.json" + audit_target.write_text("[]", encoding="utf-8") + audit_path.symlink_to(audit_target) + with mock.patch.object(sys, "argv", argv): + self.assertEqual(OSV.main(), 1) + lock_path.unlink() + with mock.patch.object(sys, "argv", argv): + self.assertEqual(OSV.main(), 1) + + def test_missing_authoritative_affected_range_fails_closed(self) -> None: + """Retain findings when the advisory lacks a machine-checkable range.""" + payload = results("0.20.3", [vulnerability("GHSA-unknown-range", None)]) + reconciled, audit = self.run_case(payload, direct_lock("0.20.3")) + self.assertEqual(remaining_ids(reconciled), ["GHSA-unknown-range"]) + self.assertEqual(audit[0]["status"], "SCANNER_METADATA_CONFLICT") + + def test_reusable_security_scan_reconciles_before_reporter_verdict(self) -> None: + """Require provenance reconciliation before the reusable scan publishes its verdict.""" + workflow = SECURITY_WORKFLOW.read_text(encoding="utf-8") + preserve = workflow.index("Preserve base OSV evidence and direct-source provenance") + head_scan = workflow.index("Scan head with OSV") + policy = workflow.index("Checkout exact central provenance policy") + reconcile_step = workflow.index("Reconcile immutable direct-source provenance") + require_output = workflow.index("Require OSV scan output") + reporter = workflow.index("Report PR-introduced OSV findings") + + self.assertLess(preserve, head_scan) + self.assertLess(head_scan, policy) + self.assertLess(policy, reconcile_step) + self.assertLess(reconcile_step, require_output) + self.assertLess(require_output, reporter) + self.assertIn("github.workflow_sha", workflow) + self.assertIn( + "github.event.pull_request.base.repo.full_name == 'ContextualWisdomLab/.github'", + workflow, + ) + self.assertIn("github.event.pull_request.base.sha || github.workflow_sha", workflow) + self.assertNotIn( + "github.repository == 'ContextualWisdomLab/.github' && github.event.pull_request.head.sha", + workflow, + ) + self.assertIn("osv_direct_source_reconcile.py", workflow) + self.assertIn("osv-provenance-audit.json", workflow) + self.assertIn('if [ ! -f "$reconciler" ]; then', workflow) + self.assertIn( + "trusted provenance policy is not yet present; retaining raw OSV evidence", + workflow, + ) + self.assertNotIn('test -f "$reconciler"', workflow) + self.assertEqual(workflow.count("--source-path source/pnpm-lock.yaml"), 2) + self.assertIn("--lockfile source/pnpm-lock.yaml", workflow) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_required_workflow_queue_contract.py b/tests/test_required_workflow_queue_contract.py index 80d34abdf..70e53d82d 100644 --- a/tests/test_required_workflow_queue_contract.py +++ b/tests/test_required_workflow_queue_contract.py @@ -32,6 +32,14 @@ def workflow_step(workflow: str, name: str) -> str: return workflow[start:end] +def osv_result_classifier_script(workflow: str) -> str: + step = workflow_step(workflow, "Install trusted OSV result evidence classifier") + marker = " cat >\"$RUNNER_TEMP/classify-osv-result.py\" <<'PY'\n" + start = step.index(marker) + len(marker) + end = step.index("\n PY", start) + return textwrap.dedent(step[start:end]) + + def test_merge_scheduler_dispatches_one_review_by_default() -> None: """Keep the default scheduler dispatch bounded to one review.""" workflow = workflow_text("pr-review-merge-scheduler.yml") @@ -1344,10 +1352,14 @@ def test_security_scan_preserves_base_output_across_cross_fork_checkout() -> Non workflow = workflow_text("security-scan.yml") assert workflow.count("--allow-no-lockfiles") == 4 + assert "--output-file=old-results.json" in workflow + assert "--output-file=new-results.json" in workflow + assert "--output-files=sarif:results.sarif" in workflow + assert "--output-file=results.sarif" not in workflow + assert "--output=old-results.json" not in workflow + assert "--output=new-results.json" not in workflow assert workflow.count("path: source") == 2 - assert workflow.count("--output=old-results.json") == 2 - assert workflow.count("--output=new-results.json") == 2 - assert workflow.count("source/") == 4 + assert workflow.count("\n source/\n") == 4 assert "clean: false" not in workflow assert "test -s old-results.json" in workflow assert "test -s new-results.json" in workflow @@ -1399,14 +1411,18 @@ def test_osv_scan_logs_and_retries_without_transitive_resolution_on_resolver_fai ) assert "id: osv_base" in workflow assert "id: osv_head" in workflow - assert "steps.osv_base.outcome == 'failure'" in workflow - assert "steps.osv_head.outcome == 'failure'" in workflow + assert "id: osv_base_retry" in workflow + assert "id: osv_head_retry" in workflow + assert "Classify base OSV result evidence" in workflow + assert "Classify head OSV result evidence" in workflow + assert "Classify retried base OSV result evidence" in workflow + assert "Classify retried head OSV result evidence" in workflow assert "Retry base OSV without transitive resolution" in workflow assert "Retry head OSV without transitive resolution" in workflow assert workflow.count("timeout-minutes: 8") == 2 assert workflow.count("timeout-minutes: 4") == 2 assert workflow.count("\n --no-resolve\n") == 4 - assert workflow.count("failed or timed out before reporter output was trusted") == 2 + assert workflow.count("did not produce authoritative result evidence") == 2 assert ( "Direct manifest and lockfile vulnerability evidence remains enforced" in workflow @@ -1415,19 +1431,158 @@ def test_osv_scan_logs_and_retries_without_transitive_resolution_on_resolver_fai "external transitive registry resolution is intentionally avoided" in workflow ) assert ( - "Retry base OSV without transitive resolution\n if: steps.osv_base.outcome == 'failure'\n continue-on-error: true" + "Retry base OSV without transitive resolution\n" + " if: steps.osv_base_evidence.outputs.complete != 'true'\n" + " id: osv_base_retry\n continue-on-error: true" in workflow ) assert ( - "Retry head OSV without transitive resolution\n if: steps.osv_head.outcome == 'failure'\n continue-on-error: true" + "Retry head OSV without transitive resolution\n" + " if: steps.osv_head_evidence.outputs.complete != 'true'\n" + " id: osv_head_retry\n continue-on-error: true" in workflow ) - assert "--output=old-results.json" in workflow - assert "--output=new-results.json" in workflow + assert "--output-file=old-results.json" in workflow + assert "--output-file=new-results.json" in workflow + assert "--output-files=sarif:results.sarif" in workflow + assert "--output=old-results.json" not in workflow + assert "--output=new-results.json" not in workflow + assert "Require authoritative base and head OSV evidence" in workflow + assert "Require successful base and head OSV scans" not in workflow + assert "Normalize successful empty OSV result documents" not in workflow + assert workflow.count("failure with authoritative vulnerability evidence") == 1 + assert workflow.count("completed successfully without findings output") == 1 + assert workflow.count('runpy.run_path(os.path.join(os.environ["RUNNER_TEMP"]') == 4 + assert "Preserve base OSV evidence and direct-source provenance" in workflow + assert 'cp -- old-results.json "$RUNNER_TEMP/osv-base-provenance/old-results.json"' in workflow + assert "Restore authoritative base OSV evidence" in workflow + assert workflow.index("Preserve base OSV evidence and direct-source provenance") < workflow.index( + "Checkout head" + ) < workflow.index("Restore authoritative base OSV evidence") + assert workflow.index("Require authoritative base and head OSV evidence") < workflow.index( + "Require OSV scan output" + ) assert "Print OSV findings being compared" in workflow assert "OSV {label} scan produced {len(findings)} finding(s)" in workflow +@pytest.mark.parametrize( + ("outcome", "payload", "expected_complete", "expected_message"), + [ + ( + "failure", + {"results": [{"packages": [{"vulnerabilities": [{"id": "PYSEC-1"}]}]}]}, + "true", + "failure with authoritative vulnerability evidence", + ), + ( + "failure", + {"results": []}, + "false", + "failed without authoritative vulnerability evidence", + ), + ( + "failure", + {"results": "malformed"}, + "false", + "results must be a list", + ), + ], +) +def test_osv_result_evidence_classifier_distinguishes_findings_from_scan_failure( + tmp_path: Path, + outcome: str, + payload: dict[str, object], + expected_complete: str, + expected_message: str, +) -> None: + workflow = workflow_text("security-scan.yml") + script = osv_result_classifier_script(workflow) + result_path = tmp_path / "old-results.json" + result_path.write_text(json.dumps(payload), encoding="utf-8") + output_path = tmp_path / "github-output" + + result = subprocess.run( + [sys.executable, "-"], + input=script, + text=True, + capture_output=True, + env={ + **os.environ, + "SCAN_OUTCOME": outcome, + "RESULT_FILE": str(result_path), + "GITHUB_OUTPUT": str(output_path), + }, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert f"complete={expected_complete}" in output_path.read_text(encoding="utf-8") + assert expected_message in result.stdout + + +def test_osv_result_evidence_classifier_normalizes_only_successful_empty_scan( + tmp_path: Path, +) -> None: + workflow = workflow_text("security-scan.yml") + script = osv_result_classifier_script(workflow) + result_path = tmp_path / "old-results.json" + output_path = tmp_path / "github-output" + + result = subprocess.run( + [sys.executable, "-"], + input=script, + text=True, + capture_output=True, + env={ + **os.environ, + "SCAN_OUTCOME": "success", + "RESULT_FILE": str(result_path), + "GITHUB_OUTPUT": str(output_path), + }, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert output_path.read_text(encoding="utf-8").strip() == "complete=true" + assert json.loads(result_path.read_text(encoding="utf-8")) == {"results": []} + + +def test_osv_result_evidence_classifier_rejects_symlinked_finding_document( + tmp_path: Path, +) -> None: + workflow = workflow_text("security-scan.yml") + script = osv_result_classifier_script(workflow) + target_path = tmp_path / "target.json" + target_path.write_text( + json.dumps( + {"results": [{"packages": [{"vulnerabilities": [{"id": "spoof"}]}]}]} + ), + encoding="utf-8", + ) + result_path = tmp_path / "old-results.json" + result_path.symlink_to(target_path) + output_path = tmp_path / "github-output" + + result = subprocess.run( + [sys.executable, "-"], + input=script, + text=True, + capture_output=True, + env={ + **os.environ, + "SCAN_OUTCOME": "failure", + "RESULT_FILE": str(result_path), + "GITHUB_OUTPUT": str(output_path), + }, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert output_path.read_text(encoding="utf-8").strip() == "complete=false" + assert "result document must be a regular file" in result.stdout + + def test_osv_sarif_upload_is_marked_comprehensive_after_clean_comparison( tmp_path: Path, ) -> None: