Local account #{{ target.pk }}: {{ target.account_email }} — {{ target.tyler_jurisdiction }}
+Created {{ target.created_at }}. Last sign-in {{ target.last_login|default:"Unknown" }}. Plans: {{ plans }}. Archived cases: {{ archived }}.
+| Draft | +Status | +Created | +Updated | +Documents | +Parties | +
|---|---|---|---|---|---|
| #{{ draft.pk }} | {{ draft.status }} | {{ draft.created_at }} | {{ draft.updated_at }} | {{ draft.documents.count }} | {{ draft.parties.count }} |
| No filings. | |||||
Activity is estimated from session expiry and the configured session lifetime.
+ {% for session in sessions %}{% empty %}No stored sessions.
{% endfor %} +Choose what to delete, then select Review deletion to see the inventory. On the next screen, verify the requester's authority and confirm permanent deletion.
+ {% if not deletion_enabled %}Deletion is disabled on this deployment. You can review and verify a request, but the final delete action is unavailable until a deployment administrator enables deletion.
{% endif %} +Use the existing contact channel to verify identity and authority. Select this local account only. Other jurisdictions require separate requests. Do not collect passwords or filing documents for verification.
+ +{% endblock %} diff --git a/efile_app/efile/templates/efile/staff/accounts.html b/efile_app/efile/templates/efile/staff/accounts.html new file mode 100644 index 00000000..4276e6ed --- /dev/null +++ b/efile_app/efile/templates/efile/staff/accounts.html @@ -0,0 +1,20 @@ +{% extends "efile/staff/base.html" %} +{% block staff_content %} + +| Account | +Jurisdiction | +Created | +|
|---|---|---|---|
| #{{ account.pk }} | {{ account.account_email }} | {{ account.tyler_jurisdiction }} | {{ account.created_at }} |
| No matching accounts. | |||
Reporting timezone: UTC. Collection enabled: {{ collecting|yesno:"Yes,No" }}. First retained coverage: {{ health.first_counter|default:"No collected data" }}. Latest update: {{ health.freshness|default:"None" }}. Pending events: {{ health.pending }}.
+Successful transmissions measure delivery to the filing service, not court acceptance. Matters and repeat use are counted per local account; they do not identify unique people. Case types use configured categories; unmatched types are grouped as other or unknown.
+Small cohorts suppress the whole requested breakdown and period, including its subtotals and rare labels. Detailed breakdowns require complete UTC calendar months.
+ +| Period | +Jurisdiction | +Metric | +New/existing | +Breakdown | +Value | +Count | +
|---|---|---|---|---|---|---|
| {{ row.period }} | {{ row.jurisdiction }} | {{ row.metric }} | {{ row.filing_kind }} | {{ row.dimension }} | {{ row.value }} | {{ row.count }} |
| No rows in this range. | ||||||
Provision TOTP for local staff account #{{ target.pk }} ({{ target.username }}). Verify the recipient through an approved staff onboarding channel.
+ {% if config_url %}This secret is shown once. Transfer it securely to the administrator and have them add it to their authenticator. Do not put it in email, tickets, screenshots, or application logs.
{{ config_url }}
Reference {{ item.reference }} — {{ item.jurisdiction }} — {{ item.get_status_display }}.
+Outcome: {{ item.outcome|default:"Pending" }}. {% if item.account_wide %}Scope: delete this local account and all of its LITEFile data.{% else %}Scope: delete selected filings and keep the account.{% endif %}
+ {% if item.target_id %}Account #{{ item.target_id }} — {{ item.target.account_email }}.
{% endif %} +Shared objects and correction chains outside this scope are blockers. Submitting filings and uncertain transmission outcomes require reconciliation; active extraction workers must stop before deletion.
+Deleting LITEFile's copies does not withdraw a filing, erase court records, or delete a Tyler account. Backup and provider cleanup need a separate operator disposition.
+ {% if item.status != 'completed' and not deletion_enabled %}Deletion is disabled on this deployment. You can review and verify this request. Ask a deployment administrator to enable deletion before the final confirmation.
{% endif %} + {% if item.status == 'received' %} +Review the inventory above and verify the requester's identity and authority. Then continue to the final deletion confirmation.
+ + {% elif item.status != 'completed' %} +Resolve the items under Requires operator review before deleting this data.
{% endif %} + + {% endif %} + {% endif %} + {% if item.status == 'completed' and item.external_cleanup_pending %}Live-system deletion completed. Backup/provider disposition remains outstanding; do not describe the request as fully erased.
{% endif %} +{% endblock %} diff --git a/efile_app/efile/templates/efile/staff/requests.html b/efile_app/efile/templates/efile/staff/requests.html new file mode 100644 index 00000000..13e85cb3 --- /dev/null +++ b/efile_app/efile/templates/efile/staff/requests.html @@ -0,0 +1,18 @@ +{% extends "efile/staff/base.html" %} +{% block staff_content %} +| Reference | +Jurisdiction | +Status | +Created | +Outcome | +
|---|---|---|---|---|
| {{ item.reference }} | {{ item.jurisdiction }} | {{ item.get_status_display }} | {{ item.created_at }} | {{ item.outcome }} |
| No requests. | ||||
+ {% translate "We keep aggregate usage counts to understand which kinds of filings people make, which jurisdictions they use, whether they start or respond to cases, whether they file for themselves or someone else, filer ZIP codes, and whether uploaded PDFs contain form fields. Reports do not include names, email addresses, account identifiers, case numbers, documents, or authentication credentials. We restrict access to these reports and hide small groups to reduce the risk of identifying a person. Aggregate counts may be retained indefinitely after account or filing data is deleted." %} +
{% translate "To provide e-filing, we may collect information you enter or upload, including your name, contact information, addresses, party information, case details, filing documents, payment information, and account credentials. We may also collect browser, device, IP address, and activity information needed for security, troubleshooting, and service improvement." %}
@@ -129,6 +132,9 @@+ {% translate "You can ask us to delete a particular filing's data or the data associated with your LITEFile account through the contact channel below. We verify your identity and authority before processing a request. Accounts in different jurisdictions are handled separately. Deleting LITEFile's copies does not withdraw a filing, erase a court record, or delete your account with the court's filing service. We may need to resolve a filing still in progress, shared records, retention requirements, or backup and provider-held copies before confirming what has been deleted." %} +
{% translate "Contact the Suffolk LIT Lab at" %}
{{ config.jurisdiction.contact_email|default:"litlab@suffolk.edu" }}
diff --git a/efile_app/efile/tests/test_extraction_claims.py b/efile_app/efile/tests/test_extraction_claims.py
index 0fe2628b..d639c481 100644
--- a/efile_app/efile/tests/test_extraction_claims.py
+++ b/efile_app/efile/tests/test_extraction_claims.py
@@ -219,7 +219,8 @@ def test_supervisor_terminates_a_timed_out_child_and_requeues(lead):
"efile.management.commands.process_document_extractions.multiprocessing.get_context",
return_value=Mock(Process=Mock(return_value=child)),
),
- patch("efile.management.commands.process_document_extractions.time.monotonic", side_effect=[0, 2, 2]),
+ # Initial rollup, child deadline, join deadline, rollup check, timeout.
+ patch("efile.management.commands.process_document_extractions.time.monotonic", side_effect=[0, 0, 2, 2, 2]),
):
call_command("process_document_extractions", once=True)
child.terminate.assert_called_once()
diff --git a/efile_app/efile/tests/test_staff_bootstrap.py b/efile_app/efile/tests/test_staff_bootstrap.py
new file mode 100644
index 00000000..ee9b6c1f
--- /dev/null
+++ b/efile_app/efile/tests/test_staff_bootstrap.py
@@ -0,0 +1,125 @@
+"""Initial staff credentials come from explicit secrets, never defaults."""
+
+import base64
+import io
+import secrets
+
+import pytest
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from django_otp.plugins.otp_totp.models import TOTPDevice
+
+from efile.models import FilingDocument, FilingDraft, UsageCounter, UserProfile
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture(autouse=True)
+def bootstrap_environment(monkeypatch):
+ for suffix in ("USERNAME", "PASSWORD", "EMAIL", "TOTP_SECRET"):
+ monkeypatch.delenv(f"LITEFILE_STAFF_BOOTSTRAP_{suffix}", raising=False)
+
+
+def configure(monkeypatch):
+ password = secrets.token_urlsafe(24)
+ monkeypatch.setenv("LITEFILE_STAFF_BOOTSTRAP_USERNAME", "environment-admin")
+ monkeypatch.setenv("LITEFILE_STAFF_BOOTSTRAP_PASSWORD", password)
+ return password
+
+
+def test_bootstrap_has_no_default_account():
+ with pytest.raises(CommandError, match="no default account"):
+ call_command("bootstrap_staff")
+ assert not UserProfile.objects.exists()
+
+
+@pytest.mark.parametrize("password", ["", "password"])
+def test_bootstrap_rejects_missing_or_weak_password(monkeypatch, password):
+ configure(monkeypatch)
+ monkeypatch.setenv("LITEFILE_STAFF_BOOTSTRAP_PASSWORD", password)
+ with pytest.raises(CommandError):
+ call_command("bootstrap_staff")
+ assert not UserProfile.objects.exists()
+ assert not TOTPDevice.objects.exists()
+
+
+def test_generated_totp_and_password_are_created_together(monkeypatch):
+ password = configure(monkeypatch)
+ output = io.StringIO()
+ call_command("bootstrap_staff", stdout=output)
+ user = UserProfile.objects.get(username="environment-admin")
+ device = TOTPDevice.objects.get(user=user)
+ assert user.is_active and user.is_staff and user.is_superuser
+ assert user.check_password(password)
+ assert device.confirmed and len(bytes.fromhex(device.key)) == 20
+ assert device.config_url in output.getvalue()
+ assert password not in output.getvalue()
+
+
+def test_supplied_secret_and_repeated_bootstrap_never_echo_or_overwrite(monkeypatch):
+ password = configure(monkeypatch)
+ key = b"t" * 20
+ secret = base64.b32encode(key).decode()
+ monkeypatch.setenv("LITEFILE_STAFF_BOOTSTRAP_TOTP_SECRET", secret)
+ output = io.StringIO()
+ call_command("bootstrap_staff", stdout=output)
+ device = TOTPDevice.objects.get()
+ assert device.key == key.hex()
+ assert secret not in output.getvalue()
+ assert "otpauth://" not in output.getvalue()
+ monkeypatch.setenv("LITEFILE_STAFF_BOOTSTRAP_PASSWORD", secrets.token_urlsafe(24))
+ monkeypatch.setenv("LITEFILE_STAFF_BOOTSTRAP_TOTP_SECRET", base64.b32encode(b"x" * 20).decode())
+ call_command("bootstrap_staff", stdout=output)
+ device.refresh_from_db()
+ assert device.key == key.hex()
+ assert TOTPDevice.objects.count() == 1
+ assert UserProfile.objects.get().check_password(password)
+
+
+@pytest.mark.parametrize("secret", ["not-a-base32-secret!", "JBSWY3DPEHPK3PXP"])
+def test_invalid_or_short_totp_rolls_back_all_creation(monkeypatch, secret):
+ configure(monkeypatch)
+ monkeypatch.setenv("LITEFILE_STAFF_BOOTSTRAP_TOTP_SECRET", secret)
+ with pytest.raises(CommandError):
+ call_command("bootstrap_staff")
+ assert not UserProfile.objects.exists()
+
+
+def test_bootstrap_never_promotes_existing_filer(monkeypatch):
+ configure(monkeypatch)
+ original_password = secrets.token_urlsafe(24)
+ user = UserProfile.objects.create_user(username="environment-admin", password=original_password)
+ with pytest.raises(CommandError, match="cannot promote"):
+ call_command("bootstrap_staff")
+ user.refresh_from_db()
+ assert not user.is_staff and not user.is_superuser
+ assert user.check_password(original_password)
+
+
+def test_demo_is_local_idempotent_and_has_no_uploads(settings, monkeypatch):
+ settings.DEBUG = True
+ settings.LITEFILE_ANALYTICS_ENABLED = False
+ monkeypatch.delenv("FLY_APP_NAME", raising=False)
+ output = io.StringIO()
+ call_command("seed_staff_demo", stdout=output)
+ before = list(UsageCounter.objects.values_list("pk", "count", "contributors").order_by("pk"))
+ assert before and all(contributors == 5 for _, _, contributors in before)
+ draft_count = FilingDraft.objects.count()
+ assert draft_count > 0
+ assert not FilingDocument.objects.exists()
+ assert all(user.analytics_excluded and not user.has_usable_password() for user in UserProfile.objects.all())
+ call_command("seed_staff_demo", stdout=output)
+ assert list(UsageCounter.objects.values_list("pk", "count", "contributors").order_by("pk")) == before
+ assert FilingDraft.objects.count() == draft_count
+
+
+@pytest.mark.parametrize("deployed", [False, True])
+def test_demo_refuses_production_or_deployed_host(settings, monkeypatch, deployed):
+ settings.DEBUG = deployed
+ if deployed:
+ monkeypatch.setenv("FLY_APP_NAME", "deployed-app")
+ else:
+ monkeypatch.delenv("FLY_APP_NAME", raising=False)
+ with pytest.raises(CommandError):
+ call_command("seed_staff_demo")
+ assert not UserProfile.objects.exists()
diff --git a/efile_app/efile/tests/test_staff_tools.py b/efile_app/efile/tests/test_staff_tools.py
new file mode 100644
index 00000000..5e2663e2
--- /dev/null
+++ b/efile_app/efile/tests/test_staff_tools.py
@@ -0,0 +1,614 @@
+"""Exercise real permissions, TOTP login, privacy cleanup, and usage reports."""
+
+import csv
+import io
+import json
+import uuid
+from datetime import UTC, date, datetime, timedelta
+from unittest.mock import Mock, patch
+
+import pytest
+from django.contrib.sessions.backends.db import SessionStore
+from django.contrib.sessions.models import Session
+from django.core.exceptions import PermissionDenied
+from django.urls import reverse
+from django.utils import timezone
+from django_otp.oath import totp
+from django_otp.plugins.otp_totp.models import TOTPDevice
+
+from efile.models import (
+ ArchivedCase,
+ DocumentExtraction,
+ FilingDocument,
+ FilingDraft,
+ FilingMetadataEvent,
+ FilingParty,
+ FilingPlan,
+ InterviewHandoff,
+ PendingActivation,
+ PrivacyRequest,
+ StaffAudit,
+ StaffRoleGrant,
+ StoredUpload,
+ UsageContributor,
+ UsageCounter,
+ UsageEvent,
+ UsageMatter,
+)
+from efile.services.analytics import count_event, drain_events, record_event, record_matter, report_rows
+from efile.services.document_extractions import claim_next_extraction, process_document_extraction
+from efile.services.document_uploads import upload_files
+from efile.services.privacy import create_request, preview, process_request, verify_request
+from efile.utils.s3_upload_handler import S3UploadHandler
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture
+def owner(django_user_model):
+ return django_user_model.objects.create_user(
+ username="litigant",
+ email="private@example.com",
+ tyler_username="private@example.com",
+ tyler_jurisdiction="illinois",
+ )
+
+
+@pytest.fixture
+def administrator(django_user_model):
+ user = django_user_model.objects.create_user(username="staff", password="a-strong-local-password", is_staff=True)
+ StaffRoleGrant.objects.create(user=user, jurisdiction="illinois", role="accounts")
+ return user
+
+
+def verified_client(client, user):
+ device = TOTPDevice.objects.create(user=user, name="Test authenticator")
+ client.force_login(user, backend="django.contrib.auth.backends.ModelBackend")
+ session = client.session
+ session["otp_device_id"] = device.persistent_id
+ session["staff_verified_at"] = timezone.now().timestamp()
+ session.save()
+ return client
+
+
+def make_session(user, draft=None):
+ session = SessionStore()
+ session["_auth_user_id"] = str(user.pk)
+ session["auth_tokens"] = {"private": "SECRET-COURT-TOKEN"}
+ if draft:
+ session["filing_draft_id"] = draft.pk
+ session.save()
+ return session.session_key
+
+
+@pytest.mark.parametrize("endpoint", ["accounts", "requests", "analytics"])
+def test_staff_pages_require_totp_and_role(client, owner, administrator, endpoint):
+ url = reverse(f"litefile_staff:{endpoint}")
+ assert client.get(url).status_code == 302
+ client.force_login(owner)
+ assert client.get(url).status_code == 302
+ client.force_login(administrator, backend="django.contrib.auth.backends.ModelBackend")
+ assert client.get(url).status_code == 302
+ verified_client(client, administrator)
+ assert client.get(url).status_code == (403 if endpoint == "analytics" else 200)
+
+
+def test_analytics_only_role_cannot_read_delete_or_grant(client, django_user_model, owner, settings):
+ user = django_user_model.objects.create_user(username="reporter", is_staff=True)
+ StaffRoleGrant.objects.create(user=user, jurisdiction="illinois", role="analytics")
+ verified_client(client, user)
+ assert client.get(reverse("litefile_staff:analytics")).status_code == 200
+ for url in [
+ reverse("litefile_staff:account", args=[owner.pk]),
+ reverse("litefile_staff:requests"),
+ reverse("litefile_staff:efile_staffrolegrant_add"),
+ ]:
+ assert client.get(url).status_code == 403
+ assert client.post(url, {"action": "process"}).status_code == 403
+ settings.LITEFILE_ANALYTICS_EXPORT_ENABLED = True
+ response = client.get(
+ reverse("litefile_staff:analytics"),
+ {
+ "jurisdiction": "vermont",
+ "start": "2026-09-01",
+ "end": "2026-09-30",
+ "grouping": "month",
+ "dimension": "all",
+ "format": "csv",
+ },
+ )
+ assert response["Content-Type"].startswith("text/html") # Invalid scoped choice never exports.
+
+
+def test_totp_login_local_only_replay_and_expiry(client, administrator, settings):
+ device = TOTPDevice.objects.create(user=administrator, name="Authenticator")
+ url = reverse("litefile_staff:login")
+ with patch("efile.authentication.auth_with_tyler_api") as tyler:
+ response = client.post(url, {"username": administrator.username, "password": "a-strong-local-password"})
+ assert response.status_code == 200
+ assert "_auth_user_id" not in client.session
+ token = str(totp(device.bin_key, step=device.step, t0=device.t0, digits=device.digits))
+ response = client.post(
+ url, {"username": administrator.username, "password": "a-strong-local-password", "otp_token": token}
+ )
+ assert response.status_code == 302
+ assert client.get(reverse("litefile_staff:accounts")).status_code == 200
+ tyler.assert_not_called()
+ client.logout()
+ assert (
+ client.post(
+ url, {"username": administrator.username, "password": "a-strong-local-password", "otp_token": token}
+ ).status_code
+ == 200
+ )
+ verified_client(client, administrator)
+ session = client.session
+ session["staff_verified_at"] = timezone.now().timestamp() - settings.LITEFILE_STAFF_SESSION_SECONDS - 1
+ session.save()
+ assert client.get(reverse("litefile_staff:accounts")).status_code == 302
+
+
+def test_role_revocation_and_private_headers(client, administrator):
+ verified_client(client, administrator)
+ response = client.get(reverse("litefile_staff:accounts"))
+ assert response["Cache-Control"] == "no-store"
+ assert response["Referrer-Policy"] == "same-origin"
+ assert "noindex" in response["X-Robots-Tag"]
+ administrator.staff_roles.all().delete()
+ assert client.get(reverse("litefile_staff:accounts")).status_code == 302
+ client.logout()
+ assert client.get("/admin/").status_code == 404
+
+
+@pytest.mark.parametrize("secure", [False, True])
+def test_staff_login_preserves_same_origin_csrf_checks(administrator, secure):
+ from django.test import Client
+
+ device = TOTPDevice.objects.create(user=administrator, name="CSRF test authenticator")
+ browser = Client(enforce_csrf_checks=True)
+ url = reverse("litefile_staff:login")
+ host = "localhost:8001"
+ origin = f"{'https' if secure else 'http'}://{host}"
+ response = browser.get(url, secure=secure, HTTP_HOST=host)
+ assert response["Referrer-Policy"] == "same-origin"
+ data = {
+ "username": administrator.username,
+ "password": "a-strong-local-password",
+ "otp_token": str(totp(device.bin_key)).zfill(6),
+ "csrfmiddlewaretoken": browser.cookies["csrftoken"].value,
+ }
+ for untrusted_origin in ("null", "https://untrusted.invalid"):
+ assert browser.post(url, data, secure=secure, HTTP_HOST=host, HTTP_ORIGIN=untrusted_origin).status_code == 403
+ assert (
+ browser.post(
+ url,
+ {key: value for key, value in data.items() if key != "csrfmiddlewaretoken"},
+ secure=secure,
+ HTTP_HOST=host,
+ HTTP_ORIGIN=origin,
+ ).status_code
+ == 403
+ )
+ with patch("efile.authentication.auth_with_tyler_api") as court_auth:
+ assert browser.post(url, data, secure=secure, HTTP_HOST=host, HTTP_ORIGIN=origin).status_code == 302
+ assert browser.get(reverse("litefile_staff:accounts"), secure=secure, HTTP_HOST=host).status_code == 200
+ court_auth.assert_not_called()
+
+
+def test_lookup_scoped_minimal_and_get_does_not_delete(client, owner, administrator, django_user_model):
+ other = django_user_model.objects.create_user(
+ username="other-state", email=owner.email, tyler_jurisdiction="vermont"
+ )
+ draft = FilingDraft.objects.create(
+ user=owner, jurisdiction="illinois", submission_response={"payment": "PRIVATE-PAYMENT"}
+ )
+ key = make_session(owner, draft)
+ verified_client(client, administrator)
+ response = client.post(reverse("litefile_staff:accounts"), {"jurisdiction": "illinois", "email": owner.email})
+ assert response.status_code == 200
+ assert list(response.context["accounts"]) == [owner]
+ assert client.get(reverse("litefile_staff:account", args=[other.pk])).status_code == 403
+ response = client.get(reverse("litefile_staff:account", args=[owner.pk]), {"action": "revoke"})
+ assert response.status_code == 200
+ assert b"SECRET-COURT-TOKEN" not in response.content
+ assert b"PRIVATE-PAYMENT" not in response.content
+ assert Session.objects.filter(pk=key).exists()
+ token = response.context["sessions"][0]["token"]
+ client.post(reverse("litefile_staff:account", args=[owner.pk]), {"action": "revoke", "session": token})
+ assert not Session.objects.filter(pk=key).exists()
+
+
+def test_full_deletion_inventory_and_retry(owner, administrator, django_user_model, settings):
+ settings.LITEFILE_PRIVACY_DELETION_ENABLED = True
+ other = django_user_model.objects.create_user(username="unrelated", tyler_jurisdiction="illinois")
+ plan = FilingPlan.objects.create(user=owner, jurisdiction="illinois", title="Private matter")
+ draft = FilingDraft.objects.create(user=owner, jurisdiction="illinois", plan=plan)
+ correction = FilingDraft.objects.create(user=owner, jurisdiction="illinois", correction_of=draft)
+ document = FilingDocument.objects.create(draft=draft, role="lead", s3_key="prepared", original_s3_key="original")
+ FilingDocument.objects.create(draft=correction, role="lead", s3_key="prepared", original_s3_key="original")
+ FilingParty.objects.create(draft=draft, role="filer", first_name="Private name", email=owner.email)
+ DocumentExtraction.objects.create(document=document, evidence={"private": "data"})
+ InterviewHandoff.objects.create(
+ draft=draft, source="test", source_id="1", idempotency_key="1", fingerprint="hash", payload={"private": "data"}
+ )
+ FilingMetadataEvent.objects.create(draft=draft, path="case_title", kind="user_edit", value={"private": "data"})
+ ArchivedCase.objects.create(user=owner, jurisdiction="illinois", case_tracking_id="private-case")
+ PendingActivation.remember(owner.email, "illinois")
+ PendingActivation.remember(owner.email, "vermont")
+ owned_session = make_session(owner, draft)
+ unrelated_session = make_session(other)
+ request = create_request(administrator, owner, account_wide=True)
+ handler = Mock()
+ with pytest.raises(ValueError, match="Verify"):
+ process_request(request.pk, administrator, handler=handler)
+ verify_request(request.pk, administrator)
+ assert preview(request)["counts"]["objects"] == 2
+ handler.erase_file.side_effect = [{"success": True}, {"success": False}]
+ request = process_request(request.pk, administrator, handler=handler)
+ assert request.status == "attention" and request.outcome == "storage_failed"
+ assert len(request.deleted_keys) == 1
+ assert FilingDraft.objects.filter(pk=draft.pk, deletion_pending=True).exists()
+ assert not Session.objects.filter(pk=owned_session).exists()
+ with pytest.raises(PermissionDenied):
+ document.name = "Restore erased data"
+ document.save()
+ with pytest.raises(PermissionDenied):
+ FilingDraft.objects.create(user=owner, jurisdiction="illinois")
+ handler.erase_file.side_effect = None
+ handler.erase_file.return_value = {"success": True}
+ request = process_request(request.pk, administrator, handler=handler)
+ assert request.status == "completed" and request.external_cleanup_pending
+ assert not django_user_model.objects.filter(pk=owner.pk).exists()
+ with pytest.raises(PermissionDenied):
+ make_session(owner) # A late login cannot recreate deleted credentials.
+ assert Session.objects.filter(pk=unrelated_session).exists()
+ assert PendingActivation.exists_for(owner.email, "vermont")
+ assert not PendingActivation.exists_for(owner.email, "illinois")
+ assert not any([request.target_id, request.object_keys, request.deleted_keys, request.draft_ids])
+ assert handler.erase_file.call_count == 3
+ assert process_request(request.pk, administrator, handler=handler).status == "completed"
+ assert "private" not in json.dumps(list(StaffAudit.objects.values("counts", "outcome")))
+
+
+def test_selected_scope_preserves_unrelated_data_and_blocks_shared_objects(owner, administrator, settings):
+ settings.LITEFILE_PRIVACY_DELETION_ENABLED = True
+ draft = FilingDraft.objects.create(user=owner, jurisdiction="illinois")
+ preserved = FilingDraft.objects.create(user=owner, jurisdiction="illinois")
+ removed = FilingDocument.objects.create(draft=draft, role="lead", s3_key="old-upload")
+ removed.delete() # The upload inventory must outlive the document row.
+ assert StoredUpload.objects.filter(draft=draft, key="old-upload").exists()
+ first_session = make_session(owner, draft)
+ preserved_session = make_session(owner, preserved)
+ request = create_request(administrator, owner, account_wide=False, draft_ids=[draft.pk])
+ verify_request(request.pk, administrator)
+ handler = Mock(erase_file=Mock(return_value={"success": True}))
+ process_request(request.pk, administrator, handler=handler)
+ handler.erase_file.assert_called_once_with("old-upload")
+ assert owner.filing_drafts.count() == 1
+ assert Session.objects.filter(pk=preserved_session).exists()
+ assert not Session.objects.filter(pk=first_session).exists()
+ shared = FilingDraft.objects.create(user=owner, jurisdiction="illinois")
+ FilingDocument.objects.create(draft=shared, role="lead", s3_key="shared")
+ FilingDocument.objects.create(draft=preserved, role="lead", s3_key="shared")
+ request = create_request(administrator, owner, account_wide=False, draft_ids=[shared.pk])
+ verify_request(request.pk, administrator)
+ assert "shared_objects" in preview(request)["blockers"]
+ assert process_request(request.pk, administrator, handler=handler).status == "attention"
+ assert handler.erase_file.call_count == 1
+
+
+@pytest.mark.parametrize("status", ["submitting", "error"])
+def test_submission_blocks_deletion(owner, administrator, settings, status):
+ settings.LITEFILE_PRIVACY_DELETION_ENABLED = True
+ FilingDraft.objects.create(user=owner, jurisdiction="illinois", status=status)
+ request = create_request(administrator, owner, account_wide=True)
+ verify_request(request.pk, administrator)
+ handler = Mock()
+ assert process_request(request.pk, administrator, handler=handler).outcome == "submission_requires_reconciliation"
+ handler.erase_file.assert_not_called()
+
+
+def test_s3_permanent_erasure_exact_versions(monkeypatch):
+ handler = S3UploadHandler()
+ handler.s3_client = Mock()
+ handler.bucket_name = "private-test-bucket"
+ monkeypatch.setattr(handler, "_ensure_initialized", Mock(return_value=True))
+ handler.s3_client.get_bucket_versioning.return_value = {"Status": "Enabled"}
+ paginator = handler.s3_client.get_paginator.return_value
+ paginator.paginate.side_effect = [
+ [
+ {
+ "Versions": [{"Key": "key", "VersionId": "v1"}, {"Key": "key-other", "VersionId": "v2"}],
+ "DeleteMarkers": [{"Key": "key", "VersionId": "marker"}],
+ }
+ ],
+ [],
+ ]
+ handler.s3_client.delete_objects.return_value = {}
+ assert handler.erase_file("key") == {"success": True}
+ assert handler.s3_client.delete_objects.call_args.kwargs["Delete"]["Objects"] == [
+ {"Key": "key", "VersionId": "v1"},
+ {"Key": "key", "VersionId": "marker"},
+ ]
+ handler.s3_client.delete_object.assert_not_called()
+ paginator.paginate.side_effect = [[{"Versions": [{"Key": "key", "VersionId": "locked"}]}]]
+ handler.s3_client.delete_objects.return_value = {"Errors": [{"Code": "AccessDenied"}]}
+ assert handler.erase_file("key") == {"success": False}
+
+
+def test_analytics_idempotency_dates_deletion_and_suppression(owner, administrator, settings):
+ settings.LITEFILE_ANALYTICS_ENABLED = True
+ settings.LITEFILE_PRIVACY_DELETION_ENABLED = True
+ at = datetime(2026, 9, 30, 23, 59, tzinfo=UTC)
+ with patch("efile.services.analytics.timezone.now", return_value=at):
+ draft = FilingDraft.objects.create(
+ user=owner, jurisdiction="illinois", existing_case="new", case_type_name="Name Change"
+ )
+ FilingParty.objects.create(
+ draft=draft, role="filer", party_type="plaintiff", party_side="initiating", zip_code="02108-9999"
+ )
+ record_event(draft, "review")
+ record_event(draft, "review")
+ record_event(draft, "submission_attempt", operation="logical-operation")
+ record_event(draft, "submission_attempt", operation="logical-operation")
+ with patch("efile.services.analytics.timezone.now", return_value=at + timedelta(minutes=1)):
+ draft.mark_submitted({})
+ drain_events()
+ before = list(UsageCounter.objects.order_by("pk").values("day", "metric", "count", "dimension", "value"))
+ drain_events()
+ assert before == list(UsageCounter.objects.order_by("pk").values("day", "metric", "count", "dimension", "value"))
+ assert UsageCounter.objects.get(metric="review", granularity="day", dimension="all").day == date(2026, 9, 30)
+ assert UsageCounter.objects.get(metric="transmitted", granularity="day", dimension="all").day == date(2026, 10, 1)
+ assert set(UsageEvent.objects.get(metric="review").dimensions) <= {
+ "case_type",
+ "filer_side",
+ "filing_for",
+ "zip_code",
+ "usage_frequency",
+ "filing_kind",
+ }
+ rows = report_rows(["illinois"], date(2026, 9, 1), date(2026, 9, 30), dimension="zip_code")
+ assert rows and all(row["count"] == "suppressed" and row["value"] == "suppressed" for row in rows)
+ request = create_request(administrator, owner, account_wide=True)
+ verify_request(request.pk, administrator)
+ process_request(request.pk, administrator, handler=Mock())
+ assert (
+ not UsageEvent.objects.exists() and not UsageContributor.objects.exists() and not UsageMatter.objects.exists()
+ )
+ assert before == list(UsageCounter.objects.order_by("pk").values("day", "metric", "count", "dimension", "value"))
+
+
+def test_analytics_worker_failure_retry_allowlist_and_staff_exclusion(owner, administrator, settings):
+ settings.LITEFILE_ANALYTICS_ENABLED = True
+ draft = FilingDraft.objects.create(user=owner, jurisdiction="illinois")
+ event = UsageEvent.objects.get(draft=draft)
+ with patch("efile.services.analytics.UsageCounter.objects.get_or_create", side_effect=RuntimeError("unavailable")):
+ with pytest.raises(RuntimeError):
+ count_event(event.pk)
+ assert not UsageCounter.objects.exists()
+ drain_events()
+ assert UsageCounter.objects.get(dimension="all", granularity="day").count == 1
+ event.dimensions["email"] = owner.email
+ event.counted_at = None
+ event.save()
+ with pytest.raises(ValueError, match="Disallowed"):
+ count_event(event.pk)
+ staff_draft = FilingDraft.objects.create(user=administrator, jurisdiction="illinois")
+ assert not UsageEvent.objects.filter(draft=staff_draft).exists()
+ settings.DEBUG = True
+ other = FilingDraft.objects.create(user=owner, jurisdiction="illinois")
+ assert not UsageEvent.objects.filter(draft=other).exists()
+
+
+def test_extraction_supervisor_rolls_usage_and_survives_rollup_failure(owner, settings):
+ from django.core.management import call_command
+
+ settings.LITEFILE_ANALYTICS_ENABLED = True
+ draft = FilingDraft.objects.create(user=owner, jurisdiction="illinois")
+ event = UsageEvent.objects.get(draft=draft)
+ with patch("efile.services.document_extractions.claim_next_extraction", return_value=None):
+ with patch("efile.management.commands.process_document_extractions.call_command", side_effect=RuntimeError):
+ call_command("process_document_extractions", once=True)
+ event.refresh_from_db()
+ assert event.counted_at is None
+ assert not UsageCounter.objects.exists()
+ call_command("process_document_extractions", once=True)
+ event.refresh_from_db()
+ assert event.counted_at is not None
+ assert event.dimensions == {}
+ assert UsageCounter.objects.get(dimension="all", granularity="day").count == 1
+
+
+def test_dashboard_csv_match(client, administrator, settings):
+ StaffRoleGrant.objects.create(user=administrator, jurisdiction="illinois", role="analytics")
+ settings.LITEFILE_ANALYTICS_EXPORT_ENABLED = True
+ UsageCounter.objects.create(
+ day=date(2026, 9, 1),
+ granularity="month",
+ jurisdiction="illinois",
+ metric="started",
+ filing_kind="new",
+ count=17,
+ contributors=10,
+ )
+ verified_client(client, administrator)
+ filters = {
+ "jurisdiction": "illinois",
+ "start": "2026-09-01",
+ "end": "2026-09-30",
+ "grouping": "month",
+ "dimension": "all",
+ }
+ url = reverse("litefile_staff:analytics")
+ response = client.get(url, filters)
+ assert response.status_code == 200
+ csv_response = client.get(url, {**filters, "format": "csv"})
+ rows = list(csv.DictReader(io.StringIO(csv_response.content.decode())))
+ assert rows == [{key: str(value) for key, value in row.items()} for row in response.context["rows"]]
+ UsageCounter.objects.create(
+ day=date(2026, 9, 1),
+ granularity="month",
+ jurisdiction="illinois",
+ metric="started",
+ filing_kind="new",
+ dimension="zip_code",
+ value="02108",
+ count=1,
+ contributors=1,
+ )
+ assert client.get(url, filters).context["rows"][0]["count"] == 17
+ details = {**filters, "dimension": "zip_code"}
+ assert client.get(url, details).context["rows"][0]["count"] == "suppressed"
+ assert "02108" not in client.get(url, {**details, "format": "csv"}).content.decode()
+
+
+def test_staff_privacy_workflow_csrf(client, owner, administrator, settings):
+ settings.LITEFILE_PRIVACY_DELETION_ENABLED = True
+ verified_client(client, administrator)
+ client.post(reverse("litefile_staff:accounts"), {"jurisdiction": "illinois", "email": owner.email})
+ assert client.session["staff_lookup"]["email"] == owner.email
+ assert "Review deletion" in client.get(reverse("litefile_staff:account", args=[owner.pk])).content.decode()
+ response = client.post(
+ reverse("litefile_staff:account", args=[owner.pk]), {"action": "request", "account_wide": "on"}
+ )
+ assert response.status_code == 302
+ request = PrivacyRequest.objects.get(target=owner)
+ assert type(owner).objects.filter(pk=owner.pk).exists()
+ url = reverse("litefile_staff:privacy_request", args=[request.pk])
+ assert client.get(url, {"action": "process", "confirmed": str(request.reference)}).status_code == 200
+ request.refresh_from_db()
+ assert request.status == "received"
+ assert "Verify and continue" in client.get(url).content.decode()
+ client.post(url, {"action": "process", "confirmed": str(request.reference)})
+ request.refresh_from_db()
+ assert request.status == "received"
+ client.post(url, {"action": "verify", "verified": "yes"})
+ settings.LITEFILE_PRIVACY_DELETION_ENABLED = False
+ disabled_page = client.get(url).content.decode()
+ assert "Deletion is disabled on this deployment" in disabled_page
+ assert "Permanently delete account" not in disabled_page
+ settings.LITEFILE_PRIVACY_DELETION_ENABLED = True
+ assert "Permanently delete account" in client.get(url).content.decode()
+ token = client.get(url).context["preview_token"]
+ client.post(url, {"action": "process", "confirmed": str(request.reference), "preview": token})
+ request.refresh_from_db()
+ assert request.status == "completed"
+ assert "staff_lookup" not in client.session
+ assert client.get(reverse("litefile_staff:account", args=[owner.pk])).status_code == 404
+ from django.test import Client
+
+ csrf_client = verified_client(Client(enforce_csrf_checks=True), administrator)
+ assert csrf_client.post(url, {"action": "external_resolved", "resolved": "yes"}).status_code == 403
+
+
+def test_superuser_can_onboard_staff_with_multiple_scoped_roles(client, django_user_model):
+ root = django_user_model.objects.create_superuser(
+ username="root", email="root@invalid.example", password="a-strong-local-password"
+ )
+ client.force_login(root, backend="django.contrib.auth.backends.ModelBackend")
+ assert client.get(reverse("litefile_staff:efile_userprofile_add")).status_code == 302
+ verified_client(client, root)
+ response = client.post(
+ reverse("litefile_staff:efile_userprofile_add"),
+ {
+ "username": "new-staff",
+ "password1": "a-different-strong-password",
+ "password2": "a-different-strong-password",
+ "usable_password": "true",
+ },
+ )
+ assert response.status_code == 302
+ staff = django_user_model.objects.get(username="new-staff")
+ assert staff.is_staff and not staff.is_superuser and staff.check_password("a-different-strong-password")
+ for role in ("accounts", "analytics"):
+ assert (
+ client.post(
+ reverse("litefile_staff:efile_staffrolegrant_add"),
+ {"user": staff.pk, "jurisdiction": "vermont", "role": role},
+ ).status_code
+ == 302
+ )
+ assert staff.staff_roles.count() == 2
+ url = reverse("litefile_staff:authenticator", args=[staff.pk])
+ assert client.get(url).status_code == 200
+ assert not TOTPDevice.objects.filter(user=staff).exists()
+ response = client.post(url, {"confirmed": "yes"})
+ assert response.status_code == 200 and b"otpauth://" in response.content
+ assert TOTPDevice.objects.filter(user=staff, confirmed=True).count() == 1
+
+
+def test_extraction_and_stale_upload_cannot_restore_frozen_data(owner, administrator, settings):
+ settings.LITEFILE_PRIVACY_DELETION_ENABLED = True
+ draft = FilingDraft.objects.create(user=owner, jurisdiction="illinois")
+ document = FilingDocument.objects.create(draft=draft, role="lead", s3_key="prepared")
+ job = DocumentExtraction.objects.create(
+ document=document,
+ status="processing",
+ claim_token=uuid.uuid4(),
+ lease_expires_at=timezone.now() + timedelta(minutes=15),
+ )
+ request = create_request(administrator, owner, account_wide=True)
+ verify_request(request.pk, administrator)
+ handler = Mock()
+ assert process_request(request.pk, administrator, handler=handler).outcome == "extraction_worker_in_flight"
+ handler.erase_file.assert_not_called()
+ FilingDraft.objects.filter(pk=draft.pk).update(deletion_pending=True)
+ assert claim_next_extraction() is None
+ with patch("efile.services.document_extractions.S3UploadHandler") as storage:
+ assert process_document_extraction(job.pk, job.claim_token) is None
+ storage.assert_not_called()
+ with patch("efile.services.document_uploads.store_prepared_document") as prepare:
+ with pytest.raises(ValueError, match="no longer available"):
+ upload_files(draft, [], "illinois")
+ prepare.assert_not_called()
+
+
+def test_failed_upload_retains_ownership_for_later_erasure(owner):
+ draft = FilingDraft.objects.create(user=owner, jurisdiction="illinois")
+ handler = Mock()
+ handler.delete_file.return_value = {"success": False}
+
+ def partial_upload(*args, keys, **kwargs):
+ keys.append("partially-saved-original")
+ raise ValueError("Preparation failed")
+
+ with (
+ patch("efile.services.document_uploads.S3UploadHandler", return_value=handler),
+ patch("efile.services.document_uploads.store_prepared_document", side_effect=partial_upload),
+ ):
+ with pytest.raises(ValueError, match="Preparation failed"):
+ upload_files(draft, [Mock(name="upload")], "illinois")
+ assert StoredUpload.objects.filter(draft=draft, key="partially-saved-original").exists()
+ assert not draft.documents.exists()
+
+
+def test_known_case_and_plan_aliases_do_not_inflate_matters(owner, settings):
+ settings.LITEFILE_ANALYTICS_ENABLED = True
+ plan = FilingPlan.objects.create(user=owner, jurisdiction="illinois", title="Matter")
+ first = FilingDraft.objects.create(user=owner, jurisdiction="illinois", court_code="court", plan=plan)
+ first.mark_submitted({})
+ plan.case_tracking_id = "known-case"
+ plan.save()
+ later = FilingDraft.objects.create(
+ user=owner, jurisdiction="illinois", court_code="court", plan=plan, previous_case_id="known-case"
+ )
+ later.mark_submitted({})
+ other_plan = FilingPlan.objects.create(user=owner, jurisdiction="illinois", title="Same matter")
+ third = FilingDraft.objects.create(
+ user=owner, jurisdiction="illinois", court_code="court", plan=other_plan, previous_case_id="known-case"
+ )
+ third.mark_submitted({})
+ record_matter(third)
+ assert UsageEvent.objects.filter(metric="matter_first_used").count() == 1
+ drain_events()
+ assert UsageCounter.objects.get(metric="matter_first_used", granularity="month", dimension="all").count == 1
+
+
+def test_outbox_rejects_free_text_in_an_allowed_dimension(owner, settings):
+ settings.LITEFILE_ANALYTICS_ENABLED = True
+ draft = FilingDraft.objects.create(user=owner, jurisdiction="illinois")
+ event = UsageEvent.objects.get(draft=draft)
+ event.dimensions["case_type"] = owner.email
+ event.save()
+ with pytest.raises(ValueError, match="Disallowed usage value"):
+ count_event(event.pk)
diff --git a/efile_app/efile/urls.py b/efile_app/efile/urls.py
index 03407c59..fdbfd4fe 100644
--- a/efile_app/efile/urls.py
+++ b/efile_app/efile/urls.py
@@ -3,6 +3,7 @@
from django.urls import include, path
from django.views.i18n import JavaScriptCatalog
+from efile.staff_admin import staff_site
from efile.utils.config_loader import config_loader
from efile.utils.jurisdiction_stuff import has_jurisdiction_login
@@ -65,6 +66,7 @@ def jurisdiction_homepage(request, jurisdiction):
urlpatterns = [
+ path(f"{settings.LITEFILE_STAFF_PATH}/", staff_site.urls),
path("api/handoffs/v1/", external_handoff, name="external_handoff"),
path("api/handoffs/v1/documents/", replace_documents, name="handoff_replace_documents"),
path("handoff/claim/