stackchain-dashboard/src/dashboard_auth.py

"""Signed, short-lived single-operator sessions for the dashboard boundary."""

import base64
import hashlib
import hmac
import ipaddress
import json
import os
import secrets
import time
from dataclasses import dataclass

from fastapi import Request

from src.session_store import SessionStore, SessionStoreError

SESSION_COOKIE = "stackchain_session"
CSRF_COOKIE = "stackchain_csrf"
DEFAULT_TTL_SECONDS = 8 * 60 * 60
MIN_SECRET_LENGTH = 24
OPERATOR_MODE = "operator"
INSECURE_LOCAL_MODE = "insecure-local"


@dataclass(frozen=True)
class Session:
    session_id: str
    csrf: str
    expires_at: int


def access_token() -> str:
    return os.getenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "")


def mode() -> str:
    return os.getenv("STACKCHAIN_DASHBOARD_AUTH_MODE", OPERATOR_MODE).strip().lower()


def configuration_error() -> str | None:
    configured_mode = mode()
    if configured_mode == INSECURE_LOCAL_MODE:
        return None
    if configured_mode != OPERATOR_MODE:
        return "unsupported authentication mode"
    operator_secret = access_token()
    signing_secret = os.getenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "")
    if len(operator_secret) < MIN_SECRET_LENGTH or len(signing_secret) < MIN_SECRET_LENGTH:
        return "operator mode requires two sufficiently long secrets"
    if hmac.compare_digest(operator_secret, signing_secret):
        return "operator and session secrets must be independent"
    return None


def enabled() -> bool:
    return mode() == OPERATOR_MODE and configuration_error() is None


def is_loopback_request(request: Request) -> bool:
    host = request.client.host if request.client else ""
    try:
        return ipaddress.ip_address(host).is_loopback
    except ValueError:
        return host == "localhost"


def _secret() -> bytes:
    configured = os.getenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "")
    if configuration_error() is not None or mode() != OPERATOR_MODE:
        raise RuntimeError("Dashboard authentication is not configured")
    return configured.encode()


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


def _decode(value: str) -> bytes:
    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))


def _session_store(now: int | None = None) -> SessionStore:
    database = os.getenv(
        "STACKCHAIN_SESSION_DB",
        os.path.join(os.getenv("STACKCHAIN_STATE_DIR", ".stackchain-state"), "sessions.sqlite3"),
    )
    current = time.time if now is None else lambda: now
    return SessionStore(database, clock=current)


def issue_session(now: int | None = None) -> tuple[str, Session]:
    issued_at = int(time.time() if now is None else now)
    ttl = int(os.getenv("STACKCHAIN_DASHBOARD_SESSION_TTL_SECONDS", str(DEFAULT_TTL_SECONDS)))
    session = Session(
        session_id=secrets.token_urlsafe(32),
        csrf=secrets.token_urlsafe(24),
        expires_at=issued_at + max(1, ttl),
    )
    payload = json.dumps(
        {"csrf": session.csrf, "exp": session.expires_at, "sid": session.session_id},
        separators=(",", ":"),
        sort_keys=True,
    ).encode()
    encoded = _encode(payload)
    signature = _encode(hmac.new(_secret(), encoded.encode(), hashlib.sha256).digest())
    _session_store(now).activate(session.session_id, session.expires_at)
    return f"{encoded}.{signature}", session


def verify_session(value: str | None, now: int | None = None) -> Session | None:
    if not value or "." not in value or not enabled():
        return None
    encoded, supplied_signature = value.rsplit(".", 1)
    expected = _encode(hmac.new(_secret(), encoded.encode(), hashlib.sha256).digest())
    if not hmac.compare_digest(supplied_signature, expected):
        return None
    try:
        payload = json.loads(_decode(encoded))
        session = Session(
            session_id=payload["sid"],
            csrf=payload["csrf"],
            expires_at=int(payload["exp"]),
        )
    except (ValueError, TypeError, KeyError, json.JSONDecodeError):
        return None
    current = int(time.time() if now is None else now)
    if (
        session.expires_at <= current
        or not isinstance(session.session_id, str)
        or not session.session_id
        or not isinstance(session.csrf, str)
        or not session.csrf
    ):
        return None
    return session if _session_store(now).is_active(session.session_id, session.expires_at) else None


def revoke_session(session: Session) -> None:
    _session_store().revoke(session.session_id)


def request_session(request: Request) -> Session | None:
    return verify_session(request.cookies.get(SESSION_COOKIE))


def cookie_path(request: Request) -> str:
    root_path = request.scope.get("root_path", "").rstrip("/")
    return root_path or "/"


def application_path(request: Request) -> str:
    path = request.url.path
    root_path = request.scope.get("root_path", "").rstrip("/")
    if root_path and (path == root_path or path.startswith(root_path + "/")):
        return path[len(root_path):] or "/"
    return path


def same_origin(request: Request) -> bool:
    origin = request.headers.get("origin", "")
    expected = f"{request.url.scheme}://{request.url.netloc}"
    return bool(origin) and hmac.compare_digest(origin.rstrip("/"), expected)