feat: HMAC-signed ingest completion webhooks (#14)

.env.example

@@ -105,6 +105,14 @@ INGEST_PUSH_MAX_LINES=5000
 TAIL_POLL_INTERVAL=30
 TAIL_ERROR_THRESHOLD=5
 
+# HMAC-signed ingest completion callbacks (optional callback_url on POST /v1/ingestions).
+# Per-key secrets (whsec_…) are minted with `raglogs keys create` and used when the
+# request was authenticated with that API key. WEBHOOK_SECRET is the fallback when
+# AUTH_ENABLED=false or the key has no stored signing secret (legacy keys).
+# WEBHOOK_SECRET=
+# WEBHOOK_MAX_RETRIES=5
+# WEBHOOK_TIMEOUT=10
+
 
 # ── Adapters ──────────────────────────────────────────────────────────────────
 # AWS region for the cloudwatch source adapter (raglogs ingest --adapter cloudwatch).

README.md

@@ -632,7 +632,7 @@ raglogs config llm_provider
 
 ### `raglogs keys`
 
-Mint, list, and revoke HTTP API keys. The plaintext secret is printed **once** at create time and is never stored or logged — only an argon2 hash and a short prefix are kept.
+Mint, list, and revoke HTTP API keys. The API bearer token is printed **once** at create time and is never stored or logged — only an argon2 hash and a short prefix are kept. A separate **webhook signing secret** (`whsec_…`) is also printed once; it is stored server-side so ingest completion callbacks can be HMAC-signed. It is not the bearer token. `raglogs keys list` shows `whsec_****` when a signing secret exists (legacy keys minted before this column fall back to `WEBHOOK_SECRET`).
 
 ```bash
 raglogs keys create --role query --scope default --name "ci"
@@ -685,6 +685,9 @@ All settings are read from `.env`, environment variables, or CLI flags. Priority
 | `OIDC_JWKS_URL` | _(empty)_ | Optional JWKS URL; default is `{issuer}/.well-known/openid-configuration` then `{issuer}/.well-known/jwks.json` |
 | `API_BIND_HOST` | `127.0.0.1` | Host the startup guard treats as the bind address. `make api` sets this to `0.0.0.0` to match uvicorn |
 | `AUTH_REFUSE_INSECURE_BIND` | `false` | If `true`, refuse to start when auth is off and the bind host is not loopback; if `false`, log a warning only |
+| `WEBHOOK_SECRET` | _(empty)_ | Fallback HMAC secret for ingest completion callbacks when auth is off or the API key has no per-key `whsec_` |
+| `WEBHOOK_MAX_RETRIES` | `5` | Extra webhook POST attempts after the first (6 POSTs by default) on 5xx / 429 / connect errors |
+| `WEBHOOK_TIMEOUT` | `10` | Per-attempt HTTP timeout in seconds for completion callbacks |
 
 ---
 
@@ -913,7 +916,7 @@ Auth is **off by default** so local demo and existing clients keep working. Set
 ```bash
 export AUTH_ENABLED=true
 raglogs keys create --role admin --name "local"
-# copy the rlk_… secret from the panel — it is shown only once
+# copy the rlk_… API key and the whsec_… webhook secret from the panels — each is shown only once
 
 curl -X POST http://localhost:8000/v1/query/explain \
   -H "Authorization: Bearer rlk_…" \
@@ -941,7 +944,7 @@ A valid key with the wrong role returns **403**:
 {"error_code": "AUTH_FORBIDDEN", "message": "…"}
 ```
 
-Keys are stored argon2-hashed with a short indexed prefix. `scope` defaults to `default` and is stored for later isolation work — this release does **not** filter clusters or logs by scope.
+Keys are stored argon2-hashed with a short indexed prefix. `scope` defaults to `default` and is stored for later isolation work — this release does **not** filter clusters or logs by scope. Each new key also gets a `whsec_…` webhook signing secret (shown once; `keys list` shows `whsec_****` only).
 
 Optional OIDC: set `AUTH_MODE=oidc` or `both` and `OIDC_ISSUER`. A JWT (three dotted segments) is validated via JWKS (`iss`, `exp`, and `aud` when `OIDC_AUDIENCE` is set). Role comes from claim `raglogs_role` or `roles`, defaulting to `query`. When `AUTH_MODE=api_key`, JWTs are rejected.
 
@@ -990,6 +993,49 @@ curl -X POST http://localhost:8000/v1/ingestions/$ID:stop
 
 **Backpressure.** When pending worker jobs ≥ `INGEST_QUEUE_MAX` (default 100), `POST /v1/ingestions` and `POST /v1/ingestions/lines` return **429** with `Retry-After` (`INGEST_RETRY_AFTER_SECONDS`, default 5) and body `{"error_code":"INGEST_QUEUE_FULL","message":"..."}`. This is a queue-depth stand-in, not full API/LLM rate limiting.
 
+**Completion callbacks.** Optional `callback_url` on `POST /v1/ingestions` (http or https only; `file:` and empty hosts are rejected). When a **batch** worker job reaches a terminal state (`done` / `failed`), raglogs POSTs an HMAC-SHA256-signed JSON body to that URL. Delivery is fail-open: retries with jittered exponential backoff (`WEBHOOK_MAX_RETRIES`, default 5 extra attempts) on 5xx, 429, and connect errors; 4xx other than 429 are not retried. Failures are logged and **do not** change ingest status — poll `GET /v1/ingestions/jobs/{worker_job_id}` still works.
+
+`job_id` in the payload is the **ingestion_job_id** when ingest created a row; if the worker failed before that, it is the `worker_job_id`. `scope` is the authenticated key's scope, or `"default"` when auth is off. `counts.clusters` is `0` (clustering is not part of ingest). Worker `done` maps to `"succeeded"` (or `"partial"` when `error_count > 0`); `failed` maps to `"failed"`.
+
+```bash
+curl -X POST http://localhost:8000/v1/ingestions \
+  -H "Content-Type: application/json" \
+  -d '{"paths":["/var/log/app"],"callback_url":"https://example.com/hooks/raglogs"}'
+```
+
+Example body:
+
+```json
+{
+  "job_id": "a1b2c3d4-…",
+  "status": "succeeded",
+  "scope": "default",
+  "counts": { "lines": 48213, "clusters": 0 },
+  "partial": false
+}
+```
+
+The signature is **not** inside the JSON. It is sent as:
+
+```
+X-Raglogs-Signature: sha256=<hex>
+```
+
+Verify by HMAC-SHA256 of the **raw request body** with the webhook secret, then compare to the header (constant-time). Use the per-key `whsec_…` from `raglogs keys create` when the ingest request was authenticated with that API key; otherwise `WEBHOOK_SECRET`. Do not HMAC the `rlk_…` bearer token — it is not stored.
+
+Python:
+
+```python
+import hmac, hashlib
+
+def verify(secret: str, body: bytes, header: str) -> bool:
+    digest = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
+    expected = "sha256=" + digest
+    return hmac.compare_digest(expected, header)
+```
+
+Tail jobs and `POST /v1/ingestions/lines` (sync push) do **not** fire callbacks in this release — tail jobs are long-lived (poll or `:stop`); push completes in the HTTP response.
+
 Overlapping tail poll windows can **double-count** the same physical lines until content dedup (G6) lands. Prefer a single tail job per source and avoid also batch-ingesting the same window.
 
 Unversioned `/ingestions`, `/query/*`, and `/config` remain as **deprecated aliases** for one release. They behave the same as the `/v1` paths and send `Deprecation: true` plus a `Link: </v1/...>; rel="successor-version"` header. `/health`, the web UI (`/`), and `/static` stay unversioned.
@@ -1124,7 +1170,7 @@ raglogs/
 │   │   ├── compare/         Window diffing — new, disappeared, increased, decreased
 │   │   ├── embeddings/      Provider abstraction + ingest persist helper
 │   │   ├── explain/         Evidence assembly, templates, confidence, summarizer
-│   │   ├── ingestion/       Ingestion orchestration and batch persistence
+│   │   ├── ingestion/       Ingestion orchestration, webhooks, batch persistence
 │   │   ├── llm/             Provider abstraction (OpenAI, Ollama, noop)
 │   │   ├── normalization/   Message normalization, fingerprinting, trigger patterns
 │   │   ├── parsing/         JSON and text parsers, field extractors, timestamps

clients/openapi.json

@@ -53,7 +53,7 @@
           "ingestions"
         ],
         "summary": "Create Ingestion",
-        "description": "Enqueue an ingest job. Returns immediately with worker_job_id.\nPoll GET /ingestions/jobs/{worker_job_id} for progress.\nWhen done, use ingestion_job_id from the result to scope /query/explain.\n\n``mode=tail`` creates a long-lived job that the worker polls; the response\nincludes ``ingestion_job_id`` and no worker_job_id.",
+        "description": "Enqueue an ingest job. Returns immediately with worker_job_id.\nPoll GET /ingestions/jobs/{worker_job_id} for progress.\nWhen done, use ingestion_job_id from the result to scope /query/explain.\n\n``mode=tail`` creates a long-lived job that the worker polls; the response\nincludes ``ingestion_job_id`` and no worker_job_id.\n\nOptional ``callback_url`` (http/https) is stored on batch worker jobs. The\nworker POSTs an HMAC-signed completion payload on terminal state. Tail jobs\ndo not fire callbacks (long-lived; poll or ``:stop`` instead).",
         "operationId": "v1_ingestions_create_ingestion__v1_ingestions",
         "requestBody": {
           "content": {
@@ -392,7 +392,7 @@
           "ingestions"
         ],
         "summary": "Create Ingestion",
-        "description": "Enqueue an ingest job. Returns immediately with worker_job_id.\nPoll GET /ingestions/jobs/{worker_job_id} for progress.\nWhen done, use ingestion_job_id from the result to scope /query/explain.\n\n``mode=tail`` creates a long-lived job that the worker polls; the response\nincludes ``ingestion_job_id`` and no worker_job_id.",
+        "description": "Enqueue an ingest job. Returns immediately with worker_job_id.\nPoll GET /ingestions/jobs/{worker_job_id} for progress.\nWhen done, use ingestion_job_id from the result to scope /query/explain.\n\n``mode=tail`` creates a long-lived job that the worker polls; the response\nincludes ``ingestion_job_id`` and no worker_job_id.\n\nOptional ``callback_url`` (http/https) is stored on batch worker jobs. The\nworker POSTs an HMAC-signed completion payload on terminal state. Tail jobs\ndo not fire callbacks (long-lived; poll or ``:stop`` instead).",
         "operationId": "legacy_ingestions_create_ingestion__ingestions",
         "requestBody": {
           "content": {
@@ -1758,6 +1758,17 @@
             ],
             "title": "Mode",
             "default": "batch"
+          },
+          "callback_url": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Callback Url"
           }
         },
         "type": "object",

migrations/versions/0006_api_key_webhook_secret.py

@@ -0,0 +1,28 @@
+"""per-key webhook signing secret on api_keys
+
+Revision ID: 0006_api_key_webhook_secret
+Revises: 0005_ingest_modes
+Create Date: 2026-08-17 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "0006_api_key_webhook_secret"
+down_revision: Union[str, None] = "0005_ingest_modes"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "api_keys",
+        sa.Column("webhook_secret", sa.Text(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("api_keys", "webhook_secret")

src/api/auth/keys.py

@@ -6,13 +6,14 @@
 import uuid
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Sequence
+from typing import Any, Sequence
 
 from argon2 import PasswordHasher
 from argon2.exceptions import InvalidHashError, VerifyMismatchError
 
 KEY_TOKEN_PREFIX = "rlk_"
 KEY_PREFIX_LEN = 12
+WEBHOOK_SECRET_PREFIX = "whsec_"
 VALID_ROLES = frozenset({"ingest", "query", "admin"})
 DEFAULT_SCOPE = "default"
 
@@ -21,7 +22,7 @@
 
 @dataclass(frozen=True)
 class ApiKeyInfo:
-    """Public key metadata — never includes the hash or plaintext."""
+    """Public key metadata — never includes the hash or plaintext secrets."""
 
     id: uuid.UUID
     name: str | None
@@ -30,6 +31,7 @@ class ApiKeyInfo:
     scope: str
     revoked_at: datetime | None
     created_at: datetime | None
+    webhook_secret_preview: str | None = None
 
 
 @dataclass(frozen=True)
@@ -51,6 +53,18 @@ def generate_api_key() -> str:
     return KEY_TOKEN_PREFIX + secrets.token_urlsafe(32)
 
 
+def generate_webhook_secret() -> str:
+    """Return a per-key HMAC secret (`whsec_` + url-safe). Store server-side."""
+    return WEBHOOK_SECRET_PREFIX + secrets.token_urlsafe(32)
+
+
+def mask_webhook_secret(secret: str | None) -> str | None:
+    """List/display form — prefix only, never the full signing secret."""
+    if not secret:
+        return None
+    return "whsec_****"
+
+
 def key_prefix(token: str) -> str:
     """Indexed lookup prefix — first 12 characters, not the full secret."""
     return token[:KEY_PREFIX_LEN]
@@ -69,7 +83,9 @@ def verify_api_key(token: str, key_hash: str) -> bool:
         return False
 
 
-def find_verified_key(token: str, candidates: Sequence[ApiKeyRecord]) -> ApiKeyRecord | None:
+def find_verified_key(
+    token: str, candidates: Sequence[ApiKeyRecord]
+) -> ApiKeyRecord | None:
     """Pick the non-revoked candidate whose prefix matches and whose hash verifies."""
     prefix = key_prefix(token)
     for candidate in candidates:
@@ -91,12 +107,16 @@ def lookup_api_key(token: str) -> ApiKeyRecord | None:
 
     prefix = key_prefix(token)
     with get_db() as db:
-        rows = db.execute(
-            select(ApiKey).where(
-                ApiKey.key_prefix == prefix,
-                ApiKey.revoked_at.is_(None),
+        rows = (
+            db.execute(
+                select(ApiKey).where(
+                    ApiKey.key_prefix == prefix,
+                    ApiKey.revoked_at.is_(None),
+                )
             )
-        ).scalars().all()
+            .scalars()
+            .all()
+        )
         records = [
             ApiKeyRecord(
                 id=row.id,
@@ -118,26 +138,49 @@ def create_api_key(
     role: str,
     scope: str = DEFAULT_SCOPE,
     name: str | None = None,
-) -> tuple[str, ApiKeyInfo]:
-    """Persist a hashed key and return (plaintext, metadata). Caller must not log plaintext."""
+) -> tuple[str, str, ApiKeyInfo]:
+    """Persist a hashed key and return (api_key, webhook_secret, metadata).
+
+    Caller must show both secrets once and must not log them. The webhook
+    secret is stored plaintext so completions can be HMAC-signed; the API
+    bearer token is argon2-hashed and cannot be used for HMAC.
+    """
     if role not in VALID_ROLES:
         raise ValueError(f"role must be one of {sorted(VALID_ROLES)}")
     if not scope:
         raise ValueError("scope must be a non-empty string")
 
     plaintext = generate_api_key()
+    webhook_secret = generate_webhook_secret()
     record = _persist_key(
         plaintext=plaintext,
+        webhook_secret=webhook_secret,
         role=role,
         scope=scope,
         name=name,
     )
-    return plaintext, record
+    return plaintext, webhook_secret, record
+
+
+def _key_info(row: Any) -> ApiKeyInfo:
+    return ApiKeyInfo(
+        id=row.id,
+        name=row.name,
+        key_prefix=row.key_prefix,
+        role=row.role,
+        scope=row.scope,
+        revoked_at=row.revoked_at,
+        created_at=row.created_at,
+        webhook_secret_preview=mask_webhook_secret(
+            getattr(row, "webhook_secret", None)
+        ),
+    )
 
 
 def _persist_key(
     *,
     plaintext: str,
+    webhook_secret: str,
     role: str,
     scope: str,
     name: str | None,
@@ -151,19 +194,12 @@ def _persist_key(
         key_hash=hash_api_key(plaintext),
         role=role,
         scope=scope,
+        webhook_secret=webhook_secret,
     )
     with get_db() as db:
         db.add(row)
         db.flush()
-        return ApiKeyInfo(
-            id=row.id,
-            name=row.name,
-            key_prefix=row.key_prefix,
-            role=row.role,
-            scope=row.scope,
-            revoked_at=row.revoked_at,
-            created_at=row.created_at,
-        )
+        return _key_info(row)
 
 
 def list_api_keys() -> list[ApiKeyInfo]:
@@ -174,19 +210,12 @@ def list_api_keys() -> list[ApiKeyInfo]:
     from src.db.session import get_db
 
     with get_db() as db:
-        rows = db.execute(select(ApiKey).order_by(ApiKey.created_at.desc())).scalars().all()
-        return [
-            ApiKeyInfo(
-                id=row.id,
-                name=row.name,
-                key_prefix=row.key_prefix,
-                role=row.role,
-                scope=row.scope,
-                revoked_at=row.revoked_at,
-                created_at=row.created_at,
-            )
-            for row in rows
-        ]
+        rows = (
+            db.execute(select(ApiKey).order_by(ApiKey.created_at.desc()))
+            .scalars()
+            .all()
+        )
+        return [_key_info(row) for row in rows]
 
 
 def revoke_api_key(key_id: uuid.UUID) -> ApiKeyInfo | None:
@@ -202,12 +231,4 @@ def revoke_api_key(key_id: uuid.UUID) -> ApiKeyInfo | None:
             return None
         if row.revoked_at is None:
             row.revoked_at = datetime.now(timezone.utc)
-        return ApiKeyInfo(
-            id=row.id,
-            name=row.name,
-            key_prefix=row.key_prefix,
-            role=row.role,
-            scope=row.scope,
-            revoked_at=row.revoked_at,
-            created_at=row.created_at,
-        )
+        return _key_info(row)