diff --git a/.github/workflows/ci-test.yml b/.github/workflows/ci-test.yml index 53fbe38..e4c798a 100644 --- a/.github/workflows/ci-test.yml +++ b/.github/workflows/ci-test.yml @@ -531,3 +531,5 @@ jobs: run: | pytest -m compat2x -o addopts= -o pythonpath= \ tests/test_worker_compat_2x_spike.py -v + pytest -o addopts= -o pythonpath= tests/test_security_runtime.py \ + tests/test_security_inputs.py tests/test_security_auth.py -v diff --git a/.github/workflows/dependency-audit.yml b/.github/workflows/dependency-audit.yml index 3c7d8f2..339db24 100644 --- a/.github/workflows/dependency-audit.yml +++ b/.github/workflows/dependency-audit.yml @@ -21,22 +21,35 @@ jobs: with: python-version: "3.14" - - name: Install audit tool 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" . + 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: Audit runtime dependencies continue-on-error: true run: | - echo "## Runtime dependency audit" >> "$GITHUB_STEP_SUMMARY" - python -m pip_audit 2>&1 | tee -a "$GITHUB_STEP_SUMMARY" - - - name: Install development dependencies - run: python -m pip install ".[dev]" + site_packages=$("$RUNNER_TEMP/runtime/bin/python" -c 'import sysconfig; print(sysconfig.get_path("purelib"))') + wheels=(dist/*.whl) + make security-audit-report SITE_PACKAGES="$site_packages" CANDIDATE_WHEEL="${wheels[0]}" SECURITY_OUTPUT=security-results/runtime - name: Audit runtime and development dependencies + if: always() continue-on-error: true run: | - echo "## Runtime and development dependency audit" >> "$GITHUB_STEP_SUMMARY" - python -m pip_audit 2>&1 | tee -a "$GITHUB_STEP_SUMMARY" + site_packages=$("$RUNNER_TEMP/development/bin/python" -c 'import sysconfig; print(sysconfig.get_path("purelib"))') + wheels=(dist/*.whl) + make security-audit-report SITE_PACKAGES="$site_packages" CANDIDATE_WHEEL="${wheels[0]}" AUDIT_ARGS=--development SECURITY_OUTPUT=security-results/development + + - name: Upload redacted reporting evidence + if: always() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 + 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 f513d85..2166888 100644 --- a/.github/workflows/publish-pypi.yml +++ b/.github/workflows/publish-pypi.yml @@ -239,9 +239,51 @@ jobs: AZURE_TENANT_ID: ${{ secrets.AZURE_TENANT_ID }} AZURE_SUBSCRIPTION_ID: ${{ secrets.AZURE_SUBSCRIPTION_ID }} + runtime-dependency-audit: + name: Runtime dependency audit 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 exact candidate artifact + uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4 + with: + name: dist + path: candidate-dist + - name: Install only the candidate runtime closure + 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 dependency policy + run: | + site_packages=$("$RUNNER_TEMP/candidate/bin/python" -c 'import sysconfig; print(sysconfig.get_path("purelib"))') + wheels=(candidate-dist/*.whl) + make security-runtime-audit SITE_PACKAGES="$site_packages" CANDIDATE_WHEEL="${wheels[0]}" SECURITY_OUTPUT=security-results/runtime + - name: Upload redacted release security evidence + if: always() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 + 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 6f76dbd..c34e56c 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 @@ -29,7 +30,19 @@ jobs: make install - name: Run Bandit - run: make security + if: always() + run: make security-bandit SECURITY_PYTHON='hatch run python' + + - name: Type check security tools + run: make security-types SECURITY_PYTHON='hatch run python' + + - name: Upload redacted security evidence + if: always() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 + with: + name: security-bandit + path: security-results/** + if-no-files-found: error semgrep: runs-on: ubuntu-latest @@ -38,6 +51,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 + + - name: Upload redacted security evidence + if: always() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 + with: + name: security-semgrep + path: security-results/** + if-no-files-found: error diff --git a/.github/workflows/worker-nightly.yml b/.github/workflows/worker-nightly.yml index 67cb6b5..58d6c28 100644 --- a/.github/workflows/worker-nightly.yml +++ b/.github/workflows/worker-nightly.yml @@ -63,6 +63,7 @@ jobs: - name: Run WorkerCompat suite against pre-release worker library run: | python -m pytest tests/test_decorator.py tests/test_integration.py \ + tests/test_security_runtime.py tests/test_security_inputs.py tests/test_security_auth.py \ -v --no-header -p no:cacheprovider --no-cov - name: Explain a WorkerCompat break diff --git a/.gitignore b/.gitignore index 2ff99df..011937e 100644 --- a/.gitignore +++ b/.gitignore @@ -19,6 +19,7 @@ htmlcov/ # Build artifacts build/ dist/ +security-results/ *.egg-info/ # IDE diff --git a/AGENTS.md b/AGENTS.md index 83144be..666973f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,6 +16,10 @@ ## Working Rules +Security gates, evidence redaction and expiring exceptions: see [SECURITY.md](SECURITY.md). +Use `make security` for expanded inventory scanning and the `security-*` Makefile +entry points documented there for evidence and audit policy. + ### Test Coverage - Maintain test coverage at **95% or above** for committed changes and PRs. - Run `hatch run pytest --cov --cov-report=term-missing -q` to verify before submitting changes. diff --git a/Makefile b/Makefile index c781c5f..d23e901 100644 --- a/Makefile +++ b/Makefile @@ -59,8 +59,7 @@ lint: ensure-hatch @$(HATCH) run lint .PHONY: security -security: ensure-hatch - @$(HATCH) run security +security: security-inventory .PHONY: check-schema-hash check-schema-hash: ensure-hatch @@ -175,3 +174,18 @@ clean-all: clean help: @echo "Available commands:" && \ grep -E '^\.PHONY: ' Makefile | cut -d ':' -f2 | xargs -n1 echo " - make" + +# Security entry points use the selected CI interpreter without provisioning Hatch. +SECURITY_PYTHON ?= $(if $(CI),python,hatch run python) +SECURITY_OUTPUT ?= security-results +.PHONY: security-inventory security-bandit security-semgrep security-runtime-audit security-audit-report security-types +security-inventory: security-bandit +security-bandit: + @$(SECURITY_PYTHON) -m tools.security_evidence bandit --output "$(SECURITY_OUTPUT)" +security-semgrep: + @$(SECURITY_PYTHON) -c 'from pathlib import Path; import os; from tools.security_evidence import semgrep_evidence; from tools.security_report import write_report; code = write_report([semgrep_evidence(os.environ.get("SCAN_OUTCOME", ""))], Path("$(SECURITY_OUTPUT)")); raise SystemExit(0 if code == 3 else code)' +security-runtime-audit: + @$(SECURITY_PYTHON) -m tools.dependency_audit_policy --site-packages "$(SITE_PACKAGES)" --candidate-wheel "$(CANDIDATE_WHEEL)" --output "$(SECURITY_OUTPUT)" $(AUDIT_ARGS) +security-audit-report: security-runtime-audit +security-types: + @$(SECURITY_PYTHON) -m mypy --explicit-package-bases tools/security_report.py tools/security_exceptions.py tools/security_evidence.py tools/dependency_audit_policy.py tools/dependency_inventory.py diff --git a/SECURITY.md b/SECURITY.md index ce14ebb..6714f76 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -33,3 +33,124 @@ If you prefer email, contact: yeongseon.choe@gmail.com | ------- | --------- | | Latest release | :white_check_mark: | | Older releases | :x: | + +## Production trust boundaries + +Default 5xx responses and failed formatters contain a fixed public envelope. +Server-side logs intentionally keep the original exception with traceback (`exc_info`) +so operators can diagnose 5xx failures in Application Insights; they are an +operator-facing channel, not part of the HTTP response. Exception messages, Pydantic +input values and application-supplied data can therefore reach your log sink: do not +put secrets into exception messages, and restrict access to logs. Handlers returning +`HttpResponse` and successful custom formatters own their content. + +`expose_internal_errors=False` is the default. Enabling it passes original 5xx +exceptions to formatters and is **unsafe for production** without application +redaction; it does not automatically recognize/redact secrets. Original 4xx +exceptions are also available to custom formatters. See +[configuration safety](docs/configuration.md#expose_internal_errors). + +Authentication is host-owned. Validation preserves route auth levels, rejects the +unsupported above-route decorator order, and never authenticates direct requests. +Configure platform authentication and route auth explicitly. Body-size and JSON +content-type enforcement are opt-in. Use `max_body_bytes` and constrained models +for public input. The default Pydantic adapter rejects JSON bodies deeper than +512 object/array levels with a sanitized 400, consistently across Python versions. +Brackets inside strings do not count. No implicit error-count/response-size cap +exists: nested invalid arrays below that ceiling can amplify errors. Duplicate JSON keys are +last-value-wins; non-finite numbers require model constraints. These are follow-up +hardening topics, not demonstrated credential leaks. Schema/wheel verification +rejects drift and never rewrites pins. + +## Automated controls and permissions + +| Control | Runs where | Enforcement | +| --- | --- | --- | +| Bandit 1.9.4 | PR, main, weekly, dispatch | Existing blocking job; all severities reviewed | +| SHA-pinned Semgrep `p/python` | Same workflow | Existing blocking job; execution evidence added | +| CodeQL and SBOM | Existing workflows | Unchanged scanning/reporting controls | +| Runtime pip-audit 2.10.1 | Exact candidate wheel, Python 3.11–3.14 | Blocking dependency of `publish` | +| Runtime/development audit | Weekly/dispatch | Reporting-only with shared classifier | +| Offline trust-boundary tests | Full SDK matrix, 2.x wheel lane, worker nightly | Existing lane enforcement retained | + +Security workflows and the new release job use `contents: read` only. Existing +CodeQL security-events permission, Azure OIDC and PyPI OIDC are unchanged. The new +gate replaces no runtime/host/Azure certification and never rebuilds the artifact. +Browser/CDN checks do not apply: this package ships no Swagger UI/browser assets. +Branch-protection requirements remain repository settings, not workflow assertions. + +## Static scope and exceptions + +`make security` (also `make security-inventory` / `make security-bandit`) scans every Python file under +`src/`, `examples/` (including deployed `e2e_app`), `scripts/`, and `tools/`, with +no per-file exclusions. Tests/fixtures are excluded; shipped `conformance.py` is +included. `infra/` contains Bicep, not Python, and requires configuration review. +There is no demo directory. Bytecode/non-Python generated files are not inputs. +Scanned inventory must equal disk, include at least 30 files and have positive LOC. + +Bandit uses `--ignore-nosec`: existing conformance assertion markers cannot hide +findings. Every inline marker must match exact ID/file/line policy; unlisted or +blanket skips produce INCOMPLETE. Baseline exceptions cover test-only assertions, +shell-free tooling subprocesses and the literal TOOL_ERROR enum false positive. +No blanket rule skips or new inline suppressions replace review. Local full validation +and CI use this same expanded inventory. `make security-types` checks the helpers. + +Both policies require exact identifiers, reason, owner, added and expiry dates, +with a positive lifetime at most 90 days. Static exceptions use `reviewed-static` +and exact file/line. Dependencies use exact package/advisory and `not-affected` +(independent affectedness evidence) or `fix-not-available` (only while no fix is +listed). Expired exceptions block even when unused; stale exceptions are counted. +Maintainers review evidence before accepting exceptions. Runtime policy starts empty. + +## Runtime policy, failure semantics and recovery + +Every known runtime advisory blocks, regardless of severity; missing severity is +UNKNOWN, never assumed low. Each supported interpreter is audited because markers +change dependency closures. The candidate venv is created without pip and contains +only the wheel/runtime dependencies. `make security-runtime-audit +SITE_PACKAGES= CANDIDATE_WHEEL=` reads wheel +METADATA and requires exactly matching installed name/version before excluding only +that first-party candidate. It freezes the entire installed third-party closure to +temporary `name==version` requirements, audited outside the venv with `--no-deps +--disable-pip`. Unpublished candidates therefore do not require PyPI publication; +missing/mismatched candidate identity, empty inventories, third-party skips and +missing/changed audit inventory remain blocking. `make security-audit-report` uses +the same policy for reporting jobs; `AUDIT_ARGS=--development` labels development scope. + +Exit statuses: PASS=0; FAIL=1 (findings/expired exceptions); TOOL_ERROR=2 +(crash, timeout, malformed evidence, policy/tool/DB failure); INCOMPLETE=3 +(empty/missing scope, skipped dependencies, inventory gaps/unlisted suppressions). +Every non-PASS blocks release. Semgrep exposes only action outcome: success INCOMPLETE +("action outcome only"), +failure conservatively TOOL_ERROR (unknown finding count), skipped/cancelled +INCOMPLETE. `make security-semgrep` tolerates exactly evidence exit 3 without claiming +PASS; its original SHA-pinned action remains the gate and a failing action stays red. + +Advisory DB outages can delay releases: a deliberate availability tradeoff for +fail-closed evidence. Never bypass it. Retry an unchanged tag with +`gh workflow run publish-pypi.yml --ref vX.Y.Z`; changed code/workflows must fix +forward to a new candidate. Trusted OIDC publishing and exact-wheel reuse remain. + +## Evidence and deployment review + +Helpers emit versioned `security-results/**/summary.json` with check/scope/status, +counts, exit status, tool versions, applied exceptions and allow-listed finding +IDs, relative paths/lines and severities. Markdown summaries go to Job Summary. +Artifact uploads run `always()` and fail on missing evidence. Raw scanner output, +source, advisory descriptions, policy reasons, request bodies, authorization headers +and connection strings are not retained. Generated local evidence is gitignored. +Findings remain code/config review findings until runtime impact is demonstrated. + +Examples and anonymous host-smoke apps are local/synthetic demonstrations, not +production authentication templates. `examples/e2e_app` and `infra/main.bicep` are +ephemeral public certification infrastructure. Bicep enforces HTTPS, storage TLS +1.2 and private blobs, but retains shared-key connection strings, public ingress +and an Application Insights connection-string output. Managed identity/RBAC, +private endpoints, explicit SCM/app TLS and removing deployment outputs need +real-Azure certification. This work changes no IaC identity/auth settings. + +Azure failure diagnostics print Application Insights exceptions/traces and raw +app logs. The settings printer hides unclassified values/package SAS URLs, but +library exception logs, host logs and app logs remain outside that guarantee. Use +synthetic data only; replace raw dumps with allow-listed diagnostics in an +Azure-verified follow-up. Never send real credentials/user data to certification. diff --git a/docs/configuration.md b/docs/configuration.md index 705bd1b..06e4d48 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -94,6 +94,14 @@ def create(req: func.HttpRequest, body: CreateBody) -> dict[str, str]: protection. Configure Azure Functions host and/or API Management request-size limits as the outer enforcement boundary. +Independently of the byte limit, the default Pydantic adapter accepts at most +512 nested JSON object/array levels (the outermost container counts as one). +Deeper bodies return a sanitized `400 Invalid JSON` before model validation on +every supported Python version; brackets inside strings do not count. This +portable ceiling avoids depending on interpreter-specific parser recursion +limits. Ordinary model-validation failures remain `422`. Custom adapters own +their parsing limits. + ### `require_json_content_type` Set `require_json_content_type=True` with `body=` to reject non-empty bodies @@ -304,7 +312,20 @@ responses it receives a public `InternalServerError` containing only `expose_internal_errors` defaults to `False`. Setting it to `True` passes the original 5xx exception to `error_formatter`, restoring the old behavior. This -is unsafe because a formatter can expose internal details to clients. +is unsafe for production because a formatter can expose internal details to clients. +Formatter output is **not** automatically redacted: supply an explicitly redacting +formatter. A formatter that raises or produces unserializable output still gets +a sanitized 500 fallback. Custom 4xx formatters receive original exceptions, which +can contain Pydantic input values; do not render `str(exc)` in production. + +Authentication is Azure Functions host-owned: validation does not check function +keys or bearer tokens, or establish an authenticated principal. Configure route +`auth_level` and platform authentication explicitly. Direct test calls bypass the +host. For public input, configure `max_body_bytes`, JSON media-type enforcement +and constrained models. The default adapter caps JSON depth at 512 object/array +levels; there is no implicit error-count cap. +duplicate JSON keys use last-value-wins semantics, and model configuration controls +non-finite numbers. See the root SECURITY.md for these production boundaries. ### `status_code` diff --git a/examples/README.md b/examples/README.md index 70a09f8..7f0c606 100644 --- a/examples/README.md +++ b/examples/README.md @@ -1,5 +1,10 @@ # Examples +These are local-only demonstrations, not production security templates. Anonymous +routes, in-memory stores and custom formatter output need explicit production +authentication, limits and redaction. The E2E app is ephemeral public certification +infrastructure using synthetic data only. See [SECURITY.md](../SECURITY.md). + `azure-functions-validation` keeps a growing set of smoke-tested examples: | Role | Path | Description | diff --git a/pyproject.toml b/pyproject.toml index 1a62864..c2e6aac 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -107,7 +107,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 && mypy --explicit-package-bases tools/security_report.py tools/security_exceptions.py tools/security_evidence.py tools/dependency_audit_policy.py" 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/security/bandit-policy.json b/security/bandit-policy.json new file mode 100644 index 0000000..d67d77e --- /dev/null +++ b/security/bandit-policy.json @@ -0,0 +1,38 @@ +{ + "schema_version": 1, + "exceptions": [ + {"id":"B404","file":"scripts/check_wheel_schemas.py","line":11,"kind":"reviewed-static","reason":"Wheel verification invokes explicit interpreter commands without a shell.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B603","file":"scripts/check_wheel_schemas.py","line":71,"kind":"reviewed-static","reason":"Installs the selected local candidate in a temporary venv without a shell.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B603","file":"scripts/check_wheel_schemas.py","line":74,"kind":"reviewed-static","reason":"Runs fixed isolated import verification with the temporary interpreter, no shell.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":112,"kind":"reviewed-static","reason":"Shipped pytest mixin assertion, not runtime authorization; existing inline B101 marker is audited with ignore-nosec.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":114,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":115,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":116,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":117,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":118,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":120,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":130,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":177,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":187,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":201,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":212,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":229,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":230,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":231,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":249,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":250,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":251,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":271,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B101","file":"src/azure_functions_validation/conformance.py","line":272,"kind":"reviewed-static","reason":"Conformance-only assertion; never an application security check.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B404","file":"tools/check_pr_format.py","line":5,"kind":"reviewed-static","reason":"Contributor formatter uses shell-free git/ruff subprocesses.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B607","file":"tools/check_pr_format.py","line":10,"kind":"reviewed-static","reason":"git resolves from trusted contributor/CI PATH, no shell.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B603","file":"tools/check_pr_format.py","line":10,"kind":"reviewed-static","reason":"git revision arguments are separate and options terminate with --.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B607","file":"tools/check_pr_format.py","line":37,"kind":"reviewed-static","reason":"ruff resolves from trusted contributor/CI PATH, no shell.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B603","file":"tools/check_pr_format.py","line":37,"kind":"reviewed-static","reason":"ruff filenames follow -- as separate arguments, no shell.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B404","file":"tools/dependency_audit_policy.py","line":11,"kind":"reviewed-static","reason":"Bounded shell-free pip-audit command uses sys.executable.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B603","file":"tools/dependency_audit_policy.py","line":111,"kind":"reviewed-static","reason":"Fixed pip-audit module with private frozen requirements path passed as separate argument, no shell.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B404","file":"tools/security_evidence.py","line":11,"kind":"reviewed-static","reason":"Bounded shell-free Bandit command uses sys.executable.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B603","file":"tools/security_evidence.py","line":125,"kind":"reviewed-static","reason":"Fixed Bandit command over repository-controlled paths, no shell.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"}, + {"id":"B105","file":"tools/security_report.py","line":59,"kind":"reviewed-static","reason":"TOOL_ERROR is a public status enum, not a credential.","owner":"yeongseon","added":"2026-10-22","expires":"2027-01-20"} + ] +} diff --git a/security/dependency-audit-policy.json b/security/dependency-audit-policy.json new file mode 100644 index 0000000..edf5af5 --- /dev/null +++ b/security/dependency-audit-policy.json @@ -0,0 +1,5 @@ +{ + "schema_version": 1, + "blocking": "all-known-runtime", + "exceptions": [] +} diff --git a/src/azure_functions_validation/_json_body.py b/src/azure_functions_validation/_json_body.py new file mode 100644 index 0000000..d20f42f --- /dev/null +++ b/src/azure_functions_validation/_json_body.py @@ -0,0 +1,34 @@ +"""Portable JSON body parsing limits, independent of CPython's C stack guard.""" + +import json +from typing import Any, Final + +from .errors import MalformedRequestError + +_MAX_JSON_DEPTH: Final = 512 + + +def parse_json_body(body: str) -> Any: + depth = 0 + in_string = False + escaped = False + for character in body: + if in_string: + if escaped: + escaped = False + elif character == "\\": + escaped = True + elif character == '"': + in_string = False + elif character == '"': + in_string = True + elif character in ("[", "{"): + depth += 1 + if depth > _MAX_JSON_DEPTH: + raise MalformedRequestError("Invalid JSON") + elif character in ("]", "}"): + depth -= 1 + try: + return json.loads(body) + except (json.JSONDecodeError, RecursionError) as exc: + raise MalformedRequestError("Invalid JSON") from exc diff --git a/src/azure_functions_validation/adapter.py b/src/azure_functions_validation/adapter.py index 30b70c3..3899945 100644 --- a/src/azure_functions_validation/adapter.py +++ b/src/azure_functions_validation/adapter.py @@ -14,6 +14,7 @@ from pydantic.fields import FieldInfo from pydantic_core import PydanticSerializationError +from ._json_body import parse_json_body from .errors import AdapterValidationError, MalformedRequestError, SerializationError @@ -407,10 +408,7 @@ def parse_body(self, req: HttpRequest, model: Any) -> Any: # Empty JSON string - this is a missing body, not invalid JSON raise self._missing_body_validation_error() - try: - data = json.loads(body_str) - except json.JSONDecodeError as e: - raise MalformedRequestError("Invalid JSON") from e + data = parse_json_body(body_str) # Validate with Pydantic try: diff --git a/src/azure_functions_validation/decorator.py b/src/azure_functions_validation/decorator.py index 5a387fc..b8ab3de 100644 --- a/src/azure_functions_validation/decorator.py +++ b/src/azure_functions_validation/decorator.py @@ -61,6 +61,7 @@ def validate_http( expose_internal_errors: Unsafe opt-in that passes original server-side exceptions to *error_formatter*. Defaults to ``False`` so custom formatters receive a sanitized ``InternalServerError`` for 5xx errors. + Unsafe for production: formatter output is not automatically redacted. status_code: HTTP status code for successful responses (default 200). Use e.g. ``status_code=201`` for creation endpoints. legacy_loc: When ``True``, error ``loc`` values omit the leading diff --git a/tests/test_candidate_inventory.py b/tests/test_candidate_inventory.py new file mode 100644 index 0000000..80c3c13 --- /dev/null +++ b/tests/test_candidate_inventory.py @@ -0,0 +1,96 @@ +from pathlib import Path +import subprocess +from unittest.mock import Mock + +import pytest + +from tools import dependency_audit_policy as audit +from tools.security_report import Status + +POLICY = '{"schema_version":1,"blocking":"all-known-runtime","exceptions":[]}' + + +def installed(root: Path, name: str, version: str) -> None: + directory = root / f"{name}-{version}.dist-info" + directory.mkdir() + (directory / "METADATA").write_text( + f"Metadata-Version: 2.1\nName: {name}\nVersion: {version}\n" + ) + + +@pytest.mark.parametrize( + "raw,code,expected", + [ + ('{"dependencies":[{"name":"idna","version":"3.20","vulns":[]}]}', 0, Status.PASS), + ( + '{"dependencies":[{"name":"idna","version":"3.20","vulns":[{"id":"CVE-1","fix_versions":[]}]}]}', + 1, + Status.FAIL, + ), + ('{"dependencies":[{"name":"idna","skip_reason":"unpublished"}]}', 0, Status.INCOMPLETE), + ('{"dependencies":[]}', 0, Status.INCOMPLETE), + ('{"dependencies":[{"name":"other","version":"3.20","vulns":[]}]}', 0, Status.INCOMPLETE), + ( + '{"dependencies":[{"name":"idna","version":"3.20","vulns":[]},{"name":"azure-functions-validation","version":"9999.0.123","vulns":[]}]}', + 0, + Status.PASS, + ), + ( + '{"dependencies":[{"name":"idna","version":"3.20","vulns":[]},{"name":"azure-functions-validation","version":"1","vulns":[]}]}', + 0, + Status.TOOL_ERROR, + ), + ], +) +def test_unpublished_candidate_audits_exact_third_party_inventory( + raw: str, code: int, expected: Status, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Given a verified unpublished candidate and one installed dependency. + installed(tmp_path, "azure-functions-validation", "9999.0.123") + installed(tmp_path, "idna", "3.20") + policy = tmp_path / "policy.json" + policy.write_text(POLICY) + + def run(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: + assert "--path" not in command + assert "--no-deps" in command and "--disable-pip" in command + requirements = Path(command[command.index("--requirement") + 1]).read_text() + assert requirements == "idna==3.20\n" + return subprocess.CompletedProcess(command, code, stdout=raw) + + monkeypatch.setattr(subprocess, "run", run) + monkeypatch.setattr(audit, "version", lambda _: "2.10.1") + # When auditing, then only the verified first-party candidate is excluded. + result = audit.run_audit(tmp_path, policy, ("azure-functions-validation", "9999.0.123")) + assert result.status == expected + + +def test_candidate_identity_mismatch_blocks_before_tool_execution( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Given an installed candidate with the wrong version. + installed(tmp_path, "azure-functions-validation", "1") + run = Mock() + monkeypatch.setattr(subprocess, "run", run) + # When a different artifact identity is required, then auditing is incomplete. + result = audit.run_audit(tmp_path, tmp_path / "policy", ("azure-functions-validation", "2")) + assert result.status == Status.INCOMPLETE + run.assert_not_called() + + +def test_wheel_identity_requires_one_metadata_record(tmp_path: Path) -> None: + from zipfile import ZipFile + + from tools.dependency_inventory import wheel_identity + from tools.security_report import EvidenceError + + wheel = tmp_path / "candidate.whl" + with ZipFile(wheel, "w") as archive: + archive.writestr( + "candidate.dist-info/METADATA", "Name: azure_functions_validation\nVersion: 9999.0.1\n" + ) + assert wheel_identity(wheel) == ("azure-functions-validation", "9999.0.1") + with ZipFile(wheel, "a") as archive: + archive.writestr("other.dist-info/METADATA", "Name: other\nVersion: 1\n") + with pytest.raises(EvidenceError): + wheel_identity(wheel) diff --git a/tests/test_check_wheel_schemas.py b/tests/test_check_wheel_schemas.py index 7a61be0..3b8302f 100644 --- a/tests/test_check_wheel_schemas.py +++ b/tests/test_check_wheel_schemas.py @@ -2,6 +2,8 @@ import hashlib from pathlib import Path +import subprocess +import sys from zipfile import ZipFile import pytest @@ -71,3 +73,28 @@ def test_digest_mismatch_fails(tmp_path: Path) -> None: # Then the checker rejects the drift assert result != 0 + + +def test_tampered_copy_fails_without_repairing_any_bytes(tmp_path: Path) -> None: + # Given real pinned resources copied into a wheel with a tampered schema. + root = Path(__file__).resolve().parents[1] + resources = root / "src/azure_functions_validation/schemas" + original = { + name: (resources / name).read_bytes() + for name in ("endpoint.schema.json", "endpoint.schema.sha256") + } + wheel = tmp_path / "tampered.whl" + with ZipFile(wheel, "w") as archive: + archive.writestr(SCHEMA_PATH, original["endpoint.schema.json"] + b" ") + archive.writestr(PIN_PATH, original["endpoint.schema.sha256"]) + before = wheel.read_bytes() + + # When the actual CLI checks the copy; Then drift fails without rewriting copy or pin. + result = subprocess.run( + [sys.executable, str(root / "scripts/check_wheel_schemas.py"), str(wheel)], + capture_output=True, + check=False, + ) + assert result.returncode == 1 + assert wheel.read_bytes() == before + assert all((resources / name).read_bytes() == data for name, data in original.items()) diff --git a/tests/test_dependency_audit_policy.py b/tests/test_dependency_audit_policy.py new file mode 100644 index 0000000..9784543 --- /dev/null +++ b/tests/test_dependency_audit_policy.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +from datetime import date +import json +from pathlib import Path +import subprocess + +import pytest + +from tools import dependency_audit_policy +from tools.dependency_audit_policy import classify_audit +from tools.security_exceptions import parse_exceptions +from tools.security_report import EvidenceError, Status, load + +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": [], "description": "CANARY_SECRET"} + ] + if vulnerable + else [], + } + ] + } + ) + + +@pytest.mark.parametrize("vulnerable,expected", [(False, Status.PASS), (True, Status.FAIL)]) +def test_any_known_runtime_advisory_blocks(vulnerable: bool, expected: Status) -> None: + result = classify_audit(payload(vulnerable), int(vulnerable), policy_raw=POLICY, today=TODAY) + assert result.status == expected + assert "CANARY_SECRET" not in repr(result) + + +@pytest.mark.parametrize("raw", ["", "invalid", "{}"]) +def test_invalid_audit_is_tool_error(raw: str) -> None: + assert classify_audit(raw, 1, policy_raw=POLICY, today=TODAY).status == Status.TOOL_ERROR + + +@pytest.mark.parametrize( + "raw", + [ + '{"dependencies":[]}', + '{"dependencies":[{"name":"x","version":"1","skip_reason":"CANARY_SECRET"}]}', + ], +) +def test_empty_or_skipped_audit_is_incomplete(raw: str) -> None: + assert classify_audit(raw, 0, policy_raw=POLICY, today=TODAY).status == Status.INCOMPLETE + + +@pytest.mark.parametrize( + "expires,expected", [("2026-10-21", Status.FAIL), ("2026-11-21", Status.PASS)] +) +def test_exception_expiry_and_affectedness(expires: str, expected: Status) -> None: + policy = json.loads(POLICY) + policy["exceptions"] = [ + { + "id": "CVE-2026-12345", + "package": "pydantic", + "kind": "fix-not-available", + "reason": "no patched release", + "owner": "maintainer", + "added": "2026-10-01", + "expires": expires, + } + ] + result = classify_audit(payload(True), 1, policy_raw=json.dumps(policy), today=TODAY) + assert result.status == expected + raw = json.loads(payload(True)) + raw["dependencies"][0]["vulns"][0]["fix_versions"] = ["3"] + assert ( + classify_audit(json.dumps(raw), 1, policy_raw=json.dumps(policy), today=TODAY).status + == Status.FAIL + ) + + +@pytest.mark.parametrize( + "patch", + [ + {"expires": "2027-10-22"}, + {"added": "2026-11-01"}, + {"kind": "ignore-all"}, + {"line": -1}, + {"line": True}, + {"owner": ""}, + {"reason": ""}, + ], +) +def test_invalid_exception_policy_fails(patch: dict[str, str | int]) -> None: + rule = { + "id": "CVE-1", + "package": "example", + "kind": "not-affected", + "reason": "reviewed", + "owner": "maintainer", + "added": "2026-10-01", + "expires": "2026-11-01", + **patch, + } + with pytest.raises((EvidenceError, ValueError)): + parse_exceptions(load(json.dumps({"exceptions": [rule]}))["exceptions"], TODAY) + + +@pytest.mark.parametrize("code", [0, 2]) +def test_inconsistent_exit_status_fails(code: int) -> None: + assert ( + classify_audit(payload(True), code, policy_raw=POLICY, today=TODAY).status + == Status.TOOL_ERROR + ) + + +def test_audit_runner_targets_only_candidate_installation( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + # Given an isolated installed inventory; When audited; Then tools use only its path. + for name, resolved in (("candidate", "1"), ("pydantic", "2.0")): + directory = tmp_path / f"{name}-{resolved}.dist-info" + directory.mkdir() + (directory / "METADATA").write_text(f"Name: {name}\nVersion: {resolved}\n") + policy = tmp_path / "policy.json" + policy.write_text(POLICY) + + def complete(command: list[str], **kwargs: str) -> subprocess.CompletedProcess[str]: + assert "--requirement" in command + return subprocess.CompletedProcess(command, 0, stdout=payload(), stderr="CANARY_SECRET") + + 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, ("candidate", "1")).status + == Status.PASS + ) diff --git a/tests/test_json_scanner.py b/tests/test_json_scanner.py new file mode 100644 index 0000000..69b3e3c --- /dev/null +++ b/tests/test_json_scanner.py @@ -0,0 +1,27 @@ +from collections.abc import Iterator +import time + +import pytest + +from azure_functions_validation._json_body import parse_json_body +from azure_functions_validation.errors import MalformedRequestError + + +class CountedBody(str): + visits = 0 + + def __iter__(self) -> Iterator[str]: + for character in super().__iter__(): + self.visits += 1 + yield character + + +def test_unterminated_escaped_quotes_are_scanned_once() -> None: + # Given a 200 KB unterminated JSON string with adversarial escaped quotes. + body = CountedBody('"' + '\\"' * 100_000) + started = time.perf_counter() + # When parsing malformed input, then rejection takes a single linear scan. + with pytest.raises(MalformedRequestError): + parse_json_body(body) + assert time.perf_counter() - started < 1 + assert body.visits == len(body) diff --git a/tests/test_print_safe_app_settings.py b/tests/test_print_safe_app_settings.py index 53a8fc1..dbf68ac 100644 --- a/tests/test_print_safe_app_settings.py +++ b/tests/test_print_safe_app_settings.py @@ -112,6 +112,23 @@ def test_known_secret_bearing_names_are_not_on_the_allowlist(self) -> None: class TestMainCli: + def test_cli_redacts_credential_shapes_and_unknown_settings( + self, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch + ) -> None: + # Given synthetic credentials in unknown settings and a package SAS URL. + settings = _settings( + CUSTOM_BEARER="Bearer CANARY_SECRET_token", + AzureWebJobsStorage="DefaultEndpointsProtocol=https;AccountKey=CANARY_SECRET_key", + WEBSITE_RUN_FROM_PACKAGE="https://example.test/package.zip?sig=CANARY_SECRET_sas", + NEW_SETTING="CANARY_SECRET_future", + ) + monkeypatch.setattr(sys, "stdin", io.StringIO(json.dumps(settings))) + # When the real CLI formatter runs; Then no credential reaches either output stream. + assert main([]) == 0 + captured = capsys.readouterr() + assert "CANARY_SECRET" not in captured.out + captured.err + assert captured.out.count("") == 4 + def test_main_reads_stdin_and_prints_safe_output( self, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/test_security_auth.py b/tests/test_security_auth.py new file mode 100644 index 0000000..83d1645 --- /dev/null +++ b/tests/test_security_auth.py @@ -0,0 +1,80 @@ +from collections.abc import Callable +import inspect +import json + +import azure.functions as func +from pydantic import BaseModel +import pytest + +from azure_functions_validation import validate_http +from azure_functions_validation._endpoint import set_endpoint_metadata +from azure_functions_validation.testing import MockHttpRequest + + +class Body(BaseModel): + value: int + + +blueprint_factory: Callable[[], func.Blueprint] = func.Blueprint + + +@pytest.mark.parametrize("blueprint", [False, True]) +@pytest.mark.parametrize("metadata", [False, True]) +@pytest.mark.parametrize("auth", [func.AuthLevel.FUNCTION, func.AuthLevel.ADMIN]) +def test_validation_preserves_host_auth_binding( + blueprint: bool, metadata: bool, auth: func.AuthLevel +) -> None: + # Given an explicitly authenticated route, including Blueprint and metadata paths. + app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS) + routes = blueprint_factory() if blueprint else app + + @validate_http(body=Body) + def handler(req: func.HttpRequest) -> None: + return None + + if metadata: + set_endpoint_metadata(handler, handler, {"version": 1, "parameters": []}) + routes.route(route="secure", auth_level=auth)(handler) + if blueprint: + app.register_functions(routes) + + # When SDK registration materializes; Then validation cannot downgrade the trigger. + binding = json.loads(app.get_functions()[0].get_function_json())["bindings"][0] + assert binding["authLevel"].casefold() == auth.value + assert "auth_level" not in inspect.signature(validate_http).parameters + + +def test_invalid_decorator_order_fails_without_changing_auth() -> None: + # Given validation above a host route, an unsupported and explicitly rejected order. + app = func.FunctionApp() + + @app.route(route="secure", auth_level=func.AuthLevel.FUNCTION) + def handler(req: func.HttpRequest) -> None: + return None + + # When decoration is attempted; Then no replacement/downgrade is silently accepted. + with pytest.raises(RuntimeError, match="FunctionBuilder"): + validate_http(body=Body)(handler) + binding = json.loads(app.get_functions()[0].get_function_json())["bindings"][0] + assert binding["authLevel"].casefold() == "function" + + +@pytest.mark.parametrize("credentials", [False, True]) +def test_validation_does_not_authenticate_or_echo_function_credentials(credentials: bool) -> None: + # Given identical malformed requests with/without host-owned credentials. + @validate_http(body=Body) + def handler(req: func.HttpRequest) -> None: + pytest.fail("malformed input reached handler") + + req = MockHttpRequest( + body=b'{"value":"CANARY_SECRET_input"}', + url="https://example.test/api/secure?code=CANARY_SECRET_key" + if credentials + else "https://example.test/api/secure", + headers={"x-functions-key": "CANARY_SECRET_key"} if credentials else {}, + params={"code": "CANARY_SECRET_key"} if credentials else {}, + ) + # When invoked directly (no host); Then validation alone makes no auth decision. + response = handler(req) + assert response.status_code == 422 + assert "CANARY_SECRET" not in response.get_body().decode() diff --git a/tests/test_security_evidence.py b/tests/test_security_evidence.py new file mode 100644 index 0000000..2b84d00 --- /dev/null +++ b/tests/test_security_evidence.py @@ -0,0 +1,227 @@ +from __future__ import annotations + +from datetime import date +import json +from pathlib import Path +import subprocess +import sys + +import pytest + +from tools import dependency_audit_policy, security_evidence +from tools.security_evidence import apply_bandit_policy, classify_bandit, semgrep_evidence +from tools.security_report import Check, Status, write_report + + +def payload() -> str: + return json.dumps( + { + "errors": [], + "metrics": { + "_totals": {"loc": 10, "nosec": 0, "skipped_tests": 0}, + "src/x.py": {"loc": 10}, + }, + "results": [], + } + ) + + +@pytest.mark.parametrize( + "raw,code,status", + [ + ("", 2, Status.TOOL_ERROR), + ("invalid", 0, Status.TOOL_ERROR), + ('{"errors":[],"metrics":{"_totals":{}},"results":[]}', 0, Status.INCOMPLETE), + ], +) +def test_missing_evidence_fails_closed(raw: str, code: int, status: Status) -> None: + assert classify_bandit(raw, code, floor=1).status == status + + +def test_complete_scan_has_real_counts() -> None: + result = classify_bandit(payload(), 0, floor=1) + assert result.status == Status.PASS + assert result.counts == {"files": 1, "loc": 10, "errors": 0} + + +@pytest.mark.parametrize("severity", ["LOW", "MEDIUM", "HIGH"]) +def test_findings_block_without_disclosing_raw_diagnostics(severity: str, tmp_path: Path) -> None: + # Given secret-bearing source diagnostics; When classified; Then only metadata survives. + data = json.loads(payload()) + data["results"] = [ + { + "test_id": "B105", + "filename": "src/x.py", + "line_number": 3, + "issue_severity": severity, + "issue_confidence": "HIGH", + "code": "Bearer CANARY_SECRET_token", + "issue_text": "AccountKey=CANARY_SECRET", + } + ] + result = classify_bandit(json.dumps(data), 1, floor=1) + assert result.status == Status.FAIL + assert result.counts[f"severity.{severity}"] == 1 + summary = tmp_path / "summary.md" + assert write_report([result], tmp_path, summary) == 1 + assert "CANARY_SECRET" not in (tmp_path / "summary.json").read_text() + summary.read_text() + + +@pytest.mark.parametrize("code", [0, 2]) +def test_crash_or_error_files_are_tool_errors(code: int) -> None: + data = json.loads(payload()) + data["errors"] = [{"reason": "CANARY_SECRET"}] + assert classify_bandit(json.dumps(data), code, floor=1).status == Status.TOOL_ERROR + + +def test_static_policy_matches_exact_location_and_expires() -> None: + data = json.loads(payload()) + data["results"] = [ + { + "test_id": "B404", + "filename": "src/x.py", + "line_number": 1, + "issue_severity": "LOW", + "issue_confidence": "HIGH", + } + ] + check = classify_bandit(json.dumps(data), 1, floor=1) + rule = { + "id": "B404", + "file": "src/x.py", + "line": 1, + "kind": "reviewed-static", + "reason": "fixed shell-free command", + "owner": "maintainer", + "added": "2026-10-01", + "expires": "2026-11-01", + } + policy = json.dumps({"schema_version": 1, "exceptions": [rule]}) + assert apply_bandit_policy(check, policy, date(2026, 10, 22)).status == Status.PASS + assert apply_bandit_policy(check, policy, date(2026, 11, 2)).status == Status.FAIL + rule["line"] = 2 + assert ( + apply_bandit_policy( + check, json.dumps({"schema_version": 1, "exceptions": [rule]}), date(2026, 10, 22) + ).status + == Status.FAIL + ) + assert apply_bandit_policy(check, "invalid", date(2026, 10, 22)).status == Status.TOOL_ERROR + + +@pytest.mark.parametrize( + "outcome,status", + [ + ("success", Status.INCOMPLETE), + ("failure", Status.TOOL_ERROR), + ("skipped", Status.INCOMPLETE), + ], +) +def test_semgrep_outcome_is_not_misreported_as_vulnerability(outcome: str, status: Status) -> None: + assert semgrep_evidence(outcome).status == status + + +def test_bandit_runner_compares_full_inventory(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(command: list[str], **kwargs: str) -> subprocess.CompletedProcess[str]: + assert "--ignore-nosec" in command + return subprocess.CompletedProcess(command, 0, stdout=raw, stderr="CANARY_SECRET") + + monkeypatch.setattr(subprocess, "run", complete) + assert security_evidence.run_bandit().status == Status.PASS + del metrics[next(iter(metrics))] + raw = json.dumps({"errors": [], "metrics": metrics, "results": []}) + assert security_evidence.run_bandit().status == Status.INCOMPLETE + + +@pytest.mark.parametrize("failure", [OSError(), subprocess.TimeoutExpired("scanner", 1)]) +def test_tool_execution_failures_never_pass( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, failure: OSError | subprocess.TimeoutExpired +) -> None: + for name in ("candidate", "dependency"): + directory = tmp_path / f"{name}-1.dist-info" + directory.mkdir() + (directory / "METADATA").write_text(f"Name: {name}\nVersion: 1\n") + + def fail(*args: str, **kwargs: str) -> 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, ("candidate", "1")).status + == Status.TOOL_ERROR + ) + + +def test_evidence_cli_runs_without_runtime_dependencies(tmp_path: Path) -> None: + # Given the real stdlib-only CLI; When invoked; Then it writes machine-readable evidence. + 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())["schema_version"] == 1 + + +@pytest.mark.parametrize("marker", ["# nosec", "# nosec B999"]) +def test_unlisted_inline_suppression_is_incomplete( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, marker: str +) -> None: + # Given an unlisted inline skip in a scanned file; When scanned; Then it cannot pass. + root = tmp_path / "src" + root.mkdir() + (root / "example.py").write_text(f"value = 1 {marker}\n") + policy = tmp_path / "security" + policy.mkdir() + (policy / "bandit-policy.json").write_text('{"schema_version":1,"exceptions":[]}') + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(security_evidence, "SCOPE", ("src",)) + raw = json.dumps( + { + "errors": [], + "metrics": {"_totals": {"loc": 30}, **{f"src/{i}.py": {} for i in range(30)}}, + "results": [], + } + ) + + def complete(*args: str, **kwargs: str) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess([], 0, stdout=raw, stderr="") + + monkeypatch.setattr(subprocess, "run", complete) + result = security_evidence.run_bandit() + assert result.status == Status.INCOMPLETE + assert "Unlisted inline suppression" in result.message + + +@pytest.mark.parametrize("development", [False, True]) +def test_audit_cli_preserves_failure_status( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, development: bool +) -> None: + args = ["audit", "--site-packages", str(tmp_path), "--output", str(tmp_path)] + if development: + args.append("--development") + monkeypatch.setattr(sys, "argv", args) + monkeypatch.setattr( + dependency_audit_policy, + "run_audit", + lambda path, policy: Check("pip-audit", (), Status.INCOMPLETE, None), + ) + assert dependency_audit_policy.main() == 3 diff --git a/tests/test_security_inputs.py b/tests/test_security_inputs.py new file mode 100644 index 0000000..3eb4f61 --- /dev/null +++ b/tests/test_security_inputs.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +import json +import logging +from unittest.mock import Mock + +from azure.functions import HttpRequest +from pydantic import BaseModel, ConfigDict, Field +import pytest + +from azure_functions_validation import PydanticAdapter, validate_http +from azure_functions_validation.testing import MockHttpRequest + + +class BoundaryModel(BaseModel): + model_config = ConfigDict(strict=True, allow_inf_nan=False) + value: int + items: list[int] = Field(default_factory=list, max_length=8) + + +@pytest.mark.parametrize( + "payload,status", + [ + (b'{"value":"CANARY_SECRET_input"}', 422), + (b'{"value":"CANARY_SECRET_input",}', 400), + (b"\xffCANARY_SECRET_input", 400), + (b'{"value":NaN}', 422), + (b'{"value":Infinity}', 422), + (b'{"value":-Infinity}', 422), + (b'{"value":1,"value":"CANARY_SECRET_input"}', 422), + (b"[" * 40 + b'"CANARY_SECRET_input"' + b"]" * 40, 422), + ], +) +def test_adversarial_input_returns_sanitized_client_error( + payload: bytes, status: int, caplog: pytest.LogCaptureFixture +) -> None: + # Given constrained models and hostile inputs below a deterministic byte limit. + @validate_http(body=BoundaryModel, max_body_bytes=256, require_json_content_type=True) + def handler(req: HttpRequest) -> None: + pytest.fail("invalid input reached the handler") + + # When invoked; Then only normalized error metadata escapes. + with caplog.at_level(logging.DEBUG): + response = handler( + MockHttpRequest(body=payload, headers={"Content-Type": "application/json"}) + ) + assert response.status_code == status + assert "CANARY_SECRET_input" not in response.get_body().decode() + caplog.text + assert all( + set(entry) == {"loc", "msg", "type"} for entry in json.loads(response.get_body())["detail"] + ) + + +@pytest.mark.parametrize( + "payload", [b"[" * 2000 + b"]" * 2000, b"[0," * 2000, b'"' + b"x" * 4096 + b'"'] +) +def test_oversize_input_is_rejected_before_adapter_or_model_work(payload: bytes) -> None: + # Given oversized nesting/arrays/strings and a narrow parser call-count spy. + adapter = PydanticAdapter() + parse = Mock(wraps=adapter.parse_body) + setattr(adapter, "parse_body", parse) + + @validate_http(body=BoundaryModel, adapter=adapter, max_body_bytes=64) + def handler(req: HttpRequest) -> None: + pytest.fail("oversized input reached the handler") + + # When invoked; Then the pipeline performs no JSON/model parsing. + response = handler(MockHttpRequest(body=payload)) + assert response.status_code == 413 + parse.assert_not_called() + assert len(response.get_body()) < 256 + + +def test_mismatched_media_type_is_rejected_before_adapter_work() -> None: + # Given a JSON-looking body advertised as text and a parser spy. + adapter = PydanticAdapter() + parse = Mock(wraps=adapter.parse_body) + setattr(adapter, "parse_body", parse) + + @validate_http(body=BoundaryModel, adapter=adapter, require_json_content_type=True) + def handler(req: HttpRequest) -> None: + pytest.fail("wrong media type reached handler") + + # When invoked; Then content-type policy rejects it without parsing. + response = handler(MockHttpRequest(body=b'{"value":1}', headers={"Content-Type": "text/plain"})) + assert response.status_code == 415 + parse.assert_not_called() + + +def test_nested_expansion_has_no_implicit_error_count_cap() -> None: + # Given a finite, byte-limited nested request with many invalid array members. + class Nested(BaseModel): + values: list[BoundaryModel] + + @validate_http(body=Nested, max_body_bytes=4096) + def handler(req: HttpRequest) -> None: + pytest.fail("invalid nesting reached handler") + + # When invoked; Then characterize the existing uncapped contract, not a promised cap. + response = handler(MockHttpRequest(json={"values": [{"value": "CANARY_SECRET_input"}] * 32})) + assert response.status_code == 422 + assert len(json.loads(response.get_body())["detail"]) == 32 + assert "CANARY_SECRET_input" not in response.get_body().decode() + + +@pytest.mark.parametrize("depth", [513, 900, 1000, 1500, 2000, 10000]) +@pytest.mark.parametrize("container", ["array", "object"]) +def test_parser_recursion_limit_is_a_client_error(depth: int, container: str) -> None: + # Given valid JSON beyond the portable nesting ceiling, within the byte limit. + opening, closing = (b"[", b"]") if container == "array" else (b'{"x":', b"}") + payload = opening * depth + b'"CANARY_SECRET_input"' + closing * depth + + @validate_http(body=BoundaryModel, max_body_bytes=65536) + def handler(req: HttpRequest) -> None: + pytest.fail("deep nesting reached handler") + + # When parsed; Then client-controlled depth is malformed input, not a server fault. + response = handler(MockHttpRequest(body=payload)) + assert response.status_code == 400 + assert json.loads(response.get_body())["detail"] == [ + {"loc": ["body"], "msg": "Invalid JSON", "type": "value_error"} + ] + assert "CANARY_SECRET_input" not in response.get_body().decode() + + +@pytest.mark.parametrize("depth", [1, 40, 511, 512]) +def test_json_nesting_at_or_below_ceiling_preserves_model_validation(depth: int) -> None: + # Given a list contract accepting arbitrary nested JSON values. + @validate_http(body=list) + def handler(req: HttpRequest) -> None: + return None + + # When parsed; Then allowed nesting still reaches the handler. + response = handler(MockHttpRequest(body=b"[" * depth + b"0" + b"]" * depth)) + assert response.status_code == 204 + + +def test_brackets_in_json_strings_do_not_count_as_nesting() -> None: + # Given escaped quotes and brackets inside a valid field, not JSON containers. + @validate_http(body=BoundaryModel) + def handler(req: HttpRequest) -> None: + return None + + # When parsed; Then string contents do not trigger depth rejection. + response = handler(MockHttpRequest(json={"value": 1, "text": '\\"[' * 2000 + "]" * 2000})) + assert response.status_code == 204 diff --git a/tests/test_security_runtime.py b/tests/test_security_runtime.py new file mode 100644 index 0000000..1094a9e --- /dev/null +++ b/tests/test_security_runtime.py @@ -0,0 +1,170 @@ +"""Offline security proofs through the installed SDK and public HTTP entry point.""" + +import json +import logging +from unittest.mock import Mock + +from azure.functions import HttpRequest, HttpResponse +from pydantic import BaseModel +import pytest + +from azure_functions_validation import HttpError, PydanticAdapter, validate_http +from azure_functions_validation.errors import InternalServerError +from azure_functions_validation.testing import MockHttpRequest + +SECRETS = ( + "CANARY_SECRET_exception", + "DefaultEndpointsProtocol=https;AccountKey=CANARY_SECRET_connection", + "Bearer CANARY_SECRET_token", + "CANARY_SECRET_body", + "CANARY_SECRET_pydantic_input", +) + + +class Output(BaseModel): + value: int + + +class NativeOutput: + """Non-Pydantic marker understood only by a custom backend.""" + + +def request(helper: bool) -> HttpRequest: + payload = json.dumps({"value": SECRETS[3]}).encode() + if helper: + return MockHttpRequest(body=payload) + return HttpRequest("POST", "https://example.test", body=payload) + + +def assert_safe(response: HttpResponse, caplog: pytest.LogCaptureFixture) -> None: + # Server-side logs intentionally keep exception details (exc_info) for operators + # (#250); the trust boundary under test is what the HTTP *response* discloses. + del caplog + observed = response.get_body().decode() + repr(dict(response.headers)) + assert all(secret not in observed for secret in SECRETS) + assert "Traceback (most recent call last)" not in observed + assert "RuntimeError" not in observed + assert "ValidationError" not in observed + assert response.status_code >= 500 + + +@pytest.mark.anyio +@pytest.mark.parametrize("helper", [False, True], ids=["sdk", "testing-helper"]) +@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize( + "failure", + [ + "http-error", + "response", + "parse_body", + "parse_query", + "parse_path", + "parse_headers", + "serialize", + "serialize_response", + "format_error", + "native-backend", + ], +) +async def test_5xx_boundaries_never_echo_secrets_in_the_response( + caplog: pytest.LogCaptureFixture, helper: bool, asynchronous: bool, failure: str +) -> None: + # Given real SDK requests and a secret-bearing fault at each extension seam. + adapter = PydanticAdapter() + model = NativeOutput if failure == "native-backend" else Output + if failure.startswith("parse_") or failure in { + "serialize", + "serialize_response", + "format_error", + }: + setattr(adapter, failure, Mock(side_effect=RuntimeError(*SECRETS))) + if failure == "native-backend": + setattr(adapter, "validate_response", Mock(side_effect=RuntimeError(*SECRETS))) + + def handler(req: HttpRequest) -> dict[str, str]: + if failure == "http-error": + raise HttpError(503, list(SECRETS)) + return {"value": SECRETS[4]} + + async def async_handler(req: HttpRequest) -> dict[str, str]: + return handler(req) + + wrapped = validate_http( + adapter=adapter, + response_model=model + if failure in {"response", "serialize_response", "native-backend"} + else None, + body=Output if failure in {"parse_body", "format_error"} else None, + query=Output if failure == "parse_query" else None, + path=Output if failure == "parse_path" else None, + headers=Output if failure == "parse_headers" else None, + )(async_handler if asynchronous else handler) + # serialize_response must be reached after successful response validation. + if failure == "serialize_response": + setattr(adapter, "validate_response", Mock(return_value=Output(value=1))) + # When the public decorator is invoked; Then the transport never echoes secrets. + with caplog.at_level(logging.DEBUG): + response = await wrapped(request(helper)) if asynchronous else wrapped(request(helper)) + assert_safe(response, caplog) + + +@pytest.mark.anyio +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("expose", [False, True]) +@pytest.mark.parametrize("formatter_mode", ["raises", "unserializable"]) +async def test_failed_formatter_is_safe_even_with_debug_opt_in( + caplog: pytest.LogCaptureFixture, asynchronous: bool, expose: bool, formatter_mode: str +) -> None: + # Given a formatter that fails after receiving an original or sanitized exception. + def formatter(exc: Exception, status: int) -> dict[str, str | set[str]]: + if formatter_mode == "raises": + raise RuntimeError(*SECRETS) + return {"value": set(SECRETS)} + + def handler(req: HttpRequest) -> None: + raise HttpError(500, list(SECRETS)) + + async def async_handler(req: HttpRequest) -> None: + handler(req) + + wrapped = validate_http(error_formatter=formatter, expose_internal_errors=expose)( + async_handler if asynchronous else handler + ) + # When invoked; Then even formatter exceptions/serialization cannot leak. + with caplog.at_level(logging.DEBUG): + response = await wrapped(request(True)) if asynchronous else wrapped(request(True)) + assert_safe(response, caplog) + + +@pytest.mark.parametrize("expose", [False, True]) +def test_debug_opt_in_controls_only_formatter_access(expose: bool) -> None: + # Given an explicitly redacting formatter, including with the unsafe opt-in. + captured: list[Exception] = [] + + def formatter(exc: Exception, status: int) -> dict[str, str]: + captured.append(exc) + return {"error": "redacted"} + + original = HttpError(500, SECRETS[0]) + + @validate_http(error_formatter=formatter, expose_internal_errors=expose) + def handler(req: HttpRequest) -> None: + raise original + + # When invoked; Then opt-in never changes the fallback or performs automatic redaction. + response = handler(request(True)) + assert json.loads(response.get_body()) == {"error": "redacted"} + assert (captured[0] is original) == expose + assert expose or isinstance(captured[0], InternalServerError) + + +def test_internal_error_exposure_defaults_off() -> None: + # Given no debugging option; When a formatter sees a 5xx; Then it sees no original. + def formatter(exc: Exception, status: int) -> dict[str, str]: + return {"error": str(exc)} + + @validate_http(error_formatter=formatter) + def handler(req: HttpRequest) -> None: + raise HttpError(500, SECRETS[0]) + + assert json.loads(handler(request(True)).get_body()) == {"error": "Internal Server Error"} diff --git a/tests/test_security_workflows.py b/tests/test_security_workflows.py new file mode 100644 index 0000000..4110ce8 --- /dev/null +++ b/tests/test_security_workflows.py @@ -0,0 +1,93 @@ +import os +from pathlib import Path +import subprocess +import sys + +import yaml + +from tools.lint_release_workflows import check_pins, check_publish_needs + +ROOT = Path(__file__).resolve().parents[1] + + +def test_release_audit_is_blocking_and_uses_candidate_artifact() -> 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"]) + download = next(s for s in gate["steps"] if "download-artifact@" in s.get("uses", "")) + assert download["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-audit" in commands + assert 'CANDIDATE_WHEEL="${wheels[0]}"' in commands + + +def test_evidence_uploads_run_on_failure_and_keep_permissions_narrow() -> 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) == [] + uploads = [ + step + 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 uploads + assert all( + s["if"] == "always()" and s["with"]["if-no-files-found"] == "error" for s in uploads + ) + + +def test_worker_variants_include_offline_security_proofs() -> None: + for filename in ("worker-nightly.yml", "ci-test.yml"): + text = (ROOT / ".github/workflows" / filename).read_text() + assert "tests/test_security_runtime.py" in text + assert "tests/test_security_auth.py" in text + + +def test_reporting_lane_reuses_policy_without_weakening_release() -> None: + workflow = yaml.safe_load((ROOT / ".github/workflows/dependency-audit.yml").read_text()) + steps = workflow["jobs"]["pip-audit"]["steps"] + audits = [s for s in steps if "make security-audit-report" in s.get("run", "")] + assert len(audits) == 2 + assert all(s["continue-on-error"] for s in audits) + assert any("--development" in s["run"] for s in audits) + + +def test_collectors_run_even_after_scanner_failure() -> None: + workflow = yaml.safe_load((ROOT / ".github/workflows/security.yml").read_text()) + for job in ("bandit", "semgrep"): + collectors = [ + s for s in workflow["jobs"][job]["steps"] if f"make security-{job}" in s.get("run", "") + ] + assert len(collectors) == 1 + assert collectors[0]["if"] == "always()" + + +def test_semgrep_make_tolerates_only_incomplete_evidence(tmp_path: Path) -> None: + for outcome, succeeds in (("success", True), ("failure", False)): + result = subprocess.run( + [ + "make", + "security-semgrep", + f"SECURITY_PYTHON={sys.executable}", + f"SECURITY_OUTPUT={tmp_path / outcome}", + ], + cwd=ROOT, + env=dict(os.environ, SCAN_OUTCOME=outcome), + capture_output=True, + text=True, + ) + assert (result.returncode == 0) == succeeds + assert "action outcome only" in result.stdout diff --git a/tools/dependency_audit_policy.py b/tools/dependency_audit_policy.py new file mode 100644 index 0000000..ce4fb0c --- /dev/null +++ b/tools/dependency_audit_policy.py @@ -0,0 +1,208 @@ +"""Audit only an isolated wheel environment, with fail-closed redacted evidence.""" + +from __future__ import annotations + +import argparse +from dataclasses import replace +from datetime import date +from importlib.metadata import PackageNotFoundError, version +from pathlib import Path +import re +import subprocess +import sys +import tempfile +from zipfile import BadZipFile + +from tools.dependency_inventory import installed_inventory, normalize, wheel_identity +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 exit_status not in {0, 1}: + raise EvidenceError("tool crash") + if not dependencies or any("skip_reason" in mapping(d) for d in dependencies): + return replace( + result, status=Status.INCOMPLETE, message="Empty or skipped audit scope." + ) + findings: list[Finding] = [] + applied: set[str] = set() + blocking = 0 + for dependency in dependencies: + item = mapping(dependency) + package = re.sub(r"[-_.]+", "-", identifier(item["name"]).lower()) + text(item["version"]) + 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 re.sub(r"[-_.]+", "-", r.package.lower()) == 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": len(dependencies), + "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 runtime advisories block without a current applicable exception.", + ) + except (ValueError, KeyError, TypeError): + return replace(result, message="Invalid audit/tool/policy evidence; release blocked.") + + +def run_audit(site_packages: Path, policy: Path, candidate: tuple[str, str]) -> Check: + try: + inventory = installed_inventory(site_packages, candidate) + except (OSError, ValueError, TypeError, KeyError): + return Check( + "pip-audit", + ("resolved-runtime-wheel",), + Status.INCOMPLETE, + None, + message="Candidate identity mismatch or missing third-party inventory.", + ) + try: + with tempfile.TemporaryDirectory(prefix="candidate-audit-") as temporary: + requirements = Path(temporary) / "requirements.txt" + requirements.write_text( + "".join(f"{name}=={resolved}\n" for name, resolved in sorted(inventory.items())) + ) + 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, + ) + result = classify_audit( + process.stdout, + process.returncode, + policy_raw=policy.read_text(encoding="utf-8"), + today=date.today(), + ) + if result.status in {Status.PASS, Status.FAIL}: + observed: dict[str, str] = {} + for entry in sequence(load(process.stdout)["dependencies"]): + item = mapping(entry) + name, resolved = normalize(text(item["name"])), text(item["version"]) + if name in observed: + raise EvidenceError("duplicate audit dependency") + observed[name] = resolved + candidate_name, candidate_version = candidate + if normalize(candidate_name) in observed: + if observed.pop(normalize(candidate_name)) != candidate_version: + raise EvidenceError("audit candidate identity mismatch") + if observed != inventory: + result = replace( + result, status=Status.INCOMPLETE, message="Audit inventory mismatch." + ) + return replace(result, tool_versions={"pip-audit": version("pip-audit")}) + except ( + OSError, + subprocess.TimeoutExpired, + PackageNotFoundError, + 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( + "--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") + parser.add_argument("--candidate-wheel", type=Path) + parser.add_argument("--candidate-name") + parser.add_argument("--candidate-version") + args = parser.parse_args() + try: + if args.candidate_wheel: + candidate = wheel_identity(args.candidate_wheel) + elif args.candidate_name and args.candidate_version: + candidate = (args.candidate_name, args.candidate_version) + else: + raise EvidenceError("candidate identity required") + check = run_audit(args.site_packages, args.policy, candidate) + except (OSError, ValueError, BadZipFile): + check = Check( + "pip-audit", + ("resolved-runtime-wheel",), + Status.INCOMPLETE, + None, + message="Candidate wheel identity unavailable.", + ) + 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/dependency_inventory.py b/tools/dependency_inventory.py new file mode 100644 index 0000000..aa8f06e --- /dev/null +++ b/tools/dependency_inventory.py @@ -0,0 +1,36 @@ +"""Verified wheel identity and exact installed dependency inventory.""" + +from email.parser import BytesParser +from importlib.metadata import distributions +from pathlib import Path +import re +from zipfile import ZipFile + +from tools.security_report import EvidenceError, identifier + + +def normalize(name: str) -> str: + return re.sub(r"[-_.]+", "-", identifier(name).lower()) + + +def wheel_identity(path: Path) -> tuple[str, str]: + with ZipFile(path) as wheel: + metadata = [n for n in wheel.namelist() if n.endswith(".dist-info/METADATA")] + if len(metadata) != 1: + raise EvidenceError("candidate wheel metadata missing or ambiguous") + parsed = BytesParser().parsebytes(wheel.read(metadata[0])) + return normalize(str(parsed["Name"])), identifier(str(parsed["Version"])) + + +def installed_inventory(site: Path, candidate: tuple[str, str]) -> dict[str, str]: + inventory: dict[str, str] = {} + for distribution in distributions(path=[str(site)]): + name = normalize(distribution.metadata["Name"]) + resolved = identifier(distribution.version) + if name in inventory: + raise EvidenceError("duplicate installed distribution") + inventory[name] = resolved + name, expected = candidate + if inventory.pop(normalize(name), None) != expected or not inventory: + raise EvidenceError("candidate mismatch or empty third-party inventory") + return inventory diff --git a/tools/security_evidence.py b/tools/security_evidence.py new file mode 100644 index 0000000..551b3c9 --- /dev/null +++ b/tools/security_evidence.py @@ -0,0 +1,213 @@ +"""Execute Bandit's explicit product inventory and record Semgrep action evidence.""" + +from __future__ import annotations + +import argparse +from dataclasses import replace +from datetime import date +from importlib.metadata import PackageNotFoundError, version +from pathlib import Path +import re +import subprocess +import sys +import tokenize + +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", "scripts", "tools") +FILE_FLOOR = 30 + + +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, confidence = text(item["issue_severity"]), 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/error files; scan unusable.") + if len(files) < floor or counts["loc"] == 0: + return replace(result, status=Status.INCOMPLETE, message="Scan below file/LOC floor.") + 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="Invalid Bandit JSON; raw diagnostics discarded.") + + +def apply_bandit_policy(check: Check, policy_raw: str, today: date) -> Check: + if check.status not in {Status.PASS, Status.FAIL}: + return check + try: + policy = load(policy_raw) + if policy["schema_version"] != 1: + raise EvidenceError("unsupported policy") + rules = parse_exceptions(policy["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: + if any( + (r.id, r.file, r.line) == (finding.id, finding.file, finding.line) for r in rules + ): + 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", "--ignore-nosec"], + 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 inventory differs from declared surfaces.", + ) + policy = Path("security/bandit-policy.json").read_text(encoding="utf-8") + result = apply_bandit_policy(result, policy, date.today()) + rules = parse_exceptions(load(policy)["exceptions"], date.today()) + suppressions = 0 + for filename in expected: + with tokenize.open(filename) as stream: + for token in tokenize.generate_tokens(stream.readline): + if token.type != tokenize.COMMENT or "nosec" not in token.string: + continue + ids = re.findall(r"B\d{3}", token.string) + suppressions += 1 + if not ids or any( + not any( + (r.id, r.file, r.line) == (rule_id, filename, token.start[0]) + for r in rules + ) + for rule_id in ids + ): + result = replace( + result, + status=Status.INCOMPLETE, + message="Unlisted inline suppression; scan incomplete.", + ) + result = replace(result, suppressions=suppressions) + return replace(result, tool_versions={"bandit": version("bandit")}) + except ( + OSError, + subprocess.TimeoutExpired, + ValueError, + KeyError, + TypeError, + PackageNotFoundError, + ): + return Check( + "bandit", + SCOPE, + Status.TOOL_ERROR, + None, + message="Bandit execution/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; finding count unknown; failures 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 0000000..2162a63 --- /dev/null +++ b/tools/security_exceptions.py @@ -0,0 +1,49 @@ +"""Owned, exact exceptions with a maximum 90-day validity window.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date + +from tools.security_report import JSON, EvidenceError, identifier, integer, 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") + rule = ExceptionRule( + identifier(item["id"]), + identifier(item["package"]) if "package" in item else "", + identifier(item["file"]) if "file" in item else "", + integer(item.get("line", 0)), + text(item["kind"]), + expires, + added, + ) + if rule.kind not in {"fix-not-available", "not-affected", "reviewed-static"}: + raise EvidenceError("unknown exception kind") + if any( + (r.id, r.package, r.file, r.line) == (rule.id, rule.package, rule.file, rule.line) + for r 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 0000000..0692c5f --- /dev/null +++ b/tools/security_report.py @@ -0,0 +1,120 @@ +"""Allow-listed security evidence; raw scanner diagnostics are never retained.""" + +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): + """Invalid evidence; messages contain no external 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)