MCP Fetch 2026.10.10: URL แรกผ่านแล้ว ทำไมต้องตรวจทุก Redirect?

"""MOCK policy trace. No DNS, sockets, HTTP, MCP client, LLM, or file writes.

Authored for Python 3.12+. NOT RUN by Hybrid Mind.
This intentionally conservative teaching policy is NOT the upstream guard.
"""
from __future__ import annotations

import ipaddress
import json
from dataclasses import dataclass
from urllib.parse import urljoin, urlsplit


@dataclass(frozen=True)
class Fixture:
    addresses: tuple[str, ...]
    location: str | None = None


# A deliberately conservative fixture policy: special-purpose exceptions which
# upstream permits are still blocked here. Do not use this list as a production
# address registry; use reviewed, maintained controls in the actual transport.
DENY_RANGES = tuple(ipaddress.ip_network(value) for value in (
    "0.0.0.0/8", "10.0.0.0/8", "100.64.0.0/10", "127.0.0.0/8",
    "169.254.0.0/16", "172.16.0.0/12", "192.0.0.0/24", "192.0.2.0/24",
    "192.88.99.0/24", "192.168.0.0/16", "198.18.0.0/15",
    "198.51.100.0/24", "203.0.113.0/24", "224.0.0.0/4", "240.0.0.0/4",
    "::/128", "::1/128", "64:ff9b:1::/48", "100::/64", "2001::/23",
    "2001:db8::/32", "2002::/16", "fc00::/7", "fe80::/10", "ff00::/8",
))


def checked_host(url: str) -> str:
    if not isinstance(url, str) or not url or len(url) > 4096:
        raise ValueError("INVALID_URL")
    if any(ord(char) <= 32 or ord(char) == 127 for char in url):
        raise ValueError("INVALID_URL")
    try:
        parts = urlsplit(url)
        host = parts.hostname
        port = parts.port  # Accessing .port also validates malformed ports.
    except ValueError as exc:
        raise ValueError("INVALID_URL") from exc
    if parts.scheme not in {"https", "http"}:
        raise ValueError("UNSUPPORTED_SCHEME")
    if parts.username is not None or parts.password is not None:
        raise ValueError("USERINFO_NOT_ALLOWED")
    if not host or "%" in host or "\\" in url or parts.fragment:
        raise ValueError("INVALID_URL")
    if not host.isascii():
        raise ValueError("ASCII_HOST_REQUIRED_IN_MOCK")
    if port is not None and port != (443 if parts.scheme == "https" else 80):
        raise ValueError("PORT_NOT_ALLOWED_IN_MOCK")
    return host.lower().rstrip(".")


def public_fixture_address(value: str) -> bool:
    try:
        address = ipaddress.ip_address(value)
    except ValueError:
        return False
    if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped:
        address = address.ipv4_mapped
    if not address.is_global or address.is_multicast:
        return False
    return not any(address.version == net.version and address in net
                   for net in DENY_RANGES)


