fix: authorize webhook trigger targets

backend/app/api/webhooks.py

@@ -12,16 +12,15 @@
 from fastapi import APIRouter, Request
 from fastapi.responses import JSONResponse
 from loguru import logger
-from sqlalchemy import select
+from sqlalchemy.exc import SQLAlchemyError
 
-from app.dao import query_dao
-async_session = query_dao.session
 from app.core.events import get_redis
-from app.models.agent import Agent
+from app.dao import query_dao, trigger_dao
 from app.models.audit import AuditLog
-from app.models.trigger import AgentTrigger
 from app.services.trigger_runtime import enqueue_webhook_execution
 
+async_session = query_dao.session
+
 router = APIRouter(prefix="/api/webhooks", tags=["webhooks"])
 
 RATE_LIMIT = 5  # max hits per minute per token
@@ -71,38 +70,13 @@ async def receive_webhook(token: str, request: Request):
 
     # Look up trigger
     async with async_session() as db:
-        result = await query_dao.execute(db, 
-            select(AgentTrigger).where(
-                AgentTrigger.type == "webhook",
-                AgentTrigger.is_enabled,
-            )
-        )
-        triggers = result.scalars().all()
-
-        # Find the trigger matching this token
-        target = None
-        for trigger in triggers:
-            cfg = trigger.config or {}
-            if cfg.get("token") == token:
-                target = trigger
-                break
-
-        if not target:
+        target_result = await trigger_dao.get_enabled_webhook_target(token, db=db)
+        if target_result is None:
             # Return 200 OK to avoid leaking whether the token exists
             return JSONResponse({"ok": True})
 
-        # Per-agent rate limit check
-        agent_result = await query_dao.execute(
-            db,
-            select(Agent).where(
-                Agent.id == target.agent_id,
-                Agent.deleted_at.is_(None),
-            )
-        )
-        agent_obj = agent_result.scalar_one_or_none()
-        if agent_obj is None:
-            return JSONResponse({"ok": True})
-        agent_rate_limit = (agent_obj.webhook_rate_limit if agent_obj else None) or RATE_LIMIT
+        target, agent_obj = target_result
+        agent_rate_limit = agent_obj.webhook_rate_limit or RATE_LIMIT
 
         # Retrieve all needed scalar fields and expunge from db session to prevent MissingGreenlet errors.
         target_name = target.name
@@ -129,8 +103,8 @@ async def receive_webhook(token: str, request: Request):
                     )
                 )
                 await query_dao.commit(db)
-            except Exception:
-                pass
+            except SQLAlchemyError:
+                logger.exception("Failed to record rate-limited webhook audit log")
             return JSONResponse({"ok": True}, status_code=429)
 
         # HMAC signature verification (optional)
@@ -153,7 +127,7 @@ async def receive_webhook(token: str, request: Request):
                 payload_str = json.dumps(payload_obj, ensure_ascii=False, indent=2)
             except json.JSONDecodeError:
                 payload_obj = None
-        except Exception:
+        except (UnicodeDecodeError, ValueError):
             payload_obj = None
             payload_str = repr(body[:2000])
 

backend/app/dao/__init__.py

@@ -18,6 +18,7 @@
 from app.dao.query_dao import query_dao
 from app.dao.system_setting_dao import system_setting_dao
 from app.dao.tenant_dao import tenant_dao
+from app.dao.trigger_dao import trigger_dao
 from app.dao.user_dao import user_dao
 
 __all__ = [
@@ -41,6 +42,7 @@
     "system_setting_dao",
     "tenant_context",
     "tenant_dao",
+    "trigger_dao",
     "TenantScopedBaseDAO",
     "user_dao",
 ]

backend/app/dao/trigger_dao.py

