wehub-resource-sync
81 lines
3.2 KiB
Python
81 lines
3.2 KiB
Python
"""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)
|
|
|
|
|
|
def test_internal_auth_headers_can_carry_owner_user_id(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(owner_user_id="owner-1")
|
|
|
|
assert headers[reloaded.INTERNAL_AUTH_HEADER_NAME] == "shared-token"
|
|
assert headers[reloaded.INTERNAL_OWNER_USER_ID_HEADER_NAME] == "owner-1"
|
|
finally:
|
|
monkeypatch.delenv("DEER_FLOW_INTERNAL_AUTH_TOKEN", raising=False)
|
|
importlib.reload(reloaded)
|
|
|
|
|
|
def test_get_internal_user_normalises_unsafe_owner_user_id():
|
|
"""P2-3: X-DeerFlow-Owner-User-Id is at the trust boundary, so the
|
|
synthetic internal user must use a path-safe id. ``make_safe_user_id``
|
|
is lossy but deterministic; two distinct raw inputs never collide.
|
|
"""
|
|
import app.gateway.internal_auth as internal_auth
|
|
from deerflow.config.paths import make_safe_user_id
|
|
|
|
# Path-traversal-style payloads must be normalised away.
|
|
user_a = internal_auth.get_internal_user(owner_user_id="ou_abc/../../etc/passwd")
|
|
user_b = internal_auth.get_internal_user(owner_user_id="ou_abc/../../etc/passwd")
|
|
assert user_a.id == user_b.id
|
|
assert "/" not in user_a.id
|
|
assert ".." not in user_a.id
|
|
|
|
# Negative chat ids and unsafe punctuation must be normalised.
|
|
user_neg = internal_auth.get_internal_user(owner_user_id="-1001234567890:alice")
|
|
assert user_neg.id == make_safe_user_id("-1001234567890:alice")
|
|
assert ":" not in user_neg.id
|
|
assert user_neg.system_role == "internal"
|
|
|
|
# Already-safe ids pass through unchanged.
|
|
user_safe = internal_auth.get_internal_user(owner_user_id="alice_42")
|
|
assert user_safe.id == "alice_42"
|
|
|
|
# Empty / None falls back to default.
|
|
assert internal_auth.get_internal_user().id == "default"
|
|
assert internal_auth.get_internal_user(owner_user_id="").id == "default"
|