004 — CPython urllib.request 3.14.5

{
  "name": "urllib-opener-recut",
  "what_it_is": "A handler registry and call chain, keeping upstream's shape and making registration report what it registered.",
  "examined": {
    "project": "CPython urllib.request", "version": "3.14.5", "license": "PSF-2.0",
    "measured": {
      "module_lines": 2163,
      "OpenerDirector_methods": 6,
      "BaseHandler_methods": 3,
      "handler_classes_shipped": 18,
      "add_handler_lines": 46,
      "_call_chain_lines": 10,
      "binding_mechanism": "scan the handler's attribute names for *_open / *_error / *_request / *_response",
      "add_handler_return_value": "None, in every case"
    },
    "reproduced": {
      "data_open": "registers under 'data', 1 handler",
      "data_opne": "registers nothing, 0 handlers, returns None, no error, no warning",
      "htp_open": "registers under 'htp' - a scheme that does not exist",
      "no_recognised_method": "0 handlers, silent"
    }
  },
  "capabilities": {
    "SMS": { "chain": "The registry and the call chain. bind() returns a Binding saying what it bound, ignored and refused." },
    "SCL": { "policy": "Which handler may serve which scheme, and which schemes exist at all." },
    "TMS": { "handlers/data": "Serves data: URLs.", "handlers/file": "Serves file: URLs from a root it is handed." },
    "DMS": { "report": "Bindings, near-miss method names, refusals, and any handler that bound nothing." }
  },
  "the_finding": "The registry is right and worth keeping - handlers are passed in, the core has six methods, BaseHandler has three. The binding is by naming convention with nothing checking it, so a transposed letter produces a handler that is registered nowhere and reported as fine.",
  "non_goals": [
    "Being urllib. No redirects, no auth, no proxies, no network, two schemes.",
    "Replacing naming-based declaration. The re-cut keeps it and adds a check, because the convention is what makes a handler readable."
  ]
}

{
  "policy_version": "1.0",
  "_comment": [
    "Upstream, a handler claims a scheme by being named after it, and any handler",
    "may claim any scheme. `htp_open` registers a handler for a protocol that does",
    "not exist; nothing objects, because there is no list of schemes to object",
    "against. The name is both the declaration and the authorisation.",
    "",
    "Here the two are separated: a handler declares by naming, and SCL decides",
    "whether the declaration is honoured."
  ],
  "schemes": ["data", "file"],
  "handlers": {
    "handlers/data": { "may_serve": ["data"] },
    "handlers/file": { "may_serve": ["file"] }
  },
  "rules": [
    {
      "id": "scheme-must-exist",
      "statement": "A handler cannot bind a scheme this system does not recognise.",
      "enforced_by": "SCL/policy.py::may_serve, called by SMS/chain.py::bind"
    }
  ]
}

"""Which handler may serve which scheme, and which schemes exist at all.

The second half is what upstream has no place for. `htp_open` binds `htp`
because binding is name-driven and there is no set of real schemes to check the
name against — so a typo in the protocol produces a handler that is registered,
reachable by nothing, and reported as fine.
"""
from __future__ import annotations

import json
from pathlib import Path

_POLICY = json.loads(Path(__file__).with_name("policy.json").read_text(encoding="utf-8"))
SCHEMES: set[str] = set(_POLICY["schemes"])
HANDLERS: dict[str, dict] = _POLICY["handlers"]


def may_serve(handler: str, scheme: str) -> tuple[bool, str]:
    if scheme not in SCHEMES:
        return False, f"unknown scheme; this system serves {sorted(SCHEMES)}"
    entry = HANDLERS.get(handler)
    if entry is None:
        return False, f"unknown handler {handler!r}"
    if scheme not in entry["may_serve"]:
        return False, f"{handler} is not permitted {scheme}"
    return True, ""

