Audit Module

Audit logging infrastructure including the main logger, encryption utilities, and storage backends.

---

AuditLogger

Main audit logging interface for AI compliance.

The AuditLogger provides a high-level interface for recording AI interactions, including inputs, outputs, safety evaluations, and performance metrics. It supports optional encryption of content and configurable retention policies.

Parameters:

Name	Type	Description	Default
storage	Union[StorageBackend, str]	Either a StorageBackend instance or a file path string. If a string is provided, FileStorage is automatically created.	required
encryption	EncryptionManager | None	Optional EncryptionManager for encrypting stored content.	None
store_content	bool	If True, stores actual content. If False, only stores content hashes for privacy (default: False).	False
retention_days	int	Number of days to retain entries before cleanup (default: 365).	365

Attributes:

Name	Type	Description
storage	StorageBackend	The underlying storage backend.
encryption		The encryption manager (if provided).
store_content		Whether to store actual content.
retention_days		Retention period in days.

Example

Basic usage with file storage:

logger = AuditLogger("/var/log/audit") entry_id = await logger.log( ... input="What is 2+2?", ... output="4", ... provider="openai", ... model="gpt-4", ... )

With encryption and content storage:

from rotalabs_comply.audit import EncryptionManager encryption = EncryptionManager() logger = AuditLogger( ... "/var/log/audit", ... encryption=encryption, ... store_content=True, ... ) entry_id = await logger.log( ... input="Sensitive question", ... output="Sensitive answer", ... safety_passed=True, ... )

With custom storage backend:

from rotalabs_comply.audit import MemoryStorage storage = MemoryStorage(max_entries=10000) logger = AuditLogger(storage)

__init__(storage, encryption=None, store_content=False, retention_days=365)

Initialize the audit logger.

Parameters:

Name	Type	Description	Default
storage	Union[StorageBackend, str]	StorageBackend instance or file path string.	required
encryption	EncryptionManager | None	Optional encryption manager for content encryption.	None
store_content	bool	Whether to store actual content (default: False).	False
retention_days	int	Days to retain entries (default: 365).	365

cleanup_expired() async

Delete entries older than the retention period.

Removes all entries with timestamps older than retention_days from the current time.

Returns:

Name	Type	Description
int	int	Number of entries deleted.

Example

Delete entries older than retention_days

deleted_count = await logger.cleanup_expired() print(f"Cleaned up {deleted_count} expired entries")

Note

This operation may be slow for large datasets. Consider running during off-peak hours or using storage-native lifecycle policies (e.g., S3 lifecycle rules) for better performance.

decrypt_content(encrypted_content)

Decrypt encrypted content from an audit entry.

Convenience method for decrypting content stored in audit entries when encryption is enabled.

Parameters:

Name	Type	Description	Default
encrypted_content	str	The encrypted content string from an AuditEntry.	required

Returns:

Name	Type	Description
str	str	The decrypted original content.

Raises:

Type	Description
ValueError	If no encryption manager is configured.
InvalidToken	If decryption fails.

Example

entry = await logger.get_entry("abc-123") if entry and entry.input_content: ... original_input = logger.decrypt_content(entry.input_content)

get_entries(start, end) async

Retrieve all audit entries within a time range.

Parameters:

Name	Type	Description	Default
start	datetime	Start of the time range (inclusive).	required
end	datetime	End of the time range (inclusive).	required

Returns:

Type	Description
List[AuditEntry]	List[AuditEntry]: All entries within the specified time range.

Example

from datetime import datetime, timedelta end = datetime.utcnow() start = end - timedelta(days=7) entries = await logger.get_entries(start, end) print(f"Found {len(entries)} entries in the last week")

get_entry(entry_id) async

Retrieve an audit entry by ID.

If encryption is enabled and store_content is True, the content will still be encrypted in the returned entry. Use the encryption manager to decrypt if needed.

Parameters:

Name	Type	Description	Default
entry_id	str	The unique identifier of the entry.	required

Returns:

