Module 6: Data Privacy & PII Protection

6. Retention Policies and Encryption

Next →

Overview

You've built a pipeline that detects PII (capsule 03), redacts it (capsule 04), and minimizes the data sent to the LLM (capsule 05). But there's a question we haven't addressed yet: what happens to the data after processing?

Your AI system's logs record every request. The model's outputs are stored for auditing. RAG documents persist in your vector store. Reversible redaction mappings contain the original data. Chat conversations are kept for future context. Each of these artifacts is a storage point for potentially sensitive data — and each one needs a retention and protection policy.

In this capsule you build two components: a Retention Scheduler that implements automatic deletion policies, and the encryption practices needed to protect the data while it exists. Together, these components close the data lifecycle in your AI system.

Why retention policies for AI?

In a traditional web application, logs are rotated and data is archived. In an AI system, retention has additional considerations:

ai_retention_challenges = {
    "llm_logs": {
        "what": "Full prompts + LLM responses",
        "risk": "They contain the full context including any PII that wasn't redacted",
        "retention_consideration": "Do you really need to keep the full prompt?",
        "recommended_retention": "30 days max, ideally with PII redacted",
    },
    "chat_history": {
        "what": "Users' multi-turn conversations",
        "risk": "Users reveal personal information during conversations",
        "retention_consideration": "Do you need the full history or just a summary?",
        "recommended_retention": "90 days, then summarize and delete the raw",
    },
    "rag_documents": {
        "what": "Documents indexed in the vector store",
        "risk": "Documents can contain PII of customers, employees, or partners",
        "retention_consideration": "Are the source documents still valid?",
        "recommended_retention": "Based on the source document's lifecycle",
    },
    "redaction_mappings": {
        "what": "Placeholder → original value mappings from reversible redaction",
        "risk": "They contain the original unredacted data",
        "retention_consideration": "They should only exist during processing",
        "recommended_retention": "Immediate deletion post-response",
    },
    "embeddings": {
        "what": "Embedding vectors of processed texts",
        "risk": "Embeddings can be partially reversible",
        "retention_consideration": "Do the embeddings correspond to personal data?",
        "recommended_retention": "Aligned with the source text's retention",
    },
    "audit_logs": {
        "what": "Records of PII detection and redaction",
        "risk": "They can reveal what data was processed (metadata)",
        "retention_consideration": "Necessary for compliance, but minimize metadata",
        "recommended_retention": "1-3 years depending on the applicable regulation",
    },
}

print("Retention challenges in AI systems:\n")
for artifact, info in ai_retention_challenges.items():
    print(f"  {artifact.upper()}")
    print(f"    What: {info['what']}")
    print(f"    Risk: {info['risk']}")
    print(f"    Retention: {info['recommended_retention']}")
    print()

Retention Scheduler

import time
import json
import hashlib
from dataclasses import dataclass, field
from datetime import datetime, timezone, timedelta
from typing import Optional, Callable
from enum import Enum


class RetentionAction(Enum):
    KEEP = "keep"
    ARCHIVE = "archive"
    DELETE = "delete"
    ANONYMIZE = "anonymize"


@dataclass
class RetentionPolicy:
    name: str
    data_type: str
    retention_days: int
    action_on_expiry: RetentionAction
    description: str


@dataclass
class DataRecord:
    record_id: str
    data_type: str
    created_at: datetime
    content: Optional[str] = None
    metadata: dict = field(default_factory=dict)

    @property
    def age_days(self) -> float:
        now = datetime.now(timezone.utc)
        return (now - self.created_at).total_seconds() / 86400


@dataclass
class RetentionResult:
    records_checked: int
    records_kept: int
    records_archived: int
    records_deleted: int
    records_anonymized: int
    details: list[dict] = field(default_factory=list)


