sigil/sigil/viewcodes.py

"""One-time codes: a human looking at the new internet through their agent.

The flow a human actually uses:

    human: "agent, what's happening with the Gitea PR?"
    agent: hands back a URL and a code
    human: opens the URL, types the code, looks around

Two lifetimes, deliberately not the same number:

    code redemption window   short (5 min)  - the code is a key
    session lifetime         long  (1 hour)  - the session is the door

The code travels in a chat log, so it is made worthless quickly. The hour is
comfortable for reading, and buys no extra capability: a human_via_agent
session holds view scopes only, whatever its lifetime.
"""

from __future__ import annotations

import hashlib
import secrets
import time
from typing import List, Optional

from . import errors, sessions
from .store import Store

DEFAULT_CODE_TTL = 300  # 5 minutes to redeem
DEFAULT_VIEW_TTL = 3600  # 1 hour of access

# Unambiguous alphabet: no 0/O or 1/I. Codes get read aloud and retyped.
_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"


def _format_code() -> str:
    raw = "".join(secrets.choice(_ALPHABET) for _ in range(8))
    return f"{raw[:4]}-{raw[4:]}"


def _hash_code(code: str) -> str:
    return hashlib.sha256(normalise(code).encode("utf-8")).hexdigest()


def _code_ref(code: str) -> str:
    """How a code is referred to in the audit log.

    Never the code, and never a prefix of it. Codes carry 40 bits; the first
    four characters are 20 of those bits, so logging a prefix drops the search
    space to about a million candidates, which is nothing to a reader holding
    the log. A truncated digest identifies the code for correlation without
    carrying any of the secret.
    """
    return "code:" + _hash_code(code)[:12]


def normalise(code: str) -> str:
    """Canonical form: upper case, letters and digits only.

    Humans retype these codes, so the dash and any spacing they drop or add
    must not matter. Both issuing and redeeming hash this same form.
    """
    return "".join(ch for ch in code.upper() if ch.isalnum())


def issue_code(
    store: Store,
    session_token: str,
    scopes: Optional[List[str]] = None,
    code_ttl: int = DEFAULT_CODE_TTL,
    view_ttl: int = DEFAULT_VIEW_TTL,
    limit: Optional[int] = None,
    window: Optional[int] = None,
) -> dict:
    """Mint a one-time view code for the agent that owns `session_token`.

    The caller must present a live agent session. There is deliberately no
    `agent` argument: the agent is derived from the session, so a code can
    never be minted for some other agent, and a route that forgets to
    authenticate fails closed instead of minting a session for whoever asked.

    The code lives code_ttl seconds; the session it mints lives view_ttl
    seconds from the moment it is redeemed.

    The agent is recorded as the grantor so the audit trail shows a human was
    let in by that agent, not that the agent itself was present.
    """
    principal = sessions.resolve(store, session_token)
    agent = sessions.assert_may_act_as_agent(principal)

    if not store.allow(f"code_issue:{agent}", limit=limit, window=window):
        raise errors.RateLimited(f"too many view codes for {agent}")

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

    granted = list(scopes if scopes is not None else sessions.HUMAN_VIEW_SCOPES)
    # Refuse to bake write or act scopes into a human code at issue time.
    illegal = [s for s in granted if s not in sessions.HUMAN_VIEW_SCOPES]
    if illegal:
        raise errors.ScopeDenied(f"human view codes cannot carry: {illegal}")

    if code_ttl >= view_ttl:
        raise errors.SigilError(
            "code_ttl must be shorter than view_ttl: an unspent code must not "
            "outlive the window it can open"
        )

    code = _format_code()
    now = int(time.time())
    store.put_code(
        _hash_code(code),
        agent,
        granted,
        redeem_expires_at=now + code_ttl,
        view_ttl=view_ttl,
    )
    store.record(
        "view_code.issued",
        actor=agent,
        subject=f"{agent}'s human",
        fingerprint=agent_row["fingerprint"],
        detail={
            "code_ref": _code_ref(code),
            "scopes": granted,
            "redeem_expires_at": now + code_ttl,
        },
    )
    return {
        "code": code,
        "agent": agent,
        "scopes": granted,
        "code_ttl": code_ttl,
        "view_ttl": view_ttl,
        "redeem_expires_at": now + code_ttl,
        "view_ttl": view_ttl,
    }