Type	Description
AuditEntry | None	AuditEntry | None: The entry if found, None otherwise.

Example

entry = await logger.get_entry("abc-123-def") if entry: ... print(f"Safety passed: {entry.safety_passed}")

log(input, output, provider=None, model=None, conversation_id=None, safety_passed=True, detectors_triggered=None, block_reason=None, alerts=None, latency_ms=0.0, input_tokens=None, output_tokens=None, metadata=None) async

Log an AI interaction.

Creates an audit entry for the interaction and stores it in the configured storage backend.

Parameters:

Name	Type	Description	Default
input	str	The user input or prompt.	required
output	str	The AI-generated output or response.	required
provider	str | None	The AI provider (e.g., "openai", "anthropic").	None
model	str | None	The model identifier (e.g., "gpt-4", "claude-3-opus").	None
conversation_id	str | None	Optional ID to link related interactions.	None
safety_passed	bool	Whether the interaction passed safety checks (default: True).	True
detectors_triggered	List[str] | None	List of safety detector names that triggered.	None
block_reason	str | None	Reason for blocking, if the request was blocked.	None
alerts	List[str] | None	List of alert messages generated.	None
latency_ms	float	Time taken to process the request in milliseconds (default: 0.0).	0.0
input_tokens	int | None	Number of tokens in the input.	None
output_tokens	int | None	Number of tokens in the output.	None
metadata	Dict[str, Any] | None	Additional custom metadata dictionary.	None

Returns:

Name	Type	Description
str	str	The unique entry ID for this audit log entry.

Example

entry_id = await logger.log( ... input="Tell me a joke", ... output="Why did the chicken...", ... provider="anthropic", ... model="claude-3-opus", ... safety_passed=True, ... latency_ms=250.5, ... input_tokens=5, ... output_tokens=20, ... metadata={"session_id": "abc123"}, ... )

Main audit logging interface for AI compliance.

Constructor

AuditLogger(
    storage: Union[StorageBackend, str],
    encryption: Optional[EncryptionManager] = None,
    store_content: bool = False,
    retention_days: int = 365,
)

Parameters:

Parameter	Type	Default	Description
storage	Union[StorageBackend, str]	Required	Storage backend or file path
encryption	Optional[EncryptionManager]	None	Encryption manager for content
store_content	bool	False	Store actual content vs hashes
retention_days	int	365	Days to retain entries

Methods

log

async def log(
    input: str,
    output: str,
    provider: Optional[str] = None,
    model: Optional[str] = None,
    conversation_id: Optional[str] = None,
    safety_passed: bool = True,
    detectors_triggered: Optional[List[str]] = None,
    block_reason: Optional[str] = None,
    alerts: Optional[List[str]] = None,
    latency_ms: float = 0.0,
    input_tokens: Optional[int] = None,
    output_tokens: Optional[int] = None,
    metadata: Optional[Dict[str, Any]] = None,
) -> str

Log an AI interaction and return the entry ID.

Example:

entry_id = await logger.log(
    input="Tell me a joke",
    output="Why did the chicken...",
    provider="anthropic",
    model="claude-3-opus",
    safety_passed=True,
    latency_ms=250.5,
    metadata={"session_id": "abc123"},
)

get_entry

async def get_entry(entry_id: str) -> Optional[AuditEntry]

Retrieve an audit entry by ID.

Example:

entry = await logger.get_entry("abc-123-def")
if entry:
    print(f"Safety passed: {entry.safety_passed}")

get_entries

async def get_entries(start: datetime, end: datetime) -> List[AuditEntry]

Retrieve all entries within a time range.

Example:

from datetime import datetime, timedelta

end = datetime.utcnow()
start = end - timedelta(days=7)
entries = await logger.get_entries(start, end)

cleanup_expired

async def cleanup_expired() -> int

Delete entries older than the retention period. Returns count of deleted entries.

Example:

deleted = await logger.cleanup_expired()
print(f"Cleaned up {deleted} expired entries")

decrypt_content

