import logging import secrets from urllib.parse import urlencode from django.conf import settings from django.contrib.auth import SESSION_KEY from django.db import OperationalError, ProgrammingError from django.http import HttpResponseForbidden from django.shortcuts import redirect, render from django.urls import reverse from wagtail.models import Site from .models import OfficeSettings from .i18n import normalize_language from .security import ( clear_rate_limit, client_ip, consume_rate_limit, ip_is_allowed, rate_fingerprint, rate_limit_status, ) logger = logging.getLogger("home.security") def _admin_prefixes(): return ("/cms/", "/django-admin/") def _is_admin_path(path): return path.startswith(_admin_prefixes()) class SecurityHeadersMiddleware: """Omeji nevarne zahteve in doda stroge glave javnemu delu strani.""" def __init__(self, get_response): self.get_response = get_response def __call__(self, request): request.csp_nonce = secrets.token_urlsafe(18) path = request.path if _is_admin_path(path) and not ip_is_allowed(client_ip(request), settings.ADMIN_ALLOWED_IPS): logger.warning("Zavrnjen dostop do CMS iz nedovoljenega omrežja (%s)", rate_fingerprint(client_ip(request))[:12]) return HttpResponseForbidden("Dostop do skrbniškega dela s tega omrežja ni dovoljen.") if request.method == "POST" and path in settings.CONTACT_FORM_PATHS: content_length = request.META.get("CONTENT_LENGTH", "0") try: content_length = int(content_length or 0) except ValueError: content_length = settings.CONTACT_MAX_BODY_BYTES + 1 if content_length > settings.CONTACT_MAX_BODY_BYTES: logger.warning("Zavrnjena prevelika kontaktna zahteva (%s)", rate_fingerprint(client_ip(request))[:12]) return render(request, "security/rate_limited.html", { "title": "Zahteva je prevelika", "message": "Kontaktni obrazec ne sprejema priponk ali tako velike količine podatkov.", }, status=413) raw_limit = consume_rate_limit( "contact-raw", rate_fingerprint("contact-raw", client_ip(request)), settings.CONTACT_RAW_RATE_LIMIT, settings.CONTACT_RAW_RATE_SECONDS, settings.CONTACT_RAW_RATE_SECONDS, ) if not raw_limit.allowed: response = render(request, "security/rate_limited.html", { "title": "Preveč poskusov", "message": "Oddanih je bilo preveč zahtev. Poskusite znova pozneje.", }, status=429) response["Retry-After"] = str(raw_limit.retry_after) return response response = self.get_response(request) response.headers.setdefault("Permissions-Policy", ( "accelerometer=(), autoplay=(), camera=(), display-capture=(), " "geolocation=(self), gyroscope=(), magnetometer=(), microphone=(), " "payment=(), usb=()" )) response.headers.setdefault("X-Permitted-Cross-Domain-Policies", "none") response.headers.setdefault("Cross-Origin-Resource-Policy", "same-site") if _is_admin_path(path): response.headers.setdefault("X-Robots-Tag", "noindex, nofollow, noarchive") response.headers.setdefault("Cache-Control", "no-store") else: nonce = request.csp_nonce response.headers.setdefault("Content-Security-Policy", "; ".join([ "default-src 'self'", "base-uri 'self'", "object-src 'none'", "frame-ancestors 'none'", "form-action 'self'", f"script-src 'self' 'nonce-{nonce}' https://challenges.cloudflare.com", "script-src-attr 'none'", "style-src 'self'", "img-src 'self' data: blob:", "font-src 'self'", "connect-src 'self' https://challenges.cloudflare.com", "frame-src https://challenges.cloudflare.com https://www.google.com https://maps.google.com https://consent.google.com", "media-src 'self'", "manifest-src 'self'", "upgrade-insecure-requests", ])) if request.method == "POST" or path in settings.CONTACT_FORM_PATHS: response.headers.setdefault("Cache-Control", "no-store") return response class LoginProtectionMiddleware: """Skupno omejevanje napačnih prijav za Wagtail in izbirni Django admin.""" LOGIN_PATHS = ("/cms/login/", "/django-admin/login/") PASSWORD_RESET_PATHS = ("/cms/password_reset/", "/django-admin/password_reset/") def __init__(self, get_response): self.get_response = get_response def __call__(self, request): if request.method == "POST" and request.path in self.PASSWORD_RESET_PATHS: reset_limit = consume_rate_limit( "password-reset", rate_fingerprint( "password-reset", client_ip(request), request.POST.get("email", "")[:254], ), settings.PASSWORD_RESET_RATE_LIMIT, settings.PASSWORD_RESET_RATE_SECONDS, settings.PASSWORD_RESET_RATE_SECONDS, ) if not reset_limit.allowed: logger.warning("Blokirane prepogoste zahteve za ponastavitev gesla") response = render(request, "security/rate_limited.html", { "title": "Preveč zahtev", "message": "Zahtev za ponastavitev gesla je bilo preveč. Poskusite znova pozneje.", }, status=429) response["Retry-After"] = str(reset_limit.retry_after) return response return self.get_response(request) if request.method != "POST" or request.path not in self.LOGIN_PATHS: return self.get_response(request) ip = client_ip(request) username = request.POST.get("username", "")[:150] ip_key = rate_fingerprint("login-ip", ip) account_key = rate_fingerprint("login-account-ip", ip, username) ip_status = rate_limit_status("login-ip", ip_key, settings.LOGIN_RATE_WINDOW_SECONDS) account_status = rate_limit_status( "login-account-ip", account_key, settings.LOGIN_RATE_WINDOW_SECONDS ) if not ip_status.allowed or not account_status.allowed: retry_after = max(ip_status.retry_after, account_status.retry_after) logger.warning("Blokirana prijava zaradi preveč neuspelih poskusov (%s)", ip_key[:12]) response = render(request, "security/rate_limited.html", { "title": "Prijava je začasno zaklenjena", "message": "Zaradi več neuspelih poskusov počakajte in nato poskusite znova.", }, status=429) response["Retry-After"] = str(retry_after) return response response = self.get_response(request) if request.session.get(SESSION_KEY): clear_rate_limit("login-ip", ip_key) clear_rate_limit("login-account-ip", account_key) else: consume_rate_limit( "login-ip", ip_key, settings.LOGIN_RATE_LIMIT * 2 - 1, settings.LOGIN_RATE_WINDOW_SECONDS, settings.LOGIN_BLOCK_SECONDS, ) result = consume_rate_limit( "login-account-ip", account_key, settings.LOGIN_RATE_LIMIT - 1, settings.LOGIN_RATE_WINDOW_SECONDS, settings.LOGIN_BLOCK_SECONDS, ) logger.warning("Neuspešen poskus prijave (%s)", account_key[:12]) if not result.allowed: response.headers["Retry-After"] = str(result.retry_after) return response class AdminTwoFactorMiddleware: """Za dostop do skrbniškega dela zahteva potrjen drugi faktor.""" EXEMPT_PATHS = ( "/cms/login/", "/cms/logout/", "/cms/2fa/", "/cms/2fa/setup/", "/cms/2fa/recovery-codes/", ) def __init__(self, get_response): self.get_response = get_response def __call__(self, request): if not settings.REQUIRE_ADMIN_2FA or not _is_admin_path(request.path): return self.get_response(request) if request.path in self.EXEMPT_PATHS: return self.get_response(request) user = getattr(request, "user", None) if not user or not user.is_authenticated: return self.get_response(request) if not user.is_staff or user.is_verified(): return self.get_response(request) from django_otp.plugins.otp_totp.models import TOTPDevice target = reverse("admin-2fa") if not TOTPDevice.objects.filter(user=user, confirmed=True).exists(): target = reverse("admin-2fa-setup") query = urlencode({"next": request.get_full_path()}) return redirect(f"{target}?{query}") class MaintenanceModeMiddleware: """Prikaže javno 503 obvestilo, CMS in statične datoteke pa pusti dostopne.""" EXEMPT_PREFIXES = ("/cms/", "/django-admin/", "/static/", "/media/") def __init__(self, get_response): self.get_response = get_response def __call__(self, request): if request.path.startswith(self.EXEMPT_PREFIXES) or getattr(request.user, "is_staff", False): return self.get_response(request) try: site = Site.find_for_request(request) if site: office = OfficeSettings.for_site(site) if office.maintenance_mode: language = normalize_language(getattr(request, "LANGUAGE_CODE", "sl")) response = render(request, "maintenance.html", { "office": office, "maintenance_title": office.localized("maintenance_title", language), "maintenance_message": office.localized("maintenance_message", language), }, status=503) response["Retry-After"] = "3600" return response except (OperationalError, ProgrammingError): pass return self.get_response(request)