diff --git a/doc/architecture.md b/doc/architecture.md index ce410a0..cbbae20 100644 --- a/doc/architecture.md +++ b/doc/architecture.md @@ -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. diff --git a/doc/plyngent.example.toml b/doc/plyngent.example.toml index 3125034..30bb1ac 100644 --- a/doc/plyngent.example.toml +++ b/doc/plyngent.example.toml @@ -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): diff --git a/src/plyngent/cli/app.py b/src/plyngent/cli/app.py index 0860d3c..7fe021d 100644 --- a/src/plyngent/cli/app.py +++ b/src/plyngent/cli/app.py @@ -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). diff --git a/src/plyngent/config/__init__.py b/src/plyngent/config/__init__.py index 7287b32..a00074c 100644 --- a/src/plyngent/config/__init__.py +++ b/src/plyngent/config/__init__.py @@ -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 diff --git a/src/plyngent/config/models.py b/src/plyngent/config/models.py index 2529db5..0fe9f65 100644 --- a/src/plyngent/config/models.py +++ b/src/plyngent/config/models.py @@ -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.""" diff --git a/src/plyngent/config/store.py b/src/plyngent/config/store.py index 19a43fd..78f07f7 100644 --- a/src/plyngent/config/store.py +++ b/src/plyngent/config/store.py @@ -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() diff --git a/src/plyngent/tools/__init__.py b/src/plyngent/tools/__init__.py index 3a9b2c2..1daaca4 100644 --- a/src/plyngent/tools/__init__.py +++ b/src/plyngent/tools/__init__.py @@ -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 diff --git a/src/plyngent/tools/net/__init__.py b/src/plyngent/tools/net/__init__.py index 9695e6e..6fbf082 100644 --- a/src/plyngent/tools/net/__init__.py +++ b/src/plyngent/tools/net/__init__.py @@ -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, diff --git a/src/plyngent/tools/net/policy.py b/src/plyngent/tools/net/policy.py index 5a95aa3..0c9e4fb 100644 --- a/src/plyngent/tools/net/policy.py +++ b/src/plyngent/tools/net/policy.py @@ -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: diff --git a/tests/test_config/test_agent_section.py b/tests/test_config/test_agent_section.py index 337b912..b666c11 100644 --- a/tests/test_config/test_agent_section.py +++ b/tests/test_config/test_agent_section.py @@ -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: diff --git a/tests/test_tools/test_fetch.py b/tests/test_tools/test_fetch.py index d49c619..c88e908 100644 --- a/tests/test_tools/test_fetch.py +++ b/tests/test_tools/test_fetch.py @@ -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)