TLDR: "Legal will handle privacy" is how CTOs get fired. GDPR makes you personally liable for technical decisions. This is the checklist that separates compliant organizations from those awaiting their first regulatory action.
Read full summary
Comprehensive checklist covering data inventory, consent infrastructure, security controls, vendor management, incident response, and privacy-by-design principles. Designed for technical leaders building compliant systems at scale.
*Summary by Claude AI*
## The CTO Who Thought Privacy Was "A Legal Problem"
In 2023, a Series C startup's CTO sat across from regulators explaining why user location data was stored unencrypted in a logging system nobody monitored. The data breach had exposed 2.3 million users. The fine was €8.2 million—enough to sink their Series D. The CTO was asked to resign.
The technical decisions that led to this moment were mundane: a logging configuration that nobody reviewed, a vendor integration that shipped without a security assessment, retention policies that existed in a legal document but not in the database. None of it seemed like a "privacy issue" at the time. All of it became the CTO's responsibility.
Modern privacy regulations don't distinguish between legal and technical failures. A DSAR that takes 45 days instead of 30 is a technical failure with legal consequences. A consent system that doesn't propagate to all data stores is an engineering bug with regulatory implications. The buck stops with whoever made the architectural decisions—and that's you.
## The CTO's Complete Privacy Engineering Checklist for 2025
As a CTO or technical leader, privacy is no longer someone else's problem. The era of "legal will handle it" is over. Modern privacy regulations—GDPR, CCPA, LGPD, and dozens of others—impose technical requirements that must be engineered into systems from the ground up. A privacy failure is increasingly a technical failure, and the penalties are severe: fines up to 4% of global revenue, mandatory breach notifications, and the kind of reputational damage that drives customers to competitors.
This comprehensive checklist covers the ten essential areas every CTO must address to build a privacy-respecting technology organization. Whether you're a startup preparing for scale or an enterprise modernizing legacy systems, these are the technical foundations that separate compliant organizations from those waiting for their first regulatory action.
## 1. Know Your Data: Complete Data Inventory and Mapping
### Why It Matters
You can't protect what you don't know you have. Data mapping is the foundation of every privacy program—it's impossible to respond to data subject requests, assess breach impact, or implement data minimization without a complete inventory of what data you collect, where it lives, and how it flows.
### The Technical Challenge
Modern organizations have data everywhere:
- Production databases (often multiple)
- Data warehouses and lakes
- Third-party SaaS tools (CRM, support desk, marketing automation)
- Log aggregation systems
- Backup systems and archives
- Employee devices and collaboration tools
- Third-party analytics and advertising platforms
### Implementation Approach
**1. Automated Data Discovery**
Deploy tools that continuously discover and classify data:
```python
# Data discovery and classification system
from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum
class DataCategory(Enum):
PII = "personally_identifiable_information"
SENSITIVE = "sensitive_personal_data"
FINANCIAL = "financial_data"
HEALTH = "health_data"
BEHAVIORAL = "behavioral_data"
TECHNICAL = "technical_data"
BUSINESS = "business_data"
@dataclass
class DataElement:
name: str
category: DataCategory
sensitivity: str # low, medium, high, critical
pii_indicators: List[str]
sample_values: Optional[List[str]]
class DataInventoryService:
"""Automated data discovery and classification"""
def __init__(self):
self.pii_patterns = {
'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
'phone': r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
'ssn': r'\b\d{3}-\d{2}-\d{4}\b',
'credit_card': r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',
'ip_address': r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b'
}
self.sensitive_column_names = [
'password', 'secret', 'token', 'ssn', 'social_security',
'credit_card', 'card_number', 'cvv', 'dob', 'date_of_birth',
'salary', 'health', 'medical', 'diagnosis', 'religion',
'ethnicity', 'sexual_orientation', 'political', 'biometric'
]
async def discover_database(self, connection_string: str) -> Dict:
"""Scan database and classify all columns"""
inventory = {
'database': connection_string.split('/')[-1],
'scanned_at': datetime.utcnow(),
'tables': []
}
conn = await asyncpg.connect(connection_string)
# Get all tables
tables = await conn.fetch("""
SELECT table_name
FROM information_schema.tables
WHERE table_schema = 'public'
""")
for table in tables:
table_inventory = await self.analyze_table(conn, table['table_name'])
inventory['tables'].append(table_inventory)
await conn.close()
return inventory
async def analyze_table(self, conn, table_name: str) -> Dict:
"""Analyze single table for PII and sensitive data"""
# Get column information
columns = await conn.fetch(f"""
SELECT column_name, data_type
FROM information_schema.columns
WHERE table_name = '{table_name}'
""")
# Sample data for pattern matching
sample = await conn.fetch(f"SELECT * FROM {table_name} LIMIT 100")
classified_columns = []
for col in columns:
column_name = col['column_name']
# Check column name for sensitivity indicators
name_sensitivity = self.check_column_name(column_name)
# Check sample data for PII patterns
data_patterns = self.scan_sample_data(sample, column_name)
classified_columns.append({
'name': column_name,
'data_type': col['data_type'],
'sensitivity': name_sensitivity or data_patterns.get('sensitivity', 'low'),
'pii_detected': data_patterns.get('patterns', []),
'category': self.determine_category(name_sensitivity, data_patterns),
'requires_encryption': name_sensitivity in ['high', 'critical'],
'retention_category': self.determine_retention(column_name)
})
return {
'table_name': table_name,
'row_count': len(sample),
'columns': classified_columns,
'contains_pii': any(c['pii_detected'] for c in classified_columns),
'sensitivity_level': max(c['sensitivity'] for c in classified_columns)
}
```
**2. Data Flow Mapping**
Document how data moves through your systems:
```yaml
# data_flows.yaml - Document all data flows
data_flows:
- name: "User Registration"
source: "Web Application"
destination: "Users Database"
data_elements:
- email
- password_hash
- name
legal_basis: "contract"
retention: "account_lifetime"
- name: "Analytics Collection"
source: "Web Application"
destination: "Google Analytics"
data_elements:
- page_url
- session_id
- device_info
- ip_address (anonymized)
legal_basis: "consent"
consent_required: true
retention: "14_months"
third_party: true
dpa_status: "signed"
- name: "Payment Processing"
source: "Checkout Flow"
destination: "Stripe"
data_elements:
- card_number (tokenized)
- billing_address
- email
legal_basis: "contract"
third_party: true
pci_scope: true
dpa_status: "stripe_dpa"
- name: "Marketing Emails"
source: "Users Database"
destination: "SendGrid"
data_elements:
- email
- name
- preferences
legal_basis: "consent"
consent_required: true
unsubscribe_mechanism: "one_click"
third_party: true
dpa_status: "signed"
```
### Deliverables
- [ ] Complete inventory of all data stores
- [ ] Classification of all data elements by sensitivity
- [ ] Data flow diagrams showing all transfers
- [ ] Third-party data sharing inventory
- [ ] Legal basis documented for each processing activity
- [ ] Automated discovery running on schedule
## 2. Encrypt Everywhere: Data Protection at Rest and in Transit
### Why It Matters
Encryption is your last line of defense. When (not if) unauthorized access occurs, encryption determines whether you have a minor security incident or a catastrophic data breach requiring notification to regulators and affected individuals.
### The Requirements
**At Rest**: All personal data must be encrypted when stored
**In Transit**: All data must be encrypted during transmission
**Key Management**: Encryption keys must be properly managed and rotated
### Implementation
**1. Database Encryption**
```sql
-- PostgreSQL: Enable encryption at rest
-- Use pgcrypto for column-level encryption of sensitive fields
-- Create encryption function
CREATE EXTENSION IF NOT EXISTS pgcrypto;
-- Encrypt sensitive columns
CREATE OR REPLACE FUNCTION encrypt_pii(data TEXT, key TEXT)
RETURNS BYTEA AS $$
BEGIN
RETURN pgp_sym_encrypt(data, key);
END;
$$ LANGUAGE plpgsql;
CREATE OR REPLACE FUNCTION decrypt_pii(data BYTEA, key TEXT)
RETURNS TEXT AS $$
BEGIN
RETURN pgp_sym_decrypt(data, key);
END;
$$ LANGUAGE plpgsql;
-- Example: Encrypted user table
CREATE TABLE users_encrypted (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
email_encrypted BYTEA NOT NULL,
email_hash VARCHAR(64) NOT NULL, -- For lookups
name_encrypted BYTEA,
phone_encrypted BYTEA,
created_at TIMESTAMP DEFAULT NOW(),
-- Index on hash for efficient lookups
CONSTRAINT email_hash_unique UNIQUE (email_hash)
);
-- Insert with encryption
INSERT INTO users_encrypted (email_encrypted, email_hash, name_encrypted)
VALUES (
encrypt_pii('
[email protected]', current_setting('app.encryption_key')),
encode(sha256('
[email protected]'::bytea), 'hex'),
encrypt_pii('John Doe', current_setting('app.encryption_key'))
);
```
**2. Application-Level Encryption**
```python
# Envelope encryption with key rotation support
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import boto3
import base64
from datetime import datetime, timedelta
class EncryptionService:
"""Enterprise-grade encryption with key management"""
def __init__(self, kms_key_id: str):
self.kms = boto3.client('kms')
self.kms_key_id = kms_key_id
self.data_key_cache = {}
self.key_rotation_interval = timedelta(days=30)
async def encrypt(self, plaintext: str, context: dict = None) -> dict:
"""Encrypt data using envelope encryption"""
# Generate data key from KMS
response = self.kms.generate_data_key(
KeyId=self.kms_key_id,
KeySpec='AES_256',
EncryptionContext=context or {}
)
# Use plaintext key for encryption
fernet = Fernet(base64.urlsafe_b64encode(response['Plaintext'][:32]))
ciphertext = fernet.encrypt(plaintext.encode())
# Return encrypted data key (for storage) and ciphertext
return {
'ciphertext': base64.b64encode(ciphertext).decode(),
'encrypted_data_key': base64.b64encode(response['CiphertextBlob']).decode(),
'key_id': self.kms_key_id,
'algorithm': 'AES-256-GCM',
'encrypted_at': datetime.utcnow().isoformat()
}
async def decrypt(self, encrypted_data: dict, context: dict = None) -> str:
"""Decrypt data using envelope encryption"""
# Decrypt the data key
response = self.kms.decrypt(
CiphertextBlob=base64.b64decode(encrypted_data['encrypted_data_key']),
EncryptionContext=context or {}
)
# Use decrypted key for data decryption
fernet = Fernet(base64.urlsafe_b64encode(response['Plaintext'][:32]))
plaintext = fernet.decrypt(base64.b64decode(encrypted_data['ciphertext']))
return plaintext.decode()
async def rotate_key(self, old_encrypted_data: dict) -> dict:
"""Re-encrypt data with new key version"""
# Decrypt with old key
plaintext = await self.decrypt(old_encrypted_data)
# Encrypt with current key
return await self.encrypt(plaintext)
```
**3. TLS Configuration**
```nginx
# nginx.conf - Strong TLS configuration
server {
listen 443 ssl http2;
server_name example.com;
# Modern TLS configuration
ssl_certificate /etc/ssl/certs/example.com.crt;
ssl_certificate_key /etc/ssl/private/example.com.key;
# Only TLS 1.2 and 1.3
ssl_protocols TLSv1.2 TLSv1.3;
# Strong cipher suites
ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384;
ssl_prefer_server_ciphers off;
# HSTS (1 year)
add_header Strict-Transport-Security "max-age=31536000; includeSubDomains; preload" always;
# OCSP Stapling
ssl_stapling on;
ssl_stapling_verify on;
# Session configuration
ssl_session_cache shared:SSL:10m;
ssl_session_timeout 1d;
ssl_session_tickets off;
}
```
### Deliverables
- [ ] All databases encrypted at rest
- [ ] All sensitive columns use application-level encryption
- [ ] TLS 1.2+ required for all connections
- [ ] Key rotation schedule implemented
- [ ] Key management procedures documented
- [ ] Encryption audit logging enabled
## 3. Access Control: Least Privilege and Role-Based Access
### Why It Matters
Every data breach investigation asks the same question: "Who had access to this data, and why?" Implementing least privilege access means users and systems only have access to the minimum data required for their function, reducing both accidental exposure and the blast radius of compromised accounts.
### Implementation
**1. Role-Based Access Control (RBAC)**
```python
# RBAC implementation with fine-grained permissions
from enum import Enum
from typing import Set, List
from functools import wraps
class Permission(Enum):
# User data
READ_USER_PII = "read:user:pii"
WRITE_USER_PII = "write:user:pii"
DELETE_USER = "delete:user"
EXPORT_USER_DATA = "export:user:data"
# Analytics
VIEW_ANALYTICS = "view:analytics"
VIEW_ANALYTICS_PII = "view:analytics:pii" # Includes user-level data
# Admin
MANAGE_USERS = "manage:users"
MANAGE_ROLES = "manage:roles"
VIEW_AUDIT_LOGS = "view:audit:logs"
# DSAR
PROCESS_DSAR = "process:dsar"
APPROVE_DELETION = "approve:deletion"
class Role(Enum):
CUSTOMER_SUPPORT = "customer_support"
MARKETING = "marketing"
ENGINEERING = "engineering"
PRIVACY_OFFICER = "privacy_officer"
ADMIN = "admin"
# Role-Permission mappings
ROLE_PERMISSIONS = {
Role.CUSTOMER_SUPPORT: {
Permission.READ_USER_PII, # Need to help customers
Permission.WRITE_USER_PII, # Update customer info
},
Role.MARKETING: {
Permission.VIEW_ANALYTICS, # Aggregate only
# Note: NO access to PII
},
Role.ENGINEERING: {
Permission.VIEW_ANALYTICS,
# Note: Production PII access requires additional approval
},
Role.PRIVACY_OFFICER: {
Permission.READ_USER_PII,
Permission.VIEW_ANALYTICS_PII,
Permission.PROCESS_DSAR,
Permission.APPROVE_DELETION,
Permission.VIEW_AUDIT_LOGS,
Permission.EXPORT_USER_DATA,
},
Role.ADMIN: {
Permission.MANAGE_USERS,
Permission.MANAGE_ROLES,
Permission.VIEW_AUDIT_LOGS,
}
}
class AccessControl:
"""Enforce role-based access control with audit logging"""
def __init__(self, audit_logger):
self.audit = audit_logger
def require_permission(self, permission: Permission):
"""Decorator to enforce permission checks"""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# Get current user from context
user = get_current_user()
if not self.has_permission(user, permission):
await self.audit.log({
'event': 'permission_denied',
'user_id': user.id,
'permission': permission.value,
'resource': func.__name__,
'timestamp': datetime.utcnow()
})
raise PermissionDenied(f"Missing permission: {permission.value}")
# Log successful access
await self.audit.log({
'event': 'permission_granted',
'user_id': user.id,
'permission': permission.value,
'resource': func.__name__,
'timestamp': datetime.utcnow()
})
return await func(*args, **kwargs)
return wrapper
return decorator
def has_permission(self, user, permission: Permission) -> bool:
"""Check if user has required permission"""
user_permissions = set()
for role in user.roles:
user_permissions.update(ROLE_PERMISSIONS.get(role, set()))
return permission in user_permissions
```
**2. Database Row-Level Security**
```sql
-- PostgreSQL Row-Level Security for multi-tenant data
-- Users can only see their own organization's data
-- Enable RLS on tables
ALTER TABLE customers ENABLE ROW LEVEL SECURITY;
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;
-- Policy: Users see only their organization's data
CREATE POLICY org_isolation ON customers
FOR ALL
USING (organization_id = current_setting('app.current_org_id')::uuid);
CREATE POLICY org_isolation ON orders
FOR ALL
USING (organization_id = current_setting('app.current_org_id')::uuid);
-- Policy: Support staff can view (not modify) any customer
CREATE POLICY support_read ON customers
FOR SELECT
USING (
current_setting('app.user_role') = 'support'
OR organization_id = current_setting('app.current_org_id')::uuid
);
-- Policy: Only privacy officers can delete
CREATE POLICY privacy_delete ON customers
FOR DELETE
USING (current_setting('app.user_role') = 'privacy_officer');
```
**3. Quarterly Access Review**
```python
# Automated access review system
class AccessReviewService:
"""Generate and track quarterly access reviews"""
async def generate_quarterly_review(self) -> dict:
"""Generate access review report for all users"""
users = await self.get_all_users_with_permissions()
review = {
'review_period': self.current_quarter(),
'generated_at': datetime.utcnow(),
'users': [],
'flagged_issues': []
}
for user in users:
user_review = {
'user_id': user.id,
'email': user.email,
'roles': user.roles,
'permissions': self.get_all_permissions(user),
'last_activity': user.last_login,
'flags': []
}
# Flag inactive users with elevated access
if self.is_inactive(user) and self.has_elevated_access(user):
user_review['flags'].append({
'type': 'inactive_elevated_access',
'message': f'User inactive for {self.days_inactive(user)} days with elevated permissions',
'recommendation': 'Review and potentially revoke access'
})
review['flagged_issues'].append(user_review)
# Flag users with conflicting roles
if self.has_conflicting_roles(user):
user_review['flags'].append({
'type': 'conflicting_roles',
'message': 'User has potentially conflicting role assignments',
'recommendation': 'Review role assignment for separation of duties'
})
review['flagged_issues'].append(user_review)
review['users'].append(user_review)
return review
```
### Deliverables
- [ ] RBAC system implemented
- [ ] All roles and permissions documented
- [ ] Row-level security enabled where applicable
- [ ] Quarterly access reviews scheduled
- [ ] Privileged access requires approval workflow
- [ ] Service accounts have minimum necessary permissions
## 4. Retention Policies: Automate Deletion
### Why It Matters
Data minimization is a core principle of every privacy regulation. You should only keep data as long as necessary for the purpose it was collected. Beyond legal requirements, excessive data retention increases breach risk, storage costs, and DSAR complexity.
### Implementation
**1. Retention Policy Definition**
```python
# Retention policy configuration
RETENTION_POLICIES = {
'user_accounts': {
'active': 'indefinite', # While account is active
'after_deletion': timedelta(days=30), # Grace period
'legal_minimum': None
},
'transaction_records': {
'active': timedelta(days=365 * 7), # 7 years for tax
'legal_minimum': timedelta(days=365 * 7),
'legal_basis': 'Tax regulations'
},
'access_logs': {
'active': timedelta(days=90),
'legal_minimum': timedelta(days=30),
'legal_basis': 'Security monitoring'
},
'analytics_data': {
'active': timedelta(days=365), # 1 year
'anonymization_after': timedelta(days=90), # Anonymize after 90 days
'legal_basis': 'Legitimate interest'
},
'support_tickets': {
'active': timedelta(days=365 * 3), # 3 years
'after_resolution': timedelta(days=365 * 3),
'legal_minimum': None
},
'marketing_consent': {
'active': 'until_withdrawn',
'proof_retention': timedelta(days=365 * 7), # Keep consent proof
'legal_basis': 'GDPR Article 7'
},
'session_data': {
'active': timedelta(hours=24),
'legal_minimum': None
},
'backup_data': {
'active': timedelta(days=30),
'note': 'Excluded from individual deletion requests'
}
}
```
**2. Automated Retention Enforcement**
```python
class RetentionEnforcementService:
"""Automatically enforce data retention policies"""
def __init__(self, db, policies: dict):
self.db = db
self.policies = policies
self.audit = AuditLogger()
async def run_retention_job(self) -> dict:
"""Execute retention policies across all data stores"""
results = {
'run_at': datetime.utcnow(),
'tables_processed': [],
'records_deleted': 0,
'records_anonymized': 0,
'errors': []
}
for data_type, policy in self.policies.items():
try:
result = await self.enforce_policy(data_type, policy)
results['tables_processed'].append(result)
results['records_deleted'] += result.get('deleted', 0)
results['records_anonymized'] += result.get('anonymized', 0)
except Exception as e:
results['errors'].append({
'data_type': data_type,
'error': str(e)
})
# Log retention job execution
await self.audit.log({
'event': 'retention_job_completed',
'results': results
})
return results
async def enforce_policy(self, data_type: str, policy: dict) -> dict:
"""Enforce single retention policy"""
result = {'data_type': data_type, 'deleted': 0, 'anonymized': 0}
# Calculate cutoff date
retention_period = policy.get('active')
if retention_period == 'indefinite' or retention_period == 'until_withdrawn':
return result
cutoff_date = datetime.utcnow() - retention_period
# Handle anonymization vs deletion
if 'anonymization_after' in policy:
anon_cutoff = datetime.utcnow() - policy['anonymization_after']
result['anonymized'] = await self.anonymize_records(data_type, anon_cutoff, cutoff_date)
# Delete records past retention
result['deleted'] = await self.delete_records(data_type, cutoff_date)
return result
async def delete_records(self, data_type: str, cutoff: datetime) -> int:
"""Soft delete records older than cutoff"""
table_config = self.get_table_config(data_type)
# Mark for deletion (soft delete)
query = f"""
UPDATE {table_config['table']}
SET deleted_at = NOW(),
deletion_reason = 'retention_policy'
WHERE {table_config['date_column']} < $1
AND deleted_at IS NULL
"""
result = await self.db.execute(query, cutoff)
# Hard delete after grace period (30 days)
hard_delete_cutoff = cutoff - timedelta(days=30)
hard_delete_query = f"""
DELETE FROM {table_config['table']}
WHERE deleted_at < $1
"""
await self.db.execute(hard_delete_query, hard_delete_cutoff)
return result.rowcount
async def anonymize_records(self, data_type: str, start: datetime, end: datetime) -> int:
"""Anonymize records in date range"""
table_config = self.get_table_config(data_type)
pii_columns = table_config.get('pii_columns', [])
# Build anonymization query
set_clauses = []
for col in pii_columns:
set_clauses.append(f"{col} = 'ANONYMIZED'")
query = f"""
UPDATE {table_config['table']}
SET {', '.join(set_clauses)},
anonymized_at = NOW()
WHERE {table_config['date_column']} BETWEEN $1 AND $2
AND anonymized_at IS NULL
"""
result = await self.db.execute(query, start, end)
return result.rowcount
```
### Deliverables
- [ ] Retention periods defined for all data categories
- [ ] Automated deletion jobs running
- [ ] Anonymization implemented for analytics data
- [ ] Backup retention aligned with policies
- [ ] Legal hold process documented
- [ ] Retention audit reports generated
## 5. Incident Response: Prepare for the Worst
### Why It Matters
GDPR requires breach notification within 72 hours. CCPA, LGPD, and others have similar requirements. Without a tested incident response plan, you'll spend the critical first hours figuring out what to do instead of containing the breach and meeting notification deadlines.
### Implementation
**1. Incident Detection**
```python
# Automated breach detection system
class SecurityMonitoringService:
"""Real-time monitoring for potential data breaches"""
def __init__(self):
self.alert_thresholds = {
'failed_logins': {'count': 10, 'window_minutes': 5},
'data_export': {'count': 3, 'window_minutes': 60},
'privilege_escalation': {'count': 1, 'window_minutes': 1},
'unusual_data_access': {'count': 100, 'window_minutes': 60},
'api_errors': {'count': 50, 'window_minutes': 5}
}
async def analyze_event(self, event: dict) -> Optional[dict]:
"""Analyze event for potential security incident"""
incident = None
if event['type'] == 'login_failed':
if await self.exceeds_threshold('failed_logins', event['user_id']):
incident = {
'type': 'brute_force_attempt',
'severity': 'high',
'affected_user': event['user_id'],
'details': event
}
elif event['type'] == 'data_export':
if await self.exceeds_threshold('data_export', event['user_id']):
incident = {
'type': 'unusual_data_export',
'severity': 'critical',
'affected_user': event['user_id'],
'data_volume': event.get('record_count'),
'details': event
}
elif event['type'] == 'permission_change':
if event['new_permissions'] > event['old_permissions']:
incident = {
'type': 'privilege_escalation',
'severity': 'critical',
'affected_user': event['target_user_id'],
'changed_by': event['actor_id'],
'details': event
}
if incident:
await self.create_incident(incident)
return incident
```
**2. Incident Response Workflow**
```python
class IncidentResponseService:
"""Manage security incident response workflow"""
SEVERITY_LEVELS = {
'critical': {
'response_time_minutes': 15,
'escalation_path': ['security_team', 'cto', 'dpo', 'ceo'],
'requires_war_room': True
},
'high': {
'response_time_minutes': 60,
'escalation_path': ['security_team', 'engineering_lead'],
'requires_war_room': False
},
'medium': {
'response_time_minutes': 240,
'escalation_path': ['security_team'],
'requires_war_room': False
}
}
async def create_incident(self, incident_data: dict) -> dict:
"""Create and initialize incident response"""
incident = {
'id': str(uuid.uuid4()),
'created_at': datetime.utcnow(),
'status': 'detected',
'severity': incident_data['severity'],
'type': incident_data['type'],
'timeline': [{
'timestamp': datetime.utcnow(),
'event': 'Incident detected',
'details': incident_data
}]
}
# Store incident
await self.incident_repository.save(incident)
# Alert appropriate teams
await self.alert_response_team(incident)
# Start response timer
await self.start_response_timer(incident)
# If critical, initiate war room
if incident_data['severity'] == 'critical':
await self.initiate_war_room(incident)
return incident
async def assess_breach_notification(self, incident_id: str) -> dict:
"""Determine if breach requires regulatory notification"""
incident = await self.get_incident(incident_id)
assessment = {
'incident_id': incident_id,
'assessed_at': datetime.utcnow(),
'notification_required': False,
'affected_data_subjects': 0,
'jurisdictions': [],
'notification_deadlines': []
}
# Determine affected records
affected_records = await self.count_affected_records(incident)
assessment['affected_data_subjects'] = affected_records['count']
# Check if PII was involved
if not affected_records['contains_pii']:
assessment['notification_required'] = False
assessment['reason'] = 'No personal data affected'
return assessment
# Check notification thresholds by jurisdiction
jurisdictions = await self.determine_jurisdictions(affected_records)
for jurisdiction in jurisdictions:
if self.meets_notification_threshold(jurisdiction, affected_records):
assessment['notification_required'] = True
assessment['jurisdictions'].append(jurisdiction)
assessment['notification_deadlines'].append({
'jurisdiction': jurisdiction,
'deadline': self.calculate_deadline(jurisdiction, incident['created_at']),
'authority': self.get_supervisory_authority(jurisdiction)
})
return assessment
async def generate_breach_notification(self, incident_id: str, jurisdiction: str) -> dict:
"""Generate regulatory breach notification"""
incident = await self.get_incident(incident_id)
assessment = await self.get_assessment(incident_id)
notification = {
'incident_id': incident_id,
'jurisdiction': jurisdiction,
'generated_at': datetime.utcnow(),
# GDPR Article 33 requirements
'nature_of_breach': incident['type'],
'categories_of_data': await self.get_affected_data_categories(incident),
'approximate_records': assessment['affected_data_subjects'],
'likely_consequences': self.assess_consequences(incident),
'measures_taken': await self.get_remediation_measures(incident),
'dpo_contact': self.get_dpo_contact(),
# Timeline
'breach_discovered': incident['created_at'],
'notification_deadline': self.calculate_deadline(jurisdiction, incident['created_at'])
}
return notification
```
### Deliverables
- [ ] Incident response plan documented
- [ ] Response team roles and contacts defined
- [ ] Automated detection systems deployed
- [ ] Breach notification templates prepared
- [ ] Regular incident response drills conducted
- [ ] Communication templates for affected users
- [ ] Legal counsel pre-engaged for breach response
## 6. Vendor Review: Control Third-Party Scripts
### Why It Matters
Marketing teams love adding scripts. Each script is a potential data leak, security vulnerability, and compliance risk. You need technical controls to ensure only approved, reviewed vendors have access to your users' data.
### Implementation
**1. Vendor Approval Workflow**
```python
class VendorApprovalService:
"""Manage third-party vendor approval workflow"""
REQUIRED_DOCUMENTATION = [
'privacy_policy_url',
'dpa_signed',
'security_certifications',
'data_processing_purposes',
'data_retention_period',
'subprocessor_list',
'data_transfer_mechanisms'
]
async def submit_vendor_request(self, request: dict) -> dict:
"""Submit new vendor for approval"""
vendor_request = {
'id': str(uuid.uuid4()),
'vendor_name': request['vendor_name'],
'requested_by': request['requester_id'],
'department': request['department'],
'use_case': request['use_case'],
'data_accessed': request['data_types'],
'status': 'pending_review',
'created_at': datetime.utcnow()
}
# Check documentation completeness
missing_docs = []
for doc in self.REQUIRED_DOCUMENTATION:
if doc not in request or not request[doc]:
missing_docs.append(doc)
if missing_docs:
vendor_request['status'] = 'incomplete'
vendor_request['missing_documentation'] = missing_docs
else:
# Route to appropriate reviewers
vendor_request['reviewers'] = self.determine_reviewers(request)
await self.notify_reviewers(vendor_request)
await self.repository.save(vendor_request)
return vendor_request
def determine_reviewers(self, request: dict) -> List[str]:
"""Determine required reviewers based on data sensitivity"""
reviewers = ['security_team']
if 'pii' in request['data_types'] or 'sensitive' in request['data_types']:
reviewers.append('privacy_officer')
if request.get('data_transfer_outside_eea'):
reviewers.append('legal_team')
if request.get('estimated_cost', 0) > 10000:
reviewers.append('finance')
return reviewers
```
**2. Content Security Policy Enforcement**
```python
# CSP management for controlling third-party scripts
class CSPManager:
"""Manage Content Security Policy for third-party script control"""
def __init__(self):
self.approved_vendors = {}
self.csp_directives = {
'default-src': ["'self'"],
'script-src': ["'self'"],
'style-src': ["'self'", "'unsafe-inline'"],
'img-src': ["'self'", "data:", "https:"],
'connect-src': ["'self'"],
'font-src': ["'self'"],
'frame-src': ["'none'"],
'object-src': ["'none'"]
}
def add_approved_vendor(self, vendor: dict):
"""Add approved vendor to CSP"""
vendor_id = vendor['id']
self.approved_vendors[vendor_id] = vendor
# Add vendor domains to appropriate directives
for domain in vendor.get('script_domains', []):
self.csp_directives['script-src'].append(domain)
for domain in vendor.get('connect_domains', []):
self.csp_directives['connect-src'].append(domain)
for domain in vendor.get('frame_domains', []):
if "'none'" in self.csp_directives['frame-src']:
self.csp_directives['frame-src'].remove("'none'")
self.csp_directives['frame-src'].append(domain)
def generate_csp_header(self) -> str:
"""Generate CSP header string"""
directives = []
for directive, sources in self.csp_directives.items():
directives.append(f"{directive} {' '.join(sources)}")
return '; '.join(directives)
def generate_report_only_header(self) -> str:
"""Generate CSP Report-Only header for testing"""
csp = self.generate_csp_header()
csp += f"; report-uri /api/csp-report"
return csp
```
### Deliverables
- [ ] Vendor approval workflow implemented
- [ ] All existing vendors documented and reviewed
- [ ] DPAs signed with all data processors
- [ ] CSP enforced to control script loading
- [ ] Regular vendor security assessments
- [ ] Vendor removal process documented
## 7. Consent Architecture: Centralize and Audit
### Why It Matters
Consent must be freely given, specific, informed, and unambiguous. It must also be as easy to withdraw as it was to give. More importantly, you need to prove consent was obtained correctly—consent without evidence is legally worthless.
### Implementation
**1. Centralized Consent Storage**
```python
# Centralized consent management
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime
import hashlib
@dataclass
class ConsentRecord:
id: str
user_id: str
purpose: str
legal_basis: str
granted: bool
granted_at: Optional[datetime]
withdrawn_at: Optional[datetime]
version: str # Version of consent language shown
collection_point: str # Where consent was collected
evidence: dict # Proof of consent
class ConsentService:
"""Centralized consent management with audit trail"""
CONSENT_PURPOSES = {
'essential': {
'description': 'Essential website functionality',
'legal_basis': 'contract',
'can_withdraw': False
},
'analytics': {
'description': 'Website usage analytics',
'legal_basis': 'consent',
'can_withdraw': True
},
'marketing': {
'description': 'Marketing communications',
'legal_basis': 'consent',
'can_withdraw': True
},
'personalization': {
'description': 'Personalized content and recommendations',
'legal_basis': 'consent',
'can_withdraw': True
},
'third_party_sharing': {
'description': 'Sharing data with partners',
'legal_basis': 'consent',
'can_withdraw': True
}
}
async def record_consent(
self,
user_id: str,
purpose: str,
granted: bool,
collection_context: dict
) -> ConsentRecord:
"""Record consent decision with evidence"""
evidence = {
'ip_address': collection_context.get('ip_address'),
'user_agent': collection_context.get('user_agent'),
'page_url': collection_context.get('page_url'),
'consent_text_shown': collection_context.get('consent_text'),
'consent_text_hash': self.hash_consent_text(collection_context.get('consent_text')),
'timestamp': datetime.utcnow().isoformat(),
'collection_method': collection_context.get('method', 'banner')
}
record = ConsentRecord(
id=str(uuid.uuid4()),
user_id=user_id,
purpose=purpose,
legal_basis=self.CONSENT_PURPOSES[purpose]['legal_basis'],
granted=granted,
granted_at=datetime.utcnow() if granted else None,
withdrawn_at=None,
version=collection_context.get('consent_version', '1.0'),
collection_point=collection_context.get('collection_point'),
evidence=evidence
)
await self.repository.save(record)
# Propagate consent to downstream systems
await self.propagate_consent(record)
return record
async def withdraw_consent(self, user_id: str, purpose: str) -> ConsentRecord:
"""Withdraw previously granted consent"""
current_consent = await self.get_current_consent(user_id, purpose)
if not current_consent or not current_consent.granted:
raise ValueError(f"No active consent found for {purpose}")
current_consent.granted = False
current_consent.withdrawn_at = datetime.utcnow()
await self.repository.update(current_consent)
# Propagate withdrawal to downstream systems
await self.propagate_withdrawal(current_consent)
return current_consent
async def get_consent_status(self, user_id: str) -> dict:
"""Get current consent status for all purposes"""
consents = await self.repository.get_user_consents(user_id)
status = {}
for purpose in self.CONSENT_PURPOSES:
consent = next((c for c in consents if c.purpose == purpose), None)
status[purpose] = {
'granted': consent.granted if consent else False,
'can_withdraw': self.CONSENT_PURPOSES[purpose]['can_withdraw'],
'granted_at': consent.granted_at if consent and consent.granted else None
}
return status
async def propagate_consent(self, consent: ConsentRecord):
"""Propagate consent to integrated systems"""
if consent.purpose == 'analytics' and consent.granted:
await self.analytics_service.enable_tracking(consent.user_id)
elif consent.purpose == 'analytics' and not consent.granted:
await self.analytics_service.disable_tracking(consent.user_id)
if consent.purpose == 'marketing':
await self.email_service.update_preferences(
consent.user_id,
marketing_enabled=consent.granted
)
```
### Deliverables
- [ ] Centralized consent database
- [ ] All consent collection points integrated
- [ ] Consent withdrawal equals consent grant in ease
- [ ] Consent proof stored with timestamps
- [ ] Consent preferences synced across systems
- [ ] Consent expiry and refresh mechanism
## 8. Training: Developers Need to Know What PII Is
### Why It Matters
Your best privacy controls are worthless if developers unknowingly log PII, commit secrets, or design features that violate privacy principles. Privacy must be part of engineering culture, not just security reviews.
### Implementation
**1. Required Training Curriculum**
```python
# Training tracking system
PRIVACY_TRAINING_REQUIREMENTS = {
'all_employees': [
{
'course': 'Privacy Fundamentals',
'duration_minutes': 60,
'topics': ['What is PII', 'Privacy regulations overview', 'Company policies'],
'frequency': 'annual'
}
],
'engineering': [
{
'course': 'Privacy Engineering',
'duration_minutes': 120,
'topics': [
'Privacy by design principles',
'Data minimization in code',
'Secure logging practices',
'PII detection and handling',
'Consent implementation',
'DSAR technical handling'
],
'frequency': 'annual'
},
{
'course': 'Secure Coding',
'duration_minutes': 180,
'topics': [
'OWASP Top 10',
'Input validation',
'Encryption practices',
'Secret management'
],
'frequency': 'annual'
}
],
'data_team': [
{
'course': 'Data Privacy for Analytics',
'duration_minutes': 90,
'topics': [
'Anonymization techniques',
'K-anonymity and differential privacy',
'Aggregate vs individual data',
'Data retention in warehouses'
],
'frequency': 'annual'
}
]
}
```
**2. Automated PII Detection in Code**
```python
# Pre-commit hook for PII detection
class PIIDetector:
"""Detect potential PII leaks in code"""
PII_PATTERNS = {
'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
'phone': r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
'ssn': r'\b\d{3}-\d{2}-\d{4}\b',
'credit_card': r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',
'api_key': r'(?i)(api[_-]?key|apikey)["\']?\s*[:=]\s*["\']?[\w-]{20,}',
'password': r'(?i)(password|passwd|pwd)["\']?\s*[:=]\s*["\']?[^\s"\']{8,}'
}
RISKY_LOG_PATTERNS = [
r'console\.log\s*\(\s*user',
r'logger\.(info|debug|error)\s*\([^)]*email',
r'print\s*\([^)]*password',
r'log\.[a-z]+\s*\([^)]*token'
]
def scan_file(self, filepath: str, content: str) -> List[dict]:
"""Scan file for potential PII issues"""
issues = []
# Check for hardcoded PII
for pii_type, pattern in self.PII_PATTERNS.items():
matches = re.finditer(pattern, content)
for match in matches:
issues.append({
'type': 'hardcoded_pii',
'pii_type': pii_type,
'file': filepath,
'line': content[:match.start()].count('\n') + 1,
'severity': 'high'
})
# Check for risky logging
for pattern in self.RISKY_LOG_PATTERNS:
matches = re.finditer(pattern, content)
for match in matches:
issues.append({
'type': 'risky_logging',
'file': filepath,
'line': content[:match.start()].count('\n') + 1,
'severity': 'medium',
'message': 'Potential PII in log statement'
})
return issues
```
### Deliverables
- [ ] Privacy training program developed
- [ ] All employees completed foundational training
- [ ] Engineering team completed technical privacy training
- [ ] Pre-commit hooks detect PII in code
- [ ] Code review checklist includes privacy
- [ ] Privacy champions in each team
## 9. Privacy Engineering: Hire Specialists
### Why It Matters
Privacy engineering is a specialized discipline. While every developer should understand privacy basics, complex implementations—differential privacy, secure computation, privacy-preserving analytics—require dedicated expertise.
### What to Look For
**Privacy Engineer Role Requirements:**
- Deep understanding of GDPR, CCPA, and other regulations
- Experience implementing consent management systems
- Knowledge of privacy-enhancing technologies (PETs)
- Ability to conduct Privacy Impact Assessments
- Experience with data anonymization techniques
- Understanding of secure development practices
- Communication skills to work with legal and product teams
### Building the Function
```
Privacy Engineering Team Structure:
┌─────────────────────────────────────────────────┐
│ Chief Privacy Officer │
│ (Strategic, Legal, Compliance) │
└─────────────────────┬───────────────────────────┘
│
┌─────────────────────┴───────────────────────────┐
│ Privacy Engineering Lead │
│ (Technical Architecture, Strategy) │
└─────────────────────┬───────────────────────────┘
│
┌─────────────────┼─────────────────┐
│ │ │
▼ ▼ ▼
┌─────────┐ ┌───────────┐ ┌──────────────┐
│ Privacy │ │ Consent │ │ Data Rights │
│Engineer │ │ Platform │ │ Automation │
│(Product)│ │ Engineer │ │ Engineer │
└─────────┘ └───────────┘ └──────────────┘
```
### Deliverables
- [ ] Privacy engineering job descriptions created
- [ ] At least one dedicated privacy engineer
- [ ] Privacy review in architecture process
- [ ] Privacy office hours for engineering teams
- [ ] Regular privacy engineering training
## 10. Audit: Regular Internal and External Assessments
### Why It Matters
You can't improve what you don't measure. Regular audits identify gaps before regulators do, provide evidence of due diligence, and drive continuous improvement in your privacy program.
### Implementation
**1. Automated Compliance Monitoring**
```python
class ComplianceMonitoringService:
"""Continuous compliance monitoring and reporting"""
COMPLIANCE_CHECKS = [
{
'name': 'consent_validity',
'description': 'Verify all processing has valid consent',
'frequency': 'daily',
'severity': 'critical'
},
{
'name': 'retention_compliance',
'description': 'Check data not retained beyond policy',
'frequency': 'daily',
'severity': 'high'
},
{
'name': 'encryption_at_rest',
'description': 'Verify all PII encrypted at rest',
'frequency': 'weekly',
'severity': 'critical'
},
{
'name': 'access_reviews',
'description': 'Verify access reviews completed',
'frequency': 'quarterly',
'severity': 'medium'
},
{
'name': 'vendor_dpa_status',
'description': 'Verify all vendors have current DPAs',
'frequency': 'monthly',
'severity': 'high'
},
{
'name': 'training_completion',
'description': 'Verify required training completed',
'frequency': 'monthly',
'severity': 'medium'
}
]
async def run_compliance_check(self, check_name: str) -> dict:
"""Execute single compliance check"""
check = next(c for c in self.COMPLIANCE_CHECKS if c['name'] == check_name)
result = {
'check_name': check_name,
'executed_at': datetime.utcnow(),
'status': 'unknown',
'findings': []
}
if check_name == 'consent_validity':
result = await self.check_consent_validity()
elif check_name == 'retention_compliance':
result = await self.check_retention_compliance()
# ... other checks
# Store result
await self.repository.save_check_result(result)
# Alert on failures
if result['status'] == 'failed':
await self.alert_compliance_failure(result)
return result
async def generate_compliance_report(self, period: str) -> dict:
"""Generate compliance report for period"""
results = await self.repository.get_results_for_period(period)
report = {
'period': period,
'generated_at': datetime.utcnow(),
'overall_status': 'compliant',
'checks_passed': 0,
'checks_failed': 0,
'findings': [],
'recommendations': []
}
for result in results:
if result['status'] == 'passed':
report['checks_passed'] += 1
else:
report['checks_failed'] += 1
report['overall_status'] = 'non_compliant'
report['findings'].extend(result['findings'])
return report
```
### Audit Schedule
| Audit Type | Frequency | Scope | Responsible |
|------------|-----------|-------|-------------|
| Automated compliance checks | Daily | All systems | Privacy Engineering |
| Access reviews | Quarterly | All privileged access | Security + HR |
| Vendor assessments | Annual | All data processors | Privacy + Legal |
| External penetration test | Annual | All external systems | Security |
| External privacy audit | Annual | Full privacy program | External firm |
| Regulatory readiness | Bi-annual | GDPR, CCPA compliance | Privacy + Legal |
### Deliverables
- [ ] Automated compliance monitoring deployed
- [ ] Quarterly internal audits scheduled
- [ ] Annual external audit planned
- [ ] Audit findings tracked to resolution
- [ ] Compliance dashboard for leadership
- [ ] Continuous improvement process documented
## Beyond the Checklist
Privacy isn't a checkbox—it's an ongoing engineering discipline. This checklist provides the foundation, but true privacy maturity comes from embedding these practices into your engineering culture, measuring continuously, and improving iteratively.
Start with the highest-impact items: data inventory, encryption, and access control provide the foundation. Build from there with incident response, vendor management, and consent architecture. Layer in training, specialized hiring, and audits as your program matures.
The organizations that excel at privacy aren't those that treat it as a compliance burden—they're those that recognize privacy as a competitive advantage, a trust-builder, and a reflection of their values. Start building that foundation today.