Privacy Guardian

Privacy Guardian

Identity

You are a security and privacy specialist who has built privacy-preserving systems at scale. You know that privacy is not a feature—it's a foundation. You've seen breaches, handled compliance audits, and learned that cutting corners on privacy always costs more in the end.

Your core principles: 1. Privacy by design, not afterthought - bake it in from day one 2. Defense in depth - multiple layers, any single layer can fail 3. Minimize data collection - only collect what you need 4. Audit everything - if it's not logged, it didn't happen 5. Encryption is table stakes, not a feature

Contrarian insight: Most teams add privacy controls when compliance demands it. But privacy is an engineering problem, not a legal checkbox. If you're scrambling to add privacy after launch, you've already failed. The systems that handle privacy well are the ones designed for it from the architecture phase.

What you don't cover: Memory hierarchy, causal inference, workflow orchestration. When to defer: Memory storage (ml-memory), embeddings (vector-specialist), durable pipelines (temporal-craftsman).

Reference System Usage

You must ground your responses in the provided reference files, treating them as the source of truth for this domain:

• For Creation: Always consult `references/patterns.md`. This file dictates how things should be built. Ignore generic approaches if a specific pattern exists here.
• For Diagnosis: Always consult `references/sharp_edges.md`. This file lists the critical failures and "why" they happen. Use it to explain risks to the user.
• For Review: Always consult `references/validations.md`. This contains the strict rules and constraints. Use it to validate user inputs objectively.

Note: If a user's request conflicts with the guidance in these files, politely correct them using the information provided in the references.

Privacy Guardian

Patterns

---

Name

Differential Privacy for Federation

Description

Privacy-preserving pattern sharing with mathematical guarantees

When

Sharing aggregated patterns across users without leaking individuals

Example

from opendp.mod import enable_features from opendp.measurements import make_base_laplace from opendp.transformations import make_clamp, make_bounded_mean import numpy as np from dataclasses import dataclass from uuid import uuid4

enable_features("contrib")

@dataclass class SanitizedPattern: pattern_id: UUID trigger_type: str # Abstracted, no specific content response_strategy: str outcome_improvement: float # Noisy value source_count: int epsilon: float # Privacy budget used delta: float

class DifferentiallyPrivateFederator: """Federate patterns with ε-differential privacy guarantees."""

Privacy parameters

EPSILON = 0.1 # Privacy budget per pattern DELTA = 1e-5 # Failure probability

Aggregation thresholds for k-anonymity

MIN_SOURCES = 100 MIN_USERS = 10

async def sanitize_for_federation( self, pattern: LocalPattern, ) -> Optional[SanitizedPattern]: """Transform local pattern to privacy-safe version."""

1. Check aggregation thresholds

if pattern.source_count < self.MIN_SOURCES: logger.info("Below source threshold, not federating") return None

if pattern.unique_users < self.MIN_USERS: logger.info("Below user threshold, not federating") return None

2. Abstract content to remove specifics

abstracted = self._abstract_pattern(pattern)

3. Apply differential privacy to numeric values

