From 0291f36b013f4ba36214455bb2145ad98e8fa056 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 9 Jul 2026 11:22:55 +0800 Subject: [PATCH] feat(fetch_url): per-Connection allowed_url_hosts glob allowlist MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The default 'URL host shares >= 2 labels of suffix with yarn_rm_url host' rule is too strict for clusters whose hostnames are single-label (e.g. 'ccam1' through 'ccam99'). The user's cluster is reachable at http://ccam1:8088, and they want fetch_url to work for any ccamN host — but ccam1 and ccam50 share 0 suffix labels, so the existing rule rejects everything. Add an opt-in allowlist field on Connection: allowed_url_hosts: list[str] | None Each entry is an fnmatch glob pattern. The URL host is allowed if it matches ANY pattern, regardless of the suffix rule. Save a Connection with ['ccam*'] to allow ccam1, ccam2, ..., ccam99. SSRF safety: the implementation does NOT use vanilla fnmatch on the flat host string (because '*' in fnmatch crosses '.', so 'ccam*' would match 'ccam50.evil.com' — a security hole). Instead, both the pattern and the host are split on '.' and matched LABEL-BY-LABEL with the label counts required to match exactly. So 'ccam*' matches 'ccam50' but NOT 'ccam50.evil.com' (different label counts). Test coverage: - 8 new tests in tests/unit/test_fetch_url.py (glob allow, glob deny, dot-boundary SSRF test, multiple globs, empty list, None fallback, error message hint) - All 21 fetch_url tests pass; 377 total. - spark_executor/models.py: Connection.allowed_url_hosts (with description) - spark_executor/tools/requests.py: SaveConnectionRequest.allowed_url_hosts - spark_executor/tools/fetch_url.py: new _host_matches_any_glob helper; _validate_url_host now takes allowed_hosts and checks globs BEFORE the suffix rule - tests/unit/test_fetch_url.py: 8 new tests - README.md: fetch_url row mentions the allowlist Co-Authored-By: Claude Fable 5 --- README.md | 2 +- spark_executor/models.py | 15 ++++ spark_executor/tools/fetch_url.py | 77 ++++++++++++++----- spark_executor/tools/requests.py | 16 +++- tests/unit/test_fetch_url.py | 122 ++++++++++++++++++++++++++++++ 5 files changed, 209 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index 94e76a2..a6c68d4 100644 --- a/README.md +++ b/README.md @@ -120,7 +120,7 @@ MCP 客户端需要先执行 `initialize` 握手，拿到 `mcp-session-id` 后 | `get_external_job_status` | 查询**非本服务提交**的外部 YARN application 状态（按 `application_id` + `connection_name`） | | `get_external_job_result` | 查询外部 YARN application 终态结果视图 | | `get_external_job_logs` | 拉取外部 YARN application 的聚合日志 | -| `fetch_url` | 代理 HTTP GET 到集群内网 URL (受 `Connection.yarn_rm_url` host 围栏约束) | +| `fetch_url` | 代理 HTTP GET 到集群内网 URL (host 围栏: 默认 2 段后缀匹配 `yarn_rm_url`, 或匹配 `Connection.allowed_url_hosts` glob) | ### Files MCP 工具 diff --git a/spark_executor/models.py b/spark_executor/models.py index ce3b2ab..bf96719 100644 --- a/spark_executor/models.py +++ b/spark_executor/models.py @@ -69,6 +69,21 @@ class Connection(BaseModel): auth_principal: str | None = None auth_keytab: str | None = None + allowed_url_hosts: list[str] | None = Field( + default=None, + description=( + "Optional list of fnmatch glob patterns for hosts the fetch_url tool " + "may access, in addition to the default 'shares >= 2 labels of suffix " + "with yarn_rm_url host' rule. Useful for clusters whose hostnames do " + "NOT share a 2+ label suffix — e.g. single-label hosts like 'ccam1'-" + "'ccam99' (configure ['ccam*']) or HDFS namenode on a different " + "subdomain ('*.hadoop.internal'). Patterns are matched against the " + "URL host only (no port, no path). fnmatch rules apply: '*' does NOT " + "match '.', so 'ccam*' matches 'ccam50' but not 'ccam50.evil.com'. " + "Default None means the suffix rule alone applies." + ), + ) + @field_validator("master") @classmethod def _check_master(cls, v: str) -> str: diff --git a/spark_executor/tools/fetch_url.py b/spark_executor/tools/fetch_url.py index 6556613..63921c6 100644 --- a/spark_executor/tools/fetch_url.py +++ b/spark_executor/tools/fetch_url.py @@ -10,6 +10,7 @@ host; IP literals and non-HTTP schemes are rejected. Reuses the connection's saved auth/SSL config so the agent doesn't need cluster credentials. """ import ipaddress +import fnmatch from urllib.parse import urlparse import httpx @@ -36,40 +37,78 @@ def _host_suffix_overlap(host_a: str, host_b: str, min_labels: int = 2) -> bool: return n >= min_labels -def _validate_url_host(url: str, yarn_rm_url: str | None) -> None: - """Validate that url is an allowed http(s) hostname on the cluster network.""" +def _host_matches_any_glob(host: str, patterns: list[str]) -> bool: + """True if `host` matches any of the fnmatch glob patterns. + + fnmatch is case-sensitive on Linux (our deployment target). `*` in a + pattern does NOT cross '.' boundaries, so 'ccam*' matches 'ccam50' but + NOT 'ccam50.evil.com' (which is what we want — domain boundary matters + for SSRF). + """ + host_labels = host.split(".") + for p in patterns: + pat_labels = p.split(".") + if len(pat_labels) != len(host_labels): + continue + if all(fnmatch.fnmatchcase(h, pat) for h, pat in zip(host_labels, pat_labels)): + return True + return False + + +def _validate_url_host( + url: str, + yarn_rm_url: str | None, + allowed_hosts: list[str] | None = None, +) -> None: + """Raise ValueError if the URL is not allowed to be fetched. + + Allowed if EITHER: + - URL host matches one of `allowed_hosts` fnmatch globs (if provided) + - URL host shares >= 2 labels of suffix with `yarn_rm_url` host + """ parsed = urlparse(url) if parsed.scheme not in ("http", "https"): - raise ValueError(f"URL scheme must be http or https, got {parsed.scheme!r}") - + raise ValueError( + f"URL scheme must be http or https, got {parsed.scheme!r}" + ) host = parsed.hostname if not host: raise ValueError(f"URL has no host: {url!r}") - + # IP literal check try: ipaddress.ip_address(host) - except ValueError: - pass - else: raise ValueError( f"URL host {host!r} is an IP literal — IP targets are not allowed. " - "Use a hostname on the cluster network." + f"Use a hostname on the cluster network." ) - + except ValueError as e: + if "IP literal" in str(e): + raise + # not an IP, continue + # Glob allowlist — if any pattern matches, host is allowed regardless of suffix + if allowed_hosts and _host_matches_any_glob(host, allowed_hosts): + return if not yarn_rm_url: raise ValueError( - "Connection has no yarn_rm_url set, cannot validate URL host. " - "Save a Connection with yarn_rm_url first." + f"Connection has no yarn_rm_url set, cannot validate URL host " + f"(no allowed_url_hosts match either). Save a Connection with " + f"yarn_rm_url or allowed_url_hosts set." ) - - anchor = urlparse(yarn_rm_url).hostname or "" + anchor = urlparse(yarn_rm_url).hostname + if not anchor: + raise ValueError(f"Connection's yarn_rm_url has no host: {yarn_rm_url!r}") if not _host_suffix_overlap(host, anchor, min_labels=2): + allowed_hint = ( + f" Or set allowed_url_hosts=[''] on the connection to allow " + f"this host (e.g. ['ccam*'] for ccam1-ccam99)." + if not allowed_hosts + else f" (no allowed_url_hosts pattern matched either)" + ) raise ValueError( - f"URL host {host!r} does not share enough suffix with the connection's " - f"yarn_rm_url host {anchor!r} (need >= 2 labels of common suffix). " - "Reject this fetch to prevent SSRF." + f"URL host {host!r} does not share enough suffix with the " + f"connection's yarn_rm_url host {anchor!r} (need >= 2 labels of " + f"common suffix).{allowed_hint} Reject this fetch to prevent SSRF." ) - def fetch_url(url: str, connection_name: str) -> FetchUrlResult: """Proxy an HTTP GET to url using the auth/SSL settings of connection_name.""" @@ -78,7 +117,7 @@ def fetch_url(url: str, connection_name: str) -> FetchUrlResult: if conn is None: raise KeyError(f"Connection not found: {connection_name}") - _validate_url_host(url, conn.yarn_rm_url) + _validate_url_host(url, conn.yarn_rm_url, conn.allowed_url_hosts) config = YarnClientConfig.from_connection(conn) resp = httpx.get( diff --git a/spark_executor/tools/requests.py b/spark_executor/tools/requests.py index 20ca8da..37a4d6b 100644 --- a/spark_executor/tools/requests.py +++ b/spark_executor/tools/requests.py @@ -117,12 +117,22 @@ class SaveConnectionRequest(BaseModel): description=( "Absolute path to a Kerberos keytab file. Optional convenience " "for 'kinit -kt' workflows. The service does NOT auto-initialize " - "from the keytab — you must `kinit -kt ` " - "yourself before calling the tools." + "from the keytab — you must `kinit -kt ` " + "yourself before calling the tools." + ), + ) + + allowed_url_hosts: list[str] | None = Field( + default=None, + description=( + "Optional list of fnmatch glob patterns for hosts the fetch_url tool " + "may access. See Connection.allowed_url_hosts for full semantics. " + "Example for single-label host clusters: ['ccam*'] allows any host " + "starting with 'ccam' (ccam1, ccam2, ..., ccam99). Default None means " + "only the default 'shares >= 2 labels of suffix with yarn_rm_url' rule." ), ) - class PrepareSubmitJobRequest(BaseModel): connection: str = Field( ..., diff --git a/tests/unit/test_fetch_url.py b/tests/unit/test_fetch_url.py index 75b8e8f..29fcc3a 100644 --- a/tests/unit/test_fetch_url.py +++ b/tests/unit/test_fetch_url.py @@ -1,4 +1,5 @@ # coding=utf-8 +from pathlib import Path from unittest.mock import patch import httpx @@ -19,6 +20,12 @@ def _fresh_stores(tmp_path, monkeypatch): fetch_url.conn_store = store +@pytest.fixture +def fresh_stores(tmp_path: Path, monkeypatch): + """Reset connection store singletons for a single test.""" + _fresh_stores(tmp_path, monkeypatch) + yield + def test_fetch_url_returns_body_and_status(tmp_path, monkeypatch): _fresh_stores(tmp_path, monkeypatch) fetch_url.conn_store.save( @@ -162,3 +169,118 @@ def test_fetch_url_follows_redirects(tmp_path, monkeypatch): with patch("spark_executor.tools.fetch_url.httpx.get", return_value=resp) as m: fetch_url.fetch_url("http://nm01.prod.internal:8042/node", "prod") assert m.call_args.kwargs["follow_redirects"] is True + +def test_fetch_url_allows_host_matching_glob_pattern(fresh_stores): + fetch_url.conn_store.save( + Connection( + name="ccam", + master="yarn", + yarn_rm_url="http://ccam1:8088", + allowed_url_hosts=["ccam*"], + ) + ) + resp = httpx.Response(200, text="hello") + with patch("spark_executor.tools.fetch_url.httpx.get", return_value=resp) as m: + out = fetch_url.fetch_url("http://ccam50:8088/foo", "ccam") + assert out.status_code == 200 + assert out.body == "hello" + assert m.call_count == 1 + +def test_fetch_url_allows_host_matching_any_of_multiple_globs(fresh_stores): + fetch_url.conn_store.save( + Connection( + name="prod", + master="yarn", + yarn_rm_url="http://rm.prod.internal:8088", + allowed_url_hosts=["ccam*", "*.prod.internal"], + ) + ) + resp = httpx.Response(200, text="hello") + with patch("spark_executor.tools.fetch_url.httpx.get", return_value=resp) as m: + out = fetch_url.fetch_url("http://history.prod.internal:18080/api/v1/info", "prod") + assert out.status_code == 200 + assert out.body == "hello" + assert m.call_count == 1 + +def test_fetch_url_glob_does_not_match_unrelated_host(fresh_stores): + fetch_url.conn_store.save( + Connection( + name="ccam", + master="yarn", + yarn_rm_url="http://ccam1:8088", + allowed_url_hosts=["ccam*"], + ) + ) + with pytest.raises(ValueError, match="does not share enough suffix|no allowed_url_hosts pattern matched"): + fetch_url.fetch_url("http://evil.com/foo", "ccam") + +def test_fetch_url_glob_does_not_cross_dot_boundary(fresh_stores): + fetch_url.conn_store.save( + Connection( + name="ccam", + master="yarn", + yarn_rm_url="http://ccam1:8088", + allowed_url_hosts=["ccam*"], + ) + ) + with pytest.raises(ValueError, match="does not share enough suffix"): + fetch_url.fetch_url("http://ccam50.evil.com/", "ccam") + +def test_fetch_url_glob_match_does_not_require_suffix_overlap(fresh_stores): + fetch_url.conn_store.save( + Connection( + name="prod", + master="yarn", + yarn_rm_url="http://rm:8088", + allowed_url_hosts=["ccam*"], + ) + ) + resp = httpx.Response(200, text="hello") + with patch("spark_executor.tools.fetch_url.httpx.get", return_value=resp) as m: + out = fetch_url.fetch_url("http://ccam50:8088/", "prod") + assert out.status_code == 200 + assert out.body == "hello" + assert m.call_count == 1 + +def test_fetch_url_empty_allowed_hosts_falls_back_to_suffix_rule(fresh_stores): + fetch_url.conn_store.save( + Connection( + name="prod", + master="yarn", + yarn_rm_url="http://rm.prod.internal", + allowed_url_hosts=[], + ) + ) + resp = httpx.Response(200, text="hello") + with patch("spark_executor.tools.fetch_url.httpx.get", return_value=resp) as m: + out = fetch_url.fetch_url("http://nm.prod.internal/", "prod") + assert out.status_code == 200 + assert out.body == "hello" + assert m.call_count == 1 + +def test_fetch_url_none_allowed_hosts_falls_back_to_suffix_rule(fresh_stores): + fetch_url.conn_store.save( + Connection( + name="prod", + master="yarn", + yarn_rm_url="http://rm.prod.internal", + allowed_url_hosts=None, + ) + ) + resp = httpx.Response(200, text="hello") + with patch("spark_executor.tools.fetch_url.httpx.get", return_value=resp) as m: + out = fetch_url.fetch_url("http://nm.prod.internal/", "prod") + assert out.status_code == 200 + assert out.body == "hello" + assert m.call_count == 1 + +def test_fetch_url_error_message_hints_at_allowed_url_hosts(fresh_stores): + fetch_url.conn_store.save( + Connection( + name="prod", + master="yarn", + yarn_rm_url="http://rm.prod.internal:8088", + ) + ) + with pytest.raises(ValueError, match="allowed_url_hosts"): + fetch_url.fetch_url("http://evil.com/foo", "prod")