From ca7a16c8bc60b7cf3a566c0adfe2c9ef988fc7d7 Mon Sep 17 00:00:00 2001 From: Melvin Strobl Date: Sat, 15 Aug 2026 21:19:32 +0200 Subject: [PATCH] Guard webhooks with a per-hook secret, and allow rotating the app key Trigger hooks are mounted unauthenticated because devices cannot present a JWT. They now take a secret as a trailing path segment, so a device needs one URL and no header support. The value never enters the registered route, only a {secret} template, and is compared with compare_digest; a mismatch is a bare 404 so the endpoint does not confirm which hooks exist. Pointing the parameter at the encrypted store keeps the literal out of flow.json. Hooks without a secret keep working and now raise a validation issue saying so. Rotating SECRET_KEY made the stored secrets unreadable for good, since the Fernet key derives from it. scripts/rotate_secret_key.py re-encrypts with the new key and refuses if the old one does not decrypt. Now that recovery exists, an unreadable store fails loudly instead of coming back empty and leaving flows short of credentials with no visible cause. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01KkmeRiyeYmVZqJVwuyHq9o --- backend/app/flow/controller.py | 18 +++ backend/app/flow/nodes.py | 42 +++++-- backend/app/flow/secrets.py | 23 ++-- backend/scripts/rotate_secret_key.py | 42 +++++++ backend/tests/flow/test_http_hook_secret.py | 118 ++++++++++++++++++++ 5 files changed, 226 insertions(+), 17 deletions(-) create mode 100644 backend/scripts/rotate_secret_key.py create mode 100644 backend/tests/flow/test_http_hook_secret.py diff --git a/backend/app/flow/controller.py b/backend/app/flow/controller.py index 12a457e..74410d9 100644 --- a/backend/app/flow/controller.py +++ b/backend/app/flow/controller.py @@ -236,6 +236,24 @@ class FlowController: for entry in loaded.values() if entry.status is NodeStatus.ERROR ] + # Webhooks are mounted without authentication, so a hook without a + # secret is open to anyone who guesses its URL. Flows written before + # the parameter existed keep running, but say so. + self.issues += [ + ValidationIssue( + code="unauthenticated_hook", + message=( + f"Webhook '{entry.node.local_id}' has no secret — " + "anyone who knows its URL can trigger it." + ), + flow=entry.flow, + node=entry.id, + ) + for entry in loaded.values() + if isinstance(entry.node, HttpNode) + and entry.node.mode == HttpNode.Mode.TRIGGER + and not entry.node.secret + ] await self._activate() diff --git a/backend/app/flow/nodes.py b/backend/app/flow/nodes.py index 70a40f7..5d91d0e 100644 --- a/backend/app/flow/nodes.py +++ b/backend/app/flow/nodes.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import hmac import logging import threading import time @@ -10,7 +11,7 @@ from typing import TYPE_CHECKING, Any, Literal import httpx import numpy as np -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel, ConfigDict, Field from app.flow.messages import MessageSpec, qualify @@ -91,8 +92,8 @@ class Node: self.output_ports = self._normalize_ports(provides) self.flow = "" - self.name = name or getattr(f, "__name__", "node") - self.id = self.name + self.name: str = name or getattr(f, "__name__", "node") + self.id: str = self.name self._index_ports() @staticmethod @@ -106,8 +107,12 @@ class Node: def _index_ports(self) -> None: """Index bound ports by message name; unbound ports have no wiring.""" - self.requires = {s.name: s for s in self.input_ports if s.name} - self.provides = {s.name: s for s in self.output_ports if s.name} + self.requires: dict[str, MessageSpec] = { + s.name: s for s in self.input_ports if s.name + } + self.provides: dict[str, MessageSpec] = { + s.name: s for s in self.output_ports if s.name + } def assign_flow(self, flow: str, node_id: str) -> None: """Place this node in a flow, qualifying its identity and messages. @@ -364,6 +369,8 @@ class HttpNode(Node): :type timeout: float :param headers: Additional HTTP headers. :type headers: dict[str, str] | None + :param secret: Shared secret callers append to the webhook URL (trigger mode). + :type secret: str :raises ValueError: If both ``requires`` and ``provides`` are empty, or if the configuration is invalid. @@ -400,6 +407,7 @@ class HttpNode(Node): "mode", "timeout", "headers", + "secret", "_route_registered", "_http_client", ) @@ -411,6 +419,14 @@ class HttpNode(Node): method: Literal["GET", "POST"] = "POST" timeout: float = 30.0 headers: dict[str, str] = {} + secret: str = Field( + default="", + description=( + "Shared secret for a webhook, appended to its URL: " + "/hooks///. Empty leaves the webhook open " + 'to anyone. Use {"$secret": "name"} to read it from the store.' + ), + ) def __init__( self, @@ -442,6 +458,7 @@ class HttpNode(Node): self.method = cfg.method.upper() self.timeout = cfg.timeout self.headers = cfg.headers + self.secret = cfg.secret self._route_registered = False self._http_client: httpx.AsyncClient | None = None @@ -588,6 +605,13 @@ class HttpNode(Node): :returns: JSON response with trigger result. :rtype: JSONResponse """ + if self.secret and not hmac.compare_digest( + str(request.path_params.get("secret", "")).encode(), + self.secret.encode(), + ): + # A wrong secret must look like no such hook at all. + return JSONResponse(content={"detail": "Not Found"}, status_code=404) + try: # Parse request data if self.method == "GET": @@ -631,9 +655,13 @@ class HttpNode(Node): status_code=500, ) + # The secret is a path parameter, not part of the registered path, so + # neither the route table nor the logs below carry its value. + path = f"{self.url}/{{secret}}" if self.secret else self.url + # Create a Starlette Route and add it directly to the app's routes route = Route( - self.url, + path, handle_request, methods=[self.method], name=self.id, @@ -645,7 +673,7 @@ class HttpNode(Node): "Registered %s route for node '%s': %s", self.method, self.name, - self.url, + path, ) def unregister_route(self, app: FastAPI) -> None: diff --git a/backend/app/flow/secrets.py b/backend/app/flow/secrets.py index 886d0b8..943e2aa 100644 --- a/backend/app/flow/secrets.py +++ b/backend/app/flow/secrets.py @@ -11,17 +11,18 @@ from __future__ import annotations import base64 import hashlib import json -import logging from pathlib import Path from typing import Any from cryptography.fernet import Fernet, InvalidToken -logger = logging.getLogger(__name__) - SECRET_REF = "$secret" +class SecretsUnreadable(RuntimeError): + """The store exists but does not decrypt with the current SECRET_KEY.""" + + class SecretNotFound(KeyError): """A node asked for a secret that is not in the store.""" @@ -50,13 +51,15 @@ class SecretsStore: self._fernet.decrypt(self._path.read_bytes()) ) return data - except (InvalidToken, ValueError): - # A changed app key makes existing secrets unreadable. - logger.error( - "Cannot decrypt %s — it was written with a different SECRET_KEY", - self._path, - ) - return {} + except (InvalidToken, ValueError) as exc: + # A changed app key makes existing secrets unreadable. Fail loudly: + # flows coming up with silently missing credentials are far harder + # to diagnose than a node that reports why it did not load. + raise SecretsUnreadable( + f"Cannot decrypt {self._path} — it was written with a " + "different SECRET_KEY. Re-encrypt it with " + "scripts/rotate_secret_key.py." + ) from exc def _write(self, data: dict[str, str]) -> None: self._path.parent.mkdir(parents=True, exist_ok=True) diff --git a/backend/scripts/rotate_secret_key.py b/backend/scripts/rotate_secret_key.py new file mode 100644 index 0000000..05b70d0 --- /dev/null +++ b/backend/scripts/rotate_secret_key.py @@ -0,0 +1,42 @@ +"""Re-encrypt the secrets store after SECRET_KEY was rotated. + +The store's encryption key is derived from SECRET_KEY, so a new key leaves +``secrets.enc`` unreadable. Put the new key in `.env` first, then hand this +script the previous one: + + uv run python scripts/rotate_secret_key.py +""" + +import logging +import sys + +from app.core.config import settings +from app.flow.secrets import SecretsStore, SecretsUnreadable + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +def main(old_key: str) -> None: + path = settings.SECRETS_FILE + if not path.exists(): + sys.exit(f"No secrets store at {path} — nothing to re-encrypt.") + if old_key == settings.SECRET_KEY: + sys.exit("Old and new key are identical — set the new SECRET_KEY first.") + + # The store has no bulk export; reading and writing it whole is the point. + try: + data = SecretsStore(path, old_key)._read() + except SecretsUnreadable: + sys.exit(f"{path} does not decrypt with that key — refusing to overwrite it.") + + SecretsStore(path, settings.SECRET_KEY)._write(data) + logger.info( + "Re-encrypted %d secret(s) in %s with the current SECRET_KEY", len(data), path + ) + + +if __name__ == "__main__": + if len(sys.argv) != 2: + sys.exit("usage: rotate_secret_key.py ") + main(sys.argv[1]) diff --git a/backend/tests/flow/test_http_hook_secret.py b/backend/tests/flow/test_http_hook_secret.py new file mode 100644 index 0000000..04b637a --- /dev/null +++ b/backend/tests/flow/test_http_hook_secret.py @@ -0,0 +1,118 @@ +"""Webhooks are mounted without authentication, so their secret is a boundary. + +The secret travels as the last path segment; anything but an exact match must +be indistinguishable from a hook that does not exist. +""" + +import asyncio +from pathlib import Path + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from app.flow.controller import HOOK_PREFIX, FlowController +from app.flow.messages import DType, MessageSpec +from app.flow.nodes import HttpNode +from app.flow.pipeline import Pipeline +from app.flow.schemas import FlowDef, NodeDef +from app.flow.secrets import init_secrets, resolve_params +from app.flow.store import FlowStore + +SECRET = "d3vice-key" +HOOK = f"{HOOK_PREFIX}/house/sensor" + + +def hook_node(**params: object) -> HttpNode: + """A trigger node mounted the way the controller mounts it.""" + node = HttpNode( + provides=[MessageSpec(name="temp", dtype=DType.FLOAT)], + params={"url": "/sensor", "method": "POST", **params}, + name="sensor", + ) + node.assign_flow("house", "sensor") + node.url = HOOK + return node + + +def mount(node: HttpNode) -> tuple[TestClient, Pipeline]: + pipeline = Pipeline(nodes=[node]) + app = FastAPI() + node.register_route(app) + return TestClient(app), pipeline + + +def test_the_right_secret_carries_the_payload_into_the_pipeline(): + node = hook_node(secret=SECRET) + client, pipeline = mount(node) + + response = client.post(f"{HOOK}/{SECRET}", json={"temp": 21.5}) + + assert response.status_code == 200 + assert pipeline.values()["house.temp"]["value"] == 21.5 + + +def test_a_hook_without_a_secret_still_answers_on_its_plain_url(): + # Flows written before the parameter existed keep working unchanged. + node = hook_node() + client, pipeline = mount(node) + + assert client.post(HOOK, json={"temp": 21.5}).status_code == 200 + assert pipeline.values()["house.temp"]["value"] == 21.5 + + +@pytest.mark.parametrize( + "path", + [ + f"{HOOK}/wrong", + # A prefix of the secret must not pass either. + f"{HOOK}/{SECRET[:-1]}", + # No secret at all: the route only exists with the segment. + HOOK, + f"{HOOK}/", + ], +) +def test_anything_but_the_secret_looks_like_no_such_hook(path: str): + node = hook_node(secret=SECRET) + client, pipeline = mount(node) + + assert client.post(path, json={"temp": 21.5}).status_code == 404 + assert pipeline.values() == {} + + +def test_the_secret_can_be_read_from_the_secrets_store(tmp_path: Path): + init_secrets(tmp_path / "secrets.enc", "app-key").set("hook_key", SECRET) + + # What the controller does before building the node. + params = resolve_params({"url": "/sensor", "secret": {"$secret": "hook_key"}}) + node = hook_node(**params) + + assert node.secret == SECRET + client, _ = mount(node) + assert client.post(f"{HOOK}/{SECRET}", json={"temp": 21.5}).status_code == 200 + + +@pytest.mark.parametrize("secret, issues", [("", 1), (SECRET, 0)]) +def test_a_hook_without_a_secret_is_reported(tmp_path: Path, secret: str, issues: int): + store = FlowStore(tmp_path / "flows") + store.write_flow( + FlowDef( + name="house", + nodes=[ + NodeDef( + id="sensor", + type="http", + params={"url": "/sensor", "secret": secret}, + provides=[MessageSpec(name="temp")], + ) + ], + ) + ) + controller = FlowController(store=store, fastapi_app=FastAPI()) + + asyncio.run(controller.reload()) + + open_hooks = [i for i in controller.issues if i.code == "unauthenticated_hook"] + assert len(open_hooks) == issues + # Either way the flow loads — an existing hook keeps working. + assert controller.get_node("house.sensor") is not None