diff --git a/packages/flow/openadapt_flow/approval_mode/item_approval.py b/packages/flow/openadapt_flow/approval_mode/item_approval.py index 58cce9463b..4fb2693500 100644 --- a/packages/flow/openadapt_flow/approval_mode/item_approval.py +++ b/packages/flow/openadapt_flow/approval_mode/item_approval.py @@ -75,6 +75,14 @@ DeclineReason, ItemApprovalRef, ) +from openadapt_flow.approval_mode.named_person_binding import ( + AssertionMaterial, + ExecutionProfile, + NamedPersonBindingV1, + NamedPersonVerifier, + PersonCredentialRegistry, + require_named_person_binding, +) from openadapt_flow.approval_mode.signed_policy import ( GrantRefused, SignedApproverGrant, @@ -395,6 +403,18 @@ class ApproverContext: #: Why the run's pin can't be trusted (another root or key directory #: than the run recorded), or ``None``. A person's answer is then refused. pin_refusal: Optional[str] = None + #: Execution profile for this run. Under ``regulated``, a named-person + #: binding (:mod:`.named_person_binding`) is required and verified + #: fail-closed. ``None`` or ``demo`` skips that check (device/key + #: attestation alone still applies). + execution_profile: Optional[ExecutionProfile] = None + #: Seal-safe named-person binding for the answer being checked. + named_person_binding: Optional[NamedPersonBindingV1] = None + #: Raw assertion material held only in memory for verification. Never + #: logged; never written to the run directory or Seal. + named_person_assertion: Optional[AssertionMaterial] = None + named_person_registry: Optional[PersonCredentialRegistry] = None + named_person_verifier: Optional[NamedPersonVerifier] = None def _as_context( @@ -512,6 +532,24 @@ def answer_refusal( refusal_code = _person_refusal(answer, context) if refusal_code is not None: return refusal_code + # Regulated runs also bind the approval to a named person (WebAuthn / + # OIDC / SAML). Device/key attestation above is not enough alone. + expected_person = None + if answer.approver_id is not None: + expected_person = answer.approver_id + elif answer.approver is not None and answer.approver.id is not None: + expected_person = answer.approver.id + named_refusal = require_named_person_binding( + profile=context.execution_profile, + approval_payload_sha256=request.digest(), + expected_person_id=expected_person, + binding=context.named_person_binding, + assertion=context.named_person_assertion, + registry=context.named_person_registry, + verifier=context.named_person_verifier, + ) + if named_refusal is not None: + return named_refusal else: if answer.attestation is not None: return "answer_fields_inconsistent" diff --git a/packages/flow/openadapt_flow/approval_mode/named_person_binding.py b/packages/flow/openadapt_flow/approval_mode/named_person_binding.py new file mode 100644 index 0000000000..e2383e182b --- /dev/null +++ b/packages/flow/openadapt_flow/approval_mode/named_person_binding.py @@ -0,0 +1,718 @@ +"""Bind a regulated approval to a named person (WebAuthn, OIDC or SAML). + +Phone pairing and the local approver-roster Ed25519 keys identify a *device* +or a *registered key*, not an authenticated individual. For regulated audit +trails (21 CFR Part 11, HIPAA, SOC 2) each approval must also carry a +:class:`NamedPersonBindingV1`: a WebAuthn passkey assertion over the exact +approval payload hash, or an OIDC / SAML assertion from the organisation's +identity provider. The binding (person id + evidence digests) goes into the +Seal; raw assertion material never does. + +Fail closed under the ``regulated`` profile: missing binding, payload +mismatch, unknown credential, failed assertion, or an unavailable verifier +are all refusals. ``demo`` never requires a binding. ``standard`` accepts a +verified binding when present but does not yet require one (later slices). + +No production IdP credentials live here. Unit tests use +:class:`DeterministicPasskeyDouble` and recorded fixture shapes, clearly +labeled as test doubles — not a live WebAuthn ceremony. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import logging +from dataclasses import dataclass +from enum import Enum +from typing import ( + Any, + Literal, + Optional, + Protocol, + Sequence, + Union, +) + +from pydantic import BaseModel, ConfigDict, Field, StrictStr, model_validator + +logger = logging.getLogger(__name__) + +NAMED_PERSON_BINDING_SCHEMA: Literal["openadapt.named-person-binding/v1"] = ( + "openadapt.named-person-binding/v1" +) +SEAL_EXTENSION_KEY = "openadapt.named-person-binding/v1" + +_OPAQUE_ID = r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,255}$" +_SHA256 = r"^sha256:[0-9a-f]{64}$" +_RP_ID = r"^[a-z0-9][a-z0-9.-]{0,253}$" +_ISO_Z = r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$" + +ExecutionProfile = Literal["demo", "standard", "regulated"] + + +class BindingMethod(str, Enum): + """How the named person was authenticated for this approval.""" + + WEBAUTHN_PASSKEY = "webauthn_passkey" + OIDC = "oidc" + SAML = "saml" + + +class NamedPersonRefusal(str, Enum): + """Closed reasons a regulated named-person check can return.""" + + MISSING_BINDING = "named_person_binding_missing" + MISSING_ASSERTION = "named_person_assertion_missing" + PAYLOAD_MISMATCH = "named_person_payload_mismatch" + PERSON_MISMATCH = "named_person_id_mismatch" + CREDENTIAL_UNKNOWN = "named_person_credential_unknown" + ASSERTION_INVALID = "named_person_assertion_invalid" + METHOD_UNSUPPORTED = "named_person_method_unsupported" + VERIFIER_UNAVAILABLE = "named_person_verifier_unavailable" + REGISTRY_UNAVAILABLE = "named_person_registry_unavailable" + EVIDENCE_DIGEST_MISMATCH = "named_person_evidence_digest_mismatch" + ISSUER_UNTRUSTED = "named_person_issuer_untrusted" + SUBJECT_MISMATCH = "named_person_subject_mismatch" + + +class _Strict(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + +def _sha256_bytes(data: bytes) -> str: + return "sha256:" + hashlib.sha256(data).hexdigest() + + +def _sha256_text(value: str) -> str: + return _sha256_bytes(value.encode("utf-8")) + + +def approval_payload_digest(payload: Any) -> str: + """Canonical sha256 of the exact approval payload the person asserted over. + + Matches the compact sorted-key JSON used elsewhere in Flow approvals. + """ + + if isinstance(payload, BaseModel): + payload = payload.model_dump(mode="json") + raw = json.dumps( + payload, ensure_ascii=True, separators=(",", ":"), sort_keys=True + ).encode("utf-8") + return _sha256_bytes(raw) + + +class WebAuthnEvidenceV1(_Strict): + """Public, PHI-free digests of one WebAuthn assertion (Seal-safe). + + Raw authenticatorData / clientDataJSON / signature bytes are *not* stored + here and must never be logged. Verification sees them only in memory via + :class:`WebAuthnAssertionMaterial`. + """ + + credential_id_sha256: StrictStr = Field(pattern=_SHA256) + rp_id: StrictStr = Field(pattern=_RP_ID) + user_handle_sha256: Optional[StrictStr] = Field(default=None, pattern=_SHA256) + authenticator_data_sha256: StrictStr = Field(pattern=_SHA256) + client_data_sha256: StrictStr = Field(pattern=_SHA256) + signature_sha256: StrictStr = Field(pattern=_SHA256) + #: Challenge the authenticator signed; must equal ``approval_payload_sha256``. + challenge_sha256: StrictStr = Field(pattern=_SHA256) + + +class OidcEvidenceV1(_Strict): + """Public digests of one OIDC ID-token claim set (Seal-safe).""" + + issuer: StrictStr = Field(min_length=1, max_length=512) + subject_sha256: StrictStr = Field(pattern=_SHA256) + audience_sha256: StrictStr = Field(pattern=_SHA256) + token_sha256: StrictStr = Field(pattern=_SHA256) + #: Nonce bound to the approval payload digest. + nonce_sha256: StrictStr = Field(pattern=_SHA256) + + +class SamlEvidenceV1(_Strict): + """Public digests of one SAML Response assertion (Seal-safe).""" + + issuer: StrictStr = Field(min_length=1, max_length=512) + subject_sha256: StrictStr = Field(pattern=_SHA256) + audience_sha256: StrictStr = Field(pattern=_SHA256) + assertion_sha256: StrictStr = Field(pattern=_SHA256) + #: InResponseTo / request id bound to the approval payload digest. + in_response_to_sha256: StrictStr = Field(pattern=_SHA256) + + +class NamedPersonBindingV1(_Strict): + """Who was authenticated for one approval, for the audit trail and Seal. + + PHI-free: opaque person ids and digests only. Never carries names, emails + or raw assertion bytes. + """ + + schema_version: Literal["openadapt.named-person-binding/v1"] = ( + NAMED_PERSON_BINDING_SCHEMA + ) + person_id: StrictStr = Field(pattern=_OPAQUE_ID) + method: BindingMethod + approval_payload_sha256: StrictStr = Field(pattern=_SHA256) + evidence: Union[WebAuthnEvidenceV1, OidcEvidenceV1, SamlEvidenceV1] + verified_at: StrictStr = Field(pattern=_ISO_Z) + + @model_validator(mode="after") + def _method_matches_evidence(self) -> "NamedPersonBindingV1": + if self.method is BindingMethod.WEBAUTHN_PASSKEY: + if not isinstance(self.evidence, WebAuthnEvidenceV1): + raise ValueError("webauthn_passkey needs WebAuthnEvidenceV1") + if self.evidence.challenge_sha256 != self.approval_payload_sha256: + raise ValueError("WebAuthn challenge must equal approval payload hash") + elif self.method is BindingMethod.OIDC: + if not isinstance(self.evidence, OidcEvidenceV1): + raise ValueError("oidc needs OidcEvidenceV1") + if self.evidence.nonce_sha256 != self.approval_payload_sha256: + raise ValueError("OIDC nonce must equal approval payload hash") + elif self.method is BindingMethod.SAML: + if not isinstance(self.evidence, SamlEvidenceV1): + raise ValueError("saml needs SamlEvidenceV1") + if self.evidence.in_response_to_sha256 != self.approval_payload_sha256: + raise ValueError("SAML InResponseTo must equal approval payload hash") + return self + + +@dataclass(frozen=True) +class WebAuthnAssertionMaterial: + """Raw WebAuthn assertion bytes held only in memory for verification. + + ``repr`` and ``str`` redact every field so logs and tracebacks never print + assertion material. + """ + + credential_id: bytes + authenticator_data: bytes + client_data_json: bytes + signature: bytes + user_handle: bytes | None = None + + def __repr__(self) -> str: # noqa: D105 + return ( + "WebAuthnAssertionMaterial(credential_id=, " + "authenticator_data=, client_data_json=, " + "signature=, user_handle=)" + ) + + __str__ = __repr__ + + def evidence_digests(self, *, rp_id: str) -> WebAuthnEvidenceV1: + challenge = _client_data_challenge(self.client_data_json) + if challenge is None: + raise NamedPersonBindingError( + NamedPersonRefusal.ASSERTION_INVALID, + "WebAuthn clientDataJSON has no challenge", + ) + return WebAuthnEvidenceV1( + credential_id_sha256=_sha256_bytes(self.credential_id), + rp_id=rp_id, + user_handle_sha256=( + _sha256_bytes(self.user_handle) if self.user_handle else None + ), + authenticator_data_sha256=_sha256_bytes(self.authenticator_data), + client_data_sha256=_sha256_bytes(self.client_data_json), + signature_sha256=_sha256_bytes(self.signature), + challenge_sha256="sha256:" + challenge, + ) + + +@dataclass(frozen=True) +class FederatedAssertionMaterial: + """Raw OIDC token or SAML response held only in memory for verification.""" + + method: Literal["oidc", "saml"] + issuer: str + subject: str + audience: str + raw: bytes + binding_value: str # nonce or InResponseTo + + def __repr__(self) -> str: # noqa: D105 + return ( + f"FederatedAssertionMaterial(method={self.method!r}, " + "issuer=, subject=, audience=, " + "raw=, binding_value=)" + ) + + __str__ = __repr__ + + +AssertionMaterial = Union[WebAuthnAssertionMaterial, FederatedAssertionMaterial] + + +@dataclass(frozen=True) +class RegisteredPasskey: + """One person's enrolled passkey in the relying-party registry.""" + + person_id: str + credential_id: bytes + #: COSE / PEM / test-double public key material. Never a private key. + public_key: bytes + rp_id: str + + +@dataclass(frozen=True) +class RegisteredFederatedSubject: + """One person's IdP subject for OIDC or SAML.""" + + person_id: str + method: Literal["oidc", "saml"] + issuer: str + subject: str + audience: str + + +class PersonCredentialRegistry(Protocol): + def passkey(self, credential_id: bytes) -> RegisteredPasskey | None: ... + + def federated( + self, *, method: Literal["oidc", "saml"], issuer: str, subject: str + ) -> RegisteredFederatedSubject | None: ... + + +class NamedPersonVerifier(Protocol): + """Verifies raw assertion material against a registered credential.""" + + def verify_webauthn( + self, + *, + challenge_sha256: str, + assertion: WebAuthnAssertionMaterial, + registered: RegisteredPasskey, + ) -> None: + """Raise :class:`NamedPersonBindingError` on failure.""" + + def verify_federated( + self, + *, + challenge_sha256: str, + assertion: FederatedAssertionMaterial, + registered: RegisteredFederatedSubject, + ) -> None: + """Raise :class:`NamedPersonBindingError` on failure.""" + + +class NamedPersonBindingError(Exception): + def __init__(self, code: NamedPersonRefusal, message: str) -> None: + super().__init__(message) + self.code = code + + +@dataclass(frozen=True) +class MemoryPersonRegistry: + """In-memory registry for tests and offline runners.""" + + passkeys: Sequence[RegisteredPasskey] = () + federated_subjects: Sequence[RegisteredFederatedSubject] = () + + def passkey(self, credential_id: bytes) -> RegisteredPasskey | None: + for entry in self.passkeys: + if hmac.compare_digest(entry.credential_id, credential_id): + return entry + return None + + def federated( + self, *, method: Literal["oidc", "saml"], issuer: str, subject: str + ) -> RegisteredFederatedSubject | None: + for entry in self.federated_subjects: + if ( + entry.method == method + and entry.issuer == issuer + and entry.subject == subject + ): + return entry + return None + + +def _client_data_challenge(client_data_json: bytes) -> str | None: + try: + data = json.loads(client_data_json.decode("utf-8")) + except (UnicodeDecodeError, ValueError): + return None + if not isinstance(data, dict): + return None + challenge = data.get("challenge") + if not isinstance(challenge, str) or not challenge: + return None + # Challenge is the approval payload digest *without* the ``sha256:`` prefix, + # urlsafe-base64 or hex. We accept the hex of the 32-byte digest. + return challenge + + +def _challenge_hex(challenge_sha256: str) -> str: + return challenge_sha256.removeprefix("sha256:") + + +class DeterministicPasskeyDouble: + """Test-double WebAuthn / federated verifier (NOT a live ceremony). + + Uses HMAC-SHA256 over a domain-separated message so unit tests are + deterministic without ``py_webauthn``, a virtual authenticator, or an IdP. + Production slices replace this with a ``py_webauthn``-backed verifier. + Fixture assertions built by :meth:`sign_webauthn` / :meth:`sign_federated` + are labeled test doubles. + """ + + DOMAIN_WEBAUTHN = b"openadapt.named-person.webauthn-double/v1\x00" + DOMAIN_FEDERATED = b"openadapt.named-person.federated-double/v1\x00" + + def sign_webauthn( + self, + *, + challenge_sha256: str, + registered: RegisteredPasskey, + credential_id: bytes | None = None, + user_handle: bytes | None = None, + ) -> WebAuthnAssertionMaterial: + """Build a fixture assertion the double will accept.""" + + cred = credential_id if credential_id is not None else registered.credential_id + challenge_hex = _challenge_hex(challenge_sha256) + client_data = json.dumps( + { + "type": "webauthn.get", + "origin": f"https://{registered.rp_id}", + "challenge": challenge_hex, + "fixture": "openadapt.named-person.webauthn-double/v1", + }, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + # Minimal authenticator data: rpIdHash(32) + flags(1) + counter(4). + rp_hash = hashlib.sha256(registered.rp_id.encode("utf-8")).digest() + authenticator_data = rp_hash + bytes([0x01, 0x00, 0x00, 0x00, 0x01]) + signature = hmac.new( + registered.public_key, + self.DOMAIN_WEBAUTHN + + challenge_hex.encode("ascii") + + authenticator_data + + hashlib.sha256(client_data).digest(), + hashlib.sha256, + ).digest() + return WebAuthnAssertionMaterial( + credential_id=cred, + authenticator_data=authenticator_data, + client_data_json=client_data, + signature=signature, + user_handle=user_handle, + ) + + def sign_federated( + self, + *, + challenge_sha256: str, + registered: RegisteredFederatedSubject, + ) -> FederatedAssertionMaterial: + """Build a fixture OIDC/SAML assertion the double will accept. + + The MAC key is the registered audience string. That is a test-double + convention only — never a production IdP secret. + """ + + binding = _challenge_hex(challenge_sha256) + body = { + "iss": registered.issuer, + "sub": registered.subject, + "aud": registered.audience, + ("nonce" if registered.method == "oidc" else "in_response_to"): binding, + "fixture": "openadapt.named-person.federated-double/v1", + } + raw = json.dumps(body, separators=(",", ":"), sort_keys=True).encode("utf-8") + mac = hmac.new( + registered.audience.encode("utf-8"), + self.DOMAIN_FEDERATED + raw, + hashlib.sha256, + ).hexdigest() + return FederatedAssertionMaterial( + method=registered.method, + issuer=registered.issuer, + subject=registered.subject, + audience=registered.audience, + raw=raw + b"." + mac.encode("ascii"), + binding_value=binding, + ) + + def verify_webauthn( + self, + *, + challenge_sha256: str, + assertion: WebAuthnAssertionMaterial, + registered: RegisteredPasskey, + ) -> None: + if not hmac.compare_digest(assertion.credential_id, registered.credential_id): + raise NamedPersonBindingError( + NamedPersonRefusal.CREDENTIAL_UNKNOWN, + "assertion credential is not the registered passkey", + ) + challenge = _client_data_challenge(assertion.client_data_json) + if challenge != _challenge_hex(challenge_sha256): + raise NamedPersonBindingError( + NamedPersonRefusal.ASSERTION_INVALID, + "WebAuthn clientData challenge does not match approval payload", + ) + expected = hmac.new( + registered.public_key, + self.DOMAIN_WEBAUTHN + + _challenge_hex(challenge_sha256).encode("ascii") + + assertion.authenticator_data + + hashlib.sha256(assertion.client_data_json).digest(), + hashlib.sha256, + ).digest() + if not hmac.compare_digest(assertion.signature, expected): + raise NamedPersonBindingError( + NamedPersonRefusal.ASSERTION_INVALID, + "WebAuthn assertion signature failed (test double)", + ) + + def verify_federated( + self, + *, + challenge_sha256: str, + assertion: FederatedAssertionMaterial, + registered: RegisteredFederatedSubject, + ) -> None: + if assertion.method != registered.method: + raise NamedPersonBindingError( + NamedPersonRefusal.METHOD_UNSUPPORTED, + "federated method does not match the registered subject", + ) + if assertion.issuer != registered.issuer: + raise NamedPersonBindingError( + NamedPersonRefusal.ISSUER_UNTRUSTED, + "federated issuer is not trusted for this person", + ) + if assertion.subject != registered.subject: + raise NamedPersonBindingError( + NamedPersonRefusal.SUBJECT_MISMATCH, + "federated subject does not match the registered person", + ) + if assertion.audience != registered.audience: + raise NamedPersonBindingError( + NamedPersonRefusal.ASSERTION_INVALID, + "federated audience mismatch", + ) + if assertion.binding_value != _challenge_hex(challenge_sha256): + raise NamedPersonBindingError( + NamedPersonRefusal.PAYLOAD_MISMATCH, + "federated nonce/InResponseTo does not match approval payload", + ) + try: + raw, mac = assertion.raw.rsplit(b".", 1) + except ValueError as exc: + raise NamedPersonBindingError( + NamedPersonRefusal.ASSERTION_INVALID, + "federated assertion fixture is malformed", + ) from exc + try: + body = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, ValueError) as exc: + raise NamedPersonBindingError( + NamedPersonRefusal.ASSERTION_INVALID, + "federated assertion body is not JSON", + ) from exc + key_field = "nonce" if registered.method == "oidc" else "in_response_to" + if body.get(key_field) != _challenge_hex(challenge_sha256): + raise NamedPersonBindingError( + NamedPersonRefusal.PAYLOAD_MISMATCH, + "federated body binding field mismatch", + ) + expected = ( + hmac.new( + registered.audience.encode("utf-8"), + self.DOMAIN_FEDERATED + raw, + hashlib.sha256, + ) + .hexdigest() + .encode("ascii") + ) + if not hmac.compare_digest(mac, expected): + raise NamedPersonBindingError( + NamedPersonRefusal.ASSERTION_INVALID, + "federated assertion MAC failed (test double)", + ) + + +def build_binding( + *, + person_id: str, + method: BindingMethod, + approval_payload_sha256: str, + assertion: AssertionMaterial, + rp_id: str | None = None, + verified_at: str, +) -> NamedPersonBindingV1: + """Build a Seal-safe binding from in-memory assertion material.""" + + evidence: Union[WebAuthnEvidenceV1, OidcEvidenceV1, SamlEvidenceV1] + if isinstance(assertion, WebAuthnAssertionMaterial): + if method is not BindingMethod.WEBAUTHN_PASSKEY: + raise NamedPersonBindingError( + NamedPersonRefusal.METHOD_UNSUPPORTED, + "WebAuthn material needs method=webauthn_passkey", + ) + if not rp_id: + raise NamedPersonBindingError( + NamedPersonRefusal.ASSERTION_INVALID, + "WebAuthn binding needs an rp_id", + ) + evidence = assertion.evidence_digests(rp_id=rp_id) + if evidence.challenge_sha256 != approval_payload_sha256: + raise NamedPersonBindingError( + NamedPersonRefusal.PAYLOAD_MISMATCH, + "WebAuthn challenge does not match approval payload hash", + ) + else: + if method is BindingMethod.OIDC: + evidence = OidcEvidenceV1( + issuer=assertion.issuer, + subject_sha256=_sha256_text(assertion.subject), + audience_sha256=_sha256_text(assertion.audience), + token_sha256=_sha256_bytes(assertion.raw), + nonce_sha256=approval_payload_sha256, + ) + elif method is BindingMethod.SAML: + evidence = SamlEvidenceV1( + issuer=assertion.issuer, + subject_sha256=_sha256_text(assertion.subject), + audience_sha256=_sha256_text(assertion.audience), + assertion_sha256=_sha256_bytes(assertion.raw), + in_response_to_sha256=approval_payload_sha256, + ) + else: + raise NamedPersonBindingError( + NamedPersonRefusal.METHOD_UNSUPPORTED, + "federated material needs method=oidc or saml", + ) + return NamedPersonBindingV1( + person_id=person_id, + method=method, + approval_payload_sha256=approval_payload_sha256, + evidence=evidence, + verified_at=verified_at, + ) + + +def seal_extension(binding: NamedPersonBindingV1) -> dict[str, Any]: + """Seal ``extensions`` entry: identity + evidence digests, never raw bytes.""" + + return { + "schema": SEAL_EXTENSION_KEY, + "binding": binding.model_dump(mode="json"), + } + + +def require_named_person_binding( + *, + profile: ExecutionProfile | None, + approval_payload_sha256: str, + expected_person_id: str | None, + binding: NamedPersonBindingV1 | None, + assertion: AssertionMaterial | None, + registry: PersonCredentialRegistry | None, + verifier: NamedPersonVerifier | None, +) -> Optional[str]: + """Return a closed refusal code, or ``None`` when the check passes. + + Under ``regulated``, a missing or unverifiable named person always + refuses. Under ``demo``, the check is skipped. Under ``standard``, a + present binding is verified fail-closed, but absence is still allowed + in this first slice. + """ + + if profile is None or profile == "demo": + return None + required = profile == "regulated" + if binding is None: + return NamedPersonRefusal.MISSING_BINDING.value if required else None + if assertion is None: + return NamedPersonRefusal.MISSING_ASSERTION.value + if registry is None: + return NamedPersonRefusal.REGISTRY_UNAVAILABLE.value + if verifier is None: + return NamedPersonRefusal.VERIFIER_UNAVAILABLE.value + if binding.approval_payload_sha256 != approval_payload_sha256: + return NamedPersonRefusal.PAYLOAD_MISMATCH.value + if expected_person_id is not None and binding.person_id != expected_person_id: + return NamedPersonRefusal.PERSON_MISMATCH.value + + try: + if binding.method is BindingMethod.WEBAUTHN_PASSKEY: + if not isinstance(assertion, WebAuthnAssertionMaterial): + return NamedPersonRefusal.METHOD_UNSUPPORTED.value + if not isinstance(binding.evidence, WebAuthnEvidenceV1): + return NamedPersonRefusal.METHOD_UNSUPPORTED.value + expected_evidence = assertion.evidence_digests( + rp_id=binding.evidence.rp_id, + ) + if expected_evidence != binding.evidence: + return NamedPersonRefusal.EVIDENCE_DIGEST_MISMATCH.value + registered = registry.passkey(assertion.credential_id) + if registered is None: + return NamedPersonRefusal.CREDENTIAL_UNKNOWN.value + if registered.person_id != binding.person_id: + return NamedPersonRefusal.PERSON_MISMATCH.value + verifier.verify_webauthn( + challenge_sha256=approval_payload_sha256, + assertion=assertion, + registered=registered, + ) + elif binding.method in {BindingMethod.OIDC, BindingMethod.SAML}: + if not isinstance(assertion, FederatedAssertionMaterial): + return NamedPersonRefusal.METHOD_UNSUPPORTED.value + method: Literal["oidc", "saml"] = ( + "oidc" if binding.method is BindingMethod.OIDC else "saml" + ) + registered_f = registry.federated( + method=method, issuer=assertion.issuer, subject=assertion.subject + ) + if registered_f is None: + return NamedPersonRefusal.CREDENTIAL_UNKNOWN.value + if registered_f.person_id != binding.person_id: + return NamedPersonRefusal.PERSON_MISMATCH.value + verifier.verify_federated( + challenge_sha256=approval_payload_sha256, + assertion=assertion, + registered=registered_f, + ) + else: + return NamedPersonRefusal.METHOD_UNSUPPORTED.value + except NamedPersonBindingError as exc: + return exc.code.value + except Exception: + # Never turn an unexpected verifier crash into a pass. + logger.exception("named-person verifier failed closed") + return NamedPersonRefusal.ASSERTION_INVALID.value + return None + + +__all__ = [ + "NAMED_PERSON_BINDING_SCHEMA", + "SEAL_EXTENSION_KEY", + "AssertionMaterial", + "BindingMethod", + "DeterministicPasskeyDouble", + "ExecutionProfile", + "FederatedAssertionMaterial", + "MemoryPersonRegistry", + "NamedPersonBindingError", + "NamedPersonBindingV1", + "NamedPersonRefusal", + "NamedPersonVerifier", + "OidcEvidenceV1", + "PersonCredentialRegistry", + "RegisteredFederatedSubject", + "RegisteredPasskey", + "SamlEvidenceV1", + "WebAuthnAssertionMaterial", + "WebAuthnEvidenceV1", + "approval_payload_digest", + "build_binding", + "require_named_person_binding", + "seal_extension", +] diff --git a/packages/flow/tests/fixtures/named_person_binding/README.md b/packages/flow/tests/fixtures/named_person_binding/README.md new file mode 100644 index 0000000000..8c698a01db --- /dev/null +++ b/packages/flow/tests/fixtures/named_person_binding/README.md @@ -0,0 +1,9 @@ +# Named-person binding fixtures (test doubles) + +These files are **synthetic test doubles**, not live WebAuthn ceremonies or +IdP traffic. They record the deterministic shapes produced by +`DeterministicPasskeyDouble` in `openadapt_flow.approval_mode.named_person_binding`. + +Do not treat them as production attestation material. A later slice will add +`py_webauthn`-backed verification against a virtual authenticator; until then +every assertion here is HMAC-domain-separated fixture data. diff --git a/packages/flow/tests/fixtures/named_person_binding/webauthn_passkey_double.json b/packages/flow/tests/fixtures/named_person_binding/webauthn_passkey_double.json new file mode 100644 index 0000000000..9ce5c5c809 --- /dev/null +++ b/packages/flow/tests/fixtures/named_person_binding/webauthn_passkey_double.json @@ -0,0 +1,56 @@ +{ + "approval_payload": { + "decision": "approve", + "request_id": "req-fixture-0001", + "schema_version": "openadapt.item-approval-request/v1" + }, + "approval_payload_sha256": "sha256:0df621837e1f30a7a32576d976d80d5d5b0785fa0eeaadea232053040972607d", + "assertion_hex": { + "authenticator_data": "f032c72798a7c370874c6b07ea18aa24ed30efc30af571e0fef5057bd1b8e4a00100000001", + "client_data_json": "{\"challenge\":\"0df621837e1f30a7a32576d976d80d5d5b0785fa0eeaadea232053040972607d\",\"fixture\":\"openadapt.named-person.webauthn-double/v1\",\"origin\":\"https://approve.openadapt.test\",\"type\":\"webauthn.get\"}", + "credential_id": "c0ffee0000000000000000000000000000000000000000000000000000000000", + "signature": "6a5bdea20aa5571a021d986e13e300edf39d5c2e6edd673857cc27eda6ed8493" + }, + "binding": { + "approval_payload_sha256": "sha256:0df621837e1f30a7a32576d976d80d5d5b0785fa0eeaadea232053040972607d", + "evidence": { + "authenticator_data_sha256": "sha256:b965ac9d59d891b6f7deeda46c377a115253a1a0ca80067de56df86c92f6295a", + "challenge_sha256": "sha256:0df621837e1f30a7a32576d976d80d5d5b0785fa0eeaadea232053040972607d", + "client_data_sha256": "sha256:9350621ef5b18b257430975a707aa38ca90dea325694e6cad1018a7dbe1bef9f", + "credential_id_sha256": "sha256:5db8414a3b4b48b05f2d27dce4e768381f255881cd351d4afcf6ba690743ba9e", + "rp_id": "approve.openadapt.test", + "signature_sha256": "sha256:fdaa9d3ae9ab0baa1082f13c7bc5285cf904100d48d60b7fd826d0042cf77db5", + "user_handle_sha256": null + }, + "method": "webauthn_passkey", + "person_id": "person:fixture-alice", + "schema_version": "openadapt.named-person-binding/v1", + "verified_at": "2026-11-05T17:00:00Z" + }, + "note": "Synthetic DeterministicPasskeyDouble fixture \u2014 NOT a live WebAuthn ceremony.", + "registered_passkey": { + "credential_id_hex": "c0ffee0000000000000000000000000000000000000000000000000000000000", + "person_id": "person:fixture-alice", + "public_key_hex": "1111111111111111111111111111111111111111111111111111111111111111", + "rp_id": "approve.openadapt.test" + }, + "seal_extension": { + "binding": { + "approval_payload_sha256": "sha256:0df621837e1f30a7a32576d976d80d5d5b0785fa0eeaadea232053040972607d", + "evidence": { + "authenticator_data_sha256": "sha256:b965ac9d59d891b6f7deeda46c377a115253a1a0ca80067de56df86c92f6295a", + "challenge_sha256": "sha256:0df621837e1f30a7a32576d976d80d5d5b0785fa0eeaadea232053040972607d", + "client_data_sha256": "sha256:9350621ef5b18b257430975a707aa38ca90dea325694e6cad1018a7dbe1bef9f", + "credential_id_sha256": "sha256:5db8414a3b4b48b05f2d27dce4e768381f255881cd351d4afcf6ba690743ba9e", + "rp_id": "approve.openadapt.test", + "signature_sha256": "sha256:fdaa9d3ae9ab0baa1082f13c7bc5285cf904100d48d60b7fd826d0042cf77db5", + "user_handle_sha256": null + }, + "method": "webauthn_passkey", + "person_id": "person:fixture-alice", + "schema_version": "openadapt.named-person-binding/v1", + "verified_at": "2026-11-05T17:00:00Z" + }, + "schema": "openadapt.named-person-binding/v1" + } +} diff --git a/packages/flow/tests/test_named_person_binding.py b/packages/flow/tests/test_named_person_binding.py new file mode 100644 index 0000000000..3d84371db8 --- /dev/null +++ b/packages/flow/tests/test_named_person_binding.py @@ -0,0 +1,433 @@ +"""Named-person binding for regulated approvals (Part of #1888). + +Phone pairing / roster Ed25519 keys identify a device or key, not a person. +Under the regulated profile an approval without a verified WebAuthn / OIDC / +SAML binding to a named person must refuse. These tests use +DeterministicPasskeyDouble fixtures — not a live WebAuthn ceremony. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from openadapt_flow.approval_mode.item_approval import ( + ApproverContext, + ItemApprovalAnswer, + ItemApprovalStore, + answer_refusal, +) +from openadapt_flow.approval_mode.model import ApprovalMode +from openadapt_flow.approval_mode.named_person_binding import ( + BindingMethod, + DeterministicPasskeyDouble, + MemoryPersonRegistry, + NamedPersonBindingError, + NamedPersonBindingV1, + NamedPersonRefusal, + RegisteredFederatedSubject, + RegisteredPasskey, + WebAuthnAssertionMaterial, + approval_payload_digest, + build_binding, + require_named_person_binding, + seal_extension, +) + +FIXTURE = ( + Path(__file__).resolve().parent + / "fixtures" + / "named_person_binding" + / "webauthn_passkey_double.json" +) + +FRAME = b"\x89PNG\r\n\x1a\nmade-up filled form" +PERSON = "person:alice" +RP_ID = "approve.openadapt.test" + + +@pytest.fixture +def passkey() -> RegisteredPasskey: + return RegisteredPasskey( + person_id=PERSON, + credential_id=b"cred-alice-1", + public_key=b"P" * 32, + rp_id=RP_ID, + ) + + +@pytest.fixture +def double() -> DeterministicPasskeyDouble: + return DeterministicPasskeyDouble() + + +@pytest.fixture +def registry(passkey: RegisteredPasskey) -> MemoryPersonRegistry: + return MemoryPersonRegistry(passkeys=[passkey]) + + +def _payload() -> dict: + return { + "request_id": "req-1", + "workflow": "referral", + "decision": "approve", + "fields": ["status"], + } + + +def test_regulated_refuses_when_binding_is_missing(passkey, double, registry) -> None: + digest = approval_payload_digest(_payload()) + code = require_named_person_binding( + profile="regulated", + approval_payload_sha256=digest, + expected_person_id=PERSON, + binding=None, + assertion=None, + registry=registry, + verifier=double, + ) + assert code == NamedPersonRefusal.MISSING_BINDING.value + + +def test_demo_does_not_require_binding() -> None: + assert ( + require_named_person_binding( + profile="demo", + approval_payload_sha256="sha256:" + "0" * 64, + expected_person_id=None, + binding=None, + assertion=None, + registry=None, + verifier=None, + ) + is None + ) + + +def test_standard_allows_absent_binding_in_this_slice() -> None: + assert ( + require_named_person_binding( + profile="standard", + approval_payload_sha256="sha256:" + "0" * 64, + expected_person_id=None, + binding=None, + assertion=None, + registry=None, + verifier=None, + ) + is None + ) + + +def test_webauthn_passkey_binding_verifies(passkey, double, registry) -> None: + digest = approval_payload_digest(_payload()) + assertion = double.sign_webauthn(challenge_sha256=digest, registered=passkey) + binding = build_binding( + person_id=PERSON, + method=BindingMethod.WEBAUTHN_PASSKEY, + approval_payload_sha256=digest, + assertion=assertion, + rp_id=RP_ID, + verified_at="2026-11-05T17:00:00Z", + ) + assert ( + require_named_person_binding( + profile="regulated", + approval_payload_sha256=digest, + expected_person_id=PERSON, + binding=binding, + assertion=assertion, + registry=registry, + verifier=double, + ) + is None + ) + ext = seal_extension(binding) + assert ext["schema"] == "openadapt.named-person-binding/v1" + dumped = json.dumps(ext) + # Seal may name digest fields like signature_sha256; never the raw bytes. + assert assertion.signature.hex() not in dumped + assert "signature_sha256" in dumped + + +def test_webauthn_wrong_payload_challenge_refuses(passkey, double, registry) -> None: + digest = approval_payload_digest(_payload()) + other = approval_payload_digest({"other": True}) + assertion = double.sign_webauthn(challenge_sha256=other, registered=passkey) + with pytest.raises(NamedPersonBindingError) as excinfo: + build_binding( + person_id=PERSON, + method=BindingMethod.WEBAUTHN_PASSKEY, + approval_payload_sha256=digest, + assertion=assertion, + rp_id=RP_ID, + verified_at="2026-11-05T17:00:00Z", + ) + assert excinfo.value.code == NamedPersonRefusal.PAYLOAD_MISMATCH + + +def test_webauthn_unknown_credential_refuses(passkey, double) -> None: + digest = approval_payload_digest(_payload()) + assertion = double.sign_webauthn(challenge_sha256=digest, registered=passkey) + binding = build_binding( + person_id=PERSON, + method=BindingMethod.WEBAUTHN_PASSKEY, + approval_payload_sha256=digest, + assertion=assertion, + rp_id=RP_ID, + verified_at="2026-11-05T17:00:00Z", + ) + empty = MemoryPersonRegistry(passkeys=[]) + code = require_named_person_binding( + profile="regulated", + approval_payload_sha256=digest, + expected_person_id=PERSON, + binding=binding, + assertion=assertion, + registry=empty, + verifier=double, + ) + assert code == NamedPersonRefusal.CREDENTIAL_UNKNOWN.value + + +def test_webauthn_tampered_signature_refuses(passkey, double, registry) -> None: + digest = approval_payload_digest(_payload()) + assertion = double.sign_webauthn(challenge_sha256=digest, registered=passkey) + # Rebuild binding evidence to match the tampered assertion so the digest + # gate does not fire first — the verifier must still refuse. + tampered = WebAuthnAssertionMaterial( + credential_id=assertion.credential_id, + authenticator_data=assertion.authenticator_data, + client_data_json=assertion.client_data_json, + signature=b"\x00" * 32, + user_handle=assertion.user_handle, + ) + binding = build_binding( + person_id=PERSON, + method=BindingMethod.WEBAUTHN_PASSKEY, + approval_payload_sha256=digest, + assertion=tampered, + rp_id=RP_ID, + verified_at="2026-11-05T17:00:00Z", + ) + code = require_named_person_binding( + profile="regulated", + approval_payload_sha256=digest, + expected_person_id=PERSON, + binding=binding, + assertion=tampered, + registry=registry, + verifier=double, + ) + assert code == NamedPersonRefusal.ASSERTION_INVALID.value + + +def test_person_id_mismatch_refuses(passkey, double, registry) -> None: + digest = approval_payload_digest(_payload()) + assertion = double.sign_webauthn(challenge_sha256=digest, registered=passkey) + binding = build_binding( + person_id=PERSON, + method=BindingMethod.WEBAUTHN_PASSKEY, + approval_payload_sha256=digest, + assertion=assertion, + rp_id=RP_ID, + verified_at="2026-11-05T17:00:00Z", + ) + code = require_named_person_binding( + profile="regulated", + approval_payload_sha256=digest, + expected_person_id="person:other", + binding=binding, + assertion=assertion, + registry=registry, + verifier=double, + ) + assert code == NamedPersonRefusal.PERSON_MISMATCH.value + + +def test_oidc_and_saml_bindings_verify(double) -> None: + digest = approval_payload_digest(_payload()) + for method, enum in (("oidc", BindingMethod.OIDC), ("saml", BindingMethod.SAML)): + fed = RegisteredFederatedSubject( + person_id="person:bob", + method=method, # type: ignore[arg-type] + issuer="https://idp.example.test", + subject="sub-bob", + audience="openadapt-regulated", + ) + registry = MemoryPersonRegistry(federated_subjects=[fed]) + assertion = double.sign_federated(challenge_sha256=digest, registered=fed) + binding = build_binding( + person_id="person:bob", + method=enum, + approval_payload_sha256=digest, + assertion=assertion, + verified_at="2026-11-05T17:00:00Z", + ) + assert ( + require_named_person_binding( + profile="regulated", + approval_payload_sha256=digest, + expected_person_id="person:bob", + binding=binding, + assertion=assertion, + registry=registry, + verifier=double, + ) + is None + ) + + +def test_assertion_repr_redacts_secrets(passkey, double) -> None: + digest = approval_payload_digest(_payload()) + assertion = double.sign_webauthn(challenge_sha256=digest, registered=passkey) + text = repr(assertion) + assert "redacted" in text + assert assertion.signature.hex() not in text + assert assertion.credential_id.hex() not in text + + +def test_recorded_webauthn_fixture_round_trips(double) -> None: + data = json.loads(FIXTURE.read_text(encoding="utf-8")) + assert "Synthetic" in data["note"] or "NOT a live" in data["note"] + reg = RegisteredPasskey( + person_id=data["registered_passkey"]["person_id"], + credential_id=bytes.fromhex(data["registered_passkey"]["credential_id_hex"]), + public_key=bytes.fromhex(data["registered_passkey"]["public_key_hex"]), + rp_id=data["registered_passkey"]["rp_id"], + ) + assertion = WebAuthnAssertionMaterial( + credential_id=bytes.fromhex(data["assertion_hex"]["credential_id"]), + authenticator_data=bytes.fromhex(data["assertion_hex"]["authenticator_data"]), + client_data_json=data["assertion_hex"]["client_data_json"].encode("utf-8"), + signature=bytes.fromhex(data["assertion_hex"]["signature"]), + ) + binding = NamedPersonBindingV1.model_validate(data["binding"]) + registry = MemoryPersonRegistry(passkeys=[reg]) + assert ( + require_named_person_binding( + profile="regulated", + approval_payload_sha256=data["approval_payload_sha256"], + expected_person_id=reg.person_id, + binding=binding, + assertion=assertion, + registry=registry, + verifier=double, + ) + is None + ) + + +def _request(store: ItemApprovalStore): + return store.open_request( + run_id="run-0001", + workflow_name="referral", + step_id="save", + step_index=3, + step_intent="Click 'Save'", + mode=ApprovalMode.EACH_ITEM, + bundle_content_digest="a" * 64, + runtime_inputs_digest="b" * 64, + frame_png=FRAME, + form_state=("dom", "sha256:" + "1" * 64), + ) + + +def test_answer_refusal_regulated_missing_binding_fail_closed( + tmp_path, monkeypatch +) -> None: + """Regulated profile refuses a person answer with no named-person binding.""" + + from openadapt_flow.approval_mode import item_approval as mod + + store = ItemApprovalStore(tmp_path / "run") + request = _request(store) + answer = ItemApprovalAnswer( + request_id=request.request_id, + request_sha256=request.digest(), + decision="approve", + decided_by="human", + route="desktop_run_it", + approver_kind="unauthenticated", + approver_id=None, + auth_method="none", + answered_at=request.created_at, + attestation=None, + ) + + monkeypatch.setattr(mod, "attestation_refusal", lambda *a, **k: None) + + ctx = ApproverContext( + execution_profile="regulated", + key_dir=tmp_path / "keys", + ) + code = answer_refusal(request, answer, policy=ctx) + assert code == NamedPersonRefusal.MISSING_BINDING.value + + +def test_answer_refusal_regulated_accepts_verified_webauthn( + tmp_path, monkeypatch, passkey, double, registry +) -> None: + from openadapt_flow.approval_mode import item_approval as mod + + store = ItemApprovalStore(tmp_path / "run") + request = _request(store) + digest = request.digest() + assertion = double.sign_webauthn(challenge_sha256=digest, registered=passkey) + binding = build_binding( + person_id=PERSON, + method=BindingMethod.WEBAUTHN_PASSKEY, + approval_payload_sha256=digest, + assertion=assertion, + rp_id=RP_ID, + verified_at="2026-11-05T17:00:00Z", + ) + answer = ItemApprovalAnswer( + request_id=request.request_id, + request_sha256=request.digest(), + decision="approve", + decided_by="human", + route="desktop_run_it", + approver_kind="unauthenticated", + # Leave approver_id unset so stubbed attestation (signer="") does not + # trip answer_fields_inconsistent; the binding still names the person. + approver_id=None, + auth_method="none", + answered_at=request.created_at, + attestation=None, + ) + monkeypatch.setattr(mod, "attestation_refusal", lambda *a, **k: None) + monkeypatch.setattr(mod, "_person_refusal", lambda *a, **k: None) + ctx = ApproverContext( + execution_profile="regulated", + key_dir=tmp_path / "keys", + named_person_binding=binding, + named_person_assertion=assertion, + named_person_registry=registry, + named_person_verifier=double, + ) + assert answer_refusal(request, answer, policy=ctx) is None + + +def test_answer_refusal_demo_skips_named_person(tmp_path, monkeypatch) -> None: + from openadapt_flow.approval_mode import item_approval as mod + + store = ItemApprovalStore(tmp_path / "run") + request = _request(store) + answer = ItemApprovalAnswer( + request_id=request.request_id, + request_sha256=request.digest(), + decision="approve", + decided_by="human", + route="desktop_run_it", + approver_kind="unauthenticated", + approver_id=None, + auth_method="none", + answered_at=request.created_at, + attestation=None, + ) + monkeypatch.setattr(mod, "attestation_refusal", lambda *a, **k: None) + ctx = ApproverContext(execution_profile="demo", key_dir=tmp_path / "keys") + assert answer_refusal(request, answer, policy=ctx) is None