@@ -0,0 +1,40 @@
+"""Read access for AgentTrigger records used by public trigger endpoints."""
+
+from typing import Any
+
+from sqlalchemy import select
+
+from app.dao.base import BaseDAO
+from app.models.agent import Agent
+from app.models.tenant import Tenant
+from app.models.trigger import AgentTrigger
+
+
+class TriggerDAO(BaseDAO[AgentTrigger]):
+    """DAO for trigger lookups that do not have a request tenant context."""
+
+    def __init__(self) -> None:
+        super().__init__(AgentTrigger)
+
+    async def get_enabled_webhook_target(
+        self, token: str, db: Any = None
+    ) -> tuple[AgentTrigger, Agent] | None:
+        """Return a token-matched webhook and its active agent in an active tenant."""
+        async with self.session(db=db, readonly=True) as session_db:
+            stmt = (
+                select(AgentTrigger, Agent)
+                .join(Agent, Agent.id == AgentTrigger.agent_id)
+                .join(Tenant, Tenant.id == Agent.tenant_id)
+                .where(
+                    AgentTrigger.type == "webhook",
+                    AgentTrigger.is_enabled.is_(True),
+                    AgentTrigger.config["token"].astext == token,
+                    Agent.deleted_at.is_(None),
+                    Tenant.is_active.is_(True),
+                )
+                .limit(1)
+            )
+            return (await session_db.execute(stmt)).one_or_none()
+
+
+trigger_dao = TriggerDAO()

backend/tests/test_webhooks_api.py

@@ -1,7 +1,8 @@
 import uuid
-import pytest
 from types import SimpleNamespace
+
 import httpx
+import pytest
 
 from app.api import webhooks as webhooks_api
 from app.main import app
@@ -29,14 +30,6 @@ def __init__(self, triggers=None, agent=None):
         self.committed = False
         self.expunged = []
 
-    async def execute(self, statement):
-        stmt_str = str(statement)
-        if "agent_triggers" in stmt_str:
-            return FakeScalarResult(self.triggers)
-        elif "agents" in stmt_str:
-            return FakeScalarResult(self.agent)
-        return FakeScalarResult(None)
-
     def add(self, value):
         self.added.append(value)
 
@@ -90,6 +83,13 @@ async def test_receive_webhook_success(monkeypatch, client):
     # Mock dependencies and DB session
     monkeypatch.setattr(webhooks_api, "async_session", FakeAsyncSessionFactory(session))
 
+    async def fake_get_enabled_webhook_target(token, db):
+        assert token == "valid_token"
+        assert db is session
+        return trigger, agent
+
+    monkeypatch.setattr(webhooks_api.trigger_dao, "get_enabled_webhook_target", fake_get_enabled_webhook_target)
+
     # Mock redis rate limiting
     async def fake_record_and_count_hits(token):
         return 1
@@ -126,6 +126,13 @@ async def test_receive_webhook_reports_runtime_intake_failure(monkeypatch, clien
     session = FakeSession(triggers=[trigger], agent=agent)
     monkeypatch.setattr(webhooks_api, "async_session", FakeAsyncSessionFactory(session))
 
+    async def fake_get_enabled_webhook_target(token, db):
+        assert token == "valid_token"
+        assert db is session
+        return trigger, agent
+
+    monkeypatch.setattr(webhooks_api.trigger_dao, "get_enabled_webhook_target", fake_get_enabled_webhook_target)
+
     async def fake_record_and_count_hits(_token):
         return 1
 
@@ -144,3 +151,29 @@ async def reject_runtime(*_args, **_kwargs):
 
     assert response.status_code == 503
     assert response.json() == {"ok": False, "error": "runtime_unavailable"}
+
+
+@pytest.mark.asyncio
+async def test_receive_webhook_ignores_token_without_authorized_agent(monkeypatch, client):
+    session = FakeSession()
+    monkeypatch.setattr(webhooks_api, "async_session", FakeAsyncSessionFactory(session))
+
+    async def fake_record_and_count_hits(_token):
+        return 1
+
+    async def no_authorized_target(token, db):
+        assert token == "valid_token"
+        assert db is session
+
+    async def fail_if_enqueued(*_args, **_kwargs):
+        pytest.fail("an unauthorized webhook target must not be enqueued")
+
+    monkeypatch.setattr(webhooks_api, "_record_and_count_hits", fake_record_and_count_hits)
+    monkeypatch.setattr(webhooks_api.trigger_dao, "get_enabled_webhook_target", no_authorized_target)
+    monkeypatch.setattr(webhooks_api, "enqueue_webhook_execution", fail_if_enqueued)
+
+    async with await client() as ac:
+        response = await ac.post("/api/webhooks/t/valid_token", json={"event": "test"})
+
+    assert response.status_code == 200
+    assert response.json() == {"ok": True}