feat: PII Export & Delete Workflow (GDPR-ready) - fixes #76

packages/backend/app/routes/__init__.py

@@ -7,6 +7,7 @@
 from .categories import bp as categories_bp
 from .docs import bp as docs_bp
 from .dashboard import bp as dashboard_bp
+from .privacy import bp as privacy_bp
 
 
 def register_routes(app: Flask):
@@ -18,3 +19,4 @@ def register_routes(app: Flask):
     app.register_blueprint(categories_bp, url_prefix="/categories")
     app.register_blueprint(docs_bp, url_prefix="/docs")
     app.register_blueprint(dashboard_bp, url_prefix="/dashboard")
+    app.register_blueprint(privacy_bp, url_prefix="/privacy")

packages/backend/app/routes/privacy.py

@@ -0,0 +1,191 @@
+from flask import Blueprint, jsonify, request, Response
+from flask_jwt_extended import jwt_required, get_jwt_identity
+from ..extensions import db
+from ..models import User, Category, Expense, RecurringExpense, Bill, Reminder, AuditLog
+import json
+import csv
+import io
+from datetime import datetime
+
+bp = Blueprint("privacy", __name__)
+
+
+def _log_audit(user_id, action):
+    """Record an audit log entry."""
+    entry = AuditLog(user_id=user_id, action=action)
+    db.session.add(entry)
+    db.session.commit()
+
+
+def _collect_user_data(user_id):
+    """Gather all personal data for a user."""
+    user = db.session.get(User, user_id)
+    if not user:
+        return None
+
+    categories = Category.query.filter_by(user_id=user_id).all()
+    expenses = Expense.query.filter_by(user_id=user_id).all()
+    recurring = RecurringExpense.query.filter_by(user_id=user_id).all()
+    bills = Bill.query.filter_by(user_id=user_id).all()
+    reminders = Reminder.query.filter_by(user_id=user_id).all()
+
+    return {
+        "profile": {
+            "id": user.id,
+            "email": user.email,
+            "preferred_currency": user.preferred_currency,
+            "role": user.role,
+            "created_at": user.created_at.isoformat() if user.created_at else None,
+        },
+        "categories": [
+            {"id": c.id, "name": c.name, "created_at": c.created_at.isoformat()}
+            for c in categories
+        ],
+        "expenses": [
+            {
+                "id": e.id,
+                "amount": str(e.amount),
+                "currency": e.currency,
+                "expense_type": e.expense_type,
+                "notes": e.notes,
+                "spent_at": e.spent_at.isoformat() if e.spent_at else None,
+                "category_id": e.category_id,
+                "created_at": e.created_at.isoformat() if e.created_at else None,
+            }
+            for e in expenses
+        ],
+        "recurring_expenses": [
+            {
+                "id": r.id,
+                "amount": str(r.amount),
+                "currency": r.currency,
+                "expense_type": r.expense_type,
+                "notes": r.notes,
+                "cadence": str(r.cadence),
+                "start_date": r.start_date.isoformat() if r.start_date else None,
+                "end_date": r.end_date.isoformat() if r.end_date else None,
+                "active": r.active,
+                "created_at": r.created_at.isoformat() if r.created_at else None,
+            }
+            for r in recurring
+        ],
+        "bills": [
+            {
+                "id": b.id,
+                "name": b.name,
+                "amount": str(b.amount),
+                "currency": b.currency,
+                "next_due_date": b.next_due_date.isoformat() if b.next_due_date else None,
+                "cadence": str(b.cadence),
+                "active": b.active,
+                "created_at": b.created_at.isoformat() if b.created_at else None,
+            }
+            for b in bills
+        ],
+        "reminders": [
+            {
+                "id": r.id,
+                "bill_id": r.bill_id,
+                "message": r.message,
+                "send_at": r.send_at.isoformat() if r.send_at else None,
+                "sent": r.sent,
+                "channel": r.channel,
+            }
+            for r in reminders
+        ],
+        "export_metadata": {
+            "exported_at": datetime.utcnow().isoformat(),
+            "format_version": "1.0",
+            "gdpr_article": "Article 20 - Right to data portability",
+        },
+    }
+
+
+@bp.route("/export", methods=["GET"])
+@jwt_required()
+def export_data():
+    """
+    Export all personal data as JSON (GDPR Article 20 - Right to data portability).
+    Accepts ?format=csv for CSV export.
+    """
+    user_id = get_jwt_identity()
+    data = _collect_user_data(user_id)
+    if not data:
+        return jsonify(error="User not found"), 404
+
+    fmt = request.args.get("format", "json").lower()
+    _log_audit(user_id, f"DATA_EXPORT_REQUESTED:{fmt}")
+
+    if fmt == "csv":
+        output = io.StringIO()
+        writer = csv.writer(output)
+        writer.writerow(["Section", "Field", "Value"])
+        for section, items in data.items():
+            if section == "export_metadata":
+                for k, v in items.items():
+                    writer.writerow([section, k, v])
+            elif section == "profile":
+                for k, v in items.items():
+                    writer.writerow([section, k, v])
+            elif isinstance(items, list):
+                for item in items:
+                    for k, v in item.items():
+                        writer.writerow([section, k, v])
+        csv_data = output.getvalue()
+        return Response(
+            csv_data,
+            mimetype="text/csv",
+            headers={"Content-Disposition": "attachment; filename=finmind-data-export.csv"},
+        )
+
+    return Response(
+        json.dumps(data, indent=2, ensure_ascii=False),
+        mimetype="application/json",
+        headers={"Content-Disposition": "attachment; filename=finmind-data-export.json"},
+    )
+
+
+@bp.route("/delete", methods=["POST"])
+@jwt_required()
+def delete_account():
+    """
+    Permanently delete all user data (GDPR Article 17 - Right to erasure).
+    Requires {"confirm": "DELETE"} in request body.
+    """
+    user_id = get_jwt_identity()
+    body = request.get_json() or {}
+
+    if body.get("confirm") != "DELETE":
+        return jsonify(
+            error="Confirmation required",
+            message='Send {"confirm": "DELETE"} to proceed',
+        ), 400
+
+    user = db.session.get(User, user_id)
+    if not user:
+        return jsonify(error="User not found"), 404
+
+    try:
+        # Delete in correct order (respecting foreign keys)
+        Reminder.query.filter_by(user_id=user_id).delete()
+        Expense.query.filter_by(user_id=user_id).delete()
+        RecurringExpense.query.filter_by(user_id=user_id).delete()
+        Bill.query.filter_by(user_id=user_id).delete()
+        Category.query.filter_by(user_id=user_id).delete()
+
+        # Audit log BEFORE deleting user (so we have the record)
+        _log_audit(user_id, "ACCOUNT_DELETED:all_pii_erased")
+
+        # Finally delete the user
+        db.session.delete(user)
+        db.session.commit()
+
+        return jsonify(
+            status="deleted",
+            message="All personal data has been permanently erased",
+            deleted_at=datetime.utcnow().isoformat(),
+        )
+
+    except Exception as e:
+        db.session.rollback()
+        return jsonify(error=f"Deletion failed: {str(e)}"), 500

