270 lines
8.0 KiB
Python
270 lines
8.0 KiB
Python
"""Signed, short-lived single-operator sessions for the dashboard boundary."""
|
|
|
|
import asyncio
|
|
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
|
|
DEFAULT_IDLE_TIMEOUT_SECONDS = 15 * 60
|
|
STEP_UP_TTL_SECONDS = 90
|
|
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
|
|
idle_expires_at: int | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SessionVerification:
|
|
session: Session | None
|
|
reason: str | None = None
|
|
|
|
|
|
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 idle_timeout_seconds() -> int:
|
|
return max(
|
|
1,
|
|
int(
|
|
os.getenv(
|
|
"STACKCHAIN_DASHBOARD_IDLE_TIMEOUT_SECONDS",
|
|
str(DEFAULT_IDLE_TIMEOUT_SECONDS),
|
|
)
|
|
),
|
|
)
|
|
|
|
|
|
def issue_session(
|
|
now: int | None = None,
|
|
*,
|
|
device_label: str = "This device",
|
|
management_id: str | 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,
|
|
device_label=device_label,
|
|
management_id=management_id,
|
|
)
|
|
return f"{encoded}.{signature}", session
|
|
|
|
|
|
def verify_session_with_reason(
|
|
value: str | None, now: int | None = None
|
|
) -> SessionVerification:
|
|
if not value or "." not in value or not enabled():
|
|
return SessionVerification(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 SessionVerification(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 SessionVerification(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 SessionVerification(None)
|
|
status = _session_store(now).status(
|
|
session.session_id,
|
|
session.expires_at,
|
|
idle_timeout_seconds=idle_timeout_seconds(),
|
|
)
|
|
if status == "idle":
|
|
return SessionVerification(None, "session_idle")
|
|
if status != "active":
|
|
return SessionVerification(None, "session_revoked")
|
|
return SessionVerification(
|
|
Session(
|
|
session_id=session.session_id,
|
|
csrf=session.csrf,
|
|
expires_at=session.expires_at,
|
|
idle_expires_at=getattr(status, "idle_expires_at", None),
|
|
)
|
|
)
|
|
|
|
|
|
def verify_session(value: str | None, now: int | None = None) -> Session | None:
|
|
return verify_session_with_reason(value, now).session
|
|
|
|
|
|
def revoke_session(session: Session) -> None:
|
|
_session_store().revoke(session.session_id)
|
|
|
|
|
|
async def revoke_all_sessions() -> None:
|
|
await asyncio.to_thread(_session_store().revoke_all)
|
|
|
|
|
|
async def active_devices(session: Session):
|
|
return await asyncio.to_thread(_session_store().list_active, session.session_id)
|
|
|
|
|
|
async def session_management_id(session: Session) -> str:
|
|
return await asyncio.to_thread(_session_store().management_id, session.session_id)
|
|
|
|
|
|
async def touch_session(session: Session) -> bool:
|
|
return await asyncio.to_thread(
|
|
_session_store().touch,
|
|
session.session_id,
|
|
session.expires_at,
|
|
idle_timeout_seconds=idle_timeout_seconds(),
|
|
)
|
|
|
|
|
|
async def revoke_managed_session(management_id: str) -> bool:
|
|
return await asyncio.to_thread(_session_store().revoke_managed, management_id)
|
|
|
|
|
|
async def issue_step_up(session: Session, *, action: str, target: str) -> str:
|
|
return await asyncio.to_thread(
|
|
_session_store().mint_step_up,
|
|
session.session_id,
|
|
action=action,
|
|
target=target,
|
|
ttl_seconds=STEP_UP_TTL_SECONDS,
|
|
)
|
|
|
|
|
|
async def consume_step_up(
|
|
grant: str, session: Session, *, action: str, target: str
|
|
) -> bool:
|
|
return await asyncio.to_thread(
|
|
_session_store().consume_step_up,
|
|
grant,
|
|
session.session_id,
|
|
action=action,
|
|
target=target,
|
|
)
|
|
|
|
|
|
async def request_session(request: Request) -> Session | None:
|
|
return await asyncio.to_thread(verify_session, request.cookies.get(SESSION_COOKIE))
|
|
|
|
|
|
async def request_session_verification(request: Request) -> SessionVerification:
|
|
return await asyncio.to_thread(
|
|
verify_session_with_reason, 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)
|