Require invited login and private downloads for the company pilot

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
This commit is contained in:
tegwick 2026-09-11 16:35:13 +02:00
parent 56bf193193
commit b7d7828f30
27 changed files with 584 additions and 111 deletions

View file

@ -0,0 +1,16 @@
from django.contrib.auth import forms
class _AccountFieldStyle:
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
for field in self.fields.values():
field.widget.attrs['class'] = 'form-input'
class AuthenticationForm(_AccountFieldStyle, forms.AuthenticationForm):
pass
class PasswordChangeForm(_AccountFieldStyle, forms.PasswordChangeForm):
pass

View file

@ -0,0 +1,21 @@
from django.contrib.auth.middleware import LoginRequiredMiddleware
from django.contrib.auth.views import redirect_to_login
from django.http import HttpResponse
from django.utils.cache import add_never_cache_headers
class PilotLoginRequiredMiddleware(LoginRequiredMiddleware):
"""Use Django's default-deny gate, including full-page HTMX reauthentication."""
def handle_no_permission(self, request, view_func):
if request.headers.get('HX-Request') == 'true':
# The original URL may render only a fragment or accept only POST.
login = redirect_to_login('/', self.get_login_url(view_func))
return HttpResponse(status=401, headers={'HX-Redirect': login.url})
return super().handle_no_permission(request, view_func)
def process_response(self, request, response):
# Company records and account pages must not survive in shared caches.
if request.path != '/health/':
add_never_cache_headers(response)
return response

View file

