sigil/sigil/identity.py

"""Enrolment and challenge-response authentication.

The flow replaces a static passcode with a signature:

  1. seed_bootstrap   platform stores a one-time secret for an agent
  2. enrol            agent presents that secret once, plus its public key
  3. challenge        platform returns a random nonce
  4. verify           agent signs the nonce; platform checks against the
                      enrolled public key and issues a short session

After step 2 the secret is gone. It is a bootstrap, not a credential.
"""

from __future__ import annotations

import hashlib
import re
import secrets
import time
from typing import Optional

from . import errors, keys, sessions
from .store import Store

DEFAULT_NONCE_TTL = 120  # 2 minutes. Short: a nonce is a one-shot question.


_AGENT_ID = re.compile(r"^[a-z0-9][a-z0-9_.-]{0,63}$")


def _validate_agent_id(agent: str) -> None:
    """Agent ids appear in log lines, URLs and filenames.

    Constrain them to a safe alphabet and length rather than trusting callers,
    so an id can never carry a newline into the audit trail or a path segment
    into a file.
    """
    if not _AGENT_ID.match(agent or ""):
        raise errors.SigilError(
            f"invalid agent id {agent!r}: expected lowercase [a-z0-9._-], max 64 chars"
        )


def _hash_secret(secret: str) -> str:
    return hashlib.sha256(secret.encode("utf-8")).hexdigest()


def seed_bootstrap(store: Store, agent: str, secret: Optional[str] = None) -> str:
    """Register a one-time enrolment secret for an agent.

    This is the only moment a shared secret exists. In production this replaces
    the agent's static passcode, issued once and never transmitted again.
    """
    _validate_agent_id(agent)
    secret = secret or secrets.token_urlsafe(24)
    store.put_bootstrap(_hash_secret(secret), agent)
    store.record(
        "identity.bootstrap_seeded", actor="system", subject=agent, detail={}
    )
    return secret


def enrol(
    store: Store, agent: str, public_b64: str, secret: str
) -> str:
    """Bind a public key to an agent, once, consuming the bootstrap secret.

    Returns the key fingerprint, which is an identifier and safe to log.
    """
    _validate_agent_id(agent)
    if not store.allow(f"enrol:{agent}"):
        raise errors.RateLimited(f"too many enrolment attempts for {agent}")
    if store.get_agent(agent) is not None:
        raise errors.EnrolmentFailed(f"{agent} is already enrolled")
    owner = store.spend_bootstrap(_hash_secret(secret))
    if owner != agent:
        raise errors.EnrolmentFailed("bootstrap secret invalid or already spent")

    public = keys.import_public_b64(public_b64)
    fingerprint = keys.fingerprint(public)
    store.put_agent(agent, keys.public_bytes(public), fingerprint)
    store.record(
        "identity.enrolled",
        actor=agent,
        subject=agent,
        fingerprint=fingerprint,
        detail={"algorithm": "ed25519"},
    )
    return fingerprint


def challenge(store: Store, agent: str, ttl: int = DEFAULT_NONCE_TTL,
              limit: Optional[int] = None, window: Optional[int] = None) -> dict:
    """Issue a single-use nonce for an agent to sign."""
    if not store.allow(f"challenge:{agent}", limit=limit, window=window):
        raise errors.RateLimited(f"too many challenges for {agent}")

    row = store.get_agent(agent)
    if row is None:
        raise errors.UnknownAgent(f"{agent} is not enrolled")
    if row["revoked_at"] is not None:
        raise errors.UnknownAgent(f"{agent} is revoked")
    nonce = secrets.token_urlsafe(24)
    expires = int(time.time()) + ttl
    store.put_nonce(nonce, agent, expires)
    store.record(
        "identity.challenge_issued", actor=agent, subject=agent,
        fingerprint=row["fingerprint"], detail={"expires_at": expires},
    )
    return {
        "nonce": nonce,
        "agent": agent,
        "expires_at": expires,
        "ttl": ttl,
        "fingerprint": row["fingerprint"],
    }


def verify(
    store: Store,
    nonce: str,
    signature_b64: str,
    ttl: int = sessions.DEFAULT_AGENT_TTL,
    limit: Optional[int] = None,
    window: Optional[int] = None,
) -> dict:
    """Check a signed nonce and, if it holds, mint an agent session.

    Replay is refused: a nonce is consumed on first use whether or not the
    signature was any good, so a captured signature cannot be retried.
    """
    import base64

    row = store.get_nonce(nonce)
    if row is None:
        raise errors.NonceUnknown("no such challenge")

    if not store.allow(f"verify:{row['agent']}", limit=limit, window=window):
        raise errors.RateLimited("too many verification attempts")

    agent = row["agent"]

    # Claim the nonce before judging anything. Compare-and-set, so two
    # concurrent verifies cannot both proceed, and a wrong signature still
    # burns the nonce rather than leaving it open for another attempt.
    if not store.claim_nonce(nonce):
        store.record(
            "auth.replay_refused", actor=agent, subject=agent,
            detail={"reason": "nonce already used or claimed"},
        )
        raise errors.NonceUsed("challenge already answered")

    if time.time() >= row["expires_at"]:
        raise errors.NonceExpired("challenge expired")

    agent_row = store.get_agent(agent)
    if agent_row is None:
        raise errors.UnknownAgent(f"{agent} is not enrolled")
    if agent_row["revoked_at"] is not None:
        raise errors.UnknownAgent(f"{agent} is revoked")

    public = keys.from_public_bytes(agent_row["pubkey"])
    try:
        signature = base64.b64decode(signature_b64.encode("ascii"))
    except Exception:
        raise errors.BadSignature("signature is not valid base64")

    if not keys.verify(public, signature, nonce.encode("utf-8")):
        store.record(
            "auth.signature_refused", actor=agent, subject=agent,
            fingerprint=agent_row["fingerprint"], detail={},
        )
        raise errors.BadSignature("signature does not verify")

    store.record(
        "auth.verified",
        actor=agent,
        subject=agent,
        fingerprint=agent_row["fingerprint"],
        detail={"method": "ed25519-challenge-response"},
    )
    return sessions.issue_agent_session(
        store, agent, ttl=ttl, fingerprint=agent_row["fingerprint"]
    )


def enrol_rekey(store: Store, agent: str, public_b64: str, secret: str) -> str:
    """Re-bind a public key using a fresh one-time secret.

    Rotation cannot use the agent's existing key to authorise itself: if that
    key leaked, it would authorise its own replacement.
    """
    row = store.get_agent(agent)
    if row is None:
        raise errors.UnknownAgent(f"{agent} is not enrolled")
    owner = store.spend_bootstrap(_hash_secret(secret))
    if owner != agent:
        raise errors.EnrolmentFailed("bootstrap secret invalid or already spent")
    public = keys.import_public_b64(public_b64)
    fingerprint = keys.fingerprint(public)
    previous = row["fingerprint"]
    store.put_agent(agent, keys.public_bytes(public), fingerprint)
    # A new key must not leave old sessions alive, or rotating after a
    # compromise buys nothing.
    killed = store.revoke_sessions_for_agent(agent, "key rotated")
    store.record(
        "identity.rekeyed",
        actor=agent,
        subject=agent,
        fingerprint=fingerprint,
        detail={"previous_fingerprint": previous, "sessions_revoked": killed},
    )
    return fingerprint