"""An opener: a registry of handlers, and a call chain over it.

Upstream's shape, kept — because it is right. `OpenerDirector` has six methods,
`BaseHandler` has three, and handlers arrive through `build_opener(*handlers)`.
The core does not know what can handle anything until it is told, which is
exactly what `http.server` cannot do (see archaeology 002).

The one change is that `bind` returns what it bound. Upstream `add_handler`
returns `None` whether it registered five methods or zero, so a handler that
registered nothing is indistinguishable, from the caller, from one that worked.
"""
from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from SCL import policy  # noqa: E402

SUFFIX = "_open"


class Binding:
    """What one handler contributed, as a value."""

    def __init__(self, handler_name: str, schemes: list[str], ignored: list[str], refused: list[str]) -> None:
        self.handler = handler_name
        self.schemes = schemes
        self.ignored = ignored      # methods that look like bindings and are not
        self.refused = refused      # schemes SCL did not permit this handler

    @property
    def bound_anything(self) -> bool:
        return bool(self.schemes)


class Opener:
    def __init__(self) -> None:
        self.by_scheme: dict[str, object] = {}
        self.bindings: list[Binding] = []

    def bind(self, handler) -> Binding:
        """Register a handler, and say what happened.

        Refusing to bind nothing is the whole repair. A handler that contributes
        no scheme is a mistake every time — there is no reason to construct one —
        so it is reported rather than accepted in silence.
        """
        schemes, ignored, refused = [], [], []
        for attr in dir(handler):
            if attr.startswith("_"):
                continue
            if not attr.endswith(SUFFIX):
                # A near-miss is worth naming: `data_opne` is not a binding and
                # is also not nothing. Upstream cannot tell these apart from a
                # method that was never meant to bind.
                if _looks_like_a_near_miss(attr):
                    ignored.append(attr)
                continue
            scheme = attr[: -len(SUFFIX)]
            allowed, why = policy.may_serve(getattr(handler, "name", type(handler).__name__), scheme)
            if not allowed:
                refused.append(f"{scheme} ({why})")
                continue
            self.by_scheme[scheme] = handler
            schemes.append(scheme)

        binding = Binding(getattr(handler, "name", type(handler).__name__), schemes, ignored, refused)
        self.bindings.append(binding)
        return binding

    def open(self, url: str) -> tuple[bool, str]:
        scheme = url.split(":", 1)[0]
        handler = self.by_scheme.get(scheme)
        if handler is None:
            return False, f"no handler bound for scheme {scheme!r}"
        return True, getattr(handler, scheme + SUFFIX)(url)


def _looks_like_a_near_miss(attr: str) -> bool:
    """Cheap heuristic: an anagram of the suffix at the end of the name.

    Deliberately not clever. It catches `_opne`, `_oepn`, `_pen` — the
    transpositions that actually happen — and says so rather than guessing what
    was meant. Upstream has nothing here at all, which is the finding.
    """
    tail = attr.rsplit("_", 1)[-1] if "_" in attr else ""
    return tail != SUFFIX.strip("_") and sorted(tail) == sorted(SUFFIX.strip("_"))

"""Serves data: URLs. Declares its scheme by method name, as upstream does."""
from __future__ import annotations

import base64

name = "handlers/data"


class DataHandler:
    name = name

    def data_open(self, url: str) -> str:
        _, _, rest = url.partition(":")
        meta, _, payload = rest.partition(",")
        if "base64" in meta:
            return base64.b64decode(payload).decode("utf-8", "replace")
        return payload

"""Serves file: URLs from a root it is handed."""
from __future__ import annotations

from pathlib import Path

name = "handlers/file"


class FileHandler:
    name = name

    def __init__(self, root: Path) -> None:
        self._root = root

    def file_open(self, url: str) -> str:
        target = (self._root / url.partition(":")[2].lstrip("/")).resolve()
        if self._root.resolve() not in target.parents:
            return "refused: outside the served root"
        return target.read_text(encoding="utf-8") if target.is_file() else "no such file"

