distinct_server/security.py

"""Small cryptographic helpers for authenticating Distinct worker agents.

The server gives each paired agent a different random secret.  Requests are
signed over their identity, timestamp, nonce, method, path, and body digest so
that a signature cannot be copied to another endpoint or replayed by another
agent.
"""

from __future__ import annotations

import base64
import hashlib
import hmac
import re
import secrets
from typing import Union

_NONCE_RE = re.compile(r"^[A-Za-z0-9._~-]{16,128}$")


def generate_secret(size: int = 32) -> bytes:
    """Return a cryptographically random secret suitable for HMAC-SHA256."""

    if size < 32:
        raise ValueError("HMAC secrets must contain at least 32 bytes")
    return secrets.token_bytes(size)


def encode_secret(secret: bytes) -> str:
    """Encode a binary credential without padding for one-time display."""

    return base64.urlsafe_b64encode(secret).rstrip(b"=").decode("ascii")


def decode_secret(value: str) -> bytes:
    """Decode a credential returned by :func:`encode_secret`."""

    if not isinstance(value, str) or not value:
        raise ValueError("credential must be non-empty text")
    try:
        padding = "=" * (-len(value) % 4)
        decoded = base64.b64decode(
            value + padding,
            altchars=b"-_",
            validate=True,
        )
    except (ValueError, TypeError) as exc:
        raise ValueError("credential is not valid base64url") from exc
    if len(decoded) < 32:
        raise ValueError("credential is too short")
    return decoded


def normalize_pairing_code(value: str) -> str:
    """Normalize the human-entered pairing code before hashing it."""

    if not isinstance(value, str):
        raise ValueError("pairing code must be text")
    normalized = "".join(character for character in value.upper() if character.isalnum())
    if not 16 <= len(normalized) <= 64:
        raise ValueError("pairing code has an invalid length")
    return normalized


def pairing_code_digest(code: str, pepper: bytes) -> str:
    """Hash a pairing code with a process-local pepper for server storage."""

    normalized = normalize_pairing_code(code)
    return hmac.new(pepper, normalized.encode("ascii"), hashlib.sha256).hexdigest()


def new_pairing_code() -> str:
    """Generate a readable code with 120 bits of entropy."""

    raw = base64.b32encode(secrets.token_bytes(15)).decode("ascii").rstrip("=")
    return "-".join(raw[index : index + 4] for index in range(0, len(raw), 4))


#: How many random bytes an agent's access code carries.
#:
#: 24 bytes is 192 bits. A pairing code has 120 and is fine, because it lasts
#: ten minutes and one use; an access code lives as long as the agent and is
#: read aloud, pasted into chats and left in terminal scrollback, so it is
#: sized for a lifetime of that rather than for ten minutes.
ACCESS_CODE_BYTES = 24


def new_access_code() -> str:
    """The code an agent prints for the people allowed to use it.

    Random, not derived from anything about the agent. A code that could be
    computed from a hostname, a port, a start time or an agent id would be
    guessable by anyone who knew those, and every one of them is public or
    nearly so. `secrets` is the only source used, and nothing about the agent
    goes into it.
    """

    raw = base64.b32encode(secrets.token_bytes(ACCESS_CODE_BYTES)).decode("ascii").rstrip("=")
    return "-".join(raw[index : index + 5] for index in range(0, len(raw), 5))


def normalize_access_code(value: str) -> str:
    """Read a code the way a person retypes one: case and dashes forgiven."""

    if not isinstance(value, str):
        raise ValueError("access code must be text")
    cleaned = "".join(character for character in value.upper() if character.isalnum())
    # The floor is what stops a short guess being cheap to try; the ceiling
    # stops an unbounded string reaching the hash.
    if not 32 <= len(cleaned) <= 128:
        raise ValueError("access code has an invalid length")
    return cleaned


def access_code_digest(code: str, pepper: bytes) -> str:
    """Hash an access code for storage, with the same pepper as pairing.

    The server never keeps the code. It keeps this, compares digests, and can
    therefore be read end to end by somebody worried about what a compromised
    server would leak: the answer is a list of hashes of high-entropy strings,
    which is nothing.
    """

    cleaned = normalize_access_code(code)
    return hmac.new(pepper, cleaned.encode("ascii"), hashlib.sha256).hexdigest()


def validate_nonce(nonce: str) -> str:
    if not isinstance(nonce, str) or not _NONCE_RE.fullmatch(nonce):
        raise ValueError("nonce must be 16-128 URL-safe characters")
    return nonce


def body_digest(body: Union[bytes, bytearray, memoryview, str]) -> str:
    if isinstance(body, str):
        body = body.encode("utf-8")
    if not isinstance(body, (bytes, bytearray, memoryview)):
        raise TypeError("body must be bytes or text")
    return hashlib.sha256(bytes(body)).hexdigest()


def canonical_request(
    agent_id: str,
    timestamp: Union[int, float],
    nonce: str,
    method: str,
    path: str,
    body: Union[bytes, bytearray, memoryview, str] = b"",
) -> bytes:
    """Return the documented byte representation signed by worker requests."""

    for label, value in (("agent_id", agent_id), ("method", method), ("path", path)):
        if not isinstance(value, str) or not value or "\n" in value or "\r" in value:
            raise ValueError(f"{label} must be non-empty single-line text")
    validate_nonce(nonce)
    if isinstance(timestamp, bool) or not isinstance(timestamp, (int, float)):
        raise ValueError("timestamp must be numeric")
    seconds = int(timestamp)
    fields = (
        agent_id,
        str(seconds),
        nonce,
        method.upper(),
        path,
        body_digest(body),
    )
    return "\n".join(fields).encode("utf-8")


def sign_request(
    secret: Union[bytes, str],
    agent_id: str,
    timestamp: Union[int, float],
    nonce: str,
    method: str,
    path: str,
    body: Union[bytes, bytearray, memoryview, str] = b"",
) -> str:
    """Return a lowercase hexadecimal HMAC-SHA256 request signature."""

    if isinstance(secret, str):
        secret = decode_secret(secret)
    if not isinstance(secret, bytes):
        raise TypeError("secret must be bytes or an encoded credential")
    message = canonical_request(agent_id, timestamp, nonce, method, path, body)
    return hmac.new(secret, message, hashlib.sha256).hexdigest()



#: Domain separator for the per-agent pseudonymous user key. Deliberately
#: byte-identical to ``distinct_agent.guard.USER_KEY_DOMAIN``: the server
#: computes the key (it holds the real identity), the agent only consumes it,
#: and ``tests/test_signing_conformance.py`` pins the two together the same
#: way it pins the request-signing duplicates.
USER_KEY_DOMAIN = b"distinct/user-key/1"


def derive_user_key(agent_secret: bytes, subject: str) -> str:
    """The pseudonymous, per-agent key for an authenticated user.

    Keyed by the agent's own secret, so two agents cannot compare keys to
    correlate a user across machines, and the agent itself can be consistent
    about a person it cannot identify.
    """

    if not isinstance(agent_secret, (bytes, bytearray)) or len(agent_secret) < 32:
        raise ValueError("agent secret must contain at least 32 bytes")
    if not isinstance(subject, str) or not subject.strip():
        raise ValueError("subject must be non-empty text")
    return hmac.new(
        bytes(agent_secret),
        USER_KEY_DOMAIN + subject.strip().encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()