def trace_policy(initial_url: str, fixtures: dict[str, Fixture],
                 authorized_hosts: frozenset[str], max_redirects: int = 3) -> dict:
    """Return a synthetic trace; request_ledger means WOULD_REQUEST, not a send.

    Empty or unknown DNS fixtures fail closed here. Upstream's unresolved-host
    behavior differs: it leaves resolution to the transport/proxy.
    """
    if isinstance(max_redirects, bool) or not isinstance(max_redirects, int):
        raise ValueError("max_redirects must be an integer")
    if not 0 <= max_redirects <= 10:
        raise ValueError("max_redirects must be between 0 and 10")
    url = initial_url
    redirects = 0
    decisions: list[dict] = []
    ledger: list[dict] = []

    def finish(status: str, reason: str) -> dict:
        return {"mode": "MOCK", "status": status, "reason": reason,
                "redirects_followed": redirects,
                "request_ledger": ledger, "decisions": decisions}

    while True:
        try:
            host = checked_host(url)
        except ValueError as exc:
            decisions.append({"url": url, "decision": "BLOCK",
                              "reason": str(exc)})
            return finish("BLOCKED", str(exc))
        if host not in authorized_hosts:
            decisions.append({"url": url, "host": host,
                              "decision": "BLOCK", "reason": "HOST_NOT_AUTHORIZED"})
            return finish("BLOCKED", "HOST_NOT_AUTHORIZED")
        fixture = fixtures.get(url)
        if fixture is None or not fixture.addresses:
            decisions.append({"url": url, "host": host,
                              "decision": "BLOCK", "reason": "UNRESOLVED_FIXTURE"})
            return finish("BLOCKED", "UNRESOLVED_FIXTURE")
        refused = [address for address in fixture.addresses
                   if not public_fixture_address(address)]
        if refused:
            decisions.append({"url": url, "host": host, "decision": "BLOCK",
                              "reason": "NONPUBLIC_OR_INVALID_ADDRESS",
                              "refused_addresses": refused})
            return finish("BLOCKED", "NONPUBLIC_OR_INVALID_ADDRESS")
        decisions.append({"url": url, "host": host, "decision": "ALLOW",
                          "addresses": list(fixture.addresses)})
        # This line is deliberately after host, address, and URL validation.
        # Nothing here sends a request. It records only the modeled next action.
        ledger.append({"url": url, "action": "WOULD_REQUEST"})
        if fixture.location is None:
            return finish("COMPLETE", "END_OF_FIXTURE")
        if redirects >= max_redirects:
            return finish("BLOCKED", "REDIRECT_LIMIT")
        url = urljoin(url, fixture.location)
        redirects += 1


FIXTURES = {
    "https://docs.example/start": Fixture(("8.8.8.8",), "https://cdn.example/page"),
    "https://cdn.example/page": Fixture(("1.1.1.1",)),
    "https://docs.example/to-internal": Fixture(("8.8.8.8",), "http://metadata.example/info"),
    "http://metadata.example/info": Fixture(("169.254.169.254",)),
    "https://docs.example/mixed": Fixture(("8.8.8.8", "10.0.0.8")),
    "https://docs.example/loop": Fixture(("8.8.8.8",), "/loop"),
    "https://docs.example/unresolved": Fixture(()),
}
AUTHORIZED_HOSTS = frozenset({"docs.example", "cdn.example", "metadata.example"})


if __name__ == "__main__":
    print(json.dumps(trace_policy("https://docs.example/to-internal", FIXTURES,
                                  AUTHORIZED_HOSTS), ensure_ascii=False, indent=2))

"""Authored synthetic assertions, NOT RUN by Hybrid Mind."""
import unittest
from fetch_policy_trace import AUTHORIZED_HOSTS, FIXTURES, Fixture, trace_policy


class PolicyTraceTests(unittest.TestCase):
    def test_public_redirect_has_two_modeled_requests(self):
        result = trace_policy("https://docs.example/start", FIXTURES, AUTHORIZED_HOSTS)
        self.assertEqual(result["status"], "COMPLETE")
        self.assertEqual(len(result["request_ledger"]), 2)

    def test_private_redirect_is_not_in_request_ledger(self):
        result = trace_policy("https://docs.example/to-internal", FIXTURES, AUTHORIZED_HOSTS)
        self.assertEqual(result["status"], "BLOCKED")
        self.assertEqual(result["reason"], "NONPUBLIC_OR_INVALID_ADDRESS")
        self.assertEqual([row["url"] for row in result["request_ledger"]],
                         ["https://docs.example/to-internal"])

    def test_one_private_answer_blocks_mixed_set(self):
        result = trace_policy("https://docs.example/mixed", FIXTURES, AUTHORIZED_HOSTS)
        self.assertEqual(result["status"], "BLOCKED")
        self.assertEqual(result["request_ledger"], [])

    def test_public_host_still_needs_task_authorization(self):
        result = trace_policy("https://cdn.example/page", FIXTURES,
                              frozenset({"docs.example"}))
        self.assertEqual(result["reason"], "HOST_NOT_AUTHORIZED")
        self.assertEqual(result["request_ledger"], [])

    def test_unresolved_mock_fails_closed(self):
        result = trace_policy("https://docs.example/unresolved", FIXTURES, AUTHORIZED_HOSTS)
        self.assertEqual(result["reason"], "UNRESOLVED_FIXTURE")
        self.assertEqual(result["request_ledger"], [])

    def test_redirect_limit_bounds_the_trace(self):
        result = trace_policy("https://docs.example/loop", FIXTURES,
                              AUTHORIZED_HOSTS, max_redirects=2)
        self.assertEqual(result["reason"], "REDIRECT_LIMIT")
        self.assertEqual(result["redirects_followed"], 2)
        self.assertEqual(len(result["request_ledger"]), 3)

    def test_invalid_url_classes_never_reach_ledger(self):
        for value in ("file:///etc/passwd", "https://user:password@docs.example/a",
                      "https://docs.example:bad/a", "https://docs.example/a\n",
                      "https://docs.example:8443/a"):
            with self.subTest(value=value):
                result = trace_policy(value, FIXTURES, AUTHORIZED_HOSTS)
                self.assertEqual(result["status"], "BLOCKED")
                self.assertEqual(result["request_ledger"], [])

    def test_ipv4_mapped_loopback_is_blocked(self):
        url = "https://docs.example/mapped"
        result = trace_policy(url, {url: Fixture(("::ffff:127.0.0.1",))}, AUTHORIZED_HOSTS)
        self.assertEqual(result["request_ledger"], [])
        self.assertEqual(result["status"], "BLOCKED")


