Kyle DammannandClaude Opus 5.5
ef9deaeff7
- 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]>
270 lines
10 KiB
Python
270 lines
10 KiB
Python
from datetime import timedelta
|
|
|
|
import pytest
|
|
from django.conf import settings
|
|
from django.test import Client
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
|
|
from accounts.models import Invitation, PasswordResetToken, User
|
|
|
|
from .conftest import PASSWORD
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
NEW_PASSWORD = "Ganz-neues-Passwort-77"
|
|
|
|
|
|
def token_from(link):
|
|
return link.rstrip("/").rsplit("/", 1)[-1]
|
|
|
|
|
|
def login(client, email, password):
|
|
return client.post(reverse("accounts:login"), {"email": email, "password": password})
|
|
|
|
|
|
# --- Login ------------------------------------------------------------------------
|
|
|
|
|
|
def test_login_with_valid_credentials(user):
|
|
client = Client()
|
|
response = login(client, "[email protected]", PASSWORD) # e-mail is case-insensitive
|
|
assert response.status_code == 302
|
|
assert client.get(reverse("core:home")).status_code == 200
|
|
|
|
|
|
def test_login_wrong_password_gives_generic_error(user):
|
|
response = login(Client(), user.email, "falsch")
|
|
assert response.status_code == 400
|
|
assert "E-Mail oder Passwort ist falsch." in response.content.decode()
|
|
|
|
|
|
def test_unknown_email_gives_same_error(db):
|
|
response = login(Client(), "[email protected]", "egal")
|
|
assert "E-Mail oder Passwort ist falsch." in response.content.decode()
|
|
|
|
|
|
def test_inactive_user_cannot_login(user):
|
|
user.is_active = False
|
|
user.save()
|
|
assert login(Client(), user.email, PASSWORD).status_code == 400
|
|
|
|
|
|
def test_password_is_hashed_with_argon2(user):
|
|
assert user.password.startswith("argon2")
|
|
|
|
|
|
def test_login_rate_limit_per_email(user):
|
|
client = Client()
|
|
for _ in range(settings.LOGIN_MAX_FAILURES_PER_EMAIL):
|
|
login(client, user.email, "falsch")
|
|
# Even the correct password is refused while locked.
|
|
response = login(client, user.email, PASSWORD)
|
|
assert response.status_code == 429
|
|
|
|
|
|
def test_login_rate_limit_per_ip(db, settings):
|
|
settings.LOGIN_MAX_FAILURES_PER_IP = 3
|
|
client = Client(REMOTE_ADDR="203.0.113.9")
|
|
for i in range(3):
|
|
login(client, f"x{i}@example.test", "falsch")
|
|
assert login(client, "[email protected]", "falsch").status_code == 429
|
|
# Other clients are unaffected.
|
|
assert login(Client(REMOTE_ADDR="203.0.113.10"), "[email protected]", "falsch").status_code == 400
|
|
|
|
|
|
def test_successful_login_resets_email_counter(user):
|
|
client = Client()
|
|
for _ in range(settings.LOGIN_MAX_FAILURES_PER_EMAIL - 1):
|
|
login(client, user.email, "falsch")
|
|
assert login(client, user.email, PASSWORD).status_code == 302
|
|
client.post(reverse("accounts:logout"))
|
|
login(client, user.email, "falsch")
|
|
assert login(client, user.email, PASSWORD).status_code == 302
|
|
|
|
|
|
def test_login_redirect_rejects_external_next(user):
|
|
response = Client().post(
|
|
reverse("accounts:login") + "?next=https://evil.example/",
|
|
{"email": user.email, "password": PASSWORD, "next": "https://evil.example/"},
|
|
)
|
|
assert response.status_code == 302
|
|
assert response["Location"] == reverse("core:home")
|
|
|
|
|
|
def test_logout_requires_post(user_client):
|
|
assert user_client.get(reverse("accounts:logout")).status_code == 405
|
|
assert user_client.post(reverse("accounts:logout")).status_code == 302
|
|
assert user_client.get(reverse("core:home")).status_code == 302
|
|
|
|
|
|
def test_login_requires_csrf_token(user):
|
|
client = Client(enforce_csrf_checks=True)
|
|
response = login(client, user.email, PASSWORD)
|
|
assert response.status_code == 403
|
|
|
|
|
|
# --- Login required everywhere ----------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"url",
|
|
["/", "/konto/", "/konto/passwort/", "/verwaltung/nutzer/", "/verwaltung/einladungen/"],
|
|
)
|
|
def test_pages_require_login(db, url):
|
|
response = Client().get(url)
|
|
assert response.status_code == 302
|
|
assert response["Location"].startswith(reverse("accounts:login"))
|
|
|
|
|
|
def test_no_open_registration_route(db):
|
|
for url in ["/registrieren/", "/register/", "/signup/", "/einladung/"]:
|
|
assert Client().get(url).status_code in (302, 404)
|
|
|
|
|
|
# --- Invitations ------------------------------------------------------------------
|
|
|
|
|
|
def register(client, token, **data):
|
|
payload = {"email": "[email protected]", "display_name": "Neu", "password1": NEW_PASSWORD, "password2": NEW_PASSWORD}
|
|
payload.update(data)
|
|
return client.post(reverse("accounts:register", args=[token]), payload)
|
|
|
|
|
|
def test_invitation_link_uses_app_base_url(admin_user):
|
|
_, link = Invitation.issue(created_by=admin_user)
|
|
assert link.startswith(settings.APP_BASE_URL + "/einladung/")
|
|
|
|
|
|
def test_register_with_valid_invitation(admin_user):
|
|
invitation, link = Invitation.issue(created_by=admin_user)
|
|
client = Client()
|
|
response = register(client, token_from(link))
|
|
assert response.status_code == 302
|
|
new_user = User.objects.get(email="[email protected]")
|
|
assert new_user.role == User.Role.USER
|
|
invitation.refresh_from_db()
|
|
assert invitation.used_by == new_user and invitation.status == "used"
|
|
assert client.get(reverse("core:home")).status_code == 200
|
|
|
|
|
|
def test_invitation_is_single_use(admin_user):
|
|
_, link = Invitation.issue(created_by=admin_user)
|
|
register(Client(), token_from(link))
|
|
response = register(Client(), token_from(link), email="[email protected]")
|
|
assert response.status_code == 404
|
|
assert not User.objects.filter(email="[email protected]").exists()
|
|
|
|
|
|
def test_expired_invitation_rejected(admin_user):
|
|
invitation, link = Invitation.issue(created_by=admin_user)
|
|
invitation.expires_at = timezone.now() - timedelta(minutes=1)
|
|
invitation.save()
|
|
assert Client().get(reverse("accounts:register", args=[token_from(link)])).status_code == 404
|
|
|
|
|
|
def test_revoked_invitation_rejected(admin_client, admin_user):
|
|
invitation, link = Invitation.issue(created_by=admin_user)
|
|
admin_client.post(reverse("accounts:admin_invitation_revoke", args=[invitation.pk]))
|
|
assert Client().get(reverse("accounts:register", args=[token_from(link)])).status_code == 404
|
|
|
|
|
|
def test_random_token_rejected(db):
|
|
assert Client().get(reverse("accounts:register", args=["nicht-gueltig"])).status_code == 404
|
|
|
|
|
|
def test_invitation_token_not_stored_in_plain_text(admin_user):
|
|
invitation, link = Invitation.issue(created_by=admin_user)
|
|
assert token_from(link) not in invitation.token_hash
|
|
|
|
|
|
def test_invitation_bound_to_email(admin_user):
|
|
_, link = Invitation.issue(created_by=admin_user, email="[email protected]")
|
|
register(Client(), token_from(link), email="[email protected]")
|
|
assert User.objects.filter(email="[email protected]").exists()
|
|
assert not User.objects.filter(email="[email protected]").exists()
|
|
|
|
|
|
def test_invitation_assigns_role(admin_user):
|
|
_, link = Invitation.issue(created_by=admin_user, role=User.Role.ADMIN)
|
|
register(Client(), token_from(link))
|
|
assert User.objects.get(email="[email protected]").role == User.Role.ADMIN
|
|
|
|
|
|
def test_register_rejects_existing_email(admin_user, user):
|
|
_, link = Invitation.issue(created_by=admin_user)
|
|
response = register(Client(), token_from(link), email=user.email.upper())
|
|
assert response.status_code == 200
|
|
assert User.objects.filter(email=user.email).count() == 1
|
|
|
|
|
|
def test_register_rejects_weak_password(admin_user):
|
|
_, link = Invitation.issue(created_by=admin_user)
|
|
response = register(Client(), token_from(link), password1="12345678", password2="12345678")
|
|
assert response.status_code == 200
|
|
assert not User.objects.filter(email="[email protected]").exists()
|
|
|
|
|
|
# --- Password reset by admin ------------------------------------------------------
|
|
|
|
|
|
def test_reset_link_flow(admin_user, user):
|
|
_, link = PasswordResetToken.issue(user, created_by=admin_user)
|
|
client = Client()
|
|
response = client.post(
|
|
reverse("accounts:password_reset", args=[token_from(link)]),
|
|
{"password1": NEW_PASSWORD, "password2": NEW_PASSWORD},
|
|
)
|
|
assert response.status_code == 302
|
|
user.refresh_from_db()
|
|
assert user.check_password(NEW_PASSWORD)
|
|
# Single use
|
|
assert Client().get(reverse("accounts:password_reset", args=[token_from(link)])).status_code == 404
|
|
|
|
|
|
def test_new_reset_link_invalidates_old(admin_user, user):
|
|
_, old = PasswordResetToken.issue(user, created_by=admin_user)
|
|
PasswordResetToken.issue(user, created_by=admin_user)
|
|
assert Client().get(reverse("accounts:password_reset", args=[token_from(old)])).status_code == 404
|
|
|
|
|
|
def test_reset_link_expires(admin_user, user):
|
|
token, link = PasswordResetToken.issue(user, created_by=admin_user)
|
|
token.expires_at = timezone.now() - timedelta(seconds=1)
|
|
token.save()
|
|
assert Client().get(reverse("accounts:password_reset", args=[token_from(link)])).status_code == 404
|
|
|
|
|
|
def test_reset_link_ends_existing_sessions(admin_user, user):
|
|
old_session = Client()
|
|
old_session.force_login(user)
|
|
_, link = PasswordResetToken.issue(user, created_by=admin_user)
|
|
Client().post(
|
|
reverse("accounts:password_reset", args=[token_from(link)]),
|
|
{"password1": NEW_PASSWORD, "password2": NEW_PASSWORD},
|
|
)
|
|
assert old_session.get(reverse("core:home")).status_code == 302
|
|
|
|
|
|
def test_temp_password_forces_change(admin_client, user):
|
|
response = admin_client.post(reverse("accounts:admin_user_action", args=[user.pk, "temp_password"]))
|
|
temp = response.context["value"]
|
|
client = Client()
|
|
assert login(client, user.email, temp).status_code == 302
|
|
# Every page redirects to the password change until a new password is set.
|
|
assert client.get(reverse("core:home"))["Location"] == reverse("accounts:password_change")
|
|
client.post(reverse("accounts:password_change"), {"password1": NEW_PASSWORD, "password2": NEW_PASSWORD})
|
|
assert client.get(reverse("core:home")).status_code == 200
|
|
user.refresh_from_db()
|
|
assert not user.must_change_password and user.check_password(NEW_PASSWORD)
|
|
|
|
|
|
def test_own_password_change_requires_old_password(user_client, user):
|
|
response = user_client.post(
|
|
reverse("accounts:password_change"),
|
|
{"old_password": "falsch", "password1": NEW_PASSWORD, "password2": NEW_PASSWORD},
|
|
)
|
|
assert response.status_code == 200
|
|
user.refresh_from_db()
|
|
assert user.check_password(PASSWORD)
|