Privacy Policy DSGVO with sheets for categories, export-, check- and first erase-functions

This commit is contained in:
2026-08-05 20:34:38 +00:00
parent c9e6438222
commit fc82a28290
19 changed files with 1917 additions and 5 deletions
+458 -1
View File
@@ -51,6 +51,8 @@ from .software_control import token_hash, detect_platform, execute_job, build_re
from .job_state import backfill_asset_job_states, filter_state_key, sync_asset_job_state
from .presence import mesh_presence_loop
from .version import APP_VERSION
from .privacy import merge_privacy_settings, localized_privacy_settings, normalize_category, append_privacy_audit, privacy_audit_tail, PROTECTED_PRIVACY_CATEGORY_KEYS
from .privacy_retention import check_retention_category, delete_retention_category, append_deletion_audit, deletion_audit_tail, IMPLEMENTED_RETENTION_KEYS
from .backup import (BACKUP_DIR, BACKUP_INTERVAL_HOURS, BACKUP_RETENTION_DAYS, backup_path, create_backup, delete_backup, list_backups, restore_backup, store_uploaded_backup, automatic_backup_loop, system_storage_information)
from openpyxl import Workbook, load_workbook
from openpyxl.styles import Font, PatternFill, Alignment
@@ -2220,6 +2222,14 @@ def _seed_default_charts(db: Session) -> None:
@app.get("/")
def dashboard(request: Request, db: Session = Depends(get_db)):
categories = db.query(Category).options(joinedload(Category.visible_fields)).all()
category_asset_counts = dict(
_apply_asset_access(
db.query(Asset.category_id, func.count(Asset.id)),
request,
)
.group_by(Asset.category_id)
.all()
)
assets = (
_apply_asset_access(db.query(Asset), request)
.options(joinedload(Asset.category).joinedload(Category.visible_fields))
@@ -2239,7 +2249,7 @@ def dashboard(request: Request, db: Session = Depends(get_db)):
dashboard_charts = [_chart_payload(chart, chart_assets) for chart in db.query(ChartDefinition).filter(ChartDefinition.dashboard_visible.is_(True)).order_by(ChartDefinition.id).all()]
return templates.TemplateResponse(
"dashboard.html",
{"request": request, "categories": categories, "assets": assets, "fields": fields, "is_admin": _is_admin(request), "issue_status_name": (_action_status(db, "issue").name if _action_status(db, "issue") else None), "dashboard_charts": dashboard_charts},
{"request": request, "categories": categories, "category_asset_counts": category_asset_counts, "assets": assets, "fields": fields, "is_admin": _is_admin(request), "issue_status_name": (_action_status(db, "issue").name if _action_status(db, "issue") else None), "dashboard_charts": dashboard_charts},
)
@@ -7147,6 +7157,453 @@ def settings_import_export_page(request: Request):
)
@app.get("/settings/privacy")
def settings_privacy_page(request: Request):
config = load_config()
privacy_settings = localized_privacy_settings(config.get("privacy"), _request_language(request))
preview_results = request.session.pop("privacy_retention_preview", {})
return templates.TemplateResponse(
"settings_privacy.html",
{
"request": request,
"privacy_settings": privacy_settings,
"privacy_audit": privacy_audit_tail(50),
"deletion_audit": deletion_audit_tail(50),
"can_edit": _is_admin(request),
"implemented_retention_keys": IMPLEMENTED_RETENTION_KEYS,
"preview_results": preview_results if isinstance(preview_results, dict) else {},
},
)
@app.post("/settings/privacy")
async def settings_privacy_save(request: Request):
_require_admin(request)
form = await request.form()
current = merge_privacy_settings(load_config().get("privacy"))
activity = dict(current.get("processing_activity", {}))
text_fields = (
"title", "responsible_department", "responsible_people", "controller",
"data_protection_contact", "legal_basis", "data_subject_groups",
"internal_recipients", "external_recipients", "third_country_details",
"systems", "backup_system", "restore_deletion_procedure",
"last_reviewed_at", "approved_by", "document_version",
)
for field in text_fields:
activity[field] = str(form.get(field, "") or "").strip()[:8000]
activity["third_country_transfer"] = form.get("third_country_transfer") == "on"
activity["backup_retention_days"] = _optional_form_int(
form.get("backup_retention_days"), field_name="backup_retention_days", minimum=0, maximum=36500
)
activity["review_interval_months"] = _optional_form_int(
form.get("review_interval_months"), field_name="review_interval_months", minimum=1, maximum=120
) or 12
activity["purposes"] = [str(value).strip()[:2000] for value in form.getlist("purposes") if str(value).strip()]
activity["toms"] = [str(value).strip()[:3000] for value in form.getlist("toms") if str(value).strip()]
category_keys = form.getlist("category_key")
categories = []
for index, key in enumerate(category_keys):
def value(name: str, default: str = "") -> str:
values = form.getlist(name)
return str(values[index] if index < len(values) else default)
category = normalize_category({
"key": value("category_key", f"custom_{index + 1}"),
"name": value("category_name"),
"system": value("category_system"),
"data_subjects": value("category_data_subjects"),
"personal_data": value("category_personal_data"),
"purpose": value("category_purpose"),
"retention_type": value("category_retention_type", "manual"),
"retention_days": value("category_retention_days"),
"retention_trigger": value("category_retention_trigger"),
"automatic_possible": value("category_automatic_possible") == "1",
"manual_procedure": value("category_manual_procedure"),
"backup_handling": value("category_backup_handling"),
"responsible": value("category_responsible"),
"notes": value("category_notes"),
}, index)
if category["name"]:
categories.append(category)
# System categories are part of executable retention handlers and cannot be removed.
submitted_keys = {str(item.get("key") or "") for item in categories}
current_by_key = {str(item.get("key") or ""): item for item in current.get("categories", [])}
default_by_key = {str(item.get("key") or ""): item for item in merge_privacy_settings(None).get("categories", [])}
for protected_key in PROTECTED_PRIVACY_CATEGORY_KEYS:
if protected_key not in submitted_keys:
restored = current_by_key.get(protected_key) or default_by_key.get(protected_key)
if restored:
categories.append(normalize_category(restored, len(categories)))
updated = {
"processing_activity": activity,
"categories": categories,
"phase": 2,
"automatic_deletion_available": False,
}
save_config({"privacy": updated})
append_privacy_audit(
"privacy_policy_updated",
_changed_by(request) or "admin",
{"categories": len(categories), "purposes": len(activity["purposes"]), "toms": len(activity["toms"])},
)
return RedirectResponse(
"/settings/privacy?toast_success=" + quote(_translate_request(request, "privacy.saved", "Privacy settings saved.")),
status_code=303,
)
def _privacy_category_by_key(category_key: str) -> dict[str, Any]:
settings = merge_privacy_settings(load_config().get("privacy"))
for category in settings.get("categories", []):
if str(category.get("key") or "") == category_key:
return category
raise HTTPException(404, "Privacy category not found.")
@app.post("/settings/privacy/retention/{category_key}/check")
def settings_privacy_retention_check(request: Request, category_key: str):
_require_admin(request)
category = _privacy_category_by_key(category_key)
result = check_retention_category(category_key, category.get("retention_days"))
result["category_name"] = category.get("name") or category_key
request.session["privacy_retention_preview"] = {category_key: result}
append_deletion_audit(
"retention_check",
_changed_by(request) or "admin",
category_key=category_key,
retention_days=category.get("retention_days"),
result=result,
)
message_key = "privacy.check_completed" if result.get("supported") else "privacy.check_not_supported"
fallback = "Retention check completed." if result.get("supported") else "No executable retention handler is available for this category."
return RedirectResponse(
"/settings/privacy?toast_success=" + quote(_translate_request(request, message_key, fallback)) + "#privacy-category-list",
status_code=303,
)
@app.post("/settings/privacy/retention/{category_key}/delete")
async def settings_privacy_retention_delete(request: Request, category_key: str):
_require_admin(request)
form = await request.form()
confirmation = str(form.get("confirmation") or "").strip().upper()
if confirmation != "DELETE":
return RedirectResponse(
"/settings/privacy?toast_error=" + quote(_translate_request(request, "privacy.delete_confirmation_invalid", "Enter DELETE to confirm the deletion run.")),
status_code=303,
)
category = _privacy_category_by_key(category_key)
try:
result = delete_retention_category(category_key, category.get("retention_days"))
except ValueError as exc:
return RedirectResponse(
"/settings/privacy?toast_error=" + quote(str(exc)),
status_code=303,
)
result["category_name"] = category.get("name") or category_key
append_deletion_audit(
"manual_deletion",
_changed_by(request) or "admin",
category_key=category_key,
retention_days=category.get("retention_days"),
result=result,
)
append_privacy_audit(
"privacy_manual_deletion_executed",
_changed_by(request) or "admin",
{
"category_key": category_key,
"retention_days": category.get("retention_days"),
"records": result.get("deleted_or_anonymised_records", 0),
"status": result.get("status"),
},
)
return RedirectResponse(
"/settings/privacy?toast_success=" + quote(_translate_request(request, "privacy.delete_completed", "Manual deletion run completed.")) + "#privacy-category-list",
status_code=303,
)
def _privacy_report_text(settings: dict[str, Any], language_code: str = "en") -> str:
activity = settings["processing_activity"]
de = str(language_code or "").lower().startswith("de")
labels = {
"title": "VERZEICHNIS VON VERARBEITUNGSTÄTIGKEITEN / LÖSCHKONZEPT" if de else "RECORD OF PROCESSING ACTIVITIES / RETENTION CONCEPT",
"document_version": "Dokumentversion" if de else "Document version",
"activity": "Verarbeitungstätigkeit" if de else "Processing activity",
"controller": "Verantwortlicher" if de else "Controller",
"department": "Verantwortlicher Fachbereich" if de else "Responsible department",
"people": "Verantwortliche Personen" if de else "Responsible people",
"dpo": "Datenschutzkontakt" if de else "Data protection contact",
"purposes": "1. ZWECKE" if de else "1. PURPOSES",
"legal": "2. RECHTSGRUNDLAGE" if de else "2. LEGAL BASIS",
"subjects": "3. BETROFFENE PERSONENGRUPPEN" if de else "3. DATA SUBJECT GROUPS",
"recipients": "4. EMPFÄNGER" if de else "4. RECIPIENTS",
"internal": "Intern" if de else "Internal",
"external": "Extern" if de else "External",
"third": "5. DRITTLANDÜBERMITTLUNGEN" if de else "5. THIRD-COUNTRY TRANSFERS",
"intended": "Übermittlung vorgesehen" if de else "Transfer intended",
"yes": "Ja" if de else "Yes", "no": "Nein" if de else "No",
"systems": "6. SYSTEME UND SICHERUNGEN" if de else "6. SYSTEMS AND BACKUPS",
"system_label": "Systeme" if de else "Systems",
"backup_system": "Backup-System" if de else "Backup system",
"backup_retention": "Backup-Aufbewahrung" if de else "Backup retention",
"restore": "Verfahren nach Wiederherstellung" if de else "Restore procedure",
"categories": "7. DATENKATEGORIEN UND LÖSCHFRISTEN" if de else "7. DATA CATEGORIES AND RETENTION",
"system": "System", "data_subjects": "Betroffene Personen" if de else "Data subjects",
"personal_data": "Personenbezogene Daten" if de else "Personal data",
"purpose": "Zweck" if de else "Purpose", "retention_type": "Löschart" if de else "Retention type",
"retention_period": "Aufbewahrungsfrist" if de else "Retention period",
"trigger": "Fristbeginn/Auslöser" if de else "Trigger",
"automatic_possible": "Automatische Löschung technisch möglich" if de else "Automatic deletion technically possible",
"automatic_active": "Automatische Löschung aktiv: Nein; Prüfläufe und manuelle Löschung für ausgewählte Kategorien verfügbar" if de else "Automatic deletion active: No; checks and manual deletion are available for selected categories",
"manual": "Manuelles Verfahren" if de else "Manual procedure",
"backup": "Behandlung in Sicherungen" if de else "Backup handling",
"responsible": "Verantwortlich" if de else "Responsible", "notes": "Hinweise" if de else "Notes",
"toms": "8. TECHNISCHE UND ORGANISATORISCHE MASSNAHMEN" if de else "8. TECHNICAL AND ORGANISATIONAL MEASURES",
"review": "9. DOKUMENTENPRÜFUNG" if de else "9. DOCUMENT REVIEW",
"review_interval": "Prüfintervall" if de else "Review interval",
"last_reviewed": "Zuletzt geprüft" if de else "Last reviewed",
"approved": "Freigegeben durch" if de else "Approved by",
"notice": "HINWEIS" if de else "NOTICE",
"notice_text": f"Dieser Export dokumentiert konfigurierte Löschregeln. Version {APP_VERSION} führt keine automatische Löschung aus; Prüfläufe und manuell bestätigte Löschläufe sind für ausgewählte Kategorien verfügbar." if de else f"This export documents configured retention rules. Version {APP_VERSION} does not execute automatic deletion; checks and manually confirmed deletion runs are available for selected categories.",
}
lines = [
labels["title"],
"=" * 64,
f"{labels['document_version']}: {activity.get('document_version') or '-'}",
f"{labels['activity']}: {activity.get('title') or '-'}",
f"{labels['controller']}: {activity.get('controller') or '-'}",
f"{labels['department']}: {activity.get('responsible_department') or '-'}",
f"{labels['people']}: {activity.get('responsible_people') or '-'}",
f"{labels['dpo']}: {activity.get('data_protection_contact') or '-'}",
"",
labels["purposes"],
]
lines.extend(f"- {item}" for item in activity.get("purposes", []))
lines.extend([
"", labels["legal"], activity.get("legal_basis") or "-",
"", labels["subjects"], activity.get("data_subject_groups") or "-",
"", labels["recipients"],
f"{labels['internal']}: {activity.get('internal_recipients') or '-'}",
f"{labels['external']}: {activity.get('external_recipients') or '-'}",
"", labels["third"],
f"{labels['intended']}: {labels['yes'] if activity.get('third_country_transfer') else labels['no']}",
activity.get("third_country_details") or "-",
"", labels["systems"],
f"{labels['system_label']}: {activity.get('systems') or '-'}",
f"{labels['backup_system']}: {activity.get('backup_system') or '-'}",
f"{labels['backup_retention']}: {activity.get('backup_retention_days') if activity.get('backup_retention_days') is not None else '-'} Tage" if de else f"{labels['backup_retention']}: {activity.get('backup_retention_days') if activity.get('backup_retention_days') is not None else '-'} days",
f"{labels['restore']}: {activity.get('restore_deletion_procedure') or '-'}",
"", labels["categories"],
])
for number, category in enumerate(settings.get("categories", []), 1):
lines.extend([
"", f"7.{number} {category.get('name')}",
f"{labels['system']}: {category.get('system') or '-'}",
f"{labels['data_subjects']}: {category.get('data_subjects') or '-'}",
f"{labels['personal_data']}: {category.get('personal_data') or '-'}",
f"{labels['purpose']}: {category.get('purpose') or '-'}",
f"{labels['retention_type']}: {category.get('retention_type') or '-'}",
f"{labels['retention_period']}: {category.get('retention_days') if category.get('retention_days') is not None else '-'} Tage" if de else f"{labels['retention_period']}: {category.get('retention_days') if category.get('retention_days') is not None else '-'} days",
f"{labels['trigger']}: {category.get('retention_trigger') or '-'}",
f"{labels['automatic_possible']}: {labels['yes'] if category.get('automatic_possible') else labels['no']}",
labels["automatic_active"],
f"{labels['manual']}: {category.get('manual_procedure') or '-'}",
f"{labels['backup']}: {category.get('backup_handling') or '-'}",
f"{labels['responsible']}: {category.get('responsible') or '-'}",
f"{labels['notes']}: {category.get('notes') or '-'}",
])
lines.extend(["", labels["toms"]])
lines.extend(f"- {item}" for item in activity.get("toms", []))
lines.extend([
"", labels["review"],
f"{labels['review_interval']}: {activity.get('review_interval_months') or 12} Monate" if de else f"{labels['review_interval']}: {activity.get('review_interval_months') or 12} months",
f"{labels['last_reviewed']}: {activity.get('last_reviewed_at') or '-'}",
f"{labels['approved']}: {activity.get('approved_by') or '-'}",
"", labels["notice"],
labels["notice_text"],
])
return "\n".join(lines) + "\n"
@app.get("/settings/privacy/export.txt")
def settings_privacy_export_text(request: Request):
settings = localized_privacy_settings(load_config().get("privacy"), _request_language(request))
append_privacy_audit("privacy_report_exported", _changed_by(request) or "authenticated-user", {"format": "text"})
return Response(
_privacy_report_text(settings, _request_language(request)),
media_type="text/plain; charset=utf-8",
headers={"Content-Disposition": "attachment; filename=assetmanager-privacy-processing-record.txt"},
)
@app.get("/settings/privacy/export.html")
def settings_privacy_export_html(request: Request):
language_code = _request_language(request)
settings = localized_privacy_settings(load_config().get("privacy"), language_code)
rendered = templates.env.get_template("privacy_report_export.html").render(
settings=settings,
activity=settings["processing_activity"],
privacy_audit=list(reversed(privacy_audit_tail(1000))),
language_code=language_code,
app_version=APP_VERSION,
generated_at=datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S %Z"),
)
payload = rendered.encode("utf-8")
append_privacy_audit("privacy_report_exported", _changed_by(request) or "authenticated-user", {"format": "html"})
return Response(
content=payload,
media_type="text/html; charset=utf-8",
headers={
"Content-Disposition": "attachment; filename=assetmanager-privacy-processing-record.html",
"Content-Length": str(len(payload)),
"X-Content-Type-Options": "nosniff",
"Cache-Control": "no-store",
},
)
def _privacy_excel_value(value: Any) -> Any:
"""Return a safe Excel cell value without control characters or oversized text."""
if value is None:
return ""
if isinstance(value, (bool, int, float, date, datetime)):
return value
if isinstance(value, (dict, list)):
value = json.dumps(value, ensure_ascii=False, sort_keys=True)
text_value = str(value)
text_value = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F]", "", text_value)
return text_value[:32767]
def _style_privacy_workbook(workbook: Workbook) -> None:
header_fill = PatternFill("solid", fgColor="D9EAF7")
for sheet in workbook.worksheets:
sheet.freeze_panes = "A2"
sheet.auto_filter.ref = sheet.dimensions
for cell in sheet[1]:
cell.font = Font(bold=True)
cell.fill = header_fill
cell.alignment = Alignment(vertical="top", wrap_text=True)
for column in sheet.columns:
width = min(max((len(str(cell.value or "")) for cell in column), default=10) + 2, 60)
sheet.column_dimensions[column[0].column_letter].width = max(width, 12)
for cell in column:
cell.alignment = Alignment(vertical="top", wrap_text=True)
@app.get("/settings/privacy/export.xlsx")
def settings_privacy_export_excel(request: Request):
language_code = _request_language(request)
de = str(language_code or "").lower().startswith("de")
settings = localized_privacy_settings(load_config().get("privacy"), language_code)
activity = settings["processing_activity"]
workbook = Workbook()
overview = workbook.active
overview.title = "Übersicht" if de else "Overview"
overview.append(["Feld" if de else "Field", "Wert" if de else "Value"])
overview_fields = (
(("Verarbeitungstätigkeit" if de else "Processing activity"), "title"),
(("Verantwortlicher" if de else "Controller"), "controller"),
(("Verantwortlicher Fachbereich" if de else "Responsible department"), "responsible_department"),
(("Verantwortliche Personen" if de else "Responsible people"), "responsible_people"),
(("Datenschutzkontakt" if de else "Data protection contact"), "data_protection_contact"),
(("Rechtsgrundlage" if de else "Legal basis"), "legal_basis"),
(("Betroffene Personengruppen" if de else "Data subject groups"), "data_subject_groups"),
(("Interne Empfänger" if de else "Internal recipients"), "internal_recipients"),
(("Externe Empfänger" if de else "External recipients"), "external_recipients"),
(("Drittlandübermittlung" if de else "Third-country transfer"), "third_country_transfer"),
(("Details zur Drittlandübermittlung" if de else "Third-country details"), "third_country_details"),
(("Systeme" if de else "Systems"), "systems"),
(("Backup-System" if de else "Backup system"), "backup_system"),
(("Backup-Aufbewahrungstage" if de else "Backup retention days"), "backup_retention_days"),
(("Löschverfahren nach Wiederherstellung" if de else "Restore deletion procedure"), "restore_deletion_procedure"),
(("Prüfintervall in Monaten" if de else "Review interval months"), "review_interval_months"),
(("Zuletzt geprüft" if de else "Last reviewed"), "last_reviewed_at"),
(("Freigegeben durch" if de else "Approved by"), "approved_by"),
(("Dokumentversion" if de else "Document version"), "document_version"),
(("AssetManager-Version" if de else "AssetManager version"), None),
)
for label, key in overview_fields:
value = APP_VERSION if key is None else activity.get(key, "")
if isinstance(value, bool):
value = ("Ja" if value else "Nein") if de else ("Yes" if value else "No")
overview.append([_privacy_excel_value(label), _privacy_excel_value(value)])
purposes = workbook.create_sheet("Zwecke" if de else "Purposes")
purposes.append(["Nr." if de else "No.", "Zweck" if de else "Purpose"])
for index, value in enumerate(activity.get("purposes", []), 1):
purposes.append([index, _privacy_excel_value(value)])
categories = workbook.create_sheet("Datenkategorien" if de else "Data categories")
category_headers = [
"Name", "System", "Betroffene Personen" if de else "Data subjects",
"Personenbezogene Daten" if de else "Personal data", "Zweck" if de else "Purpose",
"Löschart" if de else "Retention type", "Löschfrist (Tage)" if de else "Retention days",
"Fristbeginn/Auslöser" if de else "Retention trigger",
"Automatisch technisch möglich" if de else "Automatic possible",
"Automatische Löschung aktiv" if de else "Automatic active",
"Manuelles Verfahren" if de else "Manual procedure",
"Behandlung in Sicherungen" if de else "Backup handling",
"Verantwortlich" if de else "Responsible", "Hinweise" if de else "Notes",
]
categories.append(category_headers)
for item in settings.get("categories", []):
categories.append([_privacy_excel_value(value) for value in [
item.get("name"), item.get("system"), item.get("data_subjects"), item.get("personal_data"),
item.get("purpose"), item.get("retention_type"), item.get("retention_days"), item.get("retention_trigger"),
("Ja" if item.get("automatic_possible") else "Nein") if de else ("Yes" if item.get("automatic_possible") else "No"),
"Nein - manuell für ausgewählte Kategorien" if de else "No - manual for selected categories", item.get("manual_procedure"),
item.get("backup_handling"), item.get("responsible"), item.get("notes"),
]])
toms = workbook.create_sheet("TOMs")
toms.append(["Nr." if de else "No.", "Technische und organisatorische Maßnahme" if de else "Technical and organisational measure"])
for index, value in enumerate(activity.get("toms", []), 1):
toms.append([index, _privacy_excel_value(value)])
audit = workbook.create_sheet("Änderungsprotokoll" if de else "Policy audit")
audit.append([
"Zeitstempel UTC" if de else "Timestamp UTC", "Aktion" if de else "Action",
"Geändert von" if de else "Changed by", "Details",
])
for row in reversed(privacy_audit_tail(1000)):
audit.append([_privacy_excel_value(value) for value in [
row.get("timestamp_utc"), row.get("action"), row.get("changed_by"), row.get("details", {}),
]])
_style_privacy_workbook(workbook)
output = io.BytesIO()
workbook.save(output)
payload = output.getvalue()
# Verify the generated archive before sending it to the browser.
try:
load_workbook(io.BytesIO(payload), read_only=True, data_only=True).close()
except Exception as exc:
logging.exception("Privacy XLSX validation failed")
raise HTTPException(500, f"Privacy Excel export could not be validated: {exc}") from exc
append_privacy_audit("privacy_report_exported", _changed_by(request) or "authenticated-user", {"format": "xlsx"})
filename = "assetmanager-datenschutz-verarbeitungstaetigkeit.xlsx" if de else "assetmanager-privacy-processing-record.xlsx"
disposition = f"attachment; filename={filename}; filename*=UTF-8''{quote(filename)}"
return Response(
content=payload,
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
headers={
"Content-Disposition": disposition,
"Content-Length": str(len(payload)),
"X-Content-Type-Options": "nosniff",
"Cache-Control": "no-store",
},
)
@app.get("/settings/software")
def settings_software_page(request: Request):
_require_admin(request)