if __name__ == "__main__":
    unittest.main()

# MCP Fetch redirect policy trace — MOCK

Testing status: **NOT RUN**. Hybrid Mind has authored and read the code as text; no Python process, imports, compile step, tests, DNS, HTTP, MCP tool call, or upstream package installation was executed.

## Runtime / dependencies / configuration

Python **3.12+**. Standard library only: `ipaddress`, `json`, `dataclasses`, `urllib.parse`, `unittest`. No requirements installation, external SDK, environment variables, credentials, model provider, GPU, paid API, or network access is required. The companion `requirements.txt` is intentionally comment-only.

## Installation and running (instructions for the reader)

Save the two Python files in the same directory. No `pip install` step is needed. With a separately installed Python 3.12+ runtime, the reader can choose to run:

```text
python fetch_policy_trace.py
python -m unittest -v test_fetch_policy_trace.py
```

Those commands are supplied as instructions, not a claim that the publisher ran them.

## Input / expected output

Input is `initial_url`, an in-memory dictionary of `Fixture(addresses, location)` values, an explicit set of authorized hosts, and `max_redirects` (default 3; bounded 0–10). `.example` hostnames and the fixture IPs are synthetic associations: they are not resolved and do not describe the current DNS of any real site.

The default demonstration models a public fixture redirecting to a link-local metadata fixture. Expected output is `mode=MOCK`, `status=BLOCKED`, `reason=NONPUBLIC_OR_INVALID_ADDRESS`, `redirects_followed=1`; `request_ledger` contains only `https://docs.example/to-internal` with `action=WOULD_REQUEST`. The metadata URL appears only in the blocked decision. No actual request occurred, including to the first URL.

The eight authored synthetic test methods cover public-to-public, private redirect absent from ledger, mixed public/private answers, unauthorized public host, unresolved fixture, redirect budget, malformed URL classes, and mapped IPv6 loopback. They are **NOT RUN**, so there is no actual pass count or coverage measurement.

## Permission / cost

This example has no tools that write to external systems, no network capability, no secrets, and no model selection. `request_ledger` is a returned list in memory; it is not a network trace. A reader running it uses only local CPU/RAM; there is no provider charge in the authored design. Runtime latency and memory usage were not measured.

## Intentional differences from mcp-server-fetch 2026.10.10

- This is a teaching policy, not the upstream implementation or a security library.
- It fails closed when a DNS fixture is empty or missing. Upstream leaves locally unresolvable names to its transport/proxy.
- It uses a reviewed host set and restricts ports, userinfo, fragments, and ASCII hosts for simplicity. These are application choices, not upstream defaults.
- Its conservative range list blocks special-purpose exceptions that upstream permits. It is not an authoritative replacement for current IANA classification.
- It never resolves or pins an address, so it cannot exercise DNS rebinding, proxy routing, HTTP redirects, robots.txt, parser compatibility, TLS, network egress, prompt injection, or real SSRF prevention.

For real integration, use the pinned current server, inspect `tools/list`, review network egress and transport/proxy controls, and test permitted sandbox scenarios separately. Do not infer production readiness from the authored fixture tests.

# Python 3.12+ standard library only. No third-party dependencies to install.