Webhook Security

This guide covers secure authentication for the WebhookSink.

Authentication Modes

WebhookSink supports two authentication modes:

Mode	Header	Description
`hmac`	`X-Fapilog-Signature-256`	HMAC-SHA256 signature (recommended)
`header`	`X-Webhook-Secret`	Raw secret in header (deprecated)

HMAC Signature Mode (Default)

HMAC mode computes a signature of the payload using your secret key. The secret is never transmitted over the wire. This is the default mode as of v0.4.

Configuration

from fapilog.plugins.sinks.webhook import WebhookSink, WebhookSinkConfig

config = WebhookSinkConfig(
    endpoint="https://your-server.com/webhook",
    secret="your-secret-key",
    # signature_mode defaults to "hmac"
)
sink = WebhookSink(config=config)

Headers Sent

Header	Description
`X-Fapilog-Signature-256`	HMAC-SHA256 signature: `sha256=<hex-digest>`
`X-Fapilog-Timestamp`	Unix timestamp when request was signed

How It Works

Fapilog captures the current Unix timestamp
Serializes the payload as compact JSON (separators=(",", ":"))
Computes HMAC-SHA256(secret, "{timestamp}.{json_payload}")
Sends signature in X-Fapilog-Signature-256 and timestamp in X-Fapilog-Timestamp
Receiver verifies signature and rejects stale requests

Replay Protection

The timestamp in the signature prevents replay attacks. Receivers should:

Extract the timestamp from X-Fapilog-Timestamp
Reject requests where the timestamp is too old or too far in the future
Verify the signature includes the timestamp

A tolerance of 5 minutes (300 seconds) is recommended to account for clock skew.

Receiver-Side Verification

FastAPI Example

import hmac
import hashlib
import time
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()
WEBHOOK_SECRET = "your-secret-key"
TIMESTAMP_TOLERANCE = 300  # 5 minutes


def verify_webhook(
    payload: bytes,
    signature: str,
    timestamp_str: str,
    secret: str,
    tolerance_seconds: int = 300,
) -> None:
    """Verify Fapilog webhook signature and timestamp.

    Raises HTTPException if verification fails.
    """
    # Check timestamp freshness
    try:
        timestamp = int(timestamp_str)
    except (ValueError, TypeError):
        raise HTTPException(status_code=401, detail="Invalid timestamp")

    if abs(time.time() - timestamp) > tolerance_seconds:
        raise HTTPException(status_code=401, detail="Request too old or too new")

    # Verify signature includes timestamp
    if not signature.startswith("sha256="):
        raise HTTPException(status_code=401, detail="Invalid signature format")

    message = f"{timestamp}.{payload.decode()}".encode()
    expected = "sha256=" + hmac.new(
        secret.encode(),
        message,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(expected, signature):
        raise HTTPException(status_code=401, detail="Invalid signature")


@app.post("/webhook")
async def receive_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-Fapilog-Signature-256", "")
    timestamp = request.headers.get("X-Fapilog-Timestamp", "")

    verify_webhook(body, signature, timestamp, WEBHOOK_SECRET, TIMESTAMP_TOLERANCE)

    # Process the verified payload
    import json
    data = json.loads(body)
    return {"status": "received", "events": len(data) if isinstance(data, list) else 1}

Flask Example

import hmac
import hashlib
import time
from flask import Flask, request, abort

app = Flask(__name__)
WEBHOOK_SECRET = "your-secret-key"
TIMESTAMP_TOLERANCE = 300  # 5 minutes


def verify_webhook(
    payload: bytes,
    signature: str,
    timestamp_str: str,
    secret: str,
    tolerance_seconds: int = 300,
) -> bool:
    """Verify Fapilog webhook signature and timestamp."""
    try:
        timestamp = int(timestamp_str)
    except (ValueError, TypeError):
        return False

    if abs(time.time() - timestamp) > tolerance_seconds:
        return False

    if not signature.startswith("sha256="):
        return False

    message = f"{timestamp}.{payload.decode()}".encode()
    expected = "sha256=" + hmac.new(
        secret.encode(),
        message,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, signature)


@app.route("/webhook", methods=["POST"])
def receive_webhook():
    signature = request.headers.get("X-Fapilog-Signature-256", "")
    timestamp = request.headers.get("X-Fapilog-Timestamp", "")

    if not verify_webhook(request.data, signature, timestamp, WEBHOOK_SECRET):
        abort(401)
    return {"status": "received"}

Legacy Header Mode (Deprecated)

The legacy header mode sends the secret directly in the X-Webhook-Secret header. This mode is deprecated and will emit a warning.

Security Risks

Sending secrets in headers increases exposure via:

Proxy server logs (many log headers by default)
CDN/WAF request logging
Network monitoring tools
Accidental logging in receiving applications

Migration Path

Update your webhook receivers to verify HMAC signatures with timestamp
Handle the new X-Fapilog-Timestamp header and include it in signature verification
Add replay protection by rejecting stale requests
Remove legacy X-Webhook-Secret handling from receivers

Note: As of v0.4, HMAC is the default. The signature format changed from HMAC(payload) to HMAC(timestamp.payload) for replay protection.

Best Practices

Use HMAC mode for all new webhooks
Rotate secrets regularly and update both sender and receiver
Use constant-time comparison (hmac.compare_digest) to prevent timing attacks
Validate payload structure after signature verification
Log signature failures for security monitoring (without logging the secret)

Troubleshooting

Signature Mismatch

Common causes:

Missing timestamp in signature: The signature is computed over {timestamp}.{payload}, not just the payload. Ensure you include the timestamp from the X-Fapilog-Timestamp header.
Different JSON serialization: Fapilog uses compact JSON (separators=(",", ":")). Ensure your verification uses the raw request body, not re-serialized JSON.
Encoding issues: Ensure both sides use UTF-8 encoding for the secret.
Whitespace differences: The signature is computed on the exact bytes sent. Don’t strip or modify the payload before verification.

Testing Signatures

import hmac
import hashlib
import json
import time

secret = "test-secret"
payload = {"message": "hello", "level": "info"}
timestamp = int(time.time())

# Compute signature the same way Fapilog does
json_body = json.dumps(payload, separators=(",", ":"))
message = f"{timestamp}.{json_body}".encode()
signature = hmac.new(secret.encode(), message, hashlib.sha256).hexdigest()

print(f"X-Fapilog-Timestamp: {timestamp}")
print(f"X-Fapilog-Signature-256: sha256={signature}")

Clock Skew

If you see “Request too old or too new” errors:

Ensure both sender and receiver have synchronized clocks (use NTP)
Increase the tolerance if needed (default 300 seconds is generous)
Check for timezone issues in timestamp handling