Compliance
Regulatory compliance for agent systems: GDPR, SOC2, HIPAA, PCI-DSS, AI regulations, and audit readiness.
Compliance Framework
Compliance Program
├── Governance
│ ├── Policy Management
│ ├── Risk Assessment
│ ├── Control Framework
│ └── Continuous Monitoring
├── Regulatory Requirements
│ ├── Data Protection (GDPR, CCPA, HIPAA)
│ ├── Financial (PCI-DSS, SOX)
│ ├── AI-Specific (EU AI Act, NIST AI RMF)
│ └── Industry-Specific (FedRAMP, StateRAMP)
├── Technical Controls
│ ├── Data Protection
│ ├── Access Control
│ ├── Audit Logging
│ ├── Encryption
│ └── Vulnerability Management
├── Operational
│ ├── Incident Response
│ ├── Vendor Management
│ ├── Training & Awareness
│ └── Audit & Assessment
└── Documentation
├── Policies & Procedures
├── Evidence Collection
├── Audit Trails
└── Compliance Reporting
Data Protection Regulations
GDPR Compliance
class GDPRCompliance:
def __init__(self, encryption: EncryptionService, audit_logger: AuditLogger):
self.encryption = encryption
self.audit = audit_logger
self.retention_policies = RetentionPolicyManager()
# Article 5: Lawfulness, fairness, transparency
async def process_lawfully(self, processing: DataProcessing) -> LawfulBasis:
"""Determine lawful basis for processing"""
bases = {
"consent": await self.check_consent(processing),
"contract": await self.check_contractual_necessity(processing),
"legal_obligation": await self.check_legal_obligation(processing),
"vital_interests": await self.check_vital_interests(processing),
"public_task": await self.check_public_task(processing),
"legitimate_interests": await self.assess_legitimate_interests(processing)
)
valid = [b for b, valid in bases.items() if valid]
if not valid:
raise GDPRViolation("No lawful basis for processing")
return LawfulBasis(primary=valid[0], all_bases=valid)
# Article 12-14: Transparency & Information
async def provide_privacy_notice(self, data_subject: DataSubject) -> PrivacyNotice:
return PrivacyNotice(
controller=CONTROLLER_INFO,
purposes=self.get_processing_purposes(),
lawful_basis=await self.get_lawful_basis(),
recipients=self.get_recipients(),
transfers=self.get_international_transfers(),
retention=self.get_retention_periods(),
rights=self.get_data_subject_rights(),
contact=DPO_CONTACT
)
# Article 15: Right of Access
async def handle_access_request(self, request: AccessRequest) -> AccessResponse:
# Verify identity
await self.verify_identity(request.data_subject)
# Collect data
data = await self.collect_personal_data(request.data_subject)
# Log request
await self.audit.log(AccessLog(
data_subject=request.data_subject,
request_type="access",
data_categories=self.categorize_data(data),
provided=True
))
return AccessResponse(data=data)
# Article 17: Right to Erasure
async def handle_erasure_request(self, request: ErasureRequest) -> ErasureResponse:
# Check if erasure applies
if not await self.erasure_applies(request):
return ErasureResponse(granted=False, reason="Erasure not applicable")
# Execute erasure
result = await self.execute_erasure(request.data_subject)
# Log
await self.audit.log(AccessLog(
data_subject=request.data_subject,
request_type="erasure",
result="completed",
records_affected=result.records_deleted
))
return ErasureResponse(granted=True, records_deleted=result.records_deleted)
# Article 20: Data Portability
async def handle_portability_request(self, request: PortabilityRequest) -> PortabilityResponse:
data = await self.export_portable_data(request.data_subject)
return PortabilityResponse(
data=data,
format="json",
includes=self.get_portable_categories()
)
# Article 25: Data Protection by Design & Default
async def privacy_impact_assessment(self, processing: DataProcessing) -> DPIA:
"""Data Protection Impact Assessment"""
risks = await self.identify_risks(processing)
mitigations = await self.design_mitigations(risks)
return DPIA(
processing=processing,
risks=risks,
mitigations=mitigations,
residual_risk=self.assess_residual_risk(risks, mitigations),
dpo_consulted=await self.consult_dpo(processing)
)
# Article 32: Security of Processing
async def implement_security_measures(self) -> SecurityMeasures:
return SecurityMeasures(
encryption=self.encryption.get_status(),
access_control=await self.audit_access_controls(),
pseudonymization=self.pseudonymization_status(),
resilience=self.test_resilience(),
restore_time=self.measure_restore_time(),
testing=self.schedule_security_testing()
)
# Article 33-34: Breach Notification
async def handle_breach(self, breach: DataBreach) -> BreachResponse:
# Assess risk
risk = await self.assess_breach_risk(breach)
# Notify supervisory authority (72 hours)
if risk.level in [RiskLevel.HIGH, RiskLevel.CRITICAL]:
await self.notify_authority(breach, within_hours=72)
# Notify data subjects (without undue delay)
if risk.level == RiskLevel.CRITICAL:
await self.notify_data_subjects(breach)
# Document
await self.document_breach(breach)
return BreachResponse(notified_authority=risk.level in [RiskLevel.HIGH, RiskLevel.CRITICAL],
notified_subjects=risk.level == RiskLevel.CRITICAL)
CCPA/CPRA Compliance
class CCPACompliance:
def __init__(self):
self.consumer_requests = ConsumerRequestManager()
self.sale_tracker = SaleTracker()
async def handle_consumer_request(self, request: ConsumerRequest) -> ConsumerResponse:
if request.type == RequestType.KNOW:
return await self.handle_know_request(request)
elif request.type == RequestType.DELETE:
return await self.handle_delete_request(request)
elif request.type == RequestType.OPT_OUT:
return await self.handle_opt_out_request(request)
elif request.type == RequestType.LIMIT_USE:
return await self.handle_limit_use_request(request)
async def handle_know_request(self, request: KnowRequest) -> KnowResponse:
# Categories of personal information collected
categories = await self.get_collection_categories(request.consumer)
# Specific pieces of personal information
pieces = await self.get_specific_pieces(request.consumer)
# Sources
sources = await self.get_sources(request.consumer)
# Business purpose
purposes = await self.get_business_purposes(request.consumer)
# Third parties
third_parties = await self.get_third_parties(request.consumer)
return KnowResponse(
categories=categories,
pieces=pieces,
sources=sources,
purposes=purposes,
third_parties=third_parties
)
async def handle_opt_out_request(self, request: OptOutRequest) -> OptOutResponse:
# Stop sale/sharing
await self.sale_tracker.stop_sale(request.consumer)
# Notify downstream
await self.notify_downstream(request.consumer)
return OptOutResponse(success=True, effective_date=datetime.utcnow())
AI-Specific Regulations
EU AI Act Compliance
class EUAIActCompliance:
def __init__(self):
self.risk_classifier = AIRiskClassifier()
self.conformity = ConformityAssessment()
async def classify_system(self, system: AISystem) -> RiskClassification:
"""Classify AI system per EU AI Act"""
# Unacceptable risk
if self.is_unacceptable_risk(system):
return RiskClassification.UNACCEPTABLE
# High risk
if self.is_high_risk(system):
return RiskClassification.HIGH_RISK
# Limited risk
if self.is_limited_risk(system):
return RiskClassification.LIMITED
# Minimal risk
return RiskClassification.MINIMAL
def is_high_risk(self, system: AISystem) -> bool:
# Annex III categories
high_risk_categories = [
"biometric_identification",
"critical_infrastructure",
"education_vocational_training",
"employment_worker_management",
"essential_services",
"law_enforcement",
"migration_asylum_border",
"administration_of_justice"
]
return system.domain in high_risk_categories
async def compliance_requirements(self, system: AISystem) -> ComplianceRequirements:
classification = await self.classify_system(system)
if classification == RiskClassification.UNACCEPTABLE:
return ComplianceRequirements(
allowed=False,
reason="Unacceptable risk under EU AI Act"
)
if classification == RiskClassification.HIGH_RISK:
return ComplianceRequirements(
requirements=[
"risk_management_system",
"data_governance",
"technical_documentation",
"record_keeping",
"transparency_user_info",
"human_oversight",
"accuracy_robustness_cybersecurity",
"conformity_assessment",
"ce_marking",
"registration"
],
conformity_assessment=await self.conformity.assess(system)
)
return ComplianceRequirements(
requirements=["transparency_obligations"] if classification == RiskClassification.LIMITED else []
)
NIST AI Risk Management Framework
class NISTAIRMFCompliance:
def __init__(self):
self.functions = {
"govern": GovernFunction(),
"map": MapFunction(),
"measure": MeasureFunction(),
"manage": ManageFunction()
}
async def assess(self, system: AISystem) -> NISTAssessment:
results = {}
for name, func in self.functions.items():
results[name] = await func.assess(system)
return NISTAssessment(
govern=results["govern"],
map=results["map"],
measure=results["measure"],
manage=results["manage"],
overall_score=self.calculate_score(results)
)
class GovernFunction:
async def assess(self, system: AISystem) -> GovernResult:
return GovernResult(
policies_exist=self.check_policies(system),
roles_defined=self.check_roles(system),
accountability=self.check_accountability(system),
culture=self.assess_culture(system)
)
class MapFunction:
async def assess(self, system: AISystem) -> MapResult:
return MapResult(
context_documented=self.check_context(system),
stakeholders_identified=self.check_stakeholders(system),
risks_identified=self.identify_risks(system),
benefits_documented=self.check_benefits(system)
)
class MeasureFunction:
async def assess(self, system: AISystem) -> MeasureResult:
return MeasureResult(
metrics_defined=self.check_metrics(system),
testing_performed=self.check_testing(system),
benchmarks_established=self.check_benchmarks(system),
monitoring_implemented=self.check_monitoring(system)
)
class ManageFunction:
async def assess(self, system: AISystem) -> ManageResult:
return ManageResult(
risks_prioritized=self.check_risk_prioritization(system),
resources_allocated=self.check_resources(system),
monitoring_continuous=self.check_continuous_monitoring(system),
improvement_process=self.check_improvement_process(system)
)
Industry-Specific Compliance
HIPAA (Healthcare)
class HIPAACompliance:
def __init__(self, encryption: EncryptionService, audit: AuditLogger):
self.encryption = encryption
self.audit = audit
async def ensure_compliance(self, system: System) -> HIPAAReport:
checks = await asyncio.gather(
self.check_encryption_at_rest(),
self.check_encryption_in_transit(),
self.check_access_controls(),
self.check_audit_controls(),
self.check_integrity_controls(),
self.check_transmission_security(),
self.check_business_associate_agreements(),
self.check_workforce_training(),
self.check_incident_response(),
self.check_contingency_plan()
)
return HIPAAReport(
compliant=all(c.passed for c in checks),
details=checks
)
async def check_encryption_at_rest(self) -> CheckResult:
# Verify AES-256 encryption for PHI at rest
return CheckResult(
passed=await self.verify_aes_256(),
details="All PHI encrypted with AES-256 at rest"
)
async def check_access_controls(self) -> CheckResult:
# Unique user IDs, emergency access, automatic logoff
return CheckResult(
passed=await self.verify_access_controls(),
details="Unique user IDs, automatic logoff, emergency access procedures"
)
PCI-DSS (Payment Cards)
class PCIDSSCompliance:
def __init__(self):
self.requirements = {
"1": "Firewall configuration",
"2": "Default passwords changed",
"3": "Protect stored cardholder data",
"4": "Encrypt transmission",
"5": "Anti-virus software",
"6": "Secure systems/applications",
"7": "Restrict access by need-to-know",
"8": "Unique IDs for access",
"9": "Physical access restriction",
"10": "Track and monitor access",
"11": "Regular testing",
"12": "Information security policy"
}
async def assess(self, system: System) -> PCIReport:
results = {}
for req_id, description in self.requirements.items():
check = getattr(self, f"check_req_{req_id}")()
results[req_id] = await check
return PCIReport(
compliant=all(r.passed for r in results.values()),
details=results,
attestation=self.generate_attestation(results)
)
SOX (Financial Reporting)
class SOXCompliance:
async def assess_financial_controls(self, system: System) -> SOXReport:
# Section 302: Corporate responsibility for financial reports
# Section 404: Management assessment of internal controls
# Section 409: Real-time issuer disclosures
controls = await asyncio.gather(
self.check_access_controls(),
self.check_change_management(),
self.check_segregation_of_duties(),
self.check_audit_trail_integrity(),
self.check_data_integrity(),
self.check_backup_recovery()
)
return SOXReport(
section_302=controls[0],
section_404=controls[1],
section_409=controls[2],
overall=all(c.passed for c in controls)
)
Compliance Automation
Continuous Compliance Monitoring
class ComplianceMonitor:
def __init__(self, frameworks: list[str]):
self.frameworks = frameworks
self.checks = self.load_checks(frameworks)
self.scheduler = AsyncScheduler()
async def start_monitoring(self):
# Schedule continuous checks
self.scheduler.every(1).hour.do(self.run_checks)
self.scheduler.every(1).day.do(self.generate_daily_report)
self.scheduler.every(7).days.do(self.generate_weekly_report)
async def run_checks(self):
results = await asyncio.gather(*[
check.execute() for check in self.checks
])
# Alert on failures
for result in results:
if not result.passed:
await self.alert_on_failure(result)
async def generate_evidence_package(self, framework: str) -> EvidencePackage:
"""Generate audit-ready evidence package"""
checks = await self.run_framework_checks(framework)
return EvidencePackage(
framework=framework,
generated_at=datetime.utcnow(),
controls=self.map_controls_to_evidence(framework),
test_results=self.get_test_results(),
artifacts=self.collect_artifacts(),
attestation=self.generate_attestation()
)
Policy as Code (OPA)
# policies/compliance.rego
package compliance
# GDPR: Personal data must be encrypted at rest
deny[msg] {
resource := input.resource
resource.type == "database"
not resource.encryption_at_rest
msg := sprintf("Database %s must have encryption at rest", [resource.name])
}
# GDPR: Personal data encrypted in transit
deny[msg] {
resource := input.resource
resource.type == "api_endpoint"
not resource.tls_enabled
msg := sprintf("API %s must use TLS", [resource.name])
}
# PCI-DSS: No default passwords
deny[msg] {
resource := input.resource
resource.type == "database"
resource.default_password == true
msg := sprintf("Database %s uses default password", [resource.name])
}
# HIPAA: Access control
deny[msg] {
resource := input.resource
resource.type == "phi_datastore"
not resource.unique_user_ids
msg := sprintf("PHI datastore %s lacks unique user IDs", [resource.name])
}
# SOX: Change management
deny[msg] {
change := input.change
change.type == "production_deployment"
not change.approved_by_manager
msg := sprintf("Production change %s not approved by manager", [change.id])
}
# Test
test_gdpr_encryption {
input := {
resource: {"type": "database", "name": "users_db", "encryption_at_rest": false}
}
deny[0] == "Database users_db must have encryption at rest"
}
Audit & Evidence
Audit Trail Generation
class AuditTrailGenerator:
def __init__(self, event_store: EventStore):
self.events = event_store
async def generate_audit_trail(self,
entity: str,
entity_id: str,
start: datetime,
end: datetime) -> AuditTrail:
events = await self.events.query(
entity=entity,
entity_id=entity_id,
start=start,
end=end
)
return AuditTrail(
entity=entity,
entity_id=entity_id,
period=(start, end),
events=events,
summary=self.generate_summary(events),
integrity_hash=self.compute_integrity_hash(events)
)
def generate_integrity_hash(self, events: list[Event]) -> str:
"""Hash chain for integrity verification"""
chain = ""
for event in events:
chain = hashlib.sha256(f"{chain}{event.id}{event.timestamp}".encode()).hexdigest()
return chain
Evidence Collection for Audits
class EvidenceCollector:
def __init__(self, storage: EvidenceStorage):
self.storage = storage
async def collect_evidence(self, framework: str,
control_id: str) -> EvidencePackage:
"""Collect all evidence for a specific control"""
evidence = EvidencePackage(
control_id=control_id,
framework=self.get_framework(control_id),
collected_at=datetime.utcnow(),
artifacts=[]
)
# Collect different evidence types
evidence.artifacts.extend(await self.collect_configs())
evidence.artifacts.extend(await self.collect_logs())
evidence.artifacts.extend(await self.collect_test_results())
evidence.artifacts.extend(await self.collect_procedures())
evidence.artifacts.extend(await self.collect_training_records())
evidence.artifacts.extend(await self.collect_incident_records())
# Package and sign
evidence.package_hash = self.compute_hash(evidence)
evidence.signed_by = await self.sign(evidence)
await self.storage.store(evidence)
return evidence
async def collect_configs(self) -> list[Artifact]:
artifacts = []
for config in await self.get_all_configs():
artifacts.append(Artifact(
type="configuration",
name=config.name,
content=config.content,
hash=self.hash(config.content),
collected_at=datetime.utcnow()
))
return artifacts
Vendor Risk Management
Vendor Assessment
class VendorRiskManager:
def __init__(self):
self.vendors: dict[str, VendorProfile] = {}
self.questionnaires = {
"security": SecurityQuestionnaire(),
"privacy": PrivacyQuestionnaire(),
"business_continuity": BCQuestionnaire(),
"financial": FinancialQuestionnaire()
}
async def assess_vendor(self, vendor: Vendor) -> VendorRiskAssessment:
# Send questionnaires
responses = await asyncio.gather(*[
q.send(vendor.contact) for q in self.questionnaires.values()
])
# Analyze responses
scores = {name: q.score(r) for (name, q), r in zip(self.questionnaires.items(), responses)}
# Security certifications
certs = await self.verify_certifications(vendor)
# Subprocessor review
subprocessors = await self.review_subprocessors(vendor)
return VendorRiskAssessment(
vendor=vendor,
overall_score=statistics.mean(scores.values()),
scores=scores,
certifications=certs,
subprocessors=subprocessors,
risk_level=self.calculate_risk_level(scores),
recommendations=self.generate_recommendations(scores)
)
async def continuous_monitoring(self, vendor_id: str):
"""Continuous monitoring of vendor risk"""
while True:
# Check for security incidents
incidents = await self.check_security_incidents(vendor_id)
# Check financial health
financial = await self.check_financial_health(vendor_id)
# Check certification status
certs = await self.verify_certifications(vendor_id)
if incidents or financial.declining or certs.expiring:
await self.escalate(vendor_id, incidents, financial, certs)
await asyncio.sleep(86400) # Daily
Training & Awareness
Compliance Training Program
class ComplianceTrainingProgram:
def __init__(self, lms: LMSClient):
self.lms = lms
self.courses = {
"gdpr_basics": Course("GDPR Basics", duration="2h", mandatory=True),
"hipaa_compliance": Course("HIPAA Compliance", duration="3h", mandatory=True),
"pci_dss": Course("PCI-DSS Basics", duration="2h", mandatory="role:payment"),
"secure_coding": Course("Secure Coding Practices", duration="4h", mandatory="role:developer"),
"incident_response": Course("Incident Response", duration="2h", mandatory="role:engineer"),
"ai_ethics": Course("AI Ethics & Safety", duration="2h", mandatory="role:ai_engineer"),
"data_handling": Course("Data Handling & Privacy", duration="1h", mandatory=True)
}
async def assign_training(self, user: User, role: str):
required = [c for c in self.courses.values()
if c.mandatory == True or c.mandatory == f"role:{role}"]
for course in required:
await self.lms.assign(user.id, course.id, deadline=timedelta(days=30))
async def track_completion(self, user_id: str) -> TrainingStatus:
completions = await self.lms.get_completions(user_id)
required = self.get_required_courses(user_id)
return TrainingStatus(
completed=[c for c in required if c.id in completions],
pending=[c for c in required if c.id not in completions],
overdue=[c for c in required if c.id not in completions and c.deadline < datetime.utcnow()],
compliance_rate=len([c for c in required if c.id in completions]) / len(required)
)
Related Notes
- Security — Security controls
- Observability — Audit logging
- Reliability — Incident response
- Deployment Architecture — Deployment compliance
- Cost Optimization — Cost compliance
- Secure Coding — Secure coding
- Supply Chain Security — Supply chain compliance
- Deployment Architecture — Deployment compliance
- Reliability — Incident response
- Observability — Audit logging
- Secure Coding — Secure coding
- Supply Chain Security — Supply chain compliance
- CI CD Best Practices — CI/CD compliance
- Infrastructure as Code — IaC compliance
- Research Directions — Compliance research
- Agent Societies — AI governance
- Secure Coding — Secure coding
- Supply Chain Security — Supply chain compliance
Sources: GDPR, CCPA/CPRA, HIPAA, PCI-DSS, SOX, EU AI Act, NIST AI RMF, ISO 27001, SOC 2, FedRAMP, StateRAMP, CMMC, ISO 42001, NIST AI RMF, ISO 22989, ISO 23894, ISO 42001, OWASP LLM Top 10, NIST AI RMF Playbook, MITRE ATLAS, OWASP LLM AI Security, FAIR Model, FAIR Institute, Factor Analysis of Information Risk, FAIR-U, OpenFAIR, RiskLens, RiskOptics, Vanta, Drata, Secureframe, Tugboat Logic, Hyperproof, AuditBoard, Workiva, AuditBoard, LogicGate, Riskonnect, MetricStream, RSA Archer, ServiceNow GRC, MetricStream, NAVEX Global, OneTrust, TrustArc, DataGrail, Ketch, Osano, Ethyca, Transcend, Mine, BigID, OneTrust, TrustArc, DataGrail.