← Back to Research
Overview
GDPR, CCPA, and privacy regulations built directly into message format and protocol, not bolted on afterward.
Problem Statement
Current email compliance is an afterthought: - GDPR/CCPA retrofitted onto 40-year-old protocols - No standard for data retention policies - No automated right-to-erasure - No consent tracking - Audit trails are custom implementations
msgs.global already has: app/compliance.py, app/gdpr_tools.py, app/compliance_reports.py
Vision
Compliance-Native Email Protocol where every message includes:
✅ Explicit consent tracking
✅ Automated retention policies
✅ Built-in right-to-erasure
✅ Cryptographic audit trail
✅ Data classification metadata
✅ Cross-border transfer controls
Compliance Headers
1. Data Processing Basis
X-GDPR-Lawful-Basis: consent|contract|legal_obligation|vital_interest|public_task|legitimate_interest
X-GDPR-Consent-ID: consent-abc123
X-GDPR-Consent-Timestamp: 2026-03-07T20:00:00Z
X-GDPR-Consent-Purpose: marketing|transactional|customer_service
X-GDPR-Consent-Scope: email_address,name,preferences
2. Retention Policy
X-Retention-Policy: 90-days|7-years|indefinite
X-Retention-Start: message_sent|message_read|last_accessed
X-Auto-Delete-After: 2026-06-07T00:00:00Z
X-Retention-Justification: legal_requirement|business_need|user_request
3. Data Classification
X-Data-Classification: public|internal|confidential|restricted
X-Contains-PII: true|false
X-PII-Types: name,email,phone,ssn,medical,financial
X-Data-Subject-Rights: erasure,portability,rectification,restriction
4. Cross-Border Transfer
X-Data-Location: US-East|EU-West|Asia-Pacific
X-Transfer-Mechanism: standard_contractual_clauses|adequacy_decision|binding_corporate_rules
X-Restricted-Countries: [list of countries message cannot enter]
Technical Implementation
Database Schema Extensions
-- Extend existing compliance tables
ALTER TABLE messages ADD COLUMN compliance_metadata JSONB;
-- Consent tracking (enhance existing)
CREATE TABLE consent_records (
id SERIAL PRIMARY KEY,
user_id INTEGER REFERENCES users(id),
consent_id VARCHAR(64) UNIQUE,
purpose VARCHAR(100), -- marketing, transactional, etc.
scope JSONB, -- {email_address: true, name: true, ...}
lawful_basis VARCHAR(50),
granted_at TIMESTAMP,
withdrawn_at TIMESTAMP,
ip_address INET,
user_agent TEXT,
proof_hash VARCHAR(64) -- Cryptographic proof
);
-- Retention policy enforcement
CREATE TABLE retention_policies (
id SERIAL PRIMARY KEY,
policy_id VARCHAR(64) UNIQUE,
message_pattern JSONB, -- Match criteria
retention_period INTERVAL,
deletion_method VARCHAR(50), -- soft_delete|hard_delete|anonymize
justification TEXT,
created_by INTEGER REFERENCES users(id),
created_at TIMESTAMP DEFAULT NOW()
);
-- Automated deletion queue
CREATE TABLE deletion_queue (
id SERIAL PRIMARY KEY,
message_id VARCHAR(255),
deletion_reason VARCHAR(100), -- retention_expired|user_request|legal_hold_lifted
scheduled_at TIMESTAMP,
executed_at TIMESTAMP,
verification_hash VARCHAR(64) -- Proof of deletion
);
-- Data subject requests (enhance GDPR tools)
CREATE TABLE data_subject_requests (
id SERIAL PRIMARY KEY,
request_id VARCHAR(64) UNIQUE,
user_id INTEGER REFERENCES users(id),
request_type VARCHAR(50), -- access|erasure|portability|rectification|restriction
status VARCHAR(30), -- pending|in_progress|completed|rejected
submitted_at TIMESTAMP DEFAULT NOW(),
completed_at TIMESTAMP,
response_data JSONB,
audit_trail JSONB
);
-- Compliance audit log
CREATE TABLE compliance_audit_log (
id SERIAL PRIMARY KEY,
event_type VARCHAR(50), -- consent_granted|data_deleted|policy_applied|cross_border_transfer
entity_type VARCHAR(30), -- message|user|consent|policy
entity_id VARCHAR(255),
event_data JSONB,
actor_id INTEGER REFERENCES users(id),
timestamp TIMESTAMP DEFAULT NOW(),
proof_signature VARCHAR(128) -- Tamper-proof audit
);
Compliance Engine
File: app/compliance_engine.py
from flask import Blueprint, request, jsonify
from app.models import db, ConsentRecord, RetentionPolicy, ComplianceAuditLog
from datetime import datetime, timedelta
compliance_engine_bp = Blueprint('compliance_engine', __name__)
class ComplianceEngine:
"""Automated compliance enforcement"""
def validate_message_compliance(self, message):
"""Validate message meets compliance requirements"""
errors = []
# Check consent
if not self.has_valid_consent(message.from_user, message.purpose):
errors.append('No valid consent for communication purpose')
# Check data classification
if self.contains_sensitive_data(message):
if not message.headers.get('X-Data-Classification'):
errors.append('Sensitive data requires classification header')
# Check cross-border transfer
if self.is_cross_border_transfer(message):
if not self.has_transfer_mechanism(message):
errors.append('Cross-border transfer requires legal mechanism')
# Check retention policy
if not self.has_retention_policy(message):
errors.append('No retention policy specified')
return {'valid': len(errors) == 0, 'errors': errors}
def has_valid_consent(self, user, purpose):
"""Check if user has granted consent for this purpose"""
consent = ConsentRecord.query.filter_by(
user_id=user.id,
purpose=purpose,
withdrawn_at=None
).first()
return consent is not None
def apply_retention_policy(self, message):
"""Apply retention policy to message"""
# Find matching policy
policy = self.find_retention_policy(message)
if policy:
# Calculate deletion date
if policy.retention_start == 'message_sent':
delete_after = message.sent_at + policy.retention_period
elif policy.retention_start == 'message_read':
delete_after = message.read_at + policy.retention_period
else:
delete_after = datetime.now() + policy.retention_period
# Schedule deletion
db.session.add(DeletionQueue(
message_id=message.message_id,
deletion_reason='retention_expired',
scheduled_at=delete_after
))
db.session.commit()
# Audit log
self.log_compliance_event(
'retention_policy_applied',
'message',
message.message_id,
{'policy': policy.policy_id, 'delete_after': delete_after}
)
def process_data_subject_request(self, request):
"""Process GDPR/CCPA data subject request"""
if request.request_type == 'access':
return self.export_user_data(request.user_id)
elif request.request_type == 'erasure':
return self.erase_user_data(request.user_id)
elif request.request_type == 'portability':
return self.export_portable_data(request.user_id)
elif request.request_type == 'rectification':
return self.rectify_user_data(request.user_id, request.corrections)
elif request.request_type == 'restriction':
return self.restrict_processing(request.user_id)
def erase_user_data(self, user_id):
"""Right to erasure (GDPR Article 17)"""
user = User.query.get(user_id)
# 1. Soft delete messages (anonymize)
messages = Message.query.filter_by(from_user_id=user_id).all()
for msg in messages:
msg.from_email = f'deleted-user-{user_id}@msgs.global'
msg.from_name = '[Deleted User]'
msg.content = '[Content deleted per user request]'
msg.compliance_metadata = {'erased': True, 'erased_at': datetime.now()}
# 2. Delete personal data
user.email = f'deleted-{user_id}@msgs.global'
user.phone = None
user.first_name = '[Deleted]'
user.last_name = '[User]'
# 3. Revoke consents
consents = ConsentRecord.query.filter_by(user_id=user_id).all()
for consent in consents:
consent.withdrawn_at = datetime.now()
# 4. Audit trail (prove deletion)
self.log_compliance_event(
'data_erased',
'user',
user_id,
{'erased_at': datetime.now(), 'verification_hash': self.hash_deletion_proof(user_id)}
)
db.session.commit()
return {'status': 'completed', 'user_id': user_id}
def log_compliance_event(self, event_type, entity_type, entity_id, event_data):
"""Tamper-proof compliance audit log"""
event = ComplianceAuditLog(
event_type=event_type,
entity_type=entity_type,
entity_id=entity_id,
event_data=event_data,
timestamp=datetime.now()
)
# Sign event for tamper-proofing
event.proof_signature = self.sign_audit_event(event)
db.session.add(event)
db.session.commit()
# API Endpoints
@compliance_engine_bp.route('/api/v1/compliance/validate', methods=['POST'])
def validate_compliance():
"""Validate message compliance before sending"""
message_data = request.json
engine = ComplianceEngine()
result = engine.validate_message_compliance(message_data)
if not result['valid']:
return jsonify(result), 400
return jsonify(result), 200
@compliance_engine_bp.route('/api/v1/compliance/consent', methods=['POST'])
def grant_consent():
"""User grants consent"""
data = request.json
consent = ConsentRecord(
user_id=request.current_user.id,
consent_id=generate_consent_id(),
purpose=data['purpose'],
scope=data['scope'],
lawful_basis=data.get('lawful_basis', 'consent'),
granted_at=datetime.now(),
ip_address=request.remote_addr,
user_agent=request.headers.get('User-Agent')
)
# Proof hash (verifiable consent)
consent.proof_hash = hash_consent_record(consent)
db.session.add(consent)
db.session.commit()
return jsonify(consent.to_dict()), 201
@compliance_engine_bp.route('/api/v1/compliance/data-subject-request', methods=['POST'])
def submit_data_subject_request():
"""Submit GDPR/CCPA request"""
data = request.json
dsr = DataSubjectRequest(
request_id=generate_request_id(),
user_id=request.current_user.id,
request_type=data['type'],
status='pending'
)
db.session.add(dsr)
db.session.commit()
# Process asynchronously
process_dsr_async.delay(dsr.id)
return jsonify({
'request_id': dsr.request_id,
'status': 'submitted',
'eta': '30 days' # GDPR requirement
}), 202
@compliance_engine_bp.route('/api/v1/compliance/audit-log')
def get_audit_log():
"""Get compliance audit log (admin only)"""
if not request.current_user.is_admin:
return jsonify({'error': 'Unauthorized'}), 403
logs = ComplianceAuditLog.query.order_by(
ComplianceAuditLog.timestamp.desc()
).limit(1000).all()
return jsonify([log.to_dict() for log in logs])
Automated Retention Enforcement
Background Job: scripts/enforce_retention.py
#!/usr/bin/env python3
"""Automated data retention enforcement"""
from app import app, db
from app.models import Message, DeletionQueue
from datetime import datetime
def enforce_retention_policies():
"""Run daily to enforce retention"""
with app.app_context():
# Get messages scheduled for deletion
due_for_deletion = DeletionQueue.query.filter(
DeletionQueue.scheduled_at <= datetime.now(),
DeletionQueue.executed_at == None
).all()
for item in due_for_deletion:
# Delete or anonymize message
message = Message.query.filter_by(
message_id=item.message_id
).first()
if message:
# Hard delete or anonymize based on policy
if item.deletion_method == 'hard_delete':
db.session.delete(message)
else:
# Anonymize
message.content = '[Deleted per retention policy]'
message.compliance_metadata = {
'anonymized': True,
'anonymized_at': datetime.now(),
'reason': item.deletion_reason
}
# Mark as executed
item.executed_at = datetime.now()
item.verification_hash = hash_deletion_proof(message)
db.session.commit()
print(f"Processed {len(due_for_deletion)} retention deletions")
if __name__ == '__main__':
enforce_retention_policies()
Webmail Compliance UI
Feature: Compliance dashboard in webmail
// Display consent status
function showConsentStatus() {
fetch('/api/v1/compliance/consent/status')
.then(r => r.json())
.then(consents => {
consents.forEach(c => {
displayConsent({
purpose: c.purpose,
granted: c.granted_at,
status: c.withdrawn_at ? 'withdrawn' : 'active',
canWithdraw: true
});
});
});
}
// Data subject request UI
function submitDataSubjectRequest(type) {
fetch('/api/v1/compliance/data-subject-request', {
method: 'POST',
body: JSON.stringify({type: type}),
headers: {'Content-Type': 'application/json'}
})
.then(r => r.json())
.then(result => {
showNotification(`Request submitted: ${result.request_id}. ETA: ${result.eta}`);
});
}
// Show data retention status
function showRetentionInfo(message) {
const retention = message.compliance_metadata?.retention;
if (retention) {
displayBadge({
text: `Auto-deletes: ${retention.delete_after}`,
color: 'warning'
});
}
}
Compliance Certifications
This enables msgs.global to achieve: - ✅ GDPR Compliance (EU) - ✅ CCPA Compliance (California) - ✅ HIPAA (with additional controls) - ✅ SOC 2 Type II - ✅ ISO 27001
Advantages
- Automated compliance (not manual processes)
- Cryptographic proof (tamper-proof audit logs)
- Built-in by design (not retrofitted)
- User self-service (data subject requests automated)
- Transparent (users see exactly what data you have)
Status
🚀 FOUNDATION ALREADY EXISTS (compliance.py, gdpr_tools.py)
Next Steps
- [ ] Add compliance headers to message format
- [ ] Implement automated retention enforcement
- [ ] Build consent management UI
- [ ] Cryptographic audit log signing
- [ ] Data subject request automation
- [ ] Cross-border transfer controls
Integration with Existing Code
- Enhance
app/compliance.pywith automated enforcement - Extend
app/gdpr_tools.pywith DSR automation - Upgrade
app/compliance_reports.pywith real-time metrics