diff --git a/.github/workflows/dependency-audit.yml b/.github/workflows/dependency-audit.yml index dea9fd29..eea49137 100644 --- a/.github/workflows/dependency-audit.yml +++ b/.github/workflows/dependency-audit.yml @@ -15,49 +15,44 @@ jobs: steps: - name: Checkout code uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 - - name: Set up Python 3.14 uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 with: python-version: "3.14" - - - name: Install pip-audit and runtime dependencies + - name: Build wheel and isolated reporting environments run: | - python -m pip install --upgrade pip - python -m pip install 'pip-audit==2.10.1' . - - - name: Audit runtime dependencies + python -m pip install 'pip-audit==2.10.1' build + python -m build --wheel + python -m venv --without-pip "$RUNNER_TEMP/runtime" + python -m pip --python "$RUNNER_TEMP/runtime/bin/python" install dist/*.whl + python -m venv --without-pip "$RUNNER_TEMP/development" + python -m pip --python "$RUNNER_TEMP/development/bin/python" install '.[dev]' + - name: Report runtime dependency policy id: runtime continue-on-error: true run: | - set +e - python -m pip_audit --format markdown > runtime-audit.md - status=$? - set -e - cat runtime-audit.md >> "$GITHUB_STEP_SUMMARY" - exit "$status" - - - name: Install development dependencies - run: python -m pip install '.[dev]' - - - name: Audit runtime and development dependencies - id: development + site_packages=$("$RUNNER_TEMP/runtime/bin/python" -c 'import sysconfig; print(sysconfig.get_path("purelib"))') + wheels=(dist/*.whl) + test "${#wheels[@]}" -eq 1 + make security-runtime SECURITY_PYTHON=python SITE_PACKAGES="$site_packages" CANDIDATE_WHEEL="${wheels[0]}" SECURITY_OUTPUT=security-results/runtime + - name: Report development dependencies with the same classifier + if: always() continue-on-error: true run: | - set +e - python -m pip_audit --format markdown > development-audit.md - status=$? - set -e - cat development-audit.md >> "$GITHUB_STEP_SUMMARY" - exit "$status" - - - name: Summarize audit outcomes - env: - RUNTIME_OUTCOME: ${{ steps.runtime.outcome }} - DEVELOPMENT_OUTCOME: ${{ steps.development.outcome }} + site_packages=$("$RUNNER_TEMP/development/bin/python" -c 'import sysconfig; print(sysconfig.get_path("purelib"))') + wheels=(dist/*.whl) + test "${#wheels[@]}" -eq 1 + make security-development SECURITY_PYTHON=python SITE_PACKAGES="$site_packages" CANDIDATE_WHEEL="${wheels[0]}" SECURITY_OUTPUT=security-results/development + - name: Verify pinned CDN integrity + if: always() + continue-on-error: true run: | - { - echo "## Dependency audit outcomes" - echo "- Runtime dependencies: ${RUNTIME_OUTCOME}" - echo "- Runtime + development dependencies: ${DEVELOPMENT_OUTCOME}" - } >> "$GITHUB_STEP_SUMMARY" + site_packages=$("$RUNNER_TEMP/runtime/bin/python" -c 'import sysconfig; print(sysconfig.get_path("purelib"))') + PYTHONPATH="$site_packages" make security-browser SECURITY_PYTHON="$RUNNER_TEMP/runtime/bin/python" SECURITY_OUTPUT=security-results/browser + - name: Upload redacted reporting evidence + if: always() + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: dependency-security-report + path: security-results/** + if-no-files-found: error diff --git a/.github/workflows/publish-pypi.yml b/.github/workflows/publish-pypi.yml index a0465e63..ec8406d5 100644 --- a/.github/workflows/publish-pypi.yml +++ b/.github/workflows/publish-pypi.yml @@ -235,9 +235,58 @@ jobs: AZURE_SUBSCRIPTION_ID: ${{ secrets.AZURE_SUBSCRIPTION_ID }} + runtime-dependency-audit: + name: Runtime dependency and browser integrity gate + needs: build + runs-on: ubuntu-latest + permissions: + contents: read + strategy: + fail-fast: false + matrix: + interpreter: ["3.11", "3.12", "3.13", "3.14"] + steps: + - name: Checkout security policy and tooling + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - name: Set up audit interpreter + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: ${{ matrix.interpreter }} + - name: Download the same candidate wheel + uses: actions/download-artifact@9000827ccba6bdab643e8b6fd33ac0654aef8333 # v8.0.2 + with: + name: dist + path: candidate-dist + - name: Install only the candidate runtime closure in an isolated environment + run: | + python -m pip install 'pip-audit==2.10.1' + python -m venv --without-pip "$RUNNER_TEMP/candidate" + wheels=(candidate-dist/*.whl) + test "${#wheels[@]}" -eq 1 + python -m pip --python "$RUNNER_TEMP/candidate/bin/python" install "${wheels[0]}" + - name: Enforce runtime advisory policy + if: always() + run: | + site_packages=$("$RUNNER_TEMP/candidate/bin/python" -c 'import sysconfig; print(sysconfig.get_path("purelib"))') + wheels=(candidate-dist/*.whl) + test "${#wheels[@]}" -eq 1 + make security-runtime SECURITY_PYTHON=python SITE_PACKAGES="$site_packages" CANDIDATE_WHEEL="${wheels[0]}" SECURITY_OUTPUT=security-results/runtime + - name: Verify browser CDN integrity from the candidate wheel + if: always() + run: | + site_packages=$("$RUNNER_TEMP/candidate/bin/python" -c 'import sysconfig; print(sysconfig.get_path("purelib"))') + PYTHONPATH="$site_packages" make security-browser SECURITY_PYTHON="$RUNNER_TEMP/candidate/bin/python" SECURITY_OUTPUT=security-results/browser + - name: Upload redacted release security evidence + if: always() + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: release-security-${{ matrix.interpreter }} + path: security-results/** + if-no-files-found: error + publish: name: Publish to PyPI - needs: [build, lib-tests, cookbook-smoke, cookbook-host-smoke, azure-e2e] + needs: [build, lib-tests, cookbook-smoke, cookbook-host-smoke, azure-e2e, runtime-dependency-audit] runs-on: ubuntu-latest environment: pypi permissions: diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml index 6f76dbd3..ecd33e9f 100644 --- a/.github/workflows/security.yml +++ b/.github/workflows/security.yml @@ -7,6 +7,7 @@ on: branches: [main] schedule: - cron: "0 2 * * 1" + workflow_dispatch: permissions: contents: read @@ -23,13 +24,23 @@ jobs: with: python-version: "3.11" - - name: Install Hatch and dependencies via Makefile - run: | - pip install hatch - make install + - name: Install scanner dependencies + run: pip install -e ".[dev]" - - name: Run Bandit - run: make security + - name: Run Bandit on declared executable surfaces with fail-closed evidence + if: always() + run: make security-bandit SECURITY_PYTHON=python + + - name: Type check security tooling + run: make security-types SECURITY_PYTHON=python + + - name: Upload redacted security evidence + if: always() + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: security-bandit + path: security-results/** + if-no-files-found: error semgrep: runs-on: ubuntu-latest @@ -38,6 +49,27 @@ jobs: uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 - name: Run Semgrep + id: scan uses: semgrep/semgrep-action@713efdd345f3035192eaa63f56867b88e63e4e5d # v1 with: config: p/python + + - name: Set up evidence interpreter + if: always() + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.11" + + - name: Record Semgrep execution evidence + if: always() + env: + SCAN_OUTCOME: ${{ steps.scan.outcome }} + run: make security-semgrep SECURITY_PYTHON=python + + - name: Upload redacted security evidence + if: always() + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: security-semgrep + path: security-results/** + if-no-files-found: error diff --git a/.gitignore b/.gitignore index 80c3d126..7c3a86f5 100644 --- a/.gitignore +++ b/.gitignore @@ -22,6 +22,7 @@ htmlcov/ # Build artifacts build/ dist/ +security-results/ *.egg-info/ # IDE diff --git a/AGENTS.md b/AGENTS.md index d2133590..7b54b471 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -266,6 +266,7 @@ Every tier runs inside `publish-pypi.yml` on the tag, and **all of them gate the | `cookbook-smoke` | downstream import/registration drift | | `cookbook-host-smoke` | candidate wheel installs cleanly and a real `func` host + Azurite boots with it present, no cloud. NOTE: the cookbook HTTP examples do not import this package, so this is a host-boot smoke, **not** proof of this package's own runtime behavior (tracked separately) | | `azure-e2e` | cloud-only drift — deploys to real Azure, runs the live e2e suite, uploads an `azure-cert` record | +| `runtime-dependency-audit` | blocks unexcepted runtime advisories and CDN/SRI failures on the candidate wheel; see [SECURITY.md](SECURITY.md) for evidence, expiry and tool-error policy | | `publish` | uploads the exact artifact `build` produced; it never rebuilds | ### Recovery @@ -291,6 +292,10 @@ Every tier runs inside `publish-pypi.yml` on the tag, and **all of them gate the ## Golden Commands +Security entry points and candidate-inventory verification are documented in +[SECURITY.md](SECURITY.md#automated-security-controls). `make security` includes the +expanded executable-surface inventory scan used in CI. + Use Makefile entry points only. Do not bypass the Makefile in CI or contributor guidance. | Purpose | Command | diff --git a/Makefile b/Makefile index 115203b1..35eaa74d 100644 --- a/Makefile +++ b/Makefile @@ -67,8 +67,34 @@ lint: ensure-hatch @$(HATCH) run lint .PHONY: security -security: ensure-hatch - @$(HATCH) run security +security: security-bandit + +# Shared security entry points; SECURITY_PYTHON selects an already provisioned interpreter. +SECURITY_PYTHON ?= hatch run python +SECURITY_OUTPUT ?= security-results +SITE_PACKAGES ?= +CANDIDATE_WHEEL ?= +SCAN_OUTCOME ?= +SECURITY_TOOLS := tools/security_report.py tools/security_exceptions.py tools/security_evidence.py tools/dependency_audit_policy.py tools/candidate_inventory.py tools/browser_integrity.py + +.PHONY: security-bandit security-semgrep security-types security-runtime security-development security-browser +security-bandit: + @$(SECURITY_PYTHON) -m tools.security_evidence bandit --output "$(SECURITY_OUTPUT)" + +security-semgrep: + @$(SECURITY_PYTHON) -m tools.security_evidence semgrep --outcome "$(SCAN_OUTCOME)" --output "$(SECURITY_OUTPUT)"; code=$$?; test $$code -eq 0 -o $$code -eq 3 + +security-types: + @$(SECURITY_PYTHON) -m mypy --explicit-package-bases $(SECURITY_TOOLS) + +security-runtime: + @$(SECURITY_PYTHON) -m tools.dependency_audit_policy --site-packages "$(SITE_PACKAGES)" --candidate-wheel "$(CANDIDATE_WHEEL)" --output "$(SECURITY_OUTPUT)" + +security-development: + @$(SECURITY_PYTHON) -m tools.dependency_audit_policy --site-packages "$(SITE_PACKAGES)" --candidate-wheel "$(CANDIDATE_WHEEL)" --development --output "$(SECURITY_OUTPUT)" + +security-browser: + @$(SECURITY_PYTHON) -m tools.browser_integrity --output "$(SECURITY_OUTPUT)" .PHONY: check check: ensure-hatch diff --git a/SECURITY.md b/SECURITY.md index b409e5ba..a9193d41 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -33,3 +33,142 @@ If you prefer email, contact: yeongseon.choe@gmail.com | ------- | --------- | | Latest release | :white_check_mark: | | Older releases | :x: | + +## Automated security controls + +| Control | Where it runs | Enforcement | +| --- | --- | --- | +| Bandit 1.9.4 | PR, main push, weekly, dispatch (`security.yml`) | Required `bandit` merge check; all severities block unless explicitly excepted | +| Semgrep `p/python` | Same workflow | Required `semgrep` merge check; existing SHA-pinned action retained | +| CodeQL Python `security-extended` | PR, main push, weekly | Required `Analyze` merge check; results published to code scanning | +| Runtime dependency audit | Release candidate wheel on Python 3.11–3.14 | Blocking `runtime-dependency-audit` dependency of `publish` | +| Browser CDN/SRI | Release candidate wheel, plus weekly/dispatch dependency reporting | Blocks release on mismatched bytes or unavailable verification | +| Runtime + development dependency audit | Weekly/dispatch (`dependency-audit.yml`) | Reporting-only; each lane's classified result is visible in Job Summary | +| SBOM | Weekly/dispatch (`sbom.yml`) | Reporting inventory of the installed built wheel; verifies expected packages exist | + +Existing `ci-required`, `PR title`, conversation resolution, wheel reuse, host smoke, +and real-Azure certification remain unchanged. The new release security gate does +not replace any existing gate. PyPI still receives only the `build` job's artifact +through trusted OIDC publishing. Security jobs have `contents: read` only; CodeQL's +existing `security-events: write`, Azure OIDC and PyPI OIDC are unchanged. + +### Static scan coverage + +`make security-bandit` scans **all Python files** in `src/`, +`examples/` (including `e2e_app`), `demo/`, `scripts/`, and `tools/`. There are no +per-file exclusions within these surfaces. The helper compares Bandit's file +inventory to the on-disk Python inventory, requires at least 40 files and positive +LOC, and rejects scan errors and inline suppressions (`nosec`/`skipped_tests`). +Tests and fixtures under `tests/` are intentionally outside this executable-product +scan. Generated HTML/JSON/PNG previews, bytecode and other non-Python files are not +Python scan inputs; their Python generators are scanned. Bicep and workflow review +complement these checks; Bandit does not claim to analyze those languages. + +`make security` and `make check-all` include this same expanded-scope check. +Low-severity reviewed tooling findings are retained +in evidence, not hidden with global Bandit skips. Exact ID/file/line exceptions in +`security/bandit-policy.json` explain shell-free subprocess use and literal-marker +false positives. No example/demo exception was required by the baseline scan. + +### Runtime advisory and exception policy + +`security/dependency-audit-policy.json` blocks **any known advisory** in the resolved +runtime closure of the candidate wheel. Advisory severity is recorded when exposed +by pip-audit; missing severity is `UNKNOWN` and still blocks. This deliberately +avoids treating absent CVSS data as low risk. Each supported Python version is +audited because environment markers change the Azure Functions dependency closure. +The candidate environment contains only the wheel and its runtime dependencies: +it is created without pip. The helper reads the candidate wheel METADATA and +requires exactly one installed distribution with the same normalized name and +exact version before excluding only that candidate. It freezes every installed +third-party distribution to `name==version` and runs pip-audit 2.10.1 outside the +environment with `--requirement`, `--no-deps`, and `--disable-pip`. The returned +inventory must match those exact pins; a returned candidate must match its verified +identity. An unpublished first-party version therefore needs no PyPI record, but +missing/mismatched identity, an empty closure, third-party skips or omitted packages +remain blocking. Development tools cannot contaminate the release runtime audit. + +Use the same Makefile entry points locally (after provisioning the audit tool): + +```bash +make security-runtime SECURITY_PYTHON=/path/to/audit/bin/python \ + SITE_PACKAGES=/path/to/candidate/lib/python3.11/site-packages \ + CANDIDATE_WHEEL=/path/to/candidate.whl SECURITY_OUTPUT=security-results/runtime +make security-development SECURITY_PYTHON=/path/to/audit/bin/python \ + SITE_PACKAGES=/path/to/development/lib/python3.11/site-packages \ + CANDIDATE_WHEEL=/path/to/candidate.whl SECURITY_OUTPUT=security-results/development +make security-browser SECURITY_PYTHON=/path/to/candidate/bin/python +make security-types +``` + +Runtime checking blocks releases; scheduled runtime/development reporting uses +the same non-zero classifier exits with explicitly reporting-only workflow steps. + +Exceptions require maintainer review and `id`, `package`, `kind`, `reason`, `owner`, +`added`, and `expires` (ISO dates), with a positive lifetime of at most 90 days. +Allowed dependency kinds are `not-affected` (document independently verified +affectedness evidence) and `fix-not-available` (valid only while the advisory lists +no fixed version). Package and advisory ID must match exactly (package separators +are normalized); aliases are not silently ignored. Expired exceptions fail even +when unused. Unused/stale exception counts are reported for removal. The runtime +policy starts empty. Static exceptions use `reviewed-static` plus exact file/line, +with the same ownership, rationale and expiry requirements. Do not add `# nosec` +or a blanket rule exclusion as an alternative to this review. + +### Failure semantics and recovery + +Helpers exit `0` for `PASS`, `1` for `FAIL` (findings, integrity mismatch or expired +policy exception), `2` for `TOOL_ERROR` (crash, invalid/missing JSON, audit collection +or network failure), and `3` for `INCOMPLETE` (empty/undersized scan, missing coverage, +unsafe suppressions or missing execution evidence). All non-PASS results block the +release gate. A pip-audit skipped/unavailable package is not a clean audit. The +Semgrep action exposes only its step outcome here: success is recorded as INCOMPLETE +with `action outcome only`, +failure conservatively as TOOL_ERROR with unknown findings count (its original +failing step remains blocking), and skipped/cancelled as INCOMPLETE. The summary +does not pretend an action failure proves a vulnerability. +`make security-semgrep SCAN_OUTCOME=success` retains exit-3 INCOMPLETE evidence +without failing the collector; it tolerates no tool-error exit. The pinned action +remains the blocking scan. Collectors and artifact uploads run with `if: always()`. + +Fail-closed network verification trades availability for evidence: advisory database +or jsDelivr outages can delay publication. Bounded tool timeouts and curl retries +limit hanging; never bypass the gate. Retry an unchanged tag via +`gh workflow run publish-pypi.yml --ref vX.Y.Z`; changed code must fix forward to a +new release candidate. No network checks run in default pytest. + +### Evidence and redaction + +`security-results/**/summary.json` (or `security-results/summary.json`) uses schema +version 1 with check name, scope, status, tool exit status, counts, suppression +counts, applied exception identifiers, tool versions, and allow-listed finding +metadata. Markdown counts/status are appended to `$GITHUB_STEP_SUMMARY`. SHA-pinned +artifact uploads run with `if: always()` and fail if evidence is missing. Each job +or matrix cell has a unique artifact name. Generated local evidence is gitignored. + +Raw scanner stdout/stderr, code snippets, advisory descriptions, exception reasons, +HTTP headers, connection strings and downloaded assets are **not** uploaded or +printed by the helpers. Only rule/advisory IDs, relative file paths, line numbers, +classification and package identifiers survive input parsing. Invalid diagnostics +are replaced by fixed messages, never echoed. Redaction regression tests use +credential-bearing canaries. Evidence is a code/config review finding until runtime +impact is independently demonstrated; a clean scan is not proof of absence. + +### Examples and deployment boundaries + +Examples are local-only demonstrations unless explicitly described as ephemeral +certification infrastructure. OpenAPI security declarations do not enforce runtime +authorization; the report bearer check and partner purge key-presence check are +placeholders, and webhook HMAC is optional when its environment secret is unset. +The demo static server binds loopback only. The e2e Function App is +intentionally anonymously reachable for CI warmup/tests/screenshots and holds only +synthetic data; it is not a production template. + +Production deployments need explicit authentication, restricted network ingress, +HTTPS/minimum TLS, private blob access, managed identities and appropriate storage +RBAC. `infra/main.bicep` already requires HTTPS, TLS 1.2 for storage and disables +public blobs. It retains shared-key connection-string deployment/storage and public +network reachability for the existing Flex Consumption certification flow. Disabling +shared keys, replacing storage authentication with managed identity, private endpoints, +or changing Function App/SCM minimum TLS needs real-Azure validation before adoption; +these are follow-up review recommendations, not confirmed exploitable vulnerabilities. diff --git a/demo/README.md b/demo/README.md index 47cdce92..416efd78 100644 --- a/demo/README.md +++ b/demo/README.md @@ -1,5 +1,10 @@ # Demo Assets +These generators and the loopback-only preview server are local screenshot tooling, +not production web servers. Use synthetic inputs only; generated HTML/spec previews +must not contain credentials or customer payloads. The examples they invoke use +local-only anonymous defaults (see `examples/README.md`). + The README demo assets are generated from code: - `run_webhook_receiver_example.py` executes the representative `webhook_receiver` example flow and generates the OpenAPI and Swagger UI artifacts used by the README. diff --git a/docs/contributing.md b/docs/contributing.md index 4306e181..b14d8756 100644 --- a/docs/contributing.md +++ b/docs/contributing.md @@ -135,7 +135,7 @@ See [Development Guide](development.md#makefile-targets) for the complete list. | --- | --- | | Ruff | Formatter, linter, and import sorter | | mypy | Static type checker | -| Bandit | Security scanner (`src/` only) | +| Bandit | Local `make security` scans `src/`; CI additionally scans examples, demo, scripts and tools with fail-closed evidence (see root `SECURITY.md`) | Pinned versions live in `.pre-commit-config.yaml` (hook `rev`) and `pyproject.toml` (the `dev` dependency pins), so they are not duplicated here. diff --git a/examples/README.md b/examples/README.md index f3534ec7..6a0961df 100644 --- a/examples/README.md +++ b/examples/README.md @@ -1,5 +1,18 @@ # Examples +## Local-only security defaults + +These runnable examples use anonymous routes for local demonstration. Their OpenAPI +security declarations are documentation, not an authentication gateway. Do not deploy +unchanged to production: configure real authentication, network restrictions and secrets +through your deployment environment. In particular, webhook signature checking is optional +when `WEBHOOK_SECRET` is unset, example bearer checks are not a production identity provider, +and the partner purge example checks only that `X-API-Key` is present, not its identity. + +`e2e_app/` is the exception to local-only hosting: CI deploys it temporarily to Azure +for certification with synthetic data, anonymous warmup/test/docs routes, and resource-group +cleanup. Keep its public reachability for certification, not as a production default. + `azure-functions-openapi` ships five smoke-tested examples: | Role | Path | Description | diff --git a/pyproject.toml b/pyproject.toml index 28ce82cd..0417ba40 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -104,7 +104,7 @@ features = ["dev", "docs"] format = "ruff format src tests" format-check = "ruff format --check src tests" style = "ruff check src tests && ruff format --check src tests" -typecheck = "mypy src tests" +typecheck = "mypy src tests && make security-types" lint = "hatch run style && hatch run typecheck" test = "pytest -v tests" cov = "pytest --cov-report=term-missing --cov-report=html --cov-report=xml tests/" diff --git a/scripts/check_screenshots.py b/scripts/check_screenshots.py index da3fb5c2..12089381 100644 --- a/scripts/check_screenshots.py +++ b/scripts/check_screenshots.py @@ -29,8 +29,8 @@ import argparse import hashlib -import re from pathlib import Path +import re import sys from typing import Any diff --git a/security/bandit-policy.json b/security/bandit-policy.json new file mode 100644 index 00000000..86e1b2d4 --- /dev/null +++ b/security/bandit-policy.json @@ -0,0 +1,19 @@ +{ + "schema_version": 1, + "inline_suppressions_allowed": 0, + "exceptions": [ + {"id": "B105", "file": "scripts/lint_mermaid.py", "line": 22, "kind": "reviewed-static", "reason": "BAD_TOKEN is the literal backslash-n marker used by the Mermaid linter, not a password.", "owner": "yeongseon", "added": "2026-10-22", "expires": "2027-01-19"}, + {"id": "B404", "file": "tools/check_pr_format.py", "line": 5, "kind": "reviewed-static", "reason": "Contributor formatter requires subprocess; both call sites are separately reviewed below.", "owner": "yeongseon", "added": "2026-10-22", "expires": "2027-01-19"}, + {"id": "B607", "file": "tools/check_pr_format.py", "line": 10, "kind": "reviewed-static", "reason": "git resolves from the contributor/CI toolchain PATH; no shell is invoked.", "owner": "yeongseon", "added": "2026-10-22", "expires": "2027-01-19"}, + {"id": "B603", "file": "tools/check_pr_format.py", "line": 10, "kind": "reviewed-static", "reason": "git diff receives a revision range as one argument and terminates options with --; shell=False.", "owner": "yeongseon", "added": "2026-10-22", "expires": "2027-01-19"}, + {"id": "B607", "file": "tools/check_pr_format.py", "line": 37, "kind": "reviewed-static", "reason": "ruff resolves from the contributor/CI toolchain PATH; no shell is invoked.", "owner": "yeongseon", "added": "2026-10-22", "expires": "2027-01-19"}, + {"id": "B603", "file": "tools/check_pr_format.py", "line": 37, "kind": "reviewed-static", "reason": "ruff receives filenames after -- as separate arguments; shell=False.", "owner": "yeongseon", "added": "2026-10-22", "expires": "2027-01-19"}, + {"id": "B404", "file": "tools/browser_integrity.py", "line": 10, "kind": "reviewed-static", "reason": "Invokes only bounded curl for pinned public CDN assets; no shell.", "owner": "yeongseon", "added": "2026-10-22", "expires": "2027-01-19"}, + {"id": "B603", "file": "tools/browser_integrity.py", "line": 37, "kind": "reviewed-static", "reason": "Absolute curl executable, fixed HTTPS URL from wheel constants, shell=False, bounded timeouts.", "owner": "yeongseon", "added": "2026-10-22", "expires": "2027-01-19"}, + {"id": "B404", "file": "tools/dependency_audit_policy.py", "line": 10, "kind": "reviewed-static", "reason": "Runs pip-audit via the current Python interpreter without a shell.", "owner": "yeongseon", "added": "2026-10-22", "expires": "2027-01-19"}, + {"id": "B603", "file": "tools/dependency_audit_policy.py", "line": 125, "kind": "reviewed-static", "reason": "Interpreter is sys.executable; verified frozen inventory is a separate --requirement file argument; shell=False.", "owner": "yeongseon", "added": "2026-10-22", "expires": "2027-01-19"}, + {"id": "B404", "file": "tools/security_evidence.py", "line": 10, "kind": "reviewed-static", "reason": "Runs Bandit via the current Python interpreter without a shell.", "owner": "yeongseon", "added": "2026-10-22", "expires": "2027-01-19"}, + {"id": "B603", "file": "tools/security_evidence.py", "line": 132, "kind": "reviewed-static", "reason": "sys.executable runs a fixed Bandit command and repository-controlled scope; shell=False.", "owner": "yeongseon", "added": "2026-10-22", "expires": "2027-01-19"}, + {"id": "B105", "file": "tools/security_report.py", "line": 59, "kind": "reviewed-static", "reason": "TOOL_ERROR is a public status enum value, not a hardcoded credential.", "owner": "yeongseon", "added": "2026-10-22", "expires": "2027-01-19"} + ] +} diff --git a/security/dependency-audit-policy.json b/security/dependency-audit-policy.json new file mode 100644 index 00000000..edf5af53 --- /dev/null +++ b/security/dependency-audit-policy.json @@ -0,0 +1,5 @@ +{ + "schema_version": 1, + "blocking": "all-known-runtime", + "exceptions": [] +} diff --git a/tests/test_browser_integrity_evidence.py b/tests/test_browser_integrity_evidence.py new file mode 100644 index 00000000..da666570 --- /dev/null +++ b/tests/test_browser_integrity_evidence.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import base64 +import hashlib +from html.parser import HTMLParser + +import pytest + +from azure_functions_openapi import swagger_ui +from tools.browser_integrity import compare_asset +from tools.security_report import Status + + +class AssetParser(HTMLParser): + def __init__(self) -> None: + super().__init__() + self.assets: list[dict[str, str | None]] = [] + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + values = dict(attrs) + if tag in {"link", "script"} and "integrity" in values: + self.assets.append(values) + + +def test_pinned_browser_assets_have_valid_integrity_and_crossorigin() -> None: + # Given imported production constants; When HTML is rendered; Then both assets enforce SRI. + parser = AssetParser() + parser.feed(swagger_ui.render_swagger_ui().get_body().decode()) + assert swagger_ui._SWAGGER_UI_CDN_BASE == ( + "https://cdn.jsdelivr.net/npm/swagger-ui-dist@" + swagger_ui._SWAGGER_UI_DIST_VERSION + ) + assert len(parser.assets) == 2 + for asset, digest in zip( + parser.assets, + [swagger_ui._SWAGGER_UI_CSS_SRI, swagger_ui._SWAGGER_UI_BUNDLE_SRI], + strict=True, + ): + assert digest.startswith("sha384-") + assert len(base64.b64decode(digest.removeprefix("sha384-"), validate=True)) == 48 + assert asset["integrity"] == digest + assert asset["crossorigin"] == "anonymous" + assert (asset.get("src") or asset.get("href") or "").startswith( + swagger_ui._SWAGGER_UI_CDN_BASE + "/" + ) + + +@pytest.mark.parametrize( + "body,expected", + [(b"asset", Status.PASS), (b"modified", Status.FAIL), (None, Status.TOOL_ERROR)], +) +def test_asset_comparison_distinguishes_mismatch_from_outage( + body: bytes | None, expected: Status +) -> None: + # Given a pinned digest; When bytes or a network failure arrive; Then classify distinctly. + digest = "sha384-" + base64.b64encode(hashlib.sha384(b"asset").digest()).decode() + assert compare_asset(body, digest) == expected diff --git a/tests/test_candidate_inventory.py b/tests/test_candidate_inventory.py new file mode 100644 index 00000000..a0749d6b --- /dev/null +++ b/tests/test_candidate_inventory.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import json +from pathlib import Path +import subprocess +from unittest.mock import Mock +from zipfile import ZipFile + +import pytest + +from tools.dependency_audit_policy import run_audit +from tools.security_report import JSON, Status + + +def candidate_environment(root: Path) -> tuple[Path, Path, Path]: + site = root / "site" + for name, version in (("unpublished-probe", "9876.1"), ("idna", "3.11")): + info = site / f"{name.replace('-', '_')}-{version}.dist-info" + info.mkdir(parents=True) + (info / "METADATA").write_text(f"Metadata-Version: 2.1\nName: {name}\nVersion: {version}\n") + wheel = root / "probe.whl" + with ZipFile(wheel, "w") as archive: + archive.writestr( + "unpublished_probe-9876.1.dist-info/METADATA", + "Metadata-Version: 2.1\nName: unpublished-probe\nVersion: 9876.1\n", + ) + policy = root / "policy.json" + policy.write_text('{"schema_version":1,"blocking":"all-known-runtime","exceptions":[]}') + return site, wheel, policy + + +@pytest.mark.parametrize( + "scenario,expected", + [ + ("clean", Status.PASS), + ("advisory", Status.FAIL), + ("skip", Status.TOOL_ERROR), + ("omitted", Status.TOOL_ERROR), + ("candidate-mismatch", Status.TOOL_ERROR), + ], +) +def test_unpublished_candidate_audits_exact_third_party_inventory( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, scenario: str, expected: Status +) -> None: + site, wheel, policy = candidate_environment(tmp_path) + + def audit(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: + requirements = Path(command[command.index("--requirement") + 1]).read_text() + assert requirements == "idna==3.11\n" + assert "--no-deps" in command and "--disable-pip" in command + dependency: dict[str, JSON] = {"name": "idna", "version": "3.11", "vulns": []} + dependencies = [dependency] + if scenario == "skip": + dependencies = [{"name": "idna", "skip_reason": "unavailable"}] + elif scenario == "advisory": + dependency["vulns"] = [{"id": "CVE-2026-12345", "fix_versions": []}] + elif scenario == "omitted": + dependencies = [{"name": "other", "version": "1", "vulns": []}] + elif scenario == "candidate-mismatch": + dependencies.append({"name": "unpublished-probe", "version": "9876.2", "vulns": []}) + return subprocess.CompletedProcess( + command, int(scenario == "advisory"), json.dumps({"dependencies": dependencies}) + ) + + monkeypatch.setattr(subprocess, "run", audit) + monkeypatch.setattr("tools.dependency_audit_policy.version", lambda _: "2.10.1") + assert run_audit(site, policy, wheel).status == expected + + +@pytest.mark.parametrize("condition", ["mismatch", "absent", "empty"]) +def test_unverified_or_empty_candidate_inventory_never_passes( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, condition: str +) -> None: + site, wheel, policy = candidate_environment(tmp_path) + if condition == "mismatch": + (site / "unpublished_probe-9876.1.dist-info/METADATA").write_text( + "Name: unpublished-probe\nVersion: 9876.2\n" + ) + elif condition == "absent": + (site / "unpublished_probe-9876.1.dist-info/METADATA").unlink() + else: + (site / "idna-3.11.dist-info/METADATA").unlink() + scanner = Mock() + monkeypatch.setattr(subprocess, "run", scanner) + assert run_audit(site, policy, wheel).status == Status.INCOMPLETE + scanner.assert_not_called() diff --git a/tests/test_dependency_audit_policy.py b/tests/test_dependency_audit_policy.py new file mode 100644 index 00000000..5fa4d3bf --- /dev/null +++ b/tests/test_dependency_audit_policy.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +from datetime import date +import json + +import pytest + +from tools.dependency_audit_policy import classify_audit +from tools.security_report import Status + +TODAY = date(2026, 10, 22) +POLICY = '{"schema_version": 1, "blocking": "all-known-runtime", "exceptions": []}' + + +def payload(*, vulnerable: bool = False) -> str: + return json.dumps( + { + "dependencies": [ + { + "name": "pydantic", + "version": "2.0", + "vulns": [{"id": "CVE-2026-12345", "fix_versions": []}] if vulnerable else [], + } + ] + } + ) + + +@pytest.mark.parametrize("vulnerable,expected", [(False, Status.PASS), (True, Status.FAIL)]) +def test_audit_blocks_any_known_runtime_vulnerability(vulnerable: bool, expected: Status) -> None: + # Given resolved runtime dependencies; When evaluated; Then all advisories block. + assert ( + classify_audit( + payload(vulnerable=vulnerable), 1 if vulnerable else 0, policy_raw=POLICY, today=TODAY + ).status + == expected + ) + + +@pytest.mark.parametrize( + "raw", + [ + "", + "invalid", + "{}", + '{"dependencies": []}', + '{"dependencies": [{"name": "x", "skip_reason": "offline"}]}', + ], +) +def test_audit_tool_errors_never_become_clean_or_vulnerable(raw: str) -> None: + assert classify_audit(raw, 1, policy_raw=POLICY, today=TODAY).status == Status.TOOL_ERROR + + +@pytest.mark.parametrize( + "expires,expected", [("2026-10-21", Status.FAIL), ("2026-11-21", Status.PASS)] +) +def test_exception_expiry_is_enforced(expires: str, expected: Status) -> None: + # Given an owned, justified exception; When audited; Then only a current exception applies. + policy = json.loads(POLICY) + policy["exceptions"] = [ + { + "id": "CVE-2026-12345", + "package": "pydantic", + "reason": "no patched release available", + "owner": "maintainer", + "kind": "fix-not-available", + "added": "2026-10-01", + "expires": expires, + } + ] + result = classify_audit(payload(vulnerable=True), 1, policy_raw=json.dumps(policy), today=TODAY) + assert result.status == expected + assert bool(result.exceptions_applied) == (expected == Status.PASS) + + +def test_unused_exception_is_reported() -> None: + policy = json.loads(POLICY) + policy["exceptions"] = [ + { + "id": "CVE-2026-12345", + "package": "pydantic", + "reason": "reviewed unreachable code", + "owner": "maintainer", + "kind": "not-affected", + "added": "2026-10-01", + "expires": "2026-11-01", + } + ] + result = classify_audit(payload(), 0, policy_raw=json.dumps(policy), today=TODAY) + assert result.status == Status.PASS + assert result.counts["stale_exceptions"] == 1 diff --git a/tests/test_security_evidence.py b/tests/test_security_evidence.py new file mode 100644 index 00000000..a61d7939 --- /dev/null +++ b/tests/test_security_evidence.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from tools.security_evidence import SCOPE, classify_bandit, run_bandit +from tools.security_report import Check, Status, write_report + + +def bandit_payload() -> str: + return json.dumps( + { + "errors": [], + "metrics": { + "_totals": {"loc": 10, "nosec": 0, "skipped_tests": 0}, + "src/example.py": {"loc": 10}, + }, + "results": [], + } + ) + + +@pytest.mark.parametrize( + ("raw", "exit_status", "expected"), + [ + ("", 2, Status.TOOL_ERROR), + ("invalid", 0, Status.TOOL_ERROR), + ('{"errors": [], "metrics": {"_totals": {}}, "results": []}', 0, Status.INCOMPLETE), + ], +) +def test_bandit_fails_closed_when_evidence_is_missing( + raw: str, exit_status: int, expected: Status +) -> None: + # Given incomplete scanner output; When classified; Then it cannot pass. + assert classify_bandit(raw, exit_status, floor=1).status == expected + + +def test_bandit_passes_when_scan_is_complete() -> None: + # Given one scanned file; When classified; Then counts reflect the scan. + result = classify_bandit(bandit_payload(), 0, floor=1) + assert result.status == Status.PASS + assert result.counts["files"] == 1 + assert result.counts["loc"] == 10 + + +def write_scan_surfaces(root: Path, file_count: int) -> None: + for index in range(file_count): + surface = root / SCOPE[index % len(SCOPE)] + surface.mkdir(exist_ok=True) + (surface / f"module_{index:03d}.py").write_text(f"VALUE = {index}\n", encoding="utf-8") + policy = root / "security" / "bandit-policy.json" + policy.parent.mkdir() + policy.write_text('{"exceptions": []}\n', encoding="utf-8") + + +@pytest.mark.parametrize( + "file_count", + [ + pytest.param(49, id="below-bandit-progress-threshold"), + pytest.param(51, id="above-bandit-progress-threshold"), + ], +) +def test_real_bandit_scan_passes_on_either_side_of_the_progress_threshold( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, file_count: int +) -> None: + # Given harmless files on every declared surface, on either side of the 50-file + # threshold above which Bandit prints a "Working..." progress bar before its JSON; + # When the real scanner is run through the evidence collector; + # Then its JSON still parses and the scan passes instead of reporting TOOL_ERROR. + write_scan_surfaces(tmp_path, file_count) + monkeypatch.chdir(tmp_path) + result = run_bandit() + assert result.status == Status.PASS + assert result.counts["files"] == file_count + + +@pytest.mark.parametrize("field", ["nosec", "skipped_tests"]) +def test_bandit_rejects_unapproved_suppressions(field: str) -> None: + # Given an unapproved suppression; When classified; Then scan is incomplete. + data = json.loads(bandit_payload()) + data["metrics"]["_totals"][field] = 1 + assert classify_bandit(json.dumps(data), 0, floor=1).status == Status.INCOMPLETE + + +def test_bandit_rejects_errors_and_crashes() -> None: + # Given valid JSON with an error; When classified; Then tool failure is distinct. + data = json.loads(bandit_payload()) + data["errors"] = [{"filename": "src/broken.py", "reason": "secret"}] + assert classify_bandit(json.dumps(data), 0, floor=1).status == Status.TOOL_ERROR + assert classify_bandit(bandit_payload(), 2, floor=1).status == Status.TOOL_ERROR + + +@pytest.mark.parametrize("severity", ["LOW", "MEDIUM", "HIGH"]) +def test_bandit_reports_findings_without_source_or_secret(severity: str) -> None: + # Given a finding containing sensitive source; When classified; Then only metadata survives. + data = json.loads(bandit_payload()) + data["results"] = [ + { + "test_id": "B105", + "filename": "src/example.py", + "line_number": 3, + "issue_severity": severity, + "issue_confidence": "HIGH", + "code": "Authorization: Bearer secret", + "issue_text": "AccountKey=secret", + } + ] + result = classify_bandit(json.dumps(data), 1, floor=1) + assert result.status == Status.FAIL + assert result.counts[f"severity.{severity}"] == 1 + assert result.counts["confidence.HIGH"] == 1 + + +def test_summary_has_versioned_schema_and_redacts_raw_diagnostics(tmp_path: Path) -> None: + # Given secret-bearing scanner diagnostics; When rendered; Then neither report leaks them. + data = json.loads(bandit_payload()) + data["errors"] = ["Authorization: Bearer canary;AccountKey=canary"] + check = classify_bandit(json.dumps(data), 2, floor=1) + summary = tmp_path / "job-summary.md" + write_report([check], tmp_path / "results", summary) + saved = (tmp_path / "results" / "summary.json").read_text() + assert json.loads(saved)["schema_version"] == 1 + assert json.loads(saved)["checks"][0]["status"] == "TOOL_ERROR" + assert "canary" not in saved + summary.read_text() + assert set(json.loads(saved)["checks"][0]) >= { + "name", + "scope", + "status", + "exit_status", + "counts", + "suppressions", + "exceptions_applied", + "tool_versions", + } + + +@pytest.mark.parametrize( + "status,code", + [(Status.PASS, 0), (Status.FAIL, 1), (Status.TOOL_ERROR, 2), (Status.INCOMPLETE, 3)], +) +def test_status_exit_codes_are_distinct(status: Status, code: int) -> None: + assert Check("test", (), status, 0).returncode == code diff --git a/tests/test_security_policy_boundaries.py b/tests/test_security_policy_boundaries.py new file mode 100644 index 00000000..da9df2f6 --- /dev/null +++ b/tests/test_security_policy_boundaries.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +from datetime import date +import json +from pathlib import Path + +import pytest + +from tools.dependency_audit_policy import classify_audit +from tools.security_exceptions import parse_exceptions +from tools.security_report import ( + Check, + EvidenceError, + Status, + identifier, + integer, + load, + sequence, + text, + write_report, +) + + +@pytest.mark.parametrize("raw", ["null", "[]", '"text"', "true"]) +def test_json_boundary_requires_object(raw: str) -> None: + with pytest.raises(EvidenceError): + load(raw) + + +@pytest.mark.parametrize("value", [None, 0, True, ""]) +def test_text_boundary_rejects_nonempty_string_violations(value: None | int | str) -> None: + with pytest.raises(EvidenceError): + text(value) + + +@pytest.mark.parametrize("value", [None, True, -1, "1"]) +def test_count_boundary_rejects_invalid_counts(value: None | int | str) -> None: + with pytest.raises(EvidenceError): + integer(value) + + +@pytest.mark.parametrize("value", [None, "[]", 0]) +def test_array_boundary_rejects_nonarrays(value: None | str | int) -> None: + with pytest.raises(EvidenceError): + sequence(value) + + +@pytest.mark.parametrize( + "value", + [ + "/tmp/file.py", + "../secret", + "https://example.com", + "Authorization: Bearer canary", + "file.py?code=canary", + ], +) +def test_identifiers_reject_sensitive_or_external_metadata(value: str) -> None: + with pytest.raises(EvidenceError): + identifier(value) + + +@pytest.mark.parametrize( + "patch", + [ + {"expires": "2027-10-22"}, + {"added": "2026-11-01"}, + {"expires": "invalid"}, + {"kind": "ignore-all"}, + {"line": -1}, + {"line": True}, + {"owner": ""}, + {"reason": ""}, + ], +) +def test_exception_boundary_rejects_invalid_policy(patch: dict[str, str | int]) -> None: + rule = { + "id": "CVE-2026-12345", + "package": "example", + "kind": "not-affected", + "reason": "reviewed unreachable path", + "owner": "maintainer", + "added": "2026-10-01", + "expires": "2026-11-01", + **patch, + } + parsed = load(json.dumps({"exceptions": [rule]})) + with pytest.raises((EvidenceError, ValueError)): + parse_exceptions(parsed["exceptions"], date(2026, 10, 22)) + + +def test_duplicate_exception_is_rejected() -> None: + rule = { + "id": "CVE-2026-12345", + "package": "example", + "kind": "not-affected", + "reason": "reviewed unreachable path", + "owner": "maintainer", + "added": "2026-10-01", + "expires": "2026-11-01", + } + parsed = load(json.dumps({"exceptions": [rule, rule]})) + with pytest.raises(EvidenceError): + parse_exceptions(parsed["exceptions"], date(2026, 10, 22)) + + +@pytest.mark.parametrize( + "policy", + [ + "{}", + "invalid", + '{"schema_version":2,"blocking":"all-known-runtime","exceptions":[]}', + '{"schema_version":1,"blocking":"high-only","exceptions":[]}', + ], +) +def test_audit_rejects_unsupported_policy(policy: str) -> None: + assert ( + classify_audit( + '{"dependencies":[{"name":"example","version":"1","vulns":[]}]}', + 0, + policy_raw=policy, + today=date(2026, 10, 22), + ).status + == Status.TOOL_ERROR + ) + + +@pytest.mark.parametrize("exit_status", [0, 2]) +def test_audit_rejects_exit_status_inconsistent_with_findings(exit_status: int) -> None: + raw = '{"dependencies":[{"name":"example","version":"1","vulns":[{"id":"CVE-1"}]}]}' + policy = '{"schema_version":1,"blocking":"all-known-runtime","exceptions":[]}' + assert ( + classify_audit(raw, exit_status, policy_raw=policy, today=date(2026, 10, 22)).status + == Status.TOOL_ERROR + ) + + +def test_fix_not_available_exception_stops_applying_when_patch_exists() -> None: + policy = { + "schema_version": 1, + "blocking": "all-known-runtime", + "exceptions": [ + { + "id": "CVE-1", + "package": "example", + "kind": "fix-not-available", + "reason": "no fix", + "owner": "maintainer", + "added": "2026-10-01", + "expires": "2026-11-01", + } + ], + } + raw = ( + '{"dependencies":[{"name":"example","version":"1",' + '"vulns":[{"id":"CVE-1","fix_versions":["2"]}]}]}' + ) + assert ( + classify_audit(raw, 1, policy_raw=json.dumps(policy), today=date(2026, 10, 22)).status + == Status.FAIL + ) + + +def test_summary_appends_to_github_summary(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + summary = tmp_path / "summary.md" + summary.write_text("existing\n") + monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(summary)) + assert write_report([Check("scanner", (), Status.PASS, 0)], tmp_path / "output") == 0 + assert summary.read_text().startswith("existing\n") + + +def test_skipped_dependency_is_tool_error_even_with_version() -> None: + raw = ( + '{"dependencies":[{"name":"example","version":"1",' + '"skip_reason":"index unavailable","vulns":[]}]}' + ) + policy = '{"schema_version":1,"blocking":"all-known-runtime","exceptions":[]}' + assert ( + classify_audit(raw, 0, policy_raw=policy, today=date(2026, 10, 22)).status + == Status.TOOL_ERROR + ) diff --git a/tests/test_security_tool_execution.py b/tests/test_security_tool_execution.py new file mode 100644 index 00000000..67f14720 --- /dev/null +++ b/tests/test_security_tool_execution.py @@ -0,0 +1,268 @@ +from __future__ import annotations + +from datetime import date +import json +from pathlib import Path +import subprocess +import sys + +import pytest + +from tools import browser_integrity, dependency_audit_policy, security_evidence +from tools.security_evidence import apply_bandit_policy, classify_bandit, semgrep_evidence +from tools.security_report import Check, Status + + +def scanner_result(raw: str, code: int = 0) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess(["scanner"], code, stdout=raw, stderr="secret") + + +def test_bandit_runner_checks_inventory_and_preserves_version( + monkeypatch: pytest.MonkeyPatch, +) -> None: + metrics = { + p.as_posix(): {"loc": 1} + for surface in security_evidence.SCOPE + for p in Path(surface).rglob("*.py") + } + metrics["_totals"] = {"loc": len(metrics)} + raw = json.dumps({"errors": [], "metrics": metrics, "results": []}) + + def complete(*args: object, **kwargs: object) -> subprocess.CompletedProcess[str]: + return scanner_result(raw) + + monkeypatch.setattr(subprocess, "run", complete) + result = security_evidence.run_bandit() + assert result.status == Status.PASS + assert result.tool_versions["bandit"] == "1.9.4" + del metrics[next(iter(metrics))] + raw = json.dumps({"errors": [], "metrics": metrics, "results": []}) + assert security_evidence.run_bandit().status == Status.INCOMPLETE + + +def test_audit_runner_uses_only_target_path( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + policy = tmp_path / "policy.json" + policy.write_text('{"schema_version":1,"blocking":"all-known-runtime","exceptions":[]}') + from tools.candidate_inventory import Inventory, Package + + monkeypatch.setattr( + dependency_audit_policy, + "verified_inventory", + lambda path, wheel: Inventory(Package("candidate", "1"), (Package("example", "1"),)), + ) + + def complete(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: + assert Path(command[command.index("--requirement") + 1]).read_text() == "example==1\n" + return scanner_result('{"dependencies":[{"name":"example","version":"1","vulns":[]}]}') + + monkeypatch.setattr(subprocess, "run", complete) + monkeypatch.setattr(dependency_audit_policy, "version", lambda name: "2.10.1") + assert ( + dependency_audit_policy.run_audit(tmp_path, policy, tmp_path / "candidate.whl").status + == Status.PASS + ) + + +def test_browser_network_exit_is_not_a_hash_mismatch(monkeypatch: pytest.MonkeyPatch) -> None: + def failed(*args: object, **kwargs: object) -> subprocess.CompletedProcess[bytes]: + return subprocess.CompletedProcess(["curl"], 22, stdout=b"error page", stderr=b"secret") + + monkeypatch.setattr(subprocess, "run", failed) + assert all(c.status == Status.TOOL_ERROR for c in browser_integrity.verify_assets()) + + +def test_browser_success_with_wrong_bytes_fails_integrity(monkeypatch: pytest.MonkeyPatch) -> None: + def complete(*args: object, **kwargs: object) -> subprocess.CompletedProcess[bytes]: + return subprocess.CompletedProcess(["curl"], 0, stdout=b"modified", stderr=b"") + + monkeypatch.setattr(subprocess, "run", complete) + assert all(c.status == Status.FAIL for c in browser_integrity.verify_assets()) + + +@pytest.mark.parametrize("development", [False, True]) +@pytest.mark.parametrize( + "status,code", [(Status.FAIL, 1), (Status.TOOL_ERROR, 2), (Status.INCOMPLETE, 3)] +) +def test_audit_cli_preserves_failure_codes_in_both_lanes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, development: bool, status: Status, code: int +) -> None: + args = [ + "audit", + "--site-packages", + str(tmp_path), + "--output", + str(tmp_path), + "--candidate-wheel", + str(tmp_path / "candidate.whl"), + ] + if development: + args.append("--development") + monkeypatch.setattr(sys, "argv", args) + monkeypatch.setattr( + dependency_audit_policy, + "run_audit", + lambda path, policy, wheel: Check("pip-audit", (), status, 1), + ) + assert dependency_audit_policy.main() == code + assert json.loads((tmp_path / "summary.json").read_text())["checks"][0]["status"] == status + + +def test_browser_cli_writes_evidence(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.setattr(sys, "argv", ["sri", "--output", str(tmp_path)]) + monkeypatch.setattr( + browser_integrity, "verify_assets", lambda: [Check("sri", (), Status.FAIL, 0)] + ) + assert browser_integrity.main() == 1 + + +def test_bandit_cli_writes_classified_failure( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setattr(sys, "argv", ["security", "bandit", "--output", str(tmp_path)]) + monkeypatch.setattr( + security_evidence, "run_bandit", lambda: Check("bandit", (), Status.INCOMPLETE, 0) + ) + assert security_evidence.main() == 3 + + +@pytest.mark.parametrize( + "severity,exit_code,expected", + [ + ("INVALID", 1, Status.TOOL_ERROR), + ("LOW", 0, Status.TOOL_ERROR), + ], +) +def test_bandit_rejects_invalid_classification_or_inconsistent_exit( + severity: str, exit_code: int, expected: Status +) -> None: + raw = json.dumps( + { + "errors": [], + "metrics": {"_totals": {"loc": 1}, "src/x.py": {}}, + "results": [ + { + "test_id": "B404", + "filename": "src/x.py", + "line_number": 1, + "issue_severity": severity, + "issue_confidence": "HIGH", + } + ], + } + ) + assert classify_bandit(raw, exit_code, floor=1).status == expected + + +def test_static_unmatched_finding_remains_blocking() -> None: + check = classify_bandit( + json.dumps( + { + "errors": [], + "metrics": {"_totals": {"loc": 1}, "src/x.py": {}}, + "results": [ + { + "test_id": "B404", + "filename": "src/x.py", + "line_number": 1, + "issue_severity": "LOW", + "issue_confidence": "HIGH", + } + ], + } + ), + 1, + floor=1, + ) + result = apply_bandit_policy(check, '{"exceptions":[]}', date(2026, 10, 22)) + assert result.status == Status.FAIL + assert result.counts["blocking"] == 1 + + +@pytest.mark.parametrize( + "outcome,status", + [ + ("success", Status.INCOMPLETE), + ("failure", Status.TOOL_ERROR), + ("skipped", Status.INCOMPLETE), + ], +) +def test_semgrep_records_action_outcome(outcome: str, status: Status) -> None: + assert semgrep_evidence(outcome).status == status + + +def test_evidence_cli_executes_without_package_dependencies(tmp_path: Path) -> None: + result = subprocess.run( + [ + sys.executable, + "-m", + "tools.security_evidence", + "semgrep", + "--outcome", + "success", + "--output", + str(tmp_path), + ], + capture_output=True, + check=False, + ) + assert result.returncode == 3 + assert json.loads((tmp_path / "summary.json").read_text())["checks"][0]["name"] == "semgrep" + + +@pytest.mark.parametrize("failure", [OSError(), subprocess.TimeoutExpired("scanner", 1)]) +def test_external_tool_failures_are_classified( + monkeypatch: pytest.MonkeyPatch, failure: OSError | subprocess.TimeoutExpired, tmp_path: Path +) -> None: + def fail(*args: object, **kwargs: object) -> subprocess.CompletedProcess[str]: + raise failure + + monkeypatch.setattr(subprocess, "run", fail) + assert security_evidence.run_bandit().status == Status.TOOL_ERROR + assert dependency_audit_policy.run_audit(tmp_path, tmp_path).status == Status.INCOMPLETE + assert all(c.status == Status.TOOL_ERROR for c in browser_integrity.verify_assets()) + + +def test_static_exceptions_match_exact_file_line_and_id() -> None: + check = classify_bandit( + json.dumps( + { + "errors": [], + "metrics": {"_totals": {"loc": 1}, "tools/example.py": {}}, + "results": [ + { + "test_id": "B404", + "filename": "tools/example.py", + "line_number": 1, + "issue_severity": "LOW", + "issue_confidence": "HIGH", + } + ], + } + ), + 1, + floor=1, + ) + policy = { + "exceptions": [ + { + "id": "B404", + "file": "tools/example.py", + "line": 1, + "kind": "reviewed-static", + "reason": "fixed shell-free command", + "owner": "maintainer", + "added": "2026-10-01", + "expires": "2026-11-01", + } + ] + } + assert apply_bandit_policy(check, json.dumps(policy), date(2026, 10, 22)).status == Status.PASS + assert apply_bandit_policy(check, json.dumps(policy), date(2026, 11, 2)).status == Status.FAIL + assert apply_bandit_policy(check, "invalid", date(2026, 10, 22)).status == Status.TOOL_ERROR + + +def test_incomplete_static_scan_cannot_be_excepted() -> None: + check = Check("bandit", (), Status.INCOMPLETE, 0) + assert apply_bandit_policy(check, '{"exceptions": []}', date(2026, 10, 22)) == check diff --git a/tests/test_security_workflows.py b/tests/test_security_workflows.py new file mode 100644 index 00000000..906095bd --- /dev/null +++ b/tests/test_security_workflows.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +from pathlib import Path +import subprocess +import sys + +import pytest +import yaml + +from tools.lint_release_workflows import check_pins, check_publish_needs +from tools.security_evidence import SCOPE + +ROOT = Path(__file__).resolve().parents[1] + + +def test_release_security_gate_preserves_all_existing_tiers() -> None: + text = (ROOT / ".github/workflows/publish-pypi.yml").read_text() + jobs = yaml.safe_load(text)["jobs"] + assert check_publish_needs(text, "publish-pypi.yml") == [] + assert "runtime-dependency-audit" in jobs["publish"]["needs"] + gate = jobs["runtime-dependency-audit"] + assert gate["permissions"] == {"contents": "read"} + assert gate["needs"] == "build" + assert gate["strategy"]["matrix"]["interpreter"] == ["3.11", "3.12", "3.13", "3.14"] + assert not gate.get("continue-on-error") + assert all(not step.get("continue-on-error") for step in gate["steps"]) + downloads = [s for s in gate["steps"] if "download-artifact@" in s.get("uses", "")] + assert downloads[0]["with"]["name"] == "dist" + commands = "\n".join(s.get("run", "") for s in gate["steps"]) + assert "pip-audit==2.10.1" in commands + assert "--without-pip" in commands + assert "make security-runtime" in commands + assert 'CANDIDATE_WHEEL="${wheels[0]}"' in commands + assert 'make security-browser SECURITY_PYTHON="$RUNNER_TEMP/candidate/bin/python"' in commands + + +def test_security_workflows_preserve_evidence_even_on_failure() -> None: + for filename in ("security.yml", "dependency-audit.yml", "publish-pypi.yml"): + text = (ROOT / ".github/workflows" / filename).read_text() + workflow = yaml.safe_load(text) + assert workflow["permissions"] == {"contents": "read"} + assert check_pins(text, filename) == [] + for job in workflow["jobs"].values(): + for step in job.get("steps", []): + if "upload-artifact@" in step.get("uses", "") and step.get("with", {}).get( + "path", "" + ).startswith("security-results"): + assert step["if"] == "always()" + assert step["with"]["if-no-files-found"] == "error" + + +def test_security_scan_declares_all_executable_surfaces() -> None: + workflow = yaml.safe_load((ROOT / ".github/workflows/security.yml").read_text()) + assert SCOPE == ("src", "examples", "demo", "scripts", "tools") + commands = [s.get("run", "") for s in workflow["jobs"]["bandit"]["steps"]] + assert "make security-bandit SECURITY_PYTHON=python" in commands + collector = next( + s for s in workflow["jobs"]["bandit"]["steps"] if "make security-bandit" in s.get("run", "") + ) + assert collector["if"] == "always()" + semgrep = workflow["jobs"]["semgrep"]["steps"] + action = next(s for s in semgrep if "semgrep/semgrep-action@" in s.get("uses", "")) + assert action["with"]["config"] == "p/python" + assert not action.get("continue-on-error") + + +def test_scheduled_dependency_lane_uses_shared_reporting_policy() -> None: + workflow = yaml.safe_load((ROOT / ".github/workflows/dependency-audit.yml").read_text()) + job = workflow["jobs"]["pip-audit"] + assert job["continue-on-error"] is True + commands = "\n".join(s.get("run", "") for s in job["steps"]) + assert "make security-runtime" in commands + assert "make security-development" in commands + assert "make security-browser" in commands + + +@pytest.mark.parametrize("outcome,success", [("success", True), ("failure", False)]) +def test_semgrep_make_collector_tolerates_only_incomplete( + tmp_path: Path, outcome: str, success: bool +) -> None: + result = subprocess.run( + [ + "make", + "security-semgrep", + f"SECURITY_PYTHON={sys.executable}", + f"SCAN_OUTCOME={outcome}", + f"SECURITY_OUTPUT={tmp_path}", + ], + cwd=ROOT, + capture_output=True, + text=True, + ) + assert (result.returncode == 0) is success + assert (tmp_path / "summary.json").exists() + + +def test_local_security_includes_expanded_inventory_scan() -> None: + result = subprocess.run(["make", "-n", "security"], cwd=ROOT, capture_output=True, text=True) + assert result.returncode == 0 + assert "tools.security_evidence bandit" in result.stdout diff --git a/tests/test_validate_http_shorthand_conflict.py b/tests/test_validate_http_shorthand_conflict.py index 44827b8e..3b7795a2 100644 --- a/tests/test_validate_http_shorthand_conflict.py +++ b/tests/test_validate_http_shorthand_conflict.py @@ -14,6 +14,7 @@ from __future__ import annotations +from collections.abc import Callable import importlib from typing import Any @@ -34,6 +35,14 @@ except ImportError: HAS_VALIDATION = False + +def _validator() -> Callable[..., Callable[[Callable[..., Any]], Callable[..., Any]]]: + decorator: Callable[..., Callable[[Callable[..., Any]], Callable[..., Any]]] = ( + importlib.import_module("azure_functions_validation").validate_http + ) + return decorator + + pytestmark = pytest.mark.skipif( not HAS_VALIDATION, reason="azure-functions-validation not installed", @@ -63,7 +72,7 @@ class _OtherValidated(BaseModel): "openapi_above_route", [True, False], ids=["openapi-above", "openapi-below"] ) def test_conflicting_request_model_shorthand_raises(openapi_above_route: bool) -> None: - from azure_functions_validation import validate_http + validate_http = _validator() app = func.FunctionApp() @@ -71,14 +80,14 @@ def test_conflicting_request_model_shorthand_raises(openapi_above_route: bool) - @openapi(requests=_Authored) @app.route(route="orders", methods=["POST"]) - @validate_http(body=_Validated) # type: ignore[untyped-decorator] + @validate_http(body=_Validated) def orders(req: Any, body: Any) -> Any: return body else: @app.route(route="orders", methods=["POST"]) @openapi(requests=_Authored) - @validate_http(body=_Validated) # type: ignore[untyped-decorator] + @validate_http(body=_Validated) def orders(req: Any, body: Any) -> Any: # noqa: F811 return body @@ -87,13 +96,13 @@ def orders(req: Any, body: Any) -> Any: # noqa: F811 def test_conflicting_response_model_shorthand_raises() -> None: - from azure_functions_validation import validate_http + validate_http = _validator() app = func.FunctionApp() @openapi(responses=_Authored) @app.route(route="orders", methods=["GET"]) - @validate_http(response_model=_Validated) # type: ignore[untyped-decorator] + @validate_http(response_model=_Validated) def orders(req: Any) -> Any: ... with pytest.raises(OpenAPISpecConfigError, match="Conflicting status 200"): @@ -101,13 +110,13 @@ def orders(req: Any) -> Any: ... def test_request_conflict_detected_even_when_response_matches() -> None: - from azure_functions_validation import validate_http + validate_http = _validator() app = func.FunctionApp() @openapi(requests=_Authored, responses=_Validated) @app.route(route="orders", methods=["POST"]) - @validate_http(body=_Validated, response_model=_Validated) # type: ignore[untyped-decorator] + @validate_http(body=_Validated, response_model=_Validated) def orders(req: Any, body: Any) -> Any: return body @@ -116,13 +125,13 @@ def orders(req: Any, body: Any) -> Any: def test_response_conflict_detected_even_when_request_matches() -> None: - from azure_functions_validation import validate_http + validate_http = _validator() app = func.FunctionApp() @openapi(requests=_Validated, responses=_Authored) @app.route(route="orders", methods=["POST"]) - @validate_http(body=_Validated, response_model=_OtherValidated) # type: ignore[untyped-decorator] + @validate_http(body=_Validated, response_model=_OtherValidated) def orders(req: Any, body: Any) -> Any: return body @@ -131,13 +140,13 @@ def orders(req: Any, body: Any) -> Any: def test_equivalent_model_declared_through_both_decorators_merges() -> None: - from azure_functions_validation import validate_http + validate_http = _validator() app = func.FunctionApp() @openapi(requests=_Validated, responses=_Validated) @app.route(route="orders", methods=["POST"]) - @validate_http(body=_Validated, response_model=_Validated) # type: ignore[untyped-decorator] + @validate_http(body=_Validated, response_model=_Validated) def orders(req: Any, body: Any) -> Any: return body @@ -150,13 +159,13 @@ def orders(req: Any, body: Any) -> Any: def test_inferred_response_is_still_superseded_by_validation_metadata() -> None: - from azure_functions_validation import validate_http + validate_http = _validator() app = func.FunctionApp() @openapi() @app.route(route="orders", methods=["GET"]) - @validate_http(response_model=_Validated) # type: ignore[untyped-decorator] + @validate_http(response_model=_Validated) def orders(req: Any) -> _Authored: raise NotImplementedError diff --git a/tools/browser_integrity.py b/tools/browser_integrity.py new file mode 100644 index 00000000..3ed563a3 --- /dev/null +++ b/tools/browser_integrity.py @@ -0,0 +1,80 @@ +"""Verify the wheel's pinned browser assets without retaining downloaded source.""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import hmac +from pathlib import Path +import subprocess + +from azure_functions_openapi.swagger_ui import ( + _SWAGGER_UI_BUNDLE_SRI, + _SWAGGER_UI_CDN_BASE, + _SWAGGER_UI_CSS_SRI, + _SWAGGER_UI_DIST_VERSION, +) +from tools.security_report import Check, Status, write_report + + +def compare_asset(body: bytes | None, expected: str) -> Status: + if body is None: + return Status.TOOL_ERROR + actual = "sha384-" + base64.b64encode(hashlib.sha384(body).digest()).decode("ascii") + return Status.PASS if hmac.compare_digest(actual, expected) else Status.FAIL + + +def verify_assets() -> list[Check]: + checks: list[Check] = [] + for asset, digest in ( + ("swagger-ui.css", _SWAGGER_UI_CSS_SRI), + ("swagger-ui-bundle.js", _SWAGGER_UI_BUNDLE_SRI), + ): + body: bytes | None = None + code: int | None = None + try: + process = subprocess.run( + [ + "/usr/bin/curl", + "--fail", + "--silent", + "--show-error", + "--proto", + "=https", + "--max-time", + "60", + "--retry", + "2", + f"{_SWAGGER_UI_CDN_BASE}/{asset}", + ], + capture_output=True, + check=False, + timeout=200, + ) + code = process.returncode + body = process.stdout if code == 0 else None + except (OSError, subprocess.TimeoutExpired): + code = None + checks.append( + Check( + f"sri-{asset}", + (asset,), + compare_asset(body, digest), + code, + counts={"assets": int(body is not None)}, + tool_versions={"swagger-ui-dist": _SWAGGER_UI_DIST_VERSION, "hash": "sha384"}, + message="Network failure is TOOL_ERROR; received bytes with wrong hash are FAIL.", + ) + ) + return checks + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output", type=Path, default=Path("security-results")) + return write_report(verify_assets(), parser.parse_args().output) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/candidate_inventory.py b/tools/candidate_inventory.py new file mode 100644 index 00000000..7bb7129b --- /dev/null +++ b/tools/candidate_inventory.py @@ -0,0 +1,54 @@ +"""Verify wheel identity before excluding it from an installed dependency inventory.""" + +from __future__ import annotations + +from dataclasses import dataclass +from email.parser import Parser +from importlib.metadata import distributions +from pathlib import Path +import re +from zipfile import ZipFile + +from tools.security_report import EvidenceError, identifier + + +@dataclass(frozen=True, slots=True) +class Package: + name: str + version: str + + +def package(name: str, version: str) -> Package: + normalized = re.sub(r"[-_.]+", "-", identifier(name)).lower() + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9.!+_-]*", version): + raise EvidenceError("invalid installed version") + return Package(normalized, version) + + +@dataclass(frozen=True, slots=True) +class Inventory: + candidate: Package + dependencies: tuple[Package, ...] + + def requirements(self) -> str: + return "".join(f"{p.name}=={p.version}\n" for p in self.dependencies) + + +def verified_inventory(site_packages: Path, wheel: Path) -> Inventory: + with ZipFile(wheel) as archive: + metadata = [n for n in archive.namelist() if n.endswith(".dist-info/METADATA")] + if len(metadata) != 1: + raise EvidenceError("candidate wheel metadata missing or ambiguous") + message = Parser().parsestr(archive.read(metadata[0]).decode("utf-8")) + candidate = package(str(message.get("Name", "")), str(message.get("Version", ""))) + installed = tuple( + package(dist.metadata["Name"], dist.version or "") + for dist in distributions(path=[str(site_packages)]) + ) + matches = tuple(p for p in installed if p.name == candidate.name) + if matches != (candidate,) or len({p.name for p in installed}) != len(installed): + raise EvidenceError("candidate absent, mismatched or duplicate installed distribution") + dependencies = tuple(sorted((p for p in installed if p != candidate), key=lambda p: p.name)) + if not dependencies: + raise EvidenceError("empty third-party inventory") + return Inventory(candidate, dependencies) diff --git a/tools/dependency_audit_policy.py b/tools/dependency_audit_policy.py new file mode 100644 index 00000000..9004f9be --- /dev/null +++ b/tools/dependency_audit_policy.py @@ -0,0 +1,183 @@ +"""Audit an isolated wheel environment; distinguish advisory findings from tool failures.""" + +from __future__ import annotations + +import argparse +from dataclasses import replace +from datetime import date +from importlib.metadata import version +from pathlib import Path +import subprocess +import sys +import tempfile +from zipfile import BadZipFile + +from tools.candidate_inventory import Inventory, package, verified_inventory +from tools.security_exceptions import parse_exceptions +from tools.security_report import ( + Check, + EvidenceError, + Finding, + Status, + identifier, + load, + mapping, + sequence, + text, + write_report, +) + + +def classify_audit(raw: str, exit_status: int, *, policy_raw: str, today: date) -> Check: + result = Check("pip-audit", ("resolved-runtime-wheel",), Status.TOOL_ERROR, exit_status) + try: + policy = load(policy_raw) + if policy["schema_version"] != 1 or policy["blocking"] != "all-known-runtime": + raise EvidenceError("unsupported policy") + rules = parse_exceptions(policy["exceptions"], today) + if any(r.file or r.line or not r.package or r.kind == "reviewed-static" for r in rules): + raise EvidenceError("invalid dependency exception") + dependencies = sequence(load(raw)["dependencies"]) + if not dependencies or exit_status not in {0, 1}: + raise EvidenceError("empty audit or tool crash") + findings: list[Finding] = [] + applied: set[str] = set() + blocking = 0 + packages = 0 + for dependency in dependencies: + item = mapping(dependency) + package = identifier(item["name"]).lower().replace("_", "-") + text(item["version"]) + if "skip_reason" in item: + raise EvidenceError("dependency could not be audited") + packages += 1 + for advisory in sequence(item["vulns"]): + vuln = mapping(advisory) + advisory_id = identifier(vuln["id"]) + severity = identifier(vuln.get("severity", "UNKNOWN")) + findings.append(Finding(advisory_id, severity=severity, package=package)) + matches = [ + r + for r in rules + if r.id == advisory_id + and r.package.lower().replace("_", "-") == package + and r.expires >= today + ] + if matches and ( + matches[0].kind != "fix-not-available" or not sequence(vuln["fix_versions"]) + ): + applied.add(f"{package}:{advisory_id}") + else: + blocking += 1 + if bool(findings) != (exit_status == 1): + raise EvidenceError("exit status disagrees with findings") + expired = sum(r.expires < today for r in rules) + return replace( + result, + status=Status.FAIL if blocking or expired else Status.PASS, + counts={ + "packages": packages, + "findings": len(findings), + "blocking": blocking, + "expired_exceptions": expired, + "stale_exceptions": len(rules) - len(applied), + }, + findings=tuple(findings), + exceptions_applied=tuple(sorted(applied)), + message="All known advisories block unless a current, applicable exception exists.", + ) + except (ValueError, KeyError, TypeError): + return replace(result, message="Audit/tool/policy incomplete or invalid; release blocked.") + + +def verify_audited_inventory(raw: str, inventory: Inventory) -> None: + audited = tuple( + package(text(item["name"]), text(item["version"])) + for value in sequence(load(raw)["dependencies"]) + for item in (mapping(value),) + ) + third_party = tuple(p for p in audited if p.name != inventory.candidate.name) + candidates = tuple(p for p in audited if p.name == inventory.candidate.name) + if candidates not in ((), (inventory.candidate,)) or ( + len(third_party) != len(inventory.dependencies) + or set(third_party) != set(inventory.dependencies) + ): + raise EvidenceError("audit differs from verified inventory") + + +def run_audit(site_packages: Path, policy: Path, candidate: Path | None = None) -> Check: + try: + if candidate is None: + raise EvidenceError("candidate wheel required") + inventory = verified_inventory(site_packages, candidate) + except (OSError, ValueError, KeyError, TypeError, BadZipFile): + return Check( + "pip-audit", + ("resolved-runtime-wheel",), + Status.INCOMPLETE, + None, + message="Candidate identity or third-party inventory incomplete; release blocked.", + ) + try: + with tempfile.TemporaryDirectory(prefix="openapi-audit-") as temporary: + requirements = Path(temporary) / "requirements.txt" + requirements.write_text(inventory.requirements(), encoding="utf-8") + process = subprocess.run( + [ + sys.executable, + "-m", + "pip_audit", + "--requirement", + str(requirements), + "--no-deps", + "--disable-pip", + "--format", + "json", + "--desc", + "off", + "--progress-spinner", + "off", + "--timeout", + "30", + ], + capture_output=True, + text=True, + check=False, + timeout=300, + ) + verify_audited_inventory(process.stdout, inventory) + result = classify_audit( + process.stdout, + process.returncode, + policy_raw=policy.read_text(encoding="utf-8"), + today=date.today(), + ) + return replace(result, tool_versions={"pip-audit": version("pip-audit")}) + except (OSError, subprocess.TimeoutExpired, ValueError, KeyError, TypeError): + return Check( + "pip-audit", + ("resolved-runtime-wheel",), + Status.TOOL_ERROR, + None, + message="Audit execution failed; release blocked.", + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--site-packages", type=Path, required=True) + parser.add_argument("--candidate-wheel", type=Path, required=True) + parser.add_argument( + "--policy", type=Path, default=Path("security/dependency-audit-policy.json") + ) + parser.add_argument("--output", type=Path, default=Path("security-results")) + parser.add_argument("--development", action="store_true") + args = parser.parse_args() + check = run_audit(args.site_packages, args.policy, args.candidate_wheel) + if args.development: + check = replace(check, name="pip-audit-development", scope=("runtime-and-development",)) + return write_report([check], args.output) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/security_evidence.py b/tools/security_evidence.py new file mode 100644 index 00000000..e89e4a36 --- /dev/null +++ b/tools/security_evidence.py @@ -0,0 +1,195 @@ +"""Run Bandit over the declared executable surfaces, or record Semgrep action evidence.""" + +from __future__ import annotations + +import argparse +from dataclasses import replace +from datetime import date +from importlib.metadata import version +from pathlib import Path +import subprocess +import sys + +from tools.security_exceptions import parse_exceptions +from tools.security_report import ( + Check, + EvidenceError, + Finding, + Status, + identifier, + integer, + load, + mapping, + sequence, + text, + write_report, +) + +SCOPE = ("src", "examples", "demo", "scripts", "tools") +FILE_FLOOR = 40 + + +def classify_bandit(raw: str, exit_status: int, *, floor: int = FILE_FLOOR) -> Check: + result = Check("bandit", SCOPE, Status.TOOL_ERROR, exit_status) + try: + data = load(raw) + metrics = mapping(data["metrics"]) + totals = mapping(metrics["_totals"]) + files = tuple(identifier(f.removeprefix("./")) for f in metrics if f != "_totals") + findings: list[Finding] = [] + counts = { + "files": len(files), + "loc": integer(totals.get("loc", 0)), + "errors": len(sequence(data["errors"])), + } + for entry in sequence(data["results"]): + item = mapping(entry) + severity = text(item["issue_severity"]) + confidence = text(item["issue_confidence"]) + if severity not in {"LOW", "MEDIUM", "HIGH", "UNDEFINED"} or confidence not in { + "LOW", + "MEDIUM", + "HIGH", + "UNDEFINED", + }: + raise EvidenceError("invalid finding classification") + findings.append( + Finding( + identifier(item["test_id"]), + identifier(text(item["filename"]).removeprefix("./")), + integer(item["line_number"]), + severity, + confidence, + ) + ) + for key in (f"severity.{severity}", f"confidence.{confidence}"): + counts[key] = counts.get(key, 0) + 1 + suppressions = integer(totals.get("nosec", 0)) + integer(totals.get("skipped_tests", 0)) + result = replace(result, counts=counts, suppressions=suppressions, findings=tuple(findings)) + if exit_status not in {0, 1} or counts["errors"]: + return replace( + result, message="Scanner failure or skipped/error files; scan is unusable." + ) + if len(files) < floor or counts["loc"] == 0 or suppressions: + return replace( + result, + status=Status.INCOMPLETE, + message="Scan below expected file/LOC floor or unapproved inline suppressions.", + ) + if bool(findings) != (exit_status == 1): + return replace(result, message="Scanner exit status disagrees with findings.") + return replace(result, status=Status.FAIL if findings else Status.PASS) + except (ValueError, KeyError, TypeError): + return replace( + result, message="Missing or invalid Bandit JSON; no raw diagnostics retained." + ) + + +def apply_bandit_policy(check: Check, policy_raw: str, today: date) -> Check: + if check.status not in {Status.PASS, Status.FAIL}: + return check + try: + rules = parse_exceptions(load(policy_raw)["exceptions"], today) + if any( + r.expires < today + or not r.file + or not r.line + or r.package + or r.kind != "reviewed-static" + for r in rules + ): + return replace( + check, status=Status.FAIL, message="Expired or invalid static exception." + ) + applied: list[str] = [] + blocking = 0 + for finding in check.findings: + matches = [ + r + for r in rules + if (r.id, r.file, r.line) == (finding.id, finding.file, finding.line) + ] + if matches: + applied.append(f"{finding.id}:{finding.file}:{finding.line}") + else: + blocking += 1 + return replace( + check, + status=Status.FAIL if blocking else Status.PASS, + exceptions_applied=tuple(applied), + counts={ + **check.counts, + "blocking": blocking, + "stale_exceptions": len(rules) - len(set(applied)), + }, + ) + except (ValueError, KeyError, TypeError): + return replace(check, status=Status.TOOL_ERROR, message="Invalid Bandit exception policy.") + + +def run_bandit() -> Check: + try: + process = subprocess.run( + [sys.executable, "-m", "bandit", "-r", *SCOPE, "-f", "json", "-q"], + capture_output=True, + text=True, + check=False, + timeout=180, + ) + result = classify_bandit(process.stdout, process.returncode) + expected = {p.as_posix() for surface in SCOPE for p in Path(surface).rglob("*.py")} + scanned = set(mapping(load(process.stdout)["metrics"])) - {"_totals"} + if {p.removeprefix("./") for p in scanned} != expected: + result = replace( + result, + status=Status.INCOMPLETE, + message="Scanned file inventory differs from declared surfaces.", + ) + result = apply_bandit_policy( + result, Path("security/bandit-policy.json").read_text(encoding="utf-8"), date.today() + ) + return replace(result, tool_versions={"bandit": version("bandit")}) + except (OSError, subprocess.TimeoutExpired, ValueError, KeyError, TypeError): + return Check( + "bandit", + SCOPE, + Status.TOOL_ERROR, + None, + message="Bandit execution or evidence collection failed.", + ) + + +def semgrep_evidence(outcome: str) -> Check: + code: int | None + match outcome: + case "success": + status, code = Status.INCOMPLETE, 0 + case "failure": + status, code = Status.TOOL_ERROR, 1 + case _: + status, code = Status.INCOMPLETE, None + return Check( + "semgrep", + ("p/python",), + status, + code, + tool_versions={"action": "713efdd345f3035192eaa63f56867b88e63e4e5d"}, + message=( + "action outcome only; findings and analysed inventory unavailable. " + "Failure may include findings." + ), + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("check", choices=("bandit", "semgrep")) + parser.add_argument("--outcome", default="") + parser.add_argument("--output", type=Path, default=Path("security-results")) + args = parser.parse_args() + check = run_bandit() if args.check == "bandit" else semgrep_evidence(args.outcome) + return write_report([check], args.output) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/security_exceptions.py b/tools/security_exceptions.py new file mode 100644 index 00000000..82d9d760 --- /dev/null +++ b/tools/security_exceptions.py @@ -0,0 +1,49 @@ +"""Parse narrow, owned exceptions with at most 90 days of validity.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date + +from tools.security_report import JSON, EvidenceError, identifier, mapping, sequence, text + + +@dataclass(frozen=True, slots=True) +class ExceptionRule: + id: str + package: str + file: str + line: int + kind: str + expires: date + added: date + + +def parse_exceptions(data: JSON, today: date) -> tuple[ExceptionRule, ...]: + rules: list[ExceptionRule] = [] + for entry in sequence(data): + item = mapping(entry) + text(item["reason"]) + text(item["owner"]) + added = date.fromisoformat(text(item["added"])) + expires = date.fromisoformat(text(item["expires"])) + if not 0 < (expires - added).days <= 90 or added > today: + raise EvidenceError("invalid exception validity window") + line = item.get("line", 0) + if not isinstance(line, int) or isinstance(line, bool) or line < 0: + raise EvidenceError("invalid exception line") + rule = ExceptionRule( + identifier(item["id"]), + identifier(item["package"]) if "package" in item else "", + identifier(item["file"]) if "file" in item else "", + line, + text(item["kind"]), + expires, + added, + ) + if rule.kind not in {"fix-not-available", "not-affected", "reviewed-static"}: + raise EvidenceError("unknown exception kind") + if rule in rules: + raise EvidenceError("duplicate exception") + rules.append(rule) + return tuple(rules) diff --git a/tools/security_report.py b/tools/security_report.py new file mode 100644 index 00000000..bdc4ac29 --- /dev/null +++ b/tools/security_report.py @@ -0,0 +1,120 @@ +"""Allow-listed security evidence: raw scanner output is never persisted.""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass, field +from enum import StrEnum +import json +import os +from pathlib import Path +import re +from typing import TypeAlias, assert_never + +JSON: TypeAlias = "None | bool | int | float | str | list[JSON] | dict[str, JSON]" + + +class EvidenceError(ValueError): + """Unusable scanner or policy input; messages never contain the input.""" + + +def mapping(value: JSON) -> dict[str, JSON]: + if not isinstance(value, dict): + raise EvidenceError("expected JSON object") + return value + + +def sequence(value: JSON) -> list[JSON]: + if not isinstance(value, list): + raise EvidenceError("expected JSON array") + return value + + +def text(value: JSON) -> str: + if not isinstance(value, str) or not value: + raise EvidenceError("expected nonempty string") + return value + + +def integer(value: JSON) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise EvidenceError("expected nonnegative integer") + return value + + +def load(raw: str) -> dict[str, JSON]: + parsed: JSON = json.loads(raw) + return mapping(parsed) + + +def identifier(value: JSON) -> str: + candidate = text(value) + if not re.fullmatch(r"[A-Za-z0-9_.:/@+-]{1,240}", candidate): + raise EvidenceError("unsafe evidence identifier") + if candidate.startswith(("/", "http:", "https:")) or ".." in candidate: + raise EvidenceError("unsafe evidence path") + return candidate + + +class Status(StrEnum): + PASS = "PASS" + FAIL = "FAIL" + TOOL_ERROR = "TOOL_ERROR" + INCOMPLETE = "INCOMPLETE" + + +@dataclass(frozen=True, slots=True) +class Finding: + id: str + file: str = "" + line: int = 0 + severity: str = "UNKNOWN" + confidence: str = "UNKNOWN" + package: str = "" + + +@dataclass(frozen=True, slots=True) +class Check: + name: str + scope: tuple[str, ...] + status: Status + exit_status: int | None + counts: dict[str, int] = field(default_factory=dict) + suppressions: int = 0 + exceptions_applied: tuple[str, ...] = () + tool_versions: dict[str, str] = field(default_factory=dict) + findings: tuple[Finding, ...] = () + message: str = "" + + @property + def returncode(self) -> int: + match self.status: + case Status.PASS: + return 0 + case Status.FAIL: + return 1 + case Status.TOOL_ERROR: + return 2 + case Status.INCOMPLETE: + return 3 + case unreachable: + assert_never(unreachable) + + +def write_report(checks: list[Check], output: Path, summary: Path | None = None) -> int: + output.mkdir(parents=True, exist_ok=True) + (output / "summary.json").write_text( + json.dumps({"schema_version": 1, "checks": [asdict(c) for c in checks]}, indent=2) + "\n", + encoding="utf-8", + ) + markdown = "## Security evidence\n\n| Check | Status | Counts |\n| --- | --- | --- |\n" + for check in checks: + markdown += f"| {check.name} | {check.status} | {json.dumps(check.counts)} |\n" + markdown += f"\n{check.message}\n\n" + if summary is None and os.environ.get("GITHUB_STEP_SUMMARY"): + summary = Path(os.environ["GITHUB_STEP_SUMMARY"]) + if summary is not None: + with summary.open("a", encoding="utf-8") as stream: + stream.write(markdown) + for check in checks: + print(f"{check.name}: {check.status} {json.dumps(check.counts)} {check.message}") + return max((c.returncode for c in checks), default=3)