feat(proxies): add Control D proxy provider

Runs a loopback forwarder per region-named profile (unshackle-<region>), so --proxy and --proxy-download can exit in different countries Thanks to Orni_ for the original code and the test account. Co-authored-by: AntCardFr <24798811+AntCardFR@users.noreply.github.com>
2026-11-14 13:05:09 +00:00 · 2026-09-23 13:23:50 -06:00
parent ed3159378f
commit 783c8ae1c2
16 changed files with 1194 additions and 40 deletions
@@ -0,0 +1,402 @@
+"""Unit tests for the Control D proxy provider: the DoH answer parser, the loopback forwarder
+that turns a resolver into something Python-Requests can use as a proxy, the pool of
+region-named profiles, and the resolver form a client sends to a remote server."""
+
+from __future__ import annotations
+
+import base64
+import http.server
+import re
+import socket
+import socketserver
+import threading
+import time
+from types import SimpleNamespace
+from typing import Iterator
+
+import click
+import pytest
+import requests
+
+from unshackle.core.proxies import controld
+from unshackle.core.proxies.controld import (
+    MIN_TTL,
+    ControlD,
+    IPv4Adapter,
+    LocalProxy,
+    Resolver,
+    build_query,
+    local_proxy,
+    parse_a_records,
+)
+from unshackle.core.proxies.resolve import REGION, resolve_proxy
+
+pytestmark = pytest.mark.unit
+
+
+def dns_response(ip: bytes, ttl: int = 300) -> bytes:
+    """A one-question, one-answer A-record response whose answer name is a compression pointer."""
+    header = b"\x00\x00\x81\x80\x00\x01\x00\x01\x00\x00\x00\x00"
+    question = b"\x01a\x04test\x00" + b"\x00\x01\x00\x01"
+    answer = b"\xc0\x0c" + b"\x00\x01\x00\x01" + ttl.to_bytes(4, "big") + b"\x00\x04" + ip
+    return header + question + answer
+
+
+def test_query_is_a_well_formed_a_record_question() -> None:
+    query = build_query("a.test.")
+    assert query[4:6] == b"\x00\x01"
+    assert query[12:] == b"\x01a\x04test\x00\x00\x01\x00\x01"
+
+
+def test_parses_the_answer_past_a_compression_pointer() -> None:
+    assert parse_a_records(dns_response(bytes([1, 2, 3, 4]))) == [("1.2.3.4", 300)]
+    assert parse_a_records(b"") == []
+
+
+def origin(body: bytes) -> http.server.HTTPServer:
+    """A local HTTP origin that answers every GET with `body`, and keeps connections alive."""
+
+    class Origin(http.server.BaseHTTPRequestHandler):
+        protocol_version = "HTTP/1.1"
+
+        def do_GET(self) -> None:
+            self.send_response(200)
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def log_message(self, *_args: object) -> None:
+            pass
+
+    server = http.server.HTTPServer(("127.0.0.1", 0), Origin)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    return server
+
+
+@pytest.fixture
+def forwarder() -> Iterator[LocalProxy]:
+    """A forwarder whose resolver sends every hostname to 127.0.0.1, as Control D sends it to its proxy."""
+    proxy = LocalProxy(SimpleNamespace(resolve=lambda host: "127.0.0.1"))
+    threading.Thread(target=proxy.serve_forever, daemon=True).start()
+    yield proxy
+    proxy.shutdown()
+    proxy.server_close()
+
+
+def test_forwards_a_request_to_the_resolved_host(forwarder: LocalProxy) -> None:
+    pong = origin(b"pong")
+    try:
+        response = requests.get(
+            f"http://unresolvable.test:{pong.server_address[1]}/", proxies={"http": forwarder.uri}, timeout=10
+        )
+        assert response.status_code == 200
+        assert response.content == b"pong"
+    finally:
+        pong.shutdown()
+
+
+def test_a_reused_connection_never_carries_a_request_to_the_wrong_host(forwarder: LocalProxy) -> None:
+    """A client pools one proxy connection for every plain-HTTP host, and the relay is bound to the first."""
+    first, second = origin(b"A"), origin(b"B")
+    try:
+        with requests.Session() as session:
+            session.proxies = {"http": forwarder.uri}
+            assert session.get(f"http://a.test:{first.server_address[1]}/", timeout=10).content == b"A"
+            assert session.get(f"http://b.test:{second.server_address[1]}/", timeout=10).content == b"B"
+    finally:
+        first.shutdown()
+        second.shutdown()
+
+
+def test_connect_tunnels_bytes_both_ways(forwarder: LocalProxy) -> None:
+    """Every real download is HTTPS, so it all goes through CONNECT rather than absolute-form."""
+    echo = socketserver.TCPServer(
+        ("127.0.0.1", 0),
+        type(
+            "Echo",
+            (socketserver.BaseRequestHandler,),
+            {"handle": lambda self: self.request.sendall(self.request.recv(64)[::-1])},
+        ),
+    )
+    threading.Thread(target=echo.serve_forever, daemon=True).start()
+
+    try:
+        with socket.create_connection(forwarder.server_address[:2], timeout=10) as client:
+            client.sendall(f"CONNECT unresolvable.test:{echo.server_address[1]} HTTP/1.1\r\n\r\n".encode())
+            assert b"200 Connection established" in client.recv(128)
+            client.sendall(b"stressed")
+            assert client.recv(64) == b"desserts"
+    finally:
+        echo.shutdown()
+        echo.server_close()
+
+
+@pytest.mark.parametrize(
+    ("target", "status"),
+    [
+        ("1.2.3.4:443", b"502"),  # an IP literal would bypass Control D
+        ("[::1]:443", b"502"),
+        ("a.test:x", b"400"),
+    ],
+)
+def test_connect_refuses_what_it_cannot_redirect(forwarder: LocalProxy, target: str, status: bytes) -> None:
+    with socket.create_connection(forwarder.server_address[:2], timeout=10) as client:
+        client.sendall(f"CONNECT {target} HTTP/1.1\r\n\r\n".encode())
+        assert client.recv(128).startswith(b"HTTP/1.1 " + status)
+
+
+def test_an_unreachable_host_answers_502() -> None:
+    def refuse(host: str) -> str:
+        raise ConnectionError("no A record")
+
+    proxy = LocalProxy(SimpleNamespace(resolve=refuse))
+    threading.Thread(target=proxy.serve_forever, daemon=True).start()
+    try:
+        response = requests.get("http://unresolvable.test/", proxies={"http": proxy.uri}, timeout=10)
+        assert response.status_code == 502
+    finally:
+        proxy.shutdown()
+        proxy.server_close()
+
+
+def stub_session(response: bytes, seen: list) -> object:
+    class Session:
+        @staticmethod
+        def get(url: str, params: dict, headers: dict, timeout: int) -> object:
+            seen.append((url, params, headers))
+
+            class Answer:
+                content = response
+
+                @staticmethod
+                def raise_for_status() -> None:
+                    pass
+
+            return Answer()
+
+    return Session()
+
+
+def test_resolve_asks_control_d_in_wireformat_and_caches_it() -> None:
+    seen: list = []
+    resolver = Resolver("https://dns.controld.com/efgh456")
+    resolver.session = stub_session(dns_response(bytes([9, 8, 7, 6]), ttl=1), seen)
+
+    assert resolver.resolve("a.test") == "9.8.7.6"
+    url, params, headers = seen[0]
+    assert url == "https://dns.controld.com/efgh456"
+    assert headers == {"accept": "application/dns-message"}
+    assert params["dns"] == base64.urlsafe_b64encode(build_query("a.test")).rstrip(b"=").decode()
+    assert "=" not in params["dns"]  # RFC 8484 wants the padding stripped
+
+    assert resolver.resolve("a.test") == "9.8.7.6"
+    assert len(seen) == 1
+    assert resolver.cache["a.test"][1] > time.monotonic() + MIN_TTL - 1  # a 1s TTL is floored
+
+
+def test_resolve_raises_when_control_d_has_no_address() -> None:
+    resolver = Resolver("https://dns.controld.com/efgh456")
+    resolver.session = stub_session(b"\x00\x00\x81\x83" + b"\x00\x00" * 4, [])
+
+    with pytest.raises(ConnectionError):
+        resolver.resolve("nowhere.test")
+
+
+def test_resolve_asks_over_ipv4_and_ignores_environment_proxies() -> None:
+    """Control D answers a redirected name only over IPv4, and authorises the address that asked."""
+    resolver = Resolver("https://dns.controld.com/efgh456")
+    assert resolver.session.trust_env is False
+    assert isinstance(resolver.session.get_adapter("https://dns.controld.com/"), IPv4Adapter)
+
+
+CATALOGUE = [
+    {"PK": "YUL", "country": "CA", "city": "Montreal"},
+    {"PK": "YYZ", "country": "CA", "city": "Toronto"},
+    {"PK": "RES_YYZ", "country": "CA", "city": "Res Toronto", "hidden": True},
+    {"PK": "JFK", "country": "US", "city": "New York"},
+]
+
+
+@pytest.fixture
+def catalogue(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Serve Control D's location list without going near the network."""
+    response = SimpleNamespace(raise_for_status=lambda: None, json=lambda: {"body": {"proxies": CATALOGUE}})
+    monkeypatch.setattr("unshackle.core.proxies.controld.requests.get", lambda *a, **kw: response)
+
+
+class Account:
+    """A fake Control D account API: holds endpoints, records every call, and answers them."""
+
+    def __init__(self) -> None:
+        self.devices: list[dict] = []
+        self.calls: list[tuple[str, str, dict]] = []
+        self.fail_device = False
+
+    def device(self, name: str, profile: str) -> None:
+        self.devices.append(
+            {"PK": f"d-{name}", "name": name, "profile": {"PK": profile}, "resolvers": {"uid": f"r{profile}"}}
+        )
+
+    def request(self, method: str, url: str, data: dict, headers: dict, timeout: int) -> SimpleNamespace:
+        assert headers == {"authorization": "Bearer secret"}
+        path = url.removeprefix("https://api.controld.com")
+        self.calls.append((method, path, data))
+        body: dict = {}
+        if (method, path) == ("GET", "/devices"):
+            body = {"devices": [dict(x) for x in self.devices]}
+        elif (method, path) == ("POST", "/profiles"):
+            body = {"profiles": [{"PK": f"p{data['name'].replace('-', '')}"}]}
+        elif (method, path) == ("POST", "/devices"):
+            if self.fail_device:
+                raise requests.HTTPError("endpoint limit")
+            self.device(data["name"], data["profile_id"])
+            body = self.devices[-1]
+        elif method == "PUT" and path.startswith("/devices/"):
+            next(x for x in self.devices if x["PK"] == path.split("/")[2])["name"] = data["name"]
+        return SimpleNamespace(raise_for_status=lambda: None, json=lambda: {"body": body})
+
+    def moves(self) -> list[tuple[str, str]]:
+        """The (profile, location) of every default-rule change, in order."""
+        return [(path.split("/")[2], data["via"]) for method, path, data in self.calls if path.endswith("/default")]
+
+
+@pytest.fixture
+def account(catalogue: None, monkeypatch: pytest.MonkeyPatch) -> Iterator[Account]:
+    """A fake account, on a fresh process state whose forwarders are shut down when the test ends."""
+    made = Account()
+    monkeypatch.setattr("unshackle.core.proxies.controld.requests.request", made.request)
+    controld.FORWARDERS.clear()
+    controld.REGIONS.clear()
+    yield made
+    for server in controld.FORWARDERS.values():
+        server.shutdown()
+        server.server_close()
+    controld.FORWARDERS.clear()
+    controld.REGIONS.clear()
+
+
+def test_locations_come_from_control_d(account: Account) -> None:
+    provider = ControlD(token="secret")
+    assert provider.location("ca") in ("YUL", "YYZ")  # never the hidden residential one
+    assert provider.location("res_yyz") == "RES_YYZ"  # unless asked for by name
+    assert provider.location("jp") is None
+    assert repr(provider) == "2 Countries (4 Servers)"
+
+
+def test_a_query_steers_the_profile_default_rule(account: Account) -> None:
+    provider = ControlD(token="secret", profile="abcd123", resolver="efgh456", max_profiles=1)
+
+    assert provider.get_proxy("jp") is None
+    assert not account.moves()
+
+    uri = provider.get_proxy("jfk")
+    assert uri and uri.startswith("http://127.0.0.1:")
+    assert ("PUT", "/profiles/abcd123/default", {"do": "3", "via": "JFK", "status": "1"}) in account.calls
+
+    assert provider.get_proxy("JFK") == uri
+    assert len(account.moves()) == 1
+
+    # one profile only: a second country moves it and must clear its cache
+    cache = controld.FORWARDERS["https://dns.controld.com/efgh456"].resolver.cache
+    cache["cached.test"] = ("1.2.3.4", float("inf"))
+    assert provider.get_proxy("yul") == uri
+    assert account.moves()[-1] == ("abcd123", "YUL")
+    assert cache == {}
+
+
+def test_each_region_gets_its_own_profile_and_forwarder(account: Account) -> None:
+    """--proxy and --proxy-download in two countries must not move each other."""
+    provider = ControlD(
+        token="secret", profile="one", resolver="r1", profiles=[{"profile": "two", "resolver": "r2"}], max_profiles=2
+    )
+    us = provider.get_proxy("jfk")
+    ca = provider.get_proxy("yul")
+    assert us != ca
+    assert account.moves() == [("one", "JFK"), ("two", "YUL")]
+
+    # out of profiles: the oldest claim moves
+    assert provider.get_proxy("yyz") == us
+    assert account.moves()[-1] == ("one", "YYZ")
+    assert provider.get_proxy("yul") == ca
+    assert len(account.moves()) == 3
+
+
+def test_a_region_uses_the_endpoint_named_after_it_or_creates_one(account: Account) -> None:
+    account.device("unshackle-us", "kept")
+    account.device("Phone", "theirs")
+    account.device("unshackle-gb", "theirs")  # shares the user's own profile, so never touched
+    provider = ControlD(token="secret")
+
+    us = provider.get_proxy("us")
+    ca = provider.get_proxy("ca")
+    assert us != ca
+    assert ("POST", "/profiles", {"name": "unshackle-ca"}) in account.calls
+    assert [x[0] for x in account.moves()] == ["kept", "punshackleca"]
+    assert "theirs" not in provider.resolvers
+
+    # the REST API makes a provider per request; it must find the endpoint by name
+    assert ControlD(token="secret").get_proxy("ca") == ca
+    assert len(account.moves()) == 2
+
+
+def test_a_moved_profile_is_renamed_to_its_new_region(account: Account) -> None:
+    account.device("unshackle-us", "kept")
+    provider = ControlD(token="secret", max_profiles=1)
+
+    provider.get_proxy("us")
+    provider.get_proxy("ca")
+    assert ("PUT", "/profiles/kept", {"name": "unshackle-ca"}) in account.calls
+    assert account.devices[0]["name"] == "unshackle-ca"
+    assert account.moves()[-1][0] == "kept"
+
+
+def test_a_failed_endpoint_leaves_no_profile_behind(account: Account) -> None:
+    account.fail_device = True
+    with pytest.raises(requests.HTTPError):
+        ControlD(token="secret").get_proxy("us")
+    assert ("DELETE", "/profiles/punshackleus", None) in [(m, p, d or None) for m, p, d in account.calls]
+
+
+def test_a_bad_config_is_refused(catalogue: None) -> None:
+    with pytest.raises(ValueError):
+        ControlD(token="secret", profile="one")
+    with pytest.raises(ValueError):
+        ControlD(token="secret", max_profiles=0)
+    with pytest.raises(ValueError):
+        ControlD(token="secret", profile="one", resolver="bad id?")
+
+
+def test_a_server_runs_a_forwarder_for_a_resolver_a_client_sends(account: Account) -> None:
+    provider = ControlD(token="secret", profile="abcd123", resolver="efgh456")
+    sent = provider.get_remote_proxy("jfk")
+    assert sent == "controld://efgh456@dns.controld.com"
+
+    uri = resolve_proxy(sent, [])
+    assert uri and uri.startswith("http://127.0.0.1:")
+    assert local_proxy(sent) == uri  # one forwarder per resolver
+    with pytest.raises(ValueError):
+        local_proxy("controld://evil.example/x@dns.controld.com")
+
+
+def test_remote_sends_the_resolver_and_a_bare_region_skips_control_d(
+    account: Account, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    from unshackle.core import remote_service
+
+    provider = ControlD(token="secret", profile="abcd123", resolver="efgh456")
+    monkeypatch.setattr("unshackle.core.proxies.resolve.initialize_proxy_providers", lambda: [provider])
+
+    assert remote_service.resolve_remote_proxy_arg("controld:us") == "controld://efgh456@dns.controld.com"
+    with pytest.raises(click.ClickException):
+        remote_service.resolve_remote_proxy_arg("us")  # a loopback forwarder is useless to a server
+
+
+@pytest.mark.parametrize("query", ["ca", "us:seattle", "us-ny", "uk1", "yul", "jfk", "res_yyz", "nordvpn:ca"])
+def test_a_region_query_matches(query: str) -> None:
+    assert re.fullmatch(rf"(?:[a-z]+:)?{REGION}", query, re.IGNORECASE)
+    assert re.fullmatch(rf"(?:[a-z]+:){{0,2}}{REGION}", f"gluetun:{query}", re.IGNORECASE)
+
+
+@pytest.mark.parametrize("uri", ["nas:3128", "localhost:8080", "proxy.example.com:8080", "1.2.3.4:8080", "us:1080"])
+def test_a_host_and_port_is_never_a_region_query(uri: str) -> None:
+    assert not re.fullmatch(rf"(?:[a-z]+:){{0,2}}{REGION}", uri, re.IGNORECASE)