class RetentionScheduler:
    """Implements retention policies for AI data."""

    DEFAULT_POLICIES = [
        RetentionPolicy(
            name="llm_logs",
            data_type="llm_log",
            retention_days=30,
            action_on_expiry=RetentionAction.DELETE,
            description="Logs of LLM prompts and responses",
        ),
        RetentionPolicy(
            name="chat_history",
            data_type="chat_message",
            retention_days=90,
            action_on_expiry=RetentionAction.ANONYMIZE,
            description="History of user conversations",
        ),
        RetentionPolicy(
            name="redaction_mappings",
            data_type="redaction_mapping",
            retention_days=0,
            action_on_expiry=RetentionAction.DELETE,
            description="Reversible redaction mappings",
        ),
        RetentionPolicy(
            name="audit_logs",
            data_type="audit_log",
            retention_days=365,
            action_on_expiry=RetentionAction.ARCHIVE,
            description="PII audit logs",
        ),
        RetentionPolicy(
            name="user_data_cache",
            data_type="user_cache",
            retention_days=7,
            action_on_expiry=RetentionAction.DELETE,
            description="Temporary cache of user data",
        ),
    ]

    def __init__(
        self,
        policies: Optional[list[RetentionPolicy]] = None,
        on_delete: Optional[Callable] = None,
        on_archive: Optional[Callable] = None,
        on_anonymize: Optional[Callable] = None,
    ):
        self.policies = {
            p.data_type: p for p in (policies or self.DEFAULT_POLICIES)
        }
        self.on_delete = on_delete or self._default_delete
        self.on_archive = on_archive or self._default_archive
        self.on_anonymize = on_anonymize or self._default_anonymize

    def evaluate(self, records: list[DataRecord]) -> RetentionResult:
        """Evaluates what action to take for each record."""
        result = RetentionResult(records_checked=len(records), records_kept=0,
                                 records_archived=0, records_deleted=0,
                                 records_anonymized=0)

        for record in records:
            policy = self.policies.get(record.data_type)
            if not policy:
                result.records_kept += 1
                continue

            if record.age_days <= policy.retention_days:
                result.records_kept += 1
                result.details.append({
                    "record_id": record.record_id,
                    "action": "keep",
                    "age_days": round(record.age_days, 1),
                    "expires_in_days": round(
                        policy.retention_days - record.age_days, 1
                    ),
                })
            else:
                action = policy.action_on_expiry
                if action == RetentionAction.DELETE:
                    self.on_delete(record)
                    result.records_deleted += 1
                elif action == RetentionAction.ARCHIVE:
                    self.on_archive(record)
                    result.records_archived += 1
                elif action == RetentionAction.ANONYMIZE:
                    self.on_anonymize(record)
                    result.records_anonymized += 1

                result.details.append({
                    "record_id": record.record_id,
                    "action": action.value,
                    "age_days": round(record.age_days, 1),
                    "policy": policy.name,
                })

        return result

    def _default_delete(self, record: DataRecord):
        record.content = None
        record.metadata["deleted_at"] = datetime.now(timezone.utc).isoformat()

    def _default_archive(self, record: DataRecord):
        record.metadata["archived_at"] = datetime.now(timezone.utc).isoformat()
        record.metadata["archived"] = True

    def _default_anonymize(self, record: DataRecord):
        if record.content:
            record.content = hashlib.sha256(
                record.content.encode()
            ).hexdigest()[:16]
        record.metadata["anonymized_at"] = datetime.now(timezone.utc).isoformat()


# --- Demo ---

scheduler = RetentionScheduler()

now = datetime.now(timezone.utc)
records = [
    DataRecord("log-1", "llm_log", now - timedelta(days=5), "Recent log"),
    DataRecord("log-2", "llm_log", now - timedelta(days=45), "Old log"),
    DataRecord("chat-1", "chat_message", now - timedelta(days=30), "Hi there"),
    DataRecord("chat-2", "chat_message", now - timedelta(days=120), "Old chat"),
    DataRecord("map-1", "redaction_mapping", now - timedelta(hours=2), "SSN:123"),
    DataRecord("audit-1", "audit_log", now - timedelta(days=200), "PII detected"),
    DataRecord("cache-1", "user_cache", now - timedelta(days=10), "user data"),
]

result = scheduler.evaluate(records)

print(f"Retention Evaluation:")
print(f"  Checked: {result.records_checked}")
print(f"  Kept: {result.records_kept}")
print(f"  Deleted: {result.records_deleted}")
print(f"  Archived: {result.records_archived}")
print(f"  Anonymized: {result.records_anonymized}")
print(f"\n  Details:")
for d in result.details:
    print(f"    {d['record_id']}: {d['action']} (age: {d['age_days']}d)")

