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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KkmeRiyeYmVZqJVwuyHq9o

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