devplacepy/devplacepy/middleware/guards.py

# retoor <retoor@molodetz.nl>

import time

from fastapi import Request
from fastapi.responses import HTMLResponse, RedirectResponse

from devplacepy.database import get_setting
from devplacepy.middleware import state
from devplacepy.responses import json_error, wants_json
from devplacepy.seo import base_seo_context
from devplacepy.services.audit import record as audit
from devplacepy.templating import templates
from devplacepy.utils import client_ip, get_current_user


async def rate_limit_middleware(request: Request, call_next):
    if state.RATE_LIMIT_DISABLED:
        return await call_next(request)
    if request.method in (
        "POST",
        "PUT",
        "DELETE",
        "PATCH",
    ) and not request.url.path.startswith(("/openai", "/xmlrpc")):
        settings = state._hot_settings()
        limit = state._worker_rate_limit(settings["rate_limit_per_minute"])
        window = settings["rate_limit_window_seconds"]
        ip = client_ip(request, default="unknown")
        now = time.time()
        window_start = now - window
        if now - state._last_rate_sweep >= state.RATE_SWEEP_INTERVAL:
            state._sweep_rate_limit_store(window_start)
            state._last_rate_sweep = now
        timestamps = [t for t in state._rate_limit_store.get(ip, ()) if t > window_start]
        if len(timestamps) >= limit:
            state._rate_limit_store[ip] = timestamps
            audit.record(
                request,
                "security.rate_limit.block",
                result="denied",
                summary=f"request from {ip} blocked by rate limit",
                metadata={"ip": ip, "limit": limit, "window_seconds": window},
            )
            retry_after = {"Retry-After": str(window)}
            if wants_json(request):
                response = json_error(429, "Rate limit exceeded. Try again later.")
                response.headers["Retry-After"] = str(window)
                return response
            return HTMLResponse(
                "Rate limit exceeded. Try again later.",
                status_code=429,
                headers=retry_after,
            )
        timestamps.append(now)
        state._rate_limit_store[ip] = timestamps
    return await call_next(request)


_MAINTENANCE_ALLOWED_PREFIXES = ("/static", "/avatar", "/auth", "/admin", "/openai")


async def maintenance_middleware(request: Request, call_next):
    if state._hot_settings()["maintenance_mode"] != "1":
        return await call_next(request)
    if request.url.path.startswith(_MAINTENANCE_ALLOWED_PREFIXES):
        return await call_next(request)
    user = get_current_user(request)
    if user and user.get("role") == "Admin":
        return await call_next(request)
    message = get_setting(
        "maintenance_message",
        "DevPlace is undergoing scheduled maintenance. Please check back shortly.",
    )
    audit.record(
        request,
        "security.maintenance.block",
        user=user,
        result="denied",
        summary="non-admin request blocked by maintenance mode",
    )
    if wants_json(request):
        return json_error(503, message)
    seo_ctx = base_seo_context(
        request, title="Maintenance - DevPlace", description=message, robots="noindex"
    )
    return templates.TemplateResponse(
        request,
        "error.html",
        {**seo_ctx, "request": request, "error_code": 503, "error_message": message},
        status_code=503,
    )


_TERMS_ALLOWED_PREFIXES = (
    "/static",
    "/avatar",
    "/auth",
    "/docs",
    "/reports",
    "/block",
    "/mute",
    "/openai",
)

_TERMS_GATED_METHODS = ("POST", "PUT", "DELETE", "PATCH")


def _terms_exempt(path: str) -> bool:
    if path.startswith(_TERMS_ALLOWED_PREFIXES):
        return True
    return path.startswith("/profile/") and (
        path.endswith("/delete") or path.endswith("/consent")
    )


async def terms_acceptance_gate(request: Request, call_next):
    if request.method not in _TERMS_GATED_METHODS or _terms_exempt(request.url.path):
        return await call_next(request)
    from devplacepy.routers.auth.terms import (
        TERMS_ACCEPTANCE_CODE,
        current_terms_version,
        needs_acceptance,
    )

    user = get_current_user(request)
    if not needs_acceptance(user):
        return await call_next(request)
    message = "Accept the updated Terms of Service to continue."
    if wants_json(request):
        return json_error(
            403,
            message,
            code=TERMS_ACCEPTANCE_CODE,
            redirect="/auth/accept-terms",
            terms_version=current_terms_version(),
        )
    return RedirectResponse(url="/auth/accept-terms", status_code=303)