# Expected output:
# Retention Evaluation:
#   Checked: 7
#   Kept: 3
#   Deleted: 3
#   Archived: 0
#   Anonymized: 1

Log sanitization

Logs are one of the most underestimated exposure vectors. A log that records the full prompt contains all the PII the user sent.

import re
import json
import logging
from typing import Optional


class SanitizedLogger:
    """Logger that sanitizes PII before writing."""

    PII_PATTERNS = {
        "email": re.compile(
            r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"
        ),
        "phone": re.compile(
            r"\b(?:\+\d{1,3}\s?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b"
        ),
        "ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
        "credit_card": re.compile(r"\b(?:\d{4}[-\s]?){3}\d{4}\b"),
        "api_key": re.compile(r"\b(?:sk|pk|api)[_-][A-Za-z0-9]{20,}\b"),
    }

    def __init__(self, logger_name: str = "ai_system"):
        self.logger = logging.getLogger(logger_name)
        if not self.logger.handlers:
            handler = logging.StreamHandler()
            handler.setFormatter(
                logging.Formatter(
                    "%(asctime)s [%(levelname)s] %(message)s"
                )
            )
            self.logger.addHandler(handler)
            self.logger.setLevel(logging.INFO)

    def _sanitize(self, message: str) -> str:
        """Replaces PII in the log message."""
        sanitized = message
        for pii_type, pattern in self.PII_PATTERNS.items():
            sanitized = pattern.sub(f"[{pii_type.upper()}_REDACTED]", sanitized)
        return sanitized

    def info(self, message: str, **kwargs):
        self.logger.info(self._sanitize(message), **kwargs)

    def warning(self, message: str, **kwargs):
        self.logger.warning(self._sanitize(message), **kwargs)

    def error(self, message: str, **kwargs):
        self.logger.error(self._sanitize(message), **kwargs)

    def log_request(
        self,
        request_id: str,
        user_input: str,
        llm_output: str,
        pii_detected: int = 0,
    ):
        """Request log with automatic sanitization."""
        entry = {
            "request_id": request_id,
            "input_length": len(user_input),
            "output_length": len(llm_output),
            "pii_detected": pii_detected,
            "input_preview": self._sanitize(user_input[:100]),
        }
        self.logger.info(json.dumps(entry))


# --- Demo ---

logger = SanitizedLogger("demo")

logger.info("User john@test.com called from 555-123-4567")
logger.info("Processing SSN 123-45-6789 for user")
logger.info("API key sk-abc123def456ghi789jkl012 detected in input")

logger.log_request(
    request_id="req-001",
    user_input="My email is maria@empresa.com and my SSN is 123-45-6789",
    llm_output="Your account status is active.",
    pii_detected=2,
)

# Expected output:
# 2026-03-13 [INFO] User [EMAIL_REDACTED] called from [PHONE_REDACTED]
# 2026-03-13 [INFO] Processing SSN [SSN_REDACTED] for user
# 2026-03-13 [INFO] API key [API_KEY_REDACTED] detected in input
# 2026-03-13 [INFO] {"request_id": "req-001", "input_length": 55, ...}

Encryption at rest

Stored data (logs, database, vector store) must be encrypted. Python provides tools to implement encryption at rest.

import os
import json
import base64
from dataclasses import dataclass
from typing import Optional

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC


