feat(remote): add --local-proxy to send server requests through the client

A reverse tunnel over the remote connection: the server's service traffic leaves through the client's own network. Only the client dials. Adds the tunnel-test command.

docs/dev/rest-api/remote-sessions.md

@@ -146,10 +146,14 @@ hood the client walks a remote session through its lifecycle.
       `controld://<resolver>@dns.controld.com`. The server counts it as a full proxy
       URI and runs its own forwarder for it. A bare region skips Control D, because its
       proxy is a forwarder on the client. See
-      [Control D](../../guide/proxies-and-vpn.md#control-d)
+      [Control D](../../guide/proxies-and-vpn.md#control-d).
       The client also downloads playlists and segments through the same proxy,
       as a local run does. `--no-proxy-download` and `--proxy-download` change
       that for downloads only.
+    - A `--local-proxy` goes as `tunnel:<id>`. The client opens
+      `GET /api/tunnel` first, and the server replaces the ID with a loopback SOCKS5
+      proxy that relays to the client's own connection. See
+      [Send the server's requests through your machine](../../guide/proxies-and-vpn.md#send-the-servers-requests-through-your-machine).
     - Track-selection hints (`range_`, `vcodec`, `quality`, `best_available`,
       `generate_dv`) so the server fetches the right manifests
     - Your language and audio codec selection (`-l`, `-vl`, `-al`, `-a`, `-fs`)
@@ -400,6 +404,11 @@ serve:
   gets a persistent per-account cache instead, so the account's tokens outlive the
   remote session. The server and the client both refuse a forwarded cache file that is not
   JSON or that names a type other than a date or time, and the service then authenticates again.
+- **A tunnel belongs to its API key.** The server accepts `tunnel:<id>` only from the API key
+  that opened the tunnel. Its SOCKS5 proxy listens on loopback only, and it needs a username and
+  password for that tunnel, so a raw `socks5h://127.0.0.1:<port>` proxy from a client fails.
+- **Only the client dials.** The server ignores a request from the client to open a
+  connection, so a tunnel never gives a client access to the server's network.
 - **The server redacts secrets in logs.** It redacts remote session IDs, service tags,
   and other user values before it writes them to the log.
 
@@ -427,6 +436,8 @@ The server mounts all these routes, even in `--remote-only` mode. Paths use the
 | Method | Path | Purpose |
 |---|---|---|
 | `POST` | `/api/session/create` | Build + authenticate a service instance; returns a session ID immediately |
+| `GET` | `/api/tunnel` | WebSocket that relays the connections of a `tunnel:<id>` proxy to the client; `?id=` reconnects |
+| `GET` | `/api/tunnel/check?id=` | Exit IP address and lookup time through the caller's tunnel, for `unshackle tunnel-test` |
 | `GET` | `/api/session/{id}/titles` | Fetch and cache the title list |
 | `POST` | `/api/session/{id}/tracks` | Resolve tracks, manifests, chapters, headers/cookies for a title |
 | `POST` | `/api/session/{id}/segments` | Resolve per-segment/track download descriptors |

docs/guide/cli-reference.md

@@ -207,6 +207,7 @@ order the service used, then renumbers the episodes into the order you asked for
 | Flag | Description |
 |---|---|
 | `--proxy` | Proxy URI, a country or location code resolved from configured providers, or `provider:region` (e.g. `nordvpn:ca`, `gluetun:us`, `protonvpn:de:berlin`, `controld:yul`). |
+| `--local-proxy` | With `--remote`, send the server's service requests through your machine's own connection. Works with no other proxy option. Check it first with `unshackle tunnel-test`. See [Proxies & VPN](proxies-and-vpn.md#send-the-servers-requests-through-your-machine). |
 | `--no-proxy` | Force-disable all proxy use. |
 | `--no-proxy-download` | Bypass the proxy for **all downloads** (manifest, licence, and auth stay proxied). |
 | `--proxy-download` | Use a different proxy for **all downloads**, in the same forms as `--proxy` (manifest, licence, and auth stay on `--proxy`). |
@@ -250,6 +251,7 @@ unshackle logs in to the service before the search. A service that declares a se
 |---|---|
 | `-p`, `--profile` | Profile for credentials and cookies. |
 | `--proxy` | Proxy URI, country or location code, or `provider:region` (e.g. `controld:ca`). |
+| `--local-proxy` | With `--remote`, send the server's search requests through your machine's own connection. Works with no other proxy option. |
 | `--no-proxy` | Force-disable all proxy use. |
 | `--remote` | Search on a remote unshackle server. The server logs in to the service. unshackle sends it your credentials, cookies and cache for the profile, and saves the cache that the server sends back. A service that searches with no login gets none of them. |
 | `--local` | Search with the local service code, also when the config sets `remote: true` for the service. |
@@ -285,6 +287,24 @@ The export file can be a `mediaexport` file from `dl --export`, an export from a
 
 ---
 
+## `tunnel-test`
+
+Check `--local-proxy` before a real `--remote` run. The command finds the IP address that this
+machine exits from, opens a tunnel to the server, and makes the server look up its exit IP address
+through the tunnel. It stops at the first step that fails.
+
+```
+unshackle tunnel-test [--server NAME]
+```
+
+| Flag | Description |
+|---|---|
+| `--server` | Name of the remote server from `remote_services` config, if you configure more than one. |
+
+See [Proxies & VPN](proxies-and-vpn.md#send-the-servers-requests-through-your-machine).
+
+---
+
 ## `setup`
 
 Install the external tools and make a first config.

docs/guide/proxies-and-vpn.md

@@ -22,11 +22,12 @@ a query to a real proxy URL: [basic static proxies](#basic-static-proxies),
 
 ## The `--proxy` flag
 
-Four options on the `dl` command control proxying at download time:
+Five options on the `dl` command control proxying at download time:
 
 | Option | Effect |
 |---|---|
 | `--proxy` | The proxy to use. Either an explicit URI, or a query that unshackle resolves against your configured providers. |
+| `--local-proxy` | With `--remote`, send the server's service requests through your machine's own connection. Works with no other proxy option. See [Send the server's requests through your machine](#send-the-servers-requests-through-your-machine). |
 | `--no-proxy` | Force **all** proxy use off. No providers are initialised and no proxy query is resolved. |
 | `--no-proxy-download` | Use the proxy for the manifest, licence, and authentication, but bypass it for the **downloads** themselves. |
 | `--proxy-download` | Use `--proxy` for the manifest, licence, and authentication, and a different proxy for the **downloads**. Takes the same forms as `--proxy`. |
@@ -204,6 +205,75 @@ sees your own IP, but the service does not.
 A `socks5://` proxy makes the DNS lookup of each host on your machine and gives no warning.
 Use `socks5h://` to make the proxy do the DNS lookups too.
 
+### Send the server's requests through your machine
+
+With `--remote`, the server sends the service requests, and they leave from the server's own
+connection. Add `--local-proxy` to send them through your machine's connection instead:
+
+```shell
+unshackle dl --remote --local-proxy EXAMPLE 81234567
+```
+
+A VPN app that covers your whole machine then also covers the server's requests, so the
+service sees your VPN's IP address. `--local-proxy` needs `--remote` (or a service with
+`remote: true`), and it works with no other proxy option: unshackle stops if you also give
+`--proxy`, `--no-proxy`, `--proxy-download` or `--no-proxy-download`. A proxy option from the `dl`
+config defaults does not count: `--local-proxy` turns it off for the run.
+
+A `--proxy` on `localhost` or a loopback address, such as the local port of a VPN app, stops a
+`--remote` run with `That proxy is on your machine, so --remote cannot use it`. Connect the VPN
+app for the whole machine, then use `--local-proxy` without `--proxy`.
+
+#### Test it first
+
+`unshackle tunnel-test` checks the route before a real run. Add `--server <name>` to test a
+server other than the default one:
+
+```shell
+unshackle tunnel-test
+```
+
+It does three steps, and it stops at the first one that fails:
+
+1. It looks up the IP address that your machine exits from.
+2. It opens a tunnel to the server.
+3. It makes the server look up its exit IP address through the tunnel. That address must be
+   the same as in step 1, or in the same country when one lookup used IPv4 and the other IPv6.
+   It fails when the lookup left from the server's own address. It also shows how long one
+   lookup takes.
+
+#### How the tunnel works
+
+1. The client opens a WebSocket on the server, on the same host and port as the other
+   remote session requests. Only a client opens a tunnel. The server never connects to
+   a host for a client, so a tunnel gives no access to the server's network.
+2. The server gives the service a SOCKS5 proxy on its loopback address, with a username and
+   password for that tunnel only. Another API key cannot use it.
+3. The server sends each connection that the service opens through the tunnel. The client
+   connects to the host and relays the bytes.
+
+The client logs `The server sends its requests through your connection on this machine`. You
+do not configure ports: the server picks a free one, and your
+machine opens none, because the client opens the connection to the server. A NAT router, a
+firewall, or a Cloudflare tunnel in front of the server does not stop it.
+
+The service traffic stays encrypted from the server to the service, so the client relays only
+encrypted bytes. The client refuses to connect to `localhost`, and to a host name or IP address
+that resolves only to a loopback, private or link-local address, so the server cannot reach
+your local network.
+
+!!! note "Each request crosses the tunnel twice"
+    A request from the server goes to your machine and then to the service, so it takes
+    longer than a request that the server sends directly. The segments do not go through the
+    tunnel: the client downloads them as before.
+
+A service that uses a server account does not work with `--local-proxy`, because the server
+logs in with its own account. If the tunnel drops, the client connects again, and a new request
+from the server waits up to 10 seconds for it. A request that was in progress when the tunnel
+dropped fails. The server closes a tunnel when its client has not connected for 60 seconds, and
+the client then stops with `The tunnel to the server closed`. An API key can have 4 tunnels
+open at once.
+
 ## Basic (static proxies)
 
 The `Basic` proxy provider is pure static configuration, with no accounts and no network

tests/remote/unit/test_remote_search.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 from types import SimpleNamespace
 from typing import Any
 
@@ -18,6 +19,7 @@ class FakeClient:
 
     def __init__(self) -> None:
         self.sent: dict[str, Any] = {}
+        self.tunnel: Any = None
 
     def post(self, endpoint: str, data: dict[str, Any]) -> dict[str, Any]:
         self.sent = {"endpoint": endpoint, **data}
@@ -69,3 +71,44 @@ def test_search_sends_no_login(server_accounts: Any, anonymous: bool) -> None:
     list(svc.search())
 
     assert client.sent == {"endpoint": "/api/search", "service": "TEST", "query": "query"}
+
+
+def test_local_proxy_sends_the_server_a_tunnel(monkeypatch: pytest.MonkeyPatch) -> None:
+    class FakeTunnel:
+        def __init__(self, server_url: str, headers: dict) -> None:
+            self.id = "abc"
+
+        def start(self) -> str:
+            return self.id
+
+        def stop(self) -> None:
+            self.stopped = True
+
+    monkeypatch.setattr(remote_service, "TunnelClient", FakeTunnel)
+    svc, client, _ = make_service({"local_proxy": True})
+    svc.log = logging.getLogger("test")
+    client.session = SimpleNamespace(headers={"X-Secret-Key": "k"})  # type: ignore[attr-defined]
+
+    list(svc.search())
+
+    assert client.sent["proxy"] == "tunnel:abc"
+    assert client.tunnel.stopped
+
+
+def test_a_loopback_proxy_points_to_local_proxy(monkeypatch: pytest.MonkeyPatch) -> None:
+    import click
+
+    monkeypatch.setattr(remote_service, "resolve_remote_proxy_arg", lambda arg, providers=None: arg)
+    svc, _, _ = make_service({"proxy": "http://127.0.0.1:8888"}, {"global": True})
+
+    with pytest.raises(click.ClickException, match="--local-proxy without --proxy"):
+        list(svc.search())
+
+
+def test_local_proxy_is_refused_for_a_server_account_service() -> None:
+    import click
+
+    svc, _, _ = make_service({"local_proxy": True}, {"global": True})
+
+    with pytest.raises(click.ClickException, match="its own account"):
+        list(svc.search())

tests/remote/unit/test_tunnel.py

@@ -0,0 +1,389 @@
+"""The reverse tunnel: a server-side SOCKS5 proxy whose connections the client dials and relays."""
+
+from __future__ import annotations
+
+import asyncio
+import http.server
+import socket
+import socketserver
+import threading
+import time
+from typing import Callable, Iterator
+
+import aiohttp
+import click
+import pytest
+import requests
+import responses
+from aiohttp import web
+
+from unshackle.core import tunnel
+from unshackle.core.api.routes import tunnel as tunnel_route
+from unshackle.core.api.routes import tunnel_check
+from unshackle.core.tunnel import (
+    CHECK_ROUTE,
+    OPEN,
+    ROUTE,
+    TUNNELS,
+    TunnelClient,
+    connect_public,
+    frame,
+    is_public_ip,
+    tunnel_proxy,
+)
+
+
+class Hello(http.server.BaseHTTPRequestHandler):
+    hits = 0
+
+    def do_GET(self) -> None:
+        Hello.hits += 1
+        body = b"hello through the tunnel"
+        self.send_response(200)
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, *_: object) -> None:
+        pass
+
+
+@pytest.fixture
+def target() -> Iterator[int]:
+    server = socketserver.ThreadingTCPServer(("127.0.0.1", 0), Hello)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    yield server.server_address[1]
+    server.shutdown()
+
+
+@pytest.fixture
+def server() -> Iterator[str]:
+    loop = asyncio.new_event_loop()
+    app = web.Application()
+    app.router.add_get(ROUTE, tunnel_route)
+    app.router.add_get(CHECK_ROUTE, tunnel_check)
+    runner = web.AppRunner(app)
+    loop.run_until_complete(runner.setup())
+    site = web.TCPSite(runner, "127.0.0.1", 0)
+    loop.run_until_complete(site.start())
+    port = site._server.sockets[0].getsockname()[1]  # type: ignore[union-attr]
+    threading.Thread(target=loop.run_forever, daemon=True).start()
+    yield f"http://127.0.0.1:{port}"
+    asyncio.run_coroutine_threadsafe(tunnel.close_tunnels(app), loop).result(5)
+    loop.call_soon_threadsafe(loop.stop)
+
+
+def test_requests_through_the_tunnel_reach_the_client_side_target(
+    server: str, target: int, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(tunnel, "is_public_ip", lambda address: True)  # the test target is on loopback
+    client = TunnelClient(server, {"X-Secret-Key": "alice"})
+    tunnel_id = client.start()
+
+    proxy = tunnel_proxy(tunnel_id, "alice")
+    assert proxy and proxy.startswith(f"socks5h://{tunnel_id}:") and "@127.0.0.1:" in proxy
+    for _ in range(3):
+        resp = requests.get(f"http://127.0.0.1:{target}/", proxies={"all": proxy}, timeout=10)
+        assert resp.text == "hello through the tunnel"
+    client.stop()
+
+
+def test_a_tunnel_is_only_usable_by_its_owner(server: str) -> None:
+    client = TunnelClient(server, {"X-Secret-Key": "alice"})
+    tunnel_id = client.start()
+    assert tunnel_proxy(tunnel_id, "mallory") is None
+    assert tunnel_proxy("unknown", "alice") is None
+    client.stop()
+
+
+def test_the_client_refuses_a_target_on_its_own_network(server: str, target: int) -> None:
+    client = TunnelClient(server, {"X-Secret-Key": "alice"})
+    proxy = tunnel_proxy(client.start(), "alice")
+    with pytest.raises(requests.exceptions.ConnectionError):
+        requests.get(f"http://127.0.0.1:{target}/", proxies={"all": proxy}, timeout=10)
+    client.stop()
+
+
+@pytest.mark.parametrize(
+    ("address", "public"),
+    [("8.8.8.8", True), ("127.0.0.1", False), ("192.168.1.10", False), ("10.0.0.1", False), ("::1", False)],
+)
+def test_is_public_ip(address: str, public: bool) -> None:
+    assert is_public_ip(address) is public
+
+
+def test_localhost_is_refused() -> None:
+    with pytest.raises(PermissionError):
+        asyncio.run(connect_public("localhost", 443))
+
+
+def test_a_name_that_resolves_to_a_local_address_is_refused(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def resolve(self: object, host: str, port: int, **_: object) -> list:
+        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.168.1.1", port))]
+
+    monkeypatch.setattr(asyncio.BaseEventLoop, "getaddrinfo", resolve)
+    with pytest.raises(PermissionError):
+        asyncio.run(connect_public("rebind.example", 443))
+
+
+def local_proxy_ctx(params: dict, typed: set) -> click.Context:
+    from click.core import ParameterSource
+
+    ctx = click.Context(click.Command("dl"))
+    ctx.params = {"local_proxy": True, **params}
+    for key in params:
+        ctx.set_parameter_source(key, ParameterSource.COMMANDLINE if key in typed else ParameterSource.DEFAULT_MAP)
+    return ctx
+
+
+def test_local_proxy_works_with_no_other_proxy_option() -> None:
+    from unshackle.core.proxies.resolve import require_alone_local_proxy
+
+    with pytest.raises(click.UsageError, match="--proxy, --proxy-download"):
+        require_alone_local_proxy(
+            local_proxy_ctx({"remote": True, "proxy": "us", "proxy_download": "us"}, {"proxy", "proxy_download"})
+        )
+    with pytest.raises(click.UsageError, match="needs --remote"):
+        require_alone_local_proxy(local_proxy_ctx({"remote": False}, set()))
+
+
+def test_local_proxy_turns_off_a_proxy_from_the_config_defaults() -> None:
+    from unshackle.core.proxies.resolve import require_alone_local_proxy
+
+    ctx = local_proxy_ctx({"remote": True, "proxy": "us", "no_proxy_download": True}, set())
+    require_alone_local_proxy(ctx)
+    assert ctx.params["proxy"] is None and ctx.params["no_proxy_download"] is False
+
+
+def test_the_tunnel_port_refuses_a_proxy_uri_without_the_tunnel_credentials(
+    server: str, target: int, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(tunnel, "is_public_ip", lambda address: True)
+    client = TunnelClient(server, {"X-Secret-Key": "alice"})
+    tunnel_id = client.start()
+    port = TUNNELS[tunnel_id].port
+    for uri in (f"socks5h://127.0.0.1:{port}", f"socks5h://{tunnel_id}:wrong@127.0.0.1:{port}"):
+        with pytest.raises(requests.exceptions.ConnectionError):
+            requests.get(f"http://127.0.0.1:{target}/", proxies={"all": uri}, timeout=10)
+    client.stop()
+
+
+def test_the_server_never_dials_for_a_client(server: str, target: int) -> None:
+    """An OPEN from the client must not make the server connect anywhere: only the client dials."""
+
+    async def send_open() -> None:
+        async with aiohttp.ClientSession(headers={"X-Secret-Key": "mallory"}) as http:
+            async with http.ws_connect(server + ROUTE) as ws:
+                await ws.receive_json()
+                await ws.send_bytes(frame(1, OPEN, target.to_bytes(2, "big") + b"127.0.0.1"))
+
+    Hello.hits = 0
+    asyncio.run(send_open())
+    assert Hello.hits == 0
+
+
+def test_the_check_route_is_only_for_the_tunnel_owner(server: str) -> None:
+    client = TunnelClient(server, {"X-Secret-Key": "alice"})
+    tunnel_id = client.start()
+    resp = requests.get(f"{server}{CHECK_ROUTE}?id={tunnel_id}", headers={"X-Secret-Key": "mallory"}, timeout=10)
+    assert resp.status_code == 404
+    client.stop()
+
+
+@pytest.mark.parametrize(
+    ("uri", "line"),
+    [
+        ("socks5://127.0.0.1:1080", "Using a local proxy on port 1080"),
+        ("http://localhost:8888", "Using a local proxy on port 8888"),
+        ("http://1.2.3.4:8080", "Using explicit Proxy: http://1.2.3.4:8080"),
+        ("http://localhost", "Using a local proxy"),
+        ("http://127.0.0.1:99999", "Using a local proxy"),
+    ],
+)
+def test_a_local_proxy_logs_only_its_port(uri: str, line: str) -> None:
+    from unshackle.core.proxies.resolve import describe_explicit_proxy
+
+    assert describe_explicit_proxy(uri) == line
+
+
+def test_another_api_key_cannot_reattach_to_a_tunnel(server: str) -> None:
+    client = TunnelClient(server, {"X-Secret-Key": "alice"})
+    tunnel_id = client.start()
+
+    async def reattach() -> None:
+        async with aiohttp.ClientSession(headers={"X-Secret-Key": "mallory"}) as http:
+            async with http.ws_connect(f"{server}{ROUTE}?id={tunnel_id}"):
+                pass
+
+    with pytest.raises(aiohttp.WSServerHandshakeError) as e:
+        asyncio.run(reattach())
+    assert e.value.status == 404
+    client.stop()
+
+
+def test_a_request_that_is_not_a_websocket_opens_no_tunnel(server: str) -> None:
+    before = len(TUNNELS)
+    resp = requests.get(server + ROUTE, headers={"X-Secret-Key": "alice"}, timeout=10)
+    assert resp.status_code == 400
+    assert len(TUNNELS) == before
+
+
+def test_a_malformed_frame_still_leaves_the_tunnel_to_expire(server: str) -> None:
+    async def send_short_frame() -> str:
+        async with aiohttp.ClientSession(headers={"X-Secret-Key": "mallory"}) as http:
+            async with http.ws_connect(server + ROUTE) as ws:
+                tunnel_id = (await ws.receive_json())["id"]
+                await ws.send_bytes(b"\x01")
+        return str(tunnel_id)
+
+    tunnel_id = asyncio.run(send_short_frame())
+    deadline = time.monotonic() + 5
+    while TUNNELS[tunnel_id].expiry is None and time.monotonic() < deadline:
+        time.sleep(0.05)
+    assert TUNNELS[tunnel_id].expiry is not None
+
+
+def test_an_api_key_has_a_tunnel_limit(server: str) -> None:
+    clients = [TunnelClient(server, {"X-Secret-Key": "bob"}) for _ in range(tunnel.MAX_TUNNELS)]
+    for client in clients:
+        client.start()
+    with pytest.raises(ConnectionError, match="429"):
+        TunnelClient(server, {"X-Secret-Key": "bob"}).start()
+    for client in clients:
+        client.stop()
+
+
+def test_a_connection_that_stops_reading_is_dropped() -> None:
+    sent: list = []
+
+    class FakeWS:
+        closed = False
+
+        async def send_bytes(self, data: bytes) -> None:
+            sent.append(data)
+
+    async def flood() -> tunnel.Link:
+        link = tunnel.Link(FakeWS())
+        link.streams[7] = tunnel.new_queue()
+        for _ in range(tunnel.BACKLOG + 5):
+            await link.feed(7, b"x")
+        return link
+
+    link = asyncio.run(flood())
+    assert 7 not in link.streams
+    assert sent == [frame(7, tunnel.CLOSE)]
+
+
+def test_a_download_job_refuses_a_tunnel_proxy(monkeypatch: pytest.MonkeyPatch) -> None:
+    from unshackle.core.api import handlers
+    from unshackle.core.api.errors import APIError
+
+    monkeypatch.setattr(handlers, "server_cdm_allowed", lambda request, service: True)
+    monkeypatch.setattr(handlers, "server_cdm_max_height", lambda request, service: None)
+    with pytest.raises(APIError, match="tunnel"):
+        handlers.enforce_download_gates({"service": "EXAMPLE", "proxy": "tunnel:abc"}, None)
+
+
+@responses.activate
+def test_a_server_error_after_the_tunnel_closed_names_the_tunnel() -> None:
+    from types import SimpleNamespace
+
+    from unshackle.core.remote_service import RemoteClient, RemoteError
+
+    responses.add(
+        responses.GET,
+        "http://srv:8786/api/session/x/titles",
+        json={"status": "error", "error_code": "SERVICE_ERROR", "message": "proxy refused"},
+        status=422,
+    )
+    client = RemoteClient("http://srv:8786", "")
+    client.tunnel = SimpleNamespace(lost=True)  # type: ignore[assignment]
+    with pytest.raises(RemoteError) as e:
+        client.get("/api/session/x/titles")
+    assert e.value.kind == "TUNNEL_CLOSED"
+
+
+def spy_links(monkeypatch: pytest.MonkeyPatch) -> list:
+    """Record every Link the client makes, so a test can look at its open streams."""
+    links: list = []
+
+    class SpyLink(tunnel.Link):
+        def __init__(self, ws: object) -> None:
+            super().__init__(ws)
+            links.append(self)
+
+    monkeypatch.setattr(tunnel, "Link", SpyLink)
+    return links
+
+
+def wait_until(check: "Callable[[], bool]", seconds: float = 3) -> bool:
+    deadline = time.monotonic() + seconds
+    while not check() and time.monotonic() < deadline:
+        time.sleep(0.05)
+    return check()
+
+
+def test_a_dial_the_server_gave_up_on_is_dropped_by_the_client(
+    server: str, target: int, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """When the server times out an OPEN, the client must not keep the connection it makes later."""
+    monkeypatch.setattr(tunnel, "is_public_ip", lambda address: True)
+    monkeypatch.setattr(tunnel, "OPEN_TIMEOUT", 0.3)
+    real_connect = tunnel.connect_public
+
+    async def slow_connect(host: str, port: int) -> tuple:
+        await asyncio.sleep(1)
+        return await real_connect(host, port)
+
+    monkeypatch.setattr(tunnel, "connect_public", slow_connect)
+    links = spy_links(monkeypatch)
+    client = TunnelClient(server, {"X-Secret-Key": "alice"})
+    proxy = tunnel_proxy(client.start(), "alice")
+    with pytest.raises(requests.exceptions.ConnectionError):
+        requests.get(f"http://127.0.0.1:{target}/", proxies={"all": proxy}, timeout=10)
+    assert wait_until(lambda: not links[-1].streams and not client.relays)
+    client.stop()
+
+
+def socks5_connect(proxy: str, host: bytes) -> int:
+    """Send one SOCKS5 CONNECT for ``host`` with the credentials in ``proxy`` and return the reply code."""
+    from urllib.parse import urlsplit
+
+    parts = urlsplit(proxy)
+    user, password = (parts.username or "").encode(), (parts.password or "").encode()
+    with socket.create_connection(("127.0.0.1", parts.port or 0), timeout=10) as sock:
+        sock.sendall(b"\x05\x01\x02")
+        assert sock.recv(2) == b"\x05\x02"
+        sock.sendall(b"\x01" + bytes([len(user)]) + user + bytes([len(password)]) + password)
+        assert sock.recv(2) == b"\x01\x00"
+        sock.sendall(b"\x05\x01\x00\x03" + bytes([len(host)]) + host + (443).to_bytes(2, "big"))
+        return sock.recv(10)[1]
+
+
+def test_a_hostname_the_resolver_rejects_closes_its_stream(server: str, monkeypatch: pytest.MonkeyPatch) -> None:
+    async def resolve(self: object, host: str, port: int, **_: object) -> list:
+        host.encode("idna")  # as socket.getaddrinfo does, so a long label raises UnicodeError
+        return []
+
+    monkeypatch.setattr(asyncio.BaseEventLoop, "getaddrinfo", resolve)
+    monkeypatch.setattr(tunnel, "OPEN_TIMEOUT", 1)
+    links = spy_links(monkeypatch)
+    client = TunnelClient(server, {"X-Secret-Key": "alice"})
+    proxy = tunnel_proxy(client.start(), "alice")
+    assert proxy
+    assert socks5_connect(proxy, b"a" * 64 + b".example") == 5
+    assert wait_until(lambda: not links[-1].streams)
+    client.stop()
+
+
+def test_a_failed_start_leaves_no_thread_behind() -> None:
+    silent = socket.socket()
+    silent.bind(("127.0.0.1", 0))
+    silent.listen()
+    try:
+        client = TunnelClient(f"http://127.0.0.1:{silent.getsockname()[1]}", {"X-Secret-Key": "alice"})
+        with pytest.raises(ConnectionError, match="timed out"):
+            client.start(timeout=0.5)
+        assert not client.thread.is_alive()
+    finally:
+        silent.close()

unshackle/commands/dl.py

@@ -62,10 +62,12 @@ from unshackle.core.providers.anilist import parse_anilist_ref
 from unshackle.core.providers.tvdb import SEASON_TYPES, parse_int
 from unshackle.core.proxies.resolve import (
     REGION,
+    describe_explicit_proxy,
     describe_proxy,
     is_loopback,
     load_proxy_providers,
     pick_proxy,
+    require_alone_local_proxy,
     resolve_proxy,
     split_proxy_query,
 )
@@ -922,6 +924,12 @@ class dl:
         default=None,
         help="Proxy URI to use. If a 2-letter country is provided, it will try to get a proxy from the config.",
     )
+    @click.option(
+        "--local-proxy",
+        is_flag=True,
+        default=False,
+        help="With --remote, the server sends its service requests out through this machine's own connection (and any VPN on it). Works with no other proxy option.",
+    )
     @click.option(
         "--tag",
         type=str,
@@ -1598,6 +1606,10 @@ class dl:
                 if cdm_info:
                     log_event("load_cdm", level="INFO", service=self.service, context={"cdm": cdm_info})
 
+        if ctx.params.get("local_proxy"):
+            require_alone_local_proxy(ctx)
+            proxy, no_proxy = ctx.params.get("proxy"), bool(ctx.params.get("no_proxy"))
+
         # A server on this machine can reach a local proxy, so only a server elsewhere is guarded
         self.remote_needs_public_proxy = self.is_remote and not is_loopback(server_url(ctx.params.get("server")))
 
@@ -1644,7 +1656,7 @@ class dl:
                     ctx.params["proxy_query"] = proxy_query
                     ctx.params["proxy_provider"] = requested_provider
                 else:
-                    self.log.info(f"Using explicit Proxy: {mask_proxy(ctx.params['proxy'])}")
+                    self.log.info(describe_explicit_proxy(ctx.params["proxy"]))
                     ctx.params["proxy_query"] = None
                     ctx.params["proxy_provider"] = None
 

unshackle/commands/search.py

@@ -16,12 +16,19 @@ from unshackle.core.config import config
 from unshackle.core.console import console
 from unshackle.core.constants import context_settings
 from unshackle.core.proxies.proxy import Proxy
-from unshackle.core.proxies.resolve import REGION, describe_proxy, load_proxy_providers, pick_proxy, split_proxy_query
+from unshackle.core.proxies.resolve import (
+    REGION,
+    describe_explicit_proxy,
+    describe_proxy,
+    load_proxy_providers,
+    pick_proxy,
+    require_alone_local_proxy,
+    split_proxy_query,
+)
 from unshackle.core.service import Service
 from unshackle.core.services import Services
 from unshackle.core.utils.click_types import ContextData
 from unshackle.core.utils.collections import ci_get, merge_dict
-from unshackle.core.utils.redact import mask_proxy
 
 
 @click.command(
@@ -38,6 +45,12 @@ from unshackle.core.utils.redact import mask_proxy
     default=None,
     help="Proxy URI to use. If a 2-letter country is provided, it will try get a proxy from the config.",
 )
+@click.option(
+    "--local-proxy",
+    is_flag=True,
+    default=False,
+    help="With --remote, the server sends its service requests out through this machine's own connection (and any VPN on it). Works with no other proxy option.",
+)
 @click.option("--no-proxy", is_flag=True, default=False, help="Force disable all proxy use.")
 @click.option(
     "--remote/--local",
@@ -58,6 +71,7 @@ def search(
     no_proxy: bool,
     profile: Optional[str] = None,
     proxy: Optional[str] = None,
+    local_proxy: bool = False,
     remote: bool = False,
     server: Optional[str] = None,
 ):
@@ -81,6 +95,10 @@ def search(
                 log.info("Service Config loaded")
     merge_dict(ci_get(config.services, service), service_config)
 
+    if ctx.params.get("local_proxy"):
+        require_alone_local_proxy(ctx)
+        proxy, no_proxy = ctx.params.get("proxy"), bool(ctx.params.get("no_proxy"))
+
     proxy_providers: list[Proxy] = []
     if no_proxy:
         ctx.params["proxy"] = None
@@ -106,7 +124,7 @@ def search(
                     ctx.params["proxy"] = proxy_uri
                     log.info(describe_proxy(proxy_provider, proxy_uri, query))
             else:
-                log.info(f"Using explicit Proxy: {mask_proxy(ctx.params['proxy'])}")
+                log.info(describe_explicit_proxy(ctx.params["proxy"]))
 
     ctx.obj = ContextData(config=service_config, cdm=None, proxy_providers=proxy_providers, profile=profile)
 

unshackle/commands/tunnel_test.py

@@ -0,0 +1,45 @@
+import logging
+from typing import Optional
+
+import click
+
+from unshackle.core.remote_service import remote_client, start_tunnel
+from unshackle.core.tunnel import CHECK_ROUTE
+from unshackle.core.utils.ip_info import get_ip_info
+from unshackle.core.utils.redact import safe_display_url
+
+
+@click.command(short_help="Check that a remote server can send its requests through this machine.")
+@click.option(
+    "--server", type=str, default=None, help="Name of the remote server from remote_services config (if multiple)."
+)
+def tunnel_test(server: Optional[str]) -> None:
+    """Check --local-proxy before a real --remote run.
+
+    The command finds the exit IP of this machine, opens a tunnel to the server, and makes the server
+    look up its exit IP through the tunnel. The two must match.
+    """
+    log = logging.getLogger("tunnel-test")
+
+    local = get_ip_info()
+    if not local:
+        raise click.ClickException("Step 1 failed: an IP lookup from this machine did not work.")
+    log.info(f"Step 1: this machine exits at {local['ip']} ({str(local.get('country', '?')).upper()})")
+
+    client = remote_client(server)
+    tunnel = start_tunnel(client)
+    log.info(f"Step 2: opened a tunnel to {safe_display_url(client.server_url)}")
+    try:
+        remote = client.get(f"{CHECK_ROUTE}?id={tunnel.id}")
+    finally:
+        tunnel.stop()
+
+    country = str(remote.get("country", "?")).upper()
+    # one exit can answer over IPv4 here and IPv6 there, so a matching country also passes
+    if remote.get("ip") != local["ip"] and remote.get("country") != local.get("country"):
+        raise click.ClickException(
+            f"Step 3 failed: the server's requests exit at {remote.get('ip')} ({country}), "
+            f"not where this machine exits. Report this with the IP addresses."
+        )
+    log.info(f"Step 3: the server's requests exit at {remote['ip']} ({country}), {remote.get('ms')} ms for one lookup")
+    log.info("The tunnel works. Add --local-proxy to a --remote command.")

unshackle/core/api/handlers.py

@@ -38,6 +38,7 @@ from unshackle.core.session import utf8_text
 from unshackle.core.titles import Episode, Movie, Song, Title_T
 from unshackle.core.titles.music import SONG_FIELDS
 from unshackle.core.tracks import Audio, Subtitle, Tracks, Video
+from unshackle.core.tunnel import HEARTBEAT, MAX_MESSAGE, MAX_TUNNELS, TUNNELS, Tunnel, tunnel_proxy
 from unshackle.core.utilities import declared_kwargs, kind_only_conflict, original_tag
 from unshackle.core.utils.click_types import AUDIO_CODEC_LIST, QUALITY_LIST, SUBTITLE_CODEC, VIDEO_CODEC_LIST
 from unshackle.core.utils.collections import ci_get
@@ -2376,6 +2377,12 @@ def enforce_download_gates(params: Dict[str, Any], request: Optional[web.Request
     if params.get("proxy_download") is not None and not isinstance(params["proxy_download"], str):
         raise APIError(APIErrorCode.INVALID_INPUT, "proxy_download must be a string.")
 
+    if any(str(params.get(key) or "").startswith("tunnel:") for key in ("proxy", "proxy_download")):
+        raise APIError(
+            APIErrorCode.INVALID_PROXY,
+            "A download job cannot use a tunnel: proxy, because a tunnel belongs to one --remote client run.",
+        )
+
     if not server_proxy_allowed(request):
         resolve_handler_proxy(params, params.get("service") or "", request)
         if params.get("proxy_download"):
@@ -4527,7 +4534,7 @@ def resolve_handler_proxy(
 
     if (
         proxy_param
-        and re.match(r"^(https?://|socks|controld://)", str(proxy_param))
+        and re.match(r"^(https?://|socks|controld://|tunnel:)", str(proxy_param))
         and server_account_for(request, normalized_service)
     ):
         raise APIError(
@@ -4536,6 +4543,17 @@ def resolve_handler_proxy(
             details={"service": normalized_service},
         )
 
+    if isinstance(proxy_param, str) and proxy_param.startswith("tunnel:") and not no_proxy:
+        proxy_param = tunnel_proxy(
+            proxy_param.removeprefix("tunnel:"), request_secret_key(request) if request else None
+        )
+        if not proxy_param:
+            raise APIError(
+                APIErrorCode.INVALID_PROXY,
+                "The tunnel for this proxy is not open. Run the command again.",
+                details={"service": normalized_service},
+            )
+
     if allowed and not no_proxy:
         proxy_providers = initialize_proxy_providers()
 
@@ -5637,3 +5655,48 @@ async def session_delete_handler(session_id: str, request: Optional[web.Request]
     if cache_data:
         response["cache"] = cache_data
     return web.json_response(response)
+
+
+async def tunnel_handler(request: web.Request) -> web.StreamResponse:
+    """Open a ``--local-proxy`` tunnel, or reattach to the caller's tunnel named by ``?id=``."""
+    owner = request_secret_key(request)
+    tunnel_id = request.query.get("id")
+    tunnel = TUNNELS.get(tunnel_id) if tunnel_id else None
+    if tunnel_id and (not tunnel or tunnel.owner != owner):
+        raise APIError(APIErrorCode.NOT_FOUND, "No open tunnel has this id.")
+    if not tunnel and sum(t.owner == owner for t in TUNNELS.values()) >= MAX_TUNNELS:
+        raise APIError(APIErrorCode.RATE_LIMITED, f"An API key can have {MAX_TUNNELS} tunnels open at once.")
+    ws = web.WebSocketResponse(heartbeat=HEARTBEAT, max_msg_size=MAX_MESSAGE)
+    await ws.prepare(request)  # a request that is not a WebSocket upgrade fails here, before a tunnel exists
+    if not tunnel:
+        tunnel = Tunnel(owner)
+        await tunnel.start()
+    await tunnel.serve(ws)
+    return ws
+
+
+async def tunnel_check_handler(request: web.Request) -> web.Response:
+    """Look up the exit IP through the caller's tunnel ``?id=``, so the client can confirm the route before a real run."""
+    import requests
+
+    from unshackle.core.utils.ip_info import verify_proxy_exit
+
+    proxy = tunnel_proxy(request.query.get("id", ""), request_secret_key(request))
+    if not proxy:
+        raise APIError(APIErrorCode.NOT_FOUND, "No open tunnel has this id.")
+    with requests.Session() as session:
+        session.proxies = {"all": proxy}
+        session.trust_env = False
+        started = time.perf_counter()
+        try:
+            info = await asyncio.to_thread(verify_proxy_exit, session)
+        except ConnectionError as e:
+            raise APIError(APIErrorCode.INVALID_PROXY, str(e), http_status=502)
+    if not info:
+        raise APIError(
+            APIErrorCode.INVALID_PROXY,
+            "The server could not look up an IP address through the tunnel.",
+            http_status=502,
+        )
+    ms = round((time.perf_counter() - started) * 1000)
+    return web.json_response({"ip": info.get("ip"), "country": info.get("country"), "ms": ms})

unshackle/core/api/routes.py

@@ -66,9 +66,14 @@ from unshackle.core.api.handlers import (
     session_segments_handler,
     session_titles_handler,
     session_tracks_handler,
+    tunnel_check_handler,
+    tunnel_handler,
 )
 from unshackle.core.drm import cbcs
 from unshackle.core.services import Services
+from unshackle.core.tunnel import CHECK_ROUTE as TUNNEL_CHECK_ROUTE
+from unshackle.core.tunnel import ROUTE as TUNNEL_ROUTE
+from unshackle.core.tunnel import close_tunnels
 from unshackle.core.update_checker import UpdateChecker
 from unshackle.core.utils.redact import redact_path
 
@@ -485,7 +490,7 @@ async def search(request: web.Request) -> web.Response:
                 description: Profile to use for credentials and cookies (default - None)
               proxy:
                 type: string
-                description: Full proxy URI, a Control D resolver as controld://<resolver>@dns.controld.com, or a country code when the API key has server_proxy (default - None)
+                description: Full proxy URI, a Control D resolver as controld://<resolver>@dns.controld.com, or a country code when the API key has server_proxy, or tunnel:<id> for a tunnel the same API key opened on /api/tunnel (default - None)
               no_proxy:
                 type: boolean
                 description: Force disable all proxy use (default - false)
@@ -1848,6 +1853,7 @@ async def session_create(request: web.Request) -> web.Response:
                 type: string
               proxy:
                 type: string
+                description: As on /api/search, including tunnel:<id> for a --local-proxy client
               no_proxy:
                 type: boolean
               profile:
@@ -1997,6 +2003,57 @@ async def session_create(request: web.Request) -> web.Response:
         )
 
 
+@api_handler
+async def tunnel(request: web.Request) -> web.StreamResponse:
+    """
+    Open a --local-proxy tunnel.
+    ---
+    summary: Open tunnel
+    description: >-
+      A WebSocket that a --remote client holds open. The server sends the requests of a remote session
+      created with proxy tunnel:<id> through it, and the client makes them from its own connection.
+      The first message is {"id": <tunnel id>}.
+    parameters:
+      - name: id
+        in: query
+        required: false
+        description: Reattach to this tunnel after a drop
+        schema:
+          type: string
+    responses:
+      '101':
+        description: WebSocket open
+      '404':
+        description: No open tunnel with this id for this API key
+      '429':
+        description: The API key has the maximum number of tunnels open
+    """
+    return await tunnel_handler(request)
+
+
+@api_handler
+async def tunnel_check(request: web.Request) -> web.Response:
+    """
+    Look up the exit IP through a tunnel.
+    ---
+    summary: Check tunnel
+    parameters:
+      - name: id
+        in: query
+        required: true
+        schema:
+          type: string
+    responses:
+      '200':
+        description: The exit IP and country through the tunnel, and the lookup time in ms
+      '404':
+        description: No open tunnel with this id for this API key
+      '502':
+        description: No IP lookup got through the tunnel, or the requests left from the server itself
+    """
+    return await tunnel_check_handler(request)
+
+
 @api_handler
 @heartbeat
 async def session_titles(request: web.Request) -> web.Response:
@@ -3260,6 +3317,8 @@ ROUTES: list[tuple[str, str, Handler, bool]] = [
     ("GET", "/api/remote/logins", remote_logins, False),
     ("POST", "/api/remote/logins/{login_id}/input", remote_login_input, False),
     ("POST", "/api/session/create", session_create, True),
+    ("GET", TUNNEL_ROUTE, tunnel, True),
+    ("GET", TUNNEL_CHECK_ROUTE, tunnel_check, True),
     ("GET", "/api/session/{session_id}/titles", session_titles, True),
     ("POST", "/api/session/{session_id}/tracks", session_tracks, True),
     ("POST", "/api/session/{session_id}/segments", session_segments, True),
@@ -3304,6 +3363,7 @@ def setup_routes(
         if remote_only and not remote:
             continue
         add[method](path, handler)
+    app.on_shutdown.append(close_tunnels)
 
 
 def setup_swagger(app: web.Application, dashboard: bool = False) -> None:

unshackle/core/proxies/resolve.py

@@ -8,6 +8,9 @@ import re
 from typing import Any, List, Optional
 from urllib.parse import urlparse
 
+import click
+from click.core import ParameterSource
+
 from unshackle.core.proxies.proxy import Proxy
 from unshackle.core.utils.redact import mask_proxy
 
@@ -189,6 +192,38 @@ def describe_proxy(provider: Any, uri: str, query: str, allow_debug: bool = True
     return f"Using {name} Proxy: {mask_proxy(uri, isinstance(provider, Basic), allow_debug)}"
 
 
+def require_alone_local_proxy(ctx: click.Context) -> None:
+    """Check ``--local-proxy``: it needs a remote run and no other proxy option.
+
+    A proxy option that comes from the ``dl`` config defaults, not the command line, is cleared instead.
+    """
+    if not ctx.params.get("remote"):
+        raise click.UsageError(
+            "--local-proxy needs --remote: it sends a remote server's requests through this machine."
+        )
+    given = []
+    for key in ("proxy", "no_proxy", "proxy_download", "no_proxy_download"):
+        if not ctx.params.get(key):
+            continue
+        if ctx.get_parameter_source(key) in (ParameterSource.DEFAULT, ParameterSource.DEFAULT_MAP):
+            ctx.params[key] = False if key.startswith("no_") else None
+        else:
+            given.append("--" + key.replace("_", "-"))
+    if given:
+        raise click.UsageError(f"--local-proxy does not work with {', '.join(given)}.")
+
+
+def describe_explicit_proxy(uri: str) -> str:
+    """The log line for a proxy URI typed in full: a proxy on this machine shows only as its port."""
+    if is_loopback(uri):
+        try:
+            port = urlparse(uri if "//" in uri else f"//{uri}").port
+        except ValueError:
+            port = None
+        return f"Using a local proxy on port {port}" if port else "Using a local proxy"
+    return f"Using explicit Proxy: {mask_proxy(uri)}"
+
+
 def resolve_proxy(proxy: str, proxy_providers: List[Any]) -> Optional[str]:
     """Change a proxy parameter to a real proxy URI.
 

unshackle/core/remote_service.py

@@ -50,6 +50,7 @@ from unshackle.core.titles.music import SONG_FIELDS, Album, Song
 from unshackle.core.tracks import Audio, Chapter, Chapters, Subtitle, Tracks, Video
 from unshackle.core.tracks.attachment import Attachment
 from unshackle.core.tracks.track import Track
+from unshackle.core.tunnel import TunnelClient
 from unshackle.core.utils import run_proxy
 from unshackle.core.utils.collections import ci_get
 from unshackle.core.utils.redact import redact_path, redact_text, safe_display_url
@@ -104,6 +105,7 @@ class RemoteClient:
         self.auth_headers = list(auth_headers) if auth_headers else list(DEFAULT_AUTH_HEADERS)
         self._auth_header_index = 0
         self._session: Optional[requests.Session] = None
+        self.tunnel: Optional[TunnelClient] = None
         # wraps every request, so the service's CDM calls get answered while the server runs its code
         self.in_flight: Callable[[str], contextlib.AbstractContextManager[Any]] = lambda _: contextlib.nullcontext()
 
@@ -192,6 +194,12 @@ class RemoteClient:
             if optional:
                 log.debug(f"Optional endpoint {endpoint} unavailable [{error_code}]: {error_msg}")
                 return {}
+            if self.tunnel and self.tunnel.lost:
+                raise RemoteError(
+                    "The tunnel to the server closed, so the server can no longer send its requests through this "
+                    "machine. Run the command again.",
+                    "TUNNEL_CLOSED",
+                )
             raise RemoteError(f"Server error [{error_code}]: {error_msg}", error_code)
         return detail
 
@@ -920,6 +928,18 @@ def resolve_remote_proxy_arg(proxy_arg: str, providers: Optional[list] = None) -
         raise click.ClickException(str(e))
 
 
+def start_tunnel(client: RemoteClient) -> TunnelClient:
+    """Open a tunnel that sends the server's requests out through this machine's own connection."""
+    headers = {k: str(v) for k, v in client.session.headers.items() if k.lower() != "connection"}
+    tunnel = TunnelClient(client.server_url, headers)
+    try:
+        tunnel.start()
+    except ConnectionError as e:
+        raise click.ClickException(f"--local-proxy needs a tunnel to the server. {e}")
+    client.tunnel = tunnel
+    return tunnel
+
+
 def server_proxy_arg(params: Dict[str, Any]) -> str:
     """The --proxy value to resolve for a server: the one typed when it resolved to a proxy on this machine."""
     proxy = params["proxy"]
@@ -1164,11 +1184,17 @@ class RemoteService:
             data.update(self.client_login(profile))
         if params.get("no_proxy"):
             data["no_proxy"] = True
+        elif params.get("local_proxy"):
+            data["proxy"] = self.tunnel()
         elif params.get("proxy"):
             providers = getattr(self.ctx.obj, "proxy_providers", None)
             data["proxy"] = resolve_remote_proxy_arg(server_proxy_arg(params), providers)
             self.require_reachable_proxy(data["proxy"])
-        response = self.client.post("/api/search", data)
+        try:
+            response = self.client.post("/api/search", data)
+        finally:
+            if self.client.tunnel:  # a search is one request, so its tunnel ends with it
+                self.client.tunnel.stop()
         self.save_returned_cache(response.get("cache") or {}, profile)
         for result in response.get("results") or []:
             yield SearchResult(
@@ -1189,7 +1215,22 @@ class RemoteService:
     def require_reachable_proxy(self, proxy: Optional[str]) -> None:
         """Refuse a proxy on this machine when the server is elsewhere: the server cannot reach it."""
         if proxy and is_loopback(proxy) and not is_loopback(self.client.server_url):
-            raise click.ClickException("That proxy is on your machine, so --remote cannot use it.")
+            raise click.ClickException(
+                "That proxy is on your machine, so --remote cannot use it. "
+                "To send the server's requests through this machine's connection, use --local-proxy without --proxy."
+            )
+
+    def tunnel(self) -> str:
+        """The ``tunnel:<id>`` proxy to send the server for ``--local-proxy``, opening the tunnel the first time."""
+        if self._server_accounts is not None:
+            raise click.ClickException(
+                f"The server logs in to {self.service_tag} with its own account, so --local-proxy does not work with it."
+            )
+        tunnel = self.client.tunnel
+        if tunnel is None:
+            tunnel = start_tunnel(self.client)
+            self.log.info("The server sends its requests through your connection on this machine")
+        return f"tunnel:{tunnel.id}"
 
     def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None:
         self.credential = credential
@@ -1219,7 +1260,10 @@ class RemoteService:
                 pass
 
         providers = getattr(self.ctx.obj, "proxy_providers", None)
-        if not no_proxy and not proxy and self._server_accounts and not self._server_accounts.get("global"):
+        local = bool(self.ctx.parent and self.ctx.parent.params.get("local_proxy"))
+        if local:
+            create_data["proxy"] = self.tunnel()
+        elif not no_proxy and not proxy and self._server_accounts and not self._server_accounts.get("global"):
             regions = [str(r).lower() for r in self._server_accounts.get("regions") or []]
             if regions and client_region and client_region not in regions:
                 try:
@@ -1241,7 +1285,7 @@ class RemoteService:
                 if country:
                     create_data["proxy_region"] = country.group(1).lower()
         self.require_reachable_proxy(create_data.get("proxy"))
-        local_proxy = None if no_proxy else (proxy or create_data.get("proxy"))
+        local_proxy = None if no_proxy or local else (proxy or create_data.get("proxy"))
         if local_proxy:
             self._session.proxies["all"] = local_proxy
             run_proxy.pin(self._session)
@@ -1334,7 +1378,12 @@ class RemoteService:
             "argv": argv,
         }
 
-        result = self.client.post("/api/session/create", create_data)
+        try:
+            result = self.client.post("/api/session/create", create_data)
+        except BaseException:
+            if local and self.client.tunnel:  # no session, so close() never runs: free the tunnel now
+                self.client.tunnel.stop()
+            raise
         self._session_id = result["session_id"]
         atexit.register(self.close)
         # an older server omits the field, and the tracks response settles it
@@ -1975,6 +2024,8 @@ class RemoteService:
                 pass
             except Exception as e:
                 self.log.warning(f"Failed to clean up remote session: {e}")
+        if self.client.tunnel:
+            self.client.tunnel.stop()
 
     def save_returned_cache(self, cache_data: Dict[str, str], profile: Optional[str] = None) -> None:
         """Save cache files returned by the server to the local cache directory.

unshackle/core/tunnel.py

@@ -0,0 +1,412 @@
+"""Reverse tunnel: a remote client's own network as the egress of its server-side service.
+
+The client holds a WebSocket open on ``/api/tunnel`` before it creates its remote session. The server gives the
+service a SOCKS5 proxy on loopback, and sends every connection that proxy accepts to the client, which
+dials it from its own connection and relays the bytes. Only the client dials. TLS stays end to end between
+the server and the service, so the client sees only ciphertext.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import ipaddress
+import itertools
+import json
+import logging
+import secrets
+import socket
+import struct
+import threading
+from typing import Any, Optional
+
+import aiohttp
+from aiohttp import WSMsgType, web
+
+log = logging.getLogger("tunnel")
+
+OPEN, OK, DATA, CLOSE = range(4)
+HEADER = struct.Struct("!IB")  # stream id, frame kind
+HEARTBEAT = 20
+GRACE = 60
+RELINK_WAIT = 10
+OPEN_TIMEOUT = 15
+MAX_TUNNELS = 4
+MAX_STREAMS = 256
+CHUNK = 65536
+BACKLOG = 64
+MAX_MESSAGE = 4 * CHUNK
+ROUTE = "/api/tunnel"
+CHECK_ROUTE = "/api/tunnel/check"
+
+
+def frame(sid: int, kind: int, data: bytes = b"") -> bytes:
+    return HEADER.pack(sid, kind) + data
+
+
+def new_queue() -> asyncio.Queue[Optional[bytes]]:
+    return asyncio.Queue(maxsize=BACKLOG + 1)  # the extra slot always takes the end marker
+
+
+class Link:
+    """One WebSocket that carries many TCP connections, each with its own id."""
+
+    def __init__(self, ws: Any) -> None:
+        self.ws = ws
+        self.lock = asyncio.Lock()
+        self.streams: dict[int, asyncio.Queue[Optional[bytes]]] = {}
+
+    async def send(self, sid: int, kind: int, data: bytes = b"") -> None:
+        async with self.lock:
+            await self.ws.send_bytes(frame(sid, kind, data))
+
+    async def pump(self, sid: int, reader: asyncio.StreamReader) -> None:
+        """Copy a socket into the WebSocket until either side closes."""
+        try:
+            while chunk := await reader.read(CHUNK):
+                await self.send(sid, DATA, chunk)
+        except (ConnectionError, RuntimeError):
+            pass
+        finally:
+            await self.close_stream(sid)  # the peer does not echo CLOSE, so this also ends our drain
+
+    async def drain(self, q: asyncio.Queue[Optional[bytes]], writer: asyncio.StreamWriter) -> None:
+        """Copy the bytes that the peer sends for one connection into a socket, then close the socket."""
+        try:
+            while (chunk := await q.get()) is not None:
+                writer.write(chunk)
+                await writer.drain()
+        except ConnectionError:
+            pass
+        finally:
+            writer.close()
+
+    def end(self, sid: int) -> bool:
+        """End one connection's drain; False when the connection was already gone."""
+        if (q := self.streams.pop(sid, None)) is None:
+            return False
+        q.put_nowait(None)
+        return True
+
+    async def close_stream(self, sid: int) -> None:
+        """End one connection here and tell the peer."""
+        if self.end(sid) and not self.ws.closed:
+            await self.send(sid, CLOSE)
+
+    async def feed(self, sid: int, body: bytes) -> None:
+        """Queue bytes for one connection; drop a connection whose socket has stopped reading them."""
+        q = self.streams.get(sid)
+        if q is None:
+            return
+        if q.qsize() < BACKLOG:
+            q.put_nowait(body)
+        else:
+            await self.close_stream(sid)
+
+    def end_all(self) -> None:
+        for sid in list(self.streams):
+            self.end(sid)
+
+    async def relay(
+        self, sid: int, q: asyncio.Queue[Optional[bytes]], reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+    ) -> None:
+        await asyncio.gather(self.pump(sid, reader), self.drain(q, writer))
+
+
+TUNNELS: dict[str, "Tunnel"] = {}
+
+
+class Tunnel:
+    """The server side of one client's tunnel: a loopback SOCKS5 listener and the client's current link."""
+
+    def __init__(self, owner: Optional[str]) -> None:
+        self.id = secrets.token_urlsafe(16)
+        self.secret = secrets.token_urlsafe(24)
+        self.owner = owner
+        self.link: Optional[Link] = None
+        self.pending: dict[int, asyncio.Future[bool]] = {}
+        self.sids = itertools.count(1)
+        self.listener: Optional[asyncio.Server] = None
+        self.expiry: Optional[asyncio.TimerHandle] = None
+        self.linked = asyncio.Event()
+        self.closing: Optional[asyncio.Task[None]] = None
+
+    @property
+    def port(self) -> int:
+        assert self.listener is not None
+        return int(self.listener.sockets[0].getsockname()[1])
+
+    @property
+    def proxy_uri(self) -> str:
+        # without credentials, another API key could use this port as a raw socks5h://127.0.0.1 proxy
+        return f"socks5h://{self.id}:{self.secret}@127.0.0.1:{self.port}"
+
+    async def start(self) -> None:
+        self.listener = await asyncio.start_server(self.accept, "127.0.0.1", 0)
+        TUNNELS[self.id] = self
+        log.info(f"Tunnel {self.id[:6]} opened on 127.0.0.1:{self.port}")
+
+    async def close(self) -> None:
+        if TUNNELS.pop(self.id, None):
+            log.info(f"Tunnel {self.id[:6]} closed")
+        if self.expiry:
+            self.expiry.cancel()
+        if self.listener:
+            self.listener.close()
+        await self.detach()
+
+    async def detach(self) -> None:
+        link, self.link = self.link, None
+        self.linked.clear()
+        for fut in self.pending.values():
+            fut.cancel()
+        self.pending.clear()
+        if link:
+            link.end_all()
+            await link.ws.close()
+
+    async def on_frame(self, link: Link, data: bytes) -> None:
+        """Act on a frame from the client. An OPEN from the client is ignored: the server never dials for it."""
+        if len(data) < HEADER.size:
+            return
+        sid, kind = HEADER.unpack_from(data)
+        if kind == OK and (fut := self.pending.pop(sid, None)) and not fut.done():
+            fut.set_result(True)
+        elif kind == DATA:
+            await link.feed(sid, data[HEADER.size :])
+        elif kind == CLOSE:
+            if (fut := self.pending.pop(sid, None)) and not fut.done():
+                fut.set_result(False)
+            link.end(sid)
+
+    async def wait_link(self) -> Optional[Link]:
+        """The current link, waiting ``RELINK_WAIT`` seconds for a dropped one to come back."""
+        if not self.link:
+            try:
+                await asyncio.wait_for(self.linked.wait(), RELINK_WAIT)
+            except asyncio.TimeoutError:
+                pass
+        return self.link
+
+    async def accept(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        """Take one SOCKS5 CONNECT with this tunnel's username and password, and open it on the client."""
+
+        def refuse(answer: bytes) -> None:
+            writer.write(answer)
+            writer.close()
+
+        def reply(code: int) -> bytes:
+            return bytes((5, code, 0, 1)) + bytes(6)
+
+        try:
+            version, methods = await reader.readexactly(2)
+            if version != 5 or 2 not in await reader.readexactly(methods):
+                return refuse(b"\x05\xff")
+            writer.write(b"\x05\x02")
+            auth_version = (await reader.readexactly(1))[0]
+            user = await reader.readexactly((await reader.readexactly(1))[0])
+            password = await reader.readexactly((await reader.readexactly(1))[0])
+            if not (
+                auth_version == 1
+                and secrets.compare_digest(user, self.id.encode())
+                and secrets.compare_digest(password, self.secret.encode())
+            ):
+                return refuse(b"\x01\x01")
+            writer.write(b"\x01\x00")
+            _, command, _, address_type = await reader.readexactly(4)
+            if address_type == 1:
+                host = socket.inet_ntop(socket.AF_INET, await reader.readexactly(4))
+            elif address_type == 4:
+                host = socket.inet_ntop(socket.AF_INET6, await reader.readexactly(16))
+            elif address_type == 3:
+                host = (await reader.readexactly((await reader.readexactly(1))[0])).decode()
+            else:
+                return refuse(reply(8))  # address type not supported
+            port = await reader.readexactly(2)
+        except (asyncio.IncompleteReadError, ConnectionError, UnicodeDecodeError):
+            writer.close()
+            return
+        if command != 1:
+            return refuse(reply(7))  # command not supported
+        link = await self.wait_link()
+        if not link:
+            return refuse(reply(5))
+        sid = next(self.sids)
+        fut = self.pending[sid] = asyncio.get_running_loop().create_future()
+        q = link.streams[sid] = new_queue()
+        ok = False
+        try:
+            await link.send(sid, OPEN, port + host.encode())
+            ok = await asyncio.wait_for(fut, OPEN_TIMEOUT)
+        except (asyncio.TimeoutError, asyncio.CancelledError, ConnectionError, RuntimeError):
+            pass
+        if not ok:
+            self.pending.pop(sid, None)
+            await link.close_stream(sid)  # a client still dialling would relay into a dead stream
+            return refuse(reply(5))  # connection refused
+        writer.write(reply(0))
+        await link.relay(sid, q, reader, writer)
+
+    async def serve(self, ws: web.WebSocketResponse) -> None:
+        """Make ``ws`` the tunnel's link until it closes, then wait ``GRACE`` seconds for a reconnect."""
+        if self.expiry:
+            self.expiry.cancel()
+            self.expiry = None
+        await self.detach()
+        link = self.link = Link(ws)
+        try:
+            await ws.send_str(json.dumps({"id": self.id}))
+            self.linked.set()
+            async for msg in ws:
+                if msg.type == WSMsgType.BINARY:
+                    await self.on_frame(link, msg.data)
+        finally:
+            if self.link is link:
+                self.link = None
+                self.linked.clear()
+                link.end_all()
+                if self.id in TUNNELS:
+                    self.expiry = asyncio.get_running_loop().call_later(GRACE, self.expire)
+
+    def expire(self) -> None:
+        self.closing = asyncio.ensure_future(self.close())  # held, so the loop cannot drop the task
+
+
+def tunnel_proxy(tunnel_id: str, owner: Optional[str]) -> Optional[str]:
+    """The loopback proxy URI for the tunnel ``tunnel_id`` that the caller owns, or None."""
+    tunnel = TUNNELS.get(tunnel_id)
+    return tunnel.proxy_uri if tunnel and tunnel.owner == owner else None
+
+
+async def close_tunnels(_: web.Application) -> None:
+    for tunnel in list(TUNNELS.values()):
+        await tunnel.close()
+
+
+def is_public_ip(address: str) -> bool:
+    try:
+        return ipaddress.ip_address(address.partition("%")[0]).is_global
+    except ValueError:
+        return False
+
+
+async def connect_public(host: str, port: int) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
+    """Connect to the first public address of ``host``. The server picks the targets, so a name that resolves
+    only to the client's own machine or LAN is refused."""
+    infos = await asyncio.get_running_loop().getaddrinfo(host, port, type=socket.SOCK_STREAM)
+    addresses = [info[4][0] for info in infos if is_public_ip(str(info[4][0]))]
+    if not addresses:
+        raise PermissionError(f"{host} resolves only to local addresses")
+    error: OSError = OSError(f"could not connect to {host}")
+    for address in addresses:
+        try:
+            return await asyncio.wait_for(asyncio.open_connection(address, port), 10)
+        except (OSError, asyncio.TimeoutError) as e:
+            error = OSError(str(e))
+    raise error
+
+
+class TunnelClient:
+    """The client side: a background thread that keeps the tunnel's WebSocket open and dials for the server."""
+
+    def __init__(self, server_url: str, headers: dict[str, str]) -> None:
+        self.url = server_url.rstrip("/") + ROUTE
+        self.headers = headers
+        self.id: Optional[str] = None
+        self.error: Optional[BaseException] = None
+        self.lost = False
+        self.ready = threading.Event()
+        self.relays: set[asyncio.Task[None]] = set()
+        self.thread = threading.Thread(target=lambda: asyncio.run(self.run()), name="tunnel", daemon=True)
+        self.loop: Optional[asyncio.AbstractEventLoop] = None
+        self.task: Optional[asyncio.Task[None]] = None
+
+    def start(self, timeout: float = 20) -> str:
+        """Open the tunnel and return its id."""
+        self.thread.start()
+        if not self.ready.wait(timeout) or not self.id:
+            self.stop()  # or the thread opens the tunnel later and nobody uses it
+            raise ConnectionError(f"Could not open a tunnel to the server: {self.error or 'timed out'}")
+        return self.id
+
+    def stop(self) -> None:
+        """Close the tunnel and wait for its connections to end, so none is left pending at exit."""
+        if self.loop and self.task and self.thread.is_alive():
+            self.loop.call_soon_threadsafe(self.task.cancel)
+            self.thread.join(5)
+
+    async def run(self) -> None:
+        self.loop, self.task = asyncio.get_running_loop(), asyncio.current_task()
+        try:
+            await self.connect()
+        except asyncio.CancelledError:
+            pass
+        finally:
+            for t in self.relays:
+                t.cancel()
+            await asyncio.gather(*self.relays, return_exceptions=True)
+
+    async def connect(self) -> None:
+        backoff = 1
+        async with aiohttp.ClientSession(headers=self.headers) as http:
+            while True:
+                try:
+                    url = f"{self.url}?id={self.id}" if self.id else self.url
+                    async with http.ws_connect(url, heartbeat=HEARTBEAT, max_msg_size=MAX_MESSAGE) as ws:
+                        hello = await ws.receive_json(timeout=15)
+                        self.id = hello["id"]
+                        self.ready.set()
+                        backoff = 1
+                        await self.serve(Link(ws))
+                except aiohttp.WSServerHandshakeError as e:
+                    self.error = e
+                    # a 404 means the tunnel expired or the server has no tunnel route; a 5xx is a proxy in between
+                    if not self.id or e.status < 500:
+                        if self.id:
+                            self.lost = True
+                            log.warning(f"The server closed the tunnel (HTTP {e.status}), so its requests now fail")
+                        self.ready.set()
+                        return
+                except (aiohttp.ClientError, asyncio.TimeoutError, ValueError, KeyError, TypeError) as e:
+                    self.error = e
+                    if not self.id:
+                        self.ready.set()
+                        return
+                log.warning(f"The tunnel to the server dropped, reconnecting in {backoff}s")
+                await asyncio.sleep(backoff)
+                backoff = min(backoff * 2, 30)
+
+    async def serve(self, link: Link) -> None:
+        try:
+            async for msg in link.ws:
+                if msg.type != WSMsgType.BINARY or len(msg.data) < HEADER.size:
+                    continue
+                sid, kind = HEADER.unpack_from(msg.data)
+                body = msg.data[HEADER.size :]
+                if kind == OPEN:
+                    if len(link.streams) >= MAX_STREAMS or sid in link.streams or len(body) < 3:
+                        await link.send(sid, CLOSE)
+                        continue
+                    link.streams[sid] = new_queue()
+                    host = body[2:].decode(errors="replace")
+                    task = asyncio.ensure_future(self.open(link, sid, host, int.from_bytes(body[:2], "big")))
+                    self.relays.add(task)
+                    task.add_done_callback(self.relays.discard)
+                elif kind == DATA:
+                    await link.feed(sid, body)
+                elif kind == CLOSE:
+                    link.end(sid)
+        finally:
+            link.end_all()
+
+    async def open(self, link: Link, sid: int, host: str, port: int) -> None:
+        try:
+            reader, writer = await connect_public(host, port)
+        except (OSError, ValueError, asyncio.TimeoutError) as e:  # the idna codec raises ValueError on a bad hostname
+            log.debug(f"Tunnel dial to {host}:{port} failed: {e}")
+            await link.close_stream(sid)
+            return
+        if (q := link.streams.get(sid)) is None:  # the server gave up while we dialled
+            writer.close()
+            return
+        await link.send(sid, OK)
+        await link.relay(sid, q, reader, writer)