def redeem_code(store: Store, code: str) -> dict:
    """Trade a one-time code for a human_via_agent session."""
    digest = _hash_code(normalise(code))
    row = store.get_code(digest)
    if row is None:
        # Unknown guesses share one anonymous fail bucket so probing still
        # saturates — but that bucket is not the same as any agent's success
        # path, so ten bad guesses cannot lock out a valid code for someone else.
        if not store.allow("code_redeem_fail:unknown"):
            raise errors.RateLimited("too many redemption attempts")
        store.record(
            "view_code.redeem_refused", actor="unknown", detail={"reason": "unrecognised"}
        )
        raise errors.CodeInvalid("that code is not recognised")

    agent = row["agent"]

    # Defense in depth: revoke/rekey already invalidate unused codes, but refuse
    # here too if the agent is currently revoked or missing so a missed
    # invalidation cannot mint a human_via_agent session.
    from . import identity as _identity
    agent_row = store.get_agent(agent)
    if agent_row is None or _identity.is_revoked(agent_row):
        if not store.allow(f"code_redeem_fail:{agent}"):
            raise errors.RateLimited("too many redemption attempts")
        store.record(
            "view_code.redeem_refused",
            actor=agent,
            subject=f"{agent}'s human",
            detail={"reason": "agent revoked or unknown"},
        )
        raise errors.CodeInvalid("that code is not redeemable")

    # Known-agent attempts use the success quota (code_redeem), not the fail
    # bucket. Failures for this agent (expired / already used) charge a
    # per-agent fail bucket so one noisy human cannot lock every agent out.
    if time.time() >= row["redeem_expires_at"]:
        if not store.allow(f"code_redeem_fail:{agent}"):
            raise errors.RateLimited("too many redemption attempts")
        store.record(
            "view_code.redeem_refused",
            actor=agent,
            subject=f"{agent}'s human",
            detail={"reason": "redemption window closed"},
        )
        raise errors.CodeExpired("that code has expired")

    if not store.allow(f"code_redeem:{agent}"):
        raise errors.RateLimited("too many redemption attempts")

    # Claim the code before issuing anything. Compare-and-set, so a code cannot
    # mint two sessions under a race, and an expired or replayed code is refused
    # before any session exists.
    if not store.claim_code(digest):
        # Already-used is a fail for this agent; refund the success slot by
        # charging the fail bucket instead so retries do not starve real redeems.
        store.allow(f"code_redeem_fail:{agent}")
        store.record(
            "view_code.redeem_refused",
            actor=agent,
            subject=f"{agent}'s human",
            detail={"reason": "already used or claimed"},
        )
        raise errors.CodeUsed("that code has already been used")

    store.record(
        "view_code.redeemed", actor=agent, subject=f"{agent}'s human", detail={}
    )

    import json as _json

    # The session window starts here, at redemption, not at issue. Waiting for a
    # code must not eat into the hour the human was given.
    grant = sessions.GrantedCode(
        agent,
        _json.loads(row["scopes"]),
        row["view_ttl"],
        sessions._ISSUER,
    )
    return sessions.issue_view_session(store, grant)


def sweep(store: Store) -> int:
    """Report how many codes and sessions have lapsed. Housekeeping aid."""
    now = int(time.time())
    codes = [r for r in store.lapsed_codes()
             if now >= r["redeem_expires_at"]]
    tokens = [r for r in store.sessions() if r["revoked_at"] is None
              and now >= r["expires_at"]]
    return len(codes) + len(tokens)