class DataEncryptor:
    """Encrypts sensitive data at rest."""

    def __init__(self, encryption_key: Optional[bytes] = None):
        if encryption_key:
            self.fernet = Fernet(encryption_key)
        else:
            self.key = Fernet.generate_key()
            self.fernet = Fernet(self.key)

    @classmethod
    def from_password(cls, password: str, salt: Optional[bytes] = None) -> "DataEncryptor":
        """Creates an encryptor deriving the key from a password."""
        salt = salt or os.urandom(16)
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=480000,
        )
        key = base64.urlsafe_b64encode(kdf.derive(password.encode()))
        instance = cls(encryption_key=key)
        instance._salt = salt
        return instance

    def encrypt(self, data: str) -> str:
        """Encrypts a string and returns the ciphertext in base64."""
        return self.fernet.encrypt(data.encode()).decode()

    def decrypt(self, encrypted_data: str) -> str:
        """Decrypts a base64 ciphertext."""
        return self.fernet.decrypt(encrypted_data.encode()).decode()

    def encrypt_dict(self, data: dict) -> str:
        """Encrypts a full dictionary."""
        json_str = json.dumps(data, ensure_ascii=False)
        return self.encrypt(json_str)

    def decrypt_dict(self, encrypted_data: str) -> dict:
        """Decrypts a dictionary."""
        json_str = self.decrypt(encrypted_data)
        return json.loads(json_str)

    def encrypt_fields(
        self,
        data: dict,
        fields_to_encrypt: list[str],
    ) -> dict:
        """Encrypts only specific fields of a dict."""
        result = data.copy()
        for field_name in fields_to_encrypt:
            if field_name in result and isinstance(result[field_name], str):
                result[field_name] = self.encrypt(result[field_name])
                result[f"_{field_name}_encrypted"] = True
        return result

    def decrypt_fields(
        self,
        data: dict,
        fields_to_decrypt: list[str],
    ) -> dict:
        """Decrypts specific fields of a dict."""
        result = data.copy()
        for field_name in fields_to_decrypt:
            if field_name in result and result.get(f"_{field_name}_encrypted"):
                result[field_name] = self.decrypt(result[field_name])
                del result[f"_{field_name}_encrypted"]
        return result


# --- Demo ---

encryptor = DataEncryptor()

sensitive_data = {
    "user_id": "usr-123",
    "name": "María García",
    "email": "maria@empresa.com",
    "query": "What's my balance?",
    "account_balance": "$15,234.50",
}

encrypted = encryptor.encrypt_fields(
    sensitive_data,
    fields_to_encrypt=["email", "account_balance"],
)

print("Original:")
for k, v in sensitive_data.items():
    print(f"  {k}: {v}")

print("\nEncrypted fields:")
for k, v in encrypted.items():
    display = v[:40] + "..." if isinstance(v, str) and len(v) > 40 else v
    print(f"  {k}: {display}")

decrypted = encryptor.decrypt_fields(
    encrypted,
    fields_to_decrypt=["email", "account_balance"],
)

print("\nDecrypted:")
for k, v in decrypted.items():
    print(f"  {k}: {v}")

print(f"\nMatch original: {decrypted['email'] == sensitive_data['email']}")

# Expected output:
# Original:
#   email: maria@empresa.com
#   account_balance: $15,234.50
#
# Encrypted fields:
#   email: gAAAAABh... (encrypted)
#   _email_encrypted: True
#
# Decrypted:
#   email: maria@empresa.com
#   account_balance: $15,234.50
#
# Match original: True

Encryption in transit

For data in transit between your application and external APIs (OpenAI, vector stores), HTTPS/TLS is mandatory. Verify that all connections use TLS.

import ssl
from urllib.parse import urlparse


def verify_tls_configuration(urls: list[str]) -> list[dict]:
    """Verifies that the URLs use TLS."""
    results = []
    for url in urls:
        parsed = urlparse(url)
        result = {
            "url": url,
            "scheme": parsed.scheme,
            "uses_tls": parsed.scheme == "https",
            "hostname": parsed.hostname,
        }

        if not result["uses_tls"]:
            result["warning"] = (
                "INSECURE: Connection does not use TLS. "
                "Data in transit is not encrypted."
            )
            result["fix"] = f"Change to https://{parsed.hostname}{parsed.path}"
        else:
            result["status"] = "OK"

        results.append(result)
    return results


# --- Demo ---

api_urls = [
    "https://api.openai.com/v1/chat/completions",
    "https://your-vector-store.com/api/search",
    "http://internal-service.local/process",
    "https://your-database.com:5432/query",
]

tls_results = verify_tls_configuration(api_urls)
for r in tls_results:
    status = "✅" if r["uses_tls"] else "❌"
    print(f"  {status} {r['url'][:50]}...")
    if not r["uses_tls"]:
        print(f"     ⚠ {r['warning']}")
        print(f"     Fix: {r['fix']}")

