SignLedger

SignLedger is a production-ready Python library for immutable audit logging with cryptographic verification. Built for compliance, security, and trust, SignLedger provides blockchain-inspired tamper-proof logging for critical applications.

Features

• Immutable Audit Logs: Write-once, tamper-proof entries with hash chain verification
• Cryptographic Signatures: RSA, ECDSA, and Ed25519 digital signatures for non-repudiation
• Hash Chain Integrity: SHA-256/SHA-512/SHA-3 cryptographic linking of entries
• Multiple Storage Backends: In-memory, SQLite, PostgreSQL, MongoDB support
• High Performance: 2,000+ transactions per second with batch operations
• Compression: 50-70% storage reduction with zlib, gzip, lz4, or zstd
• Framework Integration: Built-in support for Django, Flask, and FastAPI
• Thread-Safe: Production-ready concurrent access with RLock protection
• Type-Safe: Full type hints and Pydantic v2 validation

Installation

pip install signledger

Optional Dependencies

# PostgreSQL backend
pip install signledger[postgresql]

# MongoDB backend
pip install signledger[mongodb]

# All backends
pip install signledger[all]

# Framework integrations
pip install signledger[django]
pip install signledger[flask]
pip install signledger[fastapi]

Quick Start

from signledger import Ledger
from signledger.backends import InMemoryBackend

# Create a ledger
ledger = Ledger(backend=InMemoryBackend())

# Append entries
entry = ledger.append({
    "action": "user_login",
    "user_id": "user123",
    "ip_address": "192.168.1.1"
})

# Query entries
entries = list(ledger.get_entries(limit=10))

# Verify integrity
is_valid = ledger.verify_integrity()  # Returns True

# Get statistics
stats = ledger.get_stats()
print(f"Total entries: {stats.total_entries}")

Use Cases

1. Financial Transaction Logging

Immutable audit trails for banking and financial services compliance (SOX, PCI-DSS).

from signledger import Ledger
from signledger.backends import SQLiteBackend
from signledger.crypto.signatures import RSASigner
from datetime import datetime

# Initialize ledger with signature support
signer = RSASigner()
ledger = Ledger(
    backend=SQLiteBackend("financial_audit.db"),
    enable_signatures=True
)

# Log financial transaction
entry = ledger.append(
    data={
        "transaction_id": "TXN123456",
        "account_from": "ACC001",
        "account_to": "ACC002",
        "amount": 15000.00,
        "currency": "USD",
        "timestamp": datetime.utcnow().isoformat(),
        "operator_id": "OPR789"
    },
    metadata={
        "compliance": "SOX",
        "category": "wire_transfer"
    },
    sign=True,
    signer=signer.sign
)

# Verify signature later
is_valid = signer.verify(entry.hash, entry.signature)

2. Healthcare Access Logs (HIPAA Compliance)

Track all access to protected health information (PHI) with non-repudiation.

from signledger import Ledger
from signledger.backends import PostgreSQLBackend

# HIPAA-compliant audit logging
ledger = Ledger(
    backend=PostgreSQLBackend(
        host="localhost",
        database="healthcare_audit",
        user="audit_user",
        password="secure_password"
    )
)

# Log PHI access
ledger.append({
    "event_type": "PHI_ACCESS",
    "patient_id": "P123456",
    "accessed_by": "DR_SMITH",
    "access_type": "READ",
    "data_accessed": ["medical_history", "prescriptions"],
    "purpose": "treatment",
    "facility": "General Hospital"
})

3. Supply Chain Provenance

Track product journey from manufacture to delivery.

from signledger import Ledger

ledger = Ledger()

# Manufacture
ledger.append({
    "event": "manufactured",
    "product_id": "PROD-2024-001",
    "batch": "BATCH-A123",
    "location": "Factory Shanghai",
    "quality_check": "PASSED"
})

