sthakur369/tampertrail-server repository overview

# 1. Clone the repository
git clone https://github.com/sthakur369/TamperTrail.git
cd TamperTrail

# 2. Create your own local .env from the template
cp .env.example .env

# 3. Build and start (Ensure Port 80 is available)
docker compose up -d --build

┌─────────────────────────────────────────┐
│          TamperTrail Setup Wizard           │
│                                         │
│  Create your master admin password      │
│  to unlock the dashboard.               │
│                                         │
│  Password: ████████████████             │
│                                         │
│            [ Complete Setup ]           │
└─────────────────────────────────────────┘

export TAMPERTRAIL_API_KEY="vl_a1b2c3d4e5f6..."
export TAMPERTRAIL_URL="http://localhost"   # your TamperTrail instance

curl -X POST "$TAMPERTRAIL_URL/v1/log" \
  -H "X-API-Key: $TAMPERTRAIL_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "actor":       "user:[email protected]",
    "action":      "payment.success",
    "level":       "INFO",
    "message":     "Payment of $149.00 processed via Stripe for Pro plan upgrade.",
    "target_type": "invoice",
    "target_id":   "inv_9f2a3b4c",
    "status":      "success",
    "environment": "production",
    "source_ip":   "203.0.113.42",
    "request_id":  "req_abc123",
    "tags": {
      "payment_provider": "stripe",
      "amount_usd":       "149.00",
      "plan":             "pro"
    },
    "metadata": {
      "card_last4":    "4242",
      "stripe_charge": "ch_3abc123def",
      "billing_email": "[email protected]"
    }
  }'

// 202 Accepted
{ "status": "accepted", "message": "Log queued for processing" }

# tampertrail_logger.py — drop into your project, import everywhere
import os
import httpx
from typing import Optional

TAMPERTRAIL_URL = os.getenv("TAMPERTRAIL_URL", "http://localhost/v1/log")
TAMPERTRAIL_API_KEY = os.getenv("TAMPERTRAIL_API_KEY", "your-api-key-here")

HEADERS = {
    "X-API-Key": TAMPERTRAIL_API_KEY,
    "Content-Type": "application/json",
}

async def send_log(
    actor: str,                          # ✅ REQUIRED — who did it       (e.g. "user:[email protected]")
    action: str,                         # ✅ REQUIRED — what happened    (e.g. "order.created")
    level: Optional[str] = None,         # severity: DEBUG, INFO, WARN, ERROR, CRITICAL
    message: Optional[str] = None,       # human-readable event description
    target_type: Optional[str] = None,   # resource type  (e.g. "order", "invoice")
    target_id: Optional[str] = None,     # resource ID    (e.g. "ORD-1001")
    status: Optional[str] = None,        # outcome: "success", "failed", "200", etc.
    environment: Optional[str] = None,   # "production", "staging", "test"
    source_ip: Optional[str] = None,     # client IP address (auto-captured if omitted)
    request_id: Optional[str] = None,    # correlation ID — links related logs together
    tags: Optional[dict] = None,         # searchable key-value pairs (visible in dashboard)
    metadata: Optional[dict] = None,     # 🔒 ENCRYPTED at rest, NEVER shown in UI - (Ingest sensitive data in this field)
) -> None:
    """Send a log entry to TamperTrail. Fails silently — logging never crashes your app."""

    # Build payload — only include fields that have values
    payload = {"actor": actor, "action": action}

    optional = {
        "level": level, "message": message, "target_type": target_type,
        "target_id": target_id, "status": status, "environment": environment,
        "source_ip": source_ip, "request_id": request_id,
        "tags": tags, "metadata": metadata,
    }
    for key, value in optional.items():
        if value is not None:
            payload[key] = value

    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            await client.post(TAMPERTRAIL_URL, json=payload, headers=HEADERS)
    except Exception:
        pass  # logging should never crash your app

# YOUR route.py file
from tampertrail_logger import send_log

