"""Pydantic models for public-key challenge-response authentication. Wire protocol ------------- Step 1 — POST /api/auth/challenge Request: ChallengeRequest {"fingerprint": "", "algorithm": "ed25519"} Response: ChallengeResponse {"challenge_token": "", "is_new_key": bool} ``challenge_token`` is a signed hex nonce (5 min TTL) containing the raw bytes the client must sign with its Ed25519 private key. The client must sign ``bytes.fromhex(challenge_token)`` with its private key. Step 2 — POST /api/auth/verify Request: VerifyRequest {challenge_token, public_key_b64, signature_b64, ?handle, ?label, ?identity_type} Response: VerifyResponse {"handle": "", "identity_id": "...", "auth_method": "ed25519", "key": {...}} ``public_key_b64`` — URL-safe base64 (no padding) of the raw public key bytes. ``signature_b64`` — URL-safe base64 of the signature over the raw nonce bytes. ``handle`` — required when registering a new key (is_new_key=True). ``identity_type`` — "human" (default) or "agent". Agent provisioning — POST /api/identities/agent (operator-authenticated via MSign) Request: AgentRegistrationRequest {handle, public_key_b64, fingerprint, algorithm, ...} Response: AgentRegistrationResponse {handle, identity_id, is_new_identity} The operator proves their identity via MSign; the server creates an agent identity with ``spawned_by`` set to the operator's handle. No challenge- response is required — trust is established at provisioning time. Algorithm identifiers ---------------------- "ed25519" — RFC 8032 / FIPS 186-5; classical, not quantum-safe. "ml-dsa-65" — FIPS 204; NIST post-quantum standard. (Pending implementation.) """ from pydantic import BaseModel, Field, field_validator, model_validator from musehub.crypto.keys import KeyAlgorithm, IMPLEMENTED_ALGORITHMS, PUBLIC_KEY_SIZES # Sorted list of algorithm strings for error messages / OpenAPI docs _IMPLEMENTED_ALGO_VALUES = sorted(a.value for a in IMPLEMENTED_ALGORITHMS) class ChallengeRequest(BaseModel): """Request a challenge nonce for the given public key fingerprint.""" fingerprint: str = Field( ..., min_length=71, max_length=71, pattern=r"^sha256:[0-9a-f]{64}$", description="Canonical sha256:-prefixed fingerprint of the raw public key bytes.", ) algorithm: str = Field( default=KeyAlgorithm.ED25519.value, description=( "Signing algorithm for this key. " f"Implemented: {_IMPLEMENTED_ALGO_VALUES}." ), ) @field_validator("algorithm") @classmethod def _validate_algorithm(cls, v: str) -> str: try: algo = KeyAlgorithm(v) except ValueError: known = [a.value for a in KeyAlgorithm] raise ValueError( f"Unknown algorithm '{v}'. Known: {known}. " f"Implemented: {_IMPLEMENTED_ALGO_VALUES}." ) if algo not in IMPLEMENTED_ALGORITHMS: raise ValueError( f"Algorithm '{v}' is defined but not yet implemented. " f"Currently implemented: {_IMPLEMENTED_ALGO_VALUES}." ) return v class ChallengeResponse(BaseModel): """Challenge nonce to sign and submit to /api/auth/verify.""" challenge_token: str = Field( ..., description=( "64-char hex nonce (32 random bytes from CSPRNG). " "Sign bytes.fromhex(challenge_token) with your private key." ), ) is_new_key: bool = Field( ..., description="True when the fingerprint is not yet registered.", ) expires_in: int = Field(300, description="Challenge validity in seconds.") algorithm: str = Field(..., description="The algorithm the challenge was issued for.") class VerifyRequest(BaseModel): """Submit a signed challenge to complete authentication.""" challenge_token: str = Field(..., description="The nonce hex string from /api/auth/challenge.") public_key_b64: str = Field( ..., min_length=1, description="Canonically-prefixed public key: 'ed25519:'.", ) signature_b64: str = Field( ..., min_length=1, description=( "Signature over raw nonce bytes (bytes.fromhex(challenge_token)), " "canonically prefixed: 'ed25519:'." ), ) handle: str | None = Field( None, min_length=1, max_length=64, description="Desired handle (username). Required when registering a new key.", ) display_name: str | None = Field( None, max_length=128, description="Optional display name shown on the profile.", ) label: str | None = Field( None, max_length=255, description='Optional friendly key label, e.g. "MacBook Pro" or "CI agent".', ) identity_type: str = Field( default="human", description='Identity type: "human" (default) or "agent".', ) @field_validator("handle") @classmethod def _validate_handle(cls, v: str | None) -> str | None: if v is None: return None normalized = v.strip().lower() import re if not re.fullmatch(r"[a-z0-9_-]+", normalized): raise ValueError( "Handle must contain only lowercase letters, digits, underscores, and hyphens." ) return normalized @field_validator("identity_type") @classmethod def _validate_identity_type(cls, v: str) -> str: if v not in ("human", "agent"): raise ValueError('identity_type must be "human" or "agent".') return v class RotateKeyRequest(BaseModel): """Add a new key to an existing identity during key rotation. Requires ``Authorization: MSign`` from an existing key of the same identity (old key proves account ownership). The ``challenge_token`` + ``signature_b64`` pair proves the caller also owns the corresponding new private key. """ challenge_token: str = Field( ..., description="The nonce hex string returned by POST /api/auth/challenge.", ) public_key_b64: str = Field( ..., min_length=1, description="Canonically-prefixed public key: 'ed25519:'.", ) signature_b64: str = Field( ..., min_length=1, description=( "Signature over raw nonce bytes (bytes.fromhex(challenge_token)), " "canonically prefixed: 'ed25519:'." ), ) label: str | None = Field( None, max_length=255, description='Optional friendly key label, e.g. "rotation-v2".', ) class AuthKeyResponse(BaseModel): """Public summary of a registered auth key (never exposes key material).""" key_id: str algorithm: str fingerprint: str label: str created_at: str last_used_at: str | None class VerifyResponse(BaseModel): """Successful key registration / re-registration response. No token is returned — authentication is per-request via MSign. Use the registered key to sign subsequent requests. """ handle: str identity_id: str is_new_identity: bool = Field( ..., description="True when this verify call created a new identity.", ) auth_method: str = Field( ..., description="Algorithm used to authenticate (e.g. 'ed25519', 'ml-dsa-65').", ) key: AuthKeyResponse class AgentRegistrationRequest(BaseModel): """Operator-signed agent provisioning request — ``POST /api/identities/agent``. The caller proves their identity via ``Authorization: MSign …``. No challenge-response is required — the operator's MSign header establishes authority; the server sets ``spawned_by`` to the authenticated handle. ``public_key_b64`` is URL-safe base64 (no padding) of the raw 32-byte Ed25519 public key — identical to the format used by ``/api/auth/verify``. ``fingerprint`` is the canonical ``sha256:``-prefixed fingerprint of those 32 bytes (format: ``sha256:<64-hex>``, 71 chars). Both are derived from the same ephemeral keypair generated at agent spawn time. """ handle: str = Field( ..., min_length=1, max_length=64, description="Desired agent handle, e.g. 'agentception-abc123'.", ) public_key_b64: str = Field( ..., min_length=1, description="URL-safe base64 (no padding) of the raw Ed25519 public key.", ) fingerprint: str = Field( ..., min_length=71, max_length=71, pattern=r"^sha256:[0-9a-f]{64}$", description="Canonical sha256:-prefixed fingerprint of the raw public key bytes.", ) algorithm: str = Field( default=KeyAlgorithm.ED25519, description="Signing algorithm for this key.", ) agent_model: str = Field( default="", max_length=255, description="LLM model this agent will run (e.g. 'claude-sonnet-4-6').", ) scope: list[str] = Field( default_factory=list, description="Permitted operation scopes, e.g. ['push:agentception'].", ) expires_at: str | None = Field( None, description="ISO-8601 UTC expiry time for this agent key. null = no expiry.", ) label: str = Field( default="", max_length=255, description='Key label, e.g. "ephemeral/agentception-abc123".', ) @field_validator("handle") @classmethod def _validate_handle(cls, v: str) -> str: import re normalized = v.strip().lower() if not re.fullmatch(r"[a-z0-9_-]+", normalized): raise ValueError( "Handle must contain only lowercase letters, digits, underscores, and hyphens." ) return normalized @field_validator("algorithm") @classmethod def _validate_algorithm(cls, v: str) -> str: try: algo = KeyAlgorithm(v) except ValueError: known = [a.value for a in KeyAlgorithm] raise ValueError(f"Unknown algorithm '{v}'. Known: {known}.") if algo not in IMPLEMENTED_ALGORITHMS: raise ValueError( f"Algorithm '{v}' is defined but not yet implemented. " f"Currently implemented: {sorted(a.value for a in IMPLEMENTED_ALGORITHMS)}." ) return v class AgentRegistrationResponse(BaseModel): """Response for ``POST /api/identities/agent``.""" handle: str identity_id: str is_new_identity: bool = Field( ..., description="True when a new identity was created.", ) spawned_by: str = Field( ..., description="Handle of the operator who provisioned this agent.", ) key: AuthKeyResponse