76c867fca4
- shared/ = portable Android-origin sources vendored from deferred/desktop-server (app/build.gradle.kts srcDir repointed; PlaybackController.kt excluded as Android-only) - backend/ = bundled-lite engine (SQLite + inline queue); .venv symlinked from the old checkout, PYTHONPATH pins THIS backend's code over any editable install - repoRoot() resolves this project dir (env SHONAR_REPO still wins); desktop-dev.sh watches shared/ + backend/ - Verified: :app:compileKotlin + :app:test green (23 tests); engine boots on :8010, self-migrates, /healthz ok
98 lines
2.8 KiB
Python
98 lines
2.8 KiB
Python
"""Security primitives: password hashing, access/refresh tokens, rate limit
|
|
keying. Secrets come exclusively from settings/env."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import secrets
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
import jwt
|
|
from argon2 import PasswordHasher
|
|
from argon2.exceptions import InvalidHashError, VerificationError, VerifyMismatchError
|
|
|
|
from shonar.core.config import get_settings
|
|
|
|
_ph = PasswordHasher()
|
|
|
|
ACCESS_TOKEN_TYPE = "access"
|
|
REFRESH_TOKEN_TYPE = "refresh"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Passwords
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return _ph.hash(password)
|
|
|
|
|
|
def verify_password(password_hash: str, password: str) -> bool:
|
|
try:
|
|
return _ph.verify(password_hash, password)
|
|
except (VerifyMismatchError, VerificationError, InvalidHashError):
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Access tokens (JWT)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def create_access_token(user_id: uuid.UUID, device_id: uuid.UUID | None = None) -> tuple[str, int]:
|
|
"""Returns (token, ttl_seconds)."""
|
|
settings = get_settings()
|
|
ttl = settings.access_token_ttl_minutes * 60
|
|
now = datetime.now(UTC)
|
|
payload: dict[str, Any] = {
|
|
"sub": str(user_id),
|
|
"typ": ACCESS_TOKEN_TYPE,
|
|
"iat": now,
|
|
"exp": now + timedelta(seconds=ttl),
|
|
"jti": uuid.uuid4().hex,
|
|
}
|
|
if device_id is not None:
|
|
payload["dev"] = str(device_id)
|
|
token = jwt.encode(payload, settings.secret_key, algorithm="HS256")
|
|
return token, ttl
|
|
|
|
|
|
class TokenError(Exception):
|
|
"""Invalid or expired token."""
|
|
|
|
|
|
def decode_access_token(token: str) -> dict[str, Any]:
|
|
settings = get_settings()
|
|
try:
|
|
payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"])
|
|
except jwt.PyJWTError as exc:
|
|
raise TokenError("invalid access token") from exc
|
|
if payload.get("typ") != ACCESS_TOKEN_TYPE:
|
|
raise TokenError("wrong token type")
|
|
return payload
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Refresh tokens (opaque, rotated, hashed at rest)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def generate_refresh_token() -> str:
|
|
"""Opaque high-entropy token; only its SHA-256 hash is ever stored."""
|
|
return secrets.token_urlsafe(48)
|
|
|
|
|
|
def hash_refresh_token(token: str) -> str:
|
|
return hashlib.sha256(token.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def refresh_token_ttl() -> timedelta:
|
|
return timedelta(days=get_settings().refresh_token_ttl_days)
|
|
|
|
|
|
def constant_time_equals(a: str, b: str) -> bool:
|
|
return hmac.compare_digest(a, b)
|