DATA
6de0bf9f5b
Vendored deer-flow upstream (bytedance/deer-flow) plus prompt-injection hardening: - New deerflow.security package: content_delimiter, html_cleaner, sanitizer (8 layers — invisible chars, control chars, symbols, NFC, PUA, tag chars, horizontal whitespace collapse with newline/tab preservation, length cap) - New deerflow.community.searx package: web_search, web_fetch, image_search backed by a private SearX instance, every external string sanitized and wrapped in <<<EXTERNAL_UNTRUSTED_CONTENT>>> delimiters - All native community web providers (ddg_search, tavily, exa, firecrawl, jina_ai, infoquest, image_search) replaced with hard-fail stubs that raise NativeWebToolDisabledError at import time, so a misconfigured tool.use path fails loud rather than silently falling back to unsanitized output - Native client back-doors (jina_client.py, infoquest_client.py) stubbed too - Native-tool tests quarantined under tests/_disabled_native/ (collect_ignore_glob via local conftest.py) - Sanitizer Layer 7 fix: only collapse horizontal whitespace, preserve newlines and tabs so list/table structure survives - Hardened runtime config.yaml references only the searx-backed tools - Factory overlay (backend/) kept in sync with deer-flow tree as a reference / source See HARDENING.md for the full audit trail and verification steps.
49 lines
1.8 KiB
Python
49 lines
1.8 KiB
Python
"""Configuration for pre-tool-call authorization."""
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class GuardrailProviderConfig(BaseModel):
|
|
"""Configuration for a guardrail provider."""
|
|
|
|
use: str = Field(description="Class path (e.g. 'deerflow.guardrails.builtin:AllowlistProvider')")
|
|
config: dict = Field(default_factory=dict, description="Provider-specific settings passed as kwargs")
|
|
|
|
|
|
class GuardrailsConfig(BaseModel):
|
|
"""Configuration for pre-tool-call authorization.
|
|
|
|
When enabled, every tool call passes through the configured provider
|
|
before execution. The provider receives tool name, arguments, and the
|
|
agent's passport reference, and returns an allow/deny decision.
|
|
"""
|
|
|
|
enabled: bool = Field(default=False, description="Enable guardrail middleware")
|
|
fail_closed: bool = Field(default=True, description="Block tool calls if provider errors")
|
|
passport: str | None = Field(default=None, description="OAP passport path or hosted agent ID")
|
|
provider: GuardrailProviderConfig | None = Field(default=None, description="Guardrail provider configuration")
|
|
|
|
|
|
_guardrails_config: GuardrailsConfig | None = None
|
|
|
|
|
|
def get_guardrails_config() -> GuardrailsConfig:
|
|
"""Get the guardrails config, returning defaults if not loaded."""
|
|
global _guardrails_config
|
|
if _guardrails_config is None:
|
|
_guardrails_config = GuardrailsConfig()
|
|
return _guardrails_config
|
|
|
|
|
|
def load_guardrails_config_from_dict(data: dict) -> GuardrailsConfig:
|
|
"""Load guardrails config from a dict (called during AppConfig loading)."""
|
|
global _guardrails_config
|
|
_guardrails_config = GuardrailsConfig.model_validate(data)
|
|
return _guardrails_config
|
|
|
|
|
|
def reset_guardrails_config() -> None:
|
|
"""Reset the cached config instance. Used in tests to prevent singleton leaks."""
|
|
global _guardrails_config
|
|
_guardrails_config = None
|