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.
This commit is contained in:
Agent Zero
@@ -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"
|
||||
Reference in New Issue
Block a user