GDPR Compliance Check for Your AI Stack

# Record of Processing Activities

| Data Type | Purpose | Legal Basis | Retention | Recipients |
|-----------|---------|-------------|-----------|------------|
| User feedback text | Summarization | Legitimate interest | 30 days | Internal team |
| Email address | Notifications | Consent | Until unsubscribe | n8n mail service |
| Chat history | Model improvement | Contract | 90 days | Your-server storage |
| API logs | Security audit | Legal obligation | 1 year | Security team |

Description: Users submit questions via webhook, Ollama responds, conversation stored.

Data: User question text, generated response, timestamp, user IP (optional)

Risk: If database breached, questions exposed (may contain sensitive info)

Controls:
  - Database encrypted at rest (SQLite with encryption)
  - Access limited to admins only
  - Retention: 30 days, then auto-delete
  - No backup of personally identifiable data

Residual Risk: Low (controls are strong)

# PostgreSQL with encryption
docker run -e POSTGRES_PASSWORD=strong_password \
  -v /path/to/pgdata:/var/lib/postgresql/data \
  postgres:15

# Enable pg_tde (transparent data encryption) for production
# Or use dm-crypt on Linux:
sudo cryptsetup luksFormat /dev/sda1

# docker-compose.yml
services:
  n8n:
    environment:
      - N8N_PROTOCOL=https
      - WEBHOOK_URL=https://your-domain.com  # Not http://

# Example: Only admins can access user chat data
from functools import wraps
from flask import abort, request

def admin_only(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        if request.user.role != 'admin':
            abort(403, "Forbidden")
        return f(*args, **kwargs)
    return decorated

@app.route('/api/chat-history')
@admin_only
def get_chat_history():
    # Only admins see this
    return database.query("SELECT * FROM chats")

# Log who accesses what
n8n:
  environment:
    - N8N_LOGGER_LEVEL=debug
    - LOG_FORMAT=json  # Structured logging

# Audit log example
{
  "timestamp": "2026-03-21T10:15:00Z",
  "user": "[email protected]",
  "action": "accessed_user_chat_history",
  "user_id": "user_123",
  "result": "success"
}

{
  "personal_data": {
    "email": "[email protected]",
    "feedback_submissions": [
      {
        "text": "Great product",
        "date": "2026-03-20",
        "processed": "sent to analysis"
      }
    ],
    "chat_history": [
      {
        "question": "How to use feature X?",
        "date": "2026-03-19",
        "model_used": "mistral"
      }
    ]
  }
}

@app.route('/api/user/<user_id>/data-export', methods=['GET'])
@authenticated
def export_user_data(user_id):
    """GDPR Article 15: Right to access"""
    if request.user.id != user_id:
        abort(403, "Can only access your own data")

    data = {
        "personal_data": database.query(
            "SELECT * FROM users WHERE id = ?", user_id
        ),
        "feedback": database.query(
            "SELECT * FROM feedback WHERE user_id = ?", user_id
        ),
        "export_date": datetime.now().isoformat()
    }

    return jsonify(data)

@app.route('/api/user/<user_id>/delete', methods=['POST'])
@authenticated
def delete_user_data(user_id):
    """GDPR Article 17: Right to deletion"""
    if request.user.id != user_id:
        abort(403)

    # Delete from all tables
    database.execute("DELETE FROM chats WHERE user_id = ?", user_id)
    database.execute("DELETE FROM feedback WHERE user_id = ?", user_id)
    database.execute("DELETE FROM users WHERE id = ?", user_id)

    # Log deletion for audit
    log_audit("user_deleted", user_id=user_id, timestamp=now())

    return {"status": "deleted"}

@app.route('/api/user/<user_id>/update', methods=['PUT'])
@authenticated
def update_user_data(user_id):
    if request.user.id != user_id:
        abort(403)

    updates = request.json
    database.execute(
        "UPDATE users SET email = ? WHERE id = ?",
        updates["email"], user_id
    )
    return {"status": "updated"}

@app.route('/api/user/<user_id>/export-json', methods=['GET'])
@authenticated
def export_json(user_id):
    if request.user.id != user_id:
        abort(403)

    data = {
        "user": database.query("SELECT * FROM users WHERE id = ?", user_id),
        "feedback": database.query("SELECT * FROM feedback WHERE user_id = ?", user_id),
        "format": "JSON",
        "exported_at": datetime.now().isoformat()
    }

    response = make_response(json.dumps(data, indent=2))
    response.headers["Content-Disposition"] = f"attachment; filename=user_{user_id}_data.json"
    return response

Data Type                     | Retention | Reason
------------------------------|-----------|----------------------------------
Chat messages                 | 30 days   | Support purposes only
Error logs                     | 90 days   | Debugging, then anonymize
User authentication logs       | 1 year    | Security audit trail
Payment records               | 7 years   | Tax/accounting requirement
Model training data           | Until     | With consent, for improvement
                              | deletion  | (user can revoke)
Anonymized analytics         | Unlimited | Cannot identify individuals

# Cron job to auto-delete old data
import schedule
import time

def cleanup_old_data():
    # Delete chats older than 30 days
    database.execute("""
        DELETE FROM chats
        WHERE created_at < datetime('now', '-30 days')
    """)

    # Anonymize error logs older than 90 days
    database.execute("""
        UPDATE logs
        SET user_id = NULL, ip_address = NULL
        WHERE created_at < datetime('now', '-90 days')
    """)

schedule.every().day.at("02:00").do(cleanup_old_data)

# Privacy Notice

We collect the following personal data:
- Email address (to send notifications)
- Chat message text (to improve the model)
- Usage logs (for debugging)

**Legal basis:** Your consent (you agree to this notice)

**How long we keep it:**
- Chat messages: 30 days
- Email: until you unsubscribe
- Logs: 90 days

**Your rights:**
- Access your data: /api/my-data
- Delete your data: /api/delete-my-data
- Export your data: /api/export-my-data
- Contact: [email protected]

**Third parties:** We don't share data with anyone.

**Cookies:** We use session cookies for login only.

**Changes:** We'll notify you 30 days before privacy policy changes.

def notify_breach(user_ids, breach_type, severity):
    """Notify users of data breach"""

    # Log breach
    breach_id = log_breach(
        type=breach_type,
        severity=severity,
        affected_count=len(user_ids),
        timestamp=datetime.now()
    )

    # Notify users
    for user_id in user_ids:
        user = database.query("SELECT email FROM users WHERE id = ?", user_id)
        send_email(
            to=user.email,
            subject="Security Notice: Your data may have been exposed",
            body=f"""
We discovered a security incident on {date}.
Your data may have been affected.

What we found: {breach_type}
What we did: {remediation_steps}

Learn more: https://your-site.com/security/{breach_id}
            """
        )

    # Notify DPA
    submit_to_dpa(
        incident_id=breach_id,
        description=breach_type,
        affected_count=len(user_ids)
    )