radadmin/backend/app/auth.py

"""Password hashing, opaque login sessions, API keys, and the auth
dependencies used to protect the API.

There is no shared secret in the environment anymore. Requests authenticate as
a specific admin-portal user, either with a login session token
(``Authorization: Bearer <token>``, issued by ``POST /auth/login``) or with a
programmatic API key (``X-API-Key: <key>``, created by an admin). Both resolve
to a row in ``radadmin_admins``; an API key acts as the user who created it.
"""

import hashlib
import secrets
from datetime import datetime, timedelta

from fastapi import Depends, Request, Security
from fastapi.security import APIKeyHeader, HTTPBearer
from fastapi.security.http import HTTPAuthorizationCredentials
from sqlalchemy import select
from sqlalchemy.orm import Session as DbSession

from .database import get_db
from .errors import APIError
from .models import Admin, ApiKey, LogEntry, Session

# Sessions live this long before a fresh login is required.
SESSION_TTL = timedelta(days=7)

_PBKDF2_ITERATIONS = 240_000

bearer_scheme = HTTPBearer(auto_error=False)
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)


# ---------------------------------------------------------------------------
# Password + key hashing (PBKDF2-HMAC-SHA256, stdlib only — no native deps)
# ---------------------------------------------------------------------------
def hash_password(password: str) -> str:
    """Return ``pbkdf2_sha256$<iterations>$<salt_hex>$<hash_hex>``."""
    salt = secrets.token_bytes(16)
    dk = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, _PBKDF2_ITERATIONS)
    return f"pbkdf2_sha256${_PBKDF2_ITERATIONS}${salt.hex()}${dk.hex()}"


def verify_password(password: str, stored: str) -> bool:
    try:
        algo, iterations, salt_hex, hash_hex = stored.split("$")
        if algo != "pbkdf2_sha256":
            return False
        dk = hashlib.pbkdf2_hmac(
            "sha256", password.encode(), bytes.fromhex(salt_hex), int(iterations)
        )
    except (ValueError, TypeError):
        return False
    return secrets.compare_digest(dk.hex(), hash_hex)


# API keys reuse the same PBKDF2 scheme; the raw key is only shown once.
hash_api_key = hash_password
verify_api_key = verify_password


def generate_token() -> str:
    """A 64-char hex session token (fits ``radadmin_sessions.token CHAR(64)``)."""
    return secrets.token_hex(32)


def generate_api_key() -> str:
    """A user-facing API key, e.g. ``rak_<40 hex chars>``."""
    return "rak_" + secrets.token_hex(20)


# ---------------------------------------------------------------------------
# Sessions
# ---------------------------------------------------------------------------
def create_session(db: DbSession, admin: Admin) -> str:
    token = generate_token()
    db.add(Session(token=token, admin_id=admin.id, expires_at=datetime.utcnow() + SESSION_TTL))
    db.commit()
    return token


def _admin_from_bearer(db: DbSession, token: str) -> Admin | None:
    session = db.get(Session, token)
    if session is None:
        return None
    if session.expires_at < datetime.utcnow():
        db.delete(session)
        db.commit()
        return None
    return db.get(Admin, session.admin_id)


# A synthetic principal for API keys whose creator no longer exists. It can act
# on resources but is not a real DB user, so it can't manage users/keys/logs.
_SYSTEM_ADMIN = Admin(id=0, username="api-key", is_admin=False, must_change_password=False)


def _admin_from_api_key(db: DbSession, raw_key: str) -> Admin | None:
    prefix = raw_key[:8]
    rows = (
        db.execute(select(ApiKey).where(ApiKey.key_prefix == prefix, ApiKey.revoked.is_(False)))
        .scalars()
        .all()
    )
    for row in rows:
        if verify_api_key(raw_key, row.key_hash):
            row.last_used_at = datetime.utcnow()
            db.commit()
            creator = db.execute(
                select(Admin).where(Admin.username == row.created_by)
            ).scalar_one_or_none()
            return creator or _SYSTEM_ADMIN
    return None


# ---------------------------------------------------------------------------
# Dependencies
# ---------------------------------------------------------------------------
def get_current_admin(
    request: Request,
    bearer: HTTPAuthorizationCredentials | None = Security(bearer_scheme),
    api_key: str | None = Security(api_key_header),
    db: DbSession = Depends(get_db),
) -> Admin:
    """Resolve the caller from a Bearer session token or an X-API-Key.

    Stashes the resolved username on ``request.state`` so the activity-log
    middleware can attribute the request without re-parsing credentials.
    """
    admin: Admin | None = None
    if bearer and bearer.credentials:
        admin = _admin_from_bearer(db, bearer.credentials)
    if admin is None and api_key:
        admin = _admin_from_api_key(db, api_key)
    if admin is None:
        raise APIError(status_code=401, detail="Invalid or missing credentials")
    request.state.username = admin.username
    return admin


def require_admin(admin: Admin = Depends(get_current_admin)) -> Admin:
    """Guard for endpoints only full admins may call."""
    if not admin.is_admin:
        raise APIError(status_code=403, detail="Administrator privileges required")
    return admin


# ---------------------------------------------------------------------------
# Activity log helper
# ---------------------------------------------------------------------------
def record_log(
    db: DbSession,
    *,
    username: str | None,
    action: str,
    detail: str | None = None,
    request: Request | None = None,
) -> None:
    """Insert one activity-log row. Never raises into the caller."""
    ip = None
    if request is not None and request.client is not None:
        ip = request.client.host
    try:
        db.add(LogEntry(username=username, action=action, detail=detail, ip_address=ip))
        db.commit()
    except Exception:
        db.rollback()