packages/backend/tests/test_privacy.py

@@ -0,0 +1,50 @@
+import json
+import pytest
+
+
+class TestPrivacyExport:
+    def test_export_json(self, client, auth_headers):
+        resp = client.get("/privacy/export", headers=auth_headers)
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert "profile" in data
+        assert "expenses" in data
+        assert "categories" in data
+        assert "bills" in data
+        assert "reminders" in data
+        assert "export_metadata" in data
+        assert data["export_metadata"]["gdpr_article"] == "Article 20 - Right to data portability"
+
+    def test_export_csv(self, client, auth_headers):
+        resp = client.get("/privacy/export?format=csv", headers=auth_headers)
+        assert resp.status_code == 200
+        assert resp.content_type == "text/csv; charset=utf-8"
+        assert "Section,Field,Value" in resp.get_data(as_text=True)
+
+    def test_export_requires_auth(self, client):
+        resp = client.get("/privacy/export")
+        assert resp.status_code == 401
+
+
+class TestPrivacyDelete:
+    def test_delete_requires_confirmation(self, client, auth_headers):
+        resp = client.post(
+            "/privacy/delete",
+            headers=auth_headers,
+            json={"confirm": "WRONG"},
+        )
+        assert resp.status_code == 400
+
+    def test_delete_account(self, client, auth_headers):
+        resp = client.post(
+            "/privacy/delete",
+            headers=auth_headers,
+            json={"confirm": "DELETE"},
+        )
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert data["status"] == "deleted"
+
+    def test_delete_requires_auth(self, client):
+        resp = client.post("/privacy/delete", json={"confirm": "DELETE"})
+        assert resp.status_code == 401