sumanth-audit-sdk 1.0.0

Universal Audit Framework (UAF) — A production-ready, framework-agnostic audit logging SDK for Python.

Universal Audit Framework (UAF) SDK

A production-ready, framework-agnostic audit logging SDK for Python 3.12+.

Built with Clean Architecture, SOLID principles, and a plugin-first mindset — UAF SDK integrates seamlessly with FastAPI, or any async Python application.

---

Table of Contents

• Features
• Installation
• Quick Start
• Architecture
• Transport Backends
• Middleware Integration
• Extension Points
• Configuration
• Security
• API Reference
• Examples
• Contributing
• License

---

Features

• Zero-code audit logging — middleware auto-captures every request
• Enhanced Audit Middleware — auto-captures user, action, state, and HTTP context
• Auto action messages — human-readable messages like "Sumanth created Employee John"
• Auto state capture — previous_state, current_state, changed_fields diff
• Smart user detection — JWT, OAuth, session, request.state, headers, custom providers
• Declarative user mapping — configure once, works across all endpoints
• Multiple transports — REST, PostgreSQL, SQLite, MongoDB, Kafka
• Pydantic v2 validation — all events are validated before dispatch
• Async-first — built on asyncio + httpx
• Retry with backoff — ensures audit events are never permanently lost
• Event signing & encryption — HMAC-SHA256 + AES-256-GCM
• Plugin architecture — extend the SDK without forking
• 100% typed — mypy --strict compliant (PEP 561)
• Well-tested — comprehensive unit and integration test suite

---

Installation

# Core SDK (REST transport included)
pip install audit-sdk

# With FastAPI middleware support
pip install audit-sdk[fastapi]

# With PostgreSQL transport
pip install audit-sdk[postgres]

# With SQLite transport
pip install audit-sdk[sqlite]

# With MongoDB transport
pip install audit-sdk[mongodb]

# Install everything
pip install audit-sdk[all]

---

Quick Start

1. REST Transport (simplest)

from audit_sdk import AuditClient, AuditSettings

settings = AuditSettings(endpoint="https://audit.example.com/api/v1/events")
audit = AuditClient(settings=settings)

await audit.log(
    action="CREATE",
    entity="Employee",
    entity_id="123",
    user={"id": "1", "name": "Sumanth"},
    data={"name": "John", "department": "Engineering"},
)

2. Database Transport (direct to PostgreSQL)

from audit_sdk import AuditClient
from audit_sdk.transport.database import DatabaseTransport

transport = DatabaseTransport.for_postgres(
    host="localhost",
    port=5432,
    database="audit_db",
    username="your_user",
    password="your_password",
)
audit = AuditClient(transport=transport)

3. FastAPI Middleware (zero-code auditing)

from fastapi import FastAPI
from audit_sdk.middleware.enhanced import (
    EnhancedAuditMiddleware,
    EnhancedAuditConfig,
)
from audit_sdk.transport.simple_database import SimpleAuditTransport

app = FastAPI()

# Transport — writes to PostgreSQL audit_log table
transport = SimpleAuditTransport.for_postgres(
    host="localhost",
    port=5432,
    database="audit_db",
    username="your_user",
    password="your_password",
)

# Register middleware — ALL endpoints are automatically audited
app.add_middleware(
    EnhancedAuditMiddleware,
    transport=transport,
    config=EnhancedAuditConfig(
        capture_request_body=True,
        excluded_paths=frozenset({"/health", "/docs", "/openapi.json"}),
    ),
)

That's it. Every endpoint is automatically audited. No audit.log() calls anywhere.

---

Architecture

+---------------------------------------------------+
|                Application Layer                   |
|             AuditClient (audit.py)                 |
+---------------------------------------------------+
|                 Domain Layer                       |
|         models/  .  core/interfaces                |
+---------------------------------------------------+
|              Infrastructure Layer                  |
|   transport/  .  security/  .  plugins/            |
+---------------------------------------------------+
|             Cross-Cutting Concerns                 |
|      config/  .  utils/  .  middleware/            |
+---------------------------------------------------+

Package Structure

audit_sdk/
├── audit.py             # AuditClient — the SDK entry point
├── config/              # Pydantic-based configuration (env-driven)
├── core/                # Interfaces, exceptions, event bus
├── middleware/          # FastAPI middleware (Enhanced, Auto, Standard)
├── models/             # AuditEvent model, enums
├── plugins/            # Plugin architecture (extend without forking)
├── security/           # HMAC signing, AES-256-GCM encryption
├── transport/          # REST, PostgreSQL, SQLite, MongoDB, Kafka
└── utils/              # Retry manager, serializer, logger

