From f4a5937a4bd5b1728095a0bf41e9f65457066cdc Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Tue, 25 Aug 2026 23:48:22 +0200 Subject: [PATCH] =?UTF-8?q?feat(g1):=20DSGVO=20Art.15/17=20funktionsfaehig?= =?UTF-8?q?=20=E2=80=94=20fehlender=20process=5Fdsar=20Worker-Job=20implem?= =?UTF-8?q?entiert?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root-Cause: POST /dsar/{user_id} queued einen Job der nirgends implementiert war — DSAR-Requests verschwanden im Nirvana. Implementiert in app/core/jobs.py nach Hausmuster: _dsar_collect_user_data sammelt profile+contacts+audit_log+notifications (Art.15/20), _dsar_execute_deletion fuehrt Art.17 aus (contacts soft-delete respektiert Audit-Pflichten, notifications hard-delete, User anonymisiert+deaktiviert mit FK-Integritaet fuer Audit-Zeilen, dsar_erasure-Audit-Eintrag), process_dsar dispatcht access/deletion/rectification. Beweis: test_g1_dsar 4/4 gruen; ruff clean. --- app/core/jobs.py | 275 ++++++++++++++++++++++++++++++++++++++++++ tests/test_g1_dsar.py | 138 +++++++++++++++++++++ 2 files changed, 413 insertions(+) create mode 100644 tests/test_g1_dsar.py diff --git a/app/core/jobs.py b/app/core/jobs.py index baf00f5..cd65619 100644 --- a/app/core/jobs.py +++ b/app/core/jobs.py @@ -153,3 +153,278 @@ async def send_password_reset_email( from app.core.job_registry import register_job # noqa: E402 register_job("send_password_reset_email", send_password_reset_email) + + +# ── DSAR Processing Job (G1 DSGVO: Art. 15 Auskunft / Art. 17 Löschung) ───── + +async def _dsar_collect_user_data(db: Any, tenant_id: str, user_id: str) -> dict[str, Any]: + """Collect every data category the dsgvo-export route promises. + + Shared by type=access (full export) so both paths stay consistent. + """ + from datetime import UTC, datetime + from uuid import UUID as PyUUID + + from sqlalchemy import select as sa_select + + from app.models.audit import AuditLog + from app.models.contact import Contact + from app.models.notification import Notification + from app.models.user import User + + uid = PyUUID(user_id) + tid = PyUUID(tenant_id) + + export_data: dict[str, Any] = { + "user_id": user_id, + "exported_at": datetime.now(UTC).isoformat(), + "legal_basis": "GDPR Art. 15 (access) / Art. 20 (portability)", + "data": {}, + } + + # Profile + user = ( + await db.execute(sa_select(User).where(User.id == uid)) + ).scalar_one_or_none() + if user: + export_data["data"]["profile"] = { + "email": user.email, + "name": user.name, + "is_active": user.is_active, + "created_at": user.created_at.isoformat() if user.created_at else None, + } + + # Contacts owned by the user + contacts = ( + await db.execute( + sa_select(Contact).where( + Contact.tenant_id == tid, + Contact.owner_id == uid, + Contact.deleted_at.is_(None), + ) + ) + ).scalars().all() + export_data["data"]["contacts"] = [ + { + "id": str(c.id), + "type": c.type, + "displayname": c.displayname, + "email_1": c.email_1, + "email_2": c.email_2, + } + for c in contacts + ] + + # Audit trail entries by/about the user (bounded to keep payloads sane) + audit_entries = ( + await db.execute( + sa_select(AuditLog).where( + AuditLog.tenant_id == tid, + AuditLog.user_id == uid, + ).limit(1000) + ) + ).scalars().all() + export_data["data"]["audit_log"] = [ + { + "action": a.action, + "entity_type": a.entity_type, + "timestamp": a.timestamp.isoformat() if a.timestamp else None, + } + for a in audit_entries + ] + + # Notifications addressed to the user + notifications = ( + await db.execute( + sa_select(Notification).where( + Notification.tenant_id == tid, + Notification.owner_id == uid, + ).limit(1000) + ) + ).scalars().all() + export_data["data"]["notifications"] = [ + { + "id": str(n.id), + "type": getattr(n, "type", None), + "title": getattr(n, "title", None), + "created_at": n.created_at.isoformat() if n.created_at else None, + } + for n in notifications + ] + + return export_data + + +async def _dsar_execute_deletion(db: Any, tenant_id: str, user_id: str) -> dict[str, int]: + """Execute GDPR Art. 17 erasure for a user within one tenant. + + Strategy (respects retention duties): + - Contacts owned by the user → soft-delete via deleted_at + (audit history must remain intact — it is not personal data of the + subject but business record; retention policy governs its cleanup) + - Notifications owned by the user → hard delete + - User account → deactivate (is_active=False), clear personal fields, + scramble password hash and email (keeps FK integrity for audit rows) + Returns counters for the audit entry. + """ + from datetime import UTC, datetime + from uuid import UUID as PyUUID + + from sqlalchemy import select as sa_select + from sqlalchemy import update as sa_update + + from app.core.audit import log_audit + from app.models.contact import Contact + from app.models.notification import Notification + from app.models.user import User + + uid = PyUUID(user_id) + tid = PyUUID(tenant_id) + + counts: dict[str, int] = {} + + # 1. Soft-delete contacts owned by the user + contact_result = await db.execute( + sa_select(Contact).where( + Contact.tenant_id == tid, + Contact.owner_id == uid, + Contact.deleted_at.is_(None), + ) + ) + contacts = contact_result.scalars().all() + for c in contacts: + c.deleted_at = datetime.now(UTC) + counts["contacts_soft_deleted"] = len(contacts) + + # 2. Hard-delete notifications owned by the user + notif_result = await db.execute( + sa_select(Notification).where( + Notification.tenant_id == tid, + Notification.owner_id == uid, + ) + ) + notifications = notif_result.scalars().all() + for n in notifications: + await db.delete(n) + counts["notifications_deleted"] = len(notifications) + + # 3. Anonymize + deactivate the account (FK integrity for audit rows kept) + await db.execute( + sa_update(User) + .where(User.id == uid) + .values( + email=f"erased.{uid.hex[:16]}@anonymized.invalid", + name="[gelöscht gemäß DSGVO Art. 17]", + first_name=None, + last_name=None, + avatar_url=None, + password_hash="!dsar-erased", + is_active=False, + preferences={}, + ) + ) + counts["user_anonymized"] = 1 + + # 4. Audit the erasure itself (who/what/when — required by Art. 17 recital) + await log_audit( + db, + tid, + user_id, + "dsar_erasure", + "user", + uid, + {"target_user": user_id, **counts}, + ) + return counts + + +async def process_dsar( + ctx: dict[str, Any], + *, + user_id: str, + tenant_id: str, + request_type: str, +) -> dict[str, Any]: + """Process a GDPR Data Subject Access Request (DSAR). + + ARQ worker function registered as "process_dsar". + + request_type: + - "access": collect all data categories (Art. 15/20) and post a system + message that the export is ready (served via the existing dsgvo-export + endpoint). + - "deletion": execute Art. 17 erasure (soft-delete contacts, hard-delete + notifications, anonymize+deactivate account) and audit it. + - "rectification": post a system message asking admins to handle the + correction manually. + + Returns a summary dict for the job result. + """ + import logging + import uuid as uuid_module + + from app.core.db import get_worker_session_factory + from app.core.notifications import post_system_message + + logger = logging.getLogger(__name__) + tid = uuid_module.UUID(tenant_id) + uid = uuid_module.UUID(user_id) + + factory = get_worker_session_factory() + async with factory() as db: + try: + if request_type == "access": + data = await _dsar_collect_user_data(db, tenant_id, user_id) + await db.commit() + categories = list(data.get("data", {}).keys()) + await post_system_message( + db, + tid, + uid, + "dsar_access_ready", + "DSGVO-Auskunft bereit", + f"Datenkategorien: {', '.join(categories)}", + severity="info", + ) + await db.commit() + logger.info("DSAR access processed for user %s", user_id) + return {"type": request_type, "status": "completed", "categories": categories} + + if request_type == "deletion": + counts = await _dsar_execute_deletion(db, tenant_id, user_id) + await db.commit() + await post_system_message( + db, + tid, + uid, + "dsar_deletion_done", + "DSGVO-Löschung ausgeführt", + f"Kontakten soft-gelöscht: {counts.get('contacts_soft_deleted', 0)}; Konto anonymisiert.", + severity="info", + ) + await db.commit() + logger.info("DSAR deletion executed for user %s: %s", user_id, counts) + return {"type": request_type, "status": "completed", **counts} + + if request_type == "rectification": + await post_system_message( + db, + tid, + uid, + "dsar_rectification_requested", + "DSGVO-Berichtigung angefordert", + f"Manuelle Bearbeitung für User {user_id} erforderlich.", + severity="warning", + ) + await db.commit() + logger.info("DSAR rectification requested for user %s", user_id) + return {"type": request_type, "status": "queued_for_manual_handling"} + + logger.warning("Unknown DSAR request_type '%s' for user %s", request_type, user_id) + return {"type": request_type, "status": "unknown_type"} + except Exception: + await db.rollback() + raise + + +register_job("process_dsar", process_dsar) diff --git a/tests/test_g1_dsar.py b/tests/test_g1_dsar.py new file mode 100644 index 0000000..0c75435 --- /dev/null +++ b/tests/test_g1_dsar.py @@ -0,0 +1,138 @@ +"""Tests for the DSAR worker job (G1 DSGVO: Art. 15/17). + +Proves the previously missing "process_dsar" job works end-to-end: +- access: collects profile + contacts + audit_log + notifications +- deletion: soft-deletes contacts, hard-deletes notifications, anonymizes + and deactivates the account, writes a dsar_erasure audit entry +""" + +from __future__ import annotations + +import uuid + +import pytest +import pytest_asyncio +from sqlalchemy import select as sa_select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from app.core.db import reset_engine_for_testing +from app.core.jobs import _dsar_collect_user_data, _dsar_execute_deletion, process_dsar +from app.models.audit import AuditLog +from app.models.contact import Contact +from app.models.notification import Notification +from app.models.user import User +from tests.conftest import seed_tenant_and_users + + +@pytest_asyncio.fixture +async def dsar_session(engine) -> AsyncSession: + reset_engine_for_testing(engine) + sf = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession) + async with sf() as session: + yield session + await session.rollback() + + +def test_process_dsar_registered(): + """The job is registered so enqueue_job('process_dsar') can find it.""" + from app.core.job_registry import get_job + + assert get_job("process_dsar") is not None + + +@pytest.mark.asyncio +async def test_dsar_access_collects_all_categories(dsar_session: AsyncSession): + """Art. 15: collect_user_data returns all promised categories.""" + seed = await seed_tenant_and_users(dsar_session) + data = await _dsar_collect_user_data( + dsar_session, + str(seed["tenant_a"].id), + str(seed["admin_a"].id), + ) + assert data["legal_basis"] + for category in ("profile", "contacts", "audit_log", "notifications"): + assert category in data["data"], f"missing category {category}" + + +@pytest.mark.asyncio +async def test_dsar_deletion_anonymizes_and_soft_deletes(dsar_session: AsyncSession): + """Art. 17: contacts soft-deleted, notifications gone, user anonymized.""" + seed = await seed_tenant_and_users(dsar_session) + uid = seed["admin_a"].id + tid = seed["tenant_a"].id + + # Fixture: one contact + one notification owned by admin_a + contact = Contact( + id=uuid.uuid4(), tenant_id=tid, owner_id=uid, + type="person", displayname="To Erase", email_1="erase@test.local", + firstname="Era", surname="SeMe", + discount_crew=0, discount_transport=0, discount_rental=0, + discount_sale=0, discount_subrent=0, discount_total=0, + ) + notification = Notification( + id=uuid.uuid4(), tenant_id=tid, owner_id=uid, + user_id=uid, + type="test", + title="to erase", + ) + dsar_session.add_all([contact, notification]) + await dsar_session.commit() + + counts = await _dsar_execute_deletion(dsar_session, str(tid), str(uid)) + await dsar_session.commit() + + assert counts["contacts_soft_deleted"] == 1 + assert counts["notifications_deleted"] == 1 + assert counts["user_anonymized"] == 1 + + # Verify persisted state (fresh query, bypassing identity map) + dsar_session.expunge_all() + erased_contact = ( + await dsar_session.execute(sa_select(Contact).where(Contact.id == contact.id)) + ).scalar_one() + assert erased_contact.deleted_at is not None + + remaining_notifications = ( + await dsar_session.execute( + sa_select(Notification).where(Notification.owner_id == uid) + ) + ).scalars().all() + assert len(remaining_notifications) == 0 + + erased_user = ( + await dsar_session.execute(sa_select(User).where(User.id == uid)) + ).scalar_one() + assert erased_user.is_active is False + assert erased_user.name == "[gelöscht gemäß DSGVO Art. 17]" + assert "anonymized.invalid" in erased_user.email + + # Audit entry written + audits = ( + await dsar_session.execute( + sa_select(AuditLog).where(AuditLog.action == "dsar_erasure") + ) + ).scalars().all() + assert len(audits) >= 1 + + +@pytest.mark.asyncio +async def test_process_dsar_dispatches_by_type(dsar_session: AsyncSession): + """process_dsar dispatches by request_type without raising.""" + seed = await seed_tenant_and_users(dsar_session) + + result = await process_dsar( + {}, + user_id=str(seed["admin_a"].id), + tenant_id=str(seed["tenant_a"].id), + request_type="access", + ) + assert result["status"] == "completed" + assert "categories" in result + + result = await process_dsar( + {}, + user_id=str(seed["editor_a"].id), + tenant_id=str(seed["tenant_a"].id), + request_type="unknown_type_xyz", + ) + assert result["status"] == "unknown_type"