# Shipping
ledger.append({
    "event": "shipped",
    "product_id": "PROD-2024-001",
    "carrier": "DHL",
    "tracking": "DHL123456789",
    "destination": "Warehouse NYC"
})

# Delivery
ledger.append({
    "event": "delivered",
    "product_id": "PROD-2024-001",
    "recipient": "John Doe",
    "signature": "base64_signature_data"
})

# Query product history
product_history = ledger.query(
    lambda e: e.data.get("product_id") == "PROD-2024-001"
)

4. Software Deployment Audit

Track all production deployments with approval workflows.

from signledger import Ledger

ledger = Ledger()

# Log deployment
ledger.append({
    "deployment_id": "DEPLOY-2024-042",
    "application": "api-service",
    "version": "v2.3.1",
    "environment": "production",
    "deployed_by": "deploy_bot",
    "approved_by": ["alice@company.com", "bob@company.com"],
    "commit_sha": "a1b2c3d4",
    "rollback_plan": "automatic",
    "health_check": "PASSED"
})

5. Database Change Tracking

Immutable record of all database schema and data changes.

from signledger import Ledger

ledger = Ledger()

# Track schema change
ledger.append({
    "change_type": "schema_migration",
    "database": "production_db",
    "migration_id": "20240101_add_user_roles",
    "sql": "ALTER TABLE users ADD COLUMN role VARCHAR(50)",
    "executed_by": "admin",
    "rollback_sql": "ALTER TABLE users DROP COLUMN role"
})

Framework Integration

Django Integration

# settings.py
INSTALLED_APPS = [
    ...
    'signledger.integrations.django',
]

SIGNLEDGER = {
    'BACKEND': 'signledger.backends.PostgreSQLBackend',
    'OPTIONS': {
        'host': 'localhost',
        'database': 'audit_log',
        'user': 'postgres',
        'password': 'password'
    },
    'ENABLE_SIGNATURES': True,
    'AUTO_VERIFY': True
}

# views.py
from signledger.integrations.django import get_ledger

def transfer_money(request):
    ledger = get_ledger()

    # Perform transaction
    transaction = process_transfer(request.data)

    # Log to immutable ledger
    ledger.append({
        "action": "money_transfer",
        "user": request.user.id,
        "amount": transaction.amount,
        "status": "completed"
    })

    return JsonResponse({"status": "success"})

Flask Integration

from flask import Flask
from signledger import Ledger
from signledger.backends import SQLiteBackend

app = Flask(__name__)

# Initialize ledger
ledger = Ledger(backend=SQLiteBackend("audit.db"))

@app.before_request
def log_request():
    """Log all API requests"""
    from flask import request

    if request.endpoint:
        ledger.append({
            "event": "api_request",
            "method": request.method,
            "endpoint": request.endpoint,
            "ip": request.remote_addr,
            "user_agent": request.user_agent.string
        })

@app.route('/api/transfer', methods=['POST'])
def transfer():
    data = request.json

    # Process transfer
    result = process_payment(data)

    # Log transaction
    ledger.append({
        "action": "payment",
        "from": data['from_account'],
        "to": data['to_account'],
        "amount": data['amount'],
        "status": result['status']
    })

    return jsonify(result)

@app.route('/api/audit/verify', methods=['GET'])
def verify_integrity():
    """Verify ledger integrity"""
    is_valid = ledger.verify_integrity()
    stats = ledger.get_stats()

    return jsonify({
        "integrity": is_valid,
        "total_entries": stats.total_entries,
        "last_entry": stats.last_entry_time.isoformat() if stats.last_entry_time else None
    })

if __name__ == '__main__':
    app.run()

FastAPI Integration

from fastapi import FastAPI, HTTPException
from signledger import Ledger
from signledger.backends import PostgreSQLBackend
from pydantic import BaseModel
from datetime import datetime

app = FastAPI()

