diff --git a/README.md b/README.md index 0a7f1f0..20a2ea9 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,25 @@ Share feedback in [GitHub Discussions](https://github.com/coreytshaffer/TriageCo - Supports local benchmark fixtures and benchmark reports without hiding the evidence trail. - Enforces local-only privacy boundaries before any optional external-safe Qwen Cloud path is considered. +### Experimental WebAuthn Authorization + +CR-YK-001 prototypes request-bound WebAuthn authorization receipts. Two +independently enrolled physical YubiKeys are the optimal production and +recovery posture. When a second hardware key is unavailable, TriageCore +supports a separately enrolled and labeled cross-device phone WebAuthn +passkey as a bounded secondary pathway. The phone passkey signs the same +request-bound WebAuthn challenge and can be verified offline, but it is not a +second hardware security key and must not be described as +assurance-equivalent. Google Authenticator/TOTP codes are not a substitute +for either WebAuthn pathway. + +Primary YubiKey enrollment and assertion ceremonies verified on Windows. A phone passkey was also validated as a secondary cross-device pathway. Redundant backup-YubiKey enrollment remains unverified. + +This feature remains experimental: execution integration and atomic +capability consumption are not complete. See +[CR-YK-001](docs/change/requests/CR-YK-001-hardware-authorization-receipts.md) +for evidence and limitations. + ## Evidence-Bound Build Review Build Review turns a development request and Git comparison into a local, diff --git a/docs/change/requests/CR-YK-001-hardware-authorization-receipts.md b/docs/change/requests/CR-YK-001-hardware-authorization-receipts.md new file mode 100644 index 0000000..3f79428 --- /dev/null +++ b/docs/change/requests/CR-YK-001-hardware-authorization-receipts.md @@ -0,0 +1,350 @@ +# CR-YK-001: FIDO2 Security-Key-Backed Human Authorization Receipts (YubiKey Hackathon Lane) + +## Status + +Primary-path hardware verification completed 2026-07-24, targeting the +YubiKey hackathon on 2026-08-05. + +Primary YubiKey enrollment and assertion ceremonies verified on Windows. A phone passkey was also validated as a secondary cross-device pathway. Redundant backup-YubiKey enrollment remains unverified. + +Verification ran on Microsoft Windows 11 Pro 64-bit, version +10.0.26200, build 26200, with `fido2` 2.2.1 through Windows native WebAuthn +on branch `cr-yk-001-hardware-authz` at `0424442`, which is patch-identical +to the previously recorded `295a119`. Separate lane (CR-YK-*) so it cannot +entangle the CR-DD sequence, mirroring how Build Week ran as its own bounded +slice. + +Component states (kept honest and separate): + +| # | Component | State | +|---|-----------|-------| +| 1 | Deterministic authorization core (`triage_core/authz.py`) | **Implemented and tested** (challenge derivation, receipts, tamper detection, capability lane, credential store with revocation) | +| 2 | WebAuthn adapter (`triage_core/fido2_adapter.py`) | **Implemented and tested against python-fido2 2.2.1**; offline verification exercised with real fido2 data structures, a synthetic software credential, and the physical ceremonies below | +| 3 | Primary physical YubiKey enrollment ceremony | **Verified 2026-07-24** through Windows native WebAuthn with explicit cross-platform authenticator selection | +| 4 | Primary physical YubiKey assertion ceremony | **Verified 2026-07-24** for both PIN-required UV and corrected touch-only operation; offline verification passed | +| 5 | Secondary cross-device phone WebAuthn passkey | **Verified 2026-07-24** as the bounded hackathon-lane secondary credential; it is not a second hardware security key | +| 6 | Redundant backup-YubiKey enrollment | **UNVERIFIED** — a second physical YubiKey was unavailable; the phone passkey is not assurance-equivalent | +| 7 | CR-DD-012B execution integration (`tc run --confirmed-plan`) | **Explicitly out of scope** for this CR and not modified during hardware smoke testing | +| 8 | Atomic capability consumption | **Not implemented**; documented limitation with a strict expected-failure test; required before 012B integration | + +## Decision (bounded allowlist) + +This CR authorizes, and only authorizes: + +1. `triage_core/authz.py`: canonical versioned authorization requests, + challenge derivation, receipt sidecar artifacts with tamper detection and + restrictive permissions, structural verification, the one-use + execution-capability lane with a closed denial vocabulary, and the + public-credential store with revocation tombstones. +2. `triage_core/fido2_adapter.py`: all python-fido2 and platform-specific + code — availability probe, enrollment and assertion ceremonies, and full + offline verification via `Fido2Server`. +3. `tests/test_authz.py`: deterministic-core tests plus fido2-backed + verification tests using synthetic software credentials. +4. The optional dependency extra `authz = ["fido2>=2.0,<3"]`. Absence of + the extra must not break import, and non-Windows platforms must receive a + controlled unavailability result, never an import crash + (`fido2.client.windows` cannot even be imported off-Windows). +5. Hackathon-day CLI additions listed under "Planned CLI surface". + +Not authorized here: any change to `tc run`, the resilience router, cloud +paths, TriageDesk, CR-DD-012B implementation files, or atomic capability +storage. + +## Problem + +TriageCore separates system recommendation from human decision, but a +recorded human decision is currently a ledger event anyone with file access +could have written. For governed agent actions the missing property is +phishing-resistant evidence that the enrolled human performed a fresh, +deliberate approval ceremony over one exact plan — the standard mitigation +for excessive agency in current OWASP agentic guidance, made cryptographic. + +## Design + +### Binding chain + +``` +run-plan artifact (CR-DD-011) existing: canonical bytes + digests + │ artifact_byte_digest / plan_body_digest + ▼ +AuthorizationRequest (schema v2) versioned canonical JSON binding: + │ schema id, decision_id, both plan + │ digests, scope/policy digest (when + │ available), task_id, risk class, + │ policy version, approver, RP ID, + │ UV-required policy, nonce (UUID), + │ issued-at, expiry + ▼ +challenge = SHA-256(canonical_json(request)) + ▼ +WebAuthn get-assertion user presence always; user +(YubiKey or bounded phone secondary) verification required when the request + │ says so + ▼ +HumanAuthorizationReceipt (schema v2) request + standard WebAuthn JSON + │ wire-format assertion; sidecar file + │ under .triagecore/authz/receipts/ + │ with owner-only permissions + ▼ +ledger: human_authorization_receipt digests + privacy-safe metadata only + ▼ +execution capability (one-use, TTL) issued → consumed | denied + reasons: capability_not_found | + capability_already_consumed | + capability_expired | + artifact_digest_mismatch +``` + +Versioning and domain separation live in the canonical payload's `schema` +field rather than raw byte concatenation, so what a signature binds is +explicit and auditable in the serialized form. Key-sorted canonical JSON +makes the digest independent of field construction order (tested). + +### Identity and normalization + +`approver_identity_id` is the stable registered identifier of the approving +human and is the only approver value that is security-bound. Human-facing +display labels are informational, live in the credential store +(`EnrolledCredential.label`), and never enter the challenge. Verification +additionally requires the asserting credential to be enrolled to the same +identity the request names (`credential_approver_mismatch` otherwise), so +one enrolled human's key cannot satisfy another human's request. + +The credential store enforces the identical normalization contract: +`EnrolledCredential` normalizes and validates `human_id` at construction, +which covers enrollment, lookup (`find_by_human_id` normalizes its query), +revocation, and serialization uniformly. Cosmetic variants such as +`Operator-A` and `operator-a` therefore resolve to one governed identity — +they cannot enroll as distinct identities, and a request naming +`" operator-a "` verifies against a credential enrolled as `OPERATOR-A` +(both invariants are tested). + +Canonical JSON fixes field ordering but not semantics, so construction +normalizes before hashing (tested for equivalence and rejection): + +- `risk_class`, `policy_version`, `approver_identity_id`, `rp_id`: + strip + lowercase, then validated — risk class against the closed + `{low, medium, high}` vocabulary, the rest against a constrained + identifier grammar. `"HIGH"`, `"high"`, and `"high "` produce one + challenge; `"critical"` is rejected. +- `task_id`, `decision_id`: whitespace-stripped and non-empty with no + embedded whitespace; case is preserved deliberately because ledger task + lookups are case-exact and operator-supplied IDs (e.g. `CR-017`) must + keep their linkage. +- digests: must match `sha256:<64 lowercase hex>`; `scope_digest` may be + empty until a scope artifact exists. +- `nonce`: any accepted UUID form is stored in canonical lowercase form. +- timestamps: round-tripped through ISO-8601 parsing to one canonical + serialization; unparseable values are rejected at construction. + +### Offline verification + +Verification layers, in order: + +1. **Structural** (dependency-free): request expiry; clientDataJSON parses; + `type == "webauthn.get"`; challenge equality against the recomputed + request digest; pinned origin. +2. **Credential policy**: credential is enrolled, not revoked, matches the + request's RP, and belongs to the request's `approver_identity_id`. +3. **Cryptographic**, via `fido2.server.Fido2Server.authenticate_complete` + (python-fido2's own server-side primitive, not re-implemented WebAuthn + parsing): RP-ID hash, user-presence flag, user-verification flag when + the request required it, credential membership, and the signature over + `authenticatorData || SHA256(clientDataJSON)`. + +Signature counters are deliberately not enforced; zero counters are valid +(common on current keys). Verification is reproducible offline from the +receipt artifact, the enrolled COSE public key, and the ledger — the same +independent-verifiability posture as `tc build-review verify`. Tamper +detection closes the loop: the ledger event stores the receipt digest, and +`read_receipt_artifact(path, expected_digest=...)` rejects drifted bytes. + +### Evidence honesty + +Primary YubiKey enrollment and assertion ceremonies verified on Windows. A phone passkey was also validated as a secondary cross-device pathway. Redundant backup-YubiKey enrollment remains unverified. + +A verified receipt is *WebAuthn-backed authorization*: evidence of +possession and user interaction with an enrolled credential, over exactly +one request digest. A receipt from the primary YubiKey is additionally +security-key-backed. A receipt from the bounded secondary phone credential +is a cross-device passkey receipt and must not be described as coming from a +second hardware security key. "User verification confirmed" is claimed only +when UV was required and the flag was verified. + +Not claimed: non-repudiation; cryptographic proof of the YubiKey model or +firmware (attestation was not requested or validated); comprehension of what +was approved (which is why the approval flow must render `render_run_plan` +output, never a bare digest). The operator reports that the tested primary is +a developer early-release firmware 5.8 YubiKey and that only one unit will be +provided. No serial is recorded. + +### Threat-model limitations (explicit) + +- **Host clock manipulation**: issued-at, expiry, and capability TTLs read + the local clock. An attacker (or operator) who shifts the host clock can + extend or revive validity windows. Receipts remain digest-bound, so the + *content* approved cannot change — only the timing gates weaken. +- **Local-host compromise**: an attacker with code execution on this host + can present a misleading plan render before the ceremony, request + approval for content the human did not intend, read sidecar artifacts, + or append forged non-signed ledger events. The security key defeats + remote credential phishing and remote approval forgery; it does not + defend a compromised local machine. Mitigations (out of scope here): + signed ledger events via the existing identity registry, and reviewing + plans on a second device. +- **Concurrency**: single-use capability enforcement is sequential-only in + this slice (see below). + +### Capability safety (known gap) + +`consume_capability` is read-then-append over a shared JSONL file. Two +concurrent consumers can both observe "unconsumed" before either appends, +so an append-only "consumed" event alone does not prove race-safe single +use, and the sequential unit tests do not claim it. A strict +expected-failure test (`test_concurrent_consume_race_documented`) +deterministically demonstrates the interleaving. **Atomic claiming — e.g. a +SQLite claim table using `BEGIN IMMEDIATE`, or an O_EXCL claim file — is a +requirement for CR-DD-012B integration** and is proposed as its own bounded +slice (CR-YK-002); it is not implemented under this CR. + +Agreed CR-YK-002 design sketch (recorded for continuity, not authorized by +this CR): explicit capability states `issued → claimed → completed | +failed`, where a claim is irrevocable for the original capability — a +failed or crashed execution does not silently restore reusability; a new +authorization is required unless a future recovery policy explicitly +permits otherwise. The atomic claim is a SQLite `BEGIN IMMEDIATE` +transaction performing `UPDATE ... SET state='claimed', claimed_at=?, +claimant_id=? WHERE capability_id=? AND state='issued' AND expires_at > ?` +and requiring exactly one changed row. The store binds capability ID, +receipt digest, approved artifact digest, execution-scope digest, expiry, +state, claimant / execution-attempt ID, claimed timestamp, and terminal +outcome timestamp. SQLite becomes the authoritative concurrency-control +mechanism; the ledger remains the durable signed evidence history. + +### Sidecar retention + +Receipts persist until the operator prunes them. They are inputs to offline +re-verification, so deletion degrades auditability of the runs they +authorized; recommended practice is to retain receipts at least as long as +the ledger covering the same tasks. Credentials use revocation tombstones, +never deletion, so historical receipts stay attributable. Neither artifact +ever contains PINs, private keys, or secrets. + +## Planned CLI surface (hackathon day) + +- `tc authz enroll --human --label ` — make-credential ceremony; + appends to `.triagecore/authz/credentials.json`. For this bounded hackathon + lane, enroll the one available developer YubiKey and a separately labeled + cross-device phone WebAuthn passkey as the secondary credential. The phone + credential is not a second hardware security key. Google Authenticator TOTP + is not accepted as a substitute. +- `tc authz approve --plan-artifact ` — render plan, build request + (UV required for high risk), get assertion, record receipt, issue + capability. +- `tc authz verify --receipt ` — offline verification against the + store and ledger digest. +- `tc authz revoke --credential ` — tombstone a lost key. +- `tc authz exec --capability ...` — demo wrapper that consumes the + capability then invokes the existing governed run path. Demo scope only; + the durable integration point is CR-DD-012B's `--confirmed-plan`. + +## Hardware verification checklist (executed 2026-07-24) + +On the Windows verification machine with the physical YubiKey and the +separately labeled phone passkey: + +1. `pip install -e ".[authz]"`; confirm `fido2` reports version 2.2.1. +2. `python -c "from triage_core.fido2_adapter import ceremony_support; print(ceremony_support())"` + → expect `windows_native`, available. +3. Enroll the primary YubiKey and the bounded secondary cross-device phone + passkey as separately labeled credentials; confirm entries contain + credential ID, COSE key, AAGUID, and nothing secret. This one-YubiKey-plus- + phone policy is a bounded amendment for this lane, not a claim that the + phone is a backup hardware security key. +4. Assertion ceremony over a real run-plan request with UV required (PIN + plus touch for the YubiKey) and with UV not required (YubiKey touch only, + with no PIN). +5. `verify_receipt` passes on both; then repeat all failure paths from the + demo script (tamper, expiry, replay, revoked key, wrong key). +6. Exercise the phone credential with REQUIRED UV through Windows QR/ + cross-device WebAuthn and confirm device unlock, `user_verified == true`, + and offline verification. + +### Hardware verification record + +- The first registration selected Windows Hello because the creation options + did not constrain authenticator attachment. It enrolled AAGUID + `08987058-cadc-4b81-b6e1-30de50dcbe96`, was quarantined in the isolated + `.triagecore/hardware-smoke` store, and did not count as YubiKey enrollment. +- Enrollment was corrected to require + `AuthenticatorAttachment.CROSS_PLATFORM`. Primary YubiKey enrollment then + succeeded in the isolated `.triagecore/hardware-smoke-yubikey` store. +- The high-risk YubiKey assertion used REQUIRED UV. PIN and touch were + observed, `user_verified` was true, and offline verification passed. +- The initial non-required assertion used PREFERRED, still prompted for a + PIN, and did not count as the touch-only result. The false UV policy was + corrected to DISCOURAGED in both the live assertion request and the + reconstructed verifier state. +- The corrected touch-only YubiKey assertion presented no PIN prompt. Touch + was observed, `user_verified` was false, and offline verification passed. +- Because only one developer YubiKey is available, a separately labeled + cross-device phone WebAuthn passkey is accepted as the secondary credential + for this bounded hackathon lane. Phone enrollment through the Windows QR/ + cross-device flow succeeded. Its high-risk REQUIRED-UV assertion succeeded + with device unlock, `user_verified` true, and offline verification passed. + This is not a second hardware security key; Google Authenticator TOTP is + not an accepted substitute. +- The offline rejection matrix passed for artifact tamper, request expiry, + wrong public key, a revoked credential copy, sequential replay denial, and + capability expiry. +- Focused tests: **64 passed, 1 skipped, 1 xfailed**. Full suite: + **1151 passed, 5 skipped, 1 xfailed**. +- The main `.triagecore/authz` credential store and execution integration were + not modified. All smoke artifacts remain isolated. + +Pre-hackathon note: audit finding F1 (DangerDetector substring `auth`/ +`token`) still blocks demo prompts about authorization from reaching the +approval gate; fix it first. + +## Demo script (~3 minutes) + +1. `tc run "" --plan --plan-output plan.json` — show digests. +2. `tc authz approve --plan-artifact plan.json` — plan renders, OS WebAuthn + dialog, touch key; show receipt artifact + metadata-only ledger event. +3. `tc authz exec` — capability consumed once; action runs. +4. Re-run — denied `capability_already_consumed`; denial is evidence. +5. Tamper one byte of the plan — `client_data_challenge_mismatch`. +6. Force expiry — `capability_expired`. +7. Revoke the key, verify again — `credential_revoked`. +8. Close on `tc task show `: propose → approve → capability → execute + → deny chain. + +## Verification path + +- Deterministic + synthetic-credential: `python -m pytest tests/test_authz.py -q` + (challenge determinism, field-order independence, normalization + equivalence and non-canonical rejection, per-field binding + including UV policy, approver identity, and scope digest, + approver-credential mismatch, malformed/typed/challenge/origin + clientData rejections, expiry, artifact tamper detection and permissions, + privacy-invariant compliance of all ledger payloads, capability lifecycle + and denials, strict-xfail concurrency demonstration, dependency-missing + and platform-unavailable handling, and full fido2-backed verification + positive/negative matrix: zero counter accepted, missing UP, missing + required UV, UP-only when UV not required, wrong RP-ID hash, altered + authenticator data, altered signature, wrong key, unknown credential, + revoked credential). Latest focused result: **64 passed, 1 skipped, + 1 xfailed**. +- Full suite: **1151 passed, 5 skipped, 1 xfailed**. +- Hardware: the checklist above passed on 2026-07-24 through Windows native + WebAuthn with the primary YubiKey and bounded secondary cross-device phone + passkey. + +## Non-goals + +- No change to routing, privacy gating, cloud enablement, or TriageDesk. +- No CR-DD-012B implementation or `--confirmed-plan` wiring. +- No atomic capability storage in this slice (separate bounded change). +- No attestation-chain validation of authenticator provenance. +- No multi-approver quorum policies. diff --git a/pyproject.toml b/pyproject.toml index 437d703..c980e8e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,6 +36,9 @@ mobile = [ "pydantic>=2.0.0", "httpx2>=0.28.0" ] +authz = [ + "fido2>=2.0,<3" +] [project.scripts] triagecore = "triage_core.cli:main" diff --git a/tests/test_authz.py b/tests/test_authz.py new file mode 100644 index 0000000..c5497db --- /dev/null +++ b/tests/test_authz.py @@ -0,0 +1,727 @@ +"""Tests for the CR-YK-001 authorization-receipt lane. + +Two tiers: + +1. Deterministic-core tests: no python-fido2, no hardware, no network. +2. fido2-backed tests: use real python-fido2 2.x WebAuthn data structures + with a synthetic software credential. These skip cleanly when the + optional dependency is absent. They prove offline verification behavior + only — they do NOT prove Windows integration or a physical-YubiKey + ceremony (that remains a manual checklist in CR-YK-001). + +The concurrency test encodes a known limitation as a strict expected +failure: sequential unit tests cannot and do not prove race-safe single use. +""" + +import json +import os +import stat +import sys +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +import pytest + +from triage_core import fido2_adapter +from triage_core.authz import ( + EVENT_CAPABILITY_CONSUMED, + EVENT_CAPABILITY_DENIED, + EVENT_RECEIPT_RECORDED, + ORIGIN, + REASON_ALREADY_CONSUMED, + REASON_DIGEST_MISMATCH, + REASON_EXPIRED, + REASON_NOT_FOUND, + REASON_OK, + REQUEST_SCHEMA, + RP_ID, + STRUCTURAL_FAIL_CHALLENGE, + STRUCTURAL_FAIL_EXPIRED, + STRUCTURAL_FAIL_MALFORMED, + STRUCTURAL_FAIL_ORIGIN, + STRUCTURAL_FAIL_TYPE, + AuthorizationRequest, + AuthzArtifactTamperError, + AuthzError, + AuthzHardwareUnavailable, + AuthzVerificationError, + CredentialStore, + EnrolledCredential, + HumanAuthorizationReceipt, + b64url_encode, + build_receipt_ledger_payload, + compute_challenge, + consume_capability, + issue_capability, + read_receipt_artifact, + record_receipt, + verify_receipt_structure, + write_receipt_artifact, +) +from triage_core.privacy_invariants import assert_persistent_privacy_safe +from triage_core.task_ledger import TaskLedger + +NOW = datetime(2026, 8, 5, 15, 5, 0, tzinfo=timezone.utc) + +BASE_REQUEST_KWARGS = dict( + decision_id="gd-1234567890abcdef", + artifact_byte_digest="sha256:" + "ab" * 32, + plan_body_digest="sha256:" + "cd" * 32, + task_id="11111111-2222-3333-4444-555555555555", + risk_class="high", + policy_version="cr-yk-001-v1", + approver_identity_id="human-corey", + user_verification_required=True, + scope_digest="sha256:" + "ee" * 32, + nonce="99999999-8888-7777-6666-555555555555", + issued_at="2026-08-05T15:00:00+00:00", + expires_at="2026-08-05T15:10:00+00:00", +) + + +def _request(**overrides) -> AuthorizationRequest: + kwargs = dict(BASE_REQUEST_KWARGS) + kwargs.update(overrides) + return AuthorizationRequest(**kwargs) + + +def _wire_assertion(client_data_bytes: bytes) -> dict: + """Minimal wire-shaped assertion for structural tests (no crypto).""" + return { + "id": b64url_encode(b"cred-id"), + "rawId": b64url_encode(b"cred-id"), + "type": "public-key", + "response": { + "clientDataJSON": b64url_encode(client_data_bytes), + "authenticatorData": b64url_encode(b"auth-data"), + "signature": b64url_encode(b"sig"), + }, + } + + +def _receipt(request: AuthorizationRequest, challenge_b64: str = None, + origin: str = ORIGIN, cd_type: str = "webauthn.get", + client_data_bytes: bytes = None) -> HumanAuthorizationReceipt: + if client_data_bytes is None: + challenge = challenge_b64 or b64url_encode(compute_challenge(request)) + client_data_bytes = json.dumps( + {"type": cd_type, "challenge": challenge, "origin": origin} + ).encode("utf-8") + return HumanAuthorizationReceipt( + request=request, + credential_id=b64url_encode(b"cred-id"), + assertion_response=_wire_assertion(client_data_bytes), + user_verified=True, + recorded_at="2026-08-05T15:01:00+00:00", + ) + + +# --- Challenge derivation ----------------------------------------------------- + +def test_challenge_deterministic_versioned_and_32_bytes(): + a = compute_challenge(_request()) + b = compute_challenge(_request()) + assert a == b and len(a) == 32 + payload = json.loads(_request().canonical_bytes()) + assert payload["schema"] == REQUEST_SCHEMA + assert payload["rp_id"] == RP_ID + assert payload["user_verification_required"] is True + + +def test_challenge_field_order_independence(): + forward = AuthorizationRequest(**dict(BASE_REQUEST_KWARGS)) + reversed_kwargs = dict(reversed(list(BASE_REQUEST_KWARGS.items()))) + backward = AuthorizationRequest(**reversed_kwargs) + assert compute_challenge(forward) == compute_challenge(backward) + + +@pytest.mark.parametrize( + "override", + [ + {"artifact_byte_digest": "sha256:" + "ff" * 32}, + {"plan_body_digest": "sha256:" + "aa" * 32}, + {"decision_id": "gd-other"}, + {"nonce": "00000000-0000-0000-0000-000000000000"}, + {"issued_at": "2026-08-05T14:59:00+00:00"}, + {"expires_at": "2026-08-05T15:11:00+00:00"}, + {"approver_identity_id": "human-mallory"}, + {"rp_id": "evil.example"}, + {"user_verification_required": False}, + {"scope_digest": "sha256:" + "11" * 32}, + {"risk_class": "low"}, + {"policy_version": "cr-yk-001-v2"}, + ], +) +def test_challenge_changes_when_bound_field_changes(override): + assert compute_challenge(_request(**override)) != compute_challenge(_request()) + + +def test_unknown_request_schema_rejected(): + with pytest.raises(AuthzError): + _request(schema="triagecore.authz.request.v1") + + +def test_normalization_makes_equivalent_requests_identical(): + baseline = compute_challenge(_request()) + normalized = _request( + risk_class=" HIGH ", + policy_version="CR-YK-001-V1", + approver_identity_id="HUMAN-Corey", + rp_id="TRIAGECORE.LOCAL", + issued_at="2026-08-05T15:00:00.000000+00:00", + ) + assert compute_challenge(normalized) == baseline + assert normalized.risk_class == "high" + assert normalized.approver_identity_id == "human-corey" + assert normalized.rp_id == RP_ID + + +@pytest.mark.parametrize( + "override", + [ + {"risk_class": "critical"}, + {"artifact_byte_digest": "sha256:nothex"}, + {"plan_body_digest": "md5:" + "ab" * 32}, + {"scope_digest": "sha256:short"}, + {"task_id": "has whitespace"}, + {"task_id": " "}, + {"decision_id": ""}, + {"nonce": "not-a-uuid"}, + {"issued_at": "yesterday"}, + {"approver_identity_id": "***"}, + ], +) +def test_non_canonical_or_invalid_inputs_rejected(override): + with pytest.raises(AuthzError): + _request(**override) + + +# --- Sidecar artifact --------------------------------------------------------- + +def test_receipt_artifact_roundtrip_and_tamper_detection(tmp_path): + receipt = _receipt(_request()) + path = write_receipt_artifact(receipt, ledger_dir=str(tmp_path)) + loaded = read_receipt_artifact(path, expected_digest=receipt.receipt_digest()) + assert loaded.receipt_digest() == receipt.receipt_digest() + + raw = open(path, "r", encoding="utf-8").read() + with open(path, "w", encoding="utf-8") as handle: + handle.write(raw.replace("human-corey", "human-mallory")) + with pytest.raises(AuthzArtifactTamperError): + read_receipt_artifact(path, expected_digest=receipt.receipt_digest()) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX permission bits") +def test_receipt_artifact_permissions_owner_only(tmp_path): + path = write_receipt_artifact(_receipt(_request()), ledger_dir=str(tmp_path)) + mode = stat.S_IMODE(os.stat(path).st_mode) + assert mode & 0o077 == 0 + + +# --- Structural verification -------------------------------------------------- + +def test_structural_passes_for_honest_receipt(): + result = verify_receipt_structure(_receipt(_request()), now=NOW) + assert result.passed, result.failure_reason + + +@pytest.mark.parametrize( + "receipt_kwargs,expected", + [ + ({"client_data_bytes": b"\xff\xfenot json"}, STRUCTURAL_FAIL_MALFORMED), + ({"cd_type": "webauthn.create"}, STRUCTURAL_FAIL_TYPE), + ({"challenge_b64": b64url_encode(b"\x00" * 32)}, STRUCTURAL_FAIL_CHALLENGE), + ({"origin": "https://evil.example"}, STRUCTURAL_FAIL_ORIGIN), + ], +) +def test_structural_rejections(receipt_kwargs, expected): + result = verify_receipt_structure(_receipt(_request(), **receipt_kwargs), now=NOW) + assert not result.passed and result.failure_reason == expected + + +def test_structural_rejects_missing_client_data(): + receipt = _receipt(_request()) + del receipt.assertion_response["response"]["clientDataJSON"] + result = verify_receipt_structure(receipt, now=NOW) + assert not result.passed and result.failure_reason == STRUCTURAL_FAIL_MALFORMED + + +def test_structural_rejects_expired_request(): + result = verify_receipt_structure(_receipt(_request()), now=NOW + timedelta(hours=1)) + assert not result.passed and result.failure_reason == STRUCTURAL_FAIL_EXPIRED + + +def test_structural_rejects_tampered_request_binding(): + honest = _receipt(_request()) + swapped = HumanAuthorizationReceipt( + request=_request(artifact_byte_digest="sha256:" + "99" * 32), + credential_id=honest.credential_id, + assertion_response=honest.assertion_response, + user_verified=honest.user_verified, + recorded_at=honest.recorded_at, + ) + result = verify_receipt_structure(swapped, now=NOW) + assert not result.passed and result.failure_reason == STRUCTURAL_FAIL_CHALLENGE + + +# --- Ledger evidence ---------------------------------------------------------- + +def test_ledger_payload_privacy_safe_and_metadata_only(tmp_path): + receipt = _receipt(_request()) + payload = build_receipt_ledger_payload(receipt, "receipts/receipt_x.json") + assert_persistent_privacy_safe(payload) + + ledger = TaskLedger(ledger_dir=str(tmp_path)) + path = record_receipt(ledger, receipt, ledger_dir=str(tmp_path)) + events = ledger.get_events(receipt.request.task_id) + recorded = next(e for e in events if e["event_type"] == EVENT_RECEIPT_RECORDED) + serialized = json.dumps(recorded["payload"]) + assert "clientDataJSON" not in serialized + assert "signature" not in serialized + assert recorded["payload"]["receipt_artifact_path"] == path + assert recorded["payload"]["uv_required"] is True + assert recorded["payload"]["uv_verified"] is True + + +# --- Capability lane ---------------------------------------------------------- + +def test_capability_single_use_sequentially(tmp_path): + ledger = TaskLedger(ledger_dir=str(tmp_path)) + receipt = _receipt(_request()) + task_id = receipt.request.task_id + digest = receipt.request.artifact_byte_digest + + capability_id = issue_capability(ledger, receipt, now=NOW) + first = consume_capability(ledger, task_id, capability_id, digest, now=NOW) + assert first.allowed and first.reason_code == REASON_OK + + second = consume_capability(ledger, task_id, capability_id, digest, now=NOW) + assert not second.allowed and second.reason_code == REASON_ALREADY_CONSUMED + + events = ledger.get_events(task_id) + assert sum(e["event_type"] == EVENT_CAPABILITY_CONSUMED for e in events) == 1 + assert sum(e["event_type"] == EVENT_CAPABILITY_DENIED for e in events) == 1 + + +def test_capability_denies_expired_missing_and_mismatched(tmp_path): + ledger = TaskLedger(ledger_dir=str(tmp_path)) + receipt = _receipt(_request()) + task_id = receipt.request.task_id + digest = receipt.request.artifact_byte_digest + + missing = consume_capability(ledger, task_id, "not-a-real-id", digest, now=NOW) + assert not missing.allowed and missing.reason_code == REASON_NOT_FOUND + + expiring = issue_capability(ledger, receipt, ttl_seconds=60, now=NOW) + expired = consume_capability( + ledger, task_id, expiring, digest, now=NOW + timedelta(hours=2) + ) + assert not expired.allowed and expired.reason_code == REASON_EXPIRED + + fresh = issue_capability(ledger, receipt, now=NOW) + mismatched = consume_capability( + ledger, task_id, fresh, "sha256:" + "00" * 32, now=NOW + ) + assert not mismatched.allowed and mismatched.reason_code == REASON_DIGEST_MISMATCH + + +@pytest.mark.xfail( + strict=True, + reason=( + "Known limitation (CR-YK-001): read-then-append consumption is not " + "atomic; interleaved consumers can each observe 'unconsumed'. Atomic " + "claiming is required before CR-DD-012B execution integration." + ), +) +def test_concurrent_consume_race_documented(tmp_path, monkeypatch): + """Deterministically interleave two consumers; single-use SHOULD hold.""" + ledger = TaskLedger(ledger_dir=str(tmp_path)) + receipt = _receipt(_request()) + task_id = receipt.request.task_id + digest = receipt.request.artifact_byte_digest + capability_id = issue_capability(ledger, receipt, now=NOW) + + # Freeze both consumers' reads at the pre-consumption snapshot, modeling + # two processes that each read the ledger before either appends. + snapshot = ledger.get_events(task_id) + monkeypatch.setattr(ledger, "get_events", lambda _tid: list(snapshot)) + + first = consume_capability(ledger, task_id, capability_id, digest, now=NOW) + second = consume_capability(ledger, task_id, capability_id, digest, now=NOW) + successes = sum(1 for d in (first, second) if d.allowed) + assert successes <= 1 # violated today: both consumers are allowed + + +# --- Credential store --------------------------------------------------------- + +def _credential(**overrides) -> EnrolledCredential: + base = dict( + human_id="human-corey", + label="primary yubikey", + credential_id=b64url_encode(b"cred-1"), + public_key_cose=b64url_encode(b"cose-public-bytes"), + enrolled_at="2026-08-01T00:00:00+00:00", + ) + base.update(overrides) + return EnrolledCredential(**base) + + +def test_credential_store_roundtrip_duplicate_and_revocation(tmp_path): + store = CredentialStore(ledger_dir=str(tmp_path)) + assert store.load() == [] + credential = _credential() + store.add(credential) + assert store.find_by_credential_id(credential.credential_id).label == "primary yubikey" + with pytest.raises(AuthzError): + store.add(credential) + + store.revoke(credential.credential_id, now=NOW) + revoked = store.find_by_credential_id(credential.credential_id) + assert revoked.revoked and revoked.revoked_at.startswith("2026-08-05") + with pytest.raises(AuthzError): + store.revoke("missing-id") + + +def test_credential_store_normalizes_identity_at_every_boundary(tmp_path): + store = CredentialStore(ledger_dir=str(tmp_path)) + store.add(_credential(human_id=" Operator-A ")) + saved = store.load()[0] + assert saved.human_id == "operator-a" # serialization + assert store.find_by_human_id(" OPERATOR-A ") == [saved] # lookup + store.revoke(saved.credential_id, now=NOW) # revocation + assert store.load()[0].revoked + + +def test_identity_variants_fold_to_single_governed_identity(tmp_path): + store = CredentialStore(ledger_dir=str(tmp_path)) + store.add(_credential(human_id="Operator-A", credential_id=b64url_encode(b"key-1"))) + store.add(_credential(human_id="operator-a", credential_id=b64url_encode(b"key-2"))) + assert {c.human_id for c in store.load()} == {"operator-a"} # one identity + assert len(store.find_by_human_id("Operator-A")) == 2 # two backup keys + + +def test_invalid_identity_rejected_at_enrollment(): + with pytest.raises(AuthzError): + _credential(human_id="***") + with pytest.raises(AuthzError): + _credential(human_id=" ") + + +# --- Adapter availability (dependency / platform) ----------------------------- + +def test_enrollment_requires_cross_platform_authenticator(monkeypatch): + pytest.importorskip("fido2") + from fido2.webauthn import ( + AuthenticatorAttachment, + PublicKeyCredentialCreationOptions, + ) + + captured = {} + credential_data = SimpleNamespace( + credential_id=b"security-key-credential", + public_key={1: 2}, + aaguid=bytes(16), + ) + registration = SimpleNamespace( + response=SimpleNamespace( + attestation_object=SimpleNamespace( + auth_data=SimpleNamespace(credential_data=credential_data) + ) + ) + ) + + class FakeClient: + def make_credential(self, options): + captured["options"] = options + return registration + + monkeypatch.setattr(fido2_adapter, "_make_client", lambda: FakeClient()) + + fido2_adapter.enroll_credential( + human_id="human-corey", + label="Corey primary YubiKey", + ) + + options = captured["options"] + assert isinstance(options, PublicKeyCredentialCreationOptions) + assert ( + options.authenticator_selection.authenticator_attachment + == AuthenticatorAttachment.CROSS_PLATFORM + ) + + +@pytest.mark.parametrize( + "required,expected_name", + [(False, "DISCOURAGED"), (True, "REQUIRED")], +) +def test_assertion_options_apply_requested_uv_policy( + monkeypatch, required, expected_name +): + pytest.importorskip("fido2") + from fido2.webauthn import UserVerificationRequirement + + captured = {} + + class FakeAssertionResponse(dict): + raw_id = b"cred-1" + response = SimpleNamespace( + authenticator_data=SimpleNamespace( + is_user_verified=lambda: required + ) + ) + + class FakeClient: + def get_assertion(self, options): + captured["options"] = options + return SimpleNamespace( + get_response=lambda _index: FakeAssertionResponse() + ) + + monkeypatch.setattr(fido2_adapter, "_make_client", lambda: FakeClient()) + + fido2_adapter.get_assertion_receipt( + _request(user_verification_required=required), + _credential(), + ) + + assert captured["options"].user_verification == getattr( + UserVerificationRequirement, expected_name + ) + + +def test_missing_dependency_yields_controlled_unavailability(monkeypatch): + def _raise(): + raise AuthzHardwareUnavailable("python-fido2 >= 2.0 is required") + + monkeypatch.setattr(fido2_adapter, "_import_fido2", _raise) + support = fido2_adapter.ceremony_support() + assert support.available is False and support.mechanism == "none" + assert "python-fido2" in support.detail + + store_dir_free_store = CredentialStore(ledger_dir="unused") + receipt = _receipt(_request()) + # Structural and store checks run dependency-free; the crypto step must + # surface the controlled error, not an ImportError. + monkeypatch.setattr( + store_dir_free_store, "find_by_credential_id", lambda _cid: _credential( + credential_id=receipt.credential_id + ) + ) + with pytest.raises(AuthzHardwareUnavailable): + fido2_adapter.verify_receipt(receipt, store_dir_free_store, now=NOW) + + +def test_unsupported_platform_reports_unavailable_without_crash(monkeypatch): + pytest.importorskip("fido2") + monkeypatch.setattr(sys, "platform", "linux") + monkeypatch.setattr(fido2_adapter, "_list_ctap_devices", lambda: []) + support = fido2_adapter.ceremony_support() + assert support.available is False + assert support.mechanism == "none" + assert "no FIDO2 authenticator" in support.detail + + +# --- Full verification with real python-fido2 structures ---------------------- + +def _synthetic(request, *, up=True, uv=True, rp_for_hash=None, origin=ORIGIN, + tamper_auth=False, tamper_sig=False, wrong_key=False, + human_id="human-corey"): + """Build a receipt + enrolled credential from a software ES256 key using + real python-fido2 WebAuthn data structures. Proves offline verification + logic only — not hardware or Windows integration.""" + pytest.importorskip("fido2") + import struct + from cryptography.hazmat.primitives import hashes + from cryptography.hazmat.primitives.asymmetric import ec + from fido2 import cbor as fido2_cbor + from fido2.cose import ES256 + from fido2.webauthn import ( + AuthenticationResponse, + AuthenticatorAssertionResponse, + AuthenticatorData, + CollectedClientData, + ) + import hashlib as _hashlib + + private_key = ec.generate_private_key(ec.SECP256R1()) + cose_public = ES256.from_cryptography_key(private_key.public_key()) + credential_id = b"synthetic-credential-0001" + + enrolled = EnrolledCredential( + human_id=human_id, + label="synthetic software credential", + credential_id=b64url_encode(credential_id), + public_key_cose=b64url_encode(fido2_cbor.encode(cose_public)), + ) + + client_data = CollectedClientData.create( + type="webauthn.get", challenge=compute_challenge(request), origin=origin + ) + rp_hash = _hashlib.sha256((rp_for_hash or request.rp_id).encode()).digest() + flags = (0x01 if up else 0) | (0x04 if uv else 0) + auth_data = AuthenticatorData(rp_hash + bytes([flags]) + struct.pack(">I", 0)) + + signing_key = ec.generate_private_key(ec.SECP256R1()) if wrong_key else private_key + signature = signing_key.sign( + bytes(auth_data) + client_data.hash, ec.ECDSA(hashes.SHA256()) + ) + if tamper_sig: + signature = signature[:-1] + bytes([signature[-1] ^ 0x01]) + + wire = dict( + AuthenticationResponse( + raw_id=credential_id, + response=AuthenticatorAssertionResponse( + client_data=client_data, + authenticator_data=auth_data, + signature=signature, + ), + ) + ) + if tamper_auth: + tampered = bytearray(bytes(auth_data)) + tampered[0] ^= 0x01 + wire["response"]["authenticatorData"] = b64url_encode(bytes(tampered)) + + receipt = HumanAuthorizationReceipt( + request=request, + credential_id=b64url_encode(credential_id), + assertion_response=json.loads(json.dumps(wire)), + user_verified=uv, + recorded_at="2026-08-05T15:01:00+00:00", + ) + return receipt, enrolled + + +def _store_with(tmp_path, enrolled) -> CredentialStore: + store = CredentialStore(ledger_dir=str(tmp_path)) + store.add(enrolled) + return store + + +@pytest.mark.parametrize( + "required,expected_name", + [(False, "DISCOURAGED"), (True, "REQUIRED")], +) +def test_offline_verification_reconstructs_requested_uv_policy( + tmp_path, monkeypatch, required, expected_name +): + pytest.importorskip("fido2") + import fido2.server as fido2_server + from fido2.webauthn import UserVerificationRequirement + + captured = {} + + class FakeServer: + def __init__(self, *_args, **_kwargs): + pass + + def authenticate_complete(self, state, _credentials, _response): + captured["state"] = state + + monkeypatch.setattr(fido2_server, "Fido2Server", FakeServer) + receipt, enrolled = _synthetic( + _request(user_verification_required=required), + uv=required, + ) + store = _store_with(tmp_path, enrolled) + + fido2_adapter.verify_receipt(receipt, store, now=NOW) + + assert captured["state"]["user_verification"] == getattr( + UserVerificationRequirement, expected_name + ) + + +def test_full_verification_passes_with_zero_counter(tmp_path): + receipt, enrolled = _synthetic(_request()) + store = _store_with(tmp_path, enrolled) + result = fido2_adapter.verify_receipt(receipt, store, now=NOW) + assert result.credential_id == enrolled.credential_id + + +def test_full_verification_uv_not_required_up_only_passes(tmp_path): + request = _request(user_verification_required=False) + receipt, enrolled = _synthetic(request, uv=False) + store = _store_with(tmp_path, enrolled) + assert fido2_adapter.verify_receipt(receipt, store, now=NOW) + + +@pytest.mark.parametrize( + "synthetic_kwargs,message_fragment", + [ + ({"up": False}, "User Present"), + ({"uv": False}, "User verification required"), + ({"rp_for_hash": "evil.example"}, "RP ID hash"), + ({"tamper_auth": True}, ""), # signature no longer covers auth data + ({"tamper_sig": True}, ""), + ({"wrong_key": False, "tamper_sig": True}, ""), + ({"wrong_key": True}, ""), + ], +) +def test_full_verification_rejections(tmp_path, synthetic_kwargs, message_fragment): + receipt, enrolled = _synthetic(_request(), **synthetic_kwargs) + store = _store_with(tmp_path, enrolled) + with pytest.raises(AuthzVerificationError) as excinfo: + fido2_adapter.verify_receipt(receipt, store, now=NOW) + if message_fragment: + assert message_fragment in str(excinfo.value) + + +def test_full_verification_rejects_unknown_credential(tmp_path): + receipt, _enrolled = _synthetic(_request()) + empty_store = CredentialStore(ledger_dir=str(tmp_path)) + with pytest.raises(AuthzVerificationError) as excinfo: + fido2_adapter.verify_receipt(receipt, empty_store, now=NOW) + assert str(excinfo.value) == fido2_adapter.VERIFY_FAIL_UNKNOWN_CREDENTIAL + + +def test_enrolled_identity_and_request_approver_resolve_identically(tmp_path): + """enroll('OPERATOR-A') + verify(approver=' Operator-A ') → same identity.""" + request = _request(approver_identity_id=" Operator-A ") + receipt, enrolled = _synthetic(request, human_id="OPERATOR-A") + store = _store_with(tmp_path, enrolled) + verified = fido2_adapter.verify_receipt(receipt, store, now=NOW) + assert verified.human_id == "operator-a" + assert request.approver_identity_id == "operator-a" + + +def test_full_verification_rejects_credential_of_different_identity(tmp_path): + receipt, enrolled = _synthetic(_request()) + enrolled.human_id = "human-someone-else" + store = _store_with(tmp_path, enrolled) + with pytest.raises(AuthzVerificationError) as excinfo: + fido2_adapter.verify_receipt(receipt, store, now=NOW) + assert str(excinfo.value) == fido2_adapter.VERIFY_FAIL_APPROVER_MISMATCH + + +def test_full_verification_rejects_revoked_credential(tmp_path): + receipt, enrolled = _synthetic(_request()) + store = _store_with(tmp_path, enrolled) + store.revoke(enrolled.credential_id, now=NOW) + with pytest.raises(AuthzVerificationError) as excinfo: + fido2_adapter.verify_receipt(receipt, store, now=NOW) + assert str(excinfo.value) == fido2_adapter.VERIFY_FAIL_REVOKED_CREDENTIAL + + +def test_full_verification_rejects_wrong_origin_structurally(tmp_path): + receipt, enrolled = _synthetic(_request(), origin="https://evil.example") + store = _store_with(tmp_path, enrolled) + with pytest.raises(AuthzVerificationError) as excinfo: + fido2_adapter.verify_receipt(receipt, store, now=NOW) + assert str(excinfo.value) == STRUCTURAL_FAIL_ORIGIN + + +def test_full_verification_rejects_challenge_mismatch_structurally(tmp_path): + receipt, enrolled = _synthetic(_request()) + tampered = HumanAuthorizationReceipt( + request=_request(scope_digest="sha256:" + "77" * 32), + credential_id=receipt.credential_id, + assertion_response=receipt.assertion_response, + user_verified=receipt.user_verified, + recorded_at=receipt.recorded_at, + ) + store = _store_with(tmp_path, enrolled) + with pytest.raises(AuthzVerificationError) as excinfo: + fido2_adapter.verify_receipt(tampered, store, now=NOW) + assert str(excinfo.value) == STRUCTURAL_FAIL_CHALLENGE diff --git a/triage_core/authz.py b/triage_core/authz.py new file mode 100644 index 0000000..c87ffa7 --- /dev/null +++ b/triage_core/authz.py @@ -0,0 +1,660 @@ +"""FIDO2 security-key-backed human authorization receipts (CR-YK-001). + +Deterministic core only: canonical authorization requests, challenge +derivation, receipt sidecar artifacts with tamper detection, structural +(pre-cryptographic) receipt checks, a one-use execution-capability lane, and +a public-credential store with revocation. + +Everything that touches python-fido2 lives in ``triage_core.fido2_adapter``. +This module imports no optional dependency and must keep working on every +platform with no authenticator present. + +Evidence-honesty vocabulary used throughout: + +- A verified receipt is *hardware-backed evidence of possession and user + interaction* with an enrolled FIDO2 security key. +- "User verification confirmed" is claimed only when the request required UV + and the authenticator's UV flag was set and checked. +- Nothing here claims non-repudiation, proof of a specific authenticator + model (attestation is not validated), or protection against an attacker + who already controls this host or its clock. See the threat-model section + of CR-YK-001. + +Known limitation (documented, not yet solved): capability consumption reads +then appends over a shared JSONL ledger. Two concurrent consumers can both +observe "unconsumed" before either records consumption, so single-use is +enforced only against sequential use. Atomic claiming (e.g. SQLite +``BEGIN IMMEDIATE``) is a CR-DD-012B integration requirement; see +``consume_capability`` and the expected-failure concurrency test. +""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import os +import re +import stat +import uuid +from dataclasses import dataclass, field, asdict +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from triage_core.privacy_invariants import assert_persistent_privacy_safe +from triage_core.task_ledger import TaskLedger + +# --- Constants --------------------------------------------------------------- + +REQUEST_SCHEMA = "triagecore.authz.request.v2" +RECEIPT_SCHEMA = "triagecore.human_authorization_receipt.v2" +CREDENTIALS_SCHEMA = "triagecore.authz_credentials.v2" + +# Local relying-party identity for a CLI-first flow. The origin is never +# served; it exists so WebAuthn client data has a stable, pinned value. +RP_ID = "triagecore.local" +RP_NAME = "TriageCore Governed Authorization" +ORIGIN = "https://triagecore.local" + +DEFAULT_REQUEST_TTL_SECONDS = 600 +DEFAULT_CAPABILITY_TTL_SECONDS = 600 + +EVENT_RECEIPT_RECORDED = "human_authorization_receipt" +EVENT_CAPABILITY_ISSUED = "execution_capability_issued" +EVENT_CAPABILITY_CONSUMED = "execution_capability_consumed" +EVENT_CAPABILITY_DENIED = "execution_capability_denied" + +# Closed reason vocabularies (house style: no free-text reasons in evidence). +REASON_OK = "ok" +REASON_NOT_FOUND = "capability_not_found" +REASON_ALREADY_CONSUMED = "capability_already_consumed" +REASON_EXPIRED = "capability_expired" +REASON_DIGEST_MISMATCH = "artifact_digest_mismatch" + +STRUCTURAL_FAIL_EXPIRED = "request_expired" +STRUCTURAL_FAIL_MALFORMED = "client_data_malformed" +STRUCTURAL_FAIL_TYPE = "client_data_type_mismatch" +STRUCTURAL_FAIL_CHALLENGE = "client_data_challenge_mismatch" +STRUCTURAL_FAIL_ORIGIN = "client_data_origin_mismatch" + +AUTHZ_DIR_NAME = "authz" +RECEIPTS_DIR_NAME = "receipts" +CREDENTIALS_FILE_NAME = "credentials.json" + + +class AuthzError(Exception): + """Base error for the authorization-receipt lane.""" + + +class AuthzHardwareUnavailable(AuthzError): + """Optional dependency or platform ceremony support is unavailable.""" + + +class AuthzVerificationError(AuthzError): + """A receipt failed structural or cryptographic verification.""" + + +class AuthzArtifactTamperError(AuthzError): + """A sidecar artifact does not match its recorded digest.""" + + +# --- Small helpers ----------------------------------------------------------- + +def b64url_encode(raw: bytes) -> str: + return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") + + +def b64url_decode(text: str) -> bytes: + padding = "=" * (-len(text) % 4) + return base64.urlsafe_b64decode(text + padding) + + +def _canonical_json_bytes(value: Any) -> bytes: + """Canonical serialization: sorted keys, minimal separators, UTF-8. + + Key sorting makes the digest independent of construction order. + """ + return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8") + + +def _sha256_hex(raw: bytes) -> str: + return "sha256:" + hashlib.sha256(raw).hexdigest() + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _isoformat(moment: datetime) -> str: + return moment.astimezone(timezone.utc).isoformat() + + +def _restrict_file_permissions(path: str) -> None: + """Best-effort owner-only permissions; a no-op where unsupported.""" + try: + os.chmod(path, stat.S_IRUSR | stat.S_IWUSR) + except OSError: # pragma: no cover - platform dependent + pass + + +# --- Normalization and validation --------------------------------------------- + +_RISK_CLASSES = frozenset({"low", "medium", "high"}) +_IDENTIFIER_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]{1,63}$") +_DIGEST_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$") + + +def normalize_identity(value: str) -> str: + """Normalize a governed identifier: strip + lowercase. + + Security-bound identifiers must be stable: cosmetic variants ("Corey", + "corey ", "COREY") fold to one canonical form before any hashing or + comparison. Display labels are informational, live in the credential + store, and are never security-bound. + """ + return (value or "").strip().lower() + + +def _normalized_timestamp(value: str, field_name: str) -> str: + try: + return _isoformat(datetime.fromisoformat(value)) + except ValueError as exc: + raise AuthzError(f"invalid {field_name} timestamp: {value!r}") from exc + + +# --- Authorization request and challenge ------------------------------------- + +@dataclass(frozen=True) +class AuthorizationRequest: + """Canonical, versioned description of the one action being approved. + + Every field participates in the challenge digest, so changing the bound + plan, decision, scope, RP, UV policy, approver identity, nonce, or + validity window invalidates any assertion made over the previous value. + + Construction normalizes before anything is hashed — canonical JSON fixes + field order, not case or stray whitespace, so "HIGH", "high", and + "high " must not produce different challenges. Vocabulary and identity + fields fold to strip+lowercase and are validated against closed + grammars; ledger-linked identifiers (``task_id``, ``decision_id``) keep + their case because ledger lookups are case-exact, but reject embedded + whitespace. Digests must be ``sha256:<64 lowercase hex>``; the nonce + must be a UUID (stored in canonical form); timestamps round-trip + through ISO-8601 parsing. + + ``approver_identity_id`` is the *stable registered identifier* of the + approving human and is security-bound. Human-facing display labels are + informational only, live in the credential store, and never enter the + challenge. + + All fields are privacy-safe metadata (digests, identifiers, vocabulary + values). Never place prompt or file content here. + """ + + decision_id: str + artifact_byte_digest: str + plan_body_digest: str + task_id: str + risk_class: str # "low" | "medium" | "high" + policy_version: str # e.g. "cr-yk-001-v1" + approver_identity_id: str # stable governed identifier, e.g. "human-corey" + user_verification_required: bool = True + scope_digest: str = "" # execution-scope / policy-decision digest, when available + rp_id: str = RP_ID + schema: str = REQUEST_SCHEMA + nonce: str = field(default_factory=lambda: str(uuid.uuid4())) + issued_at: str = field(default_factory=lambda: _isoformat(_utc_now())) + expires_at: str = "" + + def __post_init__(self) -> None: + if self.schema != REQUEST_SCHEMA: + raise AuthzError(f"unsupported request schema: {self.schema!r}") + + for name in ("risk_class", "policy_version", "approver_identity_id", "rp_id"): + object.__setattr__(self, name, normalize_identity(getattr(self, name))) + for name in ("task_id", "decision_id"): + stripped = (getattr(self, name) or "").strip() + object.__setattr__(self, name, stripped) + if not stripped or any(ch.isspace() for ch in stripped): + raise AuthzError(f"{name} must be non-empty with no whitespace") + + if self.risk_class not in _RISK_CLASSES: + raise AuthzError(f"unknown risk_class: {self.risk_class!r}") + for name in ("policy_version", "approver_identity_id", "rp_id"): + if not _IDENTIFIER_PATTERN.fullmatch(getattr(self, name)): + raise AuthzError(f"{name} is not a stable governed identifier") + for name in ("artifact_byte_digest", "plan_body_digest"): + if not _DIGEST_PATTERN.fullmatch(getattr(self, name)): + raise AuthzError(f"{name} must match sha256:<64 lowercase hex>") + if self.scope_digest and not _DIGEST_PATTERN.fullmatch(self.scope_digest): + raise AuthzError("scope_digest must be empty or sha256:<64 lowercase hex>") + + try: + object.__setattr__(self, "nonce", str(uuid.UUID(self.nonce.strip()))) + except (AttributeError, ValueError) as exc: + raise AuthzError(f"nonce must be a UUID: {self.nonce!r}") from exc + + object.__setattr__( + self, "issued_at", _normalized_timestamp(self.issued_at, "issued_at") + ) + if self.expires_at: + object.__setattr__( + self, + "expires_at", + _normalized_timestamp(self.expires_at, "expires_at"), + ) + else: + issued = datetime.fromisoformat(self.issued_at) + expires = issued.timestamp() + DEFAULT_REQUEST_TTL_SECONDS + object.__setattr__( + self, + "expires_at", + _isoformat(datetime.fromtimestamp(expires, tz=timezone.utc)), + ) + + def canonical_bytes(self) -> bytes: + return _canonical_json_bytes(asdict(self)) + + def request_digest(self) -> str: + return _sha256_hex(self.canonical_bytes()) + + def is_expired(self, now: Optional[datetime] = None) -> bool: + moment = now or _utc_now() + return moment > datetime.fromisoformat(self.expires_at) + + +def compute_challenge(request: AuthorizationRequest) -> bytes: + """32-byte WebAuthn challenge. + + challenge = SHA-256(canonical_json(request)) + + Versioning and domain separation come from the ``schema`` field inside + the canonical payload rather than raw byte concatenation, so the bound + field set is explicit and auditable in the serialized form. + """ + return hashlib.sha256(request.canonical_bytes()).digest() + + +# --- Receipt ------------------------------------------------------------------ + +@dataclass +class HumanAuthorizationReceipt: + """Request plus the WebAuthn assertion in standard JSON wire format. + + ``assertion_response`` is the W3C AuthenticationResponse JSON mapping + (``id``/``rawId``/``response.clientDataJSON``/``response.authenticatorData`` + /``response.signature``, websafe-base64 values) exactly as produced by + python-fido2's serialization, so offline verification can replay it + through library primitives without bespoke parsing. + + Receipts are sidecar artifacts, never inline ledger content: the ledger + privacy invariant forbids credential-adjacent keys and value-scans + strings, and the ledger event stores digests plus privacy-safe metadata + only. This is a boundary, not a renaming trick. + """ + + request: AuthorizationRequest + credential_id: str # b64url, matches assertion_response["rawId"] + assertion_response: Dict[str, Any] + user_verified: bool # authenticator UV flag as observed + recorded_at: str = field(default_factory=lambda: _isoformat(_utc_now())) + + def to_artifact_dict(self) -> Dict[str, Any]: + return { + "schema": RECEIPT_SCHEMA, + "request": asdict(self.request), + "credential_id": self.credential_id, + "assertion_response": self.assertion_response, + "user_verified": self.user_verified, + "recorded_at": self.recorded_at, + } + + def canonical_artifact_bytes(self) -> bytes: + return _canonical_json_bytes(self.to_artifact_dict()) + + def receipt_digest(self) -> str: + return _sha256_hex(self.canonical_artifact_bytes()) + + @classmethod + def from_artifact_dict(cls, raw: Dict[str, Any]) -> "HumanAuthorizationReceipt": + if raw.get("schema") != RECEIPT_SCHEMA: + raise AuthzVerificationError( + f"unknown receipt schema: {raw.get('schema')!r}" + ) + return cls( + request=AuthorizationRequest(**raw["request"]), + credential_id=raw["credential_id"], + assertion_response=dict(raw["assertion_response"]), + user_verified=bool(raw["user_verified"]), + recorded_at=raw["recorded_at"], + ) + + +def write_receipt_artifact( + receipt: HumanAuthorizationReceipt, + ledger_dir: str = ".triagecore", +) -> str: + """Write the receipt sidecar artifact with restrictive permissions. + + Retention: receipts persist until the operator prunes them; they are + inputs to offline re-verification, so removal degrades auditability of + the runs they authorized (documented in CR-YK-001). + """ + receipts_dir = os.path.join(ledger_dir, AUTHZ_DIR_NAME, RECEIPTS_DIR_NAME) + os.makedirs(receipts_dir, exist_ok=True) + digest_fragment = receipt.request.request_digest().split(":", 1)[1][:12] + path = os.path.join(receipts_dir, f"receipt_{digest_fragment}.json") + temp_path = path + ".tmp" + with open(temp_path, "w", encoding="utf-8", newline="\n") as handle: + json.dump(receipt.to_artifact_dict(), handle, sort_keys=True, indent=2) + handle.write("\n") + _restrict_file_permissions(temp_path) + os.replace(temp_path, path) + return path + + +def read_receipt_artifact( + path: str, + expected_digest: Optional[str] = None, +) -> HumanAuthorizationReceipt: + """Load a receipt; verify tamper-evidence when a digest is supplied. + + ``expected_digest`` should come from the ledger's + ``human_authorization_receipt`` event, closing the loop between the + append-only record and the sidecar bytes. + """ + with open(path, "r", encoding="utf-8") as handle: + receipt = HumanAuthorizationReceipt.from_artifact_dict(json.load(handle)) + if expected_digest is not None and receipt.receipt_digest() != expected_digest: + raise AuthzArtifactTamperError( + f"receipt artifact digest mismatch at {path}" + ) + return receipt + + +# --- Structural (pre-cryptographic) verification ------------------------------ + +@dataclass(frozen=True) +class StructuralVerification: + passed: bool + failure_reason: str = "" + + +def _client_data_from_wire(assertion_response: Dict[str, Any]) -> Dict[str, Any]: + inner = assertion_response.get("response") or {} + raw = inner.get("clientDataJSON") + if not isinstance(raw, str): + raise ValueError("missing clientDataJSON") + return json.loads(b64url_decode(raw)) + + +def verify_receipt_structure( + receipt: HumanAuthorizationReceipt, + now: Optional[datetime] = None, +) -> StructuralVerification: + """Deterministic checks needing no cryptography and no python-fido2. + + Full verification (RP-ID hash, UP/UV flags, enrolled and non-revoked + credential, signature) is ``triage_core.fido2_adapter.verify_receipt``, + which layers python-fido2's server-side checks on top of this. + """ + if receipt.request.is_expired(now): + return StructuralVerification(False, STRUCTURAL_FAIL_EXPIRED) + + try: + client_data = _client_data_from_wire(receipt.assertion_response) + except (ValueError, json.JSONDecodeError): + return StructuralVerification(False, STRUCTURAL_FAIL_MALFORMED) + + if client_data.get("type") != "webauthn.get": + return StructuralVerification(False, STRUCTURAL_FAIL_TYPE) + + expected_challenge = b64url_encode(compute_challenge(receipt.request)) + if client_data.get("challenge") != expected_challenge: + return StructuralVerification(False, STRUCTURAL_FAIL_CHALLENGE) + + if client_data.get("origin") != ORIGIN: + return StructuralVerification(False, STRUCTURAL_FAIL_ORIGIN) + + return StructuralVerification(True) + + +# --- Ledger events ------------------------------------------------------------ + +def build_receipt_ledger_payload( + receipt: HumanAuthorizationReceipt, + artifact_path: str, +) -> Dict[str, Any]: + """Digest-and-metadata-only payload; must pass the privacy invariant.""" + payload = { + "receipt_schema": RECEIPT_SCHEMA, + "receipt_digest": receipt.receipt_digest(), + "request_digest": receipt.request.request_digest(), + "decision_id": receipt.request.decision_id, + "artifact_byte_digest": receipt.request.artifact_byte_digest, + "plan_body_digest": receipt.request.plan_body_digest, + "scope_digest": receipt.request.scope_digest, + "approver_identity_id": receipt.request.approver_identity_id, + "risk_class": receipt.request.risk_class, + "policy_version": receipt.request.policy_version, + "rp_id": receipt.request.rp_id, + "nonce": receipt.request.nonce, + "uv_required": receipt.request.user_verification_required, + "uv_verified": receipt.user_verified, + "receipt_artifact_path": artifact_path, + "request_expires_at": receipt.request.expires_at, + } + assert_persistent_privacy_safe( + payload, artifact_name="human_authorization_receipt payload" + ) + return payload + + +def record_receipt( + ledger: TaskLedger, + receipt: HumanAuthorizationReceipt, + ledger_dir: str = ".triagecore", +) -> str: + """Write the sidecar artifact and the metadata-only ledger event.""" + artifact_path = write_receipt_artifact(receipt, ledger_dir=ledger_dir) + payload = build_receipt_ledger_payload(receipt, artifact_path) + ledger.append_event(receipt.request.task_id, EVENT_RECEIPT_RECORDED, payload) + return artifact_path + + +# --- One-use execution capability --------------------------------------------- + +@dataclass(frozen=True) +class CapabilityDecision: + allowed: bool + reason_code: str + capability_id: str = "" + + +def issue_capability( + ledger: TaskLedger, + receipt: HumanAuthorizationReceipt, + ttl_seconds: int = DEFAULT_CAPABILITY_TTL_SECONDS, + now: Optional[datetime] = None, +) -> str: + """Issue a one-use capability bound to the receipt's artifact digest. + + Callers must fully verify the receipt (structural + fido2_adapter) before + issuing. ``now`` is injectable for deterministic tests. + """ + capability_id = str(uuid.uuid4()) + expires = (now or _utc_now()).timestamp() + ttl_seconds + payload = { + "capability_id": capability_id, + "receipt_digest": receipt.receipt_digest(), + "decision_id": receipt.request.decision_id, + "artifact_byte_digest": receipt.request.artifact_byte_digest, + "approver_identity_id": receipt.request.approver_identity_id, + "expires_at": _isoformat(datetime.fromtimestamp(expires, tz=timezone.utc)), + "single_use": True, + } + assert_persistent_privacy_safe( + payload, artifact_name="execution_capability_issued payload" + ) + ledger.append_event(receipt.request.task_id, EVENT_CAPABILITY_ISSUED, payload) + return capability_id + + +def consume_capability( + ledger: TaskLedger, + task_id: str, + capability_id: str, + artifact_byte_digest: str, + now: Optional[datetime] = None, +) -> CapabilityDecision: + """Consume the capability; every outcome is a ledger event. + + CONCURRENCY LIMITATION (documented, unresolved in this slice): this is a + read-then-append sequence over a shared JSONL file with no lock. Two + concurrent consumers can both observe the capability as unconsumed and + both record consumption, so single-use holds against sequential use + only. The sequential unit tests do not demonstrate race safety, and an + expected-failure test in tests/test_authz.py encodes the gap. Atomic + claiming (single-writer lock or SQLite ``BEGIN IMMEDIATE`` claim table) + is required before CR-DD-012B wires this into execution. + + Denials are evidence too: not_found, already_consumed, expired, and + digest_mismatch all append EVENT_CAPABILITY_DENIED with the reason code. + """ + moment = now or _utc_now() + issued: Optional[Dict[str, Any]] = None + for event in ledger.get_events(task_id): + etype = event.get("event_type") + payload = event.get("payload", {}) + if payload.get("capability_id") != capability_id: + continue + if etype == EVENT_CAPABILITY_ISSUED: + issued = payload + elif etype == EVENT_CAPABILITY_CONSUMED: + return _deny(ledger, task_id, capability_id, REASON_ALREADY_CONSUMED) + + if issued is None: + return _deny(ledger, task_id, capability_id, REASON_NOT_FOUND) + if moment > datetime.fromisoformat(issued["expires_at"]): + return _deny(ledger, task_id, capability_id, REASON_EXPIRED) + if issued["artifact_byte_digest"] != artifact_byte_digest: + return _deny(ledger, task_id, capability_id, REASON_DIGEST_MISMATCH) + + payload = { + "capability_id": capability_id, + "artifact_byte_digest": artifact_byte_digest, + "reason_code": REASON_OK, + } + ledger.append_event(task_id, EVENT_CAPABILITY_CONSUMED, payload) + return CapabilityDecision(True, REASON_OK, capability_id) + + +def _deny( + ledger: TaskLedger, + task_id: str, + capability_id: str, + reason_code: str, +) -> CapabilityDecision: + ledger.append_event( + task_id, + EVENT_CAPABILITY_DENIED, + {"capability_id": capability_id, "reason_code": reason_code}, + ) + return CapabilityDecision(False, reason_code, capability_id) + + +# --- Credential store --------------------------------------------------------- + +@dataclass +class EnrolledCredential: + """Public credential material only. Never PINs, private keys, or secrets. + + ``human_id`` is a stable governed identifier with the same normalization + contract as ``AuthorizationRequest.approver_identity_id``: construction + normalizes (strip + lowercase) and validates, so enrollment, lookup, + revocation, serialization, and verification all operate on one canonical + form. Cosmetic variants ("Operator-A", "operator-a") therefore resolve + to the same governed identity and cannot enroll as distinct ones. + ``label`` is informational display text and is never security-bound. + """ + + human_id: str + label: str + credential_id: str # b64url + public_key_cose: str # b64url of CBOR-encoded COSE public key + aaguid: str = "00" * 16 # hex; zeros when the authenticator omits it + rp_id: str = RP_ID + enrolled_at: str = field(default_factory=lambda: _isoformat(_utc_now())) + revoked: bool = False + revoked_at: str = "" + + def __post_init__(self) -> None: + self.human_id = normalize_identity(self.human_id) + if not _IDENTIFIER_PATTERN.fullmatch(self.human_id): + raise AuthzError("human_id is not a stable governed identifier") + self.rp_id = normalize_identity(self.rp_id) + + +class CredentialStore: + """Registry at .triagecore/authz/credentials.json (public material only). + + Enroll at least two credentials per human so a backup key is part of the + architecture. Revocation is a tombstone, not a deletion: revoked entries + stay listed so historical receipts remain attributable while new + verifications reject them. + """ + + def __init__(self, ledger_dir: str = ".triagecore"): + self.path = os.path.join(ledger_dir, AUTHZ_DIR_NAME, CREDENTIALS_FILE_NAME) + + def load(self) -> List[EnrolledCredential]: + if not os.path.exists(self.path): + return [] + with open(self.path, "r", encoding="utf-8") as handle: + raw = json.load(handle) + return [EnrolledCredential(**item) for item in raw.get("credentials", [])] + + def save(self, credentials: List[EnrolledCredential]) -> None: + os.makedirs(os.path.dirname(self.path), exist_ok=True) + temp_path = self.path + ".tmp" + body = { + "schema": CREDENTIALS_SCHEMA, + "credentials": [asdict(item) for item in credentials], + } + with open(temp_path, "w", encoding="utf-8", newline="\n") as handle: + json.dump(body, handle, sort_keys=True, indent=2) + handle.write("\n") + _restrict_file_permissions(temp_path) + os.replace(temp_path, self.path) + + def add(self, credential: EnrolledCredential) -> None: + credentials = self.load() + if any(c.credential_id == credential.credential_id for c in credentials): + raise AuthzError("credential already enrolled") + credentials.append(credential) + self.save(credentials) + + def revoke(self, credential_id: str, now: Optional[datetime] = None) -> None: + credentials = self.load() + for credential in credentials: + if credential.credential_id == credential_id: + credential.revoked = True + credential.revoked_at = _isoformat(now or _utc_now()) + self.save(credentials) + return + raise AuthzError("credential not found") + + def find_by_credential_id(self, credential_id: str) -> Optional[EnrolledCredential]: + for credential in self.load(): + if credential.credential_id == credential_id: + return credential + return None + + def find_by_human_id(self, human_id: str) -> List[EnrolledCredential]: + """All credentials (incl. revoked) for one governed identity. + + The query is normalized with the same contract as storage, so any + cosmetic variant of the identifier reaches the same entries. + """ + wanted = normalize_identity(human_id) + return [c for c in self.load() if c.human_id == wanted] diff --git a/triage_core/fido2_adapter.py b/triage_core/fido2_adapter.py new file mode 100644 index 0000000..70fb50f --- /dev/null +++ b/triage_core/fido2_adapter.py @@ -0,0 +1,326 @@ +"""python-fido2 (2.x) adapter for CR-YK-001 authorization receipts. + +All optional-dependency and platform-specific code for the authorization lane +lives here, behind narrow functions that callers and tests can replace: + +- ``ceremony_support()`` -> availability probe, never raises on import +- ``enroll_credential()`` -> make-credential ceremony (hardware) +- ``get_assertion_receipt()`` -> get-assertion ceremony (hardware) +- ``verify_receipt()`` -> offline verification via Fido2Server + +Import rules enforced here: + +- ``import triage_core`` must succeed without python-fido2 installed; this + module imports fido2 only inside functions via the ``_import_fido2`` seam. +- ``fido2.client.windows`` crashes at import time on non-Windows platforms + (its ctypes bindings require HRESULT), so it is imported only when + ``sys.platform == "win32"``. +- Non-Windows systems without a CTAP HID device get a controlled + ``CeremonySupport(available=False, ...)`` result, never an import crash. + +Written against python-fido2 2.2.x. The 1.x style ``WindowsClient(origin)`` +constructor no longer exists; 2.x clients take a ``ClientDataCollector``. + +Verification note: ``Fido2Server.authenticate_complete`` checks clientData +type, challenge equality (constant-time), pinned origin, RP-ID hash, the +user-presence flag, user-verification when required, credential membership, +and the signature over ``authenticatorData || SHA256(clientDataJSON)``. It +does not enforce signature counters; zero counters (common on current keys) +are therefore accepted, deliberately. + +Hardware status: primary YubiKey enrollment and assertion ceremonies were +physically verified on 2026-07-24 through Windows native WebAuthn. A bounded +secondary cross-device phone passkey path was also verified. Redundant +backup-YubiKey enrollment remains unverified. Execution integration remains +out of scope for CR-YK-001. +""" + +from __future__ import annotations + +import sys +from dataclasses import dataclass +from datetime import datetime +from typing import Optional + +from triage_core.authz import ( + ORIGIN, + RP_ID, + RP_NAME, + AuthorizationRequest, + AuthzError, + AuthzHardwareUnavailable, + AuthzVerificationError, + CredentialStore, + EnrolledCredential, + HumanAuthorizationReceipt, + b64url_decode, + b64url_encode, + compute_challenge, + normalize_identity, + verify_receipt_structure, +) + +VERIFY_FAIL_UNKNOWN_CREDENTIAL = "credential_not_enrolled" +VERIFY_FAIL_REVOKED_CREDENTIAL = "credential_revoked" +VERIFY_FAIL_RP_MISMATCH = "credential_rp_mismatch" +VERIFY_FAIL_APPROVER_MISMATCH = "credential_approver_mismatch" + + +def _import_fido2(): + """Import seam for the optional dependency; tests may monkeypatch this.""" + try: + import fido2.cbor + import fido2.cose + import fido2.server + import fido2.utils + import fido2.webauthn + except ImportError as exc: + raise AuthzHardwareUnavailable( + "python-fido2 >= 2.0 is required for this operation. " + 'Install the optional extra: pip install -e ".[authz]"' + ) from exc + import fido2 as _fido2 + + return _fido2 + + +def _list_ctap_devices(): + """Enumerate CTAP HID devices; seam for tests. Requires fido2 installed.""" + from fido2.hid import CtapHidDevice + + return list(CtapHidDevice.list_devices()) + + +# --- Availability ------------------------------------------------------------- + +@dataclass(frozen=True) +class CeremonySupport: + available: bool + mechanism: str # "windows_native" | "ctap_hid" | "none" + detail: str = "" + + +def ceremony_support() -> CeremonySupport: + """Report how a live ceremony could run on this host. Never raises.""" + try: + _import_fido2() + except AuthzHardwareUnavailable as exc: + return CeremonySupport(False, "none", str(exc)) + + if sys.platform == "win32": + try: + from fido2.client.windows import WindowsClient # guarded import + + if WindowsClient.is_available(): + return CeremonySupport(True, "windows_native") + return CeremonySupport( + False, "none", "Windows WebAuthn API not available on this build" + ) + except Exception as exc: # pragma: no cover - windows only + return CeremonySupport(False, "none", f"windows client error: {exc}") + + try: + devices = _list_ctap_devices() + except Exception as exc: + return CeremonySupport(False, "none", f"CTAP HID enumeration failed: {exc}") + if devices: + return CeremonySupport(True, "ctap_hid", f"{len(devices)} device(s)") + return CeremonySupport(False, "none", "no FIDO2 authenticator found") + + +def _make_client(): # pragma: no cover - requires hardware or Windows API + """Build a 2.x WebAuthn client for the current platform.""" + _import_fido2() + from fido2.client import DefaultClientDataCollector, Fido2Client + + collector = DefaultClientDataCollector(ORIGIN) + + if sys.platform == "win32": + from fido2.client.windows import WindowsClient # guarded import + + if WindowsClient.is_available(): + return WindowsClient(client_data_collector=collector) + + devices = _list_ctap_devices() + if not devices: + raise AuthzHardwareUnavailable("no FIDO2 authenticator found") + return Fido2Client(devices[0], client_data_collector=collector) + + +# --- Ceremonies (primary YubiKey + phone paths verified, backup unverified) --- + +def enroll_credential( + human_id: str, + label: str, + rp_id: Optional[str] = None, +) -> EnrolledCredential: # pragma: no cover - requires hardware + """Run a make-credential ceremony; return public material for the store. + + Stores credential ID, COSE public key, and AAGUID only — never PINs, + private keys, or secrets (those never leave the authenticator). + """ + import os as _os + + fido2 = _import_fido2() + from fido2.webauthn import ( + AuthenticatorAttachment, + AuthenticatorSelectionCriteria, + PublicKeyCredentialCreationOptions, + PublicKeyCredentialParameters, + PublicKeyCredentialRpEntity, + PublicKeyCredentialType, + PublicKeyCredentialUserEntity, + ) + + resolved_rp = rp_id or RP_ID + human_id = normalize_identity(human_id) + if not human_id: + raise AuthzError("human_id must be a non-empty stable identifier") + client = _make_client() + options = PublicKeyCredentialCreationOptions( + rp=PublicKeyCredentialRpEntity(id=resolved_rp, name=RP_NAME), + user=PublicKeyCredentialUserEntity( + id=human_id.encode("utf-8"), name=human_id, display_name=label + ), + challenge=_os.urandom(32), + pub_key_cred_params=[ + PublicKeyCredentialParameters( + type=PublicKeyCredentialType.PUBLIC_KEY, alg=-7 # ES256 + ) + ], + authenticator_selection=AuthenticatorSelectionCriteria( + authenticator_attachment=AuthenticatorAttachment.CROSS_PLATFORM, + ), + ) + registration = client.make_credential(options) + credential_data = registration.response.attestation_object.auth_data.credential_data + if credential_data is None: + raise AuthzVerificationError("registration returned no credential data") + return EnrolledCredential( + human_id=human_id, + label=label, + credential_id=b64url_encode(credential_data.credential_id), + public_key_cose=b64url_encode(fido2.cbor.encode(credential_data.public_key)), + aaguid=bytes(credential_data.aaguid).hex(), + rp_id=resolved_rp, + ) + + +def get_assertion_receipt( + request: AuthorizationRequest, + credential: EnrolledCredential, +) -> HumanAuthorizationReceipt: # pragma: no cover - requires hardware + """Run a get-assertion ceremony over ``compute_challenge(request)``. + + UV policy comes from the request: ``user_verification_required=True`` + demands a PIN/biometric ceremony; otherwise user presence suffices. + """ + _import_fido2() + from fido2.webauthn import ( + PublicKeyCredentialDescriptor, + PublicKeyCredentialRequestOptions, + PublicKeyCredentialType, + UserVerificationRequirement, + ) + + client = _make_client() + options = PublicKeyCredentialRequestOptions( + challenge=compute_challenge(request), + rp_id=request.rp_id, + allow_credentials=[ + PublicKeyCredentialDescriptor( + type=PublicKeyCredentialType.PUBLIC_KEY, + id=b64url_decode(credential.credential_id), + ) + ], + user_verification=( + UserVerificationRequirement.REQUIRED + if request.user_verification_required + else UserVerificationRequirement.DISCOURAGED + ), + ) + selection = client.get_assertion(options) + response = selection.get_response(0) + return HumanAuthorizationReceipt( + request=request, + credential_id=b64url_encode(response.raw_id), + assertion_response=dict(response), + user_verified=response.response.authenticator_data.is_user_verified(), + ) + + +# --- Offline verification ----------------------------------------------------- + +def verify_receipt( + receipt: HumanAuthorizationReceipt, + store: CredentialStore, + now: Optional[datetime] = None, +) -> EnrolledCredential: + """Full offline verification of a receipt against enrolled credentials. + + Order: + 1. Structural checks (expiry, clientData type/challenge/origin) — no + dependency required, fails fast with a stable reason code. + 2. Credential policy: enrolled, not revoked, RP matches the request. + 3. ``Fido2Server.authenticate_complete``: RP-ID hash, user presence, + user verification when the request required it, credential match, + and the assertion signature. Counters are not enforced. + + Returns the enrolled credential on success; raises + AuthzVerificationError otherwise. + """ + structural = verify_receipt_structure(receipt, now) + if not structural.passed: + raise AuthzVerificationError(structural.failure_reason) + + enrolled = store.find_by_credential_id(receipt.credential_id) + if enrolled is None: + raise AuthzVerificationError(VERIFY_FAIL_UNKNOWN_CREDENTIAL) + if enrolled.revoked: + raise AuthzVerificationError(VERIFY_FAIL_REVOKED_CREDENTIAL) + if enrolled.rp_id != receipt.request.rp_id: + raise AuthzVerificationError(VERIFY_FAIL_RP_MISMATCH) + if normalize_identity(enrolled.human_id) != receipt.request.approver_identity_id: + # The asserting credential must belong to the identity the request + # names; otherwise any enrolled human's key could satisfy it. + raise AuthzVerificationError(VERIFY_FAIL_APPROVER_MISMATCH) + + fido2 = _import_fido2() + from fido2.server import Fido2Server + from fido2.utils import websafe_encode + from fido2.webauthn import ( + AttestedCredentialData, + PublicKeyCredentialRpEntity, + UserVerificationRequirement, + ) + + server = Fido2Server( + PublicKeyCredentialRpEntity(id=receipt.request.rp_id, name=RP_NAME), + verify_origin=lambda origin: origin == ORIGIN, + ) + state = { + "challenge": websafe_encode(compute_challenge(receipt.request)), + "user_verification": ( + UserVerificationRequirement.REQUIRED + if receipt.request.user_verification_required + else UserVerificationRequirement.DISCOURAGED + ), + } + registered = AttestedCredentialData.create( + bytes.fromhex(enrolled.aaguid), + b64url_decode(enrolled.credential_id), + fido2.cose.CoseKey.parse( + fido2.cbor.decode(b64url_decode(enrolled.public_key_cose)) + ), + ) + try: + server.authenticate_complete( + state, [registered], receipt.assertion_response + ) + except ValueError as exc: + raise AuthzVerificationError(str(exc)) from exc + except Exception as exc: + raise AuthzVerificationError( + f"assertion verification failed: {exc}" + ) from exc + return enrolled