diff --git a/docs/dev/rest-api/remote-sessions.md b/docs/dev/rest-api/remote-sessions.md index 7d10c8f7..b03a4d0c 100644 --- a/docs/dev/rest-api/remote-sessions.md +++ b/docs/dev/rest-api/remote-sessions.md @@ -146,10 +146,14 @@ hood the client walks a remote session through its lifecycle. `controld://@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:`. 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:` 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:` 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:` 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 | diff --git a/docs/guide/cli-reference.md b/docs/guide/cli-reference.md index 5d986bfd..a2d0e9a1 100644 --- a/docs/guide/cli-reference.md +++ b/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. diff --git a/docs/guide/proxies-and-vpn.md b/docs/guide/proxies-and-vpn.md index ba42f418..3a615aca 100644 --- a/docs/guide/proxies-and-vpn.md +++ b/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 ` 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 diff --git a/tests/remote/unit/test_remote_search.py b/tests/remote/unit/test_remote_search.py index a76d4be2..accc7910 100644 --- a/tests/remote/unit/test_remote_search.py +++ b/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()) diff --git a/tests/remote/unit/test_tunnel.py b/tests/remote/unit/test_tunnel.py new file mode 100644 index 00000000..4864d573 --- /dev/null +++ b/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() diff --git a/unshackle/commands/dl.py b/unshackle/commands/dl.py index 1037e9eb..bd70f4ef 100644 --- a/unshackle/commands/dl.py +++ b/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 diff --git a/unshackle/commands/search.py b/unshackle/commands/search.py index 58052a2e..80bf0fe9 100644 --- a/unshackle/commands/search.py +++ b/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) diff --git a/unshackle/commands/tunnel_test.py b/unshackle/commands/tunnel_test.py new file mode 100644 index 00000000..23fb1a63 --- /dev/null +++ b/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.") diff --git a/unshackle/core/api/handlers.py b/unshackle/core/api/handlers.py index 612cfbf3..a0d51a5b 100644 --- a/unshackle/core/api/handlers.py +++ b/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}) diff --git a/unshackle/core/api/routes.py b/unshackle/core/api/routes.py index 57f9934e..0f83d616 100644 --- a/unshackle/core/api/routes.py +++ b/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://@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://@dns.controld.com, or a country code when the API key has server_proxy, or tunnel: 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: 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: through it, and the client makes them from its own connection. + The first message is {"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: diff --git a/unshackle/core/proxies/resolve.py b/unshackle/core/proxies/resolve.py index cf43e48f..43c71538 100644 --- a/unshackle/core/proxies/resolve.py +++ b/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. diff --git a/unshackle/core/remote_service.py b/unshackle/core/remote_service.py index 5c771110..ef977f86 100644 --- a/unshackle/core/remote_service.py +++ b/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:`` 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. diff --git a/unshackle/core/tunnel.py b/unshackle/core/tunnel.py new file mode 100644 index 00000000..098bb376 --- /dev/null +++ b/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)