GDPR-Ready SaaS Architecture: What You Actually Need to Build

# utils/encryption.py
from cryptography.fernet import Fernet, MultiFernet
import os

def get_fernet() -> MultiFernet:
    """
    MultiFernet supports key rotation without re-encrypting existing data.
    PRIMARY_KEY is active for encryption.
    SECONDARY_KEY (optional) is used for decrypting data encrypted with the old key.
    """
    primary = Fernet(os.environ["ENCRYPTION_KEY_PRIMARY"])
    keys = [primary]

    secondary_raw = os.environ.get("ENCRYPTION_KEY_SECONDARY")
    if secondary_raw:
        keys.append(Fernet(secondary_raw))

    return MultiFernet(keys)

def encrypt(value: str) -> str:
    return get_fernet().encrypt(value.encode()).decode()

def decrypt(value: str) -> str:
    return get_fernet().decrypt(value.encode()).decode()

# models/user.py
from sqlalchemy import event
from sqlalchemy.orm import reconstructor

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    _national_id_encrypted: Mapped[Optional[str]] = mapped_column("national_id", String(500))

    @property
    def national_id(self) -> Optional[str]:
        if self._national_id_encrypted is None:
            return None
        return decrypt(self._national_id_encrypted)

    @national_id.setter
    def national_id(self, value: Optional[str]) -> None:
        self._national_id_encrypted = encrypt(value) if value else None

# models/audit_log.py
class AuditEventType(str, enum.Enum):
    # Auth events
    USER_LOGIN = "user.login"
    USER_LOGIN_FAILED = "user.login_failed"
    USER_LOGOUT = "user.logout"
    USER_2FA_ENABLED = "user.2fa_enabled"

    # Data access
    USER_DATA_ACCESSED = "user.data_accessed"
    USER_DATA_EXPORTED = "user.data_exported"
    USER_DATA_DELETED = "user.data_deleted"

    # Admin actions
    ADMIN_USER_VIEWED = "admin.user_viewed"
    ADMIN_ROLE_CHANGED = "admin.role_changed"

class AuditLog(Base):
    __tablename__ = "audit_logs"

    id: Mapped[int] = mapped_column(primary_key=True)
    event_type: Mapped[AuditEventType] = mapped_column(nullable=False)
    actor_user_id: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id"))
    target_user_id: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id"))
    ip_address: Mapped[Optional[str]] = mapped_column(String(45))  # IPv6 max length
    user_agent: Mapped[Optional[str]] = mapped_column(String(500))
    metadata: Mapped[Optional[dict]] = mapped_column(JSONB)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow, index=True)

# services/gdpr.py
async def erase_user_data(db: AsyncSession, user_id: int, actor_id: int) -> None:
    """
    GDPR right to erasure implementation.

    We anonymise rather than delete where foreign keys require it
    (e.g., audit logs referencing this user must be kept for compliance,
    but the personal data within them can be removed).
    """
    user = await db.get(User, user_id)
    if not user:
        raise ValueError(f"User {user_id} not found")

    # Anonymise PII fields (keep the record for referential integrity)
    user.email = f"deleted_{user_id}@anonymised.invalid"
    user.full_name = "Deleted User"
    user.national_id = None
    user.phone = None
    user.is_deleted = True
    user.deleted_at = datetime.utcnow()

    # Revoke all active sessions
    await db.execute(
        update(RefreshToken)
        .where(RefreshToken.user_id == user_id)
        .values(used=True)
    )

    # Log the erasure (this itself is a compliance record)
    db.add(AuditLog(
        event_type=AuditEventType.USER_DATA_DELETED,
        actor_user_id=actor_id,
        target_user_id=user_id,
        metadata={"reason": "gdpr_erasure_request"},
    ))

    await db.commit()

@router.get("/me/export")
async def export_my_data(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_async_db),
):
    """
    GDPR Article 20 — right to data portability.
    Returns all personal data in a structured JSON format.
    """
    await audit_log(db, AuditEventType.USER_DATA_EXPORTED, actor=current_user)

    return {
        "export_date": datetime.utcnow().isoformat(),
        "data_controller": "uguraslim.com",
        "user": {
            "id": current_user.id,
            "email": current_user.email,
            "full_name": current_user.full_name,
            "created_at": current_user.created_at.isoformat(),
        },
        "activity_summary": await get_user_activity_summary(db, current_user.id),
    }