Security Architecture - Weaver AI
Version: 1.1.0 Last Updated: 2025-10-05 Classification: Internal
Table of Contents
- Security Overview
- Threat Model
- Security Controls
- Authentication & Authorization
- Data Security
- Network Security
- Audit & Compliance
- Security Testing
- Incident Response
Security Overview
Security Principles
Weaver AI is designed with defense-in-depth and security-by-default principles:
- Zero Trust: Every request is authenticated and authorized
- Least Privilege: Minimal permissions by default
- Fail Secure: Errors deny access rather than grant it
- Audit Everything: Comprehensive logging for security events
- Secure Defaults: Safe configurations out of the box
Security Architecture Diagram
┌─────────────────────────────────────────────────────────────┐
│ External Threat Surface │
└─────────────────────────────────────────────────────────────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ Layer 1: Transport Security │
│ - TLS 1.3 encryption │
│ - Certificate validation │
│ - HSTS enforcement │
└─────────────────────────────────────────────────────────────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ Layer 2: Authentication │
│ - JWT token validation (RSA-256) │
│ - API key verification (SHA-256) │
│ - User context extraction │
└─────────────────────────────────────────────────────────────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ Layer 3: Rate Limiting │
│ - Token bucket algorithm │
│ - Per-user quotas (Redis-backed) │
│ - Burst protection │
└─────────────────────────────────────────────────────────────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ Layer 4: Input Validation │
│ - Pydantic schema validation │
│ - SQL injection prevention │
│ - XSS/CSRF protection │
│ - Size limits enforcement │
└─────────────────────────────────────────────────────────────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ Layer 5: Authorization (RBAC) │
│ - Role-based access control │
│ - Capability-based permissions │
│ - Tool access enforcement │
└─────────────────────────────────────────────────────────────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ Layer 6: Policy Guards │
│ - Content filtering │
│ - Prompt injection detection │
│ - URL allowlist/denylist │
│ - PII detection (pre-processing) │
└─────────────────────────────────────────────────────────────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ Layer 7: Execution Isolation │
│ - MCP tool sandboxing │
│ - Resource limits (CPU, memory, time) │
│ - Network egress filtering │
└─────────────────────────────────────────────────────────────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ Layer 8: Output Sanitization │
│ - PII redaction (SSN, credit cards, etc.) │
│ - Sensitive data masking │
│ - Response size limits │
└─────────────────────────────────────────────────────────────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ Layer 9: Audit Logging │
│ - Structured security event logs │
│ - Immutable audit trail │
│ - Compliance reporting (GDPR, SOC2) │
└─────────────────────────────────────────────────────────────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ Layer 10: Cryptographic Signing (NEW) │
│ - RSA-256 signatures for tamper-evident logs │
│ - SHA-256 event hashing for integrity verification │
│ - Non-repudiation with cryptographic proof │
│ - Compliance: GDPR, SOX, HIPAA, PCI DSS │
└─────────────────────────────────────────────────────────────┘
Threat Model
STRIDE Analysis
Spoofing Identity
Threat: Attacker impersonates legitimate user to access system.
Mitigations:
- JWT signature validation with RSA public key
- API key cryptographic comparison (constant-time)
- Short-lived tokens (1-hour expiry recommended)
- Token rotation and revocation support
Tampering with Data
Threat: Request/response data modified in transit or at rest.
Mitigations:
- TLS 1.3 for all communications
- JWT signature prevents token tampering
- MCP tool calls signed with HMAC-SHA256
- Pydantic validation prevents malformed data
Repudiation
Threat: User denies performing action.
Mitigations:
- Comprehensive audit logging (see
weaver_ai/security/audit.py) - Cryptographically signed audit trails (RSA-256 signatures)
- SHA-256 event hashing for integrity verification
- Immutable log files with append-only mode
- Request/response logging with correlation IDs
- User ID tracked for all actions
- Non-repudiation through cryptographic proof (see
weaver_ai/telemetry.py)
Information Disclosure
Threat: Sensitive data exposed to unauthorized parties.
Mitigations:
- PII redaction enabled by default (
WEAVER_PII_REDACT=true) - Secrets stored in environment variables, never in code
- Redis connections secured with password authentication
- Error messages sanitized to avoid leaking internals
Denial of Service
Threat: System overwhelmed with requests.
Mitigations:
- Rate limiting (configurable RPS per user)
- Request size limits (10KB query, 50KB response)
- Timeout enforcement (25s default)
- Connection pooling to prevent resource exhaustion
Elevation of Privilege
Threat: Attacker gains higher permissions than authorized.
Mitigations:
- RBAC enforcement at every tool invocation
- Capability-based permissions (fail-closed)
- Tool approval workflow for sensitive operations
- Least privilege principle in role definitions
Attack Surface Analysis
| Component | Attack Vector | Risk | Mitigation |
|---|---|---|---|
| API Gateway | HTTP flood, credential stuffing | High | Rate limiting, auth required |
| LLM Integration | Prompt injection, jailbreaking | High | Input guards, output verification |
| MCP Tools | Malicious tool execution | High | Sandboxing, approval workflow |
| Redis | Unauthorized access | Medium | Password auth, network isolation |
| Environment Config | Secret exposure in logs | Medium | Redaction, secure storage |
| Dependencies | Vulnerable libraries | Medium | Automated scanning (pip-audit) |
Trust Boundaries
┌─────────────────────────────────────────────────────────┐
│ Untrusted Zone: Internet │
│ - User browsers │
│ - External applications │
│ - Malicious actors │
└─────────────────────────────────────────────────────────┘
│
↓ TLS, Auth, Rate Limit
┌─────────────────────────────────────────────────────────┐
│ DMZ: API Gateway │
│ - FastAPI endpoints │
│ - Auth/rate limit enforcement │
│ - Input validation │
└─────────────────────────────────────────────────────────┘
│
↓ Internal network
┌─────────────────────────────────────────────────────────┐
│ Trusted Zone: Internal Services │
│ - Agent orchestrator │
│ - Redis event mesh │
│ - MCP tool registry │
└─────────────────────────────────────────────────────────┘
│
↓ Controlled egress
┌─────────────────────────────────────────────────────────┐
│ Semi-Trusted: External APIs │
│ - LLM providers (OpenAI, Anthropic) │
│ - External tools (web search, databases) │
└─────────────────────────────────────────────────────────┘
Security Controls
Authentication Implementation
JWT Authentication (WEAVER_AUTH_MODE=jwt)
Token Format:
{
"sub": "user@example.com",
"roles": ["user"],
"exp": 1696368000,
"iat": 1696364400
}
Validation Steps (see weaver_ai/security/auth.py:40-60):
- Extract token from
Authorization: Bearer <token>header - Verify JWT signature using RSA public key
- Check expiration time (
expclaim) - Extract user ID from
subclaim - Load user roles from
rolesclaim - Create
UserContextobject
Key Management:
- Private key: Keep secure, never commit to repo
- Public key: Configured via
WEAVER_JWT_PUBLIC_KEYenv var - Key rotation: Update public key, grace period for old tokens
API Key Authentication (WEAVER_AUTH_MODE=api_key)
Validation Steps (see weaver_ai/security/auth.py:20-38):
- Extract key from
Authorization: Bearer <key>header - Compare against
WEAVER_ALLOWED_API_KEYS(constant-time comparison) - Create
UserContextwith key as user ID - Assign default "user" role
Best Practices:
- Generate keys with
openssl rand -hex 32 - Rotate keys quarterly
- Use different keys per environment (dev/staging/prod)
Authorization (RBAC)
Role Definition (weaver_ai/policies/roles.yaml)
user:
- tool:python_eval
- tool:web_search
- tool:documentation
admin:
- admin:all
- tool:*
analyst:
- tool:web_search
- tool:documentation
- tool:database
Permission Check Flow
# weaver_ai/security/rbac.py:30-50
def check_access(user: UserContext, scope: str, roles_path: Path) -> None:
roles = load_roles(roles_path)
user_scopes = get_user_scopes(user, roles)
# Wildcard support (e.g., "tool:*")
if scope in user_scopes or f"{scope.split(':')[0]}:*" in user_scopes:
return
# Admin override
if "admin:all" in user_scopes:
return
raise PermissionError(f"Access denied for scope: {scope}")
Tool Permissions (weaver_ai/policies/tools.yaml)
database_query:
required_scopes:
- tool:database
sensitivity: high
rate_limit: 30
requires_approval: true # Manual approval required
code_execution:
required_scopes:
- tool:code_execution
sensitivity: high
rate_limit: 10
requires_approval: true
Rate Limiting
Algorithm: Token Bucket (see weaver_ai/security/ratelimit.py)
Configuration:
WEAVER_RATELIMIT_RPS: Requests per second (default: 5)WEAVER_RATELIMIT_BURST: Burst capacity (default: 10)
Implementation:
# Pseudocode
def enforce(user_id: str, settings: AppSettings) -> None:
key = f"ratelimit:{user_id}"
bucket = redis.get(key)
if bucket is None:
# Initialize new bucket
bucket = {"tokens": settings.ratelimit_burst, "last_refill": now()}
# Refill tokens based on time elapsed
elapsed = now() - bucket["last_refill"]
refill = elapsed * settings.ratelimit_rps
bucket["tokens"] = min(bucket["tokens"] + refill, settings.ratelimit_burst)
if bucket["tokens"] < 1:
raise HTTPException(status_code=429, detail="Rate limit exceeded")
bucket["tokens"] -= 1
redis.set(key, bucket, ttl=60)
Input Validation
Pydantic Models (weaver_ai/models/api.py):
class QueryRequest(BaseModel):
user_id: str = Field(..., min_length=1, max_length=256)
query: str = Field(..., min_length=1, max_length=10000)
@field_validator("query")
def validate_query(cls, v: str) -> str:
# Prevent null byte injection
if "\x00" in v:
raise ValueError("Null bytes not allowed")
# Prevent control character injection
control_chars = [chr(i) for i in range(32) if i not in (9, 10, 13)]
if any(char in v for char in control_chars):
raise ValueError("Control characters not allowed")
return v
Policy Guards (weaver_ai/security/policy.py):
def input_guard(query: str, policies: dict) -> None:
# Blocked content filtering
blocked_patterns = policies.get("blocked_content", [])
for pattern in blocked_patterns:
if re.search(pattern, query, re.IGNORECASE):
raise HTTPException(status_code=400, detail="Query contains blocked content")
# URL filtering
urls = extract_urls(query)
for url in urls:
if not is_allowed_url(url, policies):
raise HTTPException(status_code=400, detail=f"URL not allowed: {url}")
Output Sanitization
PII Redaction (weaver_ai/security/policy.py:80-120):
Patterns Detected:
- SSN:
\d{3}-\d{2}-\d{4} - Credit card:
\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4} - Email:
[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,} - Phone:
\(\d{3}\)\s?\d{3}-\d{4}
Redaction:
def output_guard(text: str, policies: dict, redact: bool) -> GuardResult:
if not redact:
return GuardResult(text=text, redacted=False)
redacted_text = text
redacted_text = re.sub(r'\d{3}-\d{2}-\d{4}', '[SSN-REDACTED]', redacted_text)
redacted_text = re.sub(r'\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}', '[CARD-REDACTED]', redacted_text)
# ... more patterns
return GuardResult(text=redacted_text, redacted=True)
Data Security
Data Classification
| Data Type | Classification | Encryption | Retention |
|---|---|---|---|
| API Keys | Secret | Env vars only | Rotate quarterly |
| User queries | Confidential | TLS in transit | 30 days (audit logs) |
| LLM responses | Confidential | TLS in transit | 30 days (audit logs) |
| Agent memory | Confidential | Redis AUTH | Session-based |
| Audit logs | Confidential | File system permissions | 1 year |
| Metrics | Internal | TLS in transit | 90 days |
Encryption
In Transit:
- TLS 1.3 for all HTTP connections
- Redis AUTH for Redis connections (password-protected)
- HTTPS required for external API calls
At Rest:
- Environment variables for secrets (not in code)
- Redis persistence encrypted at volume level (K8s)
- Audit logs with filesystem permissions (600)
Secret Management
Development:
# .env file (gitignored)
WEAVER_MODEL_API_KEY=sk-proj-xxx
WEAVER_ALLOWED_API_KEYS=dev-key-1,dev-key-2
WEAVER_JWT_PUBLIC_KEY="-----BEGIN PUBLIC KEY-----\n..."
Production:
# Kubernetes Secret
apiVersion: v1
kind: Secret
metadata:
name: weaver-secrets
type: Opaque
data:
model-api-key: <base64-encoded>
jwt-public-key: <base64-encoded>
Best Practices:
- Use AWS Secrets Manager, HashiCorp Vault, or K8s Secrets
- Never commit secrets to version control
- Rotate secrets regularly
- Use different secrets per environment
Network Security
Firewall Rules
# Ingress
Allow: 443/tcp (HTTPS) from 0.0.0.0/0
Allow: 8000/tcp (HTTP) from internal load balancer
Deny: All other inbound
# Egress
Allow: 443/tcp (HTTPS) to LLM providers
Allow: 6379/tcp (Redis) to internal Redis cluster
Allow: 4317/tcp (OTLP) to telemetry collector
Deny: All other outbound
Network Segmentation
┌─────────────────────────────────────────────────┐
│ Public Internet (Untrusted) │
└─────────────────────────────────────────────────┘
│
↓ Firewall, TLS
┌─────────────────────────────────────────────────┐
│ DMZ: Load Balancer + WAF │
│ - TLS termination │
│ - DDoS protection │
│ - IP filtering │
└─────────────────────────────────────────────────┘
│
↓ Internal network
┌─────────────────────────────────────────────────┐
│ Application Tier: Weaver AI Pods │
│ - No direct internet access │
│ - Outbound via NAT gateway │
└─────────────────────────────────────────────────┘
│
↓ Private subnet
┌─────────────────────────────────────────────────┐
│ Data Tier: Redis Cluster │
│ - No internet access │
│ - VPC-only access │
└─────────────────────────────────────────────────┘
Audit & Compliance
Audit Logging
Events Logged (weaver_ai/security/audit.py):
- Authentication attempts (success/failure)
- Authorization failures
- Rate limit violations
- Query requests (user_id, query hash, timestamp)
- Tool executions (tool name, user, result)
- Policy violations (blocked content, disallowed URLs)
- Admin actions (policy changes, user management)
Log Format:
{
"timestamp": "2025-10-03T10:30:45.123Z",
"event_type": "auth_failure",
"user_id": "user@example.com",
"ip_address": "203.0.113.42",
"details": {
"reason": "invalid_token",
"token_expiry": "2025-10-03T09:00:00Z"
},
"correlation_id": "req-abc123"
}
Log Storage:
- File:
WEAVER_AUDIT_PATH(default:./audit.log) - Rotation: Daily, keep 365 days
- Permissions: 600 (owner read/write only)
- Immutable: Append-only mode
Signed Telemetry (Tamper-Evident Audit Trails)
Purpose: Provide cryptographic proof that audit events have not been tampered with, enabling non-repudiation and compliance.
Implementation (weaver_ai/telemetry.py):
class SignedEvent(BaseModel):
timestamp: str # ISO 8601 timestamp
event_type: str # e.g., "auth_failure", "policy_violation"
data: dict[str, Any] # Event-specific data
signature: str # RSA-256 JWT signature
event_hash: str # SHA-256 hash of event data
def log_security_event(
event_type: str,
signing_key: str | None,
**event_data
) -> SignedEvent | None:
"""Log security event with optional cryptographic signing."""
# 1. Compute SHA-256 hash of event data
event_hash = hashlib.sha256(
json.dumps(event_data, sort_keys=True).encode()
).hexdigest()
# 2. Create signed payload
payload = {
"timestamp": datetime.now(UTC).isoformat(),
"event_type": event_type,
"event_hash": event_hash
}
# 3. Sign with RSA private key
if signing_key:
signature = jwt.encode(payload, signing_key, algorithm="RS256")
return SignedEvent(
timestamp=payload["timestamp"],
event_type=event_type,
data=event_data,
signature=signature,
event_hash=event_hash
)
Verification (weaver_ai/telemetry.py:verify_event_signature):
def verify_event_signature(event: SignedEvent, public_key: str) -> bool:
"""Verify cryptographic signature of signed event."""
try:
# 1. Decode JWT signature with RSA public key
payload = jwt.decode(
event.signature,
public_key,
algorithms=["RS256"]
)
# 2. Verify timestamp matches
if payload["timestamp"] != event.timestamp:
return False
# 3. Verify event type matches
if payload["event_type"] != event.event_type:
return False
# 4. Recompute hash from data
computed_hash = hashlib.sha256(
json.dumps(event.data, sort_keys=True).encode()
).hexdigest()
# 5. Verify hash matches
if computed_hash != payload["event_hash"]:
return False
return True
except Exception:
return False
Configuration:
# Enable signed telemetry
WEAVER_TELEMETRY_SIGNING_ENABLED=true
# RSA private key for signing (PEM format)
WEAVER_TELEMETRY_SIGNING_KEY="<your-rsa-private-key-pem>"
# RSA public key for verification (PEM format)
WEAVER_TELEMETRY_VERIFICATION_KEY="<your-rsa-public-key-pem>"
Key Generation:
# Generate 2048-bit RSA key pair
openssl genrsa -out signing_key.pem 2048
openssl rsa -in signing_key.pem -pubout -out verify_key.pem
# Use in environment
export WEAVER_TELEMETRY_SIGNING_KEY="$(cat signing_key.pem)"
export WEAVER_TELEMETRY_VERIFICATION_KEY="$(cat verify_key.pem)"
Security Properties:
-
Non-Repudiation: Users cannot deny performing logged actions
- RSA-256 signature binds event to private key holder
- Only entity with private key can create valid signatures
-
Tamper-Evidence: Any modification breaks signature
- SHA-256 hash ensures data integrity
- Invalid signatures detected during verification
-
Authenticity: Events verifiable to originate from legitimate source
- Public key verification confirms signature validity
- Prevents forged events from appearing legitimate
-
Integrity: Event data hashing prevents undetected modifications
- Canonical JSON ensures consistent hashing
- Timestamp included in signed payload
Performance Characteristics:
- Signing: ~80ms per event (RSA-256)
- Verification: ~20ms per event
- Storage overhead: ~564 bytes per event (signature + hash)
Compliance Benefits:
| Standard | Requirement | How Signed Telemetry Helps |
|---|---|---|
| GDPR Article 32 | Integrity and confidentiality of processing | Tamper-evident logs prove data integrity |
| SOX Section 404 | Internal controls effectiveness | Cryptographic proof of audit trail integrity |
| HIPAA §164.312(b) | Audit controls with integrity verification | Signed events prevent unauthorized alteration |
| PCI DSS Req 10.5 | Secure audit trail storage | File integrity monitoring via signatures |
Best Practices:
-
Key Management
- Store private keys in secure vault (AWS Secrets Manager, HashiCorp Vault)
- Never commit keys to version control
- Use different keys per environment (dev/staging/prod)
- Rotate keys quarterly with grace period for verification
-
Selective Signing
- Sign security-critical events: auth failures, privilege escalations, policy violations
- Don't sign high-volume non-critical events (debug logs, metrics)
- Balance security with performance
-
Verification
- Verify signatures during security investigations
- Implement automated integrity checks
- Alert on signature verification failures
Compliance Support
GDPR:
- Right to erasure: Delete user data on request
- Right to access: Export user's audit logs
- Data minimization: Collect only necessary data
- PII redaction: Automatic in responses
SOC 2 Type II:
- Access controls: RBAC implementation
- Audit trails: Comprehensive logging
- Encryption: TLS for transit, at-rest for secrets
- Change management: Version control, ADRs
HIPAA (if applicable):
- PHI protection: PII redaction covers PHI
- Access logs: All data access logged
- Encryption: TLS 1.3 in transit
- Integrity controls: JWT signatures, hash verification
Security Testing
Automated Security Scanning
Pre-commit Hooks:
# .pre-commit-config.yaml
repos:
- repo: local
hooks:
- id: bandit
name: Bandit Security Scan
entry: bandit -r weaver_ai/
language: system
- id: pip-audit
name: Dependency Vulnerability Scan
entry: pip-audit
language: system
CI/CD Pipeline (.github/workflows/security.yml):
- name: Security Scan
run: |
pip install bandit pip-audit safety
bandit -r weaver_ai/ -f json -o bandit-report.json
pip-audit --format json -o pip-audit-report.json
safety check --json > safety-report.json
Penetration Testing
Test Scenarios (see docs/PENTEST_SUMMARY.md):
- Authentication bypass attempts
- SQL injection in query parameters
- XSS attacks in responses
- CSRF token validation
- Rate limit bypass
- JWT token manipulation
- Privilege escalation attempts
- PII exposure in logs
Tools Used:
- OWASP ZAP for automated scanning
- Burp Suite for manual testing
- Custom scripts for API testing
Security Review Checklist
Before deployment:
- All endpoints require authentication
- Rate limiting configured and tested
- PII redaction enabled
- Audit logging verified
- Secrets stored securely (not in code)
- TLS certificates valid
- Dependency vulnerabilities scanned
- Security tests passing
- Penetration test completed
- Incident response plan documented
Incident Response
Incident Classification
| Severity | Definition | Response Time |
|---|---|---|
| Critical | Data breach, system compromise | Immediate |
| High | Auth bypass, privilege escalation | 1 hour |
| Medium | Rate limit bypass, policy violation | 4 hours |
| Low | Failed login attempts, suspicious activity | 24 hours |
Response Procedure
-
Detection
- Automated alerts from audit logs
- Monitoring system anomalies
- User reports
-
Containment
- Revoke compromised tokens/keys
- Block malicious IPs
- Isolate affected services
-
Eradication
- Patch vulnerabilities
- Update security policies
- Rotate credentials
-
Recovery
- Restore from backups if needed
- Re-enable services
- Monitor for recurrence
-
Post-Incident
- Root cause analysis
- Update documentation
- Improve monitoring
- Security training
Contacts
- Security Team: security@example.com
- On-Call Engineer: Check PagerDuty rotation
- Compliance Officer: compliance@example.com
Security Roadmap
Planned Enhancements
-
Q4 2025
- Multi-factor authentication (MFA)
- OAuth 2.0 support
- Enhanced prompt injection detection
-
Q1 2026
- Zero-trust network architecture
- Hardware security module (HSM) for key storage
- Real-time security analytics dashboard
-
Q2 2026
- Bug bounty program
- Third-party security certification
- Automated compliance reporting
References
- OWASP Top 10
- NIST Cybersecurity Framework
- CIS Controls
- GDPR Compliance Guide
- Internal:
docs/PENTEST_SUMMARY.md
Related Documents
MCP Server Specification: Grok Discussion Server
A standalone Model Context Protocol (MCP) server that enables intelligent discussions with Grok-4 AI, featuring context-aware conversations, baseline document generation, and response management.
C13.6: DAG Visualization & Workflow Security
> **Parent:** [C13 Monitoring, Logging & Anomaly Detection](C13-Monitoring-and-Logging)
index
title: "Presidio: Open-Source Framework for PII Detection, Redaction & Anonymization"
Privacy Computing and Secure Execution Solutions - Comprehensive Research Report
**Date:** March 13, 2026