def decrypt_content(encrypted_content: str) -> str

Decrypt encrypted content from an audit entry.

Raises:

• ValueError: If no encryption manager is configured
• cryptography.fernet.InvalidToken: If decryption fails

Example:

entry = await logger.get_entry("abc-123")
if entry and entry.input_content:
    original = logger.decrypt_content(entry.input_content)

---

Encryption

EncryptionManager

High-level encryption manager for string data.

Provides a convenient interface for encrypting and decrypting string data, with automatic key generation if not provided.

Parameters:

Name	Type	Description	Default
key	bytes | None	Optional Fernet encryption key. If not provided, a new key is generated.	None

Attributes:

Name	Type	Description
_key		The encryption key (kept private).
_fernet		The Fernet cipher instance.

Example

manager = EncryptionManager() encrypted = manager.encrypt("sensitive data") manager.decrypt(encrypted) 'sensitive data'

Use existing key

key = generate_key() manager = EncryptionManager(key) manager.get_key() == key True

__init__(key=None)

Initialize the encryption manager.

Parameters:

Name	Type	Description	Default
key	bytes | None	Optional Fernet encryption key. If None, generates a new key.	None

decrypt(data)

Decrypt a base64-encoded encrypted string.

Parameters:

Name	Type	Description	Default
data	str	The base64-encoded encrypted string (from encrypt()).	required

Returns:

Name	Type	Description
str	str	The original decrypted string.

Raises:

Type	Description
InvalidToken	If decryption fails.

Example

manager = EncryptionManager() encrypted = manager.encrypt("secret") manager.decrypt(encrypted) 'secret'

encrypt(data)

Encrypt a string and return base64-encoded result.

Parameters:

Name	Type	Description	Default
data	str	The string to encrypt.	required

Returns:

Name	Type	Description
str	str	Base64-encoded encrypted string, safe for storage in JSON.

Example

manager = EncryptionManager() encrypted = manager.encrypt("secret") isinstance(encrypted, str) True

get_key()

Get the encryption key.

Returns:

Name	Type	Description
bytes	bytes	The Fernet encryption key. Store this securely!

Warning

The encryption key must be stored securely. If lost, encrypted data cannot be recovered.

Example

manager = EncryptionManager() key = manager.get_key() len(key) 44

High-level encryption manager for string data.

Constructor

EncryptionManager(key: Optional[bytes] = None)

Parameters:

Parameter	Type	Default	Description
key	Optional[bytes]	Auto-gen	Fernet encryption key

Methods

encrypt

def encrypt(data: str) -> str

Encrypt a string and return base64-encoded result.

decrypt

def decrypt(data: str) -> str

Decrypt a base64-encoded encrypted string.

get_key

def get_key() -> bytes

Get the encryption key. Store this securely!

Example:

from rotalabs_comply import EncryptionManager

manager = EncryptionManager()
encrypted = manager.encrypt("sensitive data")
decrypted = manager.decrypt(encrypted)

# Save key securely
key = manager.get_key()

Helper Functions

generate_key

def generate_key() -> bytes

Generate a new Fernet encryption key.

Example:

from rotalabs_comply import generate_key

key = generate_key()
print(len(key))  # 44

encrypt

def encrypt(data: bytes, key: bytes) -> bytes

Encrypt raw bytes using Fernet symmetric encryption.

decrypt

def decrypt(data: bytes, key: bytes) -> bytes

Decrypt data that was encrypted with Fernet.

hash_content

def hash_content(content: str) -> str

Compute SHA-256 hash of string content.

Example:

from rotalabs_comply import hash_content

content_hash = hash_content("hello world")
# Returns: 'b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9'

---

Storage Backends

StorageBackend Protocol

Protocol defining the interface for audit log storage backends.

All storage backends must implement these async methods to support writing, reading, listing, deleting, and counting audit entries.

count() abstractmethod async

Count total number of entries in storage.

Returns:

Name	Type	Description
int	int	Total number of stored entries.

