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 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_prefix": code[:5],
            "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."""
    # Rate-limit before the lookup so a guessing attacker cannot use response
    # differences to probe for valid codes for free.
    if not store.allow("code_redeem_fail"):
        raise errors.RateLimited("too many redemption attempts")

    digest = _hash_code(normalise(code))
    row = store.get_code(digest)
    if row is None:
        store.record(
            "view_code.redeem_refused", actor="unknown", detail={"reason": "unrecognised"}
        )
        raise errors.CodeInvalid("that code is not recognised")

    agent = row["agent"]

    if time.time() >= row["redeem_expires_at"]:
        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")

    # 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):
        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.
    return sessions.issue_view_session(
        store,
        agent,
        scopes=_json.loads(row["scopes"]),
        ttl=row["view_ttl"],
        granted_by=agent,
    )


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)