# NousResearch/hermes-agent · Webhook route filtering and denylist features

## Open now

- 1 issue, 0 PRs open on GitHub.
- #18041: feat(gateway): per-route sender and event-type denylists for webhook routes

These members concern adding payload filters, route scripts, and per-route sender/event-type denylists to gateway webhook routes. Two of the three entries appear to be duplicates of the same denylist feature.

## TL;DR

- 3 member(s): 1 open issue(s), 0 closed issue(s), 0 open PR(s), 1 merged PR(s), 1 PR(s) closed without merging.
- A fix was merged to main in PR #60944. Verify it addresses your case.
- **Recommended:** read PR #60944 (and any competing PRs below) before filing. Confirm the merged change covers your symptom.

## Maintainer resolution verdict

- Internal-tracker verdict over this cluster's issue members: `unresolved`. This is distinct from the fix status above. Weigh it yourself.

## How this cluster was grouped

- Link evidence (`STRUCTURAL_SPANNING`): every member is connected by structural evidence: shared changed files, or a closing/superseded reference.
- Pairwise link density `0.67`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked.

## Open related issues

- #18041: feat(gateway): per-route sender and event-type denylists for webhook routes

## Prior & competing PRs

- **#60944: feat(gateway): webhook payload filters and route scripts (salvage #57544)**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`
  - files: `cli-config.yaml.example`, `gateway/platforms/webhook.py`, `gateway/platforms/webhook_filters.py`, `hermes_cli/subcommands/webhook.py`, `hermes_cli/web_server.py`, `hermes_cli/webhook.py`, `scripts/release.py`, `tests/gateway/test_webhook_adapter.py`, `tests/hermes_cli/test_dashboard_admin_endpoints.py`, `tests/hermes_cli/test_webhook_cli.py`, `website/docs/guides/webhook-github-pr-review.md`, `website/docs/user-guide/messaging/webhooks.md`
  - fix hunks:
  - `cli-config.yaml.example`

    ```diff
    @@ -846,6 +846,11 @@ platform_toolsets:
     #         priority_mode: prepend
     #         priority:
     #           - my_plugin_command
    +#   webhook:
    +#     extra:
    +#       # Route scripts default to a 30 second timeout. Scripts must live under
    +#       # the active profile's scripts directory and receive webhook JSON on stdin.
    +#       script_timeout_seconds: 30
     #
     # Discord-specific settings (config.yaml top-level, not under platforms:):
     #
    ```
  - `gateway/platforms/webhook.py`

    ```diff
    @@ -58,6 +58,10 @@
         MessageType,
         SendResult,
     )
    +from gateway.platforms.webhook_filters import (
    +    DEFAULT_SCRIPT_TIMEOUT_SECONDS,
    +    WebhookRouteProcessor,
    +)
     
     logger = logging.getLogger(__name__)
     
    ```

    ```diff
    @@ -78,7 +82,6 @@
     _INSECURE_NO_AUTH = "INSECURE_NO_AUTH"
     _DYNAMIC_ROUTES_FILENAME = "webhook_subscriptions.json"
     _RATE_WINDOW_SECONDS = 60.0
    -
     # Hostnames/IP literals that only serve connections originating on the same
     # machine. Anything else is treated as a public bind for safety-rail purposes.
     _LOOPBACK_HOSTS = frozenset({
    ```

    ```diff
    @@ -155,6 +158,15 @@ def __init__(self, config: PlatformConfig):
             self._max_body_bytes: int = int(
                 config.extra.get("max_body_bytes", 1_048_576)
             )  # 1MB
    +        self._script_timeout_seconds: int = int(
    +            config.extra.get(
    +                "script_timeout_seconds",
    +                DEFAULT_SCRIPT_TIMEOUT_SECONDS,
    +            )
    +        )
    +        self._route_processor = WebhookRouteProcessor(
    +            script_timeout_seconds=self._script_timeout_seconds
    +        )
     
         # ------------------------------------------------------------------
         # Lifecycle
    ```

    ```diff
    @@ -571,6 +583,45 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
                     {"status": "ignored", "event": event_type}
                 )
     
    +        if not self._route_processor.route_filters_match(
    +            route_config, payload, event_type, request.headers
    +        ):
    +            logger.info(
    +                "[webhook] filtered event=%s route=%s",
    +                event_type,
    +                route_name,
    +            )
    +            return web.json_response(
    +                {
    +                    "status": "ignored",
    +                    "reason": "filter",
    +                    "route": route_name,
    +                }
    +            )
    +
    +        if route_config.get("script"):
    +            # run_route_script shells out (subprocess.run, up to its timeout);
    +            # run it in a worker thread so it can't block the gateway event loop.
    +            keep, transformed_payload = await asyncio.to_thread(
    +                self._route_processor.run_route_script,
    +                route_config.get("script"),
    +                payload,
    +            )
    +            if not keep:
    +                logger.info(
    +                    "[webhook] script ignored event=%s route=%s",
    +                    event_type,
    +                    route_name,
    +                )
    +                return web.json_response(
    +                    {
    +                        "status": "ignored",
    +                        "reason": "script",
    +                        "route": route_name,
    +                    }
    +                )
    +            payload = transformed_payload or payload
    +
             # Format prompt from template
             prompt_template = route_config.get("prompt", "")
             prompt = self._render_prompt(
    ```
  - `gateway/platforms/webhook_filters.py`

    ```diff
    @@ -0,0 +1,302 @@
    +"""Route-local filters and script transforms for the webhook adapter."""
    +
    +from __future__ import annotations
    +
    +import json
    +import logging
    +import os
    +import re
    +import shutil
    +import subprocess
    +import sys
    +from pathlib import Path
    +from typing import Any, Optional
    +
    +logger = logging.getLogger(__name__)
    +
    +DEFAULT_SCRIPT_TIMEOUT_SECONDS = 30
    +_MISSING = object()
    +
    +
    +def _stringify_filter_value(value: Any) -> str:
    +    if value is _MISSING:
    +        return ""
    +    if isinstance(value, (dict, list)):
    +        return json.dumps(value, sort_keys=True)
    +    return str(value)
    +
    +
    +def _resolve_profile_path(path_value: Any) -> Optional[Path]:
    +    """Resolve a user path, mapping ~/.hermes to the active profile home."""
    +    if not isinstance(path_value, str):
    +        return None
    +    raw = os.path.expandvars(path_value.strip())
    +    if not raw:
    +        return None
    +    from hermes_constants import get_hermes_home
    +
    +    hermes_home = get_hermes_home()
    +    if raw == "~/.hermes":
    +        return hermes_home
    +    if raw.startswith("~/.hermes/"):
    +        return hermes_home / raw.removeprefix("~/.hermes/")
    +    path = Path(raw).expanduser()
    +    if path.is_absolute():
    +        return path
    +    return hermes_home / path
    +
    +
    +def _resolve_script_path(script_value: Any) -> tuple[Optional[Path], Optional[str]]:
    +    """Resolve a route script under HERMES_HOME/scripts."""
    +    if not isinstance(script_value, str) or not script_value.strip():
    +        return None, "script path is empty"
    +    from hermes_constants import get_hermes_home
    +
    +    scripts_root = (get_hermes_home() / "scripts").resolve()
    +    raw_text = os.path.expandvars(script_value.strip())
    +    if raw_text == "~/.hermes" or raw_text.startswith("~/.hermes/"):
    +        mapped = _resolve_profile_path(raw_text)
    +        candidate = mapped.resolve() if mapped is not None else scripts_root
    +    else:
    +        raw = Path(raw_text).expanduser()
    +        candidate = raw.resolve() if raw.is_absolute() else (scripts_root / raw).resolve()
    +    try:
    +        candidate.relative_to(scripts_root)
    +    except ValueError:
    +        return None, f"script path resolves outside {scripts_root}"
    +    if not candidate.exists():
    +        return None, f"script not found: {candidate}"
    +    if not candidate.is_file():
    +        return None, f"script path is not a file: {candidate}"
    +    return candidate, None
    +
    +
    +def _load_filter_file_values(path_value: Any) -> list[Any]:
    +    path = _resolve_profile_path(path_value)
    +    if path is None:
    +        return []
    +    try:
    +        raw = path.read_text(encoding="utf-8")
    +    except OSError as exc:
    +        logger.warning("[webhook] filter in_file read failed for %s: %s", path, exc)
    +        return []
    +    try:
    +        data = json.loads(raw)
    +    except json.JSONDecodeError:
    +        return [line.strip() for line in raw.splitlines() if line.strip()]
    +    if isinstance(data, list):
    +        return data
    +    if isinstance(data, dict):
    +        return list(data.keys())
    +    return [data]
    +
    +
    +class WebhookRouteProcessor:
    +    """Evaluate declarative filters and optional script transforms."""
    +
    +    def __init__(
    +        self,
    +        *,
    +        script_timeout_seconds: int = DEFAULT_SCRIPT_TIMEOUT_SECONDS,
    +    ) -> None:
    +        self.script_timeout_seconds = max(1, int(script_timeout_seconds))
    +
    +    def resolve_filter_field(
    +        self,
    +        field: Any,
    +        payload: dict,
    +        event_type: str,
    +        headers: Any,
    +    ) -> Any:
    +        """Resolve a dotted filter field against payload/event/headers context."""
    +        if not isinstance(field, str) or not field.strip():
    +            return _MISSING
    +        parts = [part for part in field.strip().split(".") if part]
    +        if not parts:
    +            return _MISSING
    +        header_dict = dict(headers or {})
    +        context = {
    +            "payload": payload.get("payload", payload),
    +            "event": event_type,
    +            "event_type": event_type,
    +            "headers": header_dict,
    +        }
    +        if parts[0] in context:
    +            value: Any = context[parts[0]]
    +            parts = parts[1:]
    +        else:
    +            value = payload
    +        for part in parts:
    +            if isinstance(value, dict):
    +                value = value.get(part, _MISSING)
    +            elif isinstance(value, list) and part.isdigit():
    +                idx = int(part)
    +                value = value[idx] if 0 <= idx < len(value) else _MISSING
    +            else:
    +                return _MISSING
    +            if value is _MISSING:
    +                return _MISSING
    +        return value
    +
    +    def filter_matches(
    +        self,
    +        spec: Any,
    +        payload: dict,
    +        event_type: str,
    +        headers: Any,
    +    ) -> bool:
    +        """Evaluate one declarative webhook filter spec."""
    +        if not isinstance(spec, dict):
    +            logger.warning("[webhook] Ignoring invalid filter spec: %r", spec)
    +            return False
    +
    +        if "all" in spec:
    +            items = spec.get("all")
    +            return isinstance(items, list) and all(
    +                self.filter_matches(item, payload, event_type, headers)
    +                for item in items
    +            )
    +        if "any" in spec:
    +            items = spec.get("any")
    +            return isinstance(items, list) and any(
    +                self.filter_matches(item, payload, event_type, headers)
    +                for item in items
    +            )
    +        if "not" in spec:
    +            return not self.filter_matches(spec.get("not"), payload, event_type, headers)
    +
    +        value = self.resolve_filter_field(
    +            spec.get("field"), payload, event_type, headers
    +        )
    +
    +        if "exists" in spec:
    +            exists = value is not _MISSING
    +            return exists is bool(spec.get("exists"))
    +        if spec.get("missing") is True:
    +            return value is _MISSING
    +        if "equals" in spec:
    +            return value is not _MISSING and value == spec.get("equals")
    +        if "not_equals" in spec:
    +            return value is _MISSING or value != spec.get("not_equals")
    +        if "contains" in spec:
    +            needle = spec.get("contains")
    +            if value is _MISSING:
    +                return False
    +            if isinstance(value, (list, tuple, set, dict)):
    +                return needle in value
    +            return str(needle) in _stringify_filter_value(value)
    +        if "in" in spec:
    +            haystack = spec.get("in")
    +            return isinstance(haystack, list) and value in haystack
    +        if "in_file" in spec:
    +            return value in _load_filter_file_values(spec.get("in_file"))
    +        if "regex" in spec:
    +            if value is _MISSING:
    +                return False
    +            try:
    +                return (
    +                    re.search(str(spec.get("regex")), _stringify_filter_value(value))
    +                    is not None
    +                )
    +            except re.error as exc:
    +                logger.warning("[webhook] Invalid webhook filter regex: %s", exc)
    +                return False
    +
    +        logger.warning("[webhook] Filter spec has no supported operator: %r", spec)
    +        return False
    +
    +    def route_filters_match(
    +        self,
    +        route_config: dict,
    +        payload: dict,
    +        event_type: str,
    +        headers: Any,
    +    ) -> bool:
    +        filters = route_config.get("filters") or []
    +        if not filters:
    +            return True
    +        if isinstance(filters, dict):
    +            return self.filter_matches(filters, payload, event_type, headers)
    +        if not isinstance(filters, list):
    +            logger.warning("[webhook] filters must be a list or object")
    +            return False
    +        return all(
    +            self.filter_matches(spec, payload, event_type, headers)
    +            for spec in filters
    +        )
    +
    +    def run_route_script(self, script_value: Any, payload: dict) -> tuple[bool, Optional[dict]]:
    +        """Run a route script and return (should_continue, transformed_payload)."""
    +        path, error = _resolve_script_path(script_value)
    +        if error or path is None:
    +            logger.warning("[webhook] script ignored webhook: %s", error)
    +            return False, None
    +
    +        suffix = path.suffix.lower()
    +        if suffix in {".sh", ".bash"}:
    +            bash = shutil.which("bash") or (
    +                "/bin/bash" if os.path.isfile("/bin/bash") else None
    +            )
    +            if bash is None:
    +                logger.warning("[webhook] script ignored webhook: bash not found")
    +                return False, None
    +            argv = [bash, str(path)]
    +        else:
    +            argv = [sys.executable, str(path)]
    +
    +        try:
    +            from tools.environments.local import _sanitize_subprocess_env
    +
    +            popen_kwargs = {"creationflags": 0x08000000} if sys.platform == "win32" else {}
    +            result = subprocess.run(
    +                argv,
    +                input=json.dumps(payload),
    +                capture_output=True,
    +                text=True,
    +                timeout=self.script_timeout_seconds,
    +                cwd=str(path.parent),
    +                env=_sanitize_subprocess_env(os.environ.copy()),
    +                **popen_kwargs,
    +            )
    +        except subprocess.TimeoutExpired:
    +            logger.warning("[webhook] script timed out: %s", path)
    +            return False, None
    +        except Exception as exc:
    +            logger.warning("[webhook] script execution failed: %s", exc)
    +            return False, None
    +
    +        stdout = (result.stdout or "").strip()
    +        stderr = (result.stderr or "").strip()
    +        try:
    +            from agent.redact import redact_sensitive_text
    +
    +            stdout = redact_sensitive_text(stdout)
    +            stderr = redact_sensitive_text(stderr)
    +        except Exception as exc:
    +            logger.warning("[webhook] Failed to redact script output: %s", exc)
    +            stdout = "[REDACTED - redaction failed]"
    +            stderr = "[REDACTED - redaction failed]"
    +        if result.returncode != 0:
    +            logger.info(
    +                "[webhook] script ignored webhook path=%s code=%s stderr=%s",
    +                path.name,
    +                result.returncode,
    +                stderr[:200],
    +            )
    +            return False, None
    +        if not stdout or stdout == "[SILENT]":
    +            return False, None
    +
    +        try:
    +            transformed = json.loads(stdout)
    +        except json.JSONDecodeError:
    +            transformed = {**payload, "script_output": stdout}
    +        if not isinstance(transformed, dict):
    +            logger.warning("[webhook] script stdout must be a JSON object or text")
    +            return False, None
    +        if (
    +            transformed.get("[SILENT]") is True
    +            or transformed.get("__hermes_ignore__") is True
    +        ):
    +            return False, None
    +        return True, transformed
    ```
  - `hermes_cli/subcommands/webhook.py`

    ```diff
    @@ -55,6 +55,13 @@ def build_webhook_parser(subparsers, *, cmd_webhook: Callable) -> None:
             "message. Zero LLM cost. Requires --deliver to be a real target "
             "(not 'log').",
         )
    +    wh_sub.add_argument(
    +        "--script",
    +        default="",
    +        help="Filter/transform script under ~/.hermes/scripts/. The route "
    +        "payload is passed as JSON on stdin; empty stdout, [SILENT], or a "
    +        "nonzero exit code ignores the webhook.",
    +    )
     
         webhook_subparsers.add_parser(
             "list", aliases=["ls"], help="List all dynamic subscriptions"
    ```
  - `hermes_cli/web_server.py`

    ```diff
    @@ -10932,6 +10932,7 @@ class WebhookCreate(BaseModel):
         description: Optional[str] = None
         events: List[str] = []
         prompt: Optional[str] = None
    +    script: Optional[str] = None
         skills: List[str] = []
         deliver: str = "log"
         deliver_only: bool = False
    ```

    ```diff
    @@ -10948,6 +10949,7 @@ def _webhook_route_summary(name: str, route: Dict[str, Any], base_url: str) -> D
             "deliver": route.get("deliver", "log"),
             "deliver_only": bool(route.get("deliver_only")),
             "prompt": route.get("prompt", ""),
    +        "script": route.get("script", ""),
             "skills": list(route.get("skills") or []),
             "created_at": route.get("created_at"),
             "url": f"{base_url}/webhooks/{name}",
    ```

    ```diff
    @@ -11031,6 +11033,8 @@ async def create_webhook(body: WebhookCreate):
             "deliver": body.deliver or "log",
             "created_at": _time.strftime("%Y-%m-%dT%H:%M:%SZ", _time.gmtime()),
         }
    +    if body.script and body.script.strip():
    +        route["script"] = body.script.strip()
         if body.deliver_only:
             route["deliver_only"] = True
         if body.deliver_chat_id:
    ```
  - `hermes_cli/webhook.py`

    ```diff
    @@ -189,6 +189,10 @@ def _cmd_subscribe(args):
                 return
             route["deliver_only"] = True
     
    +    script = getattr(args, "script", "") or ""
    +    if script.strip():
    +        route["script"] = script.strip()
    +
         if args.deliver_chat_id:
             route["deliver_extra"] = {"chat_id": args.deliver_chat_id}
     
    ```

    ```diff
    @@ -212,6 +216,8 @@ def _cmd_subscribe(args):
             prompt_preview = route["prompt"][:80] + ("..." if len(route["prompt"]) > 80 else "")
             label = "Message" if route.get("deliver_only") else "Prompt"
             print(f"  {label}: {prompt_preview}")
    +    if route.get("script"):
    +        print(f"  Script: {route['script']}")
         print("\n  Configure your service to POST to the URL above.")
         print("  Use the secret for HMAC-SHA256 signature validation.")
         print("  The gateway must be running to receive events (hermes gateway run).\n")
    ```

    ```diff
    @@ -238,6 +244,8 @@ def _cmd_list(args):
             print(f"    URL:     {base_url}/webhooks/{name}")
             print(f"    Events:  {events}")
             print(f"    Deliver: {deliver}")
    +        if route.get("script"):
    +            print(f"    Script:  {route['script']}")
             print()
     
     
    ```
  - `scripts/release.py`

    ```diff
    @@ -45,6 +45,7 @@
     
     # Auto-extracted from noreply emails + manual overrides
     AUTHOR_MAP = {
    +    "[redacted:email]": "evelynburger",  # PR #57544 salvage (gateway: webhook payload filters + route scripts; commit under unlinked identity)
         "[redacted:email]": "SiteupAgencia",  # PR #57435 salvage (tui_gateway: back off notification poller when session is busy; #55578)
         "164521089+rainbowgits@users.noreply.github.com": "rainbowgore",  # PR #59405 salvage (mcp: bound stdio initialize handshake to stop subprocess/FD leak; #59349)
         "[redacted:email]": "thestudionorth",  # PR #60015 salvage (mcp: parent-death watchdog for stdio children; commit under unlinked local identity)
    ```
  - `tests/gateway/test_webhook_adapter.py`

    ```diff
    @@ -608,6 +608,285 @@ async def test_event_filter_accepts_payload_type_field(self):
                 assert resp.status == 202
     
     
    +# ===================================================================
    +# Payload filters
    +# ===================================================================
    +
    +
    +class TestPayloadFilters:
    +    """Tests for route-level payload filters in _handle_webhook."""
    +
    +    @pytest.mark.asyncio
    +    async def test_filter_rejects_before_agent_dispatch(self):
    +        """A non-matching filter returns ignored and never starts the agent."""
    +        routes = {
    +            "todoist": {
    +                "secret": [redacted:secret-assignment]
    +                "filters": [{"field": "payload.label", "equals": "urgent"}],
    +                "prompt": "Task: {payload.content}",
    +            }
    +        }
    +        adapter = _make_adapter(routes=routes)
    +        adapter.handle_message = AsyncMock()
    +
    +        app = _create_app(adapter)
    +        async with TestClient(TestServer(app)) as cli:
    +            resp = await cli.post(
    +                "/webhooks/todoist",
    +                json={"payload": {"label": "later", "content": "Buy milk"}},
    +                headers={"X-GitHub-Delivery": "filter-skip-1"},
    +            )
    +            assert resp.status == 200
    +            data = await resp.json()
    +            assert data == {
    +                "status": "ignored",
    +                "reason": "filter",
    +                "route": "todoist",
    +            }
    +
    +        adapter.handle_message.assert_not_called()
    +        assert "filter-skip-1" not in adapter._seen_deliveries
    +
    +    @pytest.mark.asyncio
    +    async def test_filter_accepts_nested_any_and_in_file(self, tmp_path, monkeypatch):
    +        """Nested any groups can match dynamic watchlists under HERMES_HOME."""
    +        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +        watchlist = tmp_path / "data" / "watchlist.json"
    +        watchlist.parent.mkdir()
    +        watchlist.write_text(json.dumps(["chat-1", "chat-2"]), encoding="utf-8")
    +        routes = {
    +            "waha": {
    +                "secret": [redacted:secret-assignment]
    +                "filters": [
    +                    {"field": "payload.fromMe", "equals": False},
    +                    {
    +                        "any": [
    +                            {
    +                                "field": "payload.chatId",
    +                                "in_file": "~/.hermes/data/watchlist.json",
    +                            },
    +                            {
    +                                "field": "payload.id.remote",
    +                                "in_file": "~/.hermes/data/watchlist.json",
    +                            },
    +                        ]
    +                    },
    +                ],
    +                "prompt": "Message from {payload.chatId}: {payload.body}",
    +            }
    +        }
    +        adapter = _make_adapter(routes=routes)
    +        captured = []
    +
    +        async def _capture(event):
    +            captured.append(event)
    +
    +        adapter.handle_message = _capture
    +
    +        app = _create_app(adapter)
    +        async with TestClient(TestServer(app)) as cli:
    +            resp = await cli.post(
    +                "/webhooks/waha",
    +                json={
    +                    "payload": {
    +                        "fromMe": False,
    +                        "chatId": "chat-2",
    +                        "body": "hello",
    +                    }
    +                },
    +                headers={"X-GitHub-Delivery": "filter-match-1"},
    +            )
    +            assert resp.status == 202
    +
    +        await asyncio.sleep(0.05)
    +        assert len(captured) == 1
    +        assert captured[0].text == "Message from chat-2: hello"
    +
    +    @pytest.mark.asyncio
    +    async def test_filter_applies_to_deliver_only_before_delivery(self):
    +        """Filtered direct-delivery routes skip target delivery too."""
    +        routes = {
    +            "alerts": {
    +                "secret": [redacted:secret-assignment]
    +                "deliver": "telegram",
    +                "deliver_only": True,
    +                "deliver_extra": {"chat_id": "123"},
    +                "filters": [{"field": "severity", "in": ["critical"]}],
    +                "prompt": "Alert: {message}",
    +            }
    +        }
    +        adapter = _make_adapter(routes=routes)
    +        mock_target = AsyncMock()
    +        mock_target.send = AsyncMock(return_value=SendResult(success=True))
    +        mock_runner = MagicMock()
    +        mock_runner.adapters = {Platform.TELEGRAM: mock_target}
    +        adapter.gateway_runner = mock_runner
    +
    +        app = _create_app(adapter)
    +        async with TestClient(TestServer(app)) as cli:
    +            resp = await cli.post(
    +                "/webhooks/alerts",
    +                json={"severity": "info", "message": "noise"},
    +                headers={"X-GitHub-Delivery": "filter-direct-1"},
    +            )
    +            assert resp.status == 200
    +            assert (await resp.json())["reason"] == "filter"
    +
    +        mock_target.send.assert_not_called()
    +
    +    @pytest.mark.asyncio
    +    async def test_script_transforms_payload_before_prompt_rendering(self, tmp_path, monkeypatch):
    +        """A script can replace the payload used by prompt templates."""
    +        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +        scripts = tmp_path / "scripts"
    +        scripts.mkdir()
    +        script = scripts / "todoist_filter.py"
    +        script.write_text(
    +            "import json, sys\n"
    +            "payload = json.load(sys.stdin)\n"
    +            "payload['body'] = payload['task']['content'].upper()\n"
    +            "print(json.dumps(payload))\n",
    +            encoding="utf-8",
    +        )
    +        routes = {
    +            "todoist": {
    +                "secret": [redacted:secret-assignment]
    +                "script": "todoist_filter.py",
    +                "prompt": "Task: {body}",
    +            }
    +        }
    +        adapter = _make_adapter(routes=routes)
    +        captured = []
    +
    +        async def _capture(event):
    +            captured.append(event)
    +
    +        adapter.handle_message = _capture
    +
    +        app = _create_app(adapter)
    +        async with TestClient(TestServer(app)) as cli:
    +            resp = await cli.post(
    +                "/webhooks/todoist",
    +                json={"task": {"content": "pay bills"}},
    +                headers={"X-GitHub-Delivery": "script-transform-1"},
    +            )
    +            assert resp.status == 202
    +
    +        await asyncio.sleep(0.05)
    +        assert captured[0].text == "Task: PAY BILLS"
    +        assert captured[0].raw_message["body"] == "PAY BILLS"
    +
    +    @pytest.mark.asyncio
    +    async def test_script_tilde_hermes_path_resolves_to_active_profile_home(self, tmp_path, monkeypatch):
    +        """~/.hermes/scripts paths must resolve through HERMES_HOME for profiles."""
    +        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +        scripts = tmp_path / "scripts"
    +        scripts.mkdir()
    +        (scripts / "todoist_filter.py").write_text(
    +            "import json, sys\n"
    +            "payload = json.load(sys.stdin)\n"
    +            "payload['body'] = 'profile-safe'\n"
    +            "print(json.dumps(payload))\n",
    +            encoding="utf-8",
    +        )
    +        routes = {
    +            "todoist": {
    +                "secret": [redacted:secret-assignment]
    +                "script": "~/.hermes/scripts/todoist_filter.py",
    +                "prompt": "Task: {body}",
    +            }
    +        }
    +        adapter = _make_adapter(routes=routes)
    +        captured = []
    +
    +        async def _capture(event):
    +            captured.append(event)
    +
    +        adapter.handle_message = _capture
    +
    +        app = _create_app(adapter)
    +        async with TestClient(TestServer(app)) as cli:
    +            resp = await cli.post(
    +                "/webhooks/todoist",
    +                json={"task": {"content": "pay bills"}},
    +                headers={"X-GitHub-Delivery": "script-profile-path-1"},
    +            )
    +            assert resp.status == 202
    +
    +        await asyncio.sleep(0.05)
    +        assert captured[0].text == "Task: profile-safe"
    +
    +    @pytest.mark.asyncio
    +    async def test_script_silent_stdout_ignores_without_idempotency_hit(self, tmp_path, monkeypatch):
    +        """Empty or [SILENT] script stdout filters the webhook out."""
    +        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +        scripts = tmp_path / "scripts"
    +        scripts.mkdir()
    +        (scripts / "skip.py").write_text("print('[SILENT]')\n", encoding="utf-8")
    +        routes = {
    +            "todoist": {
    +                "secret": [redacted:secret-assignment]
    +                "script": "skip.py",
    +                "prompt": "Task: {body}",
    +            }
    +        }
    +        adapter = _make_adapter(routes=routes)
    +        adapter.handle_message = AsyncMock()
    +
    +        app = _create_app(adapter)
    +        async with TestClient(TestServer(app)) as cli:
    +            resp = await cli.post(
    +                "/webhooks/todoist",
    +                json={"body": "ignore me"},
    +                headers={"X-GitHub-Delivery": "script-silent-1"},
    +            )
    +            assert resp.status == 200
    +            data = await resp.json()
    +            assert data == {
    +                "status": "ignored",
    +                "reason": "script",
    +                "route": "todoist",
    +            }
    +
    +        adapter.handle_message.assert_not_called()
    +        assert "script-silent-1" not in adapter._seen_deliveries
    +
    +    @pytest.mark.asyncio
    +    async def test_script_nonzero_exit_ignores_webhook(self, tmp_path, monkeypatch):
    +        """A script can fail closed by exiting nonzero."""
    +        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +        scripts = tmp_path / "scripts"
    +        scripts.mkdir()
    +        (scripts / "skip.py").write_text(
    +            "import sys\nsys.exit(2)\n",
    +            encoding="utf-8",
    +        )
    +        routes = {
    +            "todoist": {
    +                "secret": [redacted:secret-assignment]
    +                "script": "skip.py",
    +                "prompt": "Task: {body}",
    +            }
    +        }
    +        adapter = _make_adapter(routes=routes)
    +        adapter.handle_message = AsyncMock()
    +
    +        app = _create_app(adapter)
    +        async with TestClient(TestServer(app)) as cli:
    +            resp = await cli.post(
    +                "/webhooks/todoist",
    +                json={"body": "ignore me"},
    +                headers={"X-GitHub-Delivery": "script-nonzero-1"},
    +            )
    +            assert resp.status == 200
    +            data = await resp.json()
    +            assert data["status"] == "ignored"
    +            assert data["reason"] == "script"
    +
    +        adapter.handle_message.assert_not_called()
    +        assert "script-nonzero-1" not in adapter._seen_deliveries
    +
    +
     # ===================================================================
     # HTTP handling
     # ===================================================================
    ```

    ```diff
    @@ -1246,4 +1525,3 @@ async def test_connect_allows_real_secret_on_public_bind(self):
                 assert result is True
             finally:
                 await adapter.disconnect()
    -
    ```
  - `tests/hermes_cli/test_dashboard_admin_endpoints.py`

    ```diff
    @@ -223,6 +223,30 @@ def test_list_disabled_and_create_blocked(self):
             r = self.client.post("/api/webhooks", json={"name": "gh", "deliver": "log"})
             assert r.status_code == 400
     
    +    def test_create_webhook_persists_script(self):
    +        from hermes_cli.config import load_config, save_config
    +
    +        cfg = load_config()
    +        cfg.setdefault("platforms", {})["webhook"] = {
    +            "enabled": True,
    +            "extra": {"host": "0.0.0.0", "port": 8644},
    +        }
    +        save_config(cfg)
    +
    +        r = self.client.post(
    +            "/api/webhooks",
    +            json={
    +                "name": "todoist",
    +                "deliver": "log",
    +                "script": "todoist_filter.py",
    +            },
    +        )
    +        assert r.status_code == 200
    +        assert r.json()["script"] == "todoist_filter.py"
    +
    +        subs = self.client.get("/api/webhooks").json()["subscriptions"]
    +        assert subs[0]["script"] == "todoist_filter.py"
    +
         def test_enable_platform_starts_gateway_restart(self, monkeypatch):
             import hermes_cli.web_server as ws
             from hermes_cli.config import load_config
    ```
  - `tests/hermes_cli/test_webhook_cli.py`

    ```diff
    @@ -35,6 +35,7 @@ def _make_args(**kwargs):
             "deliver_chat_id": "",
             "secret": "",
             "payload": "",
    +        "script": "",
         }
         defaults.update(kwargs)
         return Namespace(**defaults)
    ```

    ```diff
    @@ -72,6 +73,12 @@ def test_custom_secret(self):
             ))
             assert _load_subscriptions()["s"]["secret"] == "my-secret"
     
    +    def test_script_option_is_persisted(self):
    +        webhook_command(_make_args(
    +            webhook_action="subscribe", name="s", script="todoist_filter.py"
    +        ))
    +        assert _load_subscriptions()["s"]["script"] == "todoist_filter.py"
    +
         def test_auto_secret(self):
             webhook_command(_make_args(webhook_action="subscribe", name="s"))
             secret = _load_subscriptions()["s"]["secret"]
    ```
  - `website/docs/guides/webhook-github-pr-review.md`

    ```diff
    @@ -182,12 +182,20 @@ tail -f "${HERMES_HOME:-$HOME/.hermes}/logs/gateway.log"
     
     ## Filtering to specific actions
     
    -GitHub sends `pull_request` events for many actions: `opened`, `synchronize`, `reopened`, `closed`, `labeled`, etc. The `events` list filters only by the `X-GitHub-Event` header value — it cannot filter by action sub-type at the routing level.
    +GitHub sends `pull_request` events for many actions: `opened`, `synchronize`, `reopened`, `closed`, `labeled`, etc. The `events` list filters by the `X-GitHub-Event` header value, and route-level `filters` can narrow by payload fields such as `action`.
     
     The prompt in Step 1 already handles this by instructing the agent to stop early for `closed` and `labeled` events.
     
     :::warning The agent still runs and consumes tokens
    -The "stop here" instruction prevents a meaningful review, but the agent still runs to completion for every `pull_request` event regardless of action. GitHub webhooks can only filter by event type (`pull_request`, `push`, `issues`, etc.) — not by action sub-type (`opened`, `closed`, `labeled`). There is no routing-level filter for sub-actions. For high-volume repos, accept this cost or filter upstream with a GitHub Actions workflow that calls your webhook URL conditionally.
    +The "stop here" instruction prevents a meaningful review, but the agent still runs to completion for every `pull_request` event regardless of action. Prefer filtering before the agent wakes:
    +
    +```yaml
    +filters:
    +  - field: "action"
    +    in: ["opened", "synchronize", "reopened"]
    +```
    +
    +For high-volume repositories, you can still filter upstream with a GitHub Actions workflow that calls your webhook URL conditionally.
     :::
     
     > There is no Jinja2 or conditional template syntax. `{field}` and `{nested.field}` are the only substitutions supported. Anything else is passed verbatim to the agent.
    ```
  - `website/docs/user-guide/messaging/webhooks.md`

    ```diff
    @@ -81,6 +81,8 @@ Routes define how different webhook sources are handled. Each route is a named e
     | `events` | No | List of event types to accept (e.g. `["pull_request"]`). If empty, all events are accepted. Event type is read from `X-GitHub-Event`, `X-GitLab-Event`, or `event_type` in the payload. |
     | `secret` | **Yes** | HMAC secret for signature validation. Falls back to the global `secret` if not set on the route. Set to `"INSECURE_NO_AUTH"` for testing only (skips validation). |
     | `prompt` | No | Template string with dot-notation payload access (e.g. `{pull_request.title}`). If omitted, the full JSON payload is dumped into the prompt. Payload fields are untrusted — see [Authenticated does not mean trusted](#authenticated-does-not-mean-trusted). |
    +| `filters` | No | Declarative payload filters evaluated after auth/body/event filtering and before agent or direct delivery work. Non-matches return `{"status":"ignored","reason":"filter"}` with HTTP 200. |
    +| `script` | No | Filter/transform script under `~/.hermes/scripts/`. The webhook payload is passed as JSON on stdin. JSON object stdout replaces the payload before templating; text stdout is exposed as `script_output`; empty stdout, `[SILENT]`, or a nonzero exit code ignores the webhook. |
     | `skills` | No | List of skill names to load for the agent run. |
     | `deliver` | No | Where to send the response: `github_comment`, `telegram`, `discord`, `slack`, `signal`, `sms`, `whatsapp`, `matrix`, `mattermost`, `homeassistant`, `email`, `dingtalk`, `feishu`, `wecom`, `weixin`, `bluebubbles`, `qqbot`, or `log` (default). |
     | `deliver_extra` | No | Additional delivery config — keys depend on `deliver` type (e.g. `repo`, `pr_number`, `chat_id`). Values support the same `{dot.notation}` templates as `prompt`. |
    ```

    ```diff
    @@ -116,9 +118,75 @@ platforms:
               events: ["push"]
               secret: "[redacted:secret-assignment]"
               prompt: "New push to {repository.full_name} branch {ref}: {head_commit.message}"
    +          filters:
    +            - field: "ref"
    +              equals: "refs/heads/main"
               deliver: "telegram"
     ```
     
    +### Payload Filters
    +
    +Use `filters` when a provider sends a broad event stream but only some payloads should wake the agent or trigger `deliver_only` delivery. Filters run after signature validation, body parsing, and `events`, but before prompt rendering, idempotency, agent dispatch, or direct delivery.
    +
    +```yaml
    +platforms:
    +  webhook:
    +    extra:
    +      routes:
    +        todoist:
    +          events: ["item:updated"]
    +          secret: "[redacted:secret-assignment]"
    +          filters:
    +            - field: "payload.labels"
    +              contains: "hermes"
    +            - any:
    +                - field: "payload.priority"
    +                  equals: 4
    +                - field: "payload.project_id"
    +                  in_file: "~/.hermes/data/todoist/watchlist.json"
    +          prompt: "Todoist task changed: {payload.content}"
    +```
    +
    +Supported operators:
    +
    +- `exists: true|false`
    +- `missing: true`
    +- `equals` / `not_equals`
    +- `contains` for strings, lists, and dict keys
    +- `in` for inline lists
    +- `in_file` for JSON arrays, JSON objects (keys are used), or newline-delimited text files
    +- `regex`
    +- `all`, `any`, and `not` groups
    +
    +Field paths use dot notation. `payload.foo` reads from a top-level `payload` object when one exists, or from the root webhook body for flat payloads. `event` / `event_type` match the resolved event type, and `headers.<Name>` reads request headers.
    +
    +### Script Filters and Transforms
    +
    +Use `script` when declarative filters are not enough. Scripts must live under `~/.hermes/scripts/` for the active profile; relative paths resolve there, and path traversal outside that directory is blocked. `.sh` and `.bash` scripts run with bash, and all other extensions run with the current Python interpreter.
    +
    +The route payload is sent to stdin as JSON:
    +
    +```python
    +# ~/.hermes/scripts/todoist-hermes-label.py
    +import json
    +import sys
    +
    +payload = json.load(sys.stdin)
    +labels = payload.get("payload", {}).get("labels", [])
    +if "hermes" not in labels:
    +    print("[SILENT]")
    +    raise SystemExit(0)
    +
    +payload["body"] = payload["payload"]["content"]
    +print(json.dumps(payload))
    +```
    +
    +Script outcomes:
    +
    +- JSON object stdout replaces the payload used by `prompt` and `deliver_extra`.
    +- Non-JSON text stdout is added to the payload as `script_output`.
    +- Empty stdout, exact `[SILENT]`, `{"__hermes_ignore__": true}`, timeout, missing script, or nonzero exit code returns HTTP 200 with `{"status":"ignored","reason":"script"}`.
    +
     ### Prompt Templates
     
     Prompts use dot-notation to access nested fields in the webhook payload:
    ```
- **#18042: feat(gateway): per-route sender and event-type denylists for webhook routes**. closed without merge
  - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `#60944`, `#57544`

## Fix status

- Merged to main in PR #60944 (cited above). No release data is tracked. Verify downstream.
- Fix last verified (merged) `2026-07-08`.
- Subsystem: `gateway`