"""What binding actually did.

Upstream `add_handler` returns None in every case: five methods bound, one
bound, none bound. This prints the same three cases differently, which is the
entire difference between a registry you can trust and one you hope about.
"""
from __future__ import annotations


def render(bindings) -> str:
    lines = ["  bindings"]
    for b in bindings:
        if b.bound_anything:
            lines.append(f"    {b.handler:<18} bound {', '.join(b.schemes)}")
        else:
            lines.append(f"    {b.handler:<18} BOUND NOTHING")
        for attr in b.ignored:
            lines.append(f"    {'':<18}   near-miss method `{attr}` - did you mean `{attr.rsplit('_', 1)[0]}_open`?")
        for scheme in b.refused:
            lines.append(f"    {'':<18}   refused {scheme}")
    return "\n".join(lines)


def inert(bindings) -> list[str]:
    """Handlers that registered nothing. Constructing one is always a mistake."""
    return [b.handler for b in bindings if not b.bound_anything]

"""The island test, the upstream measurement, and four ways to defeat the repair.

    python src/island-test.py
"""
from __future__ import annotations

import sys
import tempfile
from pathlib import Path

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))

from DMS.report import inert                  # noqa: E402
from SCL.policy import may_serve              # noqa: E402
from SMS.chain import Opener                  # noqa: E402

failures: list[str] = []


def report(label: str, ok: bool, detail: str = "") -> None:
    print(f"  {'PASS' if ok else 'FAIL'}  {label}" + (f" - {detail}" if detail else ""))
    if not ok:
        failures.append(label)


print("\n== 1. each handler alone, no sibling loaded")
from TMS.handlers.data import DataHandler     # noqa: E402

opener = Opener()
b = opener.bind(DataHandler())
ok, out = opener.open("data:text/plain,alone")
report("handlers/data binds and serves with no sibling present", b.schemes == ["data"] and ok and out == "alone",
       f"schemes={b.schemes} out={out!r}")
report("and nothing else is reachable", opener.open("file:/x")[0] is False)

from TMS.handlers.file import FileHandler     # noqa: E402

root = Path(tempfile.mkdtemp())
(root / "a.txt").write_text("only me\n", encoding="utf-8")
solo = Opener()
b2 = solo.bind(FileHandler(root=root))
report("handlers/file binds and serves with no sibling present",
       b2.schemes == ["file"] and solo.open("file:/a.txt")[1].strip() == "only me")

print("\n== 2. the upstream behaviour, measured now")
import urllib.request as ur                   # noqa: E402


class _Good(ur.BaseHandler):
    def data_open(self, req): return None


class _Typo(ur.BaseHandler):
    def data_opne(self, req): return None


class _Wrong(ur.BaseHandler):
    def htp_open(self, req): return None


measured = {}
for label, cls in (("good", _Good), ("typo", _Typo), ("wrong-scheme", _Wrong)):
    d = ur.OpenerDirector()
    returned = d.add_handler(cls())
    measured[label] = {"schemes": sorted(d.handle_open), "handlers": len(d.handlers), "returned": returned}
    print(f"     {label:<13} schemes={measured[label]['schemes']}  handlers={measured[label]['handlers']}  add_handler returned {returned!r}")

report("a correct method binds", measured["good"]["schemes"] == ["data"])
report("a transposed method binds nothing", measured["typo"]["schemes"] == [] and measured["typo"]["handlers"] == 0)
report("and add_handler reports nothing about it", measured["typo"]["returned"] is None,
       "same return value as the successful case")
report("a nonexistent scheme is accepted", measured["wrong-scheme"]["schemes"] == ["htp"],
       "there is no set of real schemes for the name to be checked against")
report("upstream's core is small, which is the part worth keeping",
       len([m for m in ur.OpenerDirector.__dict__ if not m.startswith("__")]) <= 8
       and len([m for m in ur.BaseHandler.__dict__ if not m.startswith("__")]) <= 4,
       f"OpenerDirector={len([m for m in ur.OpenerDirector.__dict__ if not m.startswith('__')])} "
       f"BaseHandler={len([m for m in ur.BaseHandler.__dict__ if not m.startswith('__')])}")

