@@ -28,7 +28,7 @@ def navigation_context(request: HttpRequest) -> dict[str, Any]:
|
||||
}
|
||||
context = {
|
||||
"app_name": "VacatureRadar",
|
||||
"app_version": "0.3.11",
|
||||
"app_version": settings.VACATURERADAR_VERSION,
|
||||
"app_owner_name": settings.VACATURERADAR_OWNER_NAME,
|
||||
"page_context_label": page_contexts.get(route_key, "Persoonlijke cockpit"),
|
||||
}
|
||||
@@ -38,4 +38,8 @@ def navigation_context(request: HttpRequest) -> dict[str, Any]:
|
||||
.only("id", "name")
|
||||
.first()
|
||||
)
|
||||
context["demo_read_only"] = bool(
|
||||
getattr(settings, "DEMO_READ_ONLY", True)
|
||||
and request.user.username == settings.DEMO_USERNAME
|
||||
)
|
||||
return context
|
||||
|
||||
+18
-2
@@ -1,9 +1,12 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import secrets
|
||||
from dataclasses import asdict, dataclass
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.cache import cache
|
||||
from django.db import connections
|
||||
from django.db.utils import OperationalError
|
||||
from django.db.utils import DatabaseError
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@@ -23,7 +26,20 @@ def readiness() -> HealthResult:
|
||||
cursor.execute("SELECT 1")
|
||||
cursor.fetchone()
|
||||
checks["database"] = "ok"
|
||||
except OperationalError as exc:
|
||||
except DatabaseError as exc:
|
||||
ok = False
|
||||
checks["database"] = f"error:{exc.__class__.__name__}"
|
||||
|
||||
if getattr(settings, "HEALTHCHECK_REQUIRE_CACHE", False):
|
||||
key = f"health:{secrets.token_hex(8)}"
|
||||
try:
|
||||
cache.set(key, "ok", timeout=10)
|
||||
if cache.get(key) != "ok":
|
||||
raise RuntimeError("cache roundtrip failed")
|
||||
cache.delete(key)
|
||||
checks["cache"] = "ok"
|
||||
except Exception as exc: # Health boundary: report category, never payload/details.
|
||||
ok = False
|
||||
checks["cache"] = f"error:{exc.__class__.__name__}"
|
||||
|
||||
return HealthResult(ok=ok, checks=checks)
|
||||
|
||||
+55
-2
@@ -2,11 +2,62 @@ from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
|
||||
from django.http import HttpRequest, HttpResponse
|
||||
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
|
||||
|
||||
|
||||
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:
|
||||
"""Kleine CSP zonder externe assets; vacature-HTML wordt bovendien gesanitized."""
|
||||
"""Small CSP without external assets; vacancy HTML is sanitized separately."""
|
||||
|
||||
def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:
|
||||
self.get_response = get_response
|
||||
@@ -33,4 +84,6 @@ class SecurityHeadersMiddleware:
|
||||
"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
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import ipaddress
|
||||
from functools import lru_cache
|
||||
|
||||
from django.conf import settings
|
||||
from django.http import HttpRequest
|
||||
|
||||
|
||||
@lru_cache(maxsize=32)
|
||||
def _trusted_networks(
|
||||
values: tuple[str, ...],
|
||||
) -> tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...]:
|
||||
return tuple(ipaddress.ip_network(value, strict=False) for value in values)
|
||||
|
||||
|
||||
def _parse_ip(value: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
|
||||
candidate = value.strip()
|
||||
if not candidate:
|
||||
return None
|
||||
try:
|
||||
return ipaddress.ip_address(candidate)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def client_ip(request: HttpRequest) -> str:
|
||||
"""Return a proxy-aware client IP without trusting arbitrary forwarded headers."""
|
||||
|
||||
remote = _parse_ip(request.META.get("REMOTE_ADDR", ""))
|
||||
if remote is None:
|
||||
return "unknown"
|
||||
|
||||
networks = _trusted_networks(tuple(getattr(settings, "TRUSTED_PROXY_CIDRS", ())))
|
||||
if not networks or not any(remote in network for network in networks):
|
||||
return str(remote)
|
||||
|
||||
forwarded = [
|
||||
parsed
|
||||
for item in request.META.get("HTTP_X_FORWARDED_FOR", "").split(",")
|
||||
if (parsed := _parse_ip(item)) is not None
|
||||
]
|
||||
if not forwarded:
|
||||
return str(remote)
|
||||
|
||||
# Walk from the nearest hop back to the client. The first address outside
|
||||
# the explicitly trusted proxy ranges is the client address.
|
||||
for candidate in reversed([*forwarded, remote]):
|
||||
if not any(candidate in network for network in networks):
|
||||
return str(candidate)
|
||||
return str(forwarded[0])
|
||||
@@ -6,12 +6,7 @@ from django.core.cache import cache
|
||||
from django.http import HttpRequest
|
||||
from django.utils import timezone
|
||||
|
||||
|
||||
def _client_ip(request: HttpRequest) -> str:
|
||||
forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR", "").strip()
|
||||
if forwarded_for:
|
||||
return forwarded_for.split(",")[0].strip()
|
||||
return request.META.get("REMOTE_ADDR", "unknown").strip() or "unknown"
|
||||
from apps.core.network import client_ip
|
||||
|
||||
|
||||
def _identity_for_user(request: HttpRequest) -> str:
|
||||
@@ -21,7 +16,7 @@ def _identity_for_user(request: HttpRequest) -> str:
|
||||
username = (
|
||||
(request.POST.get("username", "") if request.method == "POST" else "").strip().lower()
|
||||
)
|
||||
return f"anon:{username or _client_ip(request)}"
|
||||
return f"anon:{username or client_ip(request)}"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
|
||||
+10
-1
@@ -7,7 +7,7 @@ from django.contrib import messages
|
||||
from django.contrib.auth import get_user_model, login
|
||||
from django.contrib.auth.mixins import LoginRequiredMixin
|
||||
from django.contrib.auth.views import LoginView
|
||||
from django.http import JsonResponse
|
||||
from django.http import HttpResponse, JsonResponse
|
||||
from django.shortcuts import redirect
|
||||
from django.urls import reverse
|
||||
from django.utils.http import url_has_allowed_host_and_scheme
|
||||
@@ -228,10 +228,19 @@ def demo_login(request):
|
||||
logger.exception("Kon de demo-omgeving niet seeden")
|
||||
|
||||
login(request, user, backend=SSO_BACKEND)
|
||||
request.session.set_expiry(settings.DEMO_SESSION_SECONDS)
|
||||
next_url = _safe_next(request, request.POST.get("next"))
|
||||
return redirect(next_url or settings.LOGIN_REDIRECT_URL)
|
||||
|
||||
|
||||
def robots_txt(request):
|
||||
if getattr(settings, "SEARCH_ENGINE_INDEXING_ENABLED", False):
|
||||
content = "User-agent: *\nAllow: /\n"
|
||||
else:
|
||||
content = "User-agent: *\nDisallow: /\n"
|
||||
return HttpResponse(content, content_type="text/plain; charset=utf-8")
|
||||
|
||||
|
||||
def health_live(request):
|
||||
return JsonResponse({"ok": True, "service": "vacatureradar"})
|
||||
|
||||
|
||||
Reference in New Issue
Block a user