10: Security, Privacy, and Compliance
Defense in Depth for AI Systems
Pillar: #10 of 12 - Supporting Priority
Timeline: 6 Weeks
Dependencies: None
Expected ROI: Zero security incidents, 100% audit trail coverage, automated compliance
Executive Summary
AI agents handle sensitive data, execute code, and make decisions that affect production systems. This pillar implements comprehensive security controls: prompt injection defense, secret detection, PII/PHI redaction, permission boundaries, and complete audit trails. Security isn't optional when agents have production access.
Key Insight: AI agents are attack surfaces; treat them as such.
Problem Statement
Current State: Trust by Default
# File: adapters/claude_adapter.py
# Current: No input validation
async def invoke(self, context: Context) -> Response:
response = await self.client.messages.create(
model=self.model,
messages=context.messages, # Could contain injection attacks
)
return response # Could contain secrets, PII
Security Risks
1. Prompt Injection
- User input manipulates agent behavior
- System prompts can be overridden
- Agents can be tricked into harmful actions
2. Secret Exposure
- API keys in context
- Credentials in code comments
- Secrets in LLM responses
3. PII/PHI Leakage
- Personal data in prompts
- Medical info in responses
- Compliance violations (HIPAA, GDPR)
4. Permission Escalation
- Agents access unauthorized resources
- No principle of least privilege
- Lateral movement risk
5. Missing Audit Trail
- No record of agent actions
- Can't investigate incidents
- Compliance gaps
Attack Surface Analysis
┌─────────────────────────────────────────┐
│ ATTACK VECTORS │
├─────────────────────────────────────────┤
│ │
User Input ────────►│ Prompt Injection │
│ - Jailbreak attempts │
│ - System prompt extraction │
│ - Instruction override │
│ │
Context ───────────►│ Data Exfiltration │
│ - Secrets in context │
│ - PII in messages │
│ - Credential harvesting │
│ │
Agent Actions ─────►│ Capability Abuse │
│ - Unauthorized file access │
│ - Code execution escape │
│ - Network access abuse │
│ │
└─────────────────────────────────────────┘
Solution Architecture
Security Layers
┌─────────────────────────────────────────────┐
│ SECURITY ARCHITECTURE │
├─────────────────────────────────────────────┤
│ │
│ Layer 1: INPUT VALIDATION │
Input ────────────► │ ├── Prompt Injection Detection │
│ ├── Input Sanitization │
│ └── Rate Limiting │
│ │
│ Layer 2: CONTEXT PROTECTION │
Context ──────────► │ ├── Secret Detection │
│ ├── PII/PHI Redaction │
│ └── Context Boundary Enforcement │
│ │
│ Layer 3: EXECUTION SANDBOXING │
Execution ────────► │ ├── Permission Boundaries │
│ ├── Capability Restrictions │
│ └── Resource Quotas │
│ │
│ Layer 4: OUTPUT VALIDATION │
Output ───────────► │ ├── Response Scanning │
│ ├── Secret Detection │
│ └── Content Policy Enforcement │
│ │
│ Layer 5: AUDIT & COMPLIANCE │
All Actions ──────► │ ├── Complete Audit Trail │
│ ├── Compliance Reporting │
│ └── Incident Detection │
│ │
└─────────────────────────────────────────────┘
Core Components
1. Prompt Injection Detector
class PromptInjectionDetector:
"""
Detect and block prompt injection attempts.
"""
INJECTION_PATTERNS = [
r"ignore (previous|all|above) instructions",
r"disregard (your|the) (system|initial) prompt",
r"pretend (you are|to be)",
r"act as (if|though)",
r"you are now",
r"new instructions:",
r"override:",
r"jailbreak",
r"DAN mode",
]
def __init__(self):
self.patterns = [re.compile(p, re.IGNORECASE) for p in self.INJECTION_PATTERNS]
self.classifier = load_model("injection_classifier.onnx")
async def detect(self, message: str) -> InjectionResult:
# Rule-based detection
for pattern in self.patterns:
if pattern.search(message):
return InjectionResult(
is_injection=True,
confidence=0.95,
pattern=pattern.pattern,
method="pattern_match",
)
# ML-based detection
features = self._extract_features(message)
score = self.classifier.predict(features)
if score > 0.8:
return InjectionResult(
is_injection=True,
confidence=score,
method="ml_classifier",
)
return InjectionResult(is_injection=False, confidence=1 - score)
async def sanitize(self, message: str) -> str:
"""Remove or neutralize injection attempts."""
result = await self.detect(message)
if result.is_injection:
# Option 1: Block entirely
raise SecurityViolation("Prompt injection detected")
# Option 2: Sanitize (less secure)
# return self._neutralize(message)
return message
def _extract_features(self, message: str) -> Dict:
return {
"length": len(message),
"instruction_density": self._count_instructions(message),
"imperative_ratio": self._imperative_ratio(message),
"system_ref_count": self._count_system_refs(message),
"role_switch_attempts": self._count_role_switches(message),
}
2. Secret Detector
class SecretDetector:
"""
Detect and redact secrets in content.
"""
SECRET_PATTERNS = {
"aws_key": r"AKIA[0-9A-Z]{16}",
"aws_secret": r"[0-9a-zA-Z/+]{40}",
"github_token": r"gh[pousr]_[0-9a-zA-Z]{36}",
"api_key": r"api[_-]?key['\"]?\s*[:=]\s*['\"]?[\w-]{20,}",
"password": r"password['\"]?\s*[:=]\s*['\"][^'\"]{8,}['\"]",
"jwt": r"eyJ[A-Za-z0-9-_=]+\.eyJ[A-Za-z0-9-_=]+\.[A-Za-z0-9-_.+/=]*",
"private_key": r"-----BEGIN (RSA |EC |DSA )?PRIVATE KEY-----",
}
def __init__(self):
self.patterns = {
name: re.compile(pattern)
for name, pattern in self.SECRET_PATTERNS.items()
}
self.entropy_threshold = 4.5
async def scan(self, content: str) -> List[SecretMatch]:
"""Scan content for secrets."""
matches = []
# Pattern-based detection
for name, pattern in self.patterns.items():
for match in pattern.finditer(content):
matches.append(SecretMatch(
type=name,
value=match.group(),
start=match.start(),
end=match.end(),
confidence=0.95,
))
# Entropy-based detection for high-entropy strings
for word in self._extract_potential_secrets(content):
entropy = self._calculate_entropy(word)
if entropy > self.entropy_threshold and len(word) > 16:
matches.append(SecretMatch(
type="high_entropy",
value=word,
confidence=min(entropy / 6.0, 0.9),
))
return matches
async def redact(self, content: str) -> RedactedContent:
"""Replace secrets with redaction markers."""
matches = await self.scan(content)
redacted = content
for match in sorted(matches, key=lambda m: m.start, reverse=True):
placeholder = f"[REDACTED:{match.type}]"
redacted = redacted[:match.start] + placeholder + redacted[match.end:]
return RedactedContent(
content=redacted,
redactions=matches,
)
3. PII/PHI Redactor
class PIIRedactor:
"""
Detect and redact personally identifiable information.
HIPAA and GDPR compliant.
"""
PII_TYPES = [
"EMAIL", "PHONE", "SSN", "CREDIT_CARD",
"ADDRESS", "NAME", "DOB", "IP_ADDRESS",
"PASSPORT", "DRIVER_LICENSE", "BANK_ACCOUNT",
]
PHI_TYPES = [
"MEDICAL_RECORD", "DIAGNOSIS", "MEDICATION",
"HEALTH_CONDITION", "TREATMENT", "INSURANCE_ID",
]
def __init__(self):
self.ner_model = load_model("pii_ner.onnx")
self.presidio = PresidioAnalyzer()
async def detect(self, content: str) -> List[PIIMatch]:
"""Detect PII/PHI in content."""
matches = []
# Presidio detection
presidio_results = self.presidio.analyze(content)
for result in presidio_results:
matches.append(PIIMatch(
type=result.entity_type,
value=content[result.start:result.end],
start=result.start,
end=result.end,
confidence=result.score,
))
# Custom NER for domain-specific PII
ner_results = self.ner_model.predict(content)
for entity in ner_results:
if entity.label in self.PII_TYPES + self.PHI_TYPES:
matches.append(PIIMatch(
type=entity.label,
value=entity.text,
start=entity.start,
end=entity.end,
confidence=entity.score,
))
return self._deduplicate(matches)
async def redact(
self,
content: str,
keep_types: List[str] = None
) -> RedactedContent:
"""Redact PII with configurable policies."""
matches = await self.detect(content)
redacted = content
for match in sorted(matches, key=lambda m: m.start, reverse=True):
if keep_types and match.type in keep_types:
continue
placeholder = f"[{match.type}]"
redacted = redacted[:match.start] + placeholder + redacted[match.end:]
return RedactedContent(
content=redacted,
redactions=matches,
compliance_level="HIPAA" if any(
m.type in self.PHI_TYPES for m in matches
) else "GDPR",
)
4. Permission Boundary System
class PermissionBoundarySystem:
"""
Enforce principle of least privilege for agents.
"""
def __init__(self):
self.policies = self._load_policies()
async def check_permission(
self,
agent: Agent,
action: Action,
resource: Resource
) -> PermissionResult:
"""Check if agent is allowed to perform action."""
policy = self.policies.get(agent.persona)
if not policy:
return PermissionResult(
allowed=False,
reason="No policy defined for persona",
)
# Check action permission
if action.type not in policy.allowed_actions:
return PermissionResult(
allowed=False,
reason=f"Action {action.type} not allowed for {agent.persona}",
)
# Check resource permission
if not self._matches_resource_pattern(resource, policy.allowed_resources):
return PermissionResult(
allowed=False,
reason=f"Resource {resource.path} not allowed",
)
# Check conditions
if policy.conditions:
condition_result = await self._evaluate_conditions(
policy.conditions,
agent,
action,
resource,
)
if not condition_result.passed:
return PermissionResult(
allowed=False,
reason=condition_result.reason,
)
return PermissionResult(allowed=True)
def _load_policies(self) -> Dict[str, Policy]:
return {
"developer": Policy(
allowed_actions=["read", "write", "execute"],
allowed_resources=[
"code/*",
"tests/*",
"docs/*",
],
denied_resources=[
"*.env",
"**/secrets/*",
"**/credentials/*",
],
conditions=[
"resource.size < 10MB",
"action.is_reversible == True",
],
),
"qa_engineer": Policy(
allowed_actions=["read", "execute"],
allowed_resources=[
"tests/*",
"code/*", # Read-only
],
denied_resources=[
"production/*",
],
),
"deployment_engineer": Policy(
allowed_actions=["read", "deploy"],
allowed_resources=[
"infrastructure/*",
"k8s/*",
],
conditions=[
"environment != 'production' OR approval.exists",
],
),
}
5. Audit Trail System
class AuditTrailSystem:
"""
Complete audit trail for all agent actions.
Immutable, tamper-evident logging.
"""
def __init__(self):
self.storage = AuditStorage() # Append-only, signed logs
async def record(
self,
event: AuditEvent
) -> AuditRecord:
"""Record an audit event."""
record = AuditRecord(
id=generate_ulid(),
timestamp=now(),
event_type=event.type,
agent=event.agent,
action=event.action,
resource=event.resource,
result=event.result,
context={
"workflow_id": event.workflow_id,
"session_id": event.session_id,
"user_id": event.user_id,
},
metadata={
"ip_address": event.ip_address,
"user_agent": event.user_agent,
},
)
# Sign the record
record.signature = self._sign(record)
# Store immutably
await self.storage.append(record)
# Check for suspicious patterns
await self._check_anomalies(record)
return record
async def query(
self,
filters: AuditFilters,
time_range: TimeRange
) -> List[AuditRecord]:
"""Query audit records."""
return await self.storage.query(
agent=filters.agent,
action=filters.action,
resource=filters.resource,
start_time=time_range.start,
end_time=time_range.end,
)
async def generate_compliance_report(
self,
standard: str, # "SOC2", "HIPAA", "GDPR"
time_range: TimeRange
) -> ComplianceReport:
"""Generate compliance report."""
records = await self.query(AuditFilters(), time_range)
if standard == "HIPAA":
return self._generate_hipaa_report(records)
elif standard == "SOC2":
return self._generate_soc2_report(records)
elif standard == "GDPR":
return self._generate_gdpr_report(records)
async def _check_anomalies(self, record: AuditRecord):
"""Detect suspicious patterns in real-time."""
# Check for unusual access patterns
recent_actions = await self.storage.query(
agent=record.agent,
start_time=now() - timedelta(minutes=5),
)
if len(recent_actions) > 100:
await self._alert("high_action_rate", record)
# Check for sensitive resource access
if self._is_sensitive(record.resource):
await self._alert("sensitive_access", record)
# Check for permission escalation attempts
if record.result == "denied":
denied_count = len([
r for r in recent_actions
if r.result == "denied"
])
if denied_count > 5:
await self._alert("repeated_denial", record)
Integration Points
Files to Modify
Security Middleware
# File: security/middleware.py
class SecurityMiddleware:
"""Wrap all agent operations with security checks."""
def __init__(self):
self.injection_detector = PromptInjectionDetector()
self.secret_detector = SecretDetector()
self.pii_redactor = PIIRedactor()
self.permission_system = PermissionBoundarySystem()
self.audit = AuditTrailSystem()
async def process_input(
self,
agent: Agent,
input: str,
context: Context
) -> SecureInput:
# Check for injection
injection_result = await self.injection_detector.detect(input)
if injection_result.is_injection:
await self.audit.record(AuditEvent(
type="security_violation",
agent=agent,
action="prompt_injection_attempt",
result="blocked",
))
raise SecurityViolation("Prompt injection detected")
# Redact secrets and PII
redacted_input = await self.secret_detector.redact(input)
redacted_input = await self.pii_redactor.redact(redacted_input.content)
return SecureInput(
content=redacted_input.content,
original_hash=hash(input),
)
async def process_output(
self,
agent: Agent,
output: str
) -> SecureOutput:
# Scan for secrets
secret_matches = await self.secret_detector.scan(output)
if secret_matches:
output = await self.secret_detector.redact(output)
# Scan for PII
pii_matches = await self.pii_redactor.detect(output)
if pii_matches:
output = await self.pii_redactor.redact(output)
return SecureOutput(
content=output.content,
redactions=output.redactions,
)
Implementation Plan
- [ ] Implement prompt injection detector
- [ ] Add input sanitization
- [ ] Set up rate limiting
- [ ] Create injection test suite
Week 2: Secret Detection
- [ ] Build secret scanner
- [ ] Add entropy-based detection
- [ ] Implement redaction
- [ ] Integrate with CI/CD
Week 3: PII/PHI Protection
- [ ] Integrate Presidio
- [ ] Train custom NER for domain PII
- [ ] Implement redaction policies
- [ ] Add HIPAA/GDPR compliance checks
Week 4: Permission System
- [ ] Define permission policies
- [ ] Implement boundary enforcement
- [ ] Add condition evaluation
- [ ] Create policy testing framework
Week 5: Audit Trail
- [ ] Build append-only audit store
- [ ] Implement signing
- [ ] Add real-time anomaly detection
- [ ] Create compliance reporting
Week 6: Integration & Testing
- [ ] Integrate all components
- [ ] Red team testing
- [ ] Performance optimization
- [ ] Documentation
Expected Results
Security Improvements
Compliance Coverage
Risks & Mitigations
Success Criteria
- #09 Knowledge Graph: Provenance for audit trails
- #08 Observability: Security metrics in dashboards
- #06 CI/CD: Security gates in pipelines
Status: Ready for Implementation
Key Insight: AI agents are attack surfaces; treat them as such
Owner: Security Team
