Webhook: Add Fingerprinting to Webhook Payloads

backend/app/api/docs/credentials/create.md

@@ -6,6 +6,7 @@ Credentials are encrypted and stored securely for provider integrations (OpenAI,
 - **LLM:** openai, sarvamai, google(gemini)
 - **Observability:** langfuse
 - **Audio:** elevenlabs
+- **Miscellaneous** webhook_secret
 
 ### Examples:
 
@@ -40,7 +41,19 @@ Credentials are encrypted and stored securely for provider integrations (OpenAI,
       "public_key": "pk-lf-....",
       "secret_key": "sk-lf-...",
       "host": "https://cloud.langfuse.com"
-    }
+    },
+    "webhook_secret": {
+      "webhook_secret: "webhook_secret"
+    },
   }
 }
 ```
+#### For registering Webhook Secret
+```json
+{
+  "credential":{
+    "webhook_secret":"your-webhook-secret"
+  }
+
+}
+```

backend/app/api/docs/credentials/update.md

@@ -32,3 +32,4 @@ The `credential` field accepts **two formats** (both work the same):
 - **LLM:** openai, sarvamai, google(gemini)
 - **Observability:** langfuse
 - **Audio:** elevenlabs
+- **Miscellaneous** webhook_secret

backend/app/api/docs/credentials/update_by_org_project.md

@@ -36,3 +36,4 @@ The `credential` field accepts **two formats** (both work the same):
 - **LLM:** openai, sarvamai, google(gemini)
 - **Observability:** langfuse
 - **Audio:** elevenlabs
+- **Miscellaneous** webhook_secret

backend/app/core/providers.py

@@ -14,6 +14,7 @@ class Provider(str, Enum):
     GOOGLE = "google"
     SARVAMAI = "sarvamai"
     ELEVENLABS = "elevenlabs"
+    WEBHOOK_SECRET = "webhook_secret"
 
 
 @dataclass
@@ -42,6 +43,9 @@ class ProviderConfig:
     Provider.ELEVENLABS: ProviderConfig(
         required_fields=["api_key"], sensitive_fields=["api_key"]
     ),
+    Provider.WEBHOOK_SECRET: ProviderConfig(
+        required_fields=["webhook_secret"], sensitive_fields=["webhook_secret"]
+    ),
 }
 
 

backend/app/services/collections/create_collection.py

@@ -29,7 +29,7 @@
 )
 from app.services.collections.providers.registry import get_llm_provider
 from app.celery.utils import start_create_collection_job
-from app.utils import send_callback, APIResponse
+from app.utils import send_callback, get_webhook_secret, APIResponse
 
 
 logger = logging.getLogger(__name__)
@@ -274,7 +274,12 @@ def execute_job(
             )
 
             if creation_request.callback_url:
-                send_callback(creation_request.callback_url, success_payload)
+                webhook_secret = get_webhook_secret(project_id, organization_id)
+                send_callback(
+                    str(creation_request.callback_url),
+                    success_payload,
+                    webhook_secret=webhook_secret,
+                )
 
         except Exception as err:
             span.record_exception(err)
@@ -303,5 +308,10 @@ def execute_job(
 
             if creation_request and creation_request.callback_url and collection_job:
                 failure_payload = build_failure_payload(collection_job, str(err))
-                send_callback(creation_request.callback_url, failure_payload)
+                webhook_secret = get_webhook_secret(project_id, organization_id)
+                send_callback(
+                    str(creation_request.callback_url),
+                    failure_payload,
+                    webhook_secret=webhook_secret,
+                )
             raise

backend/app/services/collections/delete_collection.py

@@ -19,7 +19,7 @@
 from app.services.collections.providers.registry import get_llm_provider
 from app.celery.utils import start_delete_collection_job
 from app.core.telemetry import log_context
-from app.utils import send_callback, APIResponse
+from app.utils import send_callback, get_webhook_secret, APIResponse
 
 
 logger = logging.getLogger(__name__)
@@ -104,6 +104,7 @@ def build_failure_payload(
 
 def _mark_job_failed_and_callback(
     *,
+    organization_id: int,
     project_id: int,
     collection_id: UUID,
     job_id: UUID,
@@ -146,7 +147,8 @@ def _mark_job_failed_and_callback(
             collection_id=collection_id,
             error_message=str(err),
         )
-        send_callback(callback_url, failure_payload)
+        webhook_secret = get_webhook_secret(project_id, organization_id)
+        send_callback(callback_url, failure_payload, webhook_secret=webhook_secret)
 
 
 def execute_job(
@@ -162,8 +164,11 @@ def execute_job(
 
     deletion_request = DeletionRequest(**request)
 
-    collection_id = UUID(collection_id)
+    collection_uuid = UUID(collection_id)
     job_uuid = UUID(job_id)
+    callback_url = (
+        str(deletion_request.callback_url) if deletion_request.callback_url else None
+    )
 
     collection_job = None
 
@@ -172,12 +177,12 @@ def execute_job(
         lifecycle="collection.delete.execute_job",
         action="delete",
         collection_job_id=job_id,
-        collection_id=collection_id,
+        collection_id=str(collection_uuid),
         task_id=task_id,
         project_id=project_id,
         organization_id=organization_id,
     ), tracer.start_as_current_span("collections.delete.execute_job") as span:
-        span.set_attribute("collection.id", str(collection_id))
+        span.set_attribute("collection.id", str(collection_uuid))
         span.set_attribute("collection.job_id", str(job_uuid))
         span.set_attribute("kaapi.project_id", project_id)
         span.set_attribute("kaapi.organization_id", organization_id)
@@ -194,7 +199,9 @@ def execute_job(
                     ),
                 )
 
-                collection = CollectionCrud(session, project_id).read_one(collection_id)
+                collection = CollectionCrud(session, project_id).read_one(
+                    collection_uuid
+                )
                 span.set_attribute("collection.provider", str(collection.provider))
 
                 provider = get_llm_provider(
@@ -208,7 +215,7 @@ def execute_job(
                 provider.delete(collection)
 
             with Session(engine) as session:
-                CollectionCrud(session, project_id).delete_by_id(collection_id)
+                CollectionCrud(session, project_id).delete_by_id(collection_uuid)
 
                 collection_job_crud = CollectionJobCrud(session, project_id)
                 collection_job_crud.update(
@@ -223,24 +230,30 @@ def execute_job(
             logger.info(
                 "[delete_collection.execute_job] Collection deleted successfully | "
                 "{'collection_id': '%s', 'job_id': '%s'}",
-                str(collection_id),
+                str(collection_uuid),
                 str(job_uuid),
             )
-            if deletion_request.callback_url and collection_job:
+            if callback_url and collection_job:
                 success_payload = build_success_payload(
                     collection_job=collection_job,
-                    collection_id=collection_id,
+                    collection_id=collection_uuid,
+                )
+                webhook_secret = get_webhook_secret(project_id, organization_id)
+                send_callback(
+                    callback_url,
+                    success_payload,
+                    webhook_secret=webhook_secret,
                 )
-                send_callback(deletion_request.callback_url, success_payload)
 
         except Exception as err:
             span.record_exception(err)
             span.set_status(trace.Status(trace.StatusCode.ERROR, str(err)))
             _mark_job_failed_and_callback(
+                organization_id=organization_id,
                 project_id=project_id,
-                collection_id=collection_id,
+                collection_id=collection_uuid,
                 job_id=job_uuid,
                 err=err,
-                callback_url=deletion_request.callback_url,
+                callback_url=callback_url,
             )
             raise

backend/app/services/doctransform/job.py

@@ -20,10 +20,11 @@
     DocTransformationJobPublic,
     TransformedDocumentPublic,
     DocTransformationJob,
+    Project,
 )
 from app.core.cloud import get_cloud_storage
 from app.celery.utils import start_doctransform_job
-from app.utils import send_callback, APIResponse
+from app.utils import send_callback, get_webhook_secret, APIResponse
 from app.services.doctransform.registry import convert_document, FORMAT_TO_EXTENSION
 from app.core.db import engine
 
@@ -117,11 +118,18 @@ def execute_job(
     tmp_dir: Path | None = None
 
     job_for_payload = None  # keep latest job snapshot for payloads
+    webhook_secret: str | None = None
 
     try:
         job_uuid = UUID(job_id)
         source_uuid = UUID(source_document_id)
 
+        if callback_url:
+            with Session(engine) as db:
+                project = db.get(Project, project_id)
+            if project is not None:
+                webhook_secret = get_webhook_secret(project_id, project.organization_id)
+
         logger.info(
             "[doc_transform.execute_job] started | job_id=%s | transformer=%s | target=%s | project_id=%s",
             job_uuid,
@@ -222,7 +230,7 @@ def execute_job(
         )
 
         if callback_url:
-            send_callback(callback_url, success_payload)
+            send_callback(callback_url, success_payload, webhook_secret=webhook_secret)
 
     except Exception as e:
         logger.error(
@@ -251,7 +259,9 @@ def execute_job(
         if callback_url and job_for_payload:
             try:
                 failure_payload = build_failure_payload(job_for_payload, str(e))
-                send_callback(callback_url, failure_payload)
+                send_callback(
+                    callback_url, failure_payload, webhook_secret=webhook_secret
+                )
             except Exception as cb_error:
                 logger.error(
                     "[doc_transform.execute_job] callback failed | job_id=%s | error=%s",

backend/app/services/llm/chain/executor.py

@@ -13,7 +13,7 @@
 from app.models.llm.response import IntermediateChainResponse, LLMChainResponse
 from app.services.llm.chain.chain import ChainContext, LLMChain
 from app.services.llm.chain.types import BlockResult
-from app.utils import APIResponse, send_callback
+from app.utils import APIResponse, get_webhook_secret, send_callback
 
 logger = logging.getLogger(__name__)
 
@@ -31,6 +31,7 @@ def __init__(
         self._chain = chain
         self._context = context
         self._request = request
+        self._webhook_secret: str | None = None
 
     def run(self) -> dict:
         """Execute the full chain lifecycle. Returns serialized APIResponse."""
@@ -60,6 +61,10 @@ def _setup(self) -> None:
                 status=ChainStatus.RUNNING,
             )
 
+        self._webhook_secret = get_webhook_secret(
+            self._context.project_id, self._context.organization_id
+        )
+
     def _teardown(self, result: BlockResult) -> dict:
         """Finalize chain record, send callback, and update job status."""
 
@@ -76,6 +81,7 @@ def _teardown(self, result: BlockResult) -> dict:
                 send_callback(
                     callback_url=str(self._request.callback_url),
                     data=callback_response.model_dump(),
+                    webhook_secret=self._webhook_secret,
                 )
             with Session(engine) as session:
                 JobCrud(session).update(
@@ -107,6 +113,7 @@ def _handle_error(self, error: str) -> dict:
             send_callback(
                 callback_url=str(self._request.callback_url),
                 data=callback_response.model_dump(),
+                webhook_secret=self._webhook_secret,
             )
 
         with Session(engine) as session:
@@ -166,6 +173,7 @@ def _send_intermediate_callback(
             send_callback(
                 callback_url=str(self._request.callback_url),
                 data=callback_data.model_dump(),
+                webhook_secret=self._webhook_secret,
             )
             logger.info(
                 f"[_send_intermediate_callback] Sent intermediate callback | "