# Initialize ledger
ledger = Ledger(
    backend=PostgreSQLBackend(
        host="localhost",
        database="audit_db",
        user="postgres",
        password="password"
    )
)

class Transaction(BaseModel):
    from_account: str
    to_account: str
    amount: float
    description: str

@app.post("/api/transaction")
async def create_transaction(txn: Transaction):
    """Process and audit a transaction"""

    # Log transaction
    entry = await ledger.append_async({
        "type": "transaction",
        "from": txn.from_account,
        "to": txn.to_account,
        "amount": txn.amount,
        "description": txn.description,
        "timestamp": datetime.utcnow().isoformat()
    })

    return {
        "transaction_id": entry.id,
        "hash": entry.hash,
        "timestamp": entry.timestamp
    }

@app.get("/api/audit/entries")
async def get_audit_entries(limit: int = 100):
    """Retrieve recent audit entries"""
    entries = list(ledger.get_entries(limit=limit))

    return {
        "entries": [
            {
                "id": e.id,
                "timestamp": e.timestamp.isoformat(),
                "data": e.data,
                "hash": e.hash
            }
            for e in entries
        ]
    }

@app.get("/api/audit/verify")
async def verify_audit_integrity():
    """Verify ledger integrity"""
    try:
        is_valid = await ledger.verify_integrity_async()
        stats = ledger.get_stats()

        return {
            "valid": is_valid,
            "total_entries": stats.total_entries,
            "first_entry": stats.first_entry_time.isoformat() if stats.first_entry_time else None,
            "last_entry": stats.last_entry_time.isoformat() if stats.last_entry_time else None
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Celery Integration

from celery import Celery
from signledger import Ledger
from signledger.backends import RedisBackend

app = Celery('tasks', broker='redis://localhost:6379')
ledger = Ledger(backend=RedisBackend(host='localhost', port=6379))

@app.task
def process_order(order_id):
    """Process order and log to audit trail"""

    # Process order
    result = handle_order(order_id)

    # Log to ledger
    ledger.append({
        "task": "process_order",
        "order_id": order_id,
        "status": result['status'],
        "processed_at": datetime.utcnow().isoformat()
    })

    return result

Simple Python Program

from signledger import Ledger
from signledger.backends import InMemoryBackend
from datetime import datetime

def main():
    # Create ledger
    ledger = Ledger(backend=InMemoryBackend())

    # Simulate user activities
    activities = [
        {"user": "alice", "action": "login", "ip": "192.168.1.10"},
        {"user": "alice", "action": "view_document", "doc_id": "DOC123"},
        {"user": "bob", "action": "login", "ip": "192.168.1.11"},
        {"user": "alice", "action": "edit_document", "doc_id": "DOC123"},
        {"user": "bob", "action": "delete_document", "doc_id": "DOC456"},
    ]

    # Log activities
    for activity in activities:
        activity["timestamp"] = datetime.utcnow().isoformat()
        entry = ledger.append(activity)
        print(f"Logged: {entry.id} - {activity['action']}")

    # Query specific user's activities
    alice_activities = ledger.query(
        lambda e: e.data.get("user") == "alice"
    )

    print(f"\nAlice's activities: {len(alice_activities)}")
    for entry in alice_activities:
        print(f"  - {entry.data['action']} at {entry.timestamp}")

    # Verify integrity
    is_valid = ledger.verify_integrity()
    print(f"\nLedger integrity: {'VALID' if is_valid else 'INVALID'}")

    # Get statistics
    stats = ledger.get_stats()
    print(f"Total entries: {stats.total_entries}")
    print(f"First entry: {stats.first_entry_time}")
    print(f"Last entry: {stats.last_entry_time}")

if __name__ == "__main__":
    main()

Advanced Features

Batch Operations

from signledger.core.batch import BatchProcessor

# High-throughput batch processing
batch = BatchProcessor(ledger, batch_size=1000)

for i in range(10000):
    batch.add_entry({"transaction_id": f"TXN{i}", "amount": i * 100})

batch.flush()  # Commit remaining entries

Compression

from signledger.compression import Compressor

# Enable compression for storage optimization
ledger = Ledger(
    backend=SQLiteBackend("audit.db"),
    compression=Compressor(algorithm="zstd", level=3)
)

Merkle Tree Verification

from signledger.crypto.merkle import MerkleTree

# Build Merkle tree from entries
entries = list(ledger.get_entries())
merkle_tree = MerkleTree()

for entry in entries:
    merkle_tree.add_leaf(entry.hash)

merkle_tree.build()

# Get Merkle root
root_hash = merkle_tree.get_root()

# Verify specific entry
proof = merkle_tree.get_proof(0)
is_valid = merkle_tree.verify_proof(proof, entries[0].hash, root_hash)

Cryptographic Signatures

from signledger.crypto.signatures import RSASigner, ECDSASigner, Ed25519Signer

# RSA signatures
rsa_signer = RSASigner(key_size=2048)
entry = ledger.append(
    {"action": "critical_operation"},
    sign=True,
    signer=rsa_signer.sign
)

# ECDSA signatures (faster, smaller)
ecdsa_signer = ECDSASigner(curve="secp256k1")
entry = ledger.append(
    {"action": "blockchain_transaction"},
    sign=True,
    signer=ecdsa_signer.sign
)

# Ed25519 signatures (modern, fast)
ed_signer = Ed25519Signer()
entry = ledger.append(
    {"action": "secure_message"},
    sign=True,
    signer=ed_signer.sign
)

Storage Backends

SQLite (Default for Small-Medium Deployments)

from signledger.backends import SQLiteBackend

ledger = Ledger(backend=SQLiteBackend("audit.db"))

PostgreSQL (Recommended for Production)

from signledger.backends import PostgreSQLBackend

ledger = Ledger(
    backend=PostgreSQLBackend(
        host="localhost",
        port=5432,
        database="audit_log",
        user="postgres",
        password="secure_password"
    )
)

MongoDB (Document-Oriented)

from signledger.backends import MongoDBBackend

ledger = Ledger(
    backend=MongoDBBackend(
        host="localhost",
        port=27017,
        database="audit_db",
        collection="ledger_entries"
    )
)

In-Memory (Testing/Development)

from signledger.backends import InMemoryBackend

ledger = Ledger(backend=InMemoryBackend())

Performance

SignLedger is optimized for high-throughput audit logging:

• Write Performance: 2,000+ entries/second (batch mode: 10,000+/sec)
• Read Performance: 50,000+ entries/second with caching
• Verification: Full integrity check of 100,000 entries in ~2 seconds
• Storage Efficiency: 50-70% reduction with compression
• Memory Usage: ~1MB per 1,000 entries (without compression)

Security

• Immutability: Entries cannot be modified after creation (Pydantic frozen models)
• Hash Chain: Tamper detection through cryptographic linking
• Digital Signatures: Non-repudiation with RSA/ECDSA/Ed25519
• Integrity Verification: Automatic or on-demand verification
• Thread Safety: Production-ready concurrent access protection

Requirements

• Python 3.8+
• pydantic >= 2.0.0
• cryptography >= 41.0.0

License

MIT License

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Support

• GitHub Issues: https://github.com/vipin08/signledger/issues
• Documentation: https://github.com/vipin08/signledger

Changelog

Version 1.0.0 (Initial Release)

• Immutable audit logging with hash chain verification
• Multiple storage backends (SQLite, PostgreSQL, MongoDB, In-Memory)
• Cryptographic signatures (RSA, ECDSA, Ed25519)
• Framework integrations (Django, Flask, FastAPI)
• Batch operations for high-throughput scenarios
• Compression support (zlib, gzip, lz4, zstd)
• Thread-safe operations
• Full type hints and Pydantic v2 validation