core/config: networking ssrf_assume_public_cidrs for fetch Fake-IP

This commit is contained in:
2026-07-24 20:20:30 +08:00
parent f3cefccefc
commit 0ca92dfd7e
11 changed files with 189 additions and 7 deletions
+2 -1
View File
@@ -19,7 +19,8 @@
- components: Utilities for class composition.
- memory: Storage controlling for sessions and messages.
- router: Multi-source capability routing (Phase H; not implemented).
- config: Plyngent configuration center (TOML), including ``[plugins]``.
- config: Plyngent configuration center (TOML), including ``[plugins]`` and
``[networking]`` (e.g. fetch SSRF Fake-IP CIDR exemptions).
- agent: Tool loop, streaming, usage, compact; `@tool` / tags / registry.
- tools: Workspace file/process/VCS/chat/todo/net tools; catalog; plugins;
instance/session context and views.
+7
View File
@@ -19,6 +19,13 @@
# enable = ["*"]
# disable = ["legacy"]
# Host network / SSRF policy for tools such as ``fetch``.
# Clash TUN + Fake-IP often resolves public names to 198.18.0.0/15; list that
# pool so fetch does not treat every site as private/loopback.
# Metadata (169.254.169.254) stays forbidden even if listed.
# [networking]
# ssrf_assume_public_cidrs = ["198.18.0.0/15"]
[agent]
# Persona (omit → built-in coding-agent line). tool_directives is the tool playbook.
# Override with multi-line literals (prefer ''' so nested " is fine):
+10 -1
View File
@@ -299,6 +299,10 @@ async def _run_chat( # noqa: C901, PLR0912, PLR0915 — chat orchestration
)
# Path denylist and policy confirm live on instance.workspace (no process bag).
state.instance_state.workspace.path_denylist = tuple(store.agent_config.path_denylist or ())
from plyngent.tools.net import set_ssrf_assume_public_cidrs
# Clash Fake-IP etc.: treat configured CIDRs as public for fetch SSRF.
_ = set_ssrf_assume_public_cidrs(store.networking_config.ssrf_assume_public_cidrs)
if interactive:
from plyngent.cli.limits import prompt_policy_command_confirm, prompt_policy_fetch_confirm
from plyngent.tools.net import set_fetch_policy_confirm_hook
@@ -339,10 +343,15 @@ async def _run_chat( # noqa: C901, PLR0912, PLR0915 — chat orchestration
if isinstance(state_obj, ReplState):
state_obj.instance_state.workspace.policy_confirm_hook = None
state_obj.instance_state.workspace.policy_allowed_commands.clear()
from plyngent.tools.net import clear_private_grants, set_fetch_policy_confirm_hook
from plyngent.tools.net import (
clear_private_grants,
clear_ssrf_assume_public_cidrs,
set_fetch_policy_confirm_hook,
)
set_fetch_policy_confirm_hook(None, instance=state_obj.instance_state)
clear_private_grants(instance=state_obj.instance_state)
clear_ssrf_assume_public_cidrs()
await state_obj.instance_state.shutdown()
else:
# No ReplState: only PTY class cleanup (temps require instance allowlist).
+1
View File
@@ -13,6 +13,7 @@ from .models import DatabaseConfig as DatabaseConfig
from .models import DeepseekProvider as DeepseekProvider
from .models import HttpTimeoutConfig as HttpTimeoutConfig
from .models import ModelConfig as ModelConfig
from .models import NetworkingConfig as NetworkingConfig
from .models import OpenAICompatibleProvider as OpenAICompatibleProvider
from .models import OpenAIProvider as OpenAIProvider
from .models import PluginsConfig as PluginsConfig
+15
View File
@@ -126,6 +126,21 @@ class PluginsConfig(Struct, omit_defaults=True):
disable: list[str] = field(default_factory=list)
class NetworkingConfig(Struct, omit_defaults=True):
"""Host-side network policy for tools such as ``fetch``.
``ssrf_assume_public_cidrs``: CIDR strings whose resolved addresses are treated
as **public for SSRF policy** (skip private-host grants). Use for Clash/meta
Fake-IP pools (commonly ``198.18.0.0/15``). Empty default = no exemptions.
Metadata ranges (e.g. ``169.254.169.254``) stay forbidden even if listed here.
Real loopback/RFC1918 literals are unaffected unless you put those CIDRs here
(not recommended).
"""
ssrf_assume_public_cidrs: list[str] = field(default_factory=list)
class ModelConfig(Struct, omit_defaults=True):
"""Capability flags for a model within a provider."""
+36 -1
View File
@@ -6,7 +6,14 @@ from typing import TYPE_CHECKING, Any, cast
import msgspec
import tomlkit
from .models import AgentConfig, DatabaseConfig, ModelConfig, PluginsConfig, Provider
from .models import (
AgentConfig,
DatabaseConfig,
ModelConfig,
NetworkingConfig,
PluginsConfig,
Provider,
)
if TYPE_CHECKING:
from collections.abc import Mapping, MutableMapping, Sequence
@@ -41,6 +48,14 @@ def _parse_plugins(raw: dict[str, object]) -> PluginsConfig:
return PluginsConfig()
def _parse_networking(raw: dict[str, object]) -> NetworkingConfig:
"""Parse the ``[networking]`` section, falling back to defaults."""
try:
return msgspec.convert(raw, NetworkingConfig)
except msgspec.ValidationError:
return NetworkingConfig()
def _parse_providers(
document: tomlkit.TOMLDocument,
) -> tuple[dict[str, Provider], dict[str, object], dict[str, Provider]]:
@@ -109,6 +124,7 @@ class ConfigStore:
_database: DatabaseConfig
_agent: AgentConfig
_plugins: PluginsConfig
_networking: NetworkingConfig
_providers: dict[str, Provider]
_bad_providers: dict[str, object]
_recoverable_providers: dict[str, Provider]
@@ -120,6 +136,7 @@ class ConfigStore:
self._database = _parse_database(cast("dict[str, object]", raw.get("database", {})))
self._agent = _parse_agent(cast("dict[str, object]", raw.get("agent", {})))
self._plugins = _parse_plugins(cast("dict[str, object]", raw.get("plugins", {})))
self._networking = _parse_networking(cast("dict[str, object]", raw.get("networking", {})))
self._providers, self._bad_providers, self._recoverable_providers = _parse_providers(document)
@property
@@ -158,6 +175,18 @@ class ConfigStore:
"""Typed plugin allowlist (enable / disable entry-point names)."""
return self._plugins
# -- networking (read-only) --
@property
def networking(self) -> MappingProxyType[str, object]:
"""Read-only mapping view of networking / SSRF-related configuration."""
return MappingProxyType(msgspec.structs.asdict(self._networking))
@property
def networking_config(self) -> NetworkingConfig:
"""Typed networking section (e.g. Fake-IP SSRF exemptions)."""
return self._networking
def set_plugins_enable(self, names: Sequence[str]) -> PluginsConfig:
"""Replace the plugin enable list (in-memory; call :meth:`write` to persist)."""
cleaned = [name.strip() for name in names if name.strip()]
@@ -360,6 +389,7 @@ class ConfigStore:
self._database = _parse_database(cast("dict[str, object]", raw.get("database", {})))
self._agent = _parse_agent(cast("dict[str, object]", raw.get("agent", {})))
self._plugins = _parse_plugins(cast("dict[str, object]", raw.get("plugins", {})))
self._networking = _parse_networking(cast("dict[str, object]", raw.get("networking", {})))
self._providers, self._bad_providers, self._recoverable_providers = _parse_providers(self._document)
# -- internal sync helpers --
@@ -413,6 +443,10 @@ class ConfigStore:
"""Sync ``[plugins]`` to the document."""
self._sync_section("plugins", self._plugins)
def _sync_networking_section(self) -> None:
"""Sync ``[networking]`` to the document."""
self._sync_section("networking", self._networking)
@staticmethod
def _to_inline_table(mapping: Mapping[str, object]) -> object:
"""Encode a flat (or nested-dict) mapping as a TOML inline table."""
@@ -471,4 +505,5 @@ class ConfigStore:
self._sync_database_section()
self._sync_agent_section()
self._sync_plugins_section()
self._sync_networking_section()
self._sync_providers_section()
+2
View File
@@ -30,9 +30,11 @@ from .file import tree as tree
from .file import write_file as write_file
from .net import NET_TOOLS as NET_TOOLS
from .net import clear_private_grants as clear_private_grants
from .net import clear_ssrf_assume_public_cidrs as clear_ssrf_assume_public_cidrs
from .net import fetch as fetch
from .net import grant_private_host as grant_private_host
from .net import set_fetch_policy_confirm_hook as set_fetch_policy_confirm_hook
from .net import set_ssrf_assume_public_cidrs as set_ssrf_assume_public_cidrs
from .plugins import load_plugin_tools as load_plugin_tools
from .process import PROCESS_TOOLS as PROCESS_TOOLS
from .process import ask_into_pty as ask_into_pty
+2
View File
@@ -2,6 +2,8 @@ from .fetch import fetch as fetch
from .grants import clear_private_grants as clear_private_grants
from .grants import grant_private_host as grant_private_host
from .grants import set_fetch_policy_confirm_hook as set_fetch_policy_confirm_hook
from .policy import clear_ssrf_assume_public_cidrs as clear_ssrf_assume_public_cidrs
from .policy import set_ssrf_assume_public_cidrs as set_ssrf_assume_public_cidrs
NET_TOOLS = [
fetch,
+68 -4
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import ipaddress
import socket
from contextvars import ContextVar
from dataclasses import dataclass
from enum import Enum
from typing import TYPE_CHECKING
@@ -44,6 +45,8 @@ _NEVER_ALLOW_NETWORKS: tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...]
ipaddress.ip_network("fd00:ec2::254/128"), # AWS IMDS IPv6
)
type IpNetwork = ipaddress.IPv4Network | ipaddress.IPv6Network
class HostClass(Enum):
PUBLIC = "public"
@@ -55,6 +58,46 @@ class FetchPolicyError(ValueError):
"""Hard fetch policy violation (returned to the model as error text)."""
# Configured via ``[networking].ssrf_assume_public_cidrs`` (e.g. Clash Fake-IP).
_ssrf_assume_public_networks: ContextVar[tuple[IpNetwork, ...]] = ContextVar(
"plyngent_ssrf_assume_public_networks",
default=(),
)
def parse_ssrf_assume_public_cidrs(cidrs: Sequence[str] | None) -> tuple[IpNetwork, ...]:
"""Parse CIDR strings; invalid entries raise :class:`FetchPolicyError`."""
if not cidrs:
return ()
out: list[IpNetwork] = []
for raw in cidrs:
text = raw.strip()
if not text:
continue
try:
out.append(ipaddress.ip_network(text, strict=False))
except ValueError as exc:
msg = f"invalid ssrf_assume_public_cidrs entry {text!r}: {exc}"
raise FetchPolicyError(msg) from exc
return tuple(out)
def set_ssrf_assume_public_cidrs(cidrs: Sequence[str] | None) -> tuple[IpNetwork, ...]:
"""Install process/context networks treated as public for SSRF (config load)."""
networks = parse_ssrf_assume_public_cidrs(cidrs)
_ = _ssrf_assume_public_networks.set(networks)
return networks
def get_ssrf_assume_public_networks() -> tuple[IpNetwork, ...]:
return _ssrf_assume_public_networks.get()
def clear_ssrf_assume_public_cidrs() -> None:
"""Reset Fake-IP / assume-public exemptions (tests / chat exit)."""
_ = _ssrf_assume_public_networks.set(())
@dataclass(frozen=True, slots=True)
class ParsedFetchUrl:
"""Normalized URL pieces used for policy and the HTTP client."""
@@ -126,7 +169,15 @@ def resolve_redirect_url(current: str, location: str) -> str:
return parse_fetch_url(joined).url
def _ip_classification(address: str) -> HostClass:
def _ip_in_networks(ip: ipaddress.IPv4Address | ipaddress.IPv6Address, networks: Sequence[IpNetwork]) -> bool:
return any(ip in network for network in networks)
def _ip_classification(
address: str,
*,
assume_public: Sequence[IpNetwork] = (),
) -> HostClass:
try:
ip = ipaddress.ip_address(address)
except ValueError:
@@ -134,19 +185,32 @@ def _ip_classification(address: str) -> HostClass:
for network in _NEVER_ALLOW_NETWORKS:
if ip in network:
return HostClass.FORBIDDEN
# Clash Fake-IP (etc.): treat configured synthetic pools as public for policy.
# Metadata above still wins; do not list real loopback here.
if assume_public and _ip_in_networks(ip, assume_public):
return HostClass.PUBLIC
# IPv6 unique-local / link-local / loopback / unspecified
if ip.is_loopback or ip.is_link_local or ip.is_private or ip.is_reserved or ip.is_multicast or ip.is_unspecified:
return HostClass.PRIVATE
return HostClass.PUBLIC
def classify_ip_strings(addresses: Sequence[str]) -> HostClass:
"""Worst-class wins: FORBIDDEN > PRIVATE > PUBLIC."""
def classify_ip_strings(
addresses: Sequence[str],
*,
assume_public: Sequence[IpNetwork] | None = None,
) -> HostClass:
"""Worst-class wins: FORBIDDEN > PRIVATE > PUBLIC.
*assume_public* defaults to the process context from
:func:`set_ssrf_assume_public_cidrs` (``[networking].ssrf_assume_public_cidrs``).
"""
if not addresses:
return HostClass.FORBIDDEN
networks = get_ssrf_assume_public_networks() if assume_public is None else tuple(assume_public)
worst = HostClass.PUBLIC
for addr in addresses:
kind = _ip_classification(addr)
kind = _ip_classification(addr, assume_public=networks)
if kind is HostClass.FORBIDDEN:
return HostClass.FORBIDDEN
if kind is HostClass.PRIVATE:
+18
View File
@@ -28,6 +28,24 @@ def test_agent_section_defaults(tmp_path: Path) -> None:
assert store.agent_config.max_context_tokens == 200_000
assert store.plugins_config.enable == []
assert store.plugins_config.disable == []
assert store.networking_config.ssrf_assume_public_cidrs == []
def test_networking_ssrf_assume_public_cidrs(tmp_path: Path) -> None:
path = tmp_path / "c.toml"
_ = path.write_text(
"""
[networking]
ssrf_assume_public_cidrs = ["198.18.0.0/15", "fdfe:dcba:9876::/48"]
""",
encoding="utf-8",
)
store = load(path)
assert store.networking_config.ssrf_assume_public_cidrs == [
"198.18.0.0/15",
"fdfe:dcba:9876::/48",
]
assert "ssrf_assume_public_cidrs" in store.networking
def test_compose_defaults_join_persona_and_directives() -> None:
+28
View File
@@ -132,6 +132,34 @@ def test_classify_ips() -> None:
assert classify_ip_strings(["8.8.8.8", "10.0.0.1"]) is HostClass.PRIVATE
def test_ssrf_assume_public_cidrs_fake_ip() -> None:
from plyngent.tools.net.policy import (
clear_ssrf_assume_public_cidrs,
parse_ssrf_assume_public_cidrs,
set_ssrf_assume_public_cidrs,
)
# Clash Fake-IP style address is private without exemption.
assert classify_ip_strings(["198.18.0.1"]) is HostClass.PRIVATE
nets = set_ssrf_assume_public_cidrs(["198.18.0.0/15"])
assert nets
assert classify_ip_strings(["198.18.0.1"]) is HostClass.PUBLIC
# Explicit assume_public overrides empty context.
clear_ssrf_assume_public_cidrs()
assert classify_ip_strings(["198.18.0.1"]) is HostClass.PRIVATE
assert (
classify_ip_strings(
["198.18.0.1"],
assume_public=parse_ssrf_assume_public_cidrs(["198.18.0.0/15"]),
)
is HostClass.PUBLIC
)
# Metadata never becomes public via assume list.
set_ssrf_assume_public_cidrs(["169.254.0.0/16", "198.18.0.0/15"])
assert classify_ip_strings(["169.254.169.254"]) is HostClass.FORBIDDEN
clear_ssrf_assume_public_cidrs()
def test_soft_confirm_reason_matrix() -> None:
assert soft_confirm_reason(method="GET", url="https://ex.com/", scheme="https", body_present=False) is None
http_reason = soft_confirm_reason(method="GET", url="http://ex.com/", scheme="http", body_present=False)