@app.post("/place-order")
def place_order(order: OrderCreate, request: Request, background_tasks: BackgroundTasks):
    db_order = create_order(db, order)

    # background_tasks runs AFTER response is sent → zero latency impact on your API
    background_tasks.add_task(
        send_log,
        actor=f"user:{order.user_id}",
        action="order.created",
        level="INFO",
        message=f"Order {order.order_id} — {order.order_name} worth ₹{order.price:,.0f}",
        target_type="order",
        target_id=order.order_id,
        status="success",
        environment="production",
        source_ip=request.client.host,
        request_id=request.state.request_id,
        tags={                                      # ← visible & searchable in dashboard
            "price": str(order.price),
            "origin": order.user_location,
            "destination": order.destination,
        },
        metadata={                                   # ← 🔒 encrypted, never shown in UI
            "user_id": order.user_id,
            "full_payload": order.model_dump(),
        },
    )
    return {"status": "created"}

# YOUR middleware.py file
import time, uuid, asyncio, platform, os, inspect
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi.responses import JSONResponse
from tampertrail_logger import send_log

class LoggingMiddleware(BaseHTTPMiddleware):
    SKIP_PATHS = {"/health", "/favicon.ico"}

    async def dispatch(self, request, call_next):
        request_id = str(uuid.uuid4())                     # → request_id
        request.state.request_id = request_id
        start = time.time()

        # Execute request (catch crashes → error tag)
        error_detail = None
        try:
            response = await call_next(request)
            status_code = response.status_code               # → status
        except Exception as e:
            status_code = 500
            error_detail = f"{type(e).__name__}: {str(e)}"   # → error
            response = JSONResponse(status_code=500, content={"detail": "Internal Server Error"})

        if request.url.path in self.SKIP_PATHS:
            response.headers["X-Request-ID"] = request_id
            return response

        # → actor (from X-User-ID header or fallback)
        user_id = request.headers.get("X-User-ID")
        actor = f"user:{user_id}" if user_id else "service:my-api"

        # → client_ip, client_port (from proxy headers or direct connection)
        client_ip = (request.headers.get("X-Forwarded-For", "").split(",")[0].strip()
                     or request.client.host)

        # → handler_file, handler_function, handler_line (introspection)
        handler_file = handler_function = handler_line = None
        try:
            endpoint = request.scope.get("endpoint")
            if endpoint:
                handler_file = os.path.basename(inspect.getfile(endpoint))
                handler_function = endpoint.__name__
                handler_line = str(inspect.getsourcelines(endpoint)[1])
        except Exception:
            pass

        route = request.scope.get("route")
        latency_ms = round((time.time() - start) * 1000, 2)  # → latency_ms
        level = "ERROR" if status_code >= 500 else ("WARN" if status_code >= 400 else "INFO")

        # ── Build tags (all visible & searchable in dashboard) ─────
        tags = {
            "method":       request.method,                                         # → HTTP method
            "path":         request.url.path,                                       # → endpoint path
            "full_url":     str(request.url),                                       # → complete URL
            "scheme":       request.url.scheme,                                     # → http or https
            "http_version": request.scope.get("http_version", ""),                  # → protocol version
            "latency_ms":   str(latency_ms),                                        # → response time
            "client_ip":    client_ip,                                              # → real client IP
            "client_port":  str(request.client.port) if request.client else "",     # → client port
            "user_agent":   request.headers.get("user-agent", ""),                  # → browser/SDK
            "host":         request.headers.get("host", ""),                        # → host header
            "language":     request.headers.get("accept-language", "").split(",")[0].strip(),  # → locale
            "server_hostname": platform.node(),                                     # → server name
            "server_os":       platform.system(),                                   # → OS
            "python_version":  platform.python_version(),                           # → runtime
            "server_pid":      str(os.getpid()),                                    # → process ID
        }

        # Optional fields (added only when data exists; spacing is only for readability)
        if request.url.query:                    tags["query_string"]        = str(request.url.query)
        if request.headers.get("content-type"):  tags["request_content_type"]= request.headers["content-type"]
        if request.headers.get("content-length"):tags["request_bytes"]       = request.headers["content-length"]
        if response.headers.get("content-type"): tags["response_content_type"]= response.headers["content-type"]
        if response.headers.get("content-length"):tags["response_bytes"]     = response.headers["content-length"]
        if request.headers.get("referer"):       tags["referer"]             = request.headers["referer"]
        if request.headers.get("origin"):        tags["origin"]              = request.headers["origin"]
        if request.headers.get("authorization"): tags["authenticated"]       = "true"  # presence only, never the token!
        if handler_file:                         tags["handler_file"]        = handler_file
        if handler_function:                     tags["handler_function"]    = handler_function
        if handler_line:                         tags["handler_line"]        = handler_line
        if getattr(route, "path", None):         tags["route_pattern"]       = route.path
        if error_detail:                         tags["error"]               = error_detail[:200]

        # ── Fire log to TamperTrail ──────────────────────────
        asyncio.create_task(send_log(
            actor=actor,
            action="http.request",
            level=level,
            message=f"{request.method} {request.url.path} → {status_code}",
            status=str(status_code),
            source_ip=client_ip,
            request_id=request_id,
            tags=tags,
        ))

        response.headers["X-Request-ID"] = request_id
        return response