delete(entry_id) abstractmethod async

Delete an entry by ID.

Parameters:

Name	Type	Description	Default
entry_id	str	The unique identifier of the entry to delete.	required

Returns:

Name	Type	Description
bool	bool	True if the entry was deleted, False if not found.

list_entries(start, end) abstractmethod async

List all entries within a time range.

Parameters:

Name	Type	Description	Default
start	datetime	Start of the time range (inclusive).	required
end	datetime	End of the time range (inclusive).	required

Returns:

Type	Description
List[AuditEntry]	List[AuditEntry]: Entries within the specified time range.

read(entry_id) abstractmethod async

Read an audit entry by ID.

Parameters:

Name	Type	Description	Default
entry_id	str	The unique identifier of the entry.	required

Returns:

Type	Description
AuditEntry | None	AuditEntry | None: The entry if found, None otherwise.

write(entry) abstractmethod async

Write an audit entry to storage.

Parameters:

Name	Type	Description	Default
entry	AuditEntry	The audit entry to store.	required

Returns:

Name	Type	Description
str	str	The entry ID (same as entry.id).

Protocol defining the interface for audit log storage backends.

Required Methods:

Method	Signature	Description
write	async (entry: AuditEntry) -> str	Write entry, return ID
read	async (entry_id: str) -> Optional[AuditEntry]	Read entry by ID
list_entries	async (start: datetime, end: datetime) -> List[AuditEntry]	List entries in range
delete	async (entry_id: str) -> bool	Delete entry, return success
count	async () -> int	Count total entries

---

FileStorage

File-based storage backend using JSONL format.

Stores audit entries as JSON Lines files with automatic rotation when files exceed the configured size limit.

Parameters:

Name	Type	Description	Default
path	str	Directory path for storing audit files.	required
rotation_size_mb	int	Maximum file size in MB before rotation (default: 100).	100

Attributes:

Name	Type	Description
path		The storage directory path.
rotation_size_bytes		Maximum file size in bytes.

Example

storage = FileStorage("/var/log/audit") entry_id = await storage.write(entry) retrieved = await storage.read(entry_id)

__init__(path, rotation_size_mb=100)

Initialize file storage.

Parameters:

Name	Type	Description	Default
path	str	Directory path for storing audit files.	required
rotation_size_mb	int	Maximum file size in MB before rotation.	100

count() async

Count total number of entries in storage.

Returns:

Name	Type	Description
int	int	Total number of stored entries.

delete(entry_id) async

Delete an entry by ID.

Note: This rewrites the file without the deleted entry, which may be slow for large files. Consider using retention policies instead.

Parameters:

Name	Type	Description	Default
entry_id	str	The unique identifier of the entry to delete.	required

Returns:

Name	Type	Description
bool	bool	True if the entry was deleted, False if not found.

list_entries(start, end) async

List all entries within a time range.

Parameters:

Name	Type	Description	Default
start	datetime	Start of the time range (inclusive).	required
end	datetime	End of the time range (inclusive).	required

Returns:

Type	Description
List[AuditEntry]	List[AuditEntry]: Entries within the specified time range.

read(entry_id) async

Read an audit entry by ID.

Searches through all JSONL files if the entry is not in the index.

Parameters:

Name	Type	Description	Default
entry_id	str	The unique identifier of the entry.	required

Returns:

Type	Description
AuditEntry | None	AuditEntry | None: The entry if found, None otherwise.

write(entry) async

Write an audit entry to a JSONL file.

Auto-rotates the file if it exceeds the configured size limit.

Parameters:

Name	Type	Description	Default
entry	AuditEntry	The audit entry to store.	required

Returns:

Name	Type	Description
str	str	The entry ID.

File-based storage backend using JSONL format.

Constructor

FileStorage(path: str, rotation_size_mb: int = 100)

Parameters:

Parameter	Type	Default	Description
path	str	Required	Directory for audit files
rotation_size_mb	int	100	Max file size before rotation

File Structure:

