mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-11-12 14:00:35 +00:00
feat(weknora): allow opt-in shared retrieval for authenticated users (#7944)
* feat(weknora): allow opt-in shared authenticated retrieval * docs(weknora): clarify shared retrieval identity requirements * docs(weknora): clarify durable batch authentication limits * docs(gateway): keep agent guidance within hard size limit --------- Co-authored-by: Totoro <msy626836554@gmail.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
c738c5ef8b
commit
463361c5ba
@ -22,6 +22,11 @@ This section accumulates work toward the **2.3.0** milestone
|
||||
`hitl_approval.max_actions` configures the approval-batch count limit
|
||||
(default 32; 0 disables it). HITL is disabled by default; `config_version` 65.
|
||||
|
||||
- **knowledge:** The WeKnora retrieval extension offers opt-in shared read-only
|
||||
access for authenticated users, without maintaining a per-user allowlist.
|
||||
Exact-user allowlists remain the default; shared access requires matching
|
||||
host-verified identity and grants no management permissions. ([#7944](https://github.com/bytedance/deer-flow/pull/7944))
|
||||
|
||||
- **subagents:** Durable batch items can wait for explicitly named prerequisites.
|
||||
An optional, once-only lead-Agent follow-up completes a previously requested
|
||||
summary or delivery after the batch ends, with persisted admission, cancellation
|
||||
|
||||
@ -1887,8 +1887,9 @@ Providers remain usable without a catalog or management page. See the
|
||||
for extension registration and configuration. The installable
|
||||
[WeKnora read-only extension](examples/deerflow-extension-weknora/README.md)
|
||||
targets WeKnora v0.8.3 and adds knowledge-base/document selection and shared
|
||||
retrieval citations. Configure explicit user-to-workspace credential bindings;
|
||||
resource authorization remains with WeKnora. The extension directory includes
|
||||
retrieval citations. Configure explicit user-to-workspace credential bindings,
|
||||
or opt into shared read-only access for authenticated users without maintaining
|
||||
a user-ID list. Resource authorization remains with WeKnora. The extension directory includes
|
||||
its installation guide.
|
||||
|
||||
The optional [knowledge platform extension](examples/deerflow-extension-knowledge-platform/README.md)
|
||||
|
||||
@ -1,5 +1,7 @@
|
||||
### Gateway API (`app/gateway/`)
|
||||
|
||||
Knowledge auth: [contract](../../docs/KNOWLEDGE.md).
|
||||
|
||||
Reject external `sandbox`, `thread_data`, `viewed_images`, `goal_outcome`. Non-null goals: valid text, int counters/limits >=0, no Overwrite; keep sparse data.
|
||||
|
||||
[Viewer fence](../../../docs/full-stack-plugins.md).
|
||||
@ -11,19 +13,17 @@ HTTP and run metadata creation share `goal_thread_lock`. Lock in detached
|
||||
ownership. Its 5s budget covers admission and I/O: timeout fails strict starts;
|
||||
others omit incarnation (MCP fails closed). Tests: `test_thread_create_run_metadata_race.py`.
|
||||
|
||||
Capability Center's `business` adapter validates bundled credentials and creates
|
||||
an MCP connection. The MCP API permits only its isolated launcher and credential
|
||||
environment keys; reject arbitrary commands and manifest-policy bypasses. Keep
|
||||
credentials out of tool schemas and preserve admin checks, masked atomic edits,
|
||||
and MCP cache reloads.
|
||||
Capability Center `business`: validate bundled credentials, create MCP connection.
|
||||
MCP API allows only its isolated launcher and credential env keys; reject arbitrary
|
||||
commands and manifest-policy bypasses. No credentials in tool schemas; retain
|
||||
admin checks, masked atomic edits and MCP cache reloads.
|
||||
|
||||
Memory shutdown resolves hot-reloaded config/backend, flushes and closes in one
|
||||
`await_drained` operation. Resolve config off-loop inside the best-effort handler
|
||||
so malformed edits cannot abort teardown. Cancellation waits for owned workers.
|
||||
The host sets no close timeout: `close()` must be quick or internally bounded;
|
||||
budget it, flush timeout and other hooks within pod grace. Drain extension
|
||||
`stop_services()` across cancellation before later resources unwind; each
|
||||
service `stop()` has a 30-second `asyncio.timeout`.
|
||||
Memory shutdown: resolve hot-reloaded config/backend off-loop in the best-effort
|
||||
handler; malformed edits must not abort teardown. Flush/close in one
|
||||
`await_drained`; cancellation waits for owned workers. Host close has no timeout:
|
||||
`close()` must be quick or internally bounded. Fit close, flush timeout and other
|
||||
hooks within pod grace. Drain extension `stop_services()` across cancellation
|
||||
before later resources unwind; each `stop()` has a 30s `asyncio.timeout`.
|
||||
|
||||
Routers with persistent writes drain through
|
||||
`app.gateway.persistent_writes.run_drained_write`.
|
||||
|
||||
@ -29,7 +29,7 @@ from app.gateway.auth_disabled import (
|
||||
from app.gateway.authz import AuthContext, resolve_route_permissions
|
||||
from app.gateway.internal_auth import INTERNAL_AUTH_HEADER_NAME, get_internal_user, is_valid_internal_auth_token
|
||||
from app.gateway.request_path import get_request_route_path
|
||||
from deerflow.runtime.user_context import reset_current_user, set_current_user
|
||||
from deerflow.runtime.user_context import bind_authenticated_user, reset_current_user, set_current_user
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -234,6 +234,8 @@ class AuthMiddleware(BaseHTTPMiddleware):
|
||||
request.state.auth = AuthContext(user=user, permissions=permissions)
|
||||
token = set_current_user(user)
|
||||
try:
|
||||
return await call_next(request)
|
||||
verified_user_id = str(user.id) if auth_source in {AUTH_SOURCE_SESSION, AUTH_SOURCE_PAT} else None
|
||||
with bind_authenticated_user(verified_user_id):
|
||||
return await call_next(request)
|
||||
finally:
|
||||
reset_current_user(token)
|
||||
|
||||
@ -17,7 +17,7 @@ from deerflow.config.agents_config import load_agent_config
|
||||
from deerflow.config.app_config import AppConfig
|
||||
from deerflow.knowledge import KnowledgeRequestContext, RetrievalCatalogPage
|
||||
from deerflow.knowledge.providers import get_knowledge_capabilities, resolve_knowledge_provider
|
||||
from deerflow.runtime.user_context import get_effective_user_id
|
||||
from deerflow.runtime.user_context import get_authenticated_user_id, get_effective_user_id
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
router = APIRouter(prefix="/api/knowledge", tags=["knowledge"])
|
||||
@ -94,7 +94,8 @@ async def list_retrieval_catalog_datasets(
|
||||
config: AppConfig = Depends(get_config),
|
||||
) -> dict[str, Any]:
|
||||
provider = await _scope_catalog(config, agent_name, request, "catalog")
|
||||
return await _catalog_result(provider.catalog(context=KnowledgeRequestContext(user_id=get_effective_user_id(), agent_name=agent_name), page=page, page_size=page_size, search=search))
|
||||
context = KnowledgeRequestContext(user_id=get_effective_user_id(), agent_name=agent_name, authenticated_user_id=get_authenticated_user_id())
|
||||
return await _catalog_result(provider.catalog(context=context, page=page, page_size=page_size, search=search))
|
||||
|
||||
|
||||
@router.get("/retrieval-catalog/datasets/{dataset_id:path}/documents")
|
||||
@ -109,4 +110,5 @@ async def list_retrieval_catalog_documents(
|
||||
config: AppConfig = Depends(get_config),
|
||||
) -> dict[str, Any]:
|
||||
provider = await _scope_catalog(config, agent_name, request, "documents")
|
||||
return await _catalog_result(provider.documents(dataset_id, context=KnowledgeRequestContext(user_id=get_effective_user_id(), agent_name=agent_name), page=page, page_size=page_size, search=search))
|
||||
context = KnowledgeRequestContext(user_id=get_effective_user_id(), agent_name=agent_name, authenticated_user_id=get_authenticated_user_id())
|
||||
return await _catalog_result(provider.documents(dataset_id, context=context, page=page, page_size=page_size, search=search))
|
||||
|
||||
@ -51,6 +51,32 @@ may retain `search(query)` when it uses an operator-owned connection with a
|
||||
fixed resource visibility. User/tenant mappings and resource authorization
|
||||
remain provider responsibilities; DeerFlow does not infer tenants from client
|
||||
metadata or create a second ACL store.
|
||||
The optional `authenticated_user_id` field carries separate host authentication
|
||||
provenance and defaults to `None`. Gateway binds it only for verified sessions
|
||||
and PATs, never internal-service or auth-disabled synthetic identities. Generic
|
||||
retrieval tools and catalog routes read it from a dedicated task-local binding;
|
||||
client flags, runtime `user_id`, saved assistant configuration and storage owner
|
||||
restoration cannot create it. Child tasks, threads and ordinary isolated
|
||||
subagents launched from an authenticated execution inherit its proof; a
|
||||
lost/fresh context denies shared access. Durable batch items are launched from
|
||||
the Gateway's startup poller and do not inherit the submitting request's proof,
|
||||
even within the same process. Like scheduled-task and IM-channel runs, they
|
||||
cannot use authenticated shared retrieval merely by carrying an owner `user_id`.
|
||||
Trusted embedding hosts may use `bind_authenticated_user()` from
|
||||
`deerflow.runtime.user_context` only after authenticating the end user, and must
|
||||
bind `None` for non-user callers.
|
||||
The bound value must be exactly the same nonempty string that
|
||||
`resolve_runtime_user_id()` returns for the execution. LangGraph Server
|
||||
identities from `runtime.server_info.user.identity` or
|
||||
`configurable.langgraph_auth_user_id` take precedence over runtime context and
|
||||
are normalized with `make_safe_user_id()`. A trusted embedding host must bind
|
||||
that same resolved ID after authentication, rather than a raw identity such as
|
||||
an email address. Binding a different representation denies shared access;
|
||||
providers must not normalize the two values independently to bypass the mismatch.
|
||||
Providers opting into shared access must require a nonempty exact match with
|
||||
`context.user_id` and still enforce their connection/resource/Agent policy.
|
||||
The WeKnora extension's explicit `access_mode: authenticated` uses this field
|
||||
for dynamic-user shared retrieval; its default exact-user mode is unchanged.
|
||||
`agent_name` identifies the current executor: the selected assistant name,
|
||||
defaulting to `lead_agent`, or the executing subagent's configured name from
|
||||
its server runtime `agent_id`. It does not identify the parent assistant.
|
||||
|
||||
@ -89,6 +89,9 @@ class KnowledgeRequestContext:
|
||||
user_id: str
|
||||
agent_name: str | None = None
|
||||
scope: dict[str, Any] | None = None
|
||||
# Set only by a host authentication boundary, not from serialized run context.
|
||||
# Providers must compare it to user_id before granting shared user access.
|
||||
authenticated_user_id: str | None = None
|
||||
|
||||
|
||||
class RetrievalCatalogItem(BaseModel):
|
||||
|
||||
@ -12,6 +12,7 @@ import logging
|
||||
from typing import Any
|
||||
|
||||
from deerflow.knowledge_scope import execution_scope
|
||||
from deerflow.runtime.user_context import get_authenticated_user_id
|
||||
|
||||
from .contract import KnowledgeCapabilities, KnowledgeRequestContext, RetrievalProvider, RetrievalResult
|
||||
|
||||
@ -171,7 +172,7 @@ async def search_provider(provider: RetrievalProvider, query: str, *, scope: obj
|
||||
canonical = validate_scope_capabilities(scope, capabilities) if scope is not None else None
|
||||
if canonical is not None and canonical["mode"] == "disabled":
|
||||
raise ValueError("Knowledge search is disabled for this turn")
|
||||
context = KnowledgeRequestContext(user_id=user_id, agent_name=agent_name, scope=canonical)
|
||||
context = KnowledgeRequestContext(user_id=user_id, agent_name=agent_name, scope=canonical, authenticated_user_id=get_authenticated_user_id())
|
||||
contextual_search = getattr(provider, "search_scoped" if operation == "search" else operation, None)
|
||||
if callable(contextual_search):
|
||||
result = await contextual_search(query, context=context)
|
||||
|
||||
@ -34,7 +34,8 @@ a background task must *not* see the foreground user, wrap it with
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from collections.abc import Iterator, Mapping
|
||||
from contextlib import contextmanager
|
||||
from contextvars import ContextVar, Token
|
||||
from typing import Final, Protocol, runtime_checkable
|
||||
|
||||
@ -51,6 +52,30 @@ class CurrentUser(Protocol):
|
||||
|
||||
|
||||
_current_user: Final[ContextVar[CurrentUser | None]] = ContextVar("deerflow_current_user", default=None)
|
||||
_authenticated_user_id: Final[ContextVar[str | None]] = ContextVar("deerflow_authenticated_user_id", default=None)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def bind_authenticated_user(user_id: str | None) -> Iterator[None]:
|
||||
"""Bind verified end-user identity at a trusted host authentication boundary.
|
||||
|
||||
Storage ownership alone is not authentication: internal and auth-disabled
|
||||
callers also have user IDs. Bind ``None`` for them, and never derive this
|
||||
value from run context, configuration, request parameters or model input.
|
||||
Child tasks/threads inherit this proof; a fresh context fails closed.
|
||||
"""
|
||||
if user_id is not None and (not isinstance(user_id, str) or not user_id.strip() or user_id != user_id.strip()):
|
||||
raise ValueError("Authenticated user ID must be a nonblank exact host identity")
|
||||
token = _authenticated_user_id.set(user_id)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
_authenticated_user_id.reset(token)
|
||||
|
||||
|
||||
def get_authenticated_user_id() -> str | None:
|
||||
"""Return host-verified end-user identity, without a storage-ID fallback."""
|
||||
return _authenticated_user_id.get()
|
||||
|
||||
|
||||
def set_current_user(user: CurrentUser) -> Token[CurrentUser | None]:
|
||||
|
||||
78
backend/tests/test_authenticated_user_context.py
Normal file
78
backend/tests/test_authenticated_user_context.py
Normal file
@ -0,0 +1,78 @@
|
||||
"""Authentication provenance must remain separate from the storage user ID."""
|
||||
|
||||
import asyncio
|
||||
from contextvars import Context
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.mark.no_auto_user
|
||||
def test_user_context_or_client_metadata_is_not_authentication():
|
||||
from deerflow.runtime.user_context import get_authenticated_user_id, reset_current_user, set_current_user
|
||||
|
||||
assert get_authenticated_user_id() is None
|
||||
token = set_current_user(SimpleNamespace(id="user", authenticated=True))
|
||||
try:
|
||||
assert get_authenticated_user_id() is None
|
||||
finally:
|
||||
reset_current_user(token)
|
||||
|
||||
|
||||
def test_verified_identity_can_be_cleared_and_restored():
|
||||
from deerflow.runtime.user_context import bind_authenticated_user, get_authenticated_user_id
|
||||
|
||||
with bind_authenticated_user("alice"):
|
||||
assert get_authenticated_user_id() == "alice"
|
||||
with bind_authenticated_user(None):
|
||||
assert get_authenticated_user_id() is None
|
||||
assert get_authenticated_user_id() == "alice"
|
||||
with pytest.raises(RuntimeError), bind_authenticated_user("bob"):
|
||||
raise RuntimeError("cancelled request")
|
||||
assert get_authenticated_user_id() == "alice"
|
||||
assert get_authenticated_user_id() is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("invalid", ["", " ", " alice ", True, 1])
|
||||
def test_verified_identity_rejects_invalid_host_bindings(invalid):
|
||||
from deerflow.runtime.user_context import bind_authenticated_user
|
||||
|
||||
with pytest.raises(ValueError), bind_authenticated_user(invalid):
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_verified_identity_survives_child_tasks_and_threads_without_cross_user_leaks():
|
||||
from deerflow.runtime.user_context import bind_authenticated_user, get_authenticated_user_id
|
||||
|
||||
ready = asyncio.Event()
|
||||
|
||||
async def child(user):
|
||||
await ready.wait()
|
||||
assert get_authenticated_user_id() == user
|
||||
return await asyncio.to_thread(get_authenticated_user_id)
|
||||
|
||||
with bind_authenticated_user("alice"):
|
||||
alice = asyncio.create_task(child("alice"))
|
||||
with bind_authenticated_user("bob"):
|
||||
bob = asyncio.create_task(child("bob"))
|
||||
assert get_authenticated_user_id() is None
|
||||
ready.set()
|
||||
assert await asyncio.gather(alice, bob) == ["alice", "bob"]
|
||||
assert Context().run(get_authenticated_user_id) is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_receives_host_proof_separately_from_its_user_id():
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from deerflow.knowledge import KnowledgeCapabilities, RetrievalResult
|
||||
from deerflow.knowledge.providers import search_provider
|
||||
from deerflow.runtime.user_context import bind_authenticated_user
|
||||
|
||||
provider = SimpleNamespace(capabilities=KnowledgeCapabilities(), search_scoped=AsyncMock(return_value=RetrievalResult()))
|
||||
await search_provider(provider, "query", scope=None, user_id="alice", agent_name="lead_agent")
|
||||
assert provider.search_scoped.await_args.kwargs["context"].authenticated_user_id is None
|
||||
with bind_authenticated_user("alice"):
|
||||
await search_provider(provider, "query", scope=None, user_id="alice", agent_name="researcher")
|
||||
assert provider.search_scoped.await_args.kwargs["context"].authenticated_user_id == "alice"
|
||||
@ -4954,6 +4954,24 @@ class TestCooperativeCancellation:
|
||||
"checkpoint_ns": "parent:task",
|
||||
}
|
||||
|
||||
def test_authenticated_identity_crosses_isolated_loop_without_user_leakage(self, executor_module):
|
||||
from deerflow.runtime.user_context import bind_authenticated_user, get_authenticated_user_id
|
||||
|
||||
ready = threading.Event()
|
||||
|
||||
async def read_identity():
|
||||
await asyncio.to_thread(ready.wait)
|
||||
return get_authenticated_user_id()
|
||||
|
||||
with bind_authenticated_user("alice"):
|
||||
alice = executor_module._submit_to_isolated_loop_in_context(executor_module._copy_isolated_subagent_context(), read_identity)
|
||||
with bind_authenticated_user("bob"):
|
||||
bob = executor_module._submit_to_isolated_loop_in_context(executor_module._copy_isolated_subagent_context(), read_identity)
|
||||
assert get_authenticated_user_id() is None
|
||||
ready.set()
|
||||
assert alice.result(timeout=10) == "alice"
|
||||
assert bob.result(timeout=10) == "bob"
|
||||
|
||||
def test_isolated_context_filters_callback_manager_without_mutating_parent(self, executor_module):
|
||||
from langchain_core.callbacks.manager import AsyncCallbackManager
|
||||
from langchain_core.runnables.config import var_child_runnable_config
|
||||
|
||||
196
backend/tests/test_weknora_shared_access.py
Normal file
196
backend/tests/test_weknora_shared_access.py
Normal file
@ -0,0 +1,196 @@
|
||||
"""Dynamic users share only an explicitly configured read-only connection."""
|
||||
|
||||
import asyncio
|
||||
|
||||
import pytest
|
||||
from test_weknora_extension import adapter as adapter
|
||||
|
||||
from deerflow.knowledge import KnowledgeRequestContext
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def shared(adapter):
|
||||
from deerflow_extension_weknora import WeKnoraProvider
|
||||
|
||||
_, calls, state, config, transport = adapter
|
||||
space = config["spaces"][0]
|
||||
space.pop("user_ids")
|
||||
space["access_mode"] = "authenticated"
|
||||
space["knowledge_base_ids"] = ["kb-1"]
|
||||
return WeKnoraProvider(config, transport=transport), calls, state, config
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("user", ["new-registration-a", "new-registration-b"])
|
||||
async def test_shared_new_users_can_browse_and_search_without_user_bindings(shared, user):
|
||||
provider, calls, *_ = shared
|
||||
context = KnowledgeRequestContext(user, "lead_agent", authenticated_user_id=user)
|
||||
catalog = await provider.catalog(context=context, page=1, page_size=20, search="")
|
||||
assert len(catalog.items) == 1
|
||||
dataset = catalog.items[0].id
|
||||
documents = await provider.documents(dataset, context=context, page=1, page_size=20, search="")
|
||||
assert documents.items[0].id == "doc-1"
|
||||
selected = KnowledgeRequestContext(user, "lead_agent", {"version": 1, "mode": "selected", "dataset_ids": [dataset]}, authenticated_user_id=user)
|
||||
assert (await provider.search_scoped("Cedar", context=selected)).hits
|
||||
await provider.entries_search("Cedar", context=selected)
|
||||
assert all(r.method == "GET" or r.url.path.endswith(("/hybrid-search", "/wiki/search")) for r in calls)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(("user", "verified"), [("new-user", None), ("default", None), ("new-user", "other-user"), ("", "")])
|
||||
async def test_shared_requires_matching_verified_identity_before_http(shared, user, verified):
|
||||
provider, calls, *_ = shared
|
||||
context = KnowledgeRequestContext(user, "lead_agent", authenticated_user_id=verified)
|
||||
with pytest.raises(PermissionError):
|
||||
await provider.catalog(context=context, page=1, page_size=20, search="")
|
||||
assert not calls
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shared_still_enforces_agent_and_base_restrictions(shared):
|
||||
provider, calls, state, _ = shared
|
||||
with pytest.raises(PermissionError):
|
||||
await provider.catalog(context=KnowledgeRequestContext("new-user", "forbidden-agent", authenticated_user_id="new-user"), page=1, page_size=20, search="")
|
||||
assert not calls
|
||||
state["bases"] = [{"id": "kb-1", "name": "Allowed"}, {"id": "kb-private", "name": "Private"}]
|
||||
context = KnowledgeRequestContext("new-user", "researcher", authenticated_user_id="new-user")
|
||||
catalog = await provider.catalog(context=context, page=1, page_size=20, search="")
|
||||
assert [item.name for item in catalog.items] == ["Allowed"]
|
||||
state["document"]["knowledge_base_id"] = "kb-private"
|
||||
with pytest.raises(PermissionError):
|
||||
await provider.search_scoped("Cedar", context=context)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"changes",
|
||||
[
|
||||
{},
|
||||
{"user_ids": None},
|
||||
{"user_ids": []},
|
||||
{"access_mode": "all"},
|
||||
{"access_mode": "authenticated", "user_ids": ["alice"]},
|
||||
{"access_mode": "authenticated", "user_ids": []},
|
||||
{"access_mode": True},
|
||||
],
|
||||
)
|
||||
def test_access_mode_must_be_explicit_and_unambiguous(adapter, changes):
|
||||
from deerflow_extension_weknora import WeKnoraProvider
|
||||
|
||||
config = adapter[3]
|
||||
space = dict(config["spaces"][0])
|
||||
space.pop("user_ids")
|
||||
with pytest.raises(ValueError):
|
||||
WeKnoraProvider({"spaces": [{**space, **changes}]})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_verified_user_does_not_bypass_default_allowlist(adapter):
|
||||
provider, calls, *_ = adapter
|
||||
with pytest.raises(PermissionError):
|
||||
await provider.catalog(context=KnowledgeRequestContext("new-user", "lead_agent", authenticated_user_id="new-user"), page=1, page_size=20, search="")
|
||||
assert not calls
|
||||
|
||||
|
||||
@pytest.mark.parametrize("credential", ["session", "pat", "anonymous", "auth_disabled", "internal"])
|
||||
def test_gateway_propagates_only_verified_user_to_shared_catalog_and_tool(shared, monkeypatch, credential):
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.testclient import TestClient
|
||||
from langchain_core.messages import AIMessage
|
||||
from langgraph.graph import END, START, MessagesState, StateGraph
|
||||
from langgraph.prebuilt import ToolNode
|
||||
|
||||
from app.gateway.auth.models import User
|
||||
from app.gateway.auth_middleware import AuthMiddleware
|
||||
from app.gateway.deps import get_config
|
||||
from app.gateway.internal_auth import create_internal_auth_headers
|
||||
from app.gateway.routers import knowledge
|
||||
from deerflow.config.app_config import AppConfig
|
||||
from deerflow.config.authorization_config import AuthorizationConfig
|
||||
from deerflow.extensions import EXTENSION_SNAPSHOT_CONTEXT_KEY, ExtensionRegistry
|
||||
from deerflow.knowledge.tools import knowledge_search_tool
|
||||
|
||||
provider, calls, *_ = shared
|
||||
users = [User(email=f"shared-{i}@example.com", password_hash="unused", system_role="user") for i in range(2)]
|
||||
|
||||
async def resolve_user(request):
|
||||
return users[int(request.cookies["access_token"])]
|
||||
|
||||
async def resolve_pat(app, authorization):
|
||||
return users[0], frozenset({"threads:read"})
|
||||
|
||||
monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False)
|
||||
monkeypatch.setattr("app.gateway.deps.get_current_user_from_request", resolve_user)
|
||||
monkeypatch.setattr("app.gateway.auth.pat.authenticate_pat", resolve_pat)
|
||||
monkeypatch.setattr("app.gateway.auth.pat.is_pat_allowed_route", lambda *args: True)
|
||||
monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: AuthorizationConfig())
|
||||
registry = ExtensionRegistry()
|
||||
with registry.attributed_to("deerflow_extension_weknora:install"):
|
||||
registry.knowledge_provider("weknora", provider)
|
||||
loaded = registry.build()
|
||||
config = AppConfig.model_validate(
|
||||
{
|
||||
"models": [],
|
||||
"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
|
||||
"knowledge_base": {"enabled": True, "scope_selection_enabled": True},
|
||||
"tools": [{"name": "knowledge_search", "group": "knowledge", "use": "deerflow.knowledge.tools:knowledge_search_tool", "provider": "weknora"}],
|
||||
}
|
||||
)
|
||||
app = FastAPI()
|
||||
app.add_middleware(AuthMiddleware)
|
||||
app.state.extensions = loaded
|
||||
app.include_router(knowledge.router)
|
||||
app.dependency_overrides[get_config] = lambda: config
|
||||
graph_builder = StateGraph(MessagesState, context_schema=dict)
|
||||
graph_builder.add_node("tools", ToolNode([knowledge_search_tool]))
|
||||
graph_builder.add_edge(START, "tools")
|
||||
graph_builder.add_edge("tools", END)
|
||||
graph = graph_builder.compile()
|
||||
|
||||
@app.get("/api/shared-search")
|
||||
async def search(request: Request):
|
||||
# These forged values must never become authentication evidence.
|
||||
context = {
|
||||
"user_id": str(request.state.user.id),
|
||||
"authenticated_user_id": str(request.state.user.id),
|
||||
"authenticated": True,
|
||||
"app_config": config,
|
||||
EXTENSION_SNAPSHOT_CONTEXT_KEY: loaded,
|
||||
}
|
||||
if request.query_params.get("child"):
|
||||
context.update(is_subagent=True, agent_id="researcher")
|
||||
result = await asyncio.create_task(
|
||||
graph.ainvoke(
|
||||
{"messages": [AIMessage(content="", tool_calls=[{"id": "shared-search", "name": "knowledge_search", "args": {"query": "Cedar"}}])]},
|
||||
context=context,
|
||||
)
|
||||
)
|
||||
message = result["messages"][-1]
|
||||
return {"content": message.content, "artifact": message.artifact}
|
||||
|
||||
headers = {}
|
||||
cookies = {}
|
||||
if credential == "session":
|
||||
cookies["access_token"] = "0"
|
||||
elif credential == "pat":
|
||||
headers["Authorization"] = "Bearer synthetic-pat"
|
||||
elif credential == "auth_disabled":
|
||||
monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
|
||||
elif credential == "internal":
|
||||
headers = create_internal_auth_headers()
|
||||
|
||||
allowed = credential in {"session", "pat"}
|
||||
with TestClient(app, headers=headers, cookies=cookies) as client:
|
||||
catalog = client.get("/api/knowledge/retrieval-catalog/datasets?agent_name=lead_agent&authenticated=true&user_id=alice")
|
||||
assert catalog.status_code == (200 if allowed else 401 if credential == "anonymous" else 403)
|
||||
search = client.get("/api/shared-search")
|
||||
if allowed:
|
||||
assert search.json()["artifact"]["knowledge_sources"]
|
||||
assert client.get("/api/shared-search?child=1").json()["artifact"]["knowledge_sources"]
|
||||
dataset = catalog.json()["items"][0]["id"]
|
||||
assert client.get(f"/api/knowledge/retrieval-catalog/datasets/{dataset}/documents?agent_name=lead_agent").status_code == 200
|
||||
if credential == "session":
|
||||
client.cookies.set("access_token", "1")
|
||||
assert client.get("/api/knowledge/retrieval-catalog/datasets?agent_name=lead_agent").status_code == 200
|
||||
else:
|
||||
assert search.status_code == 401 or search.json()["artifact"] is None
|
||||
assert not calls
|
||||
@ -11,6 +11,10 @@ into an unrestricted search. Use KB-bound `hybrid-search` with optional
|
||||
`knowledge_ids` and disabled context enrichment. The global search endpoint
|
||||
rejects document targets for KB-restricted keys and unions base/document target
|
||||
lists. Never broaden credentials to work around that. Verify result membership.
|
||||
Default `access_mode: users` requires exact user IDs. Explicit `authenticated`
|
||||
mode omits user_ids and requires nonempty host authenticated_user_id == user_id;
|
||||
never infer login from a default/storage ID or serialized context. Retain Agent,
|
||||
base and upstream checks. Shared retrieval grants no management permissions.
|
||||
Reject malformed document totals and detail shapes as provider errors; reserve
|
||||
permission errors for resource identity, membership and searchability failures.
|
||||
Dataset IDs derive from endpoint/tenant/native base, not credentials or labels.
|
||||
|
||||
@ -79,22 +79,65 @@ from chunk offsets, storage paths or response indexes.
|
||||
`user_ids` are exact server-authenticated DeerFlow IDs, not display names or
|
||||
model-supplied values. Resolve them from your deployment's user administration;
|
||||
for an unauthenticated/single-user installation use its actual runtime user ID.
|
||||
There is no default or wildcard binding. To bind multiple workspaces, add a
|
||||
The default `access_mode: users` requires a nonempty list; missing, null or empty
|
||||
`user_ids` never grants shared access. There is no wildcard binding. To bind multiple workspaces, add a
|
||||
`spaces` entry per endpoint/tenant pair, each with its own explicit users and
|
||||
credential environment variable. Configure each endpoint/tenant only once.
|
||||
|
||||
### Shared read-only access for dynamic users
|
||||
|
||||
For a knowledge base intentionally shared with every authenticated DeerFlow user,
|
||||
replace `user_ids` on that connection with **explicit** `access_mode: authenticated`:
|
||||
|
||||
```yaml
|
||||
spaces:
|
||||
- base_url: https://weknora.example.com
|
||||
tenant_id: "42"
|
||||
api_key_env: WEKNORA_SHARED_READ_KEY
|
||||
access_mode: authenticated
|
||||
knowledge_base_ids: ["YOUR_SHARED_WEKNORA_KB_ID"]
|
||||
# agent_names: ["lead_agent", "researcher"]
|
||||
```
|
||||
|
||||
Set the credential environment variable and restart Gateway after this operator
|
||||
configuration change. Newly registered users with a valid session can then use
|
||||
the connection without editing a user-ID list or restarting for each new user.
|
||||
PAT identities also qualify where existing host route/scope permissions allow.
|
||||
Do not combine `user_ids` with this mode; ambiguous configurations are rejected.
|
||||
|
||||
**All authenticated users receive this credential's visibility**, narrowed by
|
||||
`knowledge_base_ids` and optional `agent_names`. Use a dedicated least-privilege
|
||||
credential and list only bases intended for sharing. This grants retrieval,
|
||||
catalog and Wiki search only; it does not grant management, space membership or
|
||||
per-user WeKnora impersonation. Existing host tool/route authorization still applies.
|
||||
|
||||
This mode requires a Gateway host that supplies the optional
|
||||
`KnowledgeRequestContext.authenticated_user_id` provenance field. Older hosts
|
||||
and standalone callers without that proof fail closed. Anonymous requests,
|
||||
auth-disabled synthetic users, internal-service identities and client-provided
|
||||
authentication flags never qualify. Scheduled-task runs, IM-channel runs and
|
||||
durable batch items currently cannot use this shared mode: their owner `user_id`
|
||||
does not supply the required authentication proof. Durable batch items are
|
||||
launched by the Gateway's startup poller, so they do not inherit the submitting
|
||||
request's proof even within the same process. A valid session is still accepted
|
||||
when the local auth-disabled switch is present. Ordinary subagents launched from
|
||||
an authenticated execution inherit its verified identity and retain their own
|
||||
Agent restrictions. If a worker loses that context, access is denied; it is not
|
||||
reconstructed from a stored user ID. Keep `access_mode: users` for intentionally
|
||||
unauthenticated single-user deployments or explicit internal-user bindings.
|
||||
|
||||
## Authorization and scope
|
||||
|
||||
Each operation resolves the trusted current user and executor against the
|
||||
operator-owned connection bindings. It sends `X-API-Key` and `X-Tenant-ID` to
|
||||
operator-owned connection access mode and bindings. It sends `X-API-Key` and `X-Tenant-ID` to
|
||||
WeKnora, whose API-key permissions and workspace/resource authorization remain
|
||||
final. The binding is a credential-routing decision: every listed user receives
|
||||
final. The binding is a credential-routing decision: every admitted user receives
|
||||
the resource visibility of that credential, further restricted by the optional
|
||||
base allowlist. It does **not** impersonate a WeKnora human user. Use dedicated
|
||||
least-privilege keys and only bind users intended to share that visibility.
|
||||
least-privilege keys and admit only users intended to share that visibility.
|
||||
|
||||
Tenant-scoped keys must match the configured tenant. Platform keys always need
|
||||
an explicit tenant and user binding here; unbound platform discovery and
|
||||
an explicit tenant and access policy here; unbound platform discovery and
|
||||
fallback-to-all behavior are not supported. Missing bindings or keys fail
|
||||
closed. API-key roles/capabilities must allow base/document reads and knowledge
|
||||
search; restricted keys that cannot enumerate the configured bases are rejected.
|
||||
|
||||
@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, field_validator, model_validator
|
||||
|
||||
@ -20,10 +21,19 @@ class Space(BaseModel):
|
||||
base_url: AnyHttpUrl
|
||||
tenant_id: str = Field(pattern=r"^[1-9][0-9]{0,19}$")
|
||||
api_key_env: str = Field(pattern=r"^[A-Za-z_][A-Za-z0-9_]*$")
|
||||
user_ids: tuple[str, ...] = Field(min_length=1)
|
||||
access_mode: Literal["users", "authenticated"] = "users"
|
||||
user_ids: tuple[str, ...] | None = Field(default=None, min_length=1)
|
||||
agent_names: tuple[str, ...] | None = Field(default=None, min_length=1)
|
||||
knowledge_base_ids: tuple[str, ...] | None = Field(default=None, min_length=1, max_length=100)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def explicit_access(self) -> Space:
|
||||
if self.access_mode == "users" and self.user_ids is None:
|
||||
raise ValueError("users access requires explicit user_ids")
|
||||
if self.access_mode == "authenticated" and self.user_ids is not None:
|
||||
raise ValueError("authenticated access must omit user_ids")
|
||||
return self
|
||||
|
||||
@field_validator("base_url")
|
||||
@classmethod
|
||||
def safe_url(cls, value: AnyHttpUrl) -> AnyHttpUrl:
|
||||
@ -58,5 +68,5 @@ class Settings(BaseModel):
|
||||
def unique_connections(self) -> Settings:
|
||||
keys = [(str(space.base_url).rstrip("/"), space.tenant_id) for space in self.spaces]
|
||||
if len(set(keys)) != len(keys):
|
||||
raise ValueError("Configure each endpoint/tenant once, with its explicit user bindings")
|
||||
raise ValueError("Configure each endpoint/tenant once, with its explicit access mode")
|
||||
return self
|
||||
|
||||
@ -76,7 +76,12 @@ class WeKnoraProvider:
|
||||
def _bindings(self, context: KnowledgeRequestContext) -> list[_Binding]:
|
||||
bindings = []
|
||||
for space in self.settings.spaces:
|
||||
if context.user_id not in space.user_ids or (space.agent_names is not None and context.agent_name not in space.agent_names):
|
||||
if space.access_mode == "users":
|
||||
user_allowed = context.user_id in (space.user_ids or ())
|
||||
else:
|
||||
# Older harness contexts lack this field; deny shared access instead of raising AttributeError.
|
||||
user_allowed = bool(context.user_id) and getattr(context, "authenticated_user_id", None) == context.user_id
|
||||
if not user_allowed or (space.agent_names is not None and context.agent_name not in space.agent_names):
|
||||
continue
|
||||
key = os.getenv(space.api_key_env, "").strip()
|
||||
if not key or not key.isascii() or any(ord(char) < 33 or ord(char) == 127 for char in key):
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "deerflow-extension-weknora"
|
||||
version = "0.1.0"
|
||||
version = "0.1.1"
|
||||
description = "Read-only WeKnora retrieval and scope provider for DeerFlow"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user