@ -1,3 +1,138 @@
from django.test import TestCase
from urllib.parse import parse_qs, urlsplit
# Create your tests here.
import pytest
from django.test import Client
from vergabe_teilnahme.apps.ausschreibungen.models import Ausschreibung
pytestmark = pytest.mark.django_db
def test_anonymous_cannot_read_tenders(client):
Ausschreibung.objects.create(titel='Confidential tender', ausschreiber='Pilot')
response = client.get('/ausschreibungen/')
assert response.status_code == 302
assert urlsplit(response.url).path == '/accounts/login/'
assert b'Confidential tender' not in response.content
def test_anonymous_cannot_create_tender(client):
response = client.post('/ausschreibungen/neu/', {
'titel': 'Unauthorised tender', 'ausschreiber': 'Pilot',
})
assert response.status_code == 302
assert not Ausschreibung.objects.exists()
def test_expired_htmx_session_requires_full_login(client):
response = client.get('/suche/?q=private', HTTP_HX_REQUEST='true')
assert response.status_code == 401
assert urlsplit(response['HX-Redirect']).path == '/accounts/login/'
# A fragment URL must never become the full page after login.
assert parse_qs(urlsplit(response['HX-Redirect']).query)['next'] == ['/']
def test_health_is_public_and_contains_no_tenant_data(client):
assert client.get('/health/').json() == {'status': 'ok'}
def test_invited_login_redirect_and_logout(mitarbeiter):
client = Client(enforce_csrf_checks=True)
response = client.get('/accounts/login/?next=/ausschreibungen/')
assert response.status_code == 200
assert b'Benutzername' in response.content
assert b'search-results' not in response.content
credentials = {'username': mitarbeiter.username, 'password': 'testpass',
'next': '/ausschreibungen/'}
assert client.post('/accounts/login/', credentials).status_code == 403
credentials['csrfmiddlewaretoken'] = client.cookies['csrftoken'].value
response = client.post('/accounts/login/', credentials)
assert response.status_code == 302
assert response.url == '/ausschreibungen/'
assert client.get(response.url).status_code == 200
assert client.get('/accounts/logout/').status_code == 405
assert client.post('/accounts/logout/').status_code == 403
response = client.post('/accounts/logout/', {
'csrfmiddlewaretoken': client.cookies['csrftoken'].value,
})
assert response.url == '/accounts/login/'
assert client.get('/ausschreibungen/').status_code == 302
def test_login_rejects_external_next(client, mitarbeiter):
response = client.post('/accounts/login/', {
'username': mitarbeiter.username, 'password': 'testpass',
'next': 'https://untrusted.example/private',
})
assert response.status_code == 302
assert response.url == '/'
def test_deactivated_member_loses_existing_session_and_cannot_login(client, mitarbeiter):
client.force_login(mitarbeiter)
mitarbeiter.is_active = False
mitarbeiter.save(update_fields=['is_active'])
assert client.get('/ausschreibungen/').status_code == 302
response = client.post('/accounts/login/', {
'username': mitarbeiter.username, 'password': 'testpass',
})
assert response.status_code == 200
assert response.context['form'].errors
assert client.get('/ausschreibungen/').status_code == 302
def test_member_password_change_keeps_session_and_replaces_password(client, mitarbeiter):
client.force_login(mitarbeiter)
assert client.get('/accounts/password-change/').status_code == 200
response = client.post('/accounts/password-change/', {
'old_password': 'testpass',
'new_password1': 'A-unique-pilot-password-914!',
'new_password2': 'A-unique-pilot-password-914!',
})
assert response.status_code == 302
assert client.get(response.url).status_code == 200
assert client.get('/ausschreibungen/').status_code == 200
mitarbeiter.refresh_from_db()
assert not mitarbeiter.check_password('testpass')
assert mitarbeiter.check_password('A-unique-pilot-password-914!')
def test_company_member_does_not_gain_django_admin_access(client, mitarbeiter):
client.force_login(mitarbeiter)
response = client.get('/admin/')
assert response.status_code == 302
assert urlsplit(response.url).path == '/admin/login/'
@pytest.mark.parametrize('debug', [False, True])
def test_media_requires_login_and_downloads_as_private_attachment(
client, mitarbeiter, tmp_path, settings, debug,
):
settings.DEBUG = debug
settings.MEDIA_ROOT = tmp_path / 'media'
settings.MEDIA_ROOT.mkdir()
(settings.MEDIA_ROOT / 'tender.pdf').write_bytes(b'%PDF-pilot')
assert client.get('/media/tender.pdf').status_code == 302
client.force_login(mitarbeiter)
response = client.get('/media/tender.pdf')
assert response.status_code == 200
assert b''.join(response.streaming_content) == b'%PDF-pilot'
assert response['Content-Disposition'] == 'attachment; filename="tender.pdf"'
assert response['X-Content-Type-Options'] == 'nosniff'
assert 'no-store' in response['Cache-Control']
def test_media_rejects_traversal_symlink_directory_and_missing_file(
client, mitarbeiter, tmp_path, settings,
):
settings.MEDIA_ROOT = tmp_path / 'media'
settings.MEDIA_ROOT.mkdir()
private = tmp_path / 'issues.db'
private.write_bytes(b'private issue state')
(settings.MEDIA_ROOT / 'escape.pdf').symlink_to(private)
(settings.MEDIA_ROOT / 'folder').mkdir()
client.force_login(mitarbeiter)
for path in ('../issues.db', '%2e%2e/issues.db', 'escape.pdf', 'folder', 'absent.pdf'):
response = client.get('/media/' + path)
assert response.status_code == 404
assert b'private issue state' not in response.content

View file

@ -0,0 +1,21 @@
from django.contrib.auth import views
from django.urls import path, reverse_lazy
from .forms import AuthenticationForm, PasswordChangeForm
app_name = 'accounts'
urlpatterns = [
path('login/', views.LoginView.as_view(
template_name='accounts/login.html', authentication_form=AuthenticationForm,
), name='login'),
path('logout/', views.LogoutView.as_view(), name='logout'),
path('password-change/', views.PasswordChangeView.as_view(
template_name='accounts/password_change.html',
form_class=PasswordChangeForm,
success_url=reverse_lazy('accounts:password_change_done'),
), name='password_change'),
path('password-change/done/', views.PasswordChangeDoneView.as_view(
template_name='accounts/password_change_done.html',
), name='password_change_done'),
]