diff --git a/docs/channels.md b/docs/channels.md index 532f5f908..9587c775a 100644 --- a/docs/channels.md +++ b/docs/channels.md @@ -410,23 +410,45 @@ The `services` table schema: ### Security -- **Authentication** — the gateway's `POST /v1/api/notify` endpoint - requires authentication. Configure `TURNSTONE_JWT_SECRET` so the - server can mint JWTs with `aud: turnstone-channel` automatically. - If the secret is not set, the gateway fails closed and rejects all - requests with 401. Server JWTs (`aud: turnstone-server`) are - rejected. +- **Authentication** — the gateway's `POST /v1/api/notify` endpoint requires authentication. + Configure the server's signing secret with `TURNSTONE_JWT_SECRET` or `[auth].jwt_secret` in + `config.toml`; the environment variable takes precedence. The gateway reads + `TURNSTONE_JWT_SECRET`, which must match the server's secret. The server automatically mints JWTs + with `aud: turnstone-channel`. If the gateway's secret is not set, it fails closed and rejects all + requests with 401. Server JWTs (`aud: turnstone-server`) are rejected. - **Rate limit** — maximum 5 notifications per turn. The counter only increments on successful delivery, so failures don't consume the budget. -- **SSRF protection** — only `http://` and `https://` service URLs - are allowed. Other schemes are silently skipped. +- **SSRF protection** — only `http://` and `https://` service URLs are allowed. Other schemes are + skipped with a warning. - **Mention sanitization** — `discord.utils.escape_mentions()` is applied before sending, preventing `@everyone` / `@here` abuse. - **Error redaction** — generic error messages are returned to the LLM. Internal details (service IDs, URLs, exception messages) are logged server-side only. +### Troubleshooting notifications + +Check both the originating server and the channel gateway logs. Each `notify.gateway_failed` event +identifies the gateway, its URL, the attempt, the workstream (`ws_id`), and the tool call (`call_id`). +The event records whether an Authorization header was present, plus the HTTP status, request exception +type, or delivery statuses such as `no_adapter`, `failed`, and `timeout`. Completion notifications log +the same gateway and workstream details under `notify_completion.*`. Gateway URLs omit credentials, +query parameters, and fragments; outbound diagnostics omit tokens, message content, and response +bodies. + +A 401 from the gateway occurs before Discord or Slack delivery. `notify.auth_missing` on the server +means it could not find notification credentials. On the gateway, `notify.auth_not_configured` means +its signing secret is missing; `notify.auth_rejected` distinguishes a missing Authorization header, +an invalid scheme or token format, and an invalid JWT. For `invalid_jwt`, check the shared secret, the +`turnstone-channel` audience, token expiry, and host clocks. `notify.auth_insufficient_scope` indicates +a valid JWT without the required `write` scope and returns 403. + +With debug logging enabled before credentials are first used, `notify.auth_configured` records the +credential source without its value. A configured `TURNSTONE_CHANNEL_AUTH_TOKEN` overrides automatic +JWT minting and must itself be a valid JWT for the channel audience with `write` scope. Working Discord +or Slack conversations do not verify this outbound notification authentication path. + --- ## Adding New Adapters diff --git a/tests/test_notify_http.py b/tests/test_notify_http.py index eea666567..10ddeb358 100644 --- a/tests/test_notify_http.py +++ b/tests/test_notify_http.py @@ -264,6 +264,32 @@ def test_whitespace_only_message(self, client): class TestNotifyAuth: """Tests for authentication on the /v1/api/notify endpoint.""" + @pytest.mark.parametrize( + ("header", "reason"), + [ + ("", "missing_authorization"), + ("Basic private-credentials", "invalid_auth_scheme"), + ("Bearer private-token", "invalid_token_format"), + ("Bearer private-token.invalid.jwt", "invalid_jwt"), + ], + ) + def test_auth_rejection_logs_reason_without_credentials(self, authed_client, header, reason): + from unittest.mock import patch + + with patch("turnstone.channels._http.log") as logger: + response = authed_client.post( + "/v1/api/notify", + json={"message": "private-message"}, + headers={"Authorization": header} if header else {}, + ) + + assert response.status_code == 401 + assert response.json() == {"error": "Unauthorized"} + logger.warning.assert_called_once_with( + "notify.auth_rejected", reason=reason, client_host="testclient" + ) + assert "private-" not in repr(logger.mock_calls) + def test_reject_when_unconfigured(self, no_auth_client): """Requests are rejected (fail closed) when no auth is configured.""" resp = no_auth_client.post( diff --git a/tests/test_notify_tool.py b/tests/test_notify_tool.py index 5aac42532..66439547f 100644 --- a/tests/test_notify_tool.py +++ b/tests/test_notify_tool.py @@ -5,6 +5,8 @@ from typing import TYPE_CHECKING from unittest.mock import MagicMock +import pytest + if TYPE_CHECKING: from turnstone.core.session import ChatSession @@ -32,6 +34,216 @@ def _make_session() -> ChatSession: return session +class TestNotifyAuthHeaders: + @pytest.fixture(autouse=True) + def auth_config(self, tmp_path, monkeypatch): + import turnstone.core.config as config + import turnstone.core.session as session + + config_path = tmp_path / "config.toml" + config_path.touch(mode=0o600) + monkeypatch.setattr(config, "_config_path", config_path) + monkeypatch.setattr(config, "_cache", None) + monkeypatch.setattr(session, "_notify_token_manager", None) + monkeypatch.delenv("TURNSTONE_JWT_SECRET", raising=False) + monkeypatch.delenv("TURNSTONE_CHANNEL_AUTH_TOKEN", raising=False) + return config_path + + def test_config_only_secret_authenticates_with_gateway(self, auth_config): + """The documented bare-metal config must authenticate outbound notifications.""" + from unittest.mock import AsyncMock + + from starlette.testclient import TestClient + + from turnstone.channels._http import create_channel_app + from turnstone.core.session import _notify_auth_headers + + secret = "a" * 32 + auth_config.write_text(f'[auth]\njwt_secret = " {secret} "\n') + adapter = AsyncMock() + adapter.send.return_value = "message-1" + app = create_channel_app({"discord": adapter}, MagicMock(), jwt_secret=secret) + with TestClient(app) as client: + response = client.post( + "/v1/api/notify", + json={ + "target": {"channel_type": "discord", "channel_id": "123"}, + "message": "Hello!", + }, + headers=_notify_auth_headers(), + ) + + assert response.status_code == 200 + assert response.json()["results"][0]["status"] == "sent" + adapter.send.assert_awaited_once_with("123", "Hello!") + + def test_environment_secret_takes_precedence(self, auth_config, monkeypatch): + from turnstone.core.auth import JWT_AUD_CHANNEL, validate_jwt + from turnstone.core.session import _notify_auth_headers + + config_secret = "a" * 32 + auth_config.write_text(f'[auth]\njwt_secret = "{config_secret}"\n') + secret = "b" * 32 + monkeypatch.setenv("TURNSTONE_JWT_SECRET", f" {secret} ") + + token = _notify_auth_headers()["Authorization"].removeprefix("Bearer ") + auth = validate_jwt(token, secret, audience=JWT_AUD_CHANNEL) + assert auth is not None + assert "write" in auth.scopes + + def test_static_token_takes_precedence(self, auth_config, monkeypatch): + from turnstone.core.session import _notify_auth_headers + + config_secret = "a" * 32 + auth_config.write_text(f'[auth]\njwt_secret = "{config_secret}"\n') + monkeypatch.setenv("TURNSTONE_JWT_SECRET", "b" * 32) + monkeypatch.setenv("TURNSTONE_CHANNEL_AUTH_TOKEN", " static-token ") + assert _notify_auth_headers() == {"Authorization": "Bearer static-token"} + + def test_missing_secret_returns_no_headers(self): + from turnstone.core.session import _notify_auth_headers + + assert _notify_auth_headers() == {} + + +class TestNotifyDiagnostics: + @pytest.fixture(params=["tool", "completion"]) + def notify_caller(self, request, monkeypatch): + """Exercise both outbound paths with the same gateway failures.""" + import turnstone.core.session as session_module + import turnstone.server as server_module + + session = _make_session() + session._ws_id = "abcdef1234567890" + monkeypatch.setattr(session, "_backoff_or_cancelled", MagicMock()) + monkeypatch.setattr(server_module.time, "sleep", MagicMock()) + monkeypatch.setattr( + session_module, + "_notify_auth_headers", + lambda: {"Authorization": "Bearer private-auth-token"}, + ) + storage = MagicMock() + monkeypatch.setattr(session_module, "get_storage", lambda: storage) + caller_module = session_module if request.param == "tool" else server_module + logger = MagicMock() + monkeypatch.setattr(caller_module, "log", logger) + post = MagicMock() + monkeypatch.setattr(session_module.httpx, "post", post) + + def deliver(): + if request.param == "tool": + _, result = session._exec_notify( + { + "call_id": "call-1", + "channel_type": "discord", + "channel_id": "123", + "message": "private-message-content", + } + ) + assert result == "Error: notification delivery failed" + assert session._notify_count == 0 + else: + server_module._deliver_notification( + storage, + {"ws_id": session._ws_id, "message": "private-message-content"}, + {"Authorization": "Bearer private-auth-token"}, + ) + + return storage, post, logger, deliver, request.param + + def test_preserves_each_gateway_failure_without_secrets(self, notify_caller): + storage, post, logger, deliver, caller = notify_caller + storage.list_services.return_value = [ + {"service_id": "gateway-1", "url": "http://user:private-password@gw.example.com:8091"}, + {"service_id": "gateway-2", "url": "http://gw2.example.com:8091"}, + ] + post.side_effect = [ + MagicMock(status_code=401), + ConnectionError("private-exception-content"), + ] * 3 + + deliver() + + failures = [c.kwargs for c in logger.warning.call_args_list if "gateway_id" in c.kwargs] + assert len(failures) == 6 + for attempt in range(1, 4): + rejected, unreachable = failures[(attempt - 1) * 2 : attempt * 2] + assert rejected["gateway_id"] == "gateway-1" + assert rejected["gateway_url"] == "http://gw.example.com:8091/v1/api/notify" + assert rejected.get("status_code", rejected.get("status")) == 401 + assert unreachable["gateway_id"] == "gateway-2" + assert unreachable["error_type"] == "ConnectionError" + for failure in (rejected, unreachable): + assert failure["attempt"] == attempt + assert failure["ws_id"] == "abcdef1234567890" + assert failure["auth_present"] is True + if caller == "tool": + assert failure["call_id"] == "call-1" + assert "private-" not in repr(logger.mock_calls) + + @pytest.mark.parametrize( + "response_kind", ["failed_deliveries", "invalid_json", "invalid_results"] + ) + def test_unsuccessful_response_logs_safe_details(self, notify_caller, response_kind): + storage, post, logger, deliver, _ = notify_caller + storage.list_services.return_value = [ + {"service_id": "gateway-1", "url": "http://gw.example.com:8091"}, + ] + response = MagicMock(status_code=200) + if response_kind == "invalid_json": + response.json.side_effect = ValueError("private-response-content") + elif response_kind == "invalid_results": + response.json.return_value = {"results": "private-response-content"} + else: + response.json.return_value = { + "results": [ + {"status": "no_adapter", "channel_id": "private-target"}, + {"status": "private-response-content"}, + {"status": ["private-response-content"]}, + ] + } + post.return_value = response + + deliver() + + failures = [c for c in logger.warning.call_args_list if "gateway_id" in c.kwargs] + assert len(failures) == 3 + for failure in failures: + if response_kind == "invalid_json": + assert ( + failure.kwargs.get("reason") == "invalid_response" + or failure.args[0] == "notify_completion.response_parse_error" + ) + else: + expected = ( + ["invalid_response"] + if response_kind == "invalid_results" + else ["no_adapter", "unknown"] + ) + assert failure.kwargs["delivery_statuses"] == expected + assert "private-" not in repr(logger.mock_calls) + + @pytest.mark.parametrize( + ("url", "expected"), + [ + ( + "https://user:password@gw.example.com/prefix?token=secret#fragment", + "https://gw.example.com/prefix", + ), + ( + "http://user:password@[2001:db8::1]:8091/v1/api/notify", + "http://[2001:db8::1]:8091/v1/api/notify", + ), + ("http://[invalid", ""), + ("file:///private-credential", ""), + ], + ) + def test_gateway_url_redaction(self, url, expected): + from turnstone.core.session import _notify_log_url + + assert _notify_log_url(url) == expected + + class TestPrepareNotify: def test_valid_username_target(self): session = _make_session() diff --git a/turnstone/channels/_http.py b/turnstone/channels/_http.py index 3412d34dd..6b89e284d 100644 --- a/turnstone/channels/_http.py +++ b/turnstone/channels/_http.py @@ -42,13 +42,19 @@ async def _handle_health(request: Request) -> JSONResponse: def _check_auth(request: Request) -> JSONResponse | None: """Validate the request's Authorization header. Returns an error response or None.""" jwt_secret: str = getattr(request.app.state, "jwt_secret", "") + client_host = request.client.host if request.client else "" if not jwt_secret: - log.warning("notify.auth_not_configured") + log.warning("notify.auth_not_configured", client_host=client_host) return JSONResponse({"error": "authentication not configured"}, status_code=401) header = request.headers.get("Authorization", "") if not header.startswith("Bearer "): + log.warning( + "notify.auth_rejected", + reason="missing_authorization" if not header else "invalid_auth_scheme", + client_host=client_host, + ) return JSONResponse({"error": "Unauthorized"}, status_code=401) token = header[7:] @@ -67,10 +73,16 @@ def _check_auth(request: Request) -> JSONResponse | None: "notify.auth_insufficient_scope", user_id=result.user_id, scopes=sorted(result.scopes), + client_host=client_host, ) return JSONResponse({"error": "insufficient scope"}, status_code=403) return None + log.warning( + "notify.auth_rejected", + reason="invalid_jwt" if "." in token else "invalid_token_format", + client_host=client_host, + ) return JSONResponse({"error": "Unauthorized"}, status_code=401) @@ -105,7 +117,7 @@ async def _handle_notify(request: Request) -> JSONResponse: if "username" in target: user = await asyncio.to_thread(storage.get_user_by_username, target["username"]) if user is None: - log.warning("notify.user_not_found", username=target["username"]) + log.warning("notify.user_not_found", username=target["username"], ws_id=ws_id) return JSONResponse( {"error": "target not found or has no linked channels"}, status_code=404, @@ -114,7 +126,7 @@ async def _handle_notify(request: Request) -> JSONResponse: for link in links: targets.append((link["channel_type"], link["channel_user_id"])) if not targets: - log.warning("notify.user_no_linked_channels", username=target["username"]) + log.warning("notify.user_no_linked_channels", username=target["username"], ws_id=ws_id) return JSONResponse( {"error": "target not found or has no linked channels"}, status_code=404, @@ -159,6 +171,7 @@ async def _handle_notify(request: Request) -> JSONResponse: "notify.no_adapter", channel_type=channel_type, channel_id=channel_id, + ws_id=ws_id, ) continue try: @@ -181,12 +194,14 @@ async def _handle_notify(request: Request) -> JSONResponse: channel_type=channel_type, channel_id=channel_id, message_id=msg_id, + ws_id=ws_id, ) except TimeoutError: log.warning( "notify.timeout", channel_type=channel_type, channel_id=channel_id, + ws_id=ws_id, ) results.append( { @@ -200,6 +215,7 @@ async def _handle_notify(request: Request) -> JSONResponse: "notify.delivery_failed", channel_type=channel_type, channel_id=channel_id, + ws_id=ws_id, ) results.append( { diff --git a/turnstone/core/session.py b/turnstone/core/session.py index b0f7b1fcf..e1dd598c9 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -2791,7 +2791,7 @@ def _format_mcp_dispatch_error(prefix: str, exc: Exception) -> str: # --------------------------------------------------------------------------- -# Notify auth helper (module-level, lazy-init) +# Notify helpers # --------------------------------------------------------------------------- _notify_token_manager: Any = None @@ -2805,11 +2805,22 @@ def _notify_auth_headers() -> dict[str, str]: # Static token from env takes precedence static_token = os.environ.get("TURNSTONE_CHANNEL_AUTH_TOKEN", "").strip() if static_token: + log.debug("notify.auth_configured", source="TURNSTONE_CHANNEL_AUTH_TOKEN") return {"Authorization": f"Bearer {static_token}"} # JWT via ServiceTokenManager jwt_secret = os.environ.get("TURNSTONE_JWT_SECRET", "").strip() + secret_source = "TURNSTONE_JWT_SECRET" if not jwt_secret: + from turnstone.core.config import load_config + + jwt_secret = str(load_config("auth").get("jwt_secret", "")).strip() + secret_source = "config.toml [auth].jwt_secret" + if not jwt_secret: + log.warning( + "notify.auth_missing", + hint="Set TURNSTONE_JWT_SECRET or [auth].jwt_secret in config.toml", + ) return {} with _notify_token_lock: @@ -2823,10 +2834,35 @@ def _notify_auth_headers() -> dict[str, str]: secret=jwt_secret, audience=JWT_AUD_CHANNEL, ) + log.debug("notify.auth_configured", source=secret_source) header: dict[str, str] = _notify_token_manager.bearer_header return header +def _notify_log_url(url: str) -> str: + """Strip URL credentials, query parameters, and fragments from gateway diagnostics.""" + from urllib.parse import urlsplit, urlunsplit + + try: + parts = urlsplit(url) + if parts.scheme not in ("http", "https") or not parts.hostname: + return "" + return urlunsplit((parts.scheme, parts.netloc.rsplit("@", 1)[-1], parts.path, "", "")) + except ValueError: + return "" + + +def _notify_delivery_statuses(results: Any) -> list[str]: + """Summarize gateway outcomes without logging response content or target details.""" + if not isinstance(results, list): + return ["invalid_response"] + statuses: set[str] = set() + for result in results: + status = result.get("status") if isinstance(result, dict) else None + statuses.add(status if status in ("sent", "failed", "timeout", "no_adapter") else "unknown") + return sorted(statuses) + + def _screen_tool_url(url: str, allow_private_network: bool) -> tuple[str | None, bool, bool]: """SSRF-screen a tool's target URL under the operator's private-network opt-in. @@ -26988,6 +27024,11 @@ def _exec_notify(self, item: dict[str, Any]) -> tuple[str, str]: # Build auth headers for service-to-service call auth_headers = _notify_auth_headers() + log_fields = { + "ws_id": self._ws_id, + "call_id": call_id, + "auth_present": bool(auth_headers.get("Authorization")), + } # Retry loop: attempt delivery, re-query services on each retry # in case a gateway comes back online between attempts. @@ -27002,13 +27043,14 @@ def _exec_notify(self, item: dict[str, Any]) -> tuple[str, str]: attempt=attempt + 1, max_retries=self._NOTIFY_MAX_RETRIES, retry_delay=delay, + **log_fields, ) # Cancel-aware: notify runs as an in-turn tool, so a # Stop aborts pending delivery retries with the turn # (the batch synthesizes the cancelled tool_result). self._backoff_or_cancelled(delay) continue - log.warning("notify.no_services_exhausted") + log.warning("notify.no_services_exhausted", **log_fields) msg = "Error: no channel gateway services available" self._report_tool_result(call_id, "notify", msg, is_error=True) return call_id, msg @@ -27017,8 +27059,16 @@ def _exec_notify(self, item: dict[str, Any]) -> tuple[str, str]: last_error: str = "" for svc in services: url = svc["url"].rstrip("/") + "/v1/api/notify" + gateway_fields = { + **log_fields, + "gateway_id": svc.get("service_id", ""), + "gateway_url": _notify_log_url(url), + "attempt": attempt + 1, + } # SSRF guard: only allow http(s) URLs if not url.startswith(("http://", "https://")): + last_error = "invalid gateway URL" + log.warning("notify.gateway_failed", reason="invalid_url", **gateway_fields) continue try: resp = httpx.post(url, json=payload, timeout=10, headers=auth_headers) @@ -27028,20 +27078,46 @@ def _exec_notify(self, item: dict[str, Any]) -> tuple[str, str]: data = resp.json() except Exception: last_error = "invalid gateway response" + log.warning( + "notify.gateway_failed", + reason="invalid_response", + status_code=resp.status_code, + **gateway_fields, + ) continue results = data.get("results") if isinstance(data, dict) else None if isinstance(results, list) and any( isinstance(r, dict) and r.get("status") == "sent" for r in results ): self._notify_count += 1 + log.info("notify.delivered", **gateway_fields) msg = "Notification sent successfully" self._report_tool_result(call_id, "notify", msg) return call_id, msg last_error = "no successful deliveries" + log.warning( + "notify.gateway_failed", + reason="no_successful_deliveries", + status_code=resp.status_code, + delivery_statuses=_notify_delivery_statuses(results), + **gateway_fields, + ) continue last_error = f"HTTP {resp.status_code}" + log.warning( + "notify.gateway_failed", + reason="http_error", + status_code=resp.status_code, + **gateway_fields, + ) except Exception as exc: last_error = type(exc).__name__ + log.warning( + "notify.gateway_failed", + reason="request_error", + error_type=last_error, + **gateway_fields, + ) continue # try next gateway # All gateways failed this attempt — retry if we have attempts left @@ -27054,6 +27130,7 @@ def _exec_notify(self, item: dict[str, Any]) -> tuple[str, str]: last_error=last_error, gateway_count=len(services), retry_delay=delay, + **log_fields, ) # Same cancel-aware backoff as the no-services arm above. self._backoff_or_cancelled(delay) @@ -27062,6 +27139,7 @@ def _exec_notify(self, item: dict[str, Any]) -> tuple[str, str]: "notify.delivery_failed", last_error=last_error, gateway_count=len(services), + **log_fields, ) msg = "Error: notification delivery failed" diff --git a/turnstone/server.py b/turnstone/server.py index 477bd3e44..c5e9492ad 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -2464,18 +2464,32 @@ def _deliver_notification( """POST to channel gateway /v1/api/notify with retry.""" import httpx + from turnstone.core.session import _notify_delivery_statuses, _notify_log_url + + log_fields = { + "ws_id": payload.get("ws_id", ""), + "auth_present": bool(auth_headers.get("Authorization")), + } for attempt in range(3): services = storage.list_services("channel", max_age_seconds=120) if not services: if attempt < 2: + log.warning("notify_completion.no_services", attempt=attempt + 1, **log_fields) time.sleep(1.0 if attempt == 0 else 3.0) continue - log.warning("notify_completion.no_services") + log.warning("notify_completion.no_services", attempt=attempt + 1, **log_fields) return for svc in services: url = svc["url"].rstrip("/") + "/v1/api/notify" + gateway_fields = { + **log_fields, + "gateway_id": svc.get("service_id", ""), + "gateway_url": _notify_log_url(url), + "attempt": attempt + 1, + } if not url.startswith(("http://", "https://")): + log.warning("notify_completion.invalid_url", **gateway_fields) continue try: resp = httpx.post(url, json=payload, timeout=10, headers=auth_headers) @@ -2487,24 +2501,37 @@ def _deliver_notification( if isinstance(results, list) and any( isinstance(r, dict) and r.get("status") == "sent" for r in results ): - log.info("notify_completion.delivered", ws_id=payload.get("ws_id")) + log.info("notify_completion.delivered", **gateway_fields) return except Exception: - log.debug("notify_completion.response_parse_error", url=url, exc_info=True) - log.warning("notify_completion.no_successful_delivery", url=url) + log.warning( + "notify_completion.response_parse_error", + status=resp.status_code, + **gateway_fields, + ) + continue + log.warning( + "notify_completion.no_successful_delivery", + delivery_statuses=_notify_delivery_statuses(results), + **gateway_fields, + ) continue log.warning( "notify_completion.failed", status=resp.status_code, - url=url, + **gateway_fields, + ) + except Exception as exc: + log.warning( + "notify_completion.error", error_type=type(exc).__name__, **gateway_fields ) - except Exception: - log.exception("notify_completion.error", url=url) continue if attempt < 2: time.sleep(1.0 if attempt == 0 else 3.0) + log.warning("notify_completion.delivery_failed", attempts=3, **log_fields) + async def _interactive_create_validate_request( request: Request,