From 957e35cc26c6f7fed25b9c8c7d284bb1c43b66e3 Mon Sep 17 00:00:00 2001 From: yaojin Date: Thu, 6 Aug 2026 14:32:13 +0800 Subject: [PATCH] fix: authorize webhook trigger targets --- backend/app/api/webhooks.py | 48 +++++++--------------------- backend/app/dao/__init__.py | 2 ++ backend/app/dao/trigger_dao.py | 40 +++++++++++++++++++++++ backend/tests/test_webhooks_api.py | 51 ++++++++++++++++++++++++------ 4 files changed, 95 insertions(+), 46 deletions(-) create mode 100644 backend/app/dao/trigger_dao.py diff --git a/backend/app/api/webhooks.py b/backend/app/api/webhooks.py index bb8094f80..c65e9cbbf 100644 --- a/backend/app/api/webhooks.py +++ b/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]) diff --git a/backend/app/dao/__init__.py b/backend/app/dao/__init__.py index 586475c35..a2e314359 100644 --- a/backend/app/dao/__init__.py +++ b/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", ] diff --git a/backend/app/dao/trigger_dao.py b/backend/app/dao/trigger_dao.py new file mode 100644 index 000000000..de24d9367 --- /dev/null +++ b/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() diff --git a/backend/tests/test_webhooks_api.py b/backend/tests/test_webhooks_api.py index e5961604f..29020ab5e 100644 --- a/backend/tests/test_webhooks_api.py +++ b/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}