# Expected output:
#   ✅ https://api.openai.com/v1/chat/completions...
#   ✅ https://your-vector-store.com/api/search...
#   ❌ http://internal-service.local/process...
#      ⚠ INSECURE: Connection does not use TLS...
#   ✅ https://your-database.com:5432/query...

Key management

Encryption keys need proper management. Don't store them in the code or in environment variables in production.

import os
import json
from datetime import datetime, timezone, timedelta


class KeyManager:
    """Basic management of encryption keys."""

    def __init__(self, keys_directory: str = ".keys"):
        self.keys_dir = keys_directory
        os.makedirs(keys_directory, exist_ok=True)

    def generate_key(self, key_name: str, ttl_days: int = 90) -> dict:
        """Generates a new key with metadata."""
        key = Fernet.generate_key()
        metadata = {
            "key_name": key_name,
            "created_at": datetime.now(timezone.utc).isoformat(),
            "expires_at": (
                datetime.now(timezone.utc) + timedelta(days=ttl_days)
            ).isoformat(),
            "ttl_days": ttl_days,
            "status": "active",
        }
        return {"key": key.decode(), "metadata": metadata}

    def rotate_key(
        self,
        old_key: str,
        new_key_name: str,
    ) -> dict:
        """Generates a new key for rotation."""
        new_key_data = self.generate_key(new_key_name)
        return {
            "old_key_status": "deprecated",
            "new_key": new_key_data,
            "action": "Re-encrypt all data with new key",
        }

    def check_key_expiry(self, key_metadata: dict) -> dict:
        """Checks whether a key is close to expiring."""
        expires = datetime.fromisoformat(key_metadata["expires_at"])
        now = datetime.now(timezone.utc)
        days_remaining = (expires - now).days

        return {
            "key_name": key_metadata["key_name"],
            "days_remaining": days_remaining,
            "expired": days_remaining <= 0,
            "needs_rotation": days_remaining <= 14,
            "status": (
                "EXPIRED" if days_remaining <= 0
                else "ROTATE_SOON" if days_remaining <= 14
                else "OK"
            ),
        }


# --- Demo ---

km = KeyManager()

key_data = km.generate_key("pii_encryption", ttl_days=90)
print(f"Key generated: {key_data['metadata']['key_name']}")
print(f"  Expires: {key_data['metadata']['expires_at'][:10]}")

expiry = km.check_key_expiry(key_data['metadata'])
print(f"  Status: {expiry['status']}")
print(f"  Days remaining: {expiry['days_remaining']}")

Secure deletion

Deleting data isn't simply del or os.remove(). In production, you need to make sure the data isn't recoverable.

import os
import secrets
from typing import Optional


