114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
|
|
from django.conf import settings
|
|
from django.contrib import messages
|
|
from django.http import HttpRequest, HttpResponse, JsonResponse
|
|
from django.shortcuts import redirect
|
|
from django.urls import reverse
|
|
from django.utils.http import url_has_allowed_host_and_scheme
|
|
|
|
from scripts.migration_worker_gate import hold_active
|
|
|
|
|
|
class MigrationMaintenanceMiddleware:
|
|
"""Block candidate traffic before durable migration activation, including GET writes."""
|
|
|
|
def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:
|
|
self.get_response = get_response
|
|
|
|
def __call__(self, request: HttpRequest) -> HttpResponse:
|
|
try:
|
|
held = hold_active()
|
|
except (OSError, RuntimeError):
|
|
held = True
|
|
health = request.method in {"GET", "HEAD"} and request.path_info in {
|
|
"/health/ready/", "/health/live/",
|
|
}
|
|
if held and not health:
|
|
response = JsonResponse({"error": "migration_maintenance"}, status=503)
|
|
response["Retry-After"] = "30"
|
|
response["Cache-Control"] = "no-store"
|
|
return response
|
|
return self.get_response(request)
|
|
|
|
|
|
class DemoReadOnlyMiddleware:
|
|
"""Prevent a shared public demo account from mutating application data."""
|
|
|
|
SAFE_METHODS = {"GET", "HEAD", "OPTIONS"}
|
|
|
|
def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:
|
|
self.get_response = get_response
|
|
|
|
def __call__(self, request: HttpRequest) -> HttpResponse:
|
|
user = getattr(request, "user", None)
|
|
is_demo = bool(
|
|
getattr(settings, "DEMO_READ_ONLY", True)
|
|
and user
|
|
and getattr(user, "is_authenticated", False)
|
|
and getattr(user, "username", "") == settings.DEMO_USERNAME
|
|
)
|
|
if not is_demo or request.method in self.SAFE_METHODS:
|
|
return self.get_response(request)
|
|
|
|
if request.path_info == reverse("logout"):
|
|
return self.get_response(request)
|
|
|
|
if "application/json" in request.headers.get("Accept", ""):
|
|
return JsonResponse(
|
|
{
|
|
"ok": False,
|
|
"error": "demo_read_only",
|
|
"message": "De publieke demo is alleen-lezen.",
|
|
},
|
|
status=403,
|
|
)
|
|
|
|
messages.warning(
|
|
request,
|
|
"De publieke demo is alleen-lezen; wijzigingen zijn uitgeschakeld.",
|
|
)
|
|
referer = request.META.get("HTTP_REFERER", "")
|
|
if referer and url_has_allowed_host_and_scheme(
|
|
referer,
|
|
allowed_hosts={request.get_host()},
|
|
require_https=request.is_secure(),
|
|
):
|
|
return redirect(referer)
|
|
return redirect("dashboard:today")
|
|
|
|
|
|
class SecurityHeadersMiddleware:
|
|
"""Small CSP without external assets; vacancy HTML is sanitized separately."""
|
|
|
|
def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:
|
|
self.get_response = get_response
|
|
|
|
def __call__(self, request: HttpRequest) -> HttpResponse:
|
|
response = self.get_response(request)
|
|
response.headers.setdefault(
|
|
"Content-Security-Policy",
|
|
"; ".join(
|
|
[
|
|
"default-src 'self'",
|
|
"img-src 'self' data:",
|
|
"style-src 'self' 'unsafe-inline'",
|
|
"script-src 'self'",
|
|
"font-src 'self'",
|
|
"connect-src 'self'",
|
|
"frame-ancestors 'none'",
|
|
"base-uri 'self'",
|
|
"form-action 'self'",
|
|
]
|
|
),
|
|
)
|
|
response.headers.setdefault(
|
|
"Permissions-Policy", "camera=(), microphone=(), geolocation=()"
|
|
)
|
|
response.headers.setdefault("Cross-Origin-Opener-Policy", "same-origin")
|
|
if not getattr(settings, "SEARCH_ENGINE_INDEXING_ENABLED", False):
|
|
response.headers.setdefault("X-Robots-Tag", "noindex, nofollow, noarchive")
|
|
return response
|