Standalone Shonar Desktop: vendor portable sources + local engine; decouple from ~/Projects/Shonar

- 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
This commit is contained in:
avi 2026-09-14 17:14:54 -05:00
commit 76c867fca4
136 changed files with 21099 additions and 0 deletions

View file

@ -0,0 +1,98 @@
"""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)