lyzr-audit-logger-dev 0.1.11
Audit logging library for microservices
License: MIT
Requires Python: >=3.10

# Audit Logger

A Python package for audit logging across microservices. Provides async, non-blocking audit log capture with MongoDB storage.

## Installation

```bash
pip install audit-logger
```

Or install from source:

```bash
# Using HTTPS
pip install git+https://github.com/NeuralgoLyzr/audit-logger.git

# Using SSH
pip install git+ssh://git@github.com/NeuralgoLyzr/audit-logger.git
```

For local development:

```bash
cd audit_logger
pip install -e .
```

## Quick Start

### 1. Initialize at Application Startup

```python
from audit_logger import init_audit_logger

# Option 1: Use environment variable AUDIT_MONGO_URL (defaults to mongodb://localhost:27017)
audit_manager, audit_logger = init_audit_logger()

# Option 2: Explicit configuration
audit_manager, audit_logger = init_audit_logger(
    mongo_url="mongodb://audit-db:27017",
    db_name="my_service_audit",
    collection_name="audit_logs"
)

# Create indexes (call once on startup)
await audit_manager.ensure_indexes()
```

### 2. Set Context in Middleware

```python
from audit_logger import set_audit_context, clear_audit_context

@app.middleware("http")
async def audit_middleware(request: Request, call_next):
    # Set context from auth info
    set_audit_context(
        org_id=request.state.auth_user.org_id,
        user_id=request.state.auth_user.user_id,
        api_key=request.headers.get("x-api-key"),
        ip_address=request.client.host,
    )
    try:
        return await call_next(request)
    finally:
        clear_audit_context()
```

### 3. Log Audit Events

```python
from audit_logger import audit_logger, AuditResource

# Log a create action (non-blocking, returns immediately)
audit_logger.log_create(
    resource_type=AuditResource.AGENT,
    resource_id="agent_123",
    resource_name="My Agent",
    metadata={"tools_count": 5}
)

# Log an update action
from audit_logger import AuditChange

audit_logger.log_update(
    resource_type=AuditResource.AGENT,
    resource_id="agent_123",
    changes=[
        AuditChange(field="name", old_value="Old Name", new_value="New Name"),
        AuditChange(field="description", old_value=None, new_value="New description"),
    ]
)

# Log a delete action
audit_logger.log_delete(
    resource_type=AuditResource.AGENT,
    resource_id="agent_123",
    resource_name="My Agent"
)
```

### 4. Shutdown Gracefully

```python
from audit_logger import shutdown_audit_logger

@app.on_event("shutdown")
async def shutdown():
    await shutdown_audit_logger()
```

## Advanced Usage

### Direct Manager Access

For more control, use the `AuditLogManager` directly:

```python
from audit_logger import get_audit_log_manager, AuditResource, AuditResult

manager = get_audit_log_manager()

# Log with full control
manager.log_create(
    org_id="org_123",
    resource_type=AuditResource.TOOL,
    resource_id="tool_456",
    user_id="user_789",
    api_key="sk-xxx",
    ip_address="192.168.1.1",
    resource_name="My Tool",
    metadata={"type": "openapi"},
    result=AuditResult.SUCCESS,
)

# Log inference events
manager.log_inference(
    org_id="org_123",
    agent_id="agent_456",
    session_id="session_789",
    user_id="user_123",
    result=AuditResult.SUCCESS,
    metadata={"model": "gpt-4", "tokens": 1500}
)

# Log guardrail violations
from audit_logger import GuardrailViolation, AuditSeverity

manager.log_guardrail_violation(
    org_id="org_123",
    session_id="session_789",
    violations=[
        GuardrailViolation(
            violation_type="pii",
            severity=AuditSeverity.MEDIUM,
            details={"entities": ["email", "phone"]}
        )
    ]
)
```

### Query Audit Logs

```python
from audit_logger import get_audit_log_manager, AuditLogQuery, AuditResource

manager = get_audit_log_manager()

# Query with filters
logs = await manager.query(AuditLogQuery(
    org_id="org_123",
    resource_type=AuditResource.AGENT,
    limit=50,
))

# Get history for a specific resource
history = await manager.get_by_resource(
    org_id="org_123",
    resource_type=AuditResource.AGENT,
    resource_id="agent_456",
)

# Get session activity
session_logs = await manager.get_by_session(session_id="session_789")

# Count matching logs
count = await manager.count(AuditLogQuery(org_id="org_123"))
```

### Using the Decorator

```python
from audit_logger import audit_action, AuditAction, AuditResource

@audit_action(
    action=AuditAction.DELETE,
    resource_type=AuditResource.AGENT,
    resource_id_param="agent_id",
)
async def delete_agent(agent_id: str, _audit_manager=None):
    # Delete logic here
    pass
```

### Context Manager

```python
from audit_logger import audit_context

async with audit_context(org_id="org_123", user_id="user_456") as request_id:
    # All audit logs in this block will have the context set
    audit_logger.log_create(AuditResource.AGENT, "agent_789")
```

## Configuration

### Environment Variables

| Variable | Default | Description |
|----------|---------|-------------|
| `AUDIT_MONGO_URL` | `mongodb://localhost:27017` | MongoDB connection URL |

### Programmatic Configuration

```python
from audit_logger import init_audit_logger

# Full configuration
manager, logger = init_audit_logger(
    mongo_url="mongodb://audit-db:27017",
    db_name="lyzr_audit_logs",
    collection_name="audit_logs",
)
```

### Using Existing Collection

If you already have a MongoDB collection configured:

```python
from motor.motor_asyncio import AsyncIOMotorClient
from audit_logger import init_audit_logger

client = AsyncIOMotorClient("mongodb://localhost:27017")
collection = client["my_db"]["audit_logs"]

manager, logger = init_audit_logger(collection=collection)
```

## Available Enums

### AuditAction
- `CREATE`, `READ`, `UPDATE`, `DELETE`
- `EXECUTE`, `LOGIN`, `LOGOUT`
- `ACCESS_DENIED`, `EXPORT`, `IMPORT`

### AuditResource
- `AGENT`, `TOOL`, `PROVIDER`, `WORKFLOW`
- `SESSION`, `MESSAGE`, `CONTEXT`, `MEMORY`
- `CREDENTIAL`, `USER`, `ORGANIZATION`, `API_KEY`
- `INFERENCE`, `GUARDRAIL`, `ARTIFACT`

### AuditResult
- `SUCCESS`, `FAILURE`, `BLOCKED`, `PARTIAL`

### AuditSeverity
- `LOW`, `MEDIUM`, `HIGH`, `CRITICAL`

## MongoDB Schema

```javascript
{
    "_id": ObjectId,
    "timestamp": ISODate,
    "actor": {
        "org_id": "org_123",
        "user_id": "user_456",
        "api_key_hash": "a1b2c3d4...xyz",
        "ip_address": "192.168.1.1"
    },
    "action": "create",
    "target": {
        "resource_type": "agent",
        "resource_id": "agent_789",
        "resource_name": "Support Bot"
    },
    "result": "success",
    "error_message": null,
    "session_id": "session_abc",
    "request_id": "req_def",
    "changes": [...],
    "guardrail_violations": [...],
    "metadata": {...},
    "severity": "low"
}
```

## Non-Blocking Writes

All logging methods are **non-blocking** by default. They use `asyncio.create_task()` to write to MongoDB in the background without blocking your request handlers.

For synchronous writes (when you need confirmation):

```python
log_id = await manager.log_sync(entry)
```

## License

MIT