diff --git a/docs/integrations/ego-source-reader.md b/docs/integrations/ego-source-reader.md
index 2fe20d64e0..4e297048ff 100644
--- a/docs/integrations/ego-source-reader.md
+++ b/docs/integrations/ego-source-reader.md
@@ -20,6 +20,12 @@ may use the existing browser's session; origin configuration does not prove that
every page on that origin is public. Follow the browser's installed skill for
TaskSpace/Page ownership and user consent.
+An origin list is insufficient for a public community host that must not reuse
+personal cookies. Its isolated native Chat can explicitly enable the separate
+[anonymous static public reader](public-source-reader.md), retaining its shell,
+history and private-MCP boundary. Dynamic signed-in pages remain outside that
+provider's coverage; do not silently substitute the personal Ego reader.
+
For example, append a uniquely named server to the **actual execution host's**
MCP configuration, preserving its existing entries and authentication:
diff --git a/docs/integrations/public-source-reader.md b/docs/integrations/public-source-reader.md
new file mode 100644
index 0000000000..a451d8c48c
--- /dev/null
+++ b/docs/integrations/public-source-reader.md
@@ -0,0 +1,66 @@
+# Anonymous public evidence for isolated Chat
+
+An isolated project Chat can read static public articles and actual image pixels
+without inheriting personal MCP servers, browser cookies or account history.
+The optional bundled `loopx.extensions.public_source_reader` provider performs
+anonymous HTTPS GETs only. It owns external IO, not Session, grant, material,
+delegation or publication authority. It is a fallback for static sources;
+[Ego's rendered reader](ego-source-reader.md) remains useful on an explicitly
+authorized personal host when a page needs browser rendering.
+
+Install the optional dependency in the **existing Chat host's** environment:
+
+```sh
+python -m pip install 'loopx[public-source-reader]'
+```
+
+On that trusted host, set `LOOPX_CHAT_PUBLIC_SOURCE_READ=on` and restart the idle
+service through its existing lifecycle. `off` is the default and rollback.
+Only native `workspace_only` Chat processes consume this setting; ordinary
+personal Chat keeps its existing MCP configuration. Existing Sessions resume
+in their original native store. Workspace grants, shell network restrictions,
+model authentication, audience and history do not change.
+
+The host first disables all effective personal/project MCP entries, then admits
+one provider with a fixed interpreter and current-release file, minimal environment and no caller
+credentials. Isolated Python excludes the writable workspace and `PYTHONPATH`
+from provider module lookup. Source-only and bundled releases use their own
+provider file for both MCP startup and observation workers; they do not need a
+second site-installed LoopX package, and cannot silently use an older one. The
+optional dependencies must still be installed in that host interpreter. The
+reserved name `loopx_public_source_read` must not appear in a
+lower native configuration layer: a collision fails before model dispatch rather
+than merging its command, environment or headers. Project files and model input
+cannot enable this provider. Missing dependencies are an installation failure,
+not authority to enable a personal reader or widen the shell sandbox.
+
+`read_public_url(url)` returns static text, image indices, observed URL, source
+digest and explicit truncation/coverage. `read_public_image(url, index)` reads
+that article again and returns one PNG image-content block with its source
+digest. A changed article may have changed image indices; verify the returned
+digest and URL. Text/alt labels do not prove pixels were read, and static HTML
+does not prove a dynamic article or a verification wall was fully read. The
+Agent must report missing coverage and inspect rendered labels and arrows.
+
+The provider sends no cookies, account headers, proxy settings or caller-defined
+headers. Every destination and redirect must resolve entirely to globally
+routable addresses; the checked address is pinned to the TLS connection while
+the original hostname is used for certificate validation. File URLs, credentials
+in URLs, nonstandard ports, local/private/link-local addresses and mixed DNS
+answers are rejected. The whole observation, including DNS and rendering, runs
+in a minimal-environment worker with a 30-second deadline. Responses are bounded
+to 4 MB, text to 100,000 characters, image inventories to 128 and image edges to
+4,096 pixels. Errors return no private interpreter paths or resolver output.
+
+PNG/JPEG/WebP and self-contained SVG images are supported. SVG geometry and
+aspect ratio are preserved; active elements, external references and stylesheet
+blocks are rejected. The bundled [resvg renderer](https://resvg-py.readthedocs.io/en/latest/api.html)
+uses only standard OS font directories, skipping the account's font database;
+no additional native Cairo installation is required. This provider never reads
+a local source file supplied by the caller or navigates an authenticated browser.
+
+Qualification must separately demonstrate native Session tool/image consumption,
+an original-topic answer with actual provider readback, outside/private-resource
+rejection, unchanged grants/history and source completeness. A successful host
+MCP probe or unit suite alone does not qualify the community Bot's golden pack.
+Keep admission on hold while mandatory cases remain failed, blocked or untested.
diff --git a/loopx/capabilities/native_chat/codex_context.py b/loopx/capabilities/native_chat/codex_context.py
index e10228a5d1..e50bd47050 100644
--- a/loopx/capabilities/native_chat/codex_context.py
+++ b/loopx/capabilities/native_chat/codex_context.py
@@ -6,7 +6,9 @@
"""
from __future__ import annotations
+import importlib.util
import os
+import sys
from pathlib import Path
from typing import Any, Mapping
@@ -97,3 +99,31 @@ def disable_mcp_servers(config: Mapping[str, Any], host_config: Mapping[str, Any
return {**host_config, "mcp_servers": {
name: {"enabled": False} for name in effective.keys() | requested.keys()
}}
+
+
+def public_source_reader(config: Mapping[str, Any], host_config: Mapping[str, Any]) -> dict[str, Any]:
+ """Admit one operator-selected anonymous provider after private MCP removal.
+
+ This is a host IO adapter, not a project-controlled grant. No inherited
+ server fields, browser state, credentials or private execution permissions.
+ """
+ setting = os.environ.get("LOOPX_CHAT_PUBLIC_SOURCE_READ", "off")
+ if setting == "off":
+ return dict(host_config)
+ if setting != "on":
+ raise ValueError("LOOPX_CHAT_PUBLIC_SOURCE_READ must be on or off")
+ if any(importlib.util.find_spec(module) is None for module in ("mcp", "resvg_py", "PIL", "defusedxml")):
+ raise ValueError("Install loopx[public-source-reader] in the existing Chat host environment")
+ name = "loopx_public_source_read"
+ if name in config.get("mcp_servers", {}) or name in host_config.get("mcp_servers", {}):
+ raise ValueError("Public source reader conflicts with native MCP configuration")
+ from ...extensions import public_source_reader as reader
+ # Managed releases put their source on the host's sys.path, not necessarily
+ # in site-packages. Bind this release's file; -m could fail or select an
+ # older site-installed LoopX once the child discards the host's sys.path.
+ return {**host_config, "mcp_servers": {**host_config.get("mcp_servers", {}), name: {
+ "enabled": True, "command": str(Path(sys.executable).absolute()),
+ "args": ["-I", str(Path(reader.__file__).resolve())],
+ "env": {"PATH": os.defpath}, "env_vars": [],
+ "startup_timeout_sec": 30, "tool_timeout_sec": 40,
+ }}}
diff --git a/loopx/chat_agent.py b/loopx/chat_agent.py
index b897d0cb4f..3cede76d30 100644
--- a/loopx/chat_agent.py
+++ b/loopx/chat_agent.py
@@ -844,6 +844,10 @@ def start(
raise session._runtime_error("Codex returned an invalid project configuration.")
if permissions_profile:
host_config = codex_context.disable_mcp_servers(effective, host_config or {})
+ try:
+ host_config = codex_context.public_source_reader(effective, host_config)
+ except ValueError as exc:
+ raise session._runtime_error(str(exc)) from None
if session._host_model_auth is not None:
# Use public native transport defaults, not the
# account's configuration or environment. Reject a
diff --git a/loopx/extensions/public_source_reader.py b/loopx/extensions/public_source_reader.py
new file mode 100644
index 0000000000..9f9aa3b9cf
--- /dev/null
+++ b/loopx/extensions/public_source_reader.py
@@ -0,0 +1,314 @@
+"""Anonymous static public-source IO for an explicitly enabled native host.
+
+No browser, account, filesystem tool, model runner or Session authority. Every
+HTTP hop uses a checked, pinned public address and carries no account state.
+"""
+from __future__ import annotations
+
+import base64
+import hashlib
+import http.client
+import ipaddress
+import io
+import json
+import os
+from pathlib import Path
+import re
+import socket
+import ssl
+import subprocess
+import sys
+import threading
+import time
+from html.parser import HTMLParser
+from urllib.parse import urljoin, urlsplit, urlunsplit
+
+MAX_BYTES = 4_000_000
+MAX_TEXT = 100_000
+MAX_IMAGES = 128
+MAX_EDGE = 4096
+TIMEOUT = 30
+_LOCK = threading.Lock()
+
+
+def _global_address(value: str) -> bool:
+ address = ipaddress.ip_address(value)
+ if not address.is_global or address.is_reserved or address.is_multicast:
+ return False
+ return not (isinstance(address, ipaddress.IPv6Address) and (address.sixtofour or address.teredo))
+
+
+def public_url(value: str) -> str:
+ if (not isinstance(value, str) or not value or len(value) > 8192
+ or not value.isascii() or "\\" in value
+ or any(ord(c) <= 32 or ord(c) == 127 for c in value)):
+ raise ValueError("invalid public URL")
+ u = urlsplit(value)
+ if (u.scheme != "https" or not u.hostname or u.username is not None
+ or u.password is not None or u.port not in (None, 443)):
+ raise ValueError("anonymous HTTPS source required")
+ # Reject private literals before even resolving DNS. Resolved names receive
+ # the same check below, including mixed public/private DNS answers.
+ try:
+ address = ipaddress.ip_address(u.hostname)
+ except ValueError:
+ address = None
+ if address is not None and not _global_address(str(address)):
+ raise ValueError("non-public source address")
+ return urlunsplit(("https", u.netloc, u.path or "/", u.query, ""))
+
+
+def public_addresses(host: str) -> list[str]:
+ values = list(dict.fromkeys(row[4][0] for row in socket.getaddrinfo(
+ host, 443, type=socket.SOCK_STREAM)))
+ if not values or any(not _global_address(v) for v in values):
+ raise ValueError("non-public source address")
+ return values
+
+
+class _PinnedHTTPS(http.client.HTTPSConnection):
+ def __init__(self, host: str, addresses: list[str], timeout: float):
+ super().__init__(host, timeout=timeout, context=ssl.create_default_context())
+ self.addresses = addresses
+
+ def connect(self) -> None:
+ # Do not resolve the hostname again after preflight (DNS rebinding).
+ for address in self.addresses:
+ raw = None
+ try:
+ raw = socket.create_connection((address, 443), min(self.timeout, 5))
+ self.sock = self._context.wrap_socket(raw, server_hostname=self.host)
+ return
+ except OSError:
+ if raw is not None:
+ raw.close()
+ raise OSError("public source connection failed")
+
+
+def fetch(value: str, *, deadline: float) -> tuple[str, str, bytes]:
+ url = public_url(value)
+ for _ in range(6):
+ u = urlsplit(url)
+ addresses = public_addresses(u.hostname or "")
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ raise TimeoutError("source timeout")
+ connection = _PinnedHTTPS(u.hostname or "", addresses, min(remaining, 10))
+ try:
+ path = urlunsplit(("", "", u.path, u.query, ""))
+ # Fixed headers only. http.client has no cookie jar, credential
+ # lookup, ambient proxy or redirect handler.
+ connection.request("GET", path, headers={"Accept-Encoding": "identity",
+ "User-Agent": "LoopX-Public-Source/1.0"})
+ response = connection.getresponse()
+ if response.status in (301, 302, 303, 307, 308):
+ location = response.getheader("Location")
+ if not location:
+ raise ValueError("missing redirect destination")
+ url = public_url(urljoin(url, location))
+ continue
+ if response.status != 200 or response.getheader("Content-Encoding", "identity") != "identity":
+ raise ValueError("public source unavailable")
+ body = bytearray()
+ while len(body) <= MAX_BYTES:
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ raise TimeoutError("source timeout")
+ if connection.sock is not None:
+ connection.sock.settimeout(min(remaining, 10))
+ chunk = response.read1(min(65536, MAX_BYTES + 1 - len(body)))
+ if not chunk:
+ break
+ body.extend(chunk)
+ if len(body) > MAX_BYTES:
+ raise ValueError("public source too large")
+ return url, response.getheader("Content-Type", "").split(";")[0].lower(), bytes(body)
+ finally:
+ connection.close()
+ raise ValueError("too many source redirects")
+
+
+class _Article(HTMLParser):
+ def __init__(self, url: str):
+ super().__init__(convert_charrefs=True)
+ self.url = url
+ self.text: list[str] = []
+ self.images: list[dict[str, object]] = []
+ self.hidden = 0
+
+ def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+ if tag in ("script", "style", "noscript", "template"):
+ self.hidden += 1
+ if self.hidden:
+ return
+ if tag in ("p", "div", "br", "li", "h1", "h2", "h3", "tr"):
+ self.text.append("\n")
+ if tag == "img":
+ a = dict(attrs)
+ src = a.get("src") or a.get("data-src")
+ if src:
+ self.images.append({"index": len(self.images), "url": urljoin(self.url, src),
+ "alt": a.get("alt") or ""})
+
+ def handle_endtag(self, tag: str) -> None:
+ if tag in ("script", "style", "noscript", "template") and self.hidden:
+ self.hidden -= 1
+
+ def handle_data(self, data: str) -> None:
+ if not self.hidden:
+ self.text.append(data)
+
+
+def article(url: str, *, deadline: float) -> dict[str, object]:
+ observed, mime, body = fetch(url, deadline=deadline)
+ text = body.decode("utf-8", errors="replace")
+ images: list[dict[str, object]] = []
+ if mime in ("text/html", "application/xhtml+xml"):
+ parser = _Article(observed)
+ parser.feed(text)
+ text = "\n".join(line.strip() for line in "".join(parser.text).splitlines() if line.strip())
+ images = parser.images
+ elif mime not in ("text/plain", "text/markdown"):
+ raise ValueError("unsupported public text format")
+ return {"ok": True, "source": "anonymous_https_static", "requested_url": url,
+ "observed_url": observed, "text": text[:MAX_TEXT], "characters": len(text),
+ "truncated": len(text) > MAX_TEXT, "sha256": hashlib.sha256(body).hexdigest(),
+ "images": images[:MAX_IMAGES], "image_count": len(images),
+ "image_inventory_truncated": len(images) > MAX_IMAGES, "images_read": False,
+ "coverage": "static response only; scripts, account state and verification walls are not read"}
+
+
+def image_png(body: bytes, mime: str) -> bytes:
+ from PIL import Image
+ if mime == "image/svg+xml":
+ from defusedxml import ElementTree
+ root = ElementTree.fromstring(body)
+ # The provider only renders self-contained SVG, with no images or
+ # external styles; reject references except same-document fragments.
+ for node in root.iter():
+ if node.tag.rsplit("}", 1)[-1] in ("script", "foreignObject", "image", "style"):
+ raise ValueError("SVG external or active content")
+ for key, value in node.attrib.items():
+ if key.rsplit("}", 1)[-1] == "base":
+ raise ValueError("SVG external base")
+ if key.rsplit("}", 1)[-1] == "href" and not value.startswith("#"):
+ raise ValueError("SVG external reference")
+ for value in [*node.attrib.values(), node.text or ""]:
+ if "\\" in value or re.search(r"@import|@font-face", value, re.I):
+ raise ValueError("SVG external style")
+ for ref in re.findall(r"url\s*\((.*?)\)", value, re.I | re.S):
+ if not ref.strip(" \t\r\n\"'").startswith("#"):
+ raise ValueError("SVG external reference")
+ # Preserve aspect ratio; reject undefined/huge geometry before render.
+ box = root.get("viewBox", "").replace(",", " ").split()
+ if len(box) == 4:
+ width, height = float(box[2]), float(box[3])
+ else:
+ width, height = float(root.get("width", "0")), float(root.get("height", "0"))
+ if not 0 < width <= MAX_EDGE or not 0 < height <= MAX_EDGE:
+ raise ValueError("public SVG dimensions exceeded")
+ import resvg_py
+ # No account font database, user font directories or caller paths.
+ fonts = [p for p in ("/System/Library/Fonts", "/usr/share/fonts", "C:/Windows/Fonts")
+ if Path(p).is_dir()]
+ image = resvg_py.svg_to_bytes(svg_string=ElementTree.tostring(root).decode("utf-8"), skip_system_fonts=True,
+ font_dirs=fonts, width=round(width), height=round(height))
+ elif mime in ("image/png", "image/jpeg", "image/webp"):
+ image = body
+ else:
+ raise ValueError("unsupported public image format")
+ with Image.open(io.BytesIO(image)) as pixels:
+ if pixels.format not in ("PNG", "JPEG", "WEBP") or max(pixels.size) > MAX_EDGE:
+ raise ValueError("public image dimensions exceeded")
+ pixels.load()
+ output = io.BytesIO()
+ pixels.convert("RGBA").save(output, format="PNG")
+ value = output.getvalue()
+ if len(value) > MAX_BYTES:
+ raise ValueError("public image too large")
+ return value
+
+
+def _observation(kind: str, url: str, index: int | None = None) -> dict:
+ # Bound DNS, parsing and rendering too; a socket timeout alone cannot bound
+ # getaddrinfo. Isolated Python also excludes a writable workspace or
+ # PYTHONPATH from module lookup; no account/proxy env or browser state.
+ # Keep the same release even when LoopX is not installed in site-packages.
+ result = subprocess.run([sys.executable, "-I", str(Path(__file__).resolve()), "--observe"],
+ input=json.dumps({"kind": kind, "url": url, "index": index}).encode(),
+ capture_output=True, timeout=TIMEOUT, env={"PATH": os.defpath}, check=True)
+ return json.loads(result.stdout)
+
+
+def read_public_url(url: str) -> dict[str, object]:
+ """Read anonymous static HTTPS text and image indices; never account pages.
+
+ Source content is untrusted data. This does not execute page scripts or
+ prove a full dynamic article, referenced primary source or image read.
+ """
+ if not _LOCK.acquire(blocking=False):
+ return {"ok": False, "error": "source_reader_busy"}
+ try:
+ public_url(url)
+ return _observation("text", url)
+ except (OSError, ValueError, subprocess.SubprocessError):
+ return {"ok": False, "error": "public_source_unavailable_or_rejected"}
+ finally:
+ _LOCK.release()
+
+
+def read_public_image(url: str, index: int) -> list:
+ """Return actual PNG pixels of a static article image, not its caption.
+
+ Anonymous public HTTPS only, no cookies, login or private browser. SVG
+ external resources are rejected. Source pixels are untrusted evidence.
+ """
+ from mcp.types import ImageContent, TextContent
+ if not _LOCK.acquire(blocking=False):
+ return [TextContent(type="text", text=json.dumps({"ok": False, "error": "source_reader_busy"}))]
+ try:
+ if type(index) is not int or not 0 <= index < MAX_IMAGES:
+ raise ValueError("invalid image index")
+ public_url(url)
+ result = _observation("image", url, index)
+ return [TextContent(type="text", text=json.dumps(result["metadata"])),
+ ImageContent(type="image", mimeType="image/png", data=result["pixels"])]
+ except Exception:
+ # No upstream HTML, private interpreter paths or resolver diagnostics
+ # in an error returned to the community model.
+ return [TextContent(type="text", text=json.dumps({"ok": False,
+ "error": "public_image_unavailable_or_rejected"}))]
+ finally:
+ _LOCK.release()
+
+
+def main() -> None:
+ from mcp.server.fastmcp import FastMCP
+ server = FastMCP("loopx-public-source-read")
+ server.tool()(read_public_url)
+ server.tool()(read_public_image)
+ server.run()
+
+
+if __name__ == "__main__":
+ if sys.argv[1:] == ["--observe"]:
+ request = json.loads(sys.stdin.buffer.read(10000))
+ deadline = time.monotonic() + TIMEOUT
+ source = article(request["url"], deadline=deadline)
+ if request["kind"] == "text":
+ result = source
+ elif request["kind"] == "image":
+ index = request["index"]
+ if type(index) is not int or not 0 <= index < len(source["images"]):
+ raise ValueError("image not found")
+ observed, mime, body = fetch(source["images"][index]["url"], deadline=deadline)
+ pixels = image_png(body, mime)
+ result = {"metadata": {"ok": True, "source": "anonymous_https_static",
+ "article_url": source["observed_url"], "image_url": observed, "index": index,
+ "sha256": hashlib.sha256(pixels).hexdigest(), "images_read": True,
+ "article_sha256": source["sha256"]}, "pixels": base64.b64encode(pixels).decode()}
+ else:
+ raise ValueError("invalid observation")
+ print(json.dumps(result))
+ else:
+ main()
diff --git a/pyproject.toml b/pyproject.toml
index 97aed4f6a1..1c1999a476 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -26,10 +26,19 @@ Changelog = "https://github.com/loopx-project/loopx/releases"
ego-source-reader = [
"mcp==1.28.1",
]
+public-source-reader = [
+ "mcp==1.28.1",
+ "resvg-py==0.5.0",
+ "defusedxml==0.7.1",
+ "pillow>=11,<13",
+]
deepseek-harness = [
"deepseek-harness-sdk==0.1.5rc1",
]
test = [
+ "resvg-py==0.5.0",
+ "defusedxml==0.7.1",
+ "pillow>=11,<13",
"jsonschema>=4.23,<5",
"mcp==1.28.1",
"pyyaml>=6,<7",
diff --git a/tests/test_chat_agent.py b/tests/test_chat_agent.py
index 552562f156..0e7e0037b3 100644
--- a/tests/test_chat_agent.py
+++ b/tests/test_chat_agent.py
@@ -177,6 +177,7 @@ def test_non_macos_workspace_does_not_discover_macos_tools(monkeypatch):
@pytest.mark.parametrize("grant", ["workspace_read", "workspace_write"])
@pytest.mark.parametrize("resume", [False, True])
+@pytest.mark.parametrize("public_reader", [False, True])
@pytest.mark.parametrize("tool_options", [
{},
{"unsafe_path": "/Library/Developer", "unsafe_kind": "owner"},
@@ -192,9 +193,10 @@ def test_non_macos_workspace_does_not_discover_macos_tools(monkeypatch):
{"selected": "/Applications/Xcode.app/Contents/Developer", "unsafe_path": "/"},
])
def test_project_filesystem_scope_is_verified_on_start_resume_and_pinned_per_turn(
- monkeypatch, tmp_path, grant, resume, tool_options):
+ monkeypatch, tmp_path, grant, resume, tool_options, public_reader):
from loopx.capabilities.native_chat.project_context import ChatProjectContexts
context = ChatProjectContexts([tmp_path], workspace_grant=grant, filesystem_scope="workspace_only").available()[0]
+ monkeypatch.setenv("LOOPX_CHAT_PUBLIC_SOURCE_READ", "on" if public_reader else "off")
toolchain, _ = _system_toolchain_fixture(monkeypatch, **tool_options)
safe_toolchain = not tool_options
profile = "loopx_workspace_only_" + ("write" if grant == "workspace_write" else "read")
@@ -235,7 +237,13 @@ def popen(command, *args, **kwargs):
assert params["config"]["default_permissions"] == profile
assert params["config"]["skills"]["include_instructions"] is False
assert params["config"]["project_doc_max_bytes"] == 0
- assert params["config"]["mcp_servers"] == {
+ servers = dict(params["config"]["mcp_servers"])
+ if public_reader:
+ from loopx.extensions import public_source_reader as source_reader
+ public = servers.pop("loopx_public_source_read")
+ assert public["enabled"] is True and public["env_vars"] == []
+ assert public["args"] == ["-I", str(Path(source_reader.__file__).resolve())]
+ assert servers == {
"managed_fixture": {"enabled": False}, "caller_fixture": {"enabled": False}}
env = launched[0]
assert "PRIVATE_FIXTURE_TOKEN" not in env
diff --git a/tests/test_public_source_reader.py b/tests/test_public_source_reader.py
new file mode 100644
index 0000000000..d630cc2289
--- /dev/null
+++ b/tests/test_public_source_reader.py
@@ -0,0 +1,186 @@
+"""Public evidence must not acquire account, local-network or file authority."""
+import io
+import ipaddress
+import socket
+import time
+from pathlib import Path
+
+import pytest
+from PIL import Image
+
+from loopx.extensions import public_source_reader as reader
+from loopx.capabilities.native_chat import codex_context
+
+
+# Synthetic RFC1918 range fixtures are expressed as integers; these are not operator addresses.
+@pytest.mark.parametrize("url", ["file:///etc/passwd", "http://example.org/", "https://user:pass@example.org/",
+ "https://127.0.0.1/", "https://[::1]/", "https://169.254.169.254/",
+ f"https://{ipaddress.IPv4Address(0x0A000001)}/",
+ "https://example.org:8443/", "https://example.org/\nheader", "https://example.org\\@127.0.0.1/",
+ "https://224.0.0.1/", "https://[2002:7f00:1::]/"])
+def test_nonpublic_or_authenticated_urls_rejected_before_network(monkeypatch, url):
+ monkeypatch.setattr(socket, "getaddrinfo", lambda *a, **k: pytest.fail("DNS must not be called"))
+ with pytest.raises(ValueError):
+ reader.public_url(url)
+
+
+@pytest.mark.parametrize("values", [["127.0.0.1"], ["8.8.8.8", str(ipaddress.IPv4Address(0xC0A80102))], ["::ffff:127.0.0.1"], ["100.64.0.1"]])
+def test_dns_private_and_mixed_results_rejected(monkeypatch, values):
+ monkeypatch.setattr(socket, "getaddrinfo", lambda *a, **k: [(0, 0, 0, "", (v, 443)) for v in values])
+ with pytest.raises(ValueError):
+ reader.public_addresses("public-name.example")
+
+
+def test_checked_address_is_the_connected_address_and_tls_hostname_stays_original(monkeypatch):
+ raw = object()
+ observed = []
+ monkeypatch.setattr(socket, "create_connection", lambda target, timeout: observed.append(target) or raw)
+ connection = reader._PinnedHTTPS("example.org", ["8.8.8.8"], 2)
+ class TLS:
+ def wrap_socket(self, sock, *, server_hostname):
+ assert sock is raw and server_hostname == "example.org"
+ return "tls-socket"
+ connection._context = TLS()
+ connection.connect()
+ assert observed == [("8.8.8.8", 443)] and connection.sock == "tls-socket"
+
+
+def test_redirect_cannot_reach_private_network_and_cookies_are_not_reused(monkeypatch):
+ calls = []
+ monkeypatch.setattr(reader, "public_addresses", lambda host: ["8.8.8.8"])
+ class Response:
+ status = 302
+ def getheader(self, name, default=None):
+ return {"Location": "https://127.0.0.1/private", "Set-Cookie": "private=fixture"}.get(name, default)
+ class Connection:
+ def __init__(self, *a): pass
+ def request(self, method, path, *, headers): calls.append((method, path, headers))
+ def getresponse(self): return Response()
+ def close(self): pass
+ monkeypatch.setattr(reader, "_PinnedHTTPS", Connection)
+ with pytest.raises(ValueError):
+ reader.fetch("https://example.org/start", deadline=time.monotonic() + 5)
+ assert len(calls) == 1 and calls[0][:2] == ("GET", "/start")
+ assert set(calls[0][2]) == {"Accept-Encoding", "User-Agent"}
+
+
+def test_static_html_coverage_and_images_are_explicit(monkeypatch):
+ body = b'
'
+ monkeypatch.setattr(reader, "fetch", lambda *a, **k: ("https://example.org/article", "text/html", body))
+ result = reader.article("https://example.org/article", deadline=time.monotonic() + 5)
+ assert result["text"] == "Public title\nFinal paragraph."
+ assert result["images"] == [{"index": 0, "url": "https://example.org/figure.svg", "alt": "Chart"}]
+ assert result["images_read"] is False and "static response" in result["coverage"]
+
+
+@pytest.mark.parametrize("element", [
+ '