---

Transport Backends

Transport	Install Extra	Target	Description
RestTransport	(core)	HTTP endpoint	REST via httpx with retry
SimpleAuditTransport	[postgres] or [sqlite]	audit_log table	Enhanced 10-column schema
DatabaseTransport	[postgres] or [sqlite]	audit_events table	Full 18-column schema
MongoDBTransport	[mongodb]	MongoDB collection	Document-based storage
KafkaTransport	[kafka]	Kafka topic	Event streaming

PostgreSQL Setup

from audit_sdk.transport.simple_database import SimpleAuditTransport

transport = SimpleAuditTransport.for_postgres(
    host="localhost",
    port=5432,
    database="audit_db",
    username="your_user",
    password="your_password",
)

The table is auto-created on first use:

CREATE TABLE audit_log (
    id              VARCHAR(36) PRIMARY KEY,
    performed_by    TEXT NOT NULL,
    action          VARCHAR(100) NOT NULL,
    action_message  TEXT,
    functionality   VARCHAR(200) NOT NULL,
    performed_on    TEXT NOT NULL,
    performed_at    TIMESTAMPTZ NOT NULL,
    previous_state  JSONB,
    current_state   JSONB,
    changed_fields  JSONB,
    http_context    JSONB
);

SQLite Setup

transport = SimpleAuditTransport.for_sqlite(path="./audit.db")

MongoDB Setup

from audit_sdk.transport.mongodb import MongoDBTransport

transport = MongoDBTransport(
    uri="mongodb://localhost:27017",
    database="audit_db",
)

---

Middleware Integration

EnhancedAuditMiddleware (Recommended)

The most feature-rich middleware. Auto-captures everything:

Field	Description	Example
performed_by	Username (auto-detected)	Sumanth
action	Business action	CREATE, UPDATE, DELETE, LOGIN
action_message	Human-readable message	Sumanth created Employee John
functionality	Module/feature area	Employee Management
performed_on	Target entity	Employee #1
performed_at	UTC timestamp	2025-01-15 13:45:00+00
previous_state	Record before change	{"salary": 100000}
current_state	Record after change	{"salary": 120000}
changed_fields	Field-level diff	{"salary": {"old": 100000, "new": 120000}}
http_context	Request metadata	{"method": "PUT", "status": 200, ...}

User Detection Priority

The middleware resolves the authenticated user automatically:

1. Custom UserProvider — your own async resolver
2. user_info mapping — declarative field mapping (configure once)
3. request.state.user — automatic detection
4. request.user — Starlette AuthenticationMiddleware
5. JWT Bearer token — decodes payload automatically
6. Request headers — X-User-ID, X-User-Name, X-User-Email
7. Anonymous — fallback when no auth is available

Declarative User Mapping

Configure user field mapping once during middleware registration:

from audit_sdk.middleware.fastapi import AuditMiddleware, UserInfoMapping

app.add_middleware(
    AuditMiddleware,
    audit_client=audit_client,
    user_info=UserInfoMapping(
        source="request.state.current_user",
        id="emp_id",
        username="name",
        email="email",
        roles="roles",
    ),
)

Custom UserProvider

from audit_sdk.middleware.enhanced import UserProvider

class MyUserProvider:
    async def get_user(self, request) -> dict | None:
        token = request.headers.get("authorization", "").replace("Bearer ", "")
        claims = decode_jwt(token)
        return {
            "user_id": claims["sub"],
            "username": claims["preferred_username"],
            "full_name": claims["name"],
            "email": claims["email"],
            "roles": claims.get("roles", []),
        }

---

Extension Points

StateProvider — Supply previous/current state

from audit_sdk.middleware.enhanced import StateProvider

class MyStateProvider:
    async def get_previous_state(self, entity: str, entity_id: str, request) -> dict | None:
        record = await db.get(entity, entity_id)
        return record.to_dict() if record else None

    async def get_current_state(self, entity: str, entity_id: str, request) -> dict | None:
        record = await db.get(entity, entity_id)
        return record.to_dict() if record else None

State Capture Flow

