kadence/backend/app/services/auth_service.py

"""Who is allowed into the browser UI.

Two ways in, both ending in the same thing — a row in `auth_sessions` and a
cookie that names it:

* a password, hashed with stdlib PBKDF2 (no new dependency for one check);
* the Telegram Login Widget, verified with the documented HMAC scheme against
  the bot token already configured for notifications.

The desktop app and scripts are untouched — they keep using `X-API-Token`
(see `app.api.deps`). This module only ever runs for the browser.
"""

from __future__ import annotations

import hashlib
import hmac
import secrets
import time
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import delete
from sqlalchemy.orm import Session

from app.models import AuthSession

PBKDF2_ALGO = "pbkdf2_sha256"
PBKDF2_ITERATIONS = 260_000
SESSION_COOKIE = "kadence_session"
TELEGRAM_AUTH_MAX_AGE = 86400  # a login link older than a day is refused


# --------------------------------------------------------------------------
# Passwords
# --------------------------------------------------------------------------
def hash_password(password: str) -> str:
    salt = secrets.token_hex(16)
    digest = hashlib.pbkdf2_hmac("sha256", password.encode(), bytes.fromhex(salt), PBKDF2_ITERATIONS)
    return f"{PBKDF2_ALGO}${PBKDF2_ITERATIONS}${salt}${digest.hex()}"


def verify_password(password: str, stored_hash: str) -> bool:
    try:
        algo, iterations, salt, expected = stored_hash.split("$")
        if algo != PBKDF2_ALGO:
            return False
        digest = hashlib.pbkdf2_hmac(
            "sha256", password.encode(), bytes.fromhex(salt), int(iterations)
        )
        return hmac.compare_digest(digest.hex(), expected)
    except (ValueError, AttributeError):
        return False


# --------------------------------------------------------------------------
# Telegram Login Widget
# --------------------------------------------------------------------------
def verify_telegram_payload(data: dict[str, Any], bot_token: str) -> bool:
    """https://core.telegram.org/widgets/login#checking-authorization"""
    received_hash = data.get("hash")
    if not received_hash or not bot_token:
        return False
    auth_date = data.get("auth_date")
    try:
        if time.time() - int(auth_date) > TELEGRAM_AUTH_MAX_AGE:
            return False
    except (TypeError, ValueError):
        return False

    check_string = "\n".join(
        f"{key}={data[key]}" for key in sorted(data) if key not in ("hash",)
    )
    secret_key = hashlib.sha256(bot_token.encode()).digest()
    computed = hmac.new(secret_key, check_string.encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(computed, received_hash)


# --------------------------------------------------------------------------
# Sessions
# --------------------------------------------------------------------------
def create_session(session: Session, method: str, session_days: int) -> AuthSession:
    _prune_expired(session)
    row = AuthSession(
        token=secrets.token_urlsafe(32),
        method=method,
        expires_at=datetime.now(UTC) + timedelta(days=session_days),
    )
    session.add(row)
    session.commit()
    return row


def _aware(dt: datetime) -> datetime:
    """SQLite drops the tzinfo it was stored with; Postgres does not."""
    return dt if dt.tzinfo else dt.replace(tzinfo=UTC)


def valid_session(session: Session, token: str | None) -> AuthSession | None:
    if not token:
        return None
    row = session.get(AuthSession, token)
    if row is None:
        return None
    if _aware(row.expires_at) < datetime.now(UTC):
        session.delete(row)
        session.commit()
        return None
    return row


def delete_session(session: Session, token: str | None) -> None:
    if not token:
        return
    session.execute(delete(AuthSession).where(AuthSession.token == token))
    session.commit()


def _prune_expired(session: Session) -> None:
    session.execute(delete(AuthSession).where(AuthSession.expires_at < datetime.now(UTC)))


# --------------------------------------------------------------------------
# Login rate limiting (in-process; one instance, one user)
# --------------------------------------------------------------------------
_MAX_ATTEMPTS = 8
_WINDOW_SECONDS = 300
_failures: dict[str, list[float]] = {}


def rate_limited(key: str) -> bool:
    now = time.monotonic()
    recent = [t for t in _failures.get(key, []) if now - t < _WINDOW_SECONDS]
    _failures[key] = recent
    return len(recent) >= _MAX_ATTEMPTS


def record_failure(key: str) -> None:
    _failures.setdefault(key, []).append(time.monotonic())


def record_success(key: str) -> None:
    _failures.pop(key, None)