File size: 2,326 Bytes
b30f068 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | """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
|