aibot88 Audit Logging
Track every important action for compliance and debugging.
When to Use This Skill
- SOC 2 / HIPAA compliance
- Financial transaction tracking
- User action history
- Security incident investigation
- Data change tracking
Architecture
┌─────────────────────────────────────────────────────┐
│ Application │
│ │
│ auditLog.record({ │
│ action: "user.login", │
│ actor: userId, │
│ resource: "session", │
│ details: { ip, userAgent } │
│ }) │
└─────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ Audit Log Service │
│ │
│ - Enrich with context │
│ - Validate schema │
│ - Queue for async write │
└─────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ Storage (Append-Only) │
│ │
│ - PostgreSQL (with triggers) │
│ - S3 (immutable objects) │
│ - CloudWatch Logs │
└─────────────────────────────────────────────────────┘
TypeScript Implementation
Audit Log Service
// audit-log.ts
import { v4 as uuid } from 'uuid';
interface AuditEvent {
action: string; // e.g., "user.login", "order.create"
actor: {
id: string;
type: 'user' | 'system' | 'api_key';
email?: string;
};
resource: {
type: string; // e.g., "user", "order", "payment"
id?: string;
};
details?: Record<string, unknown>;
outcome: 'success' | 'failure';
reason?: string; // For failures
}
interface AuditRecord extends AuditEvent {
id: string;
timestamp: Date;
requestId?: string;
ip?: string;
userAgent?: string;
organizationId?: string;
}
class AuditLogService {
private context: AsyncLocalStorage<{ requestId?: string; ip?: string; userAgent?: string }>;
constructor() {
this.context = new AsyncLocalStorage();
}
// Set request context (call from middleware)
setContext(ctx: { requestId?: string; ip?: string; userAgent?: string }) {
return this.context.run(ctx, () => {});
}
async record(event: AuditEvent): Promise<void> {
const ctx = this.context.getStore() || {};
const record: AuditRecord = {
id: uuid(),
timestamp: new Date(),
...event,
requestId: ctx.requestId,
ip: ctx.ip,
userAgent: ctx.userAgent,
};
// Write to database (append-only table)
await db.auditLogs.create({ data: record });
// Also send to external logging (CloudWatch, DataDog, etc.)
if (process.env.AUDIT_LOG_STREAM) {
await this.sendToCloudWatch(record);
}
}
// Convenience methods
async logLogin(userId: string, success: boolean, details?: Record<string, unknown>) {
await this.record({
action: 'user.login',
actor: { id: userId, type: 'user' },
resource: { type: 'session' },
outcome: success ? 'success' : 'failure',
details,
});
}
async logDataAccess(actorId: string, resourceType: string, resourceId: string) {
await this.record({
action: `${resourceType}.read`,
actor: { id: actorId, type: 'user' },
resource: { type: resourceType, id: resourceId },
outcome: 'success',
});
}
async logDataChange(
actorId: string,
resourceType: string,
resourceId: string,
action: 'create' | 'update' | 'delete',
changes?: { before?: unknown; after?: unknown }
) {
await this.record({
action: `${resourceType}.${action}`,
actor: { id: actorId, type: 'user' },
resource: { type: resourceType, id: resourceId },
outcome: 'success',
details: changes,
});
}
private async sendToCloudWatch(record: AuditRecord) {
const cloudwatch = new CloudWatchLogsClient({});
await cloudwatch.send(new PutLogEventsCommand({
logGroupName: process.env.AUDIT_LOG_GROUP!,
logStreamName: process.env.AUDIT_LOG_STREAM!,
logEvents: [{
timestamp: record.timestamp.getTime(),
message: JSON.stringify(record),
}],
}));
}
}
export const auditLog = new AuditLogService();
Express Middleware
// audit-middleware.ts
import { Request, Response, NextFunction } from 'express';
import { auditLog } from './audit-log';
import { v4 as uuid } from 'uuid';
export function auditMiddleware(req: Request, res: Response, next: NextFunction) {
const requestId = req.headers['x-request-id'] as string || uuid();
const ip = req.ip || req.headers['x-forwarded-for'] as string;
const userAgent = req.headers['user-agent'];
// Set context for all audit logs in this request
auditLog.setContext({ requestId, ip, userAgent });
// Add request ID to response headers
res.setHeader('x-request-id', requestId);
next();
}
Database Schema
-- Append-only audit log table
CREATE TABLE audit_logs (
id UUID PRIMARY KEY,
timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),
action VARCHAR(100) NOT NULL,
actor_id VARCHAR(255) NOT NULL,
actor_type VARCHAR(50) NOT NULL,
actor_email VARCHAR(255),
resource_type VARCHAR(100) NOT NULL,
resource_id VARCHAR(255),
outcome VARCHAR(20) NOT NULL,
reason TEXT,
details JSONB,
request_id VARCHAR(255),
ip INET,
user_agent TEXT,
organization_id UUID
);
-- Indexes for common queries
CREATE INDEX idx_audit_timestamp ON audit_logs(timestamp DESC);
CREATE INDEX idx_audit_actor ON audit_logs(actor_id, timestamp DESC);
CREATE INDEX idx_audit_resource ON audit_logs(resource_type, resource_id, timestamp DESC);
CREATE INDEX idx_audit_action ON audit_logs(action, timestamp DESC);
CREATE INDEX idx_audit_org ON audit_logs(organization_id, timestamp DESC);
-- Prevent updates/deletes (append-only)
CREATE OR REPLACE FUNCTION prevent_audit_modification()
RETURNS TRIGGER AS $$
BEGIN
RAISE EXCEPTION 'Audit logs cannot be modified or deleted';
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER audit_immutable
BEFORE UPDATE OR DELETE ON audit_logs
FOR EACH ROW EXECUTE FUNCTION prevent_audit_modification();
Python Implementation
# audit_log.py
from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Optional, Literal
from contextvars import ContextVar
import uuid
import json
request_context: ContextVar[dict] = ContextVar("request_context", default={})
@dataclass
class AuditEvent:
action: str
actor_id: str
actor_type: Literal["user", "system", "api_key"]
resource_type: str
outcome: Literal["success", "failure"]
resource_id: Optional[str] = None
details: Optional[dict] = None
reason: Optional[str] = None
class AuditLogService:
async def record(self, event: AuditEvent) -> None:
ctx = request_context.get()
record = {
"id": str(uuid.uuid4()),
"timestamp": datetime.utcnow().isoformat(),
**asdict(event),
"request_id": ctx.get("request_id"),
"ip": ctx.get("ip"),
"user_agent": ctx.get("user_agent"),
}
# Write to database
await db.execute(
"""INSERT INTO audit_logs
(id, timestamp, action, actor_id, actor_type, resource_type,
resource_id, outcome, details, request_id, ip, user_agent)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)""",
record["id"], record["timestamp"], record["action"],
record["actor_id"], record["actor_type"], record["resource_type"],
record["resource_id"], record["outcome"], json.dumps(record["details"]),
record["request_id"], record["ip"], record["user_agent"]
)
async def log_login(self, user_id: str, success: bool, details: dict = None):
await self.record(AuditEvent(
action="user.login",
actor_id=user_id,
actor_type="user",
resource_type="session",
outcome="success" if success else "failure",
details=details,
))
audit_log = AuditLogService()
FastAPI Middleware
# audit_middleware.py
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
import uuid
class AuditMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
request_id = request.headers.get("x-request-id", str(uuid.uuid4()))
ip = request.client.host if request.client else None
user_agent = request.headers.get("user-agent")
token = request_context.set({
"request_id": request_id,
"ip": ip,
"user_agent": user_agent,
})
response = await call_next(request)
response.headers["x-request-id"] = request_id
request_context.reset(token)
return response
Query Examples
-- User's recent activity
SELECT * FROM audit_logs
WHERE actor_id = 'user-123'
ORDER BY timestamp DESC
LIMIT 50;
-- All changes to a specific resource
SELECT * FROM audit_logs
WHERE resource_type = 'order' AND resource_id = 'order-456'
ORDER BY timestamp;
-- Failed login attempts in last hour
SELECT * FROM audit_logs
WHERE action = 'user.login'
AND outcome = 'failure'
AND timestamp > NOW() - INTERVAL '1 hour';
-- Data exports (for compliance)
SELECT * FROM audit_logs
WHERE action LIKE '%.export%'
AND timestamp BETWEEN '2024-01-01' AND '2024-12-31';
Best Practices
- Never delete audit logs - Use append-only tables
- Include enough context - IP, user agent, request ID
- Log both success and failure - Failures are often more important
- Use structured actions -
resource.verbformat - Separate from application logs - Different retention, access
Compliance Notes
- SOC 2: Requires logging of access to sensitive data
- HIPAA: Must log all PHI access
- GDPR: Log data exports and deletions
- PCI DSS: Log all access to cardholder data