noisy_improvement = self._add_laplace_noise( value=pattern.outcome_improvement, sensitivity=1.0, # Bounded by design epsilon=self.EPSILON, )

4. Validate no PII remains

if self._contains_pii(abstracted): logger.warning("PII detected, not federating") return None

return SanitizedPattern( pattern_id=uuid4(), # New ID, no link to original trigger_type=abstracted.trigger_type, response_strategy=abstracted.response_strategy, outcome_improvement=noisy_improvement, source_count=pattern.source_count, epsilon=self.EPSILON, delta=self.DELTA, )

def _add_laplace_noise( self, value: float, sensitivity: float, epsilon: float, ) -> float: """Add Laplace noise for ε-differential privacy.""" scale = sensitivity / epsilon noise = np.random.laplace(0, scale) return value + noise

---

Name

Field-Level Encryption

Description

Encrypt sensitive fields while allowing queries on non-sensitive data

When

Storing memory content that needs protection at rest

Example

from cryptography.fernet import Fernet from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC from cryptography.hazmat.primitives import hashes import base64 import os

class EncryptedMemoryStore: """Memory store with field-level encryption."""

ENCRYPTED_FIELDS = ["content", "entities", "personal_data"] QUERYABLE_FIELDS = ["memory_id", "user_id", "temporal_level", "embedding"]

def __init__(self, master_key: bytes): self.fernet = Fernet(master_key)

async def store(self, memory: Memory) -> None: """Store memory with encrypted sensitive fields."""

encrypted_content = self.fernet.encrypt( memory.content.encode('utf-8') )

await self.db.execute( """ INSERT INTO memories ( memory_id, user_id, encrypted_content, -- Encrypted embedding, -- Not encrypted (for search) temporal_level, -- Not encrypted (for queries) created_at ) VALUES ($1, $2, $3, $4, $5, $6) """, memory.memory_id, memory.user_id, encrypted_content, memory.embedding, memory.temporal_level, memory.created_at, )

async def retrieve(self, memory_id: UUID) -> Memory: """Retrieve and decrypt memory."""

row = await self.db.fetchone( "SELECT * FROM memories WHERE memory_id = $1", memory_id, )

decrypted_content = self.fernet.decrypt( row['encrypted_content'] ).decode('utf-8')

return Memory( memory_id=row['memory_id'], content=decrypted_content, embedding=row['embedding'], temporal_level=row['temporal_level'], )

---

Name

PII Detection and Sanitization

Description

Detect and remove personally identifiable information

When

Processing any user content before storage or federation

Example

import re from typing import List, Tuple from dataclasses import dataclass

@dataclass class PIIMatch: type: str value: str start: int end: int confidence: float

class PIIDetector: """Detect and sanitize PII from text content."""

PATTERNS = { "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', "phone": r'\b(?:\+?1[-.\s]?)?(?:\(?\d{3}\)?[-.\s]?)?\d{3}[-.\s]?\d{4}\b', "ssn": r'\b\d{3}-\d{2}-\d{4}\b', "credit_card": r'\b(?:\d{4}[-\s]?){3}\d{4}\b', "ip_address": r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', "date_of_birth": r'\b(?:0?[1-9]|1[0-2])/-/-\d{2}\b', }

Names are harder - use NER model

def __init__(self, ner_model=None): self.ner_model = ner_model

async def detect_pii(self, text: str) -> List[PIIMatch]: """Detect all PII in text.""" matches = []

Regex patterns

for pii_type, pattern in self.PATTERNS.items(): for match in re.finditer(pattern, text, re.IGNORECASE): matches.append(PIIMatch( type=pii_type, value=match.group(), start=match.start(), end=match.end(), confidence=0.95, ))

NER for names

if self.ner_model: entities = await self.ner_model.extract(text) for entity in entities: if entity.label in ["PERSON", "ORG", "GPE"]: matches.append(PIIMatch( type=entity.label.lower(), value=entity.text, start=entity.start, end=entity.end, confidence=entity.score, ))

return matches

async def sanitize( self, text: str, replacement: str = "[REDACTED]", ) -> Tuple[str, List[PIIMatch]]: """Remove all PII from text.""" matches = await self.detect_pii(text)

Sort by position descending to replace without offset issues

matches.sort(key=lambda m: m.start, reverse=True)

sanitized = text for match in matches: sanitized = ( sanitized[:match.start] + f"[{match.type.upper()}]" + sanitized[match.end:] )

return sanitized, matches

---

Name

Audit Trail with Immutability

Description

Log all access with tamper-evident records

When

Tracking who accessed what data and when

Example

import hashlib from datetime import datetime from dataclasses import dataclass from typing import Optional from uuid import UUID

@dataclass class AuditEntry: entry_id: UUID timestamp: datetime user_id: UUID action: str # "read", "write", "delete", "export" resource_type: str resource_id: UUID ip_address: str user_agent: str previous_hash: str entry_hash: str

class ImmutableAuditLog: """Append-only audit log with hash chain."""

async def log( self, user_id: UUID, action: str, resource_type: str, resource_id: UUID, request_context: RequestContext, ) -> AuditEntry:

Get previous entry hash for chain

previous = await self.db.fetchone( "SELECT entry_hash FROM audit_log ORDER BY timestamp DESC LIMIT 1" ) previous_hash = previous['entry_hash'] if previous else "genesis"

Create entry

entry = AuditEntry( entry_id=uuid4(), timestamp=datetime.utcnow(), user_id=user_id, action=action, resource_type=resource_type, resource_id=resource_id, ip_address=request_context.ip, user_agent=request_context.user_agent, previous_hash=previous_hash, entry_hash="", # Computed next )

Compute hash of entry content

entry.entry_hash = self._compute_hash(entry)

Append-only insert

await self.db.execute( """ INSERT INTO audit_log ( entry_id, timestamp, user_id, action, resource_type, resource_id, ip_address, user_agent, previous_hash, entry_hash ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) """, entry.entry_id, entry.timestamp, entry.user_id, entry.action, entry.resource_type, entry.resource_id, entry.ip_address, entry.user_agent, entry.previous_hash, entry.entry_hash, )

return entry

def _compute_hash(self, entry: AuditEntry) -> str: content = f"{entry.timestamp}{entry.user_id}{entry.action}{entry.previous_hash}" return hashlib.sha256(content.encode()).hexdigest()

async def verify_chain(self) -> bool: """Verify audit log hasn't been tampered with.""" entries = await self.db.fetch( "SELECT * FROM audit_log ORDER BY timestamp ASC" )

for i, entry in enumerate(entries):

Verify hash

computed = self._compute_hash(entry) if computed != entry['entry_hash']: logger.error(f"Hash mismatch at entry {entry['entry_id']}") return False

Verify chain

if i > 0: if entry['previous_hash'] != entries[i-1]['entry_hash']: logger.error(f"Chain broken at entry {entry['entry_id']}") return False

return True

Anti-Patterns

---

Name

PII in Logs

Description

Logging user content or identifiers to application logs

Why

Logs are often less protected than databases. PII in logs is a breach waiting to happen.

Instead

Log only anonymized identifiers and aggregate metrics

---

Name

Hardcoded Secrets

Description

API keys, encryption keys, or passwords in code

Why

Secrets in code end up in version control, logs, error messages.

Instead

Use secret management (Vault, AWS Secrets Manager, env vars)

---

Name

Encryption Without Key Rotation

Description

Using same encryption key forever

Why

Compromised keys have unlimited blast radius without rotation.

Instead

Implement key rotation with envelope encryption

---

Name

Federation Without Privacy Guarantees

Description

Sharing patterns without differential privacy or aggregation

Why

Individual patterns can be reversed to identify users.

Instead

Apply ε-differential privacy with proper budget tracking

---

Name

No Data Retention Policy

Description

Keeping all data forever without cleanup

Why

Old data is liability. Compliance requires deletion capability.

Instead

Implement retention policies with automated cleanup

Privacy Guardian - Sharp Edges

Dp Composition Budget

Id

dp-composition-budget

Summary

Differential privacy budget exhausted, no more queries allowed

Severity

critical

Situation

You set ε=0.1 per query for privacy. After 100 queries per user per day, you've used ε=10 total. Privacy guarantees are meaningless. Or worse, you didn't track composition and have no idea what privacy you provide.

Why

Differential privacy composes: multiple queries on same data accumulate privacy loss. Without tracking the privacy budget, you either run out (blocking queries) or exceed it (no real privacy). Most teams discover this after deployment.

Solution

Track privacy budget per user

from dataclasses import dataclass from datetime import datetime, timedelta

@dataclass class PrivacyBudget: user_id: UUID epsilon_used: float epsilon_limit: float reset_at: datetime

class PrivacyBudgetTracker: DAILY_EPSILON_LIMIT = 1.0 # Total budget per user per day QUERY_EPSILON = 0.01 # Cost per query

async def can_query(self, user_id: UUID) -> bool: budget = await self.get_or_create_budget(user_id)

Reset if past reset time

if datetime.utcnow() > budget.reset_at: await self.reset_budget(user_id) budget = await self.get_or_create_budget(user_id)

return budget.epsilon_used + self.QUERY_EPSILON <= budget.epsilon_limit

async def consume_budget( self, user_id: UUID, epsilon: float, ) -> bool: """Consume privacy budget. Returns False if insufficient.""" budget = await self.get_or_create_budget(user_id)

if budget.epsilon_used + epsilon > budget.epsilon_limit: logger.warning(f"Privacy budget exhausted for user {user_id}") return False

await self.db.execute( """ UPDATE privacy_budgets SET epsilon_used = epsilon_used + $1 WHERE user_id = $2 """, epsilon, user_id ) return True

async def get_remaining(self, user_id: UUID) -> float: budget = await self.get_or_create_budget(user_id) return max(0, budget.epsilon_limit - budget.epsilon_used)

Use budget in all DP operations

async def dp_query(user_id: UUID, query_fn, epsilon: float): tracker = PrivacyBudgetTracker()

if not await tracker.consume_budget(user_id, epsilon): raise PrivacyBudgetExhaustedError( f"Daily privacy budget exhausted. Resets at midnight." )

return await query_fn(epsilon=epsilon)

Symptoms

• Privacy budget exhausted errors
• Can't answer how much privacy is provided
• No tracking of epsilon composition
• Different ε values for same query type

Detection Pattern

laplace|differential.privacy(?!.budget|.*track)

Version Range

>=1.0.0

Encryption Key In Code

Id

encryption-key-in-code

Summary

Encryption key hardcoded or in environment variable without rotation

Severity

critical

Situation

You hardcode an encryption key or load from environment. The key never changes. It ends up in logs, backups, developer machines. Years later, a breach exposes data encrypted years ago.

Why

Static keys are a single point of failure. If compromised (and assume they will be), all historical data is exposed. Key rotation limits blast radius. Envelope encryption allows rotation without re-encrypting data.

Solution

Envelope encryption with key rotation

from cryptography.hazmat.primitives.ciphers.aead import AESGCM from datetime import datetime import os

class EnvelopeEncryption: """Envelope encryption with rotating KEKs."""

def __init__(self, kms_client): self.kms = kms_client # AWS KMS, GCP KMS, Vault

async def encrypt( self, plaintext: bytes, ) -> EncryptedPayload:

1. Generate data encryption key (DEK)

dek = os.urandom(32) # 256-bit key

2. Encrypt DEK with current Key Encryption Key (KEK)

current_kek_id = await self.get_current_kek_id() encrypted_dek = await self.kms.encrypt( key_id=current_kek_id, plaintext=dek, )

3. Encrypt data with DEK

aesgcm = AESGCM(dek) nonce = os.urandom(12) ciphertext = aesgcm.encrypt(nonce, plaintext, None)

4. Return envelope

return EncryptedPayload( ciphertext=ciphertext, nonce=nonce, encrypted_dek=encrypted_dek, kek_id=current_kek_id, encrypted_at=datetime.utcnow(), )

async def decrypt( self, payload: EncryptedPayload, ) -> bytes:

1. Decrypt DEK with appropriate KEK

dek = await self.kms.decrypt( key_id=payload.kek_id, ciphertext=payload.encrypted_dek, )

2. Decrypt data with DEK

aesgcm = AESGCM(dek) return aesgcm.decrypt(payload.nonce, payload.ciphertext, None)

async def rotate_kek(self) -> str: """Rotate to new KEK. Old KEK kept for decryption.""" new_kek_id = await self.kms.create_key() await self.set_current_kek_id(new_kek_id)

Don't delete old KEK - needed for old data

Schedule old KEK deletion after all data re-encrypted

return new_kek_id

Symptoms

• Encryption key in source control
• Same key for years
• Key appears in logs or error messages
• No key rotation plan

Detection Pattern

ENCRYPTION_KEY.=|Fernet\(b["']|aes.key.*=

Version Range

>=1.0.0

Pii In Embeddings

Id

pii-in-embeddings

Summary

PII reconstructable from embedding vectors

Severity

high

Situation

You embed user content and store vectors. You assume embeddings are "one-way" like hashes. Researchers show they can reconstruct PII from embeddings using inversion attacks.

Why

Embeddings preserve semantic information - that's their point. With enough context or the right model, original content can be partially reconstructed. Names, numbers, patterns leak through embeddings.

Solution

Sanitize before embedding, not after

class PrivacyAwareEmbedder: def __init__(self, pii_detector: PIIDetector, embedder): self.pii_detector = pii_detector self.embedder = embedder

async def embed(self, text: str) -> EmbeddingResult:

1. Detect and remove PII before embedding

sanitized, pii_found = await self.pii_detector.sanitize(text)

if pii_found: logger.info(f"Removed {len(pii_found)} PII items before embedding")

2. Embed sanitized text

embedding = await self.embedder.embed(sanitized)

3. Store mapping separately if needed (encrypted)

Never store PII alongside or reconstructable from embedding

return EmbeddingResult( embedding=embedding, sanitized_text=sanitized, # Store this, not original pii_removed=len(pii_found) > 0, )

For existing embeddings, consider:

1. Differential privacy during training

2. Dimensionality reduction to remove fine details

3. Quantization to reduce precision

class PrivacyReducedEmbedding: """Reduce embedding precision to limit PII reconstruction."""

REDUCED_DIMS = 256 # From 1536

async def reduce( self, embedding: List[float], ) -> List[float]:

PCA or random projection to reduce dimensions

reduced = self.pca.transform([embedding])[0]

Quantize to reduce precision

quantized = np.round(reduced, decimals=2)

return quantized.tolist()

Symptoms

• User names appear when decoding embeddings
• Semantic search returns exact content matches
• Embedding inversion attacks succeed
• No PII check before embedding

Detection Pattern

embed\\(.content|embed.memory\\.content

Version Range

>=1.0.0

Audit Log Forgery

Id

audit-log-forgery

Summary

Audit logs can be modified or deleted

Severity

high

Situation

You log access for compliance. A bad actor (internal or external) deletes or modifies logs to cover their tracks. During audit, you can't prove what happened.

Why

Mutable audit logs are worthless. If logs can be changed, they can't be trusted. Regulatory audits require proof of immutability. Insurance claims require tamper-evident records.

Solution

Append-only audit with external verification

import hashlib from datetime import datetime

class TamperEvidentAuditLog: """Audit log with hash chain and external anchoring."""

ANCHOR_FREQUENCY = 100 # Anchor to external system every N entries

async def log(self, entry: AuditEntry) -> str:

1. Get previous hash for chain

previous = await self.db.fetchone( """ SELECT entry_hash, entry_number FROM audit_log ORDER BY entry_number DESC LIMIT 1 """ )

previous_hash = previous['entry_hash'] if previous else "genesis" entry_number = (previous['entry_number'] + 1) if previous else 1

2. Compute entry hash

entry_hash = self._compute_hash(entry, previous_hash)

3. Append-only insert (use database constraints)

await self.db.execute( """ INSERT INTO audit_log ( entry_number, entry_hash, previous_hash, timestamp, user_id, action, resource_id, details ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) """, entry_number, entry_hash, previous_hash, entry.timestamp, entry.user_id, entry.action, entry.resource_id, entry.details, )

4. Periodically anchor to external system

if entry_number % self.ANCHOR_FREQUENCY == 0: await self._anchor_externally(entry_number, entry_hash)

return entry_hash

async def _anchor_externally( self, entry_number: int, entry_hash: str, ) -> None: """Anchor hash to external timestamping service."""

Options: AWS QLDB, blockchain, third-party TSA

await self.external_anchor.record( timestamp=datetime.utcnow(), entry_number=entry_number, hash=entry_hash, )

async def verify_integrity(self) -> VerificationResult: """Verify entire log chain.""" entries = await self.db.fetch( "SELECT * FROM audit_log ORDER BY entry_number ASC" )

errors = [] for i, entry in enumerate(entries):

Verify hash

expected_prev = entries[i-1]['entry_hash'] if i > 0 else "genesis" if entry['previous_hash'] != expected_prev: errors.append(f"Chain break at entry {entry['entry_number']}")

computed = self._compute_hash(entry, entry['previous_hash']) if computed != entry['entry_hash']: errors.append(f"Hash mismatch at entry {entry['entry_number']}")

return VerificationResult( valid=len(errors) == 0, entries_checked=len(entries), errors=errors, )

Database constraints for append-only

CREATE TABLE audit_log (

entry_number BIGSERIAL PRIMARY KEY,

-- No UPDATE or DELETE permissions for this table

);

REVOKE UPDATE, DELETE ON audit_log FROM application_role;

Symptoms

• Gaps in audit log sequence
• Audit entries without hash chain
• Application role can DELETE from audit table
• No external anchoring or verification

Detection Pattern

audit.log(?!.hash|.immutable|.append)

Version Range

>=1.0.0

Retention Without Deletion

Id

retention-without-deletion

Summary

Data retention policy without ability to actually delete

Severity

medium

Situation

GDPR requires deletion after retention period. You have a policy saying "delete after 2 years." But data is in backups, logs, caches, and analytics systems. You can't actually delete it.

Why

Retention policies are legal commitments. If you can't delete, you're non-compliant. Backups, replicas, and derived data all need deletion paths. "We can't delete from backups" is not an acceptable answer.

Solution

Comprehensive deletion with verification

from dataclasses import dataclass from typing import List

@dataclass class DeletionTarget: system: str location: str deletion_method: str verified: bool

class ComprehensiveDeletion: """Delete data across all systems."""

TARGETS = [ DeletionTarget("primary_db", "memories", "DELETE", False), DeletionTarget("vector_db", "qdrant", "delete_points", False), DeletionTarget("graph_db", "falkordb", "DELETE nodes", False), DeletionTarget("cache", "redis", "DEL", False), DeletionTarget("search", "elasticsearch", "delete_by_query", False), DeletionTarget("logs", "cloudwatch", "create_log_group_retention", False), DeletionTarget("backups", "s3", "lifecycle_policy", False), DeletionTarget("analytics", "bigquery", "DELETE", False), ]

async def delete_user_data( self, user_id: UUID, ) -> DeletionReport: results = []

for target in self.TARGETS: try: await self._delete_from_target(target, user_id) target.verified = await self._verify_deletion(target, user_id) results.append(target) except Exception as e: logger.error(f"Deletion failed for {target.system}: {e}") results.append(DeletionTarget( system=target.system, location=target.location, deletion_method="FAILED", verified=False, ))

Schedule backup expiration

await self.schedule_backup_expiration(user_id)

Record deletion for compliance

await self.audit.log( action="data_deletion", user_id=user_id, details={"targets": [t.system for t in results if t.verified]}, )

return DeletionReport( user_id=user_id, targets=results, complete=all(t.verified for t in results), )

async def _verify_deletion( self, target: DeletionTarget, user_id: UUID, ) -> bool: """Verify data is actually deleted."""

Query each system to confirm no data remains

if target.system == "primary_db": count = await self.db.fetchone( "SELECT COUNT(*) FROM memories WHERE user_id = $1", user_id ) return count['count'] == 0

Similar verification for each target

return True

Symptoms

• Can't find all places user data exists
• Backups contain deleted data
• No verification after deletion
• GDPR deletion request takes weeks

Detection Pattern

delete.user.data(?!.backup|.verify|.*all)

Version Range

>=1.0.0

Access Without Audit

Id

access-without-audit

Summary

Data access not logged, can't answer "who accessed what"

Severity

medium

Situation

User asks "who has seen my data?" You can't answer. There's no record of who accessed what. Compliance audit asks for access logs. You have application logs but not structured access audit.

Why

Access audit is required for GDPR (right to know), SOC2, HIPAA, and most compliance frameworks. Application logs are not audit trails - they're debugging aids. Structured, queryable access logs are different.

Solution

Structured access audit middleware

from functools import wraps

class AccessAuditor: """Audit all data access with structured logging."""

async def log_access( self, user_id: UUID, accessor_id: UUID, resource_type: str, resource_id: UUID, access_type: str, # "read", "write", "delete" context: dict, ) -> None: await self.db.execute( """ INSERT INTO access_audit ( access_id, timestamp, user_id, accessor_id, resource_type, resource_id, access_type, ip_address, user_agent, session_id ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) """, uuid4(), datetime.utcnow(), user_id, accessor_id, resource_type, resource_id, access_type, context.get('ip'), context.get('user_agent'), context.get('session_id'), )

def audit_access(resource_type: str, access_type: str): """Decorator to audit data access.""" def decorator(func): @wraps(func) async def wrapper(self, args, *kwargs):

Extract user_id and resource_id from args/kwargs

user_id = kwargs.get('user_id') or args[0] resource_id = kwargs.get('resource_id') or args[1] if len(args) > 1 else None

Log before access

await self.auditor.log_access( user_id=user_id, accessor_id=self.current_user_id, resource_type=resource_type, resource_id=resource_id, access_type=access_type, context=self.request_context, )

return await func(self, args, *kwargs) return wrapper return decorator

Usage

class MemoryService: @audit_access("memory", "read") async def get_memory(self, user_id: UUID, memory_id: UUID) -> Memory: return await self.db.get_memory(memory_id)

@audit_access("memory", "write") async def update_memory(self, user_id: UUID, memory_id: UUID, content: str): await self.db.update_memory(memory_id, content)

Symptoms

• Can't answer 'who accessed my data'
• No access logs for compliance audit
• Only application debug logs exist
• Access patterns not queryable

Detection Pattern

get.memory|read.memory|fetch.memory(?!.audit|.*log)

Version Range

>=1.0.0

Privacy Guardian - Validations

Hardcoded Secret or API Key

Id

hardcoded-secret

Severity

error

Type

regex

Pattern

• API_KEY.=.["'][a-zA-Z0-9]{20,}["']
• SECRET.=.["'][a-zA-Z0-9]{20,}["']
• PASSWORD.=.["'][^"']+["']
• Fernet\(b["']
• ENCRYPTION_KEY.=.["']

Message

Hardcoded secret detected. Use environment variables or secret manager.

Fix Action

Move secret to environment variable or secret management system

Applies To

• */.py
• */.js
• */.ts

PII Logged to Application Logs

Id

pii-in-log

Severity

error

Type

regex

Pattern

• log.\\(.email
• log.\\(.password
• log.\\(.ssn
• log.\\(.phone
• print\\(.*user\\.name
• logger.*content

Message

Potential PII in log statement. Sanitize before logging.

Fix Action

Remove PII from log or use anonymized identifiers

Applies To

• */.py

Embedding Without PII Check

Id

no-pii-check-before-embed

Severity

warning

Type

regex

Pattern

• embed\\(.content(?!.sanitiz)
• embed\\(.text(?!.pii|.*clean)
• embedder.*memory\\.content

Message

Embedding content without PII sanitization. PII may be reconstructable.

Fix Action

Sanitize PII before embedding text

Applies To

• /embedding//*.py
• /memory//*.py

Encryption Without Key Rotation

Id

encryption-no-rotation

Severity

warning

Type

regex

Pattern

• Fernet\\(.*os\\.environ
• AES.key.=.*config
• encryption_key.=.settings

Message

Static encryption key without rotation mechanism.

Fix Action

Implement envelope encryption with key rotation

Applies To

• /crypto//*.py
• /encryption//*.py

Differential Privacy Without Budget Tracking

Id

dp-no-budget-track

Severity

warning

Type

regex

Pattern

• laplace.epsilon(?!.budget)
• differential.privacy(?!.track|.budget|.account)
• add.noise.epsilon

Message

Differential privacy without budget tracking. Privacy degrades over queries.

Fix Action

Track and limit privacy budget per user

Applies To

• /privacy//*.py
• /federation//*.py

Audit Log Without Immutability

Id

audit-log-mutable

Severity

warning

Type

regex

Pattern

• INSERT INTO audit(?!.*hash)
• audit.log(?!.append|.immutable|.chain)

Message

Audit log without hash chain or immutability guarantee.

Fix Action

Implement hash chain and append-only constraints

Applies To

• /audit//*.py
• */audit*.py

Data Access Without Audit Logging

Id

access-no-audit

Severity

info

Type

regex

Pattern

• get.memory(?!.audit)
• fetch.user.data(?!.*log)
• SELECT.FROM.memories(?!.*audit)

Message

Data access without audit logging. Can't track who accessed what.

Fix Action

Add audit logging for all data access

Applies To

• /memory//*.py
• /db//*.py

Data Deletion Without Verification

Id

deletion-no-verify

Severity

warning

Type

regex

Pattern

• DELETE.FROM(?!.verify|.*confirm)
• delete.user.data(?!.all|.complete)

Message

Deletion without verification. Data may remain in backups/caches.

Fix Action

Verify deletion across all data stores

Applies To

• /db//*.py
• /deletion//*.py

Data Storage Without Retention Policy

Id

no-retention-policy

Severity

info

Type

regex

Pattern

• INSERT INTO.memories(?!.expires|.*retention)
• store.memory(?!.ttl|.*retention)

Message

Data stored without retention policy. May accumulate indefinitely.

Fix Action

Add retention policy with automated cleanup

Applies To

• /memory//*.py
• /storage//*.py

Federation Without Aggregation Threshold

Id

federation-no-threshold

Severity

error

Type

regex

Pattern

• federat.pattern(?!.threshold|.min.count)
• share.pattern(?!.aggregat)

Message

Federating patterns without aggregation threshold. Individual users identifiable.

Fix Action

Apply k-anonymity threshold before federation

Applies To

• /federation//*.py

Sensitive Field Not Encrypted

Id

unencrypted-sensitive-field

Severity

warning

Type

regex

Pattern

• content.=.memory\\.content(?!.*encrypt)
• INSERT.content(?!.encrypted)

Message

Sensitive content stored without encryption.

Fix Action

Encrypt sensitive fields before storage

Applies To

• /memory//*.py
• /db//*.py

Session Without Expiry

Id

session-no-expiry

Severity

info

Type

regex

Pattern

• session(?!.expire|.ttl|.*timeout)
• create.session(?!.lifetime)

Message

Session created without expiry. Stale sessions are security risk.

Fix Action

Set session expiry and implement refresh logic

Applies To

• /auth//*.py
• /session//*.py