src/accounts/session.py

"""HMAC-signed session tokens for cookie-based "stay logged in".

Token format:  ``<user_id>.<expiry_epoch>.<sig>``  where
``sig = base64url(HMAC-SHA256(secret, "<user_id>.<expiry_epoch>"))``.

The signing secret comes from the ``SESSION_SECRET`` env var. Set it as a Space
secret so tokens survive restarts; if it is missing we generate an ephemeral one
and warn — the app still works, but existing cookies are invalidated on restart.
"""

from __future__ import annotations

import base64
import hashlib
import hmac
import logging
import os
import secrets
import time
from typing import Optional

logger = logging.getLogger("agadvisor.accounts.session")

DEFAULT_TTL_SECONDS = 30 * 24 * 3600  # 30 days


def _secret() -> bytes:
    s = os.getenv("SESSION_SECRET")
    if s:
        return s.encode("utf-8")
    # Cache an ephemeral secret on the module so it is stable within a process.
    global _EPHEMERAL
    try:
        return _EPHEMERAL  # type: ignore[name-defined]
    except NameError:
        pass
    _EPHEMERAL = secrets.token_bytes(32)  # noqa: F841
    logger.warning(
        "SESSION_SECRET not set — using an ephemeral secret. Sessions will not "
        "survive a restart. Set SESSION_SECRET as a Space secret for persistence."
    )
    return _EPHEMERAL


def _b64(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")


def _sign(payload: str) -> str:
    mac = hmac.new(_secret(), payload.encode("utf-8"), hashlib.sha256).digest()
    return _b64(mac)


def issue_token(user_id: int, ttl_seconds: int = DEFAULT_TTL_SECONDS) -> str:
    expiry = int(time.time()) + int(ttl_seconds)
    payload = f"{int(user_id)}.{expiry}"
    return f"{payload}.{_sign(payload)}"


def validate_token(token: str) -> Optional[int]:
    """Return the user_id if the token is well-formed, correctly signed and not
    expired; otherwise None. Signature check is constant-time."""
    if not token or token.count(".") != 2:
        return None
    user_part, expiry_part, sig = token.split(".")
    payload = f"{user_part}.{expiry_part}"
    if not hmac.compare_digest(sig, _sign(payload)):
        return None
    try:
        if int(expiry_part) < int(time.time()):
            return None
        return int(user_part)
    except (ValueError, TypeError):
        return None