fix(auth): share internal gateway token across workers (#3184)

* fix(auth): share internal gateway token across workers

* fix: restore deploy script executable bit

* Update deploy.sh

to skip the auth_token setup for the down command

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/app/gateway/internal_auth.py

@@ -1,23 +1,34 @@
-"""Process-local authentication for Gateway internal callers."""
+"""Authentication for trusted Gateway internal callers."""
 
 from __future__ import annotations
 
+import os
 import secrets
 from types import SimpleNamespace
 
 from deerflow.runtime.user_context import DEFAULT_USER_ID
 
 INTERNAL_AUTH_HEADER_NAME = "X-DeerFlow-Internal-Token"
-_INTERNAL_AUTH_TOKEN = secrets.token_urlsafe(32)
+INTERNAL_AUTH_ENV_VAR = "DEER_FLOW_INTERNAL_AUTH_TOKEN"
+
+
+def _load_internal_auth_token() -> str:
+    token = os.environ.get(INTERNAL_AUTH_ENV_VAR)
+    if token:
+        return token
+    return secrets.token_urlsafe(32)
+
+
+_INTERNAL_AUTH_TOKEN = _load_internal_auth_token()
 
 
 def create_internal_auth_headers() -> dict[str, str]:
-    """Return headers that authenticate same-process Gateway internal calls."""
+    """Return headers that authenticate trusted Gateway internal calls."""
     return {INTERNAL_AUTH_HEADER_NAME: _INTERNAL_AUTH_TOKEN}
 
 
 def is_valid_internal_auth_token(token: str | None) -> bool:
-    """Return True when *token* matches the process-local internal token."""
+    """Return True when *token* matches this Gateway worker's internal token."""
     return bool(token) and secrets.compare_digest(token, _INTERNAL_AUTH_TOKEN)
 
 

backend/tests/test_internal_auth.py

@@ -0,0 +1,35 @@
+"""Tests for Gateway internal auth token handling."""
+
+from __future__ import annotations
+
+import importlib
+
+
+def test_internal_auth_uses_shared_env_token(monkeypatch):
+    import app.gateway.internal_auth as internal_auth
+
+    monkeypatch.setenv("DEER_FLOW_INTERNAL_AUTH_TOKEN", "shared-token")
+    reloaded = importlib.reload(internal_auth)
+    try:
+        headers = reloaded.create_internal_auth_headers()
+
+        assert headers[reloaded.INTERNAL_AUTH_HEADER_NAME] == "shared-token"
+        assert reloaded.is_valid_internal_auth_token("shared-token") is True
+        assert reloaded.is_valid_internal_auth_token("other-token") is False
+    finally:
+        monkeypatch.delenv("DEER_FLOW_INTERNAL_AUTH_TOKEN", raising=False)
+        importlib.reload(reloaded)
+
+
+def test_internal_auth_generates_process_local_fallback(monkeypatch):
+    import app.gateway.internal_auth as internal_auth
+
+    monkeypatch.delenv("DEER_FLOW_INTERNAL_AUTH_TOKEN", raising=False)
+    reloaded = importlib.reload(internal_auth)
+    try:
+        token = reloaded.create_internal_auth_headers()[reloaded.INTERNAL_AUTH_HEADER_NAME]
+
+        assert token
+        assert reloaded.is_valid_internal_auth_token(token) is True
+    finally:
+        importlib.reload(reloaded)