diff --git a/.env.example b/.env.example index c13cf00a23..f00140cf5d 100644 --- a/.env.example +++ b/.env.example @@ -107,6 +107,14 @@ UNBROWSE_API_KEY=your-unbrowse-api-key # manually when Gateway workers also run outside the bundled deploy script. # DEER_FLOW_INTERNAL_AUTH_TOKEN=your-shared-internal-token +# At-rest encryption key for stored credentials (a Fernet key). Comma-separate +# several to rotate: the first encrypts, all decrypt. Unset, a single instance +# generates DEER_FLOW_HOME/.credentials_key (`make up` persists the same file); +# instances that do not share DEER_FLOW_HOME must all set the same value. +# Back it up: losing it makes stored credentials unreadable. Generate: +# python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" +# DEER_FLOW_CREDENTIALS_KEY=your-fernet-key + # Provisioner API key for sandbox authentication (required when using provisioner/K8s sandbox mode). # The same value must be set on the provisioner container and in config.yaml sandbox.provisioner_api_key. # Generate: openssl rand -hex 32 diff --git a/CHANGELOG.md b/CHANGELOG.md index 79012c935d..24007f96f3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4388,6 +4388,22 @@ This release closes that milestone with **439 merged pull requests**. truncated at that character. These providers now split on `" "` only, as the shared parser already documents and as LocalSandbox, the AIO backend and E2B already do. ([#6595]) +- **security:** Add `DEER_FLOW_CREDENTIALS_KEY`, an env-only at-rest + encryption key for stored credentials. Per-connection IM channel credentials + had an encryption path that no production code wired up, so they could be + neither stored nor read and Slack always used the deployment bot token. The + Gateway now loads the key at startup (comma-separated Fernet keys: the first + encrypts, all decrypt, for rotation; values carry a `fernet:v2:` prefix and + earlier `fernet:v1:` values stay readable) and passes it to every channel + connection repository; undecryptable values are treated as missing. Unset, a + single instance generates `{DEER_FLOW_HOME}/.credentials_key`; a declared + multi-instance deployment with `channel_connections` enabled refuses to start + without the key, and a malformed key is refused without being echoed. The + Helm chart generates the key into its app Secret and preserves it across + upgrades, `make up` persists it next to the runtime home, and both compose + files pass it to the Gateway. `.jwt_secret` (and the managed-model key) are + now created exclusively and read back, so replicas cold-starting on a shared + volume no longer keep different session-signing secrets. ([#6611]) ### Documentation @@ -9517,3 +9533,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6588]: https://github.com/bytedance/deer-flow/pull/6588 [#6590]: https://github.com/bytedance/deer-flow/pull/6590 [#6595]: https://github.com/bytedance/deer-flow/pull/6595 +[#6611]: https://github.com/bytedance/deer-flow/pull/6611 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 5319f710f3..3088bf8777 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -3381,6 +3381,17 @@ 的匹配行会在该字符处被截断。现在这三个提供者只按 `" "` 切分，与共享解析器 既有的约定以及 LocalSandbox、AIO 后端、E2B 的行为一致。([#6595]) +- **安全：** 新增仅从环境变量读取的静态凭据加密密钥 `DEER_FLOW_CREDENTIALS_KEY`。 + 按连接存储的 IM 渠道凭据此前虽有加密路径，但没有任何生产代码接入，既无法写入 + 也无法读取，Slack 因而始终使用部署级 bot token。现在 Gateway 在启动时加载该密钥 + （逗号分隔的 Fernet 密钥：第一个用于加密，全部用于解密，以支持轮换；密文带 + `fernet:v2:` 前缀，旧的 `fernet:v1:` 密文仍可读取），并传给所有渠道连接仓库； + 无法解密的值按缺失处理。未设置时单实例会生成 `{DEER_FLOW_HOME}/.credentials_key`； + 声明为多实例且启用 `channel_connections` 的部署在缺少密钥时拒绝启动，格式错误的 + 密钥同样会被拒绝且不会回显。Helm chart 将密钥生成到 app Secret 并在升级时保留， + `make up` 将其持久化到运行时目录，两个 compose 文件都会把它传给 Gateway。 + `.jwt_secret`（以及托管模型密钥）现在以独占方式创建并回读，共享卷上同时冷启动的 + 副本不再各自保留不同的会话签名密钥。 ([#6611]) ### 文档 @@ -7653,3 +7664,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#6588]: https://github.com/bytedance/deer-flow/pull/6588 [#6590]: https://github.com/bytedance/deer-flow/pull/6590 [#6595]: https://github.com/bytedance/deer-flow/pull/6595 +[#6611]: https://github.com/bytedance/deer-flow/pull/6611 diff --git a/README.md b/README.md index 355df27cd2..308d1fd51b 100644 --- a/README.md +++ b/README.md @@ -533,7 +533,7 @@ Browser login uses `HttpOnly` session cookies. The login page offers a "keep me DeerFlow still uses `Forwarded` / `X-Forwarded-*` headers to recover the browser-facing scheme and origin behind a proxy. The bundled nginx sets `X-Forwarded-Proto`, but preserves an upstream HTTPS value and does not overwrite every forwarded header. Configure the outer trusted proxy to replace or strip client-supplied forwarding headers before traffic reaches DeerFlow. > [!IMPORTANT] -> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; process-local memory/JSONL event stores cannot enforce singleton delivery receipts across workers. Kubernetes replicas run one worker per Pod, which the worker count cannot see: declare them with `deployment.multi_instance: true` (or `DEER_FLOW_MULTI_INSTANCE=1`, which deploy tooling such as a Helm chart can set from its replica count) so the startup gate enforces the same prerequisites instead of staying inert. The bridge shares SSE delivery and bounded `Last-Event-ID` replay across workers. When a valid reconnect cursor has been trimmed, or a subscriber that already established an empty-stream wait falls behind before its first delivery, Memory and Redis emit a machine-readable SSE `gap` event instead of silently returning a partial replay; the Web UI reloads durable thread/event state and resumes from the retained tail. Lease reconciliation marks runs from dead workers as errors, persists their delivery receipts, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE, `/wait`, and internal stream consumers use `stream_bridge.heartbeat_interval_seconds` (default `15`) for idle liveness checks; changing it requires a Gateway restart. Malformed Redis reconnect IDs live-tail new events instead of replaying the retained buffer, and the rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. Failed-login counters and lockouts for `POST /api/v1/auth/login/local` (`auth.local.max_login_attempts` / `lockout_seconds`) are kept in the shared `login_throttle` table whenever the application database is SQLite or Postgres (`auth.local.throttle_storage: auto`, the default), so every replica enforces one lockout per client IP; `memory` keeps the historical per-process counter, which under N replicas hands an attacker N × `max_login_attempts` guesses and logs a startup warning. IM chat-to-thread bindings live in the shared `channel_thread_bindings` table whenever the application database is SQLite or Postgres (an existing `channels/store.json` is imported once at startup and renamed `store.json.migrated`); the remaining IM channel state (each instance's platform connections, follow-up buffers) still needs its own multi-worker coordination. +> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; process-local memory/JSONL event stores cannot enforce singleton delivery receipts across workers. Kubernetes replicas run one worker per Pod, which the worker count cannot see: declare them with `deployment.multi_instance: true` (or `DEER_FLOW_MULTI_INSTANCE=1`, which deploy tooling such as a Helm chart can set from its replica count) so the startup gate enforces the same prerequisites instead of staying inert. Declared instances that store credentials (`channel_connections`) must also share one at-rest `DEER_FLOW_CREDENTIALS_KEY` (the Helm chart and `make up` generate it; back it up, see [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#credentials-encryption-key-deer_flow_credentials_key)). The bridge shares SSE delivery and bounded `Last-Event-ID` replay across workers. When a valid reconnect cursor has been trimmed, or a subscriber that already established an empty-stream wait falls behind before its first delivery, Memory and Redis emit a machine-readable SSE `gap` event instead of silently returning a partial replay; the Web UI reloads durable thread/event state and resumes from the retained tail. Lease reconciliation marks runs from dead workers as errors, persists their delivery receipts, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE, `/wait`, and internal stream consumers use `stream_bridge.heartbeat_interval_seconds` (default `15`) for idle liveness checks; changing it requires a Gateway restart. Malformed Redis reconnect IDs live-tail new events instead of replaying the retained buffer, and the rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. Failed-login counters and lockouts for `POST /api/v1/auth/login/local` (`auth.local.max_login_attempts` / `lockout_seconds`) are kept in the shared `login_throttle` table whenever the application database is SQLite or Postgres (`auth.local.throttle_storage: auto`, the default), so every replica enforces one lockout per client IP; `memory` keeps the historical per-process counter, which under N replicas hands an attacker N × `max_login_attempts` guesses and logs a startup warning. IM chat-to-thread bindings live in the shared `channel_thread_bindings` table whenever the application database is SQLite or Postgres (an existing `channels/store.json` is imported once at startup and renamed `store.json.migrated`); the remaining IM channel state (each instance's platform connections, follow-up buffers) still needs its own multi-worker coordination. > > In single-process JSONL deployments, cancelling an admitted event-store mutation waits for its background file I/O, rollback, and bookkeeping to settle before releasing the thread write lock. This prevents an older cancelled write from recreating deleted records or rolling back a later successful write. Cancellation can therefore wait on slow storage; it does not stop an in-flight filesystem operation. Callers still waiting to acquire the lock can cancel without starting a mutation. A batch spanning multiple threads drains its current thread group before propagating cancellation; subsequent thread groups do not start. > diff --git a/backend/app/channels/service.py b/backend/app/channels/service.py index 977d982277..c1b190cd81 100644 --- a/backend/app/channels/service.py +++ b/backend/app/channels/service.py @@ -95,7 +95,8 @@ def _merge_channel_connection_runtime_config(channels_config: dict[str, Any], ap merge_runtime_channel_configs(channels_config, connection_config) -def _make_connection_repo(connection_config: ChannelConnectionsConfig | None): +def _make_connection_repo(connection_config: ChannelConnectionsConfig | None, *, credentials_cipher: Any | None = None): + """Build the connection repository; ``credentials_cipher`` encrypts per-connection credentials.""" if connection_config is None or not getattr(connection_config, "enabled", False): return None @@ -110,7 +111,7 @@ def _make_connection_repo(connection_config: ChannelConnectionsConfig | None): if session_factory is None: logger.warning("Channel connections are enabled but database persistence is not available") return None - return ChannelConnectionRepository(session_factory) + return ChannelConnectionRepository(session_factory, cipher=credentials_cipher) class ChannelService: @@ -177,13 +178,16 @@ class ChannelService: app_config: AppConfig | None = None, *, get_stream_bridge: Callable[[], StreamBridge | None] | None = None, + credentials_cipher: Any | None = None, ) -> ChannelService: """Create a ChannelService from the application config. ``get_stream_bridge`` is threaded straight through to the ``ChannelManager`` (see its docstring); it is optional so direct callers (including most tests) that don't need follow-up-buffer - auto-draining can omit it. + auto-draining can omit it. ``credentials_cipher`` is the Gateway's + ``DEER_FLOW_CREDENTIALS_KEY`` cipher for per-connection credentials; + without it stored credentials stay unavailable. """ if app_config is None: from deerflow.config.app_config import get_app_config @@ -200,7 +204,7 @@ class ChannelService: require_bound_identity = bool(connections_enabled and getattr(connection_config, "require_bound_identity", True)) return cls( channels_config=channels_config, - connection_repo=_make_connection_repo(connection_config), + connection_repo=_make_connection_repo(connection_config, credentials_cipher=credentials_cipher), require_bound_identity=require_bound_identity, app_config=app_config, get_stream_bridge=get_stream_bridge, @@ -663,6 +667,7 @@ async def start_channel_service( app_config: AppConfig | None = None, *, get_stream_bridge: Callable[[], StreamBridge | None] | None = None, + credentials_cipher: Any | None = None, ) -> ChannelService: """Create and start the global ChannelService from app config. @@ -679,7 +684,7 @@ async def start_channel_service( # from_app_config resolves the channel store path (realpath) and reads the # runtime config files; keep that disk IO off the event loop. # asyncio.to_thread forwards both args and kwargs to the target callable. - service = await asyncio.to_thread(ChannelService.from_app_config, app_config, get_stream_bridge=get_stream_bridge) + service = await asyncio.to_thread(ChannelService.from_app_config, app_config, get_stream_bridge=get_stream_bridge, credentials_cipher=credentials_cipher) _channel_service = service async def rollback_failed_start() -> None: diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py index 0747de54c5..ae06798fe2 100644 --- a/backend/app/gateway/app.py +++ b/backend/app/gateway/app.py @@ -437,7 +437,29 @@ async def _shutdown_subagent_batch_service(app: FastAPI) -> None: set_subagent_batch_submitter(None) -def _scheduled_task_notification_repos(startup_config: AppConfig): +async def _load_credentials_cipher(startup_config: AppConfig): + """Load the at-rest credentials cipher when an enabled feature stores credentials, else ``None``. + + Without ``DEER_FLOW_CREDENTIALS_KEY`` the first load creates + ``{base_dir}/.credentials_key``, so it runs in a worker thread. The startup + gate has already refused a malformed environment key; a key file that + cannot be read or created leaves stored credentials unavailable (the + behaviour before the key existed) instead of failing the Gateway. + """ + from app.gateway.deps import credentials_key_consumers + + if not credentials_key_consumers(startup_config): + return None + from deerflow.config import credentials_key + + try: + return await asyncio.to_thread(credentials_key.get_credentials_cipher) + except Exception: + logger.exception("Failed to load the credentials encryption key; stored channel connection credentials stay unavailable") + return None + + +def _scheduled_task_notification_repos(startup_config: AppConfig, credentials_cipher=None): """Return ``(connection_repo, notification_repo)`` for the scheduled-run outbox (issue #4254). Both are None unless channel connections are enabled, because the completion @@ -454,7 +476,7 @@ def _scheduled_task_notification_repos(startup_config: AppConfig): session_factory = get_session_factory() if session_factory is None: return None, None - return ChannelConnectionRepository(session_factory), NotificationDeliveryRepository(session_factory) + return ChannelConnectionRepository(session_factory, cipher=credentials_cipher), NotificationDeliveryRepository(session_factory) async def _start_scheduled_task_notification_delivery(app: FastAPI, startup_config: AppConfig, notification_repo, connection_repo) -> None: @@ -680,12 +702,18 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: except Exception: logger.exception("Failed to start summary generation workers") + # At-rest credentials cipher (DEER_FLOW_CREDENTIALS_KEY) shared by every + # ChannelConnectionRepository this process builds; None when no enabled + # feature stores credentials. + credentials_cipher = await _load_credentials_cipher(startup_config) + app.state.credentials_cipher = credentials_cipher + # Enqueue side of the scheduled-run notification outbox (issue #4254). # It only needs the durable table, so it is wired with the scheduler; # the delivery worker starts after the channel service, further down. notification_connection_repo = scheduled_notification_repo = None try: - notification_connection_repo, scheduled_notification_repo = _scheduled_task_notification_repos(startup_config) + notification_connection_repo, scheduled_notification_repo = _scheduled_task_notification_repos(startup_config, credentials_cipher) except Exception: logger.exception("Failed to prepare the scheduled-task notification outbox") @@ -736,6 +764,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: channel_service = await start_channel_service( startup_config, get_stream_bridge=lambda: getattr(app.state, "stream_bridge", None), + credentials_cipher=credentials_cipher, ) logger.info("Channel service started: %s", channel_service.get_status()) except Exception: diff --git a/backend/app/gateway/auth/config.py b/backend/app/gateway/auth/config.py index ddd002df54..c394b39f3f 100644 --- a/backend/app/gateway/auth/config.py +++ b/backend/app/gateway/auth/config.py @@ -33,29 +33,22 @@ _auth_config: AuthConfig | None = None def _load_or_create_secret() -> str: - """Load persisted JWT secret from ``{base_dir}/.jwt_secret``, or generate and persist a new one.""" + """Load persisted JWT secret from ``{base_dir}/.jwt_secret``, or generate and persist a new one. + + The file is created exclusively and read back, so processes that share the + runtime home and cold-start together (uvicorn workers, replicas on a shared + volume) converge on one secret instead of each keeping its own. + """ from deerflow.config.paths import get_paths + from deerflow.config.secret_file import read_or_create_secret_file paths = get_paths() secret_file = paths.base_dir / _SECRET_FILE try: - if secret_file.exists(): - secret = secret_file.read_text(encoding="utf-8").strip() - if secret: - return secret + return read_or_create_secret_file(secret_file, lambda: secrets.token_urlsafe(32)) except OSError as exc: - raise RuntimeError(f"Failed to read JWT secret from {secret_file}. Set AUTH_JWT_SECRET explicitly or fix DEER_FLOW_HOME/base directory permissions so DeerFlow can read its persisted auth secret.") from exc - - secret = secrets.token_urlsafe(32) - try: - secret_file.parent.mkdir(parents=True, exist_ok=True) - fd = os.open(secret_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) - with os.fdopen(fd, "w", encoding="utf-8") as fh: - fh.write(secret) - except OSError as exc: - raise RuntimeError(f"Failed to persist JWT secret to {secret_file}. Set AUTH_JWT_SECRET explicitly or fix DEER_FLOW_HOME/base directory permissions so DeerFlow can store a stable auth secret.") from exc - return secret + raise RuntimeError(f"Failed to read or persist the JWT secret at {secret_file}. Set AUTH_JWT_SECRET explicitly or fix DEER_FLOW_HOME/base directory permissions so DeerFlow can keep a stable auth secret.") from exc def get_auth_config() -> AuthConfig: diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py index 0ae9fd18f9..1451085e65 100644 --- a/backend/app/gateway/deps.py +++ b/backend/app/gateway/deps.py @@ -31,6 +31,7 @@ from langgraph.types import Checkpointer from app.gateway.auth.errors import AuthErrorCode from deerflow.community.browser_automation.session import browser_multi_worker_error from deerflow.config.app_config import AppConfig, get_app_config +from deerflow.config.credentials_key import CREDENTIALS_KEY_ENV_VAR, CREDENTIALS_KEY_FILENAME, GENERATE_KEY_COMMAND, CredentialsKeyError, parse_credentials_keys from deerflow.config.deployment_config import multi_instance_declaration from deerflow.persistence.feedback import FeedbackRepository from deerflow.runtime import ORPHAN_RECOVERY_STOP_REASON, STARTUP_ORPHAN_RECOVERY_ERROR, RunContext, RunManager, StreamBridge @@ -228,6 +229,56 @@ def _enforce_postgres_for_multi_worker(config: AppConfig) -> None: ) +def credentials_key_consumers(config: AppConfig) -> list[str]: + """Return the enabled features that store data under ``DEER_FLOW_CREDENTIALS_KEY``. + + Each entry is the setting that enabled the feature, as refusal text spells + it. The Gateway loads the key only when this list is non-empty, and the + multi-instance gate below refuses a missing key only for these features, so + a deployment that stores no credentials keeps booting without one. + """ + consumers: list[str] = [] + connections = getattr(config, "channel_connections", None) + if connections is not None and getattr(connections, "enabled", False): + # ChannelConnectionRepository encrypts per-connection credentials. + consumers.append("channel_connections.enabled=true") + return consumers + + +def _enforce_credentials_key(config: AppConfig) -> None: + """Refuse a malformed credentials key, or a missing one where instances cannot share the generated file. + + A set ``DEER_FLOW_CREDENTIALS_KEY`` must parse (any topology): starting with + a key that cannot decrypt anything would silently hide stored credentials. + Without it each instance generates ``{base_dir}/.credentials_key``. Workers + of one process tree share that file and converge on one key through its + exclusive create, so only the explicit multi-instance declaration -- Pods or + hosts that need not share a runtime home -- requires the variable, and only + when a feature from :func:`credentials_key_consumers` is enabled. + """ + raw = os.environ.get(CREDENTIALS_KEY_ENV_VAR, "") + if raw.strip(): + try: + parse_credentials_keys(raw) + except CredentialsKeyError as exc: + raise SystemExit(str(exc)) from None + return + declaration = multi_instance_declaration(config) + if declaration is None: + return + consumers = credentials_key_consumers(config) + if not consumers: + return + enabled = ", ".join(consumers) + disabled = ", ".join(consumer.removesuffix("=true") + "=false" for consumer in consumers) + raise SystemExit( + f"{declaration.knob} with {enabled} requires {CREDENTIALS_KEY_ENV_VAR}: without it every instance generates its own " + f"{{base_dir}}/{CREDENTIALS_KEY_FILENAME}, and credentials one instance stores cannot be decrypted on its peers. " + f"Generate one key with {GENERATE_KEY_COMMAND} and set the same value on every instance (the Helm chart and " + f"scripts/deploy.sh generate and inject it automatically), {declaration.rollback}, or set {disabled}." + ) + + def _validate_agent_storage(config: AppConfig) -> None: """Fail fast on an agent-storage backend the database cannot support. @@ -600,6 +651,9 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen # SQLite write-locks cannot support concurrent multi-process access. # ------------------------------------------------------------------ _enforce_postgres_for_multi_worker(startup_config) + # Reject a malformed DEER_FLOW_CREDENTIALS_KEY, or a missing one where + # declared instances store credentials but cannot share the generated file. + _enforce_credentials_key(startup_config) # Reject agent_storage.backend='db' on a non-durable database, and warn on # node-divergent file storage under multi-worker Postgres. _validate_agent_storage(startup_config) diff --git a/backend/app/gateway/routers/channel_connections.py b/backend/app/gateway/routers/channel_connections.py index e8c06e1560..1f6682f81f 100644 --- a/backend/app/gateway/routers/channel_connections.py +++ b/backend/app/gateway/routers/channel_connections.py @@ -229,7 +229,8 @@ def _get_repository(request: Request, config: ChannelConnectionsConfig) -> Chann if sf is None: raise HTTPException(status_code=503, detail="Channel connection persistence is not available") - repo = ChannelConnectionRepository(sf) + # The lifespan loads the DEER_FLOW_CREDENTIALS_KEY cipher off the event loop. + repo = ChannelConnectionRepository(sf, cipher=getattr(request.app.state, "credentials_cipher", None)) request.app.state.channel_connection_repo = repo return repo diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index 1495347eb6..9047e4a811 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -592,8 +592,27 @@ Notes: - IM chat-to-thread bindings are shared through the database too: with `database.backend: sqlite` or `postgres` the `ChannelManager` keeps the `channel_name:chat_id[:topic_id]` → thread mapping of unbound IM conversations in the `channel_thread_bindings` table (migration `0038_channel_thread_bindings`), so a conversation created on one instance continues on the same thread when its next message lands on another. On the first start after upgrading, an existing `{base_dir}/channels/store.json` is imported once — only into an empty table, with `INSERT … ON CONFLICT DO NOTHING` so concurrently starting instances cannot duplicate a binding — and renamed `store.json.migrated`; a populated table leaves the file untouched. `memory` keeps the per-process JSON file. - The declaration also drives the `agent_storage.backend: file` divergence warning, the inbound webhook dedupe warning, and the WeChat QR-login guard, which otherwise only look at the worker count. - With the default DeerMem backend, the declaration also warns when the derived SQLite retrieval index sits inside the shared `storage_path` (the default `{storage_path}/.retrieval`). Set `memory.backend_config.retrieval_index_path` to an instance-local directory (relative values resolve against `storage_path`): SQLite WAL is unsupported on network filesystems, and the index is rebuilt from the Markdown facts at startup and re-synced per user scope after a peer writes. +- A declared multi-instance deployment that enables a feature storing credentials (today `channel_connections.enabled: true`) also needs one shared `DEER_FLOW_CREDENTIALS_KEY`; see [Credentials encryption key](#credentials-encryption-key-deer_flow_credentials_key). - Restart-required: the gate runs once at startup. Restart all Gateway instances together after changing it. +### Credentials encryption key (`DEER_FLOW_CREDENTIALS_KEY`) + +DeerFlow encrypts credentials it stores at rest (today: per-connection IM channel credentials in the `channel_credentials` table) with a deployment key read **only** from the environment — there is no `config.yaml` key, and it is never derived from `AUTH_JWT_SECRET`. + +```bash +# Generate a key (urlsafe base64 of 32 random bytes, a Fernet key): +python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" +export DEER_FLOW_CREDENTIALS_KEY= +``` + +- **Rotation:** the value may be a comma-separated list. The first key encrypts new values; every listed key decrypts. Prepend the new key (`new,old`), restart, and keep the old key listed until the values it encrypted have been rewritten. Stored values carry a `fernet:v2:` prefix; `fernet:v1:` values written by earlier builds stay readable under the same key. +- **Single instance:** when the variable is unset or blank, the Gateway generates `{DEER_FLOW_HOME}/.credentials_key` (mode `0600`) on first use and reuses it. The file is created exclusively, so uvicorn workers sharing the runtime home converge on one key. It is only created when a feature that stores credentials is enabled. +- **Multi-instance:** instances that do not share a runtime home would each generate their own file and could not decrypt what their peers stored. Startup therefore refuses a declared multi-instance deployment (`deployment.multi_instance: true` / `DEER_FLOW_MULTI_INSTANCE=1`) that enables `channel_connections` without the variable. Multiple workers of one process tree (`GATEWAY_WORKERS > 1` without the declaration) share the file and are not refused. Deployments that store no credentials keep booting without a key. +- **Malformed keys** (wrong length, standard rather than urlsafe base64, any bad entry in the list) stop startup with an error that never repeats the key material. +- **Backup:** losing the key makes every stored credential unreadable. Unreadable values are treated as missing (logged as a warning), never as a crash — Slack, for example, falls back to the deployment bot token. +- **Deployment tooling:** the Helm chart generates the key into its app Secret and preserves it across upgrades; `make up` (`scripts/deploy.sh`) honors a shell or `.env` value and otherwise generates and persists `$DEER_FLOW_HOME/.credentials_key` — the same file the Gateway would generate, so whichever side created it first wins. +- The persisted `.jwt_secret` (used when `AUTH_JWT_SECRET` is unset) is created the same exclusive way, so replicas cold-starting on a shared volume no longer keep different session-signing secrets. + ### Agent Storage Custom agent **definitions** (`config.yaml` + `SOUL.md`) are stored per-user on @@ -1555,6 +1574,7 @@ models: - `DEER_FLOW_CONFIG_PATH` - Custom config file path - `DEER_FLOW_EXTENSIONS_CONFIG_PATH` - Custom extensions config file path - `DEER_FLOW_HOME` - Runtime state directory (defaults to `.deer-flow` under the project root) +- `DEER_FLOW_CREDENTIALS_KEY` - At-rest encryption key(s) for stored credentials; comma-separate to rotate. See [Credentials encryption key](#credentials-encryption-key-deer_flow_credentials_key) - `DEER_FLOW_SKILLS_PATH` - Skills directory when `skills.path` is omitted - `GATEWAY_ENABLE_DOCS` - Set to `false` to disable Swagger UI (`/docs`), ReDoc (`/redoc`), and OpenAPI schema (`/openapi.json`) endpoints (default: `true`) diff --git a/backend/docs/IM_CHANNEL_CONNECTIONS.md b/backend/docs/IM_CHANNEL_CONNECTIONS.md index 2f13041858..7b6aaeba8b 100644 --- a/backend/docs/IM_CHANNEL_CONNECTIONS.md +++ b/backend/docs/IM_CHANNEL_CONNECTIONS.md @@ -539,7 +539,7 @@ Connection records live in SQL tables under `deerflow.persistence.channel_connec - `channel_connections`: owner user, provider identity, workspace/guild/team, status, metadata. - `channel_oauth_states`: one-time connect codes and Telegram deep-link state. - `channel_conversations`: connection-scoped IM conversation to DeerFlow thread mapping. -- `channel_credentials`: reserved for future provider-token flows, not used by the local/private binding flow. +- `channel_credentials`: per-connection provider tokens for future provider-token flows, encrypted with `DEER_FLOW_CREDENTIALS_KEY` (see [CONFIGURATION.md](CONFIGURATION.md#credentials-encryption-key-deer_flow_credentials_key)); not written by the local/private binding flow. Incoming messages that resolve to a connection carry `connection_id`, `owner_user_id`, and `workspace_id`. `ChannelManager` uses `owner_user_id` as the DeerFlow run user id and preserves the raw platform user id as `channel_user_id`. @@ -561,8 +561,11 @@ the channel state directory. responses mask password fields, and mutating runtime/channel-worker APIs require an admin user. - Stored per-connection credentials use the `channel_credentials` encryption - path. If stored credential material cannot be decrypted, DeerFlow treats it - as unavailable instead of using corrupt secrets. + path, keyed by `DEER_FLOW_CREDENTIALS_KEY` (or the generated + `{DEER_FLOW_HOME}/.credentials_key` on a single instance). If stored + credential material cannot be decrypted — a lost or changed key — DeerFlow + treats it as unavailable instead of using corrupt secrets. A declared + multi-instance deployment with channel connections enabled must set the key. - The local plaintext runtime credential fallback is documented above; prefer deployment-managed environment/config secrets for non-local deployments until a dedicated secret backend is configured. diff --git a/backend/packages/harness/deerflow/config/AGENTS.md b/backend/packages/harness/deerflow/config/AGENTS.md index d9b63a7b5c..9e9b4316a5 100644 --- a/backend/packages/harness/deerflow/config/AGENTS.md +++ b/backend/packages/harness/deerflow/config/AGENTS.md @@ -124,6 +124,26 @@ cache, including AppConfig's middleware snapshot, until a readable revision retu **Shared Reset Markers** (`shared_reset_marker.py`): caches derived from the extensions config can go stale without a byte of it changing (a remote MCP server's `tools/list`, a skill installed or edited on the shared volume by another process). `SharedResetMarker(suffix)` owns one hidden JSON file beside the resolved config, `...json`; `publish()` replaces it atomically with a random generation under `extensions_config_write_lock` + `extensions_config_file_lock` (`publish_locked()` when the caller already holds both), chaining `previous_generation` and an optional `user_id`. `SharedResetMarkerTracker.poll()` re-reads the marker at most once per second (monotonic clock, never blocking behind a concurrent poller), adopts the current state silently on its first poll, and reports a `SharedResetChange`: `user_ids={user}` only when exactly one publication carrying that user happened since the last poll, otherwise `None` (retire everything), including deleted/unreadable markers and config-path switches. `note_own_publication()` lets the writer skip the redundant self-invalidation. `resolve_shared_config_path()` maps a missing explicit path to `None` (no shared directory: callers answer `scope=process`). Consumers: `mcp/cache.py` (`mcp-cache-reset`) and `agents/lead_agent/prompt.py` (`skills-cache-reset`). Tests: `tests/test_shared_reset_marker.py`. +**Credentials key** (`credentials_key.py`): `DEER_FLOW_CREDENTIALS_KEY` is env-only +(no `config.yaml` key, never derived from `AUTH_JWT_SECRET`): comma-separated +Fernet keys, first encrypts, all decrypt (`MultiFernet`). `get_credentials_cipher()` +is the process-wide entry point for consumers that store secrets at rest; it +returns a `CredentialsCipher` (`encrypt_text`/`decrypt_text`/`rotate_text`, +`fernet:v2:` prefix, legacy `fernet:v1:` readable). `decrypt_text` raises +`CredentialsDecryptError` (an `InvalidToken`); callers treat that as missing data +and never crash. Unset, it generates `{base_dir}/.credentials_key` -- file I/O, +so async callers use `asyncio.to_thread`. Error text never carries key material. +A new consumer must also join `app.gateway.deps.credentials_key_consumers`, which +drives both Gateway loading and the multi-instance refusal. + +**Secret files** (`secret_file.py`): `read_or_create_secret_file` is the one way +to create a shared secret file (`.credentials_key`, `.jwt_secret`, the managed +model key): write a `0600` temp file, then hard-link it exclusively and read +back, so the name is never empty or partial and concurrent creators converge. +No hard links (SMB) or an abandoned empty file (older releases): single-winner +`.replacing` claim plus rename. Never overwrites a non-empty invalid file. +Tests: `tests/test_secret_file.py`, `tests/test_credentials_key.py`. + ### Config Schema **`config.yaml`** key sections: diff --git a/backend/packages/harness/deerflow/config/credentials_key.py b/backend/packages/harness/deerflow/config/credentials_key.py new file mode 100644 index 0000000000..a602559b0a --- /dev/null +++ b/backend/packages/harness/deerflow/config/credentials_key.py @@ -0,0 +1,165 @@ +"""At-rest encryption key for stored credentials (``DEER_FLOW_CREDENTIALS_KEY``). + +The key is environment-only: one or more comma-separated Fernet keys +(urlsafe base64 of 32 random bytes). The first key encrypts and every key +decrypts, so rotation is "prepend a new key, re-encrypt with +:meth:`CredentialsCipher.rotate_text`, then drop the old key". It is never +derived from ``AUTH_JWT_SECRET`` or any other secret. + +Without the variable a single instance generates ``{base_dir}/.credentials_key`` +once (exclusive create, mode ``0600``), so uvicorn workers sharing the runtime +home converge on one key. Instances that do not share a runtime home cannot +rely on that file; the Gateway refuses a declared multi-instance deployment +whose enabled features store credentials without the variable +(``app.gateway.deps._enforce_credentials_key``). + +Losing the key makes every value encrypted with it unreadable; consumers treat +such values as missing rather than failing. + +Ciphertext format: ``fernet:v2:``. ``fernet:v1:`` (and bare +tokens) were written by the earlier ``ChannelCredentialCipher``; they are still +decrypted with the configured keys. +""" + +from __future__ import annotations + +import logging +import os +import re +import threading +from collections.abc import Sequence +from pathlib import Path + +from cryptography.fernet import Fernet, InvalidToken, MultiFernet + +from deerflow.config.secret_file import InvalidSecretFileError, read_or_create_secret_file + +logger = logging.getLogger(__name__) + +CREDENTIALS_KEY_ENV_VAR = "DEER_FLOW_CREDENTIALS_KEY" +CREDENTIALS_KEY_FILENAME = ".credentials_key" +CIPHERTEXT_PREFIX = "fernet:v2:" +LEGACY_CIPHERTEXT_PREFIX = "fernet:v1:" +GENERATE_KEY_COMMAND = 'python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"' + +# urlsafe base64 of exactly 32 bytes: 43 alphabet characters plus one "=" pad. +_FERNET_KEY_PATTERN = re.compile(r"[A-Za-z0-9_-]{43}=") +_KEY_FILE_SETTLE_SECONDS = 1.0 + + +class CredentialsKeyError(ValueError): + """The configured key material is unusable. The message never contains key material.""" + + +class CredentialsDecryptError(InvalidToken): + """A stored value cannot be decrypted with any configured key (wrong or lost key, or corrupt data).""" + + +def parse_credentials_keys(raw: str, *, source: str = CREDENTIALS_KEY_ENV_VAR) -> list[str]: + """Split and validate a comma-separated key list; blank entries are ignored. + + ``source`` names where the value came from in error messages. + """ + keys = [part.strip() for part in raw.split(",") if part.strip()] + if not keys: + raise CredentialsKeyError(f"{source} holds no key. Generate one with: {GENERATE_KEY_COMMAND}") + for index, key in enumerate(keys, start=1): + if not _is_fernet_key(key): + position = f"entry {index} of {len(keys)}" if len(keys) > 1 else "the value" + raise CredentialsKeyError(f"{source}: {position} is not a Fernet key (urlsafe base64 of 32 bytes, 44 characters). Generate one with: {GENERATE_KEY_COMMAND}") + return keys + + +def _is_fernet_key(key: str) -> bool: + if not _FERNET_KEY_PATTERN.fullmatch(key): + return False + try: + Fernet(key) + except (TypeError, ValueError): + return False + return True + + +class CredentialsCipher: + """Text encryption under the deployment credentials key(s). + + Built on :class:`cryptography.fernet.MultiFernet`: the first key encrypts, + every key decrypts. Values carry the ``fernet:v2:`` prefix. + """ + + def __init__(self, fernets: Sequence[Fernet]) -> None: + if not fernets: + raise CredentialsKeyError("at least one credentials key is required") + self._multi = MultiFernet(list(fernets)) + + @classmethod + def from_keys(cls, keys: Sequence[str | bytes]) -> CredentialsCipher: + return cls([Fernet(key) for key in keys]) + + def encrypt_text(self, value: str) -> str: + return CIPHERTEXT_PREFIX + self._multi.encrypt(value.encode("utf-8")).decode("ascii") + + def decrypt_text(self, value: str) -> str: + """Decrypt a ``fernet:v2:`` value, or a legacy ``fernet:v1:`` / bare token. + + Raises :class:`CredentialsDecryptError` (an ``InvalidToken``) when no + configured key can decrypt it. + """ + token = value.removeprefix(CIPHERTEXT_PREFIX) if value.startswith(CIPHERTEXT_PREFIX) else value.removeprefix(LEGACY_CIPHERTEXT_PREFIX) + try: + return self._multi.decrypt(token.encode("ascii")).decode("utf-8") + except (InvalidToken, UnicodeError): + raise CredentialsDecryptError("stored value cannot be decrypted with the configured credentials key(s)") from None + + def rotate_text(self, value: str) -> str: + """Re-encrypt ``value`` under the primary (first) key, for re-encryption after rotation.""" + return self.encrypt_text(self.decrypt_text(value)) + + +def load_credentials_cipher(*, base_dir: Path | None = None) -> CredentialsCipher: + """Build the deployment cipher from ``DEER_FLOW_CREDENTIALS_KEY``, else the generated key file. + + Performs file I/O when the variable is unset; async callers must run it in + a worker thread. Raises :class:`CredentialsKeyError` for malformed key + material in either source. + """ + raw = os.environ.get(CREDENTIALS_KEY_ENV_VAR, "") + if raw.strip(): + return CredentialsCipher.from_keys(parse_credentials_keys(raw)) + + if base_dir is None: + from deerflow.config.paths import get_paths + + base_dir = get_paths().base_dir + key_file = Path(base_dir) / CREDENTIALS_KEY_FILENAME + try: + value = read_or_create_secret_file( + key_file, + lambda: Fernet.generate_key().decode("ascii"), + validate=lambda candidate: parse_credentials_keys(candidate, source=str(key_file)), + settle_seconds=_KEY_FILE_SETTLE_SECONDS, + ) + except InvalidSecretFileError: + raise CredentialsKeyError(f"{key_file} does not hold a valid credentials key. Restore it from a backup (stored credentials are unreadable without it), or set {CREDENTIALS_KEY_ENV_VAR}.") from None + logger.info("%s is not set; using the credentials key in %s. Back this file up: losing it makes stored credentials unreadable.", CREDENTIALS_KEY_ENV_VAR, key_file) + return CredentialsCipher.from_keys(parse_credentials_keys(value, source=str(key_file))) + + +_cipher: CredentialsCipher | None = None +_cipher_lock = threading.Lock() + + +def get_credentials_cipher() -> CredentialsCipher: + """Return the process-wide cipher, loading it on first use (may perform file I/O).""" + global _cipher + with _cipher_lock: + if _cipher is None: + _cipher = load_credentials_cipher() + return _cipher + + +def reset_credentials_cipher() -> None: + """Forget the process-wide cipher so the next call reloads it (tests, key rotation on restart).""" + global _cipher + with _cipher_lock: + _cipher = None diff --git a/backend/packages/harness/deerflow/config/managed_models.py b/backend/packages/harness/deerflow/config/managed_models.py index acdebf6845..cfa01479ef 100644 --- a/backend/packages/harness/deerflow/config/managed_models.py +++ b/backend/packages/harness/deerflow/config/managed_models.py @@ -22,6 +22,7 @@ from deerflow.config.file_signature import get_config_signature from deerflow.config.managed_model_providers import resolve_managed_model_provider from deerflow.config.model_config import ModelConfig from deerflow.config.runtime_paths import runtime_home +from deerflow.config.secret_file import read_or_create_secret_file if TYPE_CHECKING: from deerflow.config.app_config import AppConfig @@ -81,10 +82,14 @@ class ManagedModelStore: def _cipher(self, *, create: bool = False): from cryptography.fernet import Fernet - if not self.key_path.exists(): + # An empty key file is an older release's interrupted first save: it + # encrypted nothing, so it is recoverable exactly when no catalog exists. + if not self.key_path.exists() or not self.key_path.read_bytes().strip(): if not create or self.path.exists(): raise ValueError("Managed model encryption key is missing; restore it from backup") - self._write(self.key_path, Fernet.generate_key()) + # Exclusive create (replacing an abandoned empty file): a peer sharing + # the runtime home that published first keeps its key. + read_or_create_secret_file(self.key_path, lambda: Fernet.generate_key().decode("ascii")) return Fernet(self.key_path.read_bytes()) def list(self) -> list[ManagedModel]: diff --git a/backend/packages/harness/deerflow/config/secret_file.py b/backend/packages/harness/deerflow/config/secret_file.py new file mode 100644 index 0000000000..aed283d4e4 --- /dev/null +++ b/backend/packages/harness/deerflow/config/secret_file.py @@ -0,0 +1,185 @@ +"""Exclusive-create helper for small secret files shared by concurrent processes. + +Gateway processes that share a runtime home -- uvicorn workers of one host, or +replicas on a shared volume -- cold-start together and may all find a secret +file missing. A read-then-truncating write lets each keep a different value, +so ``read_or_create_secret_file`` publishes exclusively instead: exactly one +creator wins and every other process reads the winner's value back. + +A creator writes and syncs the whole value into a private temporary file first +and then hard-links it to the final name, which fails if the name exists. The +name therefore never appears empty or half-written, however long a creator +stalls, and a creator that fails only ever removes its own temporary file. +Filesystems without hard links (SMB shares) publish through the replacement +claim described below instead. + +An empty or incomplete file can still come from an older release (which +created the name before writing it) or from external tampering. The helper +re-reads it for a bounded settle window; a non-empty file that never validates +is refused and left untouched, because it may be the only copy of a key that +protects stored data. An empty one is abandoned -- nothing can have been +signed or encrypted with an empty secret -- and is replaced. A rename is +last-writer-wins, so replacement is single-winner: the process that exclusively +creates ``.replacing`` re-checks the name under that claim, writes the +new value into the claim and renames it over the name, which publishes the +value and releases the claim in one step; every other process waits for that +value, for a bounded time. +""" + +from __future__ import annotations + +import errno +import logging +import os +import tempfile +import time +from collections.abc import Callable +from pathlib import Path + +logger = logging.getLogger(__name__) + +_POLL_INTERVAL_SECONDS = 0.05 +# How long to wait on a peer's replacement claim before declaring it abandoned. +_CLAIM_WAIT_SECONDS = 10.0 +# os.link() errors meaning the filesystem cannot hard-link at all. +_NO_HARD_LINK_ERRNOS = frozenset({errno.EPERM, errno.EOPNOTSUPP, errno.ENOTSUP, errno.ENOSYS}) + + +class InvalidSecretFileError(ValueError): + """The secret file holds content that never validated; its content is never included.""" + + +def read_or_create_secret_file( + path: Path, + generate: Callable[[], str], + *, + validate: Callable[[str], object] | None = None, + settle_seconds: float = 1.0, +) -> str: + """Return the secret stored at ``path``, creating it exclusively (mode ``0600``) when absent. + + ``generate`` is called only when this process attempts the create, and its + value is returned only when this process wins; otherwise the value another + process published is returned. ``validate`` raises ``ValueError`` for an + incomplete or malformed value (whitespace is stripped before both checks). + ``OSError`` from the filesystem propagates. + """ + path = Path(path) + deadline: float | None = None + claim_deadline: float | None = None + while True: + value = _read(path) + if value is None: + if _create_exclusive(path, generate): + # Read back what we published, or what a peer published first. + continue + elif value and _is_valid(value, validate): + return value + else: + if deadline is None: + deadline = time.monotonic() + settle_seconds + if time.monotonic() < deadline: + time.sleep(_POLL_INTERVAL_SECONDS) + continue + if value: + raise InvalidSecretFileError(f"{path} does not hold a valid secret; restore it from a backup or remove it") + if _publish_under_claim(path, generate, replace_empty=True): + continue + # A peer holds the replacement claim: wait for its value, bounded so a + # replacer that crashed mid-claim cannot block startup forever. + if claim_deadline is None: + claim_deadline = time.monotonic() + max(_CLAIM_WAIT_SECONDS, settle_seconds) + if time.monotonic() >= claim_deadline: + raise InvalidSecretFileError(f"{path} is missing or empty and {_claim_path(path)} was left by an interrupted creation; remove both so a new secret can be generated") + time.sleep(_POLL_INTERVAL_SECONDS) + + +def _read(path: Path) -> str | None: + try: + return path.read_text(encoding="utf-8").strip() + except FileNotFoundError: + return None + + +def _is_valid(value: str, validate: Callable[[str], object] | None) -> bool: + if validate is None: + return True + try: + validate(value) + except ValueError: + return False + return True + + +def _write_all(fd: int, data: bytes) -> None: + view = memoryview(data) + while view: + written = os.write(fd, view) + view = view[written:] + os.fsync(fd) + + +def _create_exclusive(path: Path, generate: Callable[[], str]) -> bool: + """Publish a generated value at the missing ``path`` without ever exposing it partially written. + + Returns ``False`` only when the filesystem cannot hard-link and a peer holds + the replacement claim; otherwise the caller re-reads ``path``. + """ + path.parent.mkdir(parents=True, exist_ok=True) + fd, temporary = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp") + try: + try: + _write_all(fd, generate().encode("utf-8")) + finally: + os.close(fd) + try: + os.link(temporary, path) + except FileExistsError: + pass # A peer published first. + except OSError as exc: + if exc.errno not in _NO_HARD_LINK_ERRNOS: + raise + logger.debug("%s cannot hard-link (%s); publishing through the replacement claim", path.parent, exc) + return _publish_under_claim(path, generate, replace_empty=False) + return True + finally: + Path(temporary).unlink(missing_ok=True) + + +def _claim_path(path: Path) -> Path: + return path.with_name(f"{path.name}.replacing") + + +def _publish_under_claim(path: Path, generate: Callable[[], str], *, replace_empty: bool) -> bool: + """Publish a generated value at ``path`` while holding the exclusive ``.replacing`` claim. + + ``replace_empty`` replaces an abandoned empty file; otherwise only a missing + name is filled (creation on filesystems without hard links). Returns + ``False`` only when a peer holds the claim. Holding it excludes every other + claimant until our rename publishes the value (and removes the claim), and + ``path`` is re-read under the claim, so a process that decided before a + peer's value landed backs off instead of overwriting a value peers may + already have returned. A failure removes only the claim we created. + """ + claim = _claim_path(path) + try: + fd = os.open(claim, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0), 0o600) + except FileExistsError: + return False + published = False + try: + try: + if _read(path) != ("" if replace_empty else None): + # A peer's value landed (or the name changed state) before our claim. + return True + if replace_empty: + logger.warning("Replacing the empty secret file %s left by an interrupted creation", path) + _write_all(fd, generate().encode("utf-8")) + finally: + os.close(fd) + os.replace(claim, path) + published = True + return True + finally: + if not published: + claim.unlink(missing_ok=True) diff --git a/backend/packages/harness/deerflow/persistence/channel_connections/sql.py b/backend/packages/harness/deerflow/persistence/channel_connections/sql.py index 9483e2982b..76a69be17d 100644 --- a/backend/packages/harness/deerflow/persistence/channel_connections/sql.py +++ b/backend/packages/harness/deerflow/persistence/channel_connections/sql.py @@ -15,6 +15,7 @@ from sqlalchemy import delete, func, select, text, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker +from deerflow.config.credentials_key import CredentialsCipher from deerflow.persistence.channel_connections.model import ( ChannelConnectionRow, ChannelConversationRow, @@ -32,27 +33,29 @@ logger = logging.getLogger(__name__) _UPSERT_MAX_ATTEMPTS = 3 -class ChannelCredentialCipher: - """Encrypts provider credentials before they are persisted.""" +class ChannelCredentialCipher(CredentialsCipher): + """Historical constructors for the channel-credentials cipher. + + Production passes the deployment cipher from + ``deerflow.config.credentials_key`` (``DEER_FLOW_CREDENTIALS_KEY``); this + subclass only keeps the earlier API: a single ``Fernet`` and ``from_key``'s + SHA-256 passphrase derivation. New values carry ``fernet:v2:``; the + ``fernet:v1:`` values it wrote before remain readable under the same key. + """ def __init__(self, fernet: Fernet) -> None: - self._fernet = fernet + super().__init__([fernet]) @classmethod def from_key(cls, key: str) -> ChannelCredentialCipher: digest = hashlib.sha256(key.encode("utf-8")).digest() return cls(Fernet(base64.urlsafe_b64encode(digest))) - def encrypt_text(self, value: str | None) -> str | None: - if value is None: - return None - return "fernet:v1:" + self._fernet.encrypt(value.encode("utf-8")).decode("ascii") + def encrypt_text(self, value: str | None) -> str | None: # type: ignore[override] # historical None passthrough + return None if value is None else super().encrypt_text(value) - def decrypt_text(self, value: str | None) -> str | None: - if value is None: - return None - token = value.removeprefix("fernet:v1:") - return self._fernet.decrypt(token.encode("ascii")).decode("utf-8") + def decrypt_text(self, value: str | None) -> str | None: # type: ignore[override] # historical None passthrough + return None if value is None else super().decrypt_text(value) class ChannelConnectionRepository: @@ -62,7 +65,7 @@ class ChannelConnectionRepository: self, session_factory: async_sessionmaker[AsyncSession], *, - cipher: ChannelCredentialCipher | None = None, + cipher: CredentialsCipher | None = None, ) -> None: self.session_factory = session_factory self._cipher = cipher @@ -93,6 +96,11 @@ class ChannelConnectionRepository: raise RuntimeError("channel connection encryption key is required") return self._cipher.encrypt_text(value) + def _decrypt_optional_secret(self, value: str | None) -> str | None: + if value is None or self._cipher is None: + return None + return self._cipher.decrypt_text(value) + @staticmethod def _connection_to_dict(row: ChannelConnectionRow) -> dict[str, Any]: data = row.to_dict() @@ -245,12 +253,12 @@ class ChannelConnectionRepository: if row is None: row = ChannelCredentialRow(connection_id=connection_id) session.add(row) - row.encrypted_access_token = self._cipher.encrypt_text(access_token) - row.encrypted_refresh_token = self._cipher.encrypt_text(refresh_token) + row.encrypted_access_token = self._encrypt_optional_secret(access_token) + row.encrypted_refresh_token = self._encrypt_optional_secret(refresh_token) row.token_type = token_type row.expires_at = expires_at row.refresh_expires_at = refresh_expires_at - row.encrypted_extra_json = self._cipher.encrypt_text(json.dumps(extra or {}, ensure_ascii=False)) + row.encrypted_extra_json = self._encrypt_optional_secret(json.dumps(extra or {}, ensure_ascii=False)) row.version = (row.version or 0) + 1 await session.commit() @@ -262,11 +270,11 @@ class ChannelConnectionRepository: if row is None: return None try: - extra_raw = self._cipher.decrypt_text(row.encrypted_extra_json) + extra_raw = self._decrypt_optional_secret(row.encrypted_extra_json) return { "connection_id": row.connection_id, - "access_token": self._cipher.decrypt_text(row.encrypted_access_token), - "refresh_token": self._cipher.decrypt_text(row.encrypted_refresh_token), + "access_token": self._decrypt_optional_secret(row.encrypted_access_token), + "refresh_token": self._decrypt_optional_secret(row.encrypted_refresh_token), "token_type": row.token_type, "expires_at": self._coerce_datetime(row.expires_at), "refresh_expires_at": self._coerce_datetime(row.refresh_expires_at), diff --git a/backend/tests/blocking_io/test_credentials_key_offloop.py b/backend/tests/blocking_io/test_credentials_key_offloop.py new file mode 100644 index 0000000000..cc711aa6c9 --- /dev/null +++ b/backend/tests/blocking_io/test_credentials_key_offloop.py @@ -0,0 +1,36 @@ +"""Regression anchor: the Gateway loads the credentials key off the event loop. + +``app.gateway.app._load_credentials_cipher`` runs inside the lifespan. Without +``DEER_FLOW_CREDENTIALS_KEY`` it creates ``{base_dir}/.credentials_key`` (mkdir, +exclusive create, fsync, read-back), all of which must go through +``asyncio.to_thread``; the strict Blockbuster gate fails this test otherwise. +""" + +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +import pytest + +# Imported at collection time: importing the Gateway app reads package +# metadata, which is not the startup path under test. +from app.gateway.app import _load_credentials_cipher +from deerflow.config.credentials_key import CREDENTIALS_KEY_ENV_VAR, CREDENTIALS_KEY_FILENAME, reset_credentials_cipher +from deerflow.config.paths import Paths + +pytestmark = pytest.mark.asyncio + + +async def test_credentials_key_generation_does_not_block_the_event_loop(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + home = tmp_path / "missing" / "home" + monkeypatch.delenv(CREDENTIALS_KEY_ENV_VAR, raising=False) + monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=home)) + reset_credentials_cipher() + try: + cipher = await _load_credentials_cipher(SimpleNamespace(channel_connections=SimpleNamespace(enabled=True))) + finally: + reset_credentials_cipher() + + assert cipher is not None + assert (home / CREDENTIALS_KEY_FILENAME).exists() diff --git a/backend/tests/test_auth_config.py b/backend/tests/test_auth_config.py index 61d1d7d2e0..ecc4f79641 100644 --- a/backend/tests/test_auth_config.py +++ b/backend/tests/test_auth_config.py @@ -88,3 +88,65 @@ def test_auth_config_empty_secret_file_generates_new(tmp_path): assert (tmp_path / ".jwt_secret").read_text().strip() == config.jwt_secret finally: cfg._auth_config = old + + +def test_jwt_secret_creation_keeps_a_peer_replica_value(tmp_path, monkeypatch): + """Two replicas cold-starting on a shared home volume must sign with one key. + + Deterministic interleaving: the peer creates ``.jwt_secret`` after this + process found none but before it writes. A truncating write would replace + the peer's value and both replicas would keep different secrets. + """ + from deerflow.config.paths import Paths + + secret_file = tmp_path / ".jwt_secret" + + def peer_wins_the_race(_nbytes): + secret_file.write_text("peer-replica-secret-value-0123456789", encoding="utf-8") + return "this-replica-secret-value-0123456789" + + monkeypatch.setattr(cfg.secrets, "token_urlsafe", peer_wins_the_race) + with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=tmp_path)): + secret = cfg._load_or_create_secret() + + assert secret == "peer-replica-secret-value-0123456789" + assert secret_file.read_text(encoding="utf-8") == "peer-replica-secret-value-0123456789" + + +def test_concurrent_jwt_secret_creators_converge(tmp_path): + import threading + + from deerflow.config.paths import Paths + + workers = 8 + barrier = threading.Barrier(workers) + results: list[str] = [] + lock = threading.Lock() + + def run(): + barrier.wait(timeout=10) + value = cfg._load_or_create_secret() + with lock: + results.append(value) + + with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=tmp_path)): + threads = [threading.Thread(target=run) for _ in range(workers)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=30) + + assert len(results) == workers + assert len(set(results)) == 1 + assert (tmp_path / ".jwt_secret").read_text(encoding="utf-8") == results[0] + if os.name == "posix": + assert (tmp_path / ".jwt_secret").stat().st_mode & 0o777 == 0o600 + + +def test_unusable_jwt_secret_location_keeps_the_actionable_error(tmp_path): + from deerflow.config.paths import Paths + + not_a_directory = tmp_path / "home" + not_a_directory.write_text("", encoding="utf-8") + with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=not_a_directory)), pytest.raises(RuntimeError, match="AUTH_JWT_SECRET"): + cfg._load_or_create_secret() diff --git a/backend/tests/test_channel_credentials_wiring.py b/backend/tests/test_channel_credentials_wiring.py new file mode 100644 index 0000000000..7da05f6fca --- /dev/null +++ b/backend/tests/test_channel_credentials_wiring.py @@ -0,0 +1,261 @@ +"""Per-connection channel credentials use the deployment credentials key. + +Before ``DEER_FLOW_CREDENTIALS_KEY`` no production construction site passed a +cipher to ``ChannelConnectionRepository``: ``store_credentials`` refused +(``RuntimeError``) and ``get_credentials`` always answered ``None``, so Slack +replies always fell back to the deployment bot token. The Gateway now loads +the cipher once at startup (off the event loop) for the features that need it +and hands it to the channel service, the scheduled-task notification outbox +and the channel-connections router. +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest +from cryptography.fernet import Fernet + +from app.channels.message_bus import MessageBus, OutboundMessage +from deerflow.config.credentials_key import ( + CIPHERTEXT_PREFIX, + CREDENTIALS_KEY_ENV_VAR, + CREDENTIALS_KEY_FILENAME, + CredentialsCipher, + load_credentials_cipher, + reset_credentials_cipher, +) +from deerflow.config.paths import Paths + +KEY_OLD = Fernet.generate_key().decode("ascii") +KEY_NEW = Fernet.generate_key().decode("ascii") + + +@pytest.fixture(autouse=True) +def _isolated_cipher(monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + monkeypatch.delenv(CREDENTIALS_KEY_ENV_VAR, raising=False) + monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path / "home")) + reset_credentials_cipher() + yield + reset_credentials_cipher() + + +def _startup_config(*, channel_connections: bool) -> SimpleNamespace: + return SimpleNamespace(channel_connections=SimpleNamespace(enabled=channel_connections)) + + +@pytest.fixture +async def session_factory(tmp_path: Path): + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + + await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'channels.db'}", sqlite_dir=str(tmp_path)) + try: + yield get_session_factory() + finally: + await close_engine() + + +async def _connection(repo) -> str: + connection = await repo.upsert_connection(owner_user_id="alice", provider="slack", external_account_id="U-alice", workspace_id="T1") + return connection["id"] + + +# --------------------------------------------------------------------------- +# Startup loading +# --------------------------------------------------------------------------- + + +@pytest.mark.anyio +async def test_no_key_is_loaded_or_generated_when_no_feature_needs_it(tmp_path: Path) -> None: + from app.gateway.app import _load_credentials_cipher + + assert await _load_credentials_cipher(_startup_config(channel_connections=False)) is None + assert not (tmp_path / "home" / CREDENTIALS_KEY_FILENAME).exists() + + +@pytest.mark.anyio +async def test_channel_connections_load_the_generated_key_file(tmp_path: Path) -> None: + from app.gateway.app import _load_credentials_cipher + + cipher = await _load_credentials_cipher(_startup_config(channel_connections=True)) + + assert isinstance(cipher, CredentialsCipher) + key_file = tmp_path / "home" / CREDENTIALS_KEY_FILENAME + assert key_file.exists() + assert CredentialsCipher.from_keys([key_file.read_text(encoding="utf-8").strip()]).decrypt_text(cipher.encrypt_text("x")) == "x" + + +@pytest.mark.anyio +async def test_channel_connections_prefer_the_environment_key(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + from app.gateway.app import _load_credentials_cipher + + monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, KEY_NEW) + + cipher = await _load_credentials_cipher(_startup_config(channel_connections=True)) + + assert CredentialsCipher.from_keys([KEY_NEW]).decrypt_text(cipher.encrypt_text("x")) == "x" + assert not (tmp_path / "home" / CREDENTIALS_KEY_FILENAME).exists() + + +@pytest.mark.anyio +async def test_an_unloadable_key_degrades_to_no_stored_credentials(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None: + """A broken key file must not take the Gateway down: credentials stay unavailable, as before the key existed.""" + from app.gateway import app as gateway_app + + def broken_loader(): + raise OSError("read-only file system") + + monkeypatch.setattr("deerflow.config.credentials_key.get_credentials_cipher", broken_loader) + with caplog.at_level(logging.ERROR, logger=gateway_app.logger.name): + assert await gateway_app._load_credentials_cipher(_startup_config(channel_connections=True)) is None + assert any("credentials" in record.getMessage().lower() for record in caplog.records) + + +# --------------------------------------------------------------------------- +# Construction sites +# --------------------------------------------------------------------------- + + +def test_the_channel_service_repository_gets_the_cipher(monkeypatch: pytest.MonkeyPatch) -> None: + from app.channels.service import _make_connection_repo + + sentinel_factory = MagicMock(name="session_factory") + monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: sentinel_factory) + cipher = CredentialsCipher.from_keys([KEY_NEW]) + + repo = _make_connection_repo(SimpleNamespace(enabled=True), credentials_cipher=cipher) + + assert repo is not None + assert repo.session_factory is sentinel_factory + assert repo._cipher is cipher + assert _make_connection_repo(SimpleNamespace(enabled=True))._cipher is None, "callers that pass no cipher keep the old behaviour" + + +def test_the_notification_outbox_repository_gets_the_cipher(monkeypatch: pytest.MonkeyPatch) -> None: + from app.gateway.app import _scheduled_task_notification_repos + + monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: MagicMock(name="session_factory")) + cipher = CredentialsCipher.from_keys([KEY_NEW]) + + connection_repo, notification_repo = _scheduled_task_notification_repos(_startup_config(channel_connections=True), cipher) + + assert connection_repo._cipher is cipher + assert notification_repo is not None + + +def test_the_router_repository_gets_the_cipher_from_app_state(monkeypatch: pytest.MonkeyPatch) -> None: + from app.gateway.routers import channel_connections + + monkeypatch.setattr(channel_connections, "get_session_factory", lambda: MagicMock(name="session_factory")) + cipher = CredentialsCipher.from_keys([KEY_NEW]) + request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(credentials_cipher=cipher))) + + repo = channel_connections._get_repository(request, MagicMock()) + + assert repo._cipher is cipher + assert request.app.state.channel_connection_repo is repo + + +def test_start_channel_service_threads_the_cipher_through(monkeypatch: pytest.MonkeyPatch) -> None: + import anyio + + import app.channels.service as service_module + + captured: dict[str, object] = {} + + class _FakeService: + async def start(self): + return None + + def fake_from_app_config(app_config=None, *, get_stream_bridge=None, credentials_cipher=None): + captured["credentials_cipher"] = credentials_cipher + return _FakeService() + + cipher = CredentialsCipher.from_keys([KEY_NEW]) + monkeypatch.setattr(service_module.ChannelService, "from_app_config", staticmethod(fake_from_app_config)) + monkeypatch.setattr(service_module, "_channel_service", None) + + anyio.run(lambda: service_module.start_channel_service(credentials_cipher=cipher)) + + assert captured["credentials_cipher"] is cipher + + +# --------------------------------------------------------------------------- +# Repository behaviour under the deployment key +# --------------------------------------------------------------------------- + + +@pytest.mark.anyio +async def test_credentials_round_trip_under_the_deployment_key(session_factory, tmp_path: Path) -> None: + from sqlalchemy import select + + from deerflow.persistence.channel_connections import ChannelConnectionRepository, ChannelCredentialRow + + repo = ChannelConnectionRepository(session_factory, cipher=load_credentials_cipher(base_dir=tmp_path)) + connection_id = await _connection(repo) + + await repo.store_credentials(connection_id, access_token="xoxb-connection-token", extra={"bot_user_id": "B1"}) + + async with session_factory() as session: + row = (await session.execute(select(ChannelCredentialRow))).scalar_one() + assert row.encrypted_access_token.startswith(CIPHERTEXT_PREFIX) + assert "xoxb-connection-token" not in row.encrypted_access_token + + restarted = ChannelConnectionRepository(session_factory, cipher=load_credentials_cipher(base_dir=tmp_path)) + credentials = await restarted.get_credentials(connection_id) + assert credentials is not None + assert credentials["access_token"] == "xoxb-connection-token" + assert credentials["extra"] == {"bot_user_id": "B1"} + + +@pytest.mark.anyio +async def test_rotation_keeps_rows_written_under_the_previous_key_readable(session_factory) -> None: + from deerflow.persistence.channel_connections import ChannelConnectionRepository + + before = ChannelConnectionRepository(session_factory, cipher=CredentialsCipher.from_keys([KEY_OLD])) + connection_id = await _connection(before) + await before.store_credentials(connection_id, access_token="written-before-rotation") + + after = ChannelConnectionRepository(session_factory, cipher=CredentialsCipher.from_keys([KEY_NEW, KEY_OLD])) + + assert (await after.get_credentials(connection_id))["access_token"] == "written-before-rotation" + + +@pytest.mark.anyio +async def test_a_changed_key_treats_stored_credentials_as_missing(session_factory, caplog: pytest.LogCaptureFixture) -> None: + from deerflow.persistence.channel_connections import ChannelConnectionRepository + + before = ChannelConnectionRepository(session_factory, cipher=CredentialsCipher.from_keys([KEY_OLD])) + connection_id = await _connection(before) + await before.store_credentials(connection_id, access_token="unreadable-after-key-loss") + + after = ChannelConnectionRepository(session_factory, cipher=CredentialsCipher.from_keys([KEY_NEW])) + with caplog.at_level(logging.WARNING, logger="deerflow.persistence.channel_connections.sql"): + assert await after.get_credentials(connection_id) is None + assert any("Unable to decrypt channel connection credentials" in record.getMessage() for record in caplog.records) + assert not any("unreadable-after-key-loss" in record.getMessage() for record in caplog.records) + + +@pytest.mark.anyio +async def test_slack_falls_back_to_the_deployment_token_without_stored_credentials(session_factory) -> None: + from app.channels.slack import SlackChannel + from deerflow.persistence.channel_connections import ChannelConnectionRepository + + repo = ChannelConnectionRepository(session_factory, cipher=CredentialsCipher.from_keys([KEY_NEW])) + connection_id = await _connection(repo) + operator_client = MagicMock(name="operator_client") + factory = MagicMock(name="web_client_factory") + channel = SlackChannel(bus=MessageBus(), config={"connection_repo": repo, "web_client_factory": factory}) + channel._web_client = operator_client + message = OutboundMessage(channel_name="slack", chat_id="C1", thread_id="t1", text="hi", connection_id=connection_id) + + assert await channel._get_web_client_for_message(message) is operator_client + factory.assert_not_called() + + await repo.store_credentials(connection_id, access_token="xoxb-connection-token") + connection_client = await channel._get_web_client_for_message(message) + factory.assert_called_once_with(token="xoxb-connection-token") + assert connection_client is factory.return_value diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index 8690c24f3c..16fc950215 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -6513,7 +6513,7 @@ class TestGithubFollowupBuffer: def get_status(self): return {} - def fake_from_app_config(app_config=None, *, get_stream_bridge=None): + def fake_from_app_config(app_config=None, *, get_stream_bridge=None, credentials_cipher=None): captured["get_stream_bridge"] = get_stream_bridge return _FakeService() diff --git a/backend/tests/test_compose_credentials_key.py b/backend/tests/test_compose_credentials_key.py new file mode 100644 index 0000000000..5c143deaed --- /dev/null +++ b/backend/tests/test_compose_credentials_key.py @@ -0,0 +1,104 @@ +"""Both compose stacks pass ``DEER_FLOW_CREDENTIALS_KEY`` through to the Gateway. + +``scripts/deploy.sh`` resolves the key (shell, ``.env``, the persisted +``$DEER_FLOW_HOME/.credentials_key``, or a freshly generated one) and Compose +interpolates it into the gateway service. ``:-`` keeps a stack started without +the deploy script quiet: an empty value makes the Gateway fall back to the same +``.credentials_key`` file under its runtime home, which is the host's +``DEER_FLOW_HOME`` bind mount. +""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +from pathlib import Path + +import pytest +import yaml +from support.compose import DOCKER, requires_docker_compose +from support.shell import find_script_bash + +REPO_ROOT = Path(__file__).resolve().parents[2] +COMPOSE_PATHS = { + "prod": REPO_ROOT / "docker" / "docker-compose.yaml", + "dev": REPO_ROOT / "docker" / "docker-compose-dev.yaml", +} +EXPECTED_ENTRY = "DEER_FLOW_CREDENTIALS_KEY=${DEER_FLOW_CREDENTIALS_KEY:-}" +BASH_EXECUTABLE = find_script_bash() +DOTENV_KEY = "u6Tp1Gk0bq8m6C2YkS9oWw7xH3fLr5eN1aZdQjVtXc4=" +SHELL_KEY = "Zq3Lw8Rn2Tb6Yv1Kc5Hm9Pd4Sx7Ga0Je3Uf6Ni2Ro8M=" + + +@pytest.mark.parametrize("variant", sorted(COMPOSE_PATHS)) +def test_gateway_receives_the_credentials_key(variant: str) -> None: + services = yaml.safe_load(COMPOSE_PATHS[variant].read_text(encoding="utf-8"))["services"] + + assert EXPECTED_ENTRY in services["gateway"]["environment"] + assert "DEER_FLOW_HOME=/app/backend/.deer-flow" in services["gateway"]["environment"], "the fallback key file lives in the runtime home" + + +@pytest.mark.parametrize("variant", sorted(COMPOSE_PATHS)) +def test_only_the_gateway_receives_the_credentials_key(variant: str) -> None: + services = yaml.safe_load(COMPOSE_PATHS[variant].read_text(encoding="utf-8"))["services"] + for name, service in services.items(): + if name == "gateway": + continue + environment = service.get("environment") or [] + entries = environment if isinstance(environment, list) else [f"{key}={value}" for key, value in environment.items()] + assert not any(str(entry).startswith("DEER_FLOW_CREDENTIALS_KEY") for entry in entries), name + + +def _render_through_docker_start(tmp_path: Path, env_file: str, shell_value: str | None) -> dict[str, str]: + """Render the dev gateway environment the way ``make docker-start`` interpolates it. + + ``scripts/docker.sh start`` runs Compose from ``docker/`` without + ``--env-file``; a ``docker`` shell function swaps ``up ...`` for ``config``. + """ + docker_dir = tmp_path / "docker" + shutil.copytree(REPO_ROOT / "docker", docker_dir) + (tmp_path / "frontend").mkdir() + (tmp_path / "frontend" / ".env").write_text("", encoding="utf-8") + (tmp_path / ".env").write_text(env_file, encoding="utf-8") + (tmp_path / "config.yaml").write_text("sandbox:\n use: deerflow.sandbox.local:LocalSandboxProvider\n", encoding="utf-8") + (tmp_path / "extensions_config.json").write_text("{}", encoding="utf-8") + rendered = tmp_path / "rendered.json" + env = {key: value for key, value in os.environ.items() if not key.startswith(("DEER_FLOW_", "COMPOSE_", "AUTH_"))} + if shell_value is not None: + env["DEER_FLOW_CREDENTIALS_KEY"] = shell_value + script = f""" +source '{REPO_ROOT / "scripts" / "docker.sh"}' +PROJECT_ROOT='{tmp_path}' +DOCKER_DIR='{docker_dir}' +require_compose_version() {{ :; }} +docker() {{ + local args=() + for arg in "$@"; do + [ "$arg" = up ] && break + args+=("$arg") + done + command '{DOCKER}' "${{args[@]}}" config --format json > '{rendered}' +}} +start +""" + subprocess.run([BASH_EXECUTABLE, "-c", script], env=env, capture_output=True, text=True, timeout=120, check=True) + return json.loads(rendered.read_text(encoding="utf-8"))["services"]["gateway"]["environment"] + + +@requires_docker_compose +@pytest.mark.skipif(BASH_EXECUTABLE is None, reason="bash is required to run scripts/docker.sh") +@pytest.mark.parametrize( + ("env_file", "shell_value", "expected"), + [ + ("", None, ""), + (f"DEER_FLOW_CREDENTIALS_KEY={DOTENV_KEY}\n", None, DOTENV_KEY), + (f"DEER_FLOW_CREDENTIALS_KEY='{DOTENV_KEY}'\r\n", None, DOTENV_KEY), + (f"DEER_FLOW_CREDENTIALS_KEY={DOTENV_KEY}\n", SHELL_KEY, SHELL_KEY), + ], + ids=["unset", "dotenv", "quoted-crlf", "shell-export-wins"], +) +def test_make_docker_start_keeps_the_dotenv_credentials_key(tmp_path, env_file: str, shell_value: str | None, expected: str): + """The dev launcher must hand the checkout .env key to interpolation; an empty override would replace it.""" + assert _render_through_docker_start(tmp_path, env_file, shell_value)["DEER_FLOW_CREDENTIALS_KEY"] == expected diff --git a/backend/tests/test_credentials_key.py b/backend/tests/test_credentials_key.py new file mode 100644 index 0000000000..81f0b8fd71 --- /dev/null +++ b/backend/tests/test_credentials_key.py @@ -0,0 +1,254 @@ +"""At-rest credentials key ``DEER_FLOW_CREDENTIALS_KEY`` and the cipher built from it. + +The key is env-only: one or more comma-separated urlsafe-base64 Fernet keys, +the first encrypting and every one decrypting, so an operator rotates by +prepending a new key. Without it a single instance generates +``{base_dir}/.credentials_key`` once. It is never derived from +``AUTH_JWT_SECRET``. +""" + +from __future__ import annotations + +import base64 +import hashlib +import os +from pathlib import Path + +import pytest +from cryptography.fernet import Fernet, InvalidToken + +from deerflow.config import credentials_key as ck +from deerflow.config.credentials_key import ( + CIPHERTEXT_PREFIX, + CREDENTIALS_KEY_ENV_VAR, + CREDENTIALS_KEY_FILENAME, + GENERATE_KEY_COMMAND, + CredentialsCipher, + CredentialsDecryptError, + CredentialsKeyError, + get_credentials_cipher, + load_credentials_cipher, + parse_credentials_keys, + reset_credentials_cipher, +) +from deerflow.persistence.channel_connections import ChannelCredentialCipher + +KEY_OLD = Fernet.generate_key().decode("ascii") +KEY_NEW = Fernet.generate_key().decode("ascii") + + +@pytest.fixture(autouse=True) +def _isolated_key_env(monkeypatch: pytest.MonkeyPatch): + monkeypatch.delenv(CREDENTIALS_KEY_ENV_VAR, raising=False) + reset_credentials_cipher() + yield + reset_credentials_cipher() + + +def _std_base64_key() -> str: + """A 32-byte key spelled with the standard alphabet (``+``/``/``), as Helm's randBytes emits it.""" + raw = bytes([0xFB, 0xFF] * 16) + value = base64.b64encode(raw).decode("ascii") + assert "+" in value or "/" in value + return value + + +# --------------------------------------------------------------------------- +# Parsing +# --------------------------------------------------------------------------- + + +def test_parse_accepts_a_single_key() -> None: + assert parse_credentials_keys(KEY_NEW) == [KEY_NEW] + + +def test_parse_keeps_rotation_order_and_ignores_blank_entries() -> None: + assert parse_credentials_keys(f" {KEY_NEW} , {KEY_OLD} ,") == [KEY_NEW, KEY_OLD] + + +@pytest.mark.parametrize( + "raw", + [ + "", + " , ", + "not-a-fernet-key", + KEY_NEW[:-1], + KEY_NEW + "A", + _std_base64_key(), + "é" * 44, + f"{KEY_NEW},short-second-key", + ], + ids=["empty", "only-separators", "garbage", "truncated", "too-long", "standard-alphabet", "non-ascii", "second-entry-bad"], +) +def test_parse_rejects_malformed_keys_without_echoing_them(raw: str) -> None: + with pytest.raises(CredentialsKeyError) as exc_info: + parse_credentials_keys(raw) + + message = str(exc_info.value) + assert CREDENTIALS_KEY_ENV_VAR in message + assert GENERATE_KEY_COMMAND in message, "the refusal must say how to generate a valid key" + for entry in filter(None, (part.strip() for part in raw.split(","))): + assert entry not in message, "key material must never reach logs or refusal text" + assert exc_info.value.__cause__ is None + + +# --------------------------------------------------------------------------- +# Cipher +# --------------------------------------------------------------------------- + + +def test_cipher_writes_a_versioned_prefix_and_round_trips() -> None: + cipher = CredentialsCipher.from_keys([KEY_NEW]) + + token = cipher.encrypt_text("xoxb-secret") + + assert token.startswith(CIPHERTEXT_PREFIX) + assert CIPHERTEXT_PREFIX == "fernet:v2:" + assert "xoxb-secret" not in token + assert cipher.decrypt_text(token) == "xoxb-secret" + + +def test_the_first_key_encrypts_and_every_key_decrypts() -> None: + old_token = CredentialsCipher.from_keys([KEY_OLD]).encrypt_text("written-before-rotation") + rotated = CredentialsCipher.from_keys([KEY_NEW, KEY_OLD]) + + assert rotated.decrypt_text(old_token) == "written-before-rotation" + + new_token = rotated.encrypt_text("written-after-rotation") + assert CredentialsCipher.from_keys([KEY_NEW]).decrypt_text(new_token) == "written-after-rotation" + with pytest.raises(CredentialsDecryptError): + CredentialsCipher.from_keys([KEY_OLD]).decrypt_text(new_token) + + +def test_rotate_re_encrypts_under_the_primary_key() -> None: + old_token = CredentialsCipher.from_keys([KEY_OLD]).encrypt_text("long-lived-token") + rotated = CredentialsCipher.from_keys([KEY_NEW, KEY_OLD]) + + re_encrypted = rotated.rotate_text(old_token) + + assert re_encrypted.startswith(CIPHERTEXT_PREFIX) + assert CredentialsCipher.from_keys([KEY_NEW]).decrypt_text(re_encrypted) == "long-lived-token" + + +def test_an_undecryptable_value_raises_an_invalid_token_subclass() -> None: + token = CredentialsCipher.from_keys([KEY_OLD]).encrypt_text("secret") + cipher = CredentialsCipher.from_keys([KEY_NEW]) + + with pytest.raises(CredentialsDecryptError) as exc_info: + cipher.decrypt_text(token) + assert isinstance(exc_info.value, InvalidToken), "existing `except InvalidToken` call sites keep working" + + for garbage in ("fernet:v2:not-a-token", "fernet:v2:é", "plain text"): + with pytest.raises(CredentialsDecryptError): + cipher.decrypt_text(garbage) + + +def test_legacy_v1_values_written_by_the_passphrase_cipher_stay_readable() -> None: + """Exactly what ``ChannelCredentialCipher.from_key`` wrote before the v2 format.""" + derived = Fernet(base64.urlsafe_b64encode(hashlib.sha256(b"legacy-passphrase").digest())) + legacy_value = "fernet:v1:" + derived.encrypt(b"xoxb-legacy").decode("ascii") + + cipher = ChannelCredentialCipher.from_key("legacy-passphrase") + + assert cipher.decrypt_text(legacy_value) == "xoxb-legacy" + assert cipher.decrypt_text(legacy_value.removeprefix("fernet:v1:")) == "xoxb-legacy", "the old reader accepted a bare token" + assert cipher.encrypt_text("new-value").startswith(CIPHERTEXT_PREFIX) + assert isinstance(cipher, CredentialsCipher) + + +def test_v1_values_under_a_raw_key_are_readable_by_the_deployment_cipher() -> None: + """``ChannelCredentialCipher(Fernet(key))`` wrote ``fernet:v1:`` under a raw key; the same key in the env reads it.""" + legacy_value = "fernet:v1:" + Fernet(KEY_OLD).encrypt(b"value").decode("ascii") + + assert ChannelCredentialCipher(Fernet(KEY_OLD)).decrypt_text(legacy_value) == "value" + assert CredentialsCipher.from_keys([KEY_NEW, KEY_OLD]).decrypt_text(legacy_value) == "value" + + +# --------------------------------------------------------------------------- +# Loading +# --------------------------------------------------------------------------- + + +def test_the_environment_key_list_wins_and_no_file_is_written(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, f"{KEY_NEW},{KEY_OLD}") + old_token = CredentialsCipher.from_keys([KEY_OLD]).encrypt_text("old") + + cipher = load_credentials_cipher(base_dir=tmp_path) + + assert cipher.decrypt_text(old_token) == "old" + assert CredentialsCipher.from_keys([KEY_NEW]).decrypt_text(cipher.encrypt_text("new")) == "new" + assert not (tmp_path / CREDENTIALS_KEY_FILENAME).exists() + + +def test_a_single_instance_generates_the_key_file_once(tmp_path: Path) -> None: + first = load_credentials_cipher(base_dir=tmp_path) + key_file = tmp_path / CREDENTIALS_KEY_FILENAME + + assert key_file.exists() + persisted = key_file.read_text(encoding="utf-8").strip() + assert parse_credentials_keys(persisted) == [persisted] + if os.name == "posix": + assert key_file.stat().st_mode & 0o777 == 0o600 + + token = first.encrypt_text("survives a restart") + assert load_credentials_cipher(base_dir=tmp_path).decrypt_text(token) == "survives a restart" + assert key_file.read_text(encoding="utf-8").strip() == persisted + + +def test_a_blank_environment_value_means_unset(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """Compose renders ``${DEER_FLOW_CREDENTIALS_KEY:-}`` as an empty string.""" + monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, " ") + + load_credentials_cipher(base_dir=tmp_path) + + assert (tmp_path / CREDENTIALS_KEY_FILENAME).exists() + + +def test_a_malformed_environment_key_is_refused_before_touching_disk(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, "definitely-not-a-key") + + with pytest.raises(CredentialsKeyError): + load_credentials_cipher(base_dir=tmp_path) + + assert not (tmp_path / CREDENTIALS_KEY_FILENAME).exists() + + +def test_a_corrupt_key_file_is_refused_and_left_alone(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """The file may be the only copy of the key that encrypted stored credentials.""" + monkeypatch.setattr(ck, "_KEY_FILE_SETTLE_SECONDS", 0.05) + key_file = tmp_path / CREDENTIALS_KEY_FILENAME + key_file.write_text("corrupted-key-material", encoding="utf-8") + + with pytest.raises(CredentialsKeyError) as exc_info: + load_credentials_cipher(base_dir=tmp_path) + + assert str(key_file) in str(exc_info.value) + assert "corrupted-key-material" not in str(exc_info.value) + assert key_file.read_text(encoding="utf-8") == "corrupted-key-material" + + +def test_the_key_is_never_derived_from_the_jwt_secret(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.setenv("AUTH_JWT_SECRET", KEY_OLD) + jwt_derived_token = CredentialsCipher.from_keys([KEY_OLD]).encrypt_text("secret") + + cipher = load_credentials_cipher(base_dir=tmp_path) + + with pytest.raises(CredentialsDecryptError): + cipher.decrypt_text(jwt_derived_token) + assert (tmp_path / CREDENTIALS_KEY_FILENAME).read_text(encoding="utf-8").strip() != KEY_OLD + + +def test_the_process_cipher_is_loaded_once_until_reset(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + from deerflow.config.paths import Paths + + monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path)) + monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, KEY_OLD) + + first = get_credentials_cipher() + monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, KEY_NEW) + assert get_credentials_cipher() is first + + reset_credentials_cipher() + reloaded = get_credentials_cipher() + assert reloaded is not first + assert CredentialsCipher.from_keys([KEY_NEW]).decrypt_text(reloaded.encrypt_text("x")) == "x" diff --git a/backend/tests/test_deploy_dotenv_secrets.py b/backend/tests/test_deploy_dotenv_secrets.py index 9b4cb080de..0ea414e954 100644 --- a/backend/tests/test_deploy_dotenv_secrets.py +++ b/backend/tests/test_deploy_dotenv_secrets.py @@ -31,12 +31,16 @@ REPO_ROOT = Path(__file__).resolve().parents[2] BASH = find_script_bash() pytestmark = pytest.mark.skipif(BASH is None, reason="repo shell-script tests need Git Bash on Windows") -SECRETS = ("BETTER_AUTH_SECRET", "DEER_FLOW_INTERNAL_AUTH_TOKEN") +SECRETS = ("BETTER_AUTH_SECRET", "DEER_FLOW_INTERNAL_AUTH_TOKEN", "DEER_FLOW_CREDENTIALS_KEY") PERSISTED_FILE = { "BETTER_AUTH_SECRET": ".better-auth-secret", "DEER_FLOW_INTERNAL_AUTH_TOKEN": ".internal-auth-token", + # Same name the Gateway itself generates under its runtime home, which is + # this directory's bind mount: one key whichever side created it first. + "DEER_FLOW_CREDENTIALS_KEY": ".credentials_key", } -GENERATED = re.compile(r"set:[A-Za-z0-9_\-]{32,}") +# A Fernet key ends in one "=" of base64 padding. +GENERATED = re.compile(r"set:[A-Za-z0-9_\-]{32,}=?") # The fake docker answers `compose ... config` the way Compose renders the # script's stub project: it reads the `${KEY}` reference off stdin, looks the @@ -65,6 +69,7 @@ esac { printf 'BETTER_AUTH_SECRET=%s\\n' "${BETTER_AUTH_SECRET+set:}${BETTER_AUTH_SECRET:-}" printf 'DEER_FLOW_INTERNAL_AUTH_TOKEN=%s\\n' "${DEER_FLOW_INTERNAL_AUTH_TOKEN+set:}${DEER_FLOW_INTERNAL_AUTH_TOKEN:-}" + printf 'DEER_FLOW_CREDENTIALS_KEY=%s\\n' "${DEER_FLOW_CREDENTIALS_KEY+set:}${DEER_FLOW_CREDENTIALS_KEY:-}" } > "$CAPTURE_SECRETS" for arg in "$@"; do printf "%s\\n" "$arg"; done > "$CAPTURE_DOCKER_ARGS" exit 0 @@ -164,7 +169,8 @@ def test_deploy_asks_compose_to_interpolate_the_secret_like_the_real_project(tmp # The stub project on stdin is what gets interpolated: it must reference # the secret and nothing from the real compose file. assert config_args[config_args.index("-f") + 1] == "-" - assert "${DEER_FLOW_INTERNAL_AUTH_TOKEN}" in config_stdin + # Each probe replaces the captured stub; the last secret resolved is the last one probed. + assert f"${{{SECRETS[-1]}}}" in config_stdin assert "docker-compose.yaml" not in config_stdin @@ -384,3 +390,36 @@ def test_real_compose_empty_shell_export_still_gets_a_generated_secret(tmp_path, _, observed, _, _, _, _ = _run_deploy_build(tmp_path, worktree, real_docker=real_docker, shell_env={key: ""}) assert GENERATED.fullmatch(observed[key]), observed[key] + + +# ── DEER_FLOW_CREDENTIALS_KEY specifics ───────────────────────────────────── + + +def test_deploy_generates_a_valid_fernet_credentials_key(tmp_path): + """The Gateway refuses a malformed key, so the generated one must parse as Fernet.""" + from cryptography.fernet import Fernet + + from deerflow.config.credentials_key import parse_credentials_keys + + worktree = _worktree(tmp_path) + + _, observed, _, _, _, home = _run_deploy_build(tmp_path, worktree, compose_environment={}) + + key = observed["DEER_FLOW_CREDENTIALS_KEY"].removeprefix("set:") + Fernet(key) + assert parse_credentials_keys(key) == [key] + assert (home / ".credentials_key").read_text(encoding="utf-8").strip() == key + + +def test_deploy_reuses_a_key_file_the_gateway_already_generated(tmp_path): + """A stack first started without deploy.sh already has the Gateway's own key file: keep it.""" + worktree = _worktree(tmp_path) + home = tmp_path / "deer-flow-home" + home.mkdir() + (home / ".credentials_key").write_text("gateway-generated-key", encoding="utf-8") + + result, observed, _, _, _, _ = _run_deploy_build(tmp_path, worktree, compose_environment={}) + + assert observed["DEER_FLOW_CREDENTIALS_KEY"] == "set:gateway-generated-key" + assert (home / ".credentials_key").read_text(encoding="utf-8") == "gateway-generated-key" + assert "DEER_FLOW_CREDENTIALS_KEY loaded from" in result.stdout diff --git a/backend/tests/test_deploy_home_writability.py b/backend/tests/test_deploy_home_writability.py index 69f4b2c26a..2a9b959bc5 100644 --- a/backend/tests/test_deploy_home_writability.py +++ b/backend/tests/test_deploy_home_writability.py @@ -51,6 +51,7 @@ def deploy_fixture(tmp_path: Path): # Keep the preflight independent of secret generation and optional extras. BETTER_AUTH_SECRET="test-better-auth-secret", DEER_FLOW_INTERNAL_AUTH_TOKEN="test-internal-auth-token", + DEER_FLOW_CREDENTIALS_KEY="test-credentials-key", UV_EXTRAS="redis", ) return worktree, env, home, capture @@ -76,8 +77,12 @@ def unwritable_home(deploy_fixture): @pytest.fixture( - params=[("BETTER_AUTH_SECRET", ".better-auth-secret"), ("DEER_FLOW_INTERNAL_AUTH_TOKEN", ".internal-auth-token")], - ids=["better-auth", "internal-auth"], + params=[ + ("BETTER_AUTH_SECRET", ".better-auth-secret"), + ("DEER_FLOW_INTERNAL_AUTH_TOKEN", ".internal-auth-token"), + ("DEER_FLOW_CREDENTIALS_KEY", ".credentials_key"), + ], + ids=["better-auth", "internal-auth", "credentials-key"], ) def persisted_secret(request, deploy_fixture): _, _, home, _ = deploy_fixture @@ -150,6 +155,7 @@ def test_deploy_down_allows_unwritable_home(deploy_fixture, unwritable_home): _, env, _, capture = deploy_fixture env.pop("BETTER_AUTH_SECRET") env.pop("DEER_FLOW_INTERNAL_AUTH_TOKEN") + env.pop("DEER_FLOW_CREDENTIALS_KEY") result = _run_deploy(deploy_fixture, "down") @@ -226,6 +232,7 @@ def test_deploy_down_allows_unreadable_persisted_secret(deploy_fixture, unreadab _, env, home, capture = deploy_fixture env.pop("BETTER_AUTH_SECRET") env.pop("DEER_FLOW_INTERNAL_AUTH_TOKEN") + env.pop("DEER_FLOW_CREDENTIALS_KEY") result = _run_deploy(deploy_fixture, "down") diff --git a/backend/tests/test_docker_sandbox_mode_detection.py b/backend/tests/test_docker_sandbox_mode_detection.py index 9c1bccb28b..0e9049d067 100644 --- a/backend/tests/test_docker_sandbox_mode_detection.py +++ b/backend/tests/test_docker_sandbox_mode_detection.py @@ -572,6 +572,7 @@ def test_aio_deploy_socket_preflight_allows_windows_when_docker_reachable(tmp_pa env["DEER_FLOW_DOCKER_SOCKET"] = "/var/run/docker.sock" env["BETTER_AUTH_SECRET"] = "test-secret" env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token" + env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key" env["UV_EXTRAS"] = "redis" result = subprocess.run( @@ -597,6 +598,7 @@ def test_aio_deploy_socket_preflight_rejects_missing_socket_on_posix(tmp_path): env["DEER_FLOW_DOCKER_SOCKET"] = "/nonexistent/docker.sock" env["BETTER_AUTH_SECRET"] = "test-secret" env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token" + env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key" env["UV_EXTRAS"] = "redis" result = subprocess.run( @@ -632,6 +634,7 @@ def test_aio_deploy_socket_preflight_rejects_missing_custom_socket_on_windows(tm env["DEER_FLOW_DOCKER_SOCKET"] = "/nonexistent/docker.sock" env["BETTER_AUTH_SECRET"] = "test-secret" env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token" + env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key" env["UV_EXTRAS"] = "redis" result = subprocess.run( @@ -668,6 +671,7 @@ def test_aio_deploy_socket_preflight_rejects_windows_when_docker_unreachable(tmp env["DEER_FLOW_DOCKER_SOCKET"] = "/var/run/docker.sock" env["BETTER_AUTH_SECRET"] = "test-secret" env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token" + env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key" env["UV_EXTRAS"] = "redis" result = subprocess.run( @@ -704,6 +708,7 @@ def test_aio_deploy_socket_unsets_default_on_windows(tmp_path): env["DEER_FLOW_DOCKER_SOCKET"] = "/var/run/docker.sock" env["BETTER_AUTH_SECRET"] = "test-secret" env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token" + env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key" env["UV_EXTRAS"] = "redis" result = subprocess.run( @@ -740,6 +745,7 @@ def test_aio_deploy_socket_preserves_unset_default_on_windows(tmp_path): env.pop("DEER_FLOW_DOCKER_SOCKET", None) env["BETTER_AUTH_SECRET"] = "test-secret" env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token" + env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key" env["UV_EXTRAS"] = "redis" result = subprocess.run( @@ -779,6 +785,7 @@ def test_aio_deploy_socket_preserves_custom_socket_on_windows(tmp_path): env["DEER_FLOW_DOCKER_SOCKET"] = "/custom/docker.sock" env["BETTER_AUTH_SECRET"] = "test-secret" env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token" + env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key" env["UV_EXTRAS"] = "redis" result = subprocess.run( @@ -816,6 +823,7 @@ def test_aio_deploy_socket_reads_from_dotenv(tmp_path): env.pop("DEER_FLOW_DOCKER_SOCKET", None) env["BETTER_AUTH_SECRET"] = "test-secret" env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token" + env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key" env["UV_EXTRAS"] = "redis" result = subprocess.run( diff --git a/backend/tests/test_helm_credentials_key.py b/backend/tests/test_helm_credentials_key.py new file mode 100644 index 0000000000..8f1d45f7d8 --- /dev/null +++ b/backend/tests/test_helm_credentials_key.py @@ -0,0 +1,106 @@ +"""The Helm chart generates and injects ``DEER_FLOW_CREDENTIALS_KEY``. + +Every gateway Pod must encrypt and decrypt stored credentials with one key, so +the chart generates a Fernet key into the app Secret once, preserves it across +upgrades through ``lookup`` like ``AUTH_JWT_SECRET``, and injects it into the +gateway. Sprig's ``randBytes`` emits standard base64; Fernet keys use the +urlsafe alphabet, so the template translates ``+``/``/`` to ``-``/``_``. + +The ``helm template`` tests skip when helm is not installed; CI's runner has it. +""" + +from __future__ import annotations + +import base64 +import re +import shutil +import subprocess +from pathlib import Path + +import pytest +import yaml +from cryptography.fernet import Fernet + +from deerflow.config.credentials_key import CREDENTIALS_KEY_ENV_VAR, parse_credentials_keys + +REPO_ROOT = Path(__file__).resolve().parents[2] +CHART = REPO_ROOT / "deploy" / "helm" / "deer-flow" +APP_SECRET_TEMPLATE = CHART / "templates" / "secret-app.yaml" +README = CHART / "README.md" +NOTES = CHART / "templates" / "NOTES.txt" +VALUES = CHART / "values.yaml" + +_FERNET_KEY = re.compile(r"[A-Za-z0-9_-]{43}=") + + +def _render_chart(*settings: str) -> list[dict]: + helm = shutil.which("helm") + if helm is None: + pytest.skip("helm is unavailable") + command = [helm, "template", "deer-flow", str(CHART)] + for setting in settings: + command.extend(["--set", setting]) + rendered = subprocess.run(command, check=True, capture_output=True, text=True).stdout + return [document for document in yaml.safe_load_all(rendered) if isinstance(document, dict)] + + +def _app_secret(documents: list[dict]) -> dict: + return next(document for document in documents if document.get("kind") == "Secret" and document["metadata"]["name"].endswith("-app")) + + +def _gateway(documents: list[dict]) -> dict: + deployment = next(document for document in documents if document.get("kind") == "Deployment" and document["metadata"]["name"].endswith("-gateway")) + return next(container for container in deployment["spec"]["template"]["spec"]["containers"] if container["name"] == "gateway") + + +def _env(container: dict) -> dict[str, dict]: + return {item["name"]: item for item in container["env"]} + + +def test_rendered_app_secret_carries_a_valid_fernet_key() -> None: + key = _app_secret(_render_chart())["stringData"][CREDENTIALS_KEY_ENV_VAR] + + assert _FERNET_KEY.fullmatch(key), "must be urlsafe base64 of 32 bytes" + assert len(base64.urlsafe_b64decode(key)) == 32 + Fernet(key) + assert parse_credentials_keys(key) == [key], "the Gateway's own parser must accept the generated key" + + +def test_each_fresh_render_generates_an_independent_key() -> None: + """randBytes, not a constant: two installs must not share a key.""" + first = _app_secret(_render_chart())["stringData"][CREDENTIALS_KEY_ENV_VAR] + second = _app_secret(_render_chart())["stringData"][CREDENTIALS_KEY_ENV_VAR] + assert first != second + + +def test_app_secret_template_preserves_the_key_across_upgrades() -> None: + """``helm template`` never sees ``lookup`` results, so pin the preservation in the source.""" + template = APP_SECRET_TEMPLATE.read_text(encoding="utf-8") + assert f'index $prev.data "{CREDENTIALS_KEY_ENV_VAR}"' in template, "a regenerated key would make every stored credential unreadable" + assert f"{CREDENTIALS_KEY_ENV_VAR}: {{{{ $credentialsKey | quote }}}}" in template + assert 'randBytes 32 | replace "+" "-" | replace "/" "_"' in template + + +@pytest.mark.parametrize("replicas", ["1", "3"]) +def test_gateway_reads_the_key_from_the_app_secret(replicas: str) -> None: + gateway = _gateway(_render_chart(f"gateway.replicas={replicas}")) + reference = _env(gateway)[CREDENTIALS_KEY_ENV_VAR]["valueFrom"]["secretKeyRef"] + + assert reference["name"].endswith("-app") + assert reference["key"] == CREDENTIALS_KEY_ENV_VAR + # Optional so a user-managed existingAppSecret without the key still boots: the + # Gateway's own startup gate refuses a multi-instance deployment that stores + # credentials without it, and leaves the others alone. + assert reference["optional"] is True + + +def test_existing_app_secret_is_honored_for_the_key() -> None: + documents = _render_chart("existingAppSecret=my-app-secret") + assert not any(document.get("kind") == "Secret" and document["metadata"]["name"].endswith("-app") for document in documents) + assert _env(_gateway(documents))[CREDENTIALS_KEY_ENV_VAR]["valueFrom"]["secretKeyRef"]["name"] == "my-app-secret" + + +def test_docs_mention_the_key() -> None: + for document in (README, NOTES, VALUES): + assert CREDENTIALS_KEY_ENV_VAR in document.read_text(encoding="utf-8"), document.name + assert ".credentials_key" in README.read_text(encoding="utf-8"), "the README must say how to keep an auto-generated key" diff --git a/backend/tests/test_managed_models.py b/backend/tests/test_managed_models.py index 0d5c91115b..73fa921c06 100644 --- a/backend/tests/test_managed_models.py +++ b/backend/tests/test_managed_models.py @@ -60,6 +60,55 @@ def test_missing_encryption_key_never_replaced(store): assert not store.key_path.exists() +def test_first_save_keeps_a_key_a_peer_created_concurrently(store, monkeypatch): + """A peer that publishes the key file first must win; replacing it would orphan the peer's catalog.""" + from cryptography.fernet import Fernet + + peer_key = Fernet.generate_key() + original_generate = Fernet.generate_key + + def peer_wins_the_race(): + store.key_path.parent.mkdir(parents=True, exist_ok=True) + store.key_path.write_bytes(peer_key) + return original_generate() + + monkeypatch.setattr(Fernet, "generate_key", staticmethod(peer_wins_the_race)) + store.save(profile(), expected_revision=None) + + assert store.key_path.read_bytes() == peer_key + assert Fernet(peer_key).decrypt(store.path.read_bytes()) + + +def test_first_save_recovers_an_empty_key_left_by_an_interrupted_creation(store): + """An empty key file with no catalog encrypted no data, so the next save must replace it, not 503 forever.""" + from cryptography.fernet import Fernet + + store.key_path.parent.mkdir(parents=True, exist_ok=True) + store.key_path.write_bytes(b"") + + store.save(profile(), expected_revision=None) + + key = store.key_path.read_bytes() + assert key.strip() + assert Fernet(key).decrypt(store.path.read_bytes()) + assert ManagedModelStore().list()[0].api_key.get_secret_value() == "test-secret" + + +def test_an_empty_key_next_to_an_existing_catalog_is_never_replaced(store): + """The catalog was encrypted with the lost key; a new key would silently orphan it.""" + store.save(profile(), expected_revision=None) + catalog = store.path.read_bytes() + store.key_path.write_bytes(b"") + + with pytest.raises(ValueError): + store.save(profile(), expected_revision=None) + with pytest.raises(ValueError, match="restore it from backup"): + store._cipher(create=True) + + assert store.key_path.read_bytes() == b"" + assert store.path.read_bytes() == catalog + + @pytest.mark.parametrize("url", ["file:///tmp/test", "https://user:pass@example.com/v1", "https://example.com/v1?key=abc", "https://example.com/#fragment"]) def test_endpoint_validation(url): with pytest.raises(ValidationError): diff --git a/backend/tests/test_multi_worker_postgres_gate.py b/backend/tests/test_multi_worker_postgres_gate.py index 710c547927..d0c7700e61 100644 --- a/backend/tests/test_multi_worker_postgres_gate.py +++ b/backend/tests/test_multi_worker_postgres_gate.py @@ -18,8 +18,17 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest from fastapi import FastAPI -from app.gateway.deps import _enforce_postgres_for_multi_worker, _validate_agent_storage, _validate_login_throttle_storage, _validate_memory_retrieval_index, langgraph_runtime +from app.gateway.deps import ( + _enforce_credentials_key, + _enforce_postgres_for_multi_worker, + _validate_agent_storage, + _validate_login_throttle_storage, + _validate_memory_retrieval_index, + credentials_key_consumers, + langgraph_runtime, +) from app.gateway.routers.browser import _browser_tools_enabled +from deerflow.config.credentials_key import CREDENTIALS_KEY_ENV_VAR, GENERATE_KEY_COMMAND from deerflow.config.database_config import DatabaseConfig from deerflow.config.deployment_config import MULTI_INSTANCE_ENV_VAR, DeploymentConfig, multi_instance_declaration from deerflow.config.run_ownership_config import RunOwnershipConfig @@ -75,6 +84,7 @@ def isolated_worker_env(monkeypatch): monkeypatch.delenv("WEB_CONCURRENCY", raising=False) monkeypatch.delenv(MULTI_INSTANCE_ENV_VAR, raising=False) monkeypatch.delenv("DEER_FLOW_STREAM_BRIDGE_REDIS_URL", raising=False) + monkeypatch.delenv(CREDENTIALS_KEY_ENV_VAR, raising=False) # --------------------------------------------------------------------------- @@ -846,3 +856,100 @@ def test_retrieval_index_warning_only_covers_the_bundled_fts5_index(caplog, memo with caplog.at_level("WARNING"): _validate_memory_retrieval_index(config) assert _retrieval_index_warnings(caplog) == [] + + +# --------------------------------------------------------------------------- +# At-rest credentials key (DEER_FLOW_CREDENTIALS_KEY) +# --------------------------------------------------------------------------- + +_VALID_KEY = "QUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUE=" + + +def _with_channel_connections(config, *, enabled: bool = True): + config.channel_connections = SimpleNamespace(enabled=enabled) + return config + + +def test_credentials_key_consumers_name_the_enabled_features(): + assert credentials_key_consumers(_cluster_ready()) == [], "a config without the section uses no key" + assert credentials_key_consumers(_with_channel_connections(_cluster_ready(), enabled=False)) == [] + assert credentials_key_consumers(_with_channel_connections(_cluster_ready())) == ["channel_connections.enabled=true"] + + +def test_declared_multi_instance_with_channel_connections_requires_the_credentials_key(): + """Each instance would otherwise generate its own key file and could not read its peers' credentials.""" + with pytest.raises(SystemExit) as exc_info: + _enforce_credentials_key(_with_channel_connections(_cluster_ready(deployment_multi_instance=True))) + msg = str(exc_info.value) + assert CREDENTIALS_KEY_ENV_VAR in msg + assert "deployment.multi_instance=true" in msg + assert "channel_connections.enabled=true" in msg + assert GENERATE_KEY_COMMAND in msg, "the refusal must say how to generate a key" + assert "Helm" in msg + assert "deployment.multi_instance=false" in msg, "must name the rollback knob" + assert "or set channel_connections.enabled=false" in msg, "must name the setting that turns the consumer off" + + +def test_env_declared_multi_instance_names_the_env_knob(monkeypatch): + monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "1") + with pytest.raises(SystemExit) as exc_info: + _enforce_credentials_key(_with_channel_connections(_cluster_ready())) + msg = str(exc_info.value) + assert f"{MULTI_INSTANCE_ENV_VAR}=1" in msg + assert f"unset {MULTI_INSTANCE_ENV_VAR}" in msg + + +def test_declared_multi_instance_accepts_a_shared_credentials_key(monkeypatch): + monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, _VALID_KEY) + _enforce_credentials_key(_with_channel_connections(_cluster_ready(deployment_multi_instance=True))) + + +@pytest.mark.parametrize("channel_connections", [None, False], ids=["section-absent", "disabled"]) +def test_declared_multi_instance_without_key_consumers_keeps_booting(channel_connections): + """Existing multi-instance deployments that store no credentials must not need the key.""" + config = _cluster_ready(deployment_multi_instance=True) + if channel_connections is not None: + _with_channel_connections(config, enabled=channel_connections) + _enforce_credentials_key(config) + + +def test_multi_worker_without_a_declaration_shares_the_generated_key_file(monkeypatch): + """Workers of one process tree share base_dir, where the exclusive-create key file converges.""" + monkeypatch.setenv("GATEWAY_WORKERS", "4") + _enforce_credentials_key(_with_channel_connections(_cluster_ready())) + + +def test_single_instance_without_a_credentials_key_keeps_booting(): + _enforce_credentials_key(_with_channel_connections(_config_with_backend("sqlite"))) + + +@pytest.mark.parametrize("declared", [False, True], ids=["single-instance", "multi-instance"]) +@pytest.mark.parametrize("value", ["not-a-fernet-key", f"{_VALID_KEY},second-entry-is-garbage"], ids=["garbage", "bad-rotation-entry"]) +def test_a_malformed_credentials_key_is_refused_without_echoing_it(monkeypatch, declared, value): + monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, value) + with pytest.raises(SystemExit) as exc_info: + _enforce_credentials_key(_with_channel_connections(_cluster_ready(deployment_multi_instance=declared))) + msg = str(exc_info.value) + assert CREDENTIALS_KEY_ENV_VAR in msg + assert "not-a-fernet-key" not in msg and "second-entry-is-garbage" not in msg and _VALID_KEY not in msg + + +@pytest.mark.asyncio +async def test_langgraph_runtime_enforces_the_credentials_key_before_persistence_setup(): + init_engine_from_config = AsyncMock(name="init_engine_from_config") + + @asynccontextmanager + async def _noop_stream_bridge(_config): + yield MagicMock() + + with ( + patch("deerflow.persistence.engine.init_engine_from_config", init_engine_from_config), + patch("deerflow.runtime.make_stream_bridge", side_effect=_noop_stream_bridge) as make_stream_bridge, + ): + startup_config = _with_channel_connections(_cluster_ready(deployment_multi_instance=True)) + with pytest.raises(SystemExit, match=CREDENTIALS_KEY_ENV_VAR): + async with langgraph_runtime(FastAPI(), startup_config): + pass + + init_engine_from_config.assert_not_called() + make_stream_bridge.assert_not_called() diff --git a/backend/tests/test_secret_file.py b/backend/tests/test_secret_file.py new file mode 100644 index 0000000000..43b7e70250 --- /dev/null +++ b/backend/tests/test_secret_file.py @@ -0,0 +1,351 @@ +"""Exclusive-create helper for small secret files shared by concurrent processes. + +Gateway replicas that share a runtime home (a Helm ``ReadWriteMany`` volume, a +compose bind mount, or several uvicorn workers of one host) cold-start at the +same time and each find no secret file yet. A read-then-``O_TRUNC`` write lets +every one of them keep a different value -- for ``.jwt_secret`` that is two +replicas signing sessions with different keys, for ``.credentials_key`` it is +credentials one replica cannot decrypt. ``read_or_create_secret_file`` creates +with ``O_EXCL`` and reads the winner's value back instead. +""" + +from __future__ import annotations + +import logging +import os +import threading +from pathlib import Path + +import pytest + +from deerflow.config.secret_file import InvalidSecretFileError, read_or_create_secret_file + + +def test_creates_a_missing_file_with_owner_only_permissions(tmp_path: Path) -> None: + path = tmp_path / "nested" / ".secret" + + value = read_or_create_secret_file(path, lambda: "generated-value") + + assert value == "generated-value" + assert path.read_text(encoding="utf-8") == "generated-value" + if os.name == "posix": + assert path.stat().st_mode & 0o777 == 0o600 + + +def test_reuses_an_existing_value_without_generating(tmp_path: Path) -> None: + path = tmp_path / ".secret" + path.write_text("existing-value\n", encoding="utf-8") + + def generate() -> str: + raise AssertionError("an existing secret must never be regenerated") + + assert read_or_create_secret_file(path, generate) == "existing-value" + assert path.read_text(encoding="utf-8") == "existing-value\n" + + +def test_a_peer_that_publishes_first_wins(tmp_path: Path) -> None: + """Deterministic race: the peer creates the file between our existence check and our create. + + A read-then-truncating write would overwrite the peer's value and return + our own, leaving the two processes on different secrets. + """ + path = tmp_path / ".secret" + + def generate() -> str: + path.write_text("peer-value", encoding="utf-8") + return "our-value" + + assert read_or_create_secret_file(path, generate) == "peer-value" + assert path.read_text(encoding="utf-8") == "peer-value" + + +def test_concurrent_creators_converge_on_one_value(tmp_path: Path) -> None: + path = tmp_path / ".secret" + workers = 16 + barrier = threading.Barrier(workers) + results: list[str] = [] + errors: list[BaseException] = [] + lock = threading.Lock() + + def run(index: int) -> None: + try: + barrier.wait(timeout=10) + value = read_or_create_secret_file(path, lambda: f"value-{index}") + with lock: + results.append(value) + except BaseException as exc: # noqa: BLE001 - surface every worker failure + with lock: + errors.append(exc) + + threads = [threading.Thread(target=run, args=(index,)) for index in range(workers)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=30) + + assert not errors + assert len(results) == workers + assert len(set(results)) == 1, f"creators diverged: {sorted(set(results))}" + assert path.read_text(encoding="utf-8") == results[0] + + +def test_waits_for_a_peer_that_has_created_but_not_yet_written(tmp_path: Path) -> None: + """Older releases published the name before its content: an empty file may be one of them mid-write.""" + path = tmp_path / ".secret" + path.touch() + writer = threading.Timer(0.1, lambda: path.write_text("late-value", encoding="utf-8")) + writer.start() + try: + value = read_or_create_secret_file(path, lambda: "our-value", settle_seconds=5) + finally: + writer.join() + + assert value == "late-value" + assert path.read_text(encoding="utf-8") == "late-value" + + +def test_retries_a_partially_written_value_that_fails_validation(tmp_path: Path) -> None: + path = tmp_path / ".secret" + path.write_text("0123", encoding="utf-8") + + def validate(value: str) -> None: + if len(value) != 10: + raise ValueError("incomplete") + + writer = threading.Timer(0.1, lambda: path.write_text("0123456789", encoding="utf-8")) + writer.start() + try: + value = read_or_create_secret_file(path, lambda: "abcdefghij", validate=validate, settle_seconds=5) + finally: + writer.join() + + assert value == "0123456789" + + +def test_an_abandoned_empty_file_is_replaced(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + """A crash between create and write (or an older release's truncating write) leaves an empty file. + + Nothing can have been signed or encrypted with an empty secret, so once the + settle window passes it is replaced instead of failing startup forever. + """ + path = tmp_path / ".secret" + path.touch() + + with caplog.at_level(logging.WARNING, logger="deerflow.config.secret_file"): + value = read_or_create_secret_file(path, lambda: "fresh-value", settle_seconds=0.05) + + assert value == "fresh-value" + assert path.read_text(encoding="utf-8") == "fresh-value" + assert any(str(path) in record.getMessage() for record in caplog.records) + if os.name == "posix": + assert path.stat().st_mode & 0o777 == 0o600 + + +def test_concurrent_replacers_of_an_abandoned_empty_file_converge(tmp_path: Path) -> None: + """Replicas restarting together after a crash all find the same abandoned empty file. + + The replacement is a last-writer-wins rename, so a replacer must not trust + its own value until peers that decided to replace in the same window have + written theirs. + """ + path = tmp_path / ".secret" + path.touch() + workers = 8 + barrier = threading.Barrier(workers) + results: list[str] = [] + errors: list[BaseException] = [] + lock = threading.Lock() + + def run(index: int) -> None: + def generate() -> str: + # Stagger the replacements so an early replacer would otherwise + # read back its own value before a later one overwrites it. + threading.Event().wait(0.01 * index) + return f"value-{index}" + + try: + barrier.wait(timeout=10) + value = read_or_create_secret_file(path, generate, settle_seconds=0.2) + with lock: + results.append(value) + except BaseException as exc: # noqa: BLE001 - surface every worker failure + with lock: + errors.append(exc) + + threads = [threading.Thread(target=run, args=(index,)) for index in range(workers)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=30) + + assert not errors + assert len(results) == workers + assert len(set(results)) == 1, f"replacers diverged: {sorted(set(results))}" + assert path.read_text(encoding="utf-8") == results[0] + + +def test_waits_for_a_peer_that_holds_the_replacement_claim(tmp_path: Path) -> None: + path = tmp_path / ".secret" + path.touch() + claim = tmp_path / ".secret.replacing" + claim.write_text("peer-value", encoding="utf-8") + publisher = threading.Timer(0.3, lambda: os.replace(claim, path)) + publisher.start() + try: + value = read_or_create_secret_file(path, lambda: "our-value", settle_seconds=0.05) + finally: + publisher.join() + + assert value == "peer-value" + assert not claim.exists() + + +def test_a_claim_left_by_a_crashed_replacer_is_refused_after_a_bounded_wait(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("deerflow.config.secret_file._CLAIM_WAIT_SECONDS", 0.2) + path = tmp_path / ".secret" + path.touch() + claim = tmp_path / ".secret.replacing" + claim.touch() + + with pytest.raises(InvalidSecretFileError, match="interrupted creation"): + read_or_create_secret_file(path, lambda: "our-value", settle_seconds=0.05) + + assert path.read_text(encoding="utf-8") == "" + + +def test_a_late_replacer_does_not_overwrite_a_published_replacement(tmp_path: Path) -> None: + """A process that saw the empty file before a peer's replacement landed must back off.""" + path = tmp_path / ".secret" + path.write_text("peer-value", encoding="utf-8") + from deerflow.config.secret_file import _publish_under_claim + + assert _publish_under_claim(path, lambda: "late-value", replace_empty=True) is True + assert path.read_text(encoding="utf-8") == "peer-value" + assert not (tmp_path / ".secret.replacing").exists() + + +def test_a_persistently_invalid_file_is_refused_without_echoing_it(tmp_path: Path) -> None: + """A non-empty value may be the only copy of a key that encrypted data: never overwrite it.""" + path = tmp_path / ".secret" + path.write_text("operator-typo-in-the-key", encoding="utf-8") + + def validate(value: str) -> None: + raise ValueError(f"bad value {value}") + + with pytest.raises(InvalidSecretFileError) as exc_info: + read_or_create_secret_file(path, lambda: "replacement", validate=validate, settle_seconds=0.05) + + assert str(path) in str(exc_info.value) + assert "operator-typo-in-the-key" not in str(exc_info.value) + assert exc_info.value.__cause__ is None, "the validator's message may quote the value" + assert path.read_text(encoding="utf-8") == "operator-typo-in-the-key" + + +def test_a_failed_write_leaves_no_empty_file_for_peers_to_wait_on(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + path = tmp_path / ".secret" + + def broken_write(_fd: int, _data: bytes) -> None: + raise OSError("disk full") + + monkeypatch.setattr("deerflow.config.secret_file._write_all", broken_write) + + with pytest.raises(OSError, match="disk full"): + read_or_create_secret_file(path, lambda: "value") + + assert not path.exists() + + +def _pause_one_creator(monkeypatch: pytest.MonkeyPatch, thread_name: str, *, fail: bool = False) -> tuple[threading.Event, threading.Event]: + """Hold ``thread_name`` inside its secret write until released; the other threads write normally.""" + import deerflow.config.secret_file as secret_file + + entered = threading.Event() + release = threading.Event() + original_write_all = secret_file._write_all + + def write_all(fd: int, data: bytes) -> None: + if threading.current_thread().name == thread_name: + entered.set() + assert release.wait(timeout=10) + if fail: + raise OSError("disk full") + original_write_all(fd, data) + + monkeypatch.setattr(secret_file, "_write_all", write_all) + return entered, release + + +@pytest.mark.parametrize("fail", [False, True], ids=["paused-then-writes", "paused-then-fails"]) +def test_a_paused_creator_never_exposes_a_file_a_peer_could_replace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, fail: bool) -> None: + """A creator stalled mid-write must not leave a visible empty file behind. + + Otherwise a peer treats it as abandoned after the settle window and + publishes its own value; the resumed creator then returns a value nobody + persisted, or (when its write fails) deletes the peer's valid file. + """ + path = tmp_path / ".secret" + entered, release = _pause_one_creator(monkeypatch, "first", fail=fail) + results: dict[str, str | BaseException] = {} + + def run(name: str) -> None: + try: + results[name] = read_or_create_secret_file(path, lambda: f"{name}-value", settle_seconds=0.05) + except BaseException as exc: # noqa: BLE001 - asserted below + results[name] = exc + + first = threading.Thread(target=run, args=("first",), name="first") + first.start() + assert entered.wait(timeout=10) + peer = threading.Thread(target=run, args=("peer",), name="peer") + peer.start() + peer.join(timeout=10) + release.set() + first.join(timeout=10) + + assert results["peer"] == "peer-value" + assert path.read_text(encoding="utf-8") == "peer-value", "the peer's published value must survive" + if fail: + assert isinstance(results["first"], OSError) + else: + assert results["first"] == "peer-value", "the resumed creator must adopt the persisted value" + + +def test_creation_without_hard_links_still_converges(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Filesystems without hard links (SMB shares) publish through the replacement claim instead.""" + import errno + + def no_hard_links(*_args, **_kwargs): + raise OSError(errno.EPERM, "Operation not permitted") + + monkeypatch.setattr(os, "link", no_hard_links) + path = tmp_path / ".secret" + workers = 8 + barrier = threading.Barrier(workers) + results: list[str] = [] + lock = threading.Lock() + + def run(index: int) -> None: + barrier.wait(timeout=10) + value = read_or_create_secret_file(path, lambda: f"value-{index}", settle_seconds=0.05) + with lock: + results.append(value) + + threads = [threading.Thread(target=run, args=(index,)) for index in range(workers)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=30) + + assert len(results) == workers + assert len(set(results)) == 1, f"creators diverged: {sorted(set(results))}" + assert path.read_text(encoding="utf-8") == results[0] + assert not (tmp_path / ".secret.replacing").exists() + if os.name == "posix": + assert path.stat().st_mode & 0o777 == 0o600 + + +def test_no_temporary_files_are_left_behind(tmp_path: Path) -> None: + path = tmp_path / ".secret" + read_or_create_secret_file(path, lambda: "value") + + assert sorted(entry.name for entry in tmp_path.iterdir()) == [".secret"] diff --git a/deploy/helm/deer-flow/README.md b/deploy/helm/deer-flow/README.md index 263e0f8b38..cfde3fa142 100644 --- a/deploy/helm/deer-flow/README.md +++ b/deploy/helm/deer-flow/README.md @@ -379,6 +379,25 @@ kubectl -n deer-flow exec deploy/deer-flow-provisioner -- curl -s localhost:8002 JWT=$(kubectl -n deer-flow exec deploy/deer-flow-gateway -- cat /app/backend/.deer-flow/.jwt_secret) kubectl -n deer-flow patch secret deer-flow-app -p "{\"stringData\":{\"AUTH_JWT_SECRET\":\"$JWT\"}}" ``` +- **Credentials key.** The app Secret also holds `DEER_FLOW_CREDENTIALS_KEY`, + the Fernet key the gateway encrypts stored credentials with (per-connection + IM channel credentials today). The chart generates it once (32 random bytes, + urlsafe base64) and preserves it across upgrades via `lookup`; every gateway + Pod reads the same value. **Back it up**: deleting the Secret or the release + generates a new key, and credentials stored under the old one become + unreadable (they are treated as missing, never as an error). To rotate, + prepend a new key — `new,old` — and keep the old one listed until stored + values have been rewritten. An `existingAppSecret` should carry it; the env + entry is optional, so without it a single Pod falls back to an auto-generated + `.credentials_key` on the home volume, while a multi-instance gateway with + `channel_connections.enabled` refuses to start. If you add the key to an + `existingAppSecret` later, copy that file's value so stored credentials stay + readable: + + ```bash + KEY=$(kubectl -n deer-flow exec deploy/deer-flow-gateway -- cat /app/backend/.deer-flow/.credentials_key) + kubectl -n deer-flow patch secret my-app-secret -p "{\"stringData\":{\"DEER_FLOW_CREDENTIALS_KEY\":\"$KEY\"}}" + ``` - **Scheduled task recovery.** If a deployment explicitly enables `scheduler.multi_instance: true`, it must use shared Postgres, `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`. @@ -493,7 +512,7 @@ but not file mode — so a PVC written by an earlier **root** run (e.g. a cluste that ran the gateway as root before enabling this hardening, or a backup restore of root-owned files) will keep files like `.jwt_secret` at `0600 root:root`. The non-root gateway (uid 1000) then can't read them and crashes on the first auth -request with `RuntimeError: Failed to read JWT secret from .../​.jwt_secret`. +request with `RuntimeError: Failed to read or persist the JWT secret at .../​.jwt_secret`. **Fresh installs are unaffected** — uid 1000 creates every file as `1000:1000`. diff --git a/deploy/helm/deer-flow/templates/NOTES.txt b/deploy/helm/deer-flow/templates/NOTES.txt index e902ea73b7..5e57cb98f3 100644 --- a/deploy/helm/deer-flow/templates/NOTES.txt +++ b/deploy/helm/deer-flow/templates/NOTES.txt @@ -74,10 +74,12 @@ Provisioner / sandbox notes: Generated secrets (persisted across upgrades): • BETTER_AUTH_SECRET, DEER_FLOW_INTERNAL_AUTH_TOKEN, AUTH_JWT_SECRET (the - session-cookie signing key shared by every gateway Pod) and + session-cookie signing key shared by every gateway Pod), PROVISIONER_API_KEY (the key the gateway presents to the sandbox - provisioner, which answers 401 to every /api/* request without it) are - stored in the Secret `{{ include "deer-flow.appSecret" . }}`. + provisioner, which answers 401 to every /api/* request without it) and + DEER_FLOW_CREDENTIALS_KEY (the at-rest key for stored credentials; back it + up, losing it makes them unreadable) are stored in the Secret + `{{ include "deer-flow.appSecret" . }}`. Database ({{ if .Values.postgresql.enabled }}bundled postgres{{ else }}external postgres{{ end }}): diff --git a/deploy/helm/deer-flow/templates/_helpers.tpl b/deploy/helm/deer-flow/templates/_helpers.tpl index d0b2d38172..cd3093845f 100644 --- a/deploy/helm/deer-flow/templates/_helpers.tpl +++ b/deploy/helm/deer-flow/templates/_helpers.tpl @@ -71,7 +71,8 @@ imagePullSecrets: {{- end -}} {{/* Name of the Secret holding the app secrets (BETTER_AUTH_SECRET, - DEER_FLOW_INTERNAL_AUTH_TOKEN, AUTH_JWT_SECRET, PROVISIONER_API_KEY). + DEER_FLOW_INTERNAL_AUTH_TOKEN, AUTH_JWT_SECRET, PROVISIONER_API_KEY, + DEER_FLOW_CREDENTIALS_KEY). `existingAppSecret` points every consumer (gateway, frontend, provisioner) at a user-managed Secret instead and skips generating one; it must carry PROVISIONER_API_KEY while the provisioner is enabled and diff --git a/deploy/helm/deer-flow/templates/gateway-deployment.yaml b/deploy/helm/deer-flow/templates/gateway-deployment.yaml index b5662adf39..43da44f65c 100644 --- a/deploy/helm/deer-flow/templates/gateway-deployment.yaml +++ b/deploy/helm/deer-flow/templates/gateway-deployment.yaml @@ -150,6 +150,18 @@ spec: name: {{ include "deer-flow.appSecret" . }} key: AUTH_JWT_SECRET optional: {{ ne (include "deer-flow.gatewayMultiInstance" .) "true" }} + # At-rest key for stored credentials, shared by every Pod so one + # Pod can decrypt what another stored. Optional so a user-managed + # existingAppSecret without it still boots: a single Pod then falls + # back to .credentials_key on the home volume, and the Gateway's own + # startup gate refuses a multi-instance deployment that stores + # credentials (channel_connections) without the key. + - name: DEER_FLOW_CREDENTIALS_KEY + valueFrom: + secretKeyRef: + name: {{ include "deer-flow.appSecret" . }} + key: DEER_FLOW_CREDENTIALS_KEY + optional: true {{- if .Values.provisioner.enabled }} # Shared gateway -> bundled-provisioner key. config.yaml references # it as sandbox.provisioner_api_key: $PROVISIONER_API_KEY, and the diff --git a/deploy/helm/deer-flow/templates/secret-app.yaml b/deploy/helm/deer-flow/templates/secret-app.yaml index 23103535e8..e3a783b06a 100644 --- a/deploy/helm/deer-flow/templates/secret-app.yaml +++ b/deploy/helm/deer-flow/templates/secret-app.yaml @@ -4,11 +4,13 @@ {{- $internalToken := "" -}} {{- $jwtSecret := "" -}} {{- $provisionerKey := "" -}} +{{- $credentialsKey := "" -}} {{- if $prev -}} {{- $betterAuth = index $prev.data "BETTER_AUTH_SECRET" | default "" | b64dec -}} {{- $internalToken = index $prev.data "DEER_FLOW_INTERNAL_AUTH_TOKEN" | default "" | b64dec -}} {{- $jwtSecret = index $prev.data "AUTH_JWT_SECRET" | default "" | b64dec -}} {{- $provisionerKey = index $prev.data "PROVISIONER_API_KEY" | default "" | b64dec -}} +{{- $credentialsKey = index $prev.data "DEER_FLOW_CREDENTIALS_KEY" | default "" | b64dec -}} {{- end -}} {{- if not $betterAuth -}}{{- $betterAuth = randAlphaNum 48 | lower -}}{{- end -}} {{- if not $internalToken -}}{{- $internalToken = randAlphaNum 40 -}}{{- end -}} @@ -22,6 +24,11 @@ value. Generated once and preserved across upgrades so a rollout never leaves the two sides on different keys. */ -}} {{- if not $provisionerKey -}}{{- $provisionerKey = randAlphaNum 48 -}}{{- end -}} +{{- /* At-rest key for stored credentials (a Fernet key: urlsafe base64 of 32 + random bytes). randBytes emits standard base64, so translate its + alphabet. Generated once and preserved across upgrades: a new key would + make every credential the gateway Pods stored unreadable. */ -}} +{{- if not $credentialsKey -}}{{- $credentialsKey = randBytes 32 | replace "+" "-" | replace "/" "_" -}}{{- end -}} apiVersion: v1 kind: Secret metadata: @@ -35,4 +42,5 @@ stringData: DEER_FLOW_INTERNAL_AUTH_TOKEN: {{ $internalToken | quote }} AUTH_JWT_SECRET: {{ $jwtSecret | quote }} PROVISIONER_API_KEY: {{ $provisionerKey | quote }} + DEER_FLOW_CREDENTIALS_KEY: {{ $credentialsKey | quote }} {{- end -}} diff --git a/deploy/helm/deer-flow/values.yaml b/deploy/helm/deer-flow/values.yaml index 95bbb5d6fe..eb6f913c91 100644 --- a/deploy/helm/deer-flow/values.yaml +++ b/deploy/helm/deer-flow/values.yaml @@ -338,7 +338,10 @@ existingSecret: "" # DEER_FLOW_INTERNAL_AUTH_TOKEN, PROVISIONER_API_KEY (the shared gateway -> # sandbox-provisioner key, required while `provisioner.enabled`) and, for # more than one gateway instance, AUTH_JWT_SECRET (the session-cookie -# signing key every Pod must share). +# signing key every Pod must share). It should also hold +# DEER_FLOW_CREDENTIALS_KEY (the at-rest key for stored credentials, a +# Fernet key); a multi-instance gateway with channel_connections enabled +# refuses to start without it. existingAppSecret: "" # -- Ingress in front of nginx (port 2026). nginx preserves all internal routing. diff --git a/docker/docker-compose-dev.yaml b/docker/docker-compose-dev.yaml index a9e27f2db3..d074356713 100644 --- a/docker/docker-compose-dev.yaml +++ b/docker/docker-compose-dev.yaml @@ -239,6 +239,8 @@ services: - DEER_FLOW_CHANNELS_LANGGRAPH_URL=${DEER_FLOW_CHANNELS_LANGGRAPH_URL:-http://gateway:8001/api} - DEER_FLOW_CHANNELS_GATEWAY_URL=${DEER_FLOW_CHANNELS_GATEWAY_URL:-http://gateway:8001} - DEER_FLOW_INTERNAL_AUTH_TOKEN=${DEER_FLOW_INTERNAL_AUTH_TOKEN:-} + # At-rest key for stored credentials; empty falls back to .credentials_key in DEER_FLOW_HOME. + - DEER_FLOW_CREDENTIALS_KEY=${DEER_FLOW_CREDENTIALS_KEY:-} - DEER_FLOW_HOST_BASE_DIR=${DEER_FLOW_ROOT}/backend/.deer-flow - DEER_FLOW_SANDBOX_HOST=host.docker.internal # The login lockout counts failures per client IP, but every browser diff --git a/docker/docker-compose.yaml b/docker/docker-compose.yaml index 2ef39a7e43..0fc74a41fa 100644 --- a/docker/docker-compose.yaml +++ b/docker/docker-compose.yaml @@ -24,6 +24,7 @@ # DEER_FLOW_REPO_ROOT — repo root (used for skills host path in DooD) # BETTER_AUTH_SECRET — required for frontend auth/session security # DEER_FLOW_INTERNAL_AUTH_TOKEN — shared internal Gateway auth token for multi-worker IM channels +# DEER_FLOW_CREDENTIALS_KEY — at-rest Fernet key for stored credentials (deploy.sh persists it as $DEER_FLOW_HOME/.credentials_key) # # LangSmith tracing is disabled by default (LANGSMITH_TRACING=false). # Set LANGSMITH_TRACING=true and LANGSMITH_API_KEY in .env to enable it. @@ -165,6 +166,9 @@ services: - DEER_FLOW_CHANNELS_LANGGRAPH_URL=${DEER_FLOW_CHANNELS_LANGGRAPH_URL:-http://gateway:8001/api} - DEER_FLOW_CHANNELS_GATEWAY_URL=${DEER_FLOW_CHANNELS_GATEWAY_URL:-http://gateway:8001} - DEER_FLOW_INTERNAL_AUTH_TOKEN=${DEER_FLOW_INTERNAL_AUTH_TOKEN} + # At-rest key for stored credentials (scripts/deploy.sh resolves or generates it). + # Empty falls back to .credentials_key in the runtime home mounted above. + - DEER_FLOW_CREDENTIALS_KEY=${DEER_FLOW_CREDENTIALS_KEY:-} # DooD path/network translation - DEER_FLOW_HOST_BASE_DIR=${DEER_FLOW_HOME} - DEER_FLOW_SANDBOX_HOST=host.docker.internal diff --git a/scripts/AGENTS.md b/scripts/AGENTS.md index 1d7117cec9..8e7812dcb6 100644 --- a/scripts/AGENTS.md +++ b/scripts/AGENTS.md @@ -97,9 +97,10 @@ the `env_file` value); shell exports still win. Pinned by `deploy.sh` never sources the repo-root `.env`; Compose reads it via `--env-file`, and shell exports outrank that file during interpolation (an -exported-but-empty variable still wins). So `BETTER_AUTH_SECRET` and -`DEER_FLOW_INTERNAL_AUTH_TOKEN` resolve shell → `.env` → persisted file under -`DEER_FLOW_HOME` → freshly generated, and a `.env`-provided value is left +exported-but-empty variable still wins). So `BETTER_AUTH_SECRET`, +`DEER_FLOW_INTERNAL_AUTH_TOKEN` and `DEER_FLOW_CREDENTIALS_KEY` resolve shell → +`.env` → persisted file under `DEER_FLOW_HOME` → freshly generated, and a +`.env`-provided value is left unexported so Compose parses it itself. Whether `.env` provides one is Compose's answer, not a `KEY=VALUE` grep: Compose also accepts `KEY: VALUE` lines and interpolates `${VAR}` inside values, so the script renders a stub @@ -112,6 +113,9 @@ failing probe stops the script rather than guessing. `read_dotenv_value` stays for the end-of-run summary only. Do not export a value the script read from `.env`: that shadows Compose's own dotenv parsing and re-creates the bug where `make up` replaced the operator's secret with a generated one. +The credentials key (a Fernet key) persists as `.credentials_key`, the file the +Gateway itself generates in that runtime home, via a noclobber (`O_EXCL`) +create that reads a concurrent winner back. `backend/tests/test_deploy_dotenv_secrets.py` pins the order and the probe; its real-Compose cases run against the installed `docker` CLI and against any standalone binaries listed in `DEER_FLOW_TEST_COMPOSE_BINARIES`. diff --git a/scripts/deploy.sh b/scripts/deploy.sh index 0435308b09..d029142279 100755 --- a/scripts/deploy.sh +++ b/scripts/deploy.sh @@ -298,6 +298,53 @@ elif [ "$CMD" != "down" ] && [ -z "$DEER_FLOW_INTERNAL_AUTH_TOKEN" ]; then fi fi +# ── DEER_FLOW_CREDENTIALS_KEY ──────────────────────────────────────────────── +# At-rest Fernet key the Gateway encrypts stored credentials with. Losing it +# makes those credentials unreadable, so back the file up. It is persisted +# under the name the Gateway itself generates in its runtime home (the +# $DEER_FLOW_HOME bind mount), so a key either side created first is the one +# both keep. noclobber turns the redirection into an exclusive (O_EXCL) +# create: when a concurrent run wins, its key is read back instead of ours. + +_credentials_key_file="$DEER_FLOW_HOME/.credentials_key" +if [ "$CMD" != "down" ] && [ -z "$DEER_FLOW_CREDENTIALS_KEY" ] && dotenv_provides_secret DEER_FLOW_CREDENTIALS_KEY; then + echo -e "${GREEN}✓ DEER_FLOW_CREDENTIALS_KEY loaded from $ENV_FILE${NC}" +elif [ "$CMD" != "down" ] && [ -z "$DEER_FLOW_CREDENTIALS_KEY" ]; then + _generated_credentials_key="" + if [ ! -f "$_credentials_key_file" ]; then + # urlsafe base64 of 32 random bytes, the Fernet key format. + if command -v python3 > /dev/null 2>&1 && \ + _generated_credentials_key="$(python3 -c 'import sys; sys.version_info >= (3, 6) or sys.exit(1); import base64, os; print(base64.urlsafe_b64encode(os.urandom(32)).decode())' 2>/dev/null)"; then + true + elif command -v python > /dev/null 2>&1 && \ + _generated_credentials_key="$(python -c 'import sys; sys.version_info >= (3, 6) or sys.exit(1); import base64, os; print(base64.urlsafe_b64encode(os.urandom(32)).decode())' 2>/dev/null)"; then + true + elif command -v openssl > /dev/null 2>&1 && \ + _generated_credentials_key="$(openssl rand -base64 32 | tr '+/' '-_')"; then + true + else + echo -e "${RED}✗ Cannot generate DEER_FLOW_CREDENTIALS_KEY: python3, python, and openssl are all unavailable.${NC}" >&2 + echo -e "${RED} Set DEER_FLOW_CREDENTIALS_KEY manually before running make up.${NC}" >&2 + exit 1 + fi + if ! ( umask 077; set -o noclobber; printf '%s\n' "$_generated_credentials_key" > "$_credentials_key_file" ) 2>/dev/null; then + _generated_credentials_key="" + fi + fi + if [ -n "$_generated_credentials_key" ]; then + export DEER_FLOW_CREDENTIALS_KEY="$_generated_credentials_key" + echo -e "${GREEN}✓ DEER_FLOW_CREDENTIALS_KEY generated → $_credentials_key_file (back it up)${NC}" + else + if [ ! -r "$_credentials_key_file" ]; then + fail_home_permission "$_credentials_key_file" readable + fi + export DEER_FLOW_CREDENTIALS_KEY + DEER_FLOW_CREDENTIALS_KEY="$(cat "$_credentials_key_file")" + echo -e "${GREEN}✓ DEER_FLOW_CREDENTIALS_KEY loaded from $_credentials_key_file${NC}" + fi + unset _generated_credentials_key +fi + # ── UV_EXTRAS auto-detection ───────────────────────────────────────────────── # The production Dockerfile accepts UV_EXTRAS as a single build-arg token and # adds the --extra prefix itself. Convert the detector's uv flag string diff --git a/scripts/docker.sh b/scripts/docker.sh index 09a892265f..02a155a1db 100755 --- a/scripts/docker.sh +++ b/scripts/docker.sh @@ -150,9 +150,10 @@ ensure_env_files() { # start runs Compose from docker/ without --env-file, so ${VAR} interpolation in # docker-compose-dev.yaml sees the shell only, never the checkout .env. Export # the .env values that interpolation consumes; a value already set in the shell -# (even empty) wins, as it does for Compose. AUTH_TRUSTED_PROXIES belongs here -# because the Gateway's `environment:` entry, which defaults to nginx, outranks -# the same key loaded through env_file and would replace the operator's list. +# (even empty) wins, as it does for Compose. AUTH_TRUSTED_PROXIES and +# DEER_FLOW_CREDENTIALS_KEY belong here because the Gateway's `environment:` +# entries (defaulting to nginx and to empty) outrank the same keys loaded through +# env_file and would replace the operator's values. load_proxy_env_from_dotenv() { local env_file="$PROJECT_ROOT/.env" local var @@ -163,7 +164,7 @@ load_proxy_env_from_dotenv() { return fi - for var in HTTP_PROXY HTTPS_PROXY ALL_PROXY NO_PROXY http_proxy https_proxy all_proxy no_proxy AUTH_TRUSTED_PROXIES; do + for var in HTTP_PROXY HTTPS_PROXY ALL_PROXY NO_PROXY http_proxy https_proxy all_proxy no_proxy AUTH_TRUSTED_PROXIES DEER_FLOW_CREDENTIALS_KEY; do if [ -z "${!var+x}" ]; then line="$(grep -E "^[[:space:]]*${var}=" "$env_file" | tail -n 1 || true)" if [ -n "$line" ]; then