103 lines
3.2 KiB
Python
103 lines
3.2 KiB
Python
"""Signed, short-lived single-operator sessions for the dashboard boundary."""
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import secrets
|
|
import time
|
|
from dataclasses import dataclass
|
|
|
|
from fastapi import Request
|
|
|
|
SESSION_COOKIE = "stackchain_session"
|
|
CSRF_COOKIE = "stackchain_csrf"
|
|
DEFAULT_TTL_SECONDS = 8 * 60 * 60
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Session:
|
|
csrf: str
|
|
expires_at: int
|
|
|
|
|
|
def access_token() -> str:
|
|
return os.getenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "")
|
|
|
|
|
|
def enabled() -> bool:
|
|
return bool(access_token())
|
|
|
|
|
|
def _secret() -> bytes:
|
|
configured = os.getenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "")
|
|
if configured:
|
|
return configured.encode()
|
|
return hmac.new(
|
|
access_token().encode(), b"stackchain-dashboard-session-signing", hashlib.sha256
|
|
).digest()
|
|
|
|
|
|
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 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(csrf=secrets.token_urlsafe(24), expires_at=issued_at + max(1, ttl))
|
|
payload = json.dumps(
|
|
{"csrf": session.csrf, "exp": session.expires_at},
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
).encode()
|
|
encoded = _encode(payload)
|
|
signature = _encode(hmac.new(_secret(), encoded.encode(), hashlib.sha256).digest())
|
|
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(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.csrf, str) or not session.csrf:
|
|
return None
|
|
return session
|
|
|
|
|
|
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)
|