app/security.py

from __future__ import annotations

import os
import secrets
from collections.abc import Iterable
from typing import Any

from fastapi import HTTPException, Request, status

SENSITIVE_ENV_SUFFIXES = ("KEY", "TOKEN", "SECRET", "PASSWORD", "PASS", "CREDENTIAL", "CREDENTIALS")


def authenticate(request: Request) -> None:
    if not request.session.get("authenticated"):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Login required")


def constant_time_equal(a: str, b: str) -> bool:
    return secrets.compare_digest(a.encode("utf-8"), b.encode("utf-8"))


def known_secret_values(extra_values: Iterable[str | None] = ()) -> list[str]:
    values: list[str] = []
    for key, value in os.environ.items():
        upper = key.upper()
        if value and len(value) >= 4 and any(upper.endswith(suffix) or suffix in upper for suffix in SENSITIVE_ENV_SUFFIXES):
            values.append(value)
    for value in extra_values:
        if value and len(value) >= 4:
            values.append(value)
    return sorted(set(values), key=len, reverse=True)


def redact_text(text: str | None, extra_values: Iterable[str | None] = ()) -> str:
    if text is None:
        return ""
    safe = str(text)
    for value in known_secret_values(extra_values):
        safe = safe.replace(value, "[REDACTED]")
    return safe


def redact_obj(value: Any) -> Any:
    if isinstance(value, str):
        return redact_text(value)
    if isinstance(value, list):
        return [redact_obj(item) for item in value]
    if isinstance(value, dict):
        return {key: redact_obj(item) for key, item in value.items()}
    return value