print("\n== 3. the repair catches what upstream cannot")
class Typo:
    name = "handlers/data"
    def data_opne(self, url): return "x"      # noqa: N802


class Wrong:
    name = "handlers/data"
    def htp_open(self, url): return "x"


o = Opener()
bt = o.bind(Typo())
bw = o.bind(Wrong())
report("the transposed handler is reported as binding nothing", not bt.bound_anything)
report("and the near-miss method is named", bt.ignored == ["data_opne"], str(bt.ignored))
report("the nonexistent scheme is refused, not bound", not bw.bound_anything and "htp" not in o.by_scheme,
       f"refused={bw.refused}")
report("inert handlers are listed", sorted(set(inert(o.bindings))) == ["handlers/data"])

print("\n== 4. and it stays quiet when there is nothing to say")
good = Opener()
gb = good.bind(DataHandler())
report("a correct handler produces no near-miss and no refusal",
       gb.bound_anything and gb.ignored == [] and gb.refused == [],
       "otherwise the warnings are decoration that is always present")
report("and nothing is listed as inert", inert(good.bindings) == [])

print("\n== 5. SCL decides what a name is allowed to mean")
ok_data, _ = may_serve("handlers/data", "data")
ok_htp, why_htp = may_serve("handlers/data", "htp")
ok_cross, why_cross = may_serve("handlers/file", "data")
report("a permitted scheme passes", ok_data)
report("an unknown scheme is refused with the real set", not ok_htp and "data" in why_htp, why_htp)
report("a handler cannot claim another's scheme", not ok_cross, why_cross)

print("")
if failures:
    print(f"  {len(failures)} check(s) failed: {', '.join(failures)}")
    raise SystemExit(1)
print("  island test passed")

"""Bind four handlers, one of which is the upstream failure.

    python src/main.py
"""
from __future__ import annotations

import sys
import tempfile
from pathlib import Path

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))

from DMS.report import inert, render          # noqa: E402
from SMS.chain import Opener                  # noqa: E402
from TMS.handlers.data import DataHandler     # noqa: E402
from TMS.handlers.file import FileHandler     # noqa: E402


class TypoHandler:
    """The upstream failure, verbatim: one transposition and the handler is inert."""
    name = "handlers/data"

    def data_opne(self, url: str) -> str:      # noqa: N802 - deliberate
        return "never reached"


class WrongSchemeHandler:
    """Binds a protocol that does not exist. Upstream accepts this."""
    name = "handlers/data"

    def htp_open(self, url: str) -> str:
        return "never requested"


def main() -> int:
    root = Path(tempfile.mkdtemp())
    (root / "note.txt").write_text("served from disk\n", encoding="utf-8")

    opener = Opener()
    for handler in (DataHandler(), FileHandler(root=root), TypoHandler(), WrongSchemeHandler()):
        opener.bind(handler)

    print("\n== urllib opener re-cut")
    print(render(opener.bindings))

    print("\n  opening")
    for url in ("data:text/plain,hello", "file:/note.txt", "htp://example.com", "gopher://old"):
        ok, result = opener.open(url)
        print(f"    {url:<26} {'->' if ok else '  '} {result.strip()}")

    dead = inert(opener.bindings)
    print(f"\n  handlers that bound nothing: {', '.join(dead) if dead else 'none'}")

    # Upstream returns None for all four of these. The assertion is that this
    # version can tell the inert one from the working one.
    if not dead:
        print("\n  RE-CUT FAILED: the typo handler bound something, so nothing is being demonstrated")
        return 1
    if "htp" in opener.by_scheme:
        print("\n  RE-CUT FAILED: a nonexistent scheme was bound")
        return 1
    print("  a transposed letter is now a reported outcome rather than a silent one.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())