{path}/
├── audit_20260128.jsonl
├── audit_20260128_001.jsonl  # Rotated
├── audit_20260129.jsonl
└── ...

Example:

from rotalabs_comply.audit import FileStorage

storage = FileStorage("/var/log/audit", rotation_size_mb=50)
entry_id = await storage.write(entry)

---

MemoryStorage

In-memory storage backend for testing and development.

Stores entries in a dictionary. Data is lost when the process ends.

Parameters:

Name	Type	Description	Default
max_entries	int | None	Optional maximum number of entries to store. When exceeded, oldest entries are removed.	None

Example

storage = MemoryStorage(max_entries=1000) entry_id = await storage.write(entry) count = await storage.count()

__init__(max_entries=None)

Initialize memory storage.

Parameters:

Name	Type	Description	Default
max_entries	int | None	Optional maximum number of entries to store.	None

count() async

Count total number of entries in storage.

Returns:

Name	Type	Description
int	int	Total number of stored entries.

delete(entry_id) async

Delete an entry by ID.

Parameters:

Name	Type	Description	Default
entry_id	str	The unique identifier of the entry to delete.	required

Returns:

Name	Type	Description
bool	bool	True if the entry was deleted, False if not found.

list_entries(start, end) async

List all entries within a time range.

Parameters:

Name	Type	Description	Default
start	datetime	Start of the time range (inclusive).	required
end	datetime	End of the time range (inclusive).	required

Returns:

Type	Description
List[AuditEntry]	List[AuditEntry]: Entries within the specified time range.

read(entry_id) async

Read an audit entry by ID.

Parameters:

Name	Type	Description	Default
entry_id	str	The unique identifier of the entry.	required

Returns:

Type	Description
AuditEntry | None	AuditEntry | None: The entry if found, None otherwise.

write(entry) async

Write an audit entry to memory.

If max_entries is set and exceeded, removes the oldest entry.

Parameters:

Name	Type	Description	Default
entry	AuditEntry	The audit entry to store.	required

Returns:

Name	Type	Description
str	str	The entry ID.

In-memory storage backend for testing and development.

Constructor

MemoryStorage(max_entries: Optional[int] = None)

Parameters:

Parameter	Type	Default	Description
max_entries	Optional[int]	None	Max entries (LRU eviction)

Example:

from rotalabs_comply.audit import MemoryStorage

storage = MemoryStorage(max_entries=1000)
count = await storage.count()

Data Persistence

Data is lost when the process ends. Use only for testing.

---

S3Storage

AWS S3 storage backend for audit logs.

Stores each audit entry as a separate JSON file in S3, organized by date for easy querying and lifecycle management.

Requires boto3 to be installed (optional dependency).

Parameters:

Name	Type	Description	Default
bucket	str	S3 bucket name.	required
prefix	str	Key prefix for audit files (default: "audit/").	'audit/'
region	str | None	AWS region (optional, uses default if not specified).	None

File structure

{prefix}{YYYY-MM-DD}/{entry_id}.json

Example

storage = S3Storage("my-audit-bucket", prefix="logs/audit/") entry_id = await storage.write(entry)

Stored at: s3://my-audit-bucket/logs/audit/2024-01-15/abc123.json

__init__(bucket, prefix='audit/', region=None)

Initialize S3 storage.

Parameters:

Name	Type	Description	Default
bucket	str	S3 bucket name.	required
prefix	str	Key prefix for audit files.	'audit/'
region	str | None	AWS region (optional).	None

count() async

Count total number of entries in storage.

Returns:

Name	Type	Description
int	int	Total number of stored entries.

delete(entry_id) async

Delete an entry by ID.

Parameters:

Name	Type	Description	Default
entry_id	str	The unique identifier of the entry to delete.	required

Returns:

Name	Type	Description
bool	bool	True if the entry was deleted, False if not found.

list_entries(start, end) async

List all entries within a time range.

Parameters:

Name	Type	Description	Default
start	datetime	Start of the time range (inclusive).	required
end	datetime	End of the time range (inclusive).	required