Request arrives
    |
    +-- BEFORE handler --> get_previous_state() (PUT/PATCH/DELETE)
    |
    +-- Handler executes (modifies database)
    |
    +-- AFTER handler  --> get_current_state()  (POST/PUT/PATCH)
                           --> auto-computes changed_fields diff

Action	previous_state	current_state	changed_fields
CREATE	null	{new record}	null
UPDATE	{before}	{after}	{"field": {"old": x, "new": y}}
DELETE	{deleted record}	null	null
READ	null	null	null

Plugin System

from audit_sdk.plugins import AbstractPlugin, PluginRegistry

class PIIRedactorPlugin(AbstractPlugin):
    name = "pii-redactor"

    async def before_send(self, event):
        # Redact sensitive fields before audit log is stored
        if "ssn" in event.data:
            event.data["ssn"] = "***-**-****"
        return event

registry = PluginRegistry()
registry.register(PIIRedactorPlugin())

---

Configuration

EnhancedAuditConfig

from audit_sdk.middleware.enhanced import EnhancedAuditConfig

config = EnhancedAuditConfig(
    excluded_paths=frozenset({"/health", "/docs", "/metrics"}),
    include_methods=frozenset({"GET", "POST", "PUT", "PATCH", "DELETE"}),
    capture_request_body=True,
    max_body_size=10 * 1024,  # 10 KiB
)

AuditSettings (Environment Variables)

from audit_sdk import AuditSettings

settings = AuditSettings(
    endpoint="https://audit.example.com/api/v1/events",
    api_key="your-api-key",
    timeout=30,
    max_retries=3,
)

Configuration can also be loaded from environment variables with the AUDIT_ prefix:

export AUDIT_ENDPOINT=https://audit.example.com/api/v1/events
export AUDIT_API_KEY=your-api-key

---

Security

• Event Signing — HMAC-SHA256 signatures ensure event integrity
• Field Encryption — AES-256-GCM for sensitive audit data
• No secrets in logs — API keys and passwords are masked in all output

from audit_sdk.security import HMACSigner, AESEncryptor

signer = HMACSigner(secret="your-signing-secret")
encryptor = AESEncryptor(key="base64-encoded-32-byte-key")

---

API Reference

Core Classes

Class	Import	Description
AuditClient	from audit_sdk import AuditClient	Main SDK entry point
AuditSettings	from audit_sdk import AuditSettings	Configuration object
AuditEvent	from audit_sdk import AuditEvent	Event data model

Middleware

Class	Import	Description
EnhancedAuditMiddleware	from audit_sdk.middleware.enhanced import ...	Full-featured middleware
AuditMiddleware	from audit_sdk.middleware.fastapi import ...	Standard middleware
AutoAuditMiddleware	from audit_sdk.middleware.auto_audit import ...	Auto-detect middleware

Transports

Class	Import	Description
RestTransport	from audit_sdk.transport import RestTransport	HTTP/REST transport
SimpleAuditTransport	from audit_sdk.transport.simple_database import ...	Enhanced DB transport
DatabaseTransport	from audit_sdk.transport.database import ...	Full DB transport
MongoDBTransport	from audit_sdk.transport.mongodb import ...	MongoDB transport

Utilities

Class	Import	Description
RetryManager	from audit_sdk import RetryManager	Async retry with backoff
PluginRegistry	from audit_sdk.plugins import PluginRegistry	Plugin management
AbstractPlugin	from audit_sdk.plugins import AbstractPlugin	Plugin base class

---

Examples

Full working examples are available in the GitHub repository:

• FastAPI + PostgreSQL — Complete CRUD app with auto-auditing
• REST Transport — Sending events to an HTTP collector
• Database Verify — Querying and displaying audit records
• Auto Audit Demo — Zero-config automatic auditing

Client Application Demo

A complete sample HR application demonstrating SDK integration is available at: client_app/

---

Retry Architecture

The SDK ensures audit events are never permanently lost:

AuditClient.publish(...)
        |
        v
Transport.send(event)       <-- first attempt
        |
     Retryable failure?       Non-retryable (4xx)?
        |                     --> raise TransportError
        v
RetryManager.enqueue(event)  <-- background retry with exponential backoff

Condition	Action
Network error	Retry with backoff
Timeout	Retry with backoff
HTTP 429 / 5xx	Retry with backoff
HTTP 400/401/403/404	Abort (non-retryable)

---

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

---

License

MIT License - see LICENSE for details.

Copyright (c) 2025 Sumanth