finbot/auth.py

"""Signed, expiring tokens for the web dashboard.

The Space is public, so the dashboard cannot simply trust a user id in a query
string -- anyone could type someone else's. Instead the bot issues a token in
Telegram (where identity is already established) and the page verifies it.

Signed with ``TELEGRAM_WEBHOOK_SECRET``, so there is no additional secret to
manage or leak. Tokens are short-lived because a URL leaks easily: browser
history, referrer headers, a screenshot in a chat.

Nothing here can mutate the ledger. A stolen token, within its window, exposes
one person's spending summary and coach chat -- bad, but bounded, and the
window is minutes.
"""

from __future__ import annotations

import base64
import hashlib
import hmac
import time
from typing import NamedTuple, Optional

#: Long enough to open the link and read it, short enough that a leaked URL in
#: browser history is stale by the time anyone finds it.
DEFAULT_TTL_SECONDS = 30 * 60

_SIG_BYTES = 16


class TokenError(ValueError):
    """The token was malformed, tampered with, or expired."""


class Claims(NamedTuple):
    user_id: int
    expires_at: int

    @property
    def seconds_left(self) -> int:
        return max(0, self.expires_at - int(time.time()))


def _b64encode(raw: bytes) -> str:
    return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")


def _b64decode(text: str) -> bytes:
    padding = "=" * (-len(text) % 4)
    return base64.urlsafe_b64decode(text + padding)


def _sign(payload: bytes, secret: str) -> bytes:
    return hmac.new(secret.encode("utf-8"), payload, hashlib.sha256).digest()[
        :_SIG_BYTES
    ]


def issue(
    user_id: int,
    secret: str,
    ttl_seconds: int = DEFAULT_TTL_SECONDS,
    now: Optional[int] = None,
) -> str:
    """Mint a dashboard token for ``user_id``."""
    if not secret:
        raise TokenError("cannot issue tokens without a signing secret")

    now = now if now is not None else int(time.time())
    payload = f"{int(user_id)}.{now + int(ttl_seconds)}".encode("utf-8")
    return f"{_b64encode(payload)}.{_b64encode(_sign(payload, secret))}"


def verify(token: str, secret: str, now: Optional[int] = None) -> Claims:
    """Validate a token and return its claims, or raise :class:`TokenError`."""
    if not secret:
        raise TokenError("no signing secret configured")
    if not token or "." not in token:
        raise TokenError("malformed token")

    encoded_payload, _, encoded_sig = token.rpartition(".")
    try:
        payload = _b64decode(encoded_payload)
        signature = _b64decode(encoded_sig)
    except Exception as exc:
        raise TokenError("malformed token") from exc

    # Constant-time: a timing-variable comparison would leak the signature
    # byte by byte to anyone willing to measure.
    if not hmac.compare_digest(signature, _sign(payload, secret)):
        raise TokenError("bad signature")

    try:
        user_part, _, expiry_part = payload.decode("utf-8").partition(".")
        claims = Claims(user_id=int(user_part), expires_at=int(expiry_part))
    except Exception as exc:
        raise TokenError("malformed payload") from exc

    now = now if now is not None else int(time.time())
    if claims.expires_at < now:
        raise TokenError("token expired")

    return claims


def dashboard_url(base_url: str, token: str) -> str:
    base = (base_url or "").rstrip("/")
    return f"{base}/?t={token}"