Files
Kochbuch/tests/test_infrastructure.py
T
Kyle DammannandClaude Opus 5.5 ef9deaeff7 Add phase 1: Docker setup, auth with invitations and roles
- Django 5.2 + PostgreSQL 17 via docker compose (app + db, volumes,
  healthcheck on /healthz, test profile running pytest against Postgres)
- Custom e-mail user model with user/admin roles
- Invitation-only registration with single-use, hashed, expiring links
- Admin area: invitations, role changes, deactivation, password reset
  via one-time link or temporary password (forced change on login)
- First admin created from ADMIN_EMAIL/ADMIN_PASSWORD on startup
- Login rate limiting per e-mail and IP (DB-backed, worker-safe)
- Proxy middleware honouring X-Forwarded-* only from TRUSTED_PROXIES,
  optional CLIENT_IP_HEADER (e.g. CF-Connecting-IP)
- Security headers (CSP, HSTS, Permissions-Policy), secure __Host- cookies
- Mobile-first German UI, CLAUDE.md, README, init-env script
- 65 tests for auth, permissions, proxy handling and rate limiting

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-30 21:24:59 +01:00

152 lines
5.3 KiB
Python

from datetime import timedelta
import pytest
from django.core.management import call_command
from django.core.management.base import CommandError
from django.http import HttpResponse
from django.test import Client, RequestFactory
from accounts.models import User
from core import ratelimit
from core.middleware import ProxyMiddleware
pytestmark = pytest.mark.django_db
# --- Proxy handling --------------------------------------------------------------
def run_proxy(remote_addr, **headers):
request = RequestFactory().get("/", REMOTE_ADDR=remote_addr, **headers)
ProxyMiddleware(lambda r: HttpResponse())(request)
return request
def test_forwarded_headers_from_trusted_proxy_are_used():
request = run_proxy("172.18.0.5", HTTP_X_FORWARDED_FOR="198.51.100.7", HTTP_X_FORWARDED_PROTO="https")
assert request.META["REMOTE_ADDR"] == "198.51.100.7"
assert request.is_secure()
def test_forwarded_headers_from_untrusted_peer_are_dropped():
request = run_proxy("203.0.113.50", HTTP_X_FORWARDED_FOR="10.0.0.1", HTTP_X_FORWARDED_PROTO="https")
assert request.META["REMOTE_ADDR"] == "203.0.113.50"
assert "HTTP_X_FORWARDED_FOR" not in request.META
assert not request.is_secure()
def test_spoofed_left_entries_in_forwarded_for_are_ignored():
# Client sent "X-Forwarded-For: 1.2.3.4"; the proxy appended the real client address.
request = run_proxy("172.18.0.5", HTTP_X_FORWARDED_FOR="1.2.3.4, 198.51.100.7")
assert request.META["REMOTE_ADDR"] == "198.51.100.7"
def test_chained_trusted_proxies_are_skipped():
request = run_proxy("172.18.0.5", HTTP_X_FORWARDED_FOR="198.51.100.7, 192.168.1.2")
assert request.META["REMOTE_ADDR"] == "198.51.100.7"
def test_client_ip_header_only_from_trusted_proxy(settings):
settings.CLIENT_IP_HEADER = "CF-Connecting-IP"
trusted = run_proxy("172.18.0.5", HTTP_CF_CONNECTING_IP="198.51.100.8", HTTP_X_FORWARDED_FOR="162.158.1.1")
assert trusted.META["REMOTE_ADDR"] == "198.51.100.8"
untrusted = run_proxy("203.0.113.50", HTTP_CF_CONNECTING_IP="198.51.100.8")
assert untrusted.META["REMOTE_ADDR"] == "203.0.113.50"
# --- Security headers & health -----------------------------------------------------
def test_security_headers(db):
response = Client().get("/anmelden/")
assert "default-src 'self'" in response["Content-Security-Policy"]
assert response["X-Frame-Options"] == "DENY"
assert response["X-Content-Type-Options"] == "nosniff"
assert response["Referrer-Policy"] == "same-origin"
assert "Permissions-Policy" in response
def test_hsts_on_secure_requests(db):
response = Client().get("/anmelden/", secure=True)
assert "max-age=" in response["Strict-Transport-Security"]
def test_session_cookie_flags(user):
from django.conf import settings
from .conftest import PASSWORD
response = Client().post("/anmelden/", {"email": user.email, "password": PASSWORD})
cookie = response.cookies[settings.SESSION_COOKIE_NAME]
assert cookie["httponly"]
assert cookie["samesite"] == "Lax"
assert bool(cookie["secure"]) == settings.COOKIE_SECURE
def test_authenticated_pages_not_cacheable(user_client):
assert "no-store" in user_client.get("/")["Cache-Control"]
def test_healthz(db):
response = Client().get("/healthz")
assert response.status_code == 200
assert response.json() == {"status": "ok"}
# --- Rate limiter ------------------------------------------------------------------
def test_ratelimit_counts_and_resets():
window = timedelta(minutes=1)
assert ratelimit.hit("k", 2, window)
assert ratelimit.hit("k", 2, window)
assert ratelimit.is_limited("k", 2, window)
assert not ratelimit.hit("k", 2, window)
ratelimit.reset("k")
assert not ratelimit.is_limited("k", 2, window)
def test_ratelimit_zero_means_unlimited():
for _ in range(50):
assert ratelimit.hit("u", 0, timedelta(minutes=1))
assert not ratelimit.is_limited("u", 0, timedelta(minutes=1))
def test_ratelimit_window_expires():
window = timedelta(seconds=0)
ratelimit.hit("w", 1, window)
assert not ratelimit.is_limited("w", 1, window)
assert ratelimit.hit("w", 1, window)
# --- First admin from environment --------------------------------------------------
def test_ensure_admin_creates_admin_once(monkeypatch):
monkeypatch.setenv("ADMIN_EMAIL", "[email protected]")
monkeypatch.setenv("ADMIN_PASSWORD", "Start-Passwort-2026")
call_command("ensure_admin")
admin = User.objects.get(email="[email protected]")
assert admin.is_admin and admin.check_password("Start-Passwort-2026")
# A second run with a different password must not touch the existing account.
monkeypatch.setenv("ADMIN_PASSWORD", "Anderes-Passwort-2026")
call_command("ensure_admin")
admin.refresh_from_db()
assert admin.check_password("Start-Passwort-2026")
assert User.objects.count() == 1
def test_ensure_admin_rejects_weak_password(monkeypatch):
monkeypatch.setenv("ADMIN_EMAIL", "[email protected]")
monkeypatch.setenv("ADMIN_PASSWORD", "123")
with pytest.raises(CommandError):
call_command("ensure_admin")
assert not User.objects.exists()
def test_ensure_admin_without_env_does_nothing(monkeypatch):
monkeypatch.delenv("ADMIN_EMAIL", raising=False)
call_command("ensure_admin")
assert not User.objects.exists()