class SecureDeleter:
    """Secure deletion of data."""

    @staticmethod
    def secure_delete_string(sensitive_string: str) -> str:
        """
        Overwrites a string in memory.
        Note: Python doesn't guarantee memory immutability,
        but this is the best practice available.
        """
        length = len(sensitive_string)
        return secrets.token_hex(length // 2)

    @staticmethod
    def secure_delete_file(filepath: str, passes: int = 3):
        """
        Overwrites a file with random data before deleting it.
        """
        if not os.path.exists(filepath):
            return {"status": "not_found", "path": filepath}

        file_size = os.path.getsize(filepath)

        for pass_num in range(passes):
            with open(filepath, "wb") as f:
                f.write(os.urandom(file_size))
                f.flush()
                os.fsync(f.fileno())

        os.remove(filepath)
        return {
            "status": "deleted",
            "path": filepath,
            "passes": passes,
            "size_overwritten": file_size,
        }

    @staticmethod
    def secure_delete_dict(data: dict) -> dict:
        """Overwrites the values of a dictionary."""
        for key in list(data.keys()):
            if isinstance(data[key], str):
                data[key] = secrets.token_hex(8)
            elif isinstance(data[key], (int, float)):
                data[key] = 0
            elif isinstance(data[key], dict):
                SecureDeleter.secure_delete_dict(data[key])
            elif isinstance(data[key], list):
                data[key] = []
        return data


# --- Demo ---

deleter = SecureDeleter()

sensitive = {
    "ssn": "123-45-6789",
    "credit_card": "4111-1111-1111-1111",
    "email": "maria@empresa.com",
}

print("Before deletion:")
for k, v in sensitive.items():
    print(f"  {k}: {v}")

deleter.secure_delete_dict(sensitive)

print("\nAfter secure deletion:")
for k, v in sensitive.items():
    print(f"  {k}: {v}")

# Expected output:
# Before deletion:
#   ssn: 123-45-6789
#   credit_card: 4111-1111-1111-1111
#   email: maria@empresa.com
#
# After secure deletion:
#   ssn: a1b2c3d4e5f6g7h8
#   credit_card: 9i0j1k2l3m4n5o6p
#   email: q7r8s9t0u1v2w3x4

Connection to the project

The Retention Scheduler and the encryption practices integrate into the PII Protection Layer:

PII Protection Layer
├── PIIScanner (Capsule 03)
├── PreLLMRedactor (Capsule 04)
├── PostLLMRedactor (Capsule 04)
├── DataMinimizer (Capsule 05)
├── RetentionScheduler (THIS CAPSULE)   ← Retention
├── DataEncryptor (THIS CAPSULE)        ← Encryption
├── SanitizedLogger (THIS CAPSULE)      ← Log sanitization
└── Audit Logger

Troubleshooting

Problem 1: "The retention policy deletes data I still need for debugging"

Solution: Create a separate policy for debugging data with a longer retention (90 days), but make sure the debugging data is sanitized (no real PII). Use logs with redacted PII for debugging.

Problem 2: "Encryption adds significant latency"

Solution: Encrypt only the sensitive fields, not the entire record. Use encrypt_fields() instead of encrypt_dict(). Encrypting a typical text field takes <1ms with Fernet.

Problem 3: "I lost the encryption key and can't decrypt the data"

Solution: Implement key backup and escrow from the start. Use a key management service (AWS KMS, GCP KMS, HashiCorp Vault from Module 5) instead of managing keys manually.

Problem 4: "I don't know how long to retain data based on my jurisdiction"

Solution: As a general rule: transaction data 7 years (tax requirements), audit logs 1-3 years, user data until they request deletion or the relationship ends. Consult your legal team for specific requirements.

Exercises

Exercise 1: Retention scheduler with notifications

Extend the RetentionScheduler to send notifications when data is close to expiring.

See solution 
class NotifyingRetentionScheduler(RetentionScheduler):
    def __init__(self, warning_days: int = 7, **kwargs):
        super().__init__(**kwargs)
        self.warning_days = warning_days
        self.notifications: list[dict] = []

    def evaluate_with_warnings(
        self, records: list[DataRecord],
    ) -> RetentionResult:
        result = self.evaluate(records)

        for record in records:
            policy = self.policies.get(record.data_type)
            if not policy:
                continue
            days_remaining = policy.retention_days - record.age_days
            if 0 < days_remaining <= self.warning_days:
                self.notifications.append({
                    "record_id": record.record_id,
                    "data_type": record.data_type,
                    "expires_in_days": round(days_remaining, 1),
                    "action_on_expiry": policy.action_on_expiry.value,
                })

        return result


scheduler = NotifyingRetentionScheduler(warning_days=7)
now = datetime.now(timezone.utc)
records = [
    DataRecord("log-1", "llm_log", now - timedelta(days=25)),
    DataRecord("log-2", "llm_log", now - timedelta(days=45)),
]
scheduler.evaluate_with_warnings(records)
print(f"Notifications: {scheduler.notifications}")

Exercise 2: Encrypted audit log

Create an audit log that encrypts the entries before writing them.

See solution 
class EncryptedAuditLog:
    def __init__(self):
        self.encryptor = DataEncryptor()
        self.entries: list[str] = []

    def log(self, entry: dict):
        encrypted = self.encryptor.encrypt_dict(entry)
        self.entries.append(encrypted)

    def read(self, index: int) -> dict:
        return self.encryptor.decrypt_dict(self.entries[index])

    def read_all(self) -> list[dict]:
        return [self.encryptor.decrypt_dict(e) for e in self.entries]


audit = EncryptedAuditLog()
audit.log({"event": "pii_detected", "type": "email", "request_id": "req-1"})
audit.log({"event": "pii_redacted", "type": "ssn", "request_id": "req-2"})

print(f"Encrypted entries: {len(audit.entries)}")
print(f"Raw (encrypted): {audit.entries[0][:40]}...")
print(f"Decrypted: {audit.read(0)}")

Exercise 3: Log sanitizer with custom patterns

Extend the SanitizedLogger to accept custom PII patterns.

See solution 
class CustomSanitizedLogger(SanitizedLogger):
    def __init__(self, custom_patterns: Optional[dict] = None, **kwargs):
        super().__init__(**kwargs)
        if custom_patterns:
            for name, pattern_str in custom_patterns.items():
                self.PII_PATTERNS[name] = re.compile(pattern_str)

    def add_pattern(self, name: str, pattern: str):
        self.PII_PATTERNS[name] = re.compile(pattern)


logger = CustomSanitizedLogger(
    custom_patterns={
        "employee_id": r"\bEMP-\d{6}\b",
        "order_id": r"\bORD-\d{5,}\b",
    }
)

logger.info("Employee EMP-123456 processed order ORD-98765")
# Output: Employee [EMPLOYEE_ID_REDACTED] processed order [ORDER_ID_REDACTED]

Exercise 4: Data lifecycle tracker

Create a tracker that records the full lifecycle of a piece of data.

See solution 
@dataclass
class DataLifecycleEvent:
    timestamp: str
    event: str
    details: str

class DataLifecycleTracker:
    def __init__(self):
        self.records: dict[str, list[DataLifecycleEvent]] = {}

    def track(self, record_id: str, event: str, details: str = ""):
        if record_id not in self.records:
            self.records[record_id] = []
        self.records[record_id].append(DataLifecycleEvent(
            timestamp=datetime.now(timezone.utc).isoformat(),
            event=event,
            details=details,
        ))

    def get_lifecycle(self, record_id: str) -> list[dict]:
        events = self.records.get(record_id, [])
        return [{"event": e.event, "details": e.details, "at": e.timestamp} for e in events]


tracker = DataLifecycleTracker()
tracker.track("data-001", "created", "User submitted form")
tracker.track("data-001", "pii_detected", "Email and phone found")
tracker.track("data-001", "redacted", "PII redacted pre-LLM")
tracker.track("data-001", "processed", "LLM generated response")
tracker.track("data-001", "scheduled_deletion", "Retention: 30 days")

for event in tracker.get_lifecycle("data-001"):
    print(f"  {event['event']}: {event['details']}")

Summary

  • 🔑 Retention policies define how long AI data is stored — LLM logs, chat histories, and redaction mappings need different policies
  • 🔑 The RetentionScheduler evaluates records against policies and runs automatic actions: DELETE, ARCHIVE, or ANONYMIZE when they expire
  • 🔑 Log sanitization is critical — a logger that records full prompts exposes all the user's PII. Use SanitizedLogger with regex to redact before writing
  • 🔑 Encryption at rest with Fernet (symmetric) protects stored data — encrypt sensitive fields individually to minimize latency
  • 🔑 Encryption in transit requires HTTPS/TLS for all connections to external APIs — verify there are no unencrypted HTTP endpoints
  • 🔑 Key management needs periodic rotation (90 days), secure backup, and ideally a KMS service (covered in Module 5)
  • 🔑 Secure deletion requires overwriting data before deleting — del and os.remove() don't guarantee the data is unrecoverable
  • 🔑 The redaction mappings from reversible redaction are the most sensitive data in the pipeline — they must be deleted immediately after processing

Additional resources

  1. GDPR Art. 5(1)(e) — Storage Limitation — Storage limitation principle in GDPR
  2. Python cryptography Library — Encryption library used in this capsule (Fernet)
  3. NIST Guidelines for Media Sanitization (SP 800-88) — NIST's guide for secure data deletion
  4. OWASP Logging Cheat Sheet — Best practices for secure logging
  5. AWS KMS Documentation — AWS's key management service
  6. HashiCorp Vault — Encryption as a Service — Vault's transit secrets engine for encryption
  7. PCI DSS — Key Management — Key management requirements for credit card data

Created: March 2026 Version: 1.0

← Previous5. Data MinimizationNext →7. Compliance Basics: GDPR, CCPA, and AI