feat(g1): DSGVO Art.15/17 funktionsfaehig — fehlender process_dsar Worker-Job implementiert

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.
2026-08-25 23:48:22 +02:00
parent 38b73f5d4d
commit f4a5937a4b
2 changed files with 413 additions and 0 deletions
@@ -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)
@@ -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"