Returns:

Type	Description
List[AuditEntry]	List[AuditEntry]: Entries within the specified time range.

read(entry_id) async

Read an audit entry by ID.

Note: This searches through date prefixes which may be slow. Consider maintaining an index for production use.

Parameters:

Name	Type	Description	Default
entry_id	str	The unique identifier of the entry.	required

Returns:

Type	Description
AuditEntry | None	AuditEntry | None: The entry if found, None otherwise.

write(entry) async

Write an audit entry to S3.

Parameters:

Name	Type	Description	Default
entry	AuditEntry	The audit entry to store.	required

Returns:

Name	Type	Description
str	str	The entry ID.

AWS S3 storage backend for audit logs.

Constructor

S3Storage(
    bucket: str,
    prefix: str = "audit/",
    region: Optional[str] = None,
)

Parameters:

Parameter	Type	Default	Description
bucket	str	Required	S3 bucket name
prefix	str	"audit/"	Key prefix for files
region	Optional[str]	None	AWS region

Key Structure:

s3://{bucket}/{prefix}{YYYY-MM-DD}/{entry_id}.json

Example:

from rotalabs_comply.audit import S3Storage

storage = S3Storage(
    bucket="my-audit-bucket",
    prefix="prod/audit/",
    region="us-west-2",
)

Dependency

Requires boto3. Install with pip install rotalabs-comply[s3].

---

AuditEntry (Storage)

Represents a single audit log entry.

Captures all relevant information about an AI interaction including inputs, outputs, safety evaluations, and performance metrics.

Attributes:

Name	Type	Description
id	str	Unique identifier for this entry.
timestamp	str	When the interaction occurred (ISO format).
input_hash	str	SHA-256 hash of the input content.
output_hash	str	SHA-256 hash of the output content.
input_content	str | None	Actual input content (if store_content=True, may be encrypted).
output_content	str | None	Actual output content (if store_content=True, may be encrypted).
provider	str | None	The AI provider (e.g., "openai", "anthropic").
model	str | None	The model identifier (e.g., "gpt-4", "claude-3-opus").
conversation_id	str | None	Optional ID linking related interactions.
safety_passed	bool	Whether the interaction passed all safety checks.
detectors_triggered	List[str]	List of safety detector names that triggered.
block_reason	str | None	Reason for blocking, if the request was blocked.
alerts	List[str]	List of alert messages generated.
latency_ms	float	Time taken to process the request in milliseconds.
input_tokens	int | None	Number of tokens in the input.
output_tokens	int | None	Number of tokens in the output.
metadata	Dict[str, Any]	Additional custom metadata.

from_dict(data) classmethod

Create entry from dictionary.

to_dict()

Convert entry to dictionary for serialization.

Dataclass representing a single audit log entry in storage.

Attributes:

Attribute	Type	Description
id	str	Unique identifier
timestamp	str	ISO format timestamp
input_hash	str	SHA-256 of input
output_hash	str	SHA-256 of output
input_content	Optional[str]	Input content (if stored)
output_content	Optional[str]	Output content (if stored)
provider	Optional[str]	AI provider
model	Optional[str]	Model identifier
conversation_id	Optional[str]	Conversation ID
safety_passed	bool	Safety check result
detectors_triggered	List[str]	Triggered detectors
block_reason	Optional[str]	Block reason
alerts	List[str]	Alert messages
latency_ms	float	Response latency
input_tokens	Optional[int]	Input token count
output_tokens	Optional[int]	Output token count
metadata	Dict[str, Any]	Custom metadata

Methods:

Method	Description
to_dict()	Convert to dictionary
from_dict(data)	Create from dictionary

---

Helper Functions

create_entry_id

def create_entry_id() -> str

Generate a unique entry ID (UUID v4).

Example:

from rotalabs_comply.audit.storage import create_entry_id

entry_id = create_entry_id()
# Returns: "550e8400-e29b-41d4-a716-446655440000"