# Register: app.add_middleware(LoggingMiddleware)

┌─────────────────────────────────────────────────────────┐
│                    Your Browser                         │
└───────────────────────┬─────────────────────────────────┘
                        │ Port 80 (only exposed port)
┌───────────────────────▼─────────────────────────────────┐
│              tampertrail-client (Nginx)                     │
│                                                         │
│  • Serves React dashboard (SPA)                         │
│  • Rate limiting per IP per endpoint                    │
│  • Security headers (CSP, HSTS, X-Frame-Options)        │
│  • Reverse proxies /v1/* → tampertrail-server               │
│  • FastAPI /docs blocked from public access             │
└───────────────────────┬─────────────────────────────────┘
                        │ Internal Docker network (port 8000)
┌───────────────────────▼─────────────────────────────────┐
│              tampertrail-server (FastAPI)                   │
│                                                         │
│  • Log ingestion with WAL crash recovery                │
│  • SHA-256 hash chain computation                       │
│  • Fernet metadata encryption                           │
│  • JWT authentication + Argon2 password hashing         │
│  • Multi-tenant RLS enforcement                         │
│  • Micro-batch async DB writer                          │
│  • Monthly partition management                         │
│  • Chain verification engine                            │
└───────────────────────┬─────────────────────────────────┘
                        │ Internal Docker network (port 5432)
┌───────────────────────▼─────────────────────────────────┐
│              tampertrail-db (PostgreSQL 16)                 │
│                                                         │
│  • audit_logs: monthly range-partitioned table          │
│  • encrypted_metadata: BYTEA (Fernet ciphertext)        │
│  • tags: JSONB with GIN index (fast search)             │
│  • Row-Level Security on all sensitive tables           │
│  • Chain checkpoints for retention-safe verification    │
└─────────────────────────────────────────────────────────┘

# Quick verification — stops at first error
curl -H "Authorization: Bearer $TOKEN" \
  "http://localhost/v1/verify"

# Deep scan — finds ALL tampered entries across the entire vault
curl -H "Authorization: Bearer $TOKEN" \
  "http://localhost/v1/verify/deep"

// All clear
{
  "valid": true,
  "checked": 10482,
  "message": "All 10,482 log entries verified successfully. Chain integrity intact."
}

// Tampering detected
{
  "valid": false,
  "checked": 4201,
  "first_error_at": "2026-01-15T14:22:03.491Z",
  "message": "Chain break detected: row prev_hash does not match previous row hash. Possible tampering before 2026-01-15T14:22:03Z."
}