feat(serve)!: gate the server's own proxies behind server_proxy

The server no longer spends its configured proxy providers on a remote client unless the API key sets server_proxy. It places the client by geolocating the connection address and asks for a proxy when that region does not match its own.

docs/dev/rest-api/endpoints.md

@@ -138,7 +138,7 @@ Find titles in a service by query string. The service must have a `search()` met
 | `service` | string | yes | - | Service tag. |
 | `query` | string | yes | - | Search query. |
 | `profile` | string | no | `null` | Credential/cookie profile to use. |
-| `proxy` | string | no | `null` | Proxy URI or country code. |
+| `proxy` | string | no | `null` | Full proxy URI, or a country code when the API key has `server_proxy`. |
 | `no_proxy` | boolean | no | `false` | Force-disable all proxy use. |
 
 === "Request"
@@ -186,7 +186,7 @@ Get the list of titles behind a title ID, for example episodes or a movie, witho
 | `service` | string | yes | Service tag. |
 | `title_id` | string | yes | Title identifier. |
 | `profile` | string | no | Credential/cookie profile. |
-| `proxy` | string | no | Proxy URI or country code. |
+| `proxy` | string | no | Full proxy URI, or a country code when the API key has `server_proxy`. |
 | `no_proxy` | boolean | no | Force-disable proxy. |
 | `cdm_type` | string | no | Preferred CDM type. |
 
@@ -414,7 +414,7 @@ Make a download job. It requires `service` and `title_id`. Every other field is
 | `skip_dl` | boolean | `false` | Only fetch keys, do not download. |
 | `export` | boolean | `false` | Export manifest/keys/subs to JSON. |
 | `cdm_only` | boolean | `null` | Force CDM-only (`true`) or vault-only (`false`) key retrieval. |
-| `proxy` / `no_proxy` / `no_proxy_download` | string / bool / bool | `null` / `false` / `false` | Proxy controls. |
+| `proxy` / `no_proxy` / `no_proxy_download` | string / bool / bool | `null` / `false` / `false` | Proxy controls. `proxy` must be a full proxy URI unless the API key has `server_proxy`. |
 | `no_folder` / `no_source` / `no_mux` | boolean | `false` | Output/mux controls. |
 | `workers` | int | `null` | Threads per track. |
 | `adaptive_workers` | boolean | `false` | Scale segment workers to measured CDN throughput, up to the `workers` cap. |
@@ -834,6 +834,8 @@ DELETE /api/session/{id}        → tear down, harvest updated cache
 
 Make a remote session for a service and title. Authentication runs asynchronously in the background. This call returns immediately with `status: "authenticating"`, and you then poll the prompt endpoint. The body accepts `service` and `title_id` (both required). It also accepts a broad set of optional keys, because the body allows `additionalProperties`. These are `credentials` (`{username, password, extra?}`), `cookies` (base64 of zlib-compressed Netscape cookie file), `proxy`, `no_proxy`, `profile`, `cache` (map of forwarded cache files), `client_region`, `cdm_type`, `range_`, `vcodec`, `quality`, `best_available`, and any service CLI options.
 
+The `proxy` value must be a full proxy URI, unless the operator gives your API key `server_proxy` in the `serve.users` config. Without it, the server does not resolve country codes with its own proxy providers. It rejects the request with `INVALID_PROXY` when no `proxy` is set and the reported `client_region` differs from the server's region. A request that reports no `client_region` is not blocked. Pass `proxy` with your own proxy, or `no_proxy` to accept the server's own connection. With `server_proxy`, the server resolves a country code and picks a proxy for your `client_region` itself.
+
 === "Response `200`"
 
     ```json

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

@@ -114,7 +114,14 @@ hood the client walks a remote session through its lifecycle.
     - Local **credentials** for the service/profile (`{username, password, extra?}`)
     - Local **cookies**, compressed and base64-encoded
     - A resolved **proxy**, or, if you did not set one, your detected
-      **client region** so the server can auto-proxy to match it
+      **client region**. A client that reports no region is not blocked. The
+      proxy must be a full proxy URI unless the operator gives your API key
+      `server_proxy`. Without `server_proxy` the server does not resolve country
+      codes with its own proxy providers, and it rejects the remote session when
+      it sits in a different region from the reported one, asking the client to
+      pass `--proxy` with its own proxy, or `--no-proxy` to accept the server's
+      own connection. With `server_proxy` the server picks a proxy that matches
+      the client region itself
     - Track-selection hints (`range_`, `vcodec`, `quality`, `best_available`) so
       the server fetches the right manifests
     - Your local per-service **cache files** (e.g. refreshed tokens)

docs/reference/configuration/services.md

@@ -142,6 +142,17 @@ client that asks anyway gets a `FORBIDDEN` error. Because a download job always
 an API key without `server_cdm` for that service also cannot submit or retry `/api/download`
 jobs. Keys that have no `users` entry, such as `api_secret`, keep server CDM access.
 
+`server_proxy` decides whether that API key may use the proxy providers the server has
+configured. There is no implicit access: only the boolean `true` on the API key's `users` entry
+grants it, and an API key without a `users` entry has no access either. Without the opt-in the
+server never spends its own proxy subscriptions on a remote client: not to resolve a country code
+in `proxy`, not to match the client's region, and not for a service geofence. This applies to
+remote sessions, search, title lists, and `/api/download` jobs alike. Such a client must send a
+full proxy URI, or `no_proxy` to accept the server's own connection. Set `server_proxy: true`
+to let that API key send a country code or a `provider:country` value, and to let the server
+pick a proxy for the client's region and for a service geofence. The client reports its own
+region; a client that reports none is not blocked.
+
 ```yaml
 serve:
   api_secret: change-me
@@ -159,6 +170,7 @@ serve:
     c9d0e1f2:
       services: [EXAMPLE1, EXAMPLE2]
       server_cdm: [EXAMPLE1]      # the server licenses only EXAMPLE1; EXAMPLE2 needs a local CDM
+      server_proxy: true          # this key may also use the server's proxy providers
 ```
 
 !!! note "`dl` keys inside `serve`"

tests/remote/unit/test_handler_proxy_policy.py

@@ -0,0 +1,290 @@
+"""The server spends its own proxy subscriptions only for a key that opts in."""
+
+from types import SimpleNamespace
+
+import pytest
+
+from unshackle.core.api import handlers
+from unshackle.core.api.errors import APIError, APIErrorCode
+from unshackle.core.utils import ip_info
+
+pytestmark = pytest.mark.unit
+
+USERS = {
+    "plainkey": {"username": "plain"},
+    "proxykey": {"username": "proxy", "server_proxy": True},
+    "strkey": {"username": "str", "server_proxy": "true"},
+}
+
+
+@pytest.fixture(autouse=True)
+def serve_users(monkeypatch):
+    monkeypatch.setattr(handlers.config, "serve", {"users": USERS}, raising=False)
+    monkeypatch.setattr(handlers, "request_secret_key", lambda request: request.headers.get("X-Secret-Key"))
+
+
+def request(key):
+    return SimpleNamespace(headers={"X-Secret-Key": key} if key else {})
+
+
+def patch_server_region(monkeypatch, region):
+    monkeypatch.setattr(ip_info, "get_ip_info", lambda session=None, cached=False: region and {"country": region})
+
+
+def patch_providers(monkeypatch, proxy):
+    """Give the server one provider that answers every country code with ``proxy``."""
+    provider = SimpleNamespace(get_proxy=lambda query: proxy)
+    monkeypatch.setattr(handlers, "initialize_proxy_providers", lambda: [provider])
+    return provider
+
+
+def test_region_mismatch_raises(monkeypatch):
+    patch_server_region(monkeypatch, "IN")
+    with pytest.raises(APIError) as exc_info:
+        handlers.resolve_handler_proxy({"client_region": "ca"}, "EXAMPLE", request("plainkey"))
+    assert exc_info.value.error_code == APIErrorCode.INVALID_PROXY
+    assert "IN" not in exc_info.value.message
+    # The details must not leak the server region either.
+    assert exc_info.value.details == {"service": "EXAMPLE"}
+
+
+def test_region_match_no_proxy(monkeypatch):
+    patch_server_region(monkeypatch, "CA")
+    assert handlers.resolve_handler_proxy({"client_region": "ca"}, "EXAMPLE", request("plainkey")) == (None, [])
+
+
+def test_region_unknown_no_error(monkeypatch):
+    patch_server_region(monkeypatch, None)
+    assert handlers.resolve_handler_proxy({"client_region": "ca"}, "EXAMPLE", request("plainkey")) == (None, [])
+
+
+def test_direct_uri_passes_through(monkeypatch):
+    patch_server_region(monkeypatch, "IN")
+    proxy, providers = handlers.resolve_handler_proxy(
+        {"proxy": "socks5://user:pass@host:1080"}, "EXAMPLE", request("plainkey")
+    )
+    assert proxy == "socks5://user:pass@host:1080"
+    assert providers == []
+
+
+def test_country_code_rejected(monkeypatch):
+    patch_server_region(monkeypatch, "IN")
+    with pytest.raises(APIError) as exc_info:
+        handlers.resolve_handler_proxy({"proxy": "us"}, "EXAMPLE", request("plainkey"))
+    assert exc_info.value.error_code == APIErrorCode.INVALID_PROXY
+    assert "us" not in exc_info.value.details.values()
+
+
+def test_provider_country_rejected(monkeypatch):
+    patch_server_region(monkeypatch, "IN")
+    with pytest.raises(APIError) as exc_info:
+        handlers.resolve_handler_proxy({"proxy": "nordvpn:us"}, "EXAMPLE", request("plainkey"))
+    assert exc_info.value.error_code == APIErrorCode.INVALID_PROXY
+    assert "nordvpn" not in exc_info.value.message.lower()
+    assert exc_info.value.details == {"service": "EXAMPLE"}
+
+
+def test_no_proxy_short_circuits(monkeypatch):
+    """no_proxy skips resolution and never loads providers, even with a proxy param set."""
+    monkeypatch.setattr(handlers, "initialize_proxy_providers", lambda: pytest.fail("providers loaded"))
+    _, providers = handlers.resolve_handler_proxy(
+        {"proxy": "us", "no_proxy": True, "client_region": "ca"}, "EXAMPLE", request("proxykey")
+    )
+    assert providers == []
+
+
+def test_server_proxy_allowed_entitlements():
+    assert handlers.server_proxy_allowed(request("proxykey")) is True
+    assert handlers.server_proxy_allowed(request("plainkey")) is False
+    # yaml string "true" is not the boolean opt-in
+    assert handlers.server_proxy_allowed(request("strkey")) is False
+    # no implicit admin access: a key absent from serve.users is denied
+    assert handlers.server_proxy_allowed(request("adminsecret")) is False
+    # no request (an internal call, or a handler that dropped it) fails closed
+    assert handlers.server_proxy_allowed(None) is False
+
+
+def test_opted_in_country_code_resolves(monkeypatch):
+    patch_server_region(monkeypatch, "IN")
+    provider = patch_providers(monkeypatch, "http://us.proxy:8080")
+    proxy, providers = handlers.resolve_handler_proxy({"proxy": "us"}, "EXAMPLE", request("proxykey"))
+    assert proxy == "http://us.proxy:8080"
+    assert providers == [provider]
+
+
+def test_opted_in_region_mismatch_auto_proxies(monkeypatch):
+    patch_server_region(monkeypatch, "IN")
+    patch_providers(monkeypatch, "http://ca.proxy:8080")
+    proxy, providers = handlers.resolve_handler_proxy({"client_region": "ca"}, "EXAMPLE", request("proxykey"))
+    assert proxy == "http://ca.proxy:8080"
+    assert providers
+
+
+def test_absent_key_denied(monkeypatch):
+    patch_server_region(monkeypatch, "IN")
+    patch_providers(monkeypatch, "http://ca.proxy:8080")
+    with pytest.raises(APIError) as exc_info:
+        handlers.resolve_handler_proxy({"client_region": "ca"}, "EXAMPLE", request("adminsecret"))
+    assert exc_info.value.error_code == APIErrorCode.INVALID_PROXY
+
+
+def test_user_without_server_proxy_denied(monkeypatch):
+    patch_server_region(monkeypatch, "IN")
+    patch_providers(monkeypatch, "http://ca.proxy:8080")
+    with pytest.raises(APIError) as exc_info:
+        handlers.resolve_handler_proxy({"client_region": "ca"}, "EXAMPLE", request("plainkey"))
+    assert exc_info.value.error_code == APIErrorCode.INVALID_PROXY
+
+
+def test_no_region_sent_proceeds(monkeypatch):
+    """An out-of-date client that reports no region is unknown and never blocked."""
+    patch_server_region(monkeypatch, "IN")
+    monkeypatch.setattr(handlers, "initialize_proxy_providers", lambda: [])
+    assert handlers.resolve_handler_proxy({}, "EXAMPLE", request("plainkey")) == (None, [])
+    assert handlers.resolve_handler_proxy({}, "EXAMPLE", request("proxykey")) == (None, [])
+
+
+@pytest.mark.parametrize("key", ["plainkey", "proxykey"])
+def test_non_string_client_region_rejected(monkeypatch, key):
+    patch_providers(monkeypatch, "http://ca.proxy:8080")
+    with pytest.raises(APIError) as exc_info:
+        handlers.resolve_handler_proxy({"client_region": 1}, "EXAMPLE", request(key))
+    assert exc_info.value.error_code == APIErrorCode.INVALID_INPUT
+
+
+def test_setup_list_service_threads_request(monkeypatch):
+    """A regression that stops passing the request through would re-open the proxy gate."""
+    seen = {}
+
+    def fake_resolve(data, service, request=None):
+        seen["request"] = request
+        return None, []
+
+    class FakeService:
+        def authenticate(self, cookies, credential):
+            seen["authenticated"] = True
+
+    monkeypatch.setattr(handlers, "load_service_yaml", lambda service: {})
+    monkeypatch.setattr(handlers, "resolve_handler_proxy", fake_resolve)
+    monkeypatch.setattr(handlers, "load_full_cdm", lambda *args: None)
+    monkeypatch.setattr(handlers, "build_parent_ctx", lambda *args: None)
+    monkeypatch.setattr(handlers.Services, "load", lambda service: None)
+    monkeypatch.setattr(handlers, "instantiate_service", lambda *args: FakeService())
+    monkeypatch.setattr("unshackle.commands.dl.dl.get_cookie_jar", staticmethod(lambda service, profile: None))
+    monkeypatch.setattr("unshackle.commands.dl.dl.get_credentials", staticmethod(lambda service, profile: None))
+
+    req = request("plainkey")
+    handlers.setup_list_service({}, "EXAMPLE", None, "some-title", req)
+    assert seen["request"] is req
+    assert seen["authenticated"]
+
+
+class TestDownloadGates:
+    """/api/download submissions and retries obey the same proxy policy as sessions."""
+
+    @pytest.fixture(autouse=True)
+    def cdm_gate_open(self, monkeypatch):
+        monkeypatch.setattr(handlers, "server_cdm_allowed", lambda request=None, service=None: True)
+
+    def test_country_code_rejected_when_denied(self, monkeypatch):
+        with pytest.raises(APIError) as exc_info:
+            handlers.enforce_download_gates({"service": "EXAMPLE", "proxy": "us"}, request("plainkey"))
+        assert exc_info.value.error_code == APIErrorCode.INVALID_PROXY
+
+    def test_provider_country_rejected_when_denied(self, monkeypatch):
+        with pytest.raises(APIError) as exc_info:
+            handlers.enforce_download_gates({"service": "EXAMPLE", "proxy": "nordvpn:us"}, request("plainkey"))
+        assert exc_info.value.error_code == APIErrorCode.INVALID_PROXY
+
+    def test_full_uri_passes_when_denied(self):
+        handlers.enforce_download_gates(
+            {"service": "EXAMPLE", "proxy": "socks5://user:pass@host:1080"}, request("plainkey")
+        )
+
+    def test_no_proxy_param_passes_when_denied(self):
+        handlers.enforce_download_gates({"service": "EXAMPLE"}, request("plainkey"))
+
+    def test_country_code_passes_when_opted_in(self, monkeypatch):
+        # The gate must not even try to resolve for an opted-in key: dl resolves at run time,
+        # exactly as before the gate existed.
+        monkeypatch.setattr(handlers, "initialize_proxy_providers", lambda: pytest.fail("providers loaded"))
+        handlers.enforce_download_gates({"service": "EXAMPLE", "proxy": "us"}, request("proxykey"))
+
+    def test_region_mismatch_rejected_at_submit(self, monkeypatch):
+        """A queued job reaches the same region verdict, at submission."""
+        patch_server_region(monkeypatch, "IN")
+        with pytest.raises(APIError) as exc_info:
+            handlers.enforce_download_gates({"service": "EXAMPLE", "client_region": "gb"}, request("plainkey"))
+        assert exc_info.value.error_code == APIErrorCode.INVALID_PROXY
+
+
+class TestDownloadEnforcement:
+    """The server_proxy stamp survives from submission to the dl construction."""
+
+    @pytest.fixture
+    def created_params(self, monkeypatch):
+        """Drive download_handler with the job machinery faked out; capture create_job params."""
+        import asyncio
+        from datetime import datetime, timezone
+
+        from unshackle.core.api import download_manager as dm
+
+        created = {}
+
+        class FakeManager:
+            async def start_workers(self):
+                pass
+
+            def create_job(self, service, title_id, owner_key=None, **params):
+                created.update(params)
+                return SimpleNamespace(
+                    job_id="job1", status=SimpleNamespace(value="queued"), created_time=datetime.now(timezone.utc)
+                )
+
+        monkeypatch.setattr(dm, "get_download_manager", lambda: FakeManager())
+        monkeypatch.setattr(handlers, "validate_service", lambda service, request=None: "EXAMPLE")
+        monkeypatch.setattr(handlers, "enforce_download_gates", lambda params, request=None: None)
+        monkeypatch.setattr(handlers.Services, "load", lambda service: SimpleNamespace())
+
+        def run(body, req):
+            response = asyncio.run(handlers.download_handler(body, req))
+            assert response.status == 202
+            return created
+
+        return run
+
+    def test_client_sent_stamp_is_overwritten(self, created_params):
+        created = created_params({"service": "EXAMPLE", "title_id": "t1", "server_proxy": True}, request("plainkey"))
+        assert created["server_proxy"] is False
+
+    def test_opted_in_key_is_stamped_true(self, created_params):
+        created = created_params({"service": "EXAMPLE", "title_id": "t1"}, request("proxykey"))
+        assert created["server_proxy"] is True
+
+    @pytest.mark.parametrize(("params", "expected"), [({}, []), ({"server_proxy": True}, None)])
+    def test_perform_download_maps_stamp_to_dl_providers(self, monkeypatch, tmp_path, params, expected):
+        """An unstamped or denied job hands dl an empty provider list; a stamped one loads config."""
+        import click
+
+        from unshackle.commands import dl as dl_module
+        from unshackle.core.api import download_manager as dm
+
+        seen = {}
+
+        class Stop(Exception):
+            pass
+
+        class FakeDl:
+            cli = click.Group("dl")
+
+            def __init__(self, **kwargs):
+                seen.update(kwargs)
+                raise Stop
+
+        monkeypatch.setattr(dl_module, "dl", FakeDl)
+        monkeypatch.setattr("unshackle.core.services.Services.get_path", staticmethod(lambda service: tmp_path))
+        monkeypatch.setattr(handlers, "load_full_cdm", lambda *args: None)
+
+        with pytest.raises(Stop):
+            dm.perform_download("job1", "EXAMPLE", "t1", dict(params))
+        assert seen["proxy_providers"] == expected

unshackle/commands/dl.py

@@ -973,6 +973,7 @@ class dl:
         enrich: bool = False,
         daily: bool = False,
         output_dir: Optional[Path] = None,
+        proxy_providers: Optional[list[Any]] = None,
         *_: Any,
         **__: Any,
     ):
@@ -1314,28 +1315,31 @@ class dl:
         if no_proxy:
             ctx.params["proxy"] = None
         else:
-            with console.status("Loading Proxy Providers...", spinner="dots"):
-                if config.proxy_providers.get("basic"):
-                    self.proxy_providers.append(Basic(**config.proxy_providers["basic"]))
-                # ExpressVPN/ProtonVPN auto-load when their default cookie file exists (no yaml needed)
-                expressvpn = ExpressVPN(**(config.proxy_providers.get("expressvpn") or {}))
-                if config.proxy_providers.get("expressvpn") or expressvpn.cache_path.is_file():
-                    self.proxy_providers.append(expressvpn)
-                if config.proxy_providers.get("nordvpn"):
-                    self.proxy_providers.append(NordVPN(**config.proxy_providers["nordvpn"]))
-                proton = ProtonVPN(**(config.proxy_providers.get("protonvpn") or {}))
-                if config.proxy_providers.get("protonvpn") or proton.cookie_path.is_file():
-                    self.proxy_providers.append(proton)
-                if config.proxy_providers.get("surfsharkvpn"):
-                    self.proxy_providers.append(SurfsharkVPN(**config.proxy_providers["surfsharkvpn"]))
-                if config.proxy_providers.get("windscribevpn"):
-                    self.proxy_providers.append(WindscribeVPN(**config.proxy_providers["windscribevpn"]))
-                if config.proxy_providers.get("gluetun"):
-                    self.proxy_providers.append(Gluetun(**config.proxy_providers["gluetun"]))
-                if binaries.HolaProxy:
-                    self.proxy_providers.append(Hola())
-                for proxy_provider in self.proxy_providers:
-                    self.log.info(f"Loaded {proxy_provider.__class__.__name__}: {proxy_provider}")
+            if proxy_providers is not None:
+                self.proxy_providers = list(proxy_providers)
+            else:
+                with console.status("Loading Proxy Providers...", spinner="dots"):
+                    if config.proxy_providers.get("basic"):
+                        self.proxy_providers.append(Basic(**config.proxy_providers["basic"]))
+                    # ExpressVPN/ProtonVPN auto-load when their default cookie file exists (no yaml needed)
+                    expressvpn = ExpressVPN(**(config.proxy_providers.get("expressvpn") or {}))
+                    if config.proxy_providers.get("expressvpn") or expressvpn.cache_path.is_file():
+                        self.proxy_providers.append(expressvpn)
+                    if config.proxy_providers.get("nordvpn"):
+                        self.proxy_providers.append(NordVPN(**config.proxy_providers["nordvpn"]))
+                    proton = ProtonVPN(**(config.proxy_providers.get("protonvpn") or {}))
+                    if config.proxy_providers.get("protonvpn") or proton.cookie_path.is_file():
+                        self.proxy_providers.append(proton)
+                    if config.proxy_providers.get("surfsharkvpn"):
+                        self.proxy_providers.append(SurfsharkVPN(**config.proxy_providers["surfsharkvpn"]))
+                    if config.proxy_providers.get("windscribevpn"):
+                        self.proxy_providers.append(WindscribeVPN(**config.proxy_providers["windscribevpn"]))
+                    if config.proxy_providers.get("gluetun"):
+                        self.proxy_providers.append(Gluetun(**config.proxy_providers["gluetun"]))
+                    if binaries.HolaProxy:
+                        self.proxy_providers.append(Hola())
+                    for proxy_provider in self.proxy_providers:
+                        self.log.info(f"Loaded {proxy_provider.__class__.__name__}: {proxy_provider}")
 
             if proxy:
                 requested_provider = None

unshackle/core/api/download_manager.py

@@ -472,6 +472,7 @@ def perform_download(
     dl_instance = dl(
         ctx=ctx,
         no_proxy=params.get("no_proxy", False),
+        proxy_providers=None if params.get("server_proxy") is True else [],
         profile=params.get("profile"),
         proxy=params.get("proxy"),
         repack=params.get("repack", False),

unshackle/core/api/handlers.py

@@ -221,7 +221,13 @@ def instantiate_service(
     return parent_ctx.invoke(service_module.cli, title=title, **extras)
 
 
-def setup_list_service(data: Dict[str, Any], normalized_service: str, profile: Optional[str], title_id: str) -> Any:
+def setup_list_service(
+    data: Dict[str, Any],
+    normalized_service: str,
+    profile: Optional[str],
+    title_id: str,
+    request: Optional[web.Request] = None,
+) -> Any:
     """Assemble and authenticate a service instance for list_titles / list_tracks.
 
     Runs the shared preamble: load yaml → get proxy → load CDM → assemble ctx →
@@ -231,22 +237,8 @@ def setup_list_service(data: Dict[str, Any], normalized_service: str, profile: O
 
     service_config = load_service_yaml(normalized_service)
 
-    proxy_param = data.get("proxy")
     no_proxy = data.get("no_proxy", False)
-    proxy_providers = []
-
-    if not no_proxy:
-        proxy_providers = initialize_proxy_providers()
-
-    if proxy_param and not no_proxy:
-        try:
-            proxy_param = resolve_proxy(proxy_param, proxy_providers)
-        except ValueError as e:
-            raise APIError(
-                APIErrorCode.INVALID_PROXY,
-                f"Proxy error: {e}",
-                details={"proxy": proxy_param, "service": normalized_service},
-            )
+    proxy_param, proxy_providers = resolve_handler_proxy(data, normalized_service, request)
 
     cdm = load_full_cdm(normalized_service, profile, data.get("cdm_type"))
     parent_ctx = build_parent_ctx(profile, cdm, proxy_param, no_proxy, proxy_providers, service_config)
@@ -314,6 +306,24 @@ def server_cdm_allowed(request: Optional[web.Request] = None, service: Optional[
     return bool(allowed)
 
 
+def server_proxy_allowed(request: Optional[web.Request] = None) -> bool:
+    """Whether the calling API key may use the server's own configured proxy providers.
+
+    Configured API keys opt in with ``server_proxy: true``, a real yaml boolean, so a
+    quoted ``"true"`` or ``"no"`` does not count. It is a plain bool, not a per-service
+    list, because a proxy is not tied to a service. No API key has implicit access: an
+    API key absent from ``serve.users`` gets none, and neither does a call without a
+    request, so a handler that does not pass ``request`` through fails closed.
+    """
+    if not request:
+        return False
+    secret_key = request_secret_key(request)
+    user_config = (config.serve or {}).get("users", {}).get(secret_key)
+    if not isinstance(user_config, dict):
+        return False
+    return user_config.get("server_proxy") is True
+
+
 JOB_EVENTS_ROUTE = "/api/download/jobs/{job_id}/events"
 
 CORS_HEADERS = {
@@ -785,25 +795,11 @@ async def search_handler(data: Dict[str, Any], request: Optional[web.Request] =
         )
 
     profile = data.get("profile")
-    proxy_param = data.get("proxy")
     no_proxy = data.get("no_proxy", False)
 
     service_config = load_service_yaml(normalized_service)
 
-    proxy_providers = []
-    if not no_proxy:
-        proxy_providers = initialize_proxy_providers()
-
-    if proxy_param and not no_proxy:
-        try:
-            resolved_proxy = resolve_proxy(proxy_param, proxy_providers)
-            proxy_param = resolved_proxy
-        except ValueError as e:
-            raise APIError(
-                APIErrorCode.INVALID_PROXY,
-                f"Proxy error: {e}",
-                details={"proxy": proxy_param, "service": normalized_service},
-            )
+    proxy_param, proxy_providers = await asyncio.to_thread(resolve_handler_proxy, data, normalized_service, request)
 
     cdm = load_full_cdm(normalized_service, profile, data.get("cdm_type"))
     parent_ctx = build_parent_ctx(profile, cdm, proxy_param, no_proxy, proxy_providers, service_config)
@@ -860,7 +856,9 @@ async def list_titles_handler(data: Dict[str, Any], request: Optional[web.Reques
         )
 
     try:
-        service_instance = setup_list_service(data, normalized_service, profile, title_id)
+        service_instance = await asyncio.to_thread(
+            setup_list_service, data, normalized_service, profile, title_id, request
+        )
         titles = service_instance.get_titles()
 
         if hasattr(titles, "__iter__") and not isinstance(titles, str):
@@ -898,7 +896,9 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
         )
 
     try:
-        service_instance = setup_list_service(data, normalized_service, profile, title_id)
+        service_instance = await asyncio.to_thread(
+            setup_list_service, data, normalized_service, profile, title_id, request
+        )
         titles = service_instance.get_titles()
 
         wanted_param = data.get("wanted")
@@ -1198,7 +1198,10 @@ def enforce_download_gates(params: Dict[str, Any], request: Optional[web.Request
     secrets instead of the server-side credentials. Gate it behind `serve.allow_job_credentials`
     (default off) so a default deployment stays locked to its own credentials. This mirrors the
     CDM gate. A download job licenses DRM in-process with the server's own CDM, so an API key without
-    ``server_cdm`` cannot submit or retry jobs.
+    ``server_cdm`` cannot submit or retry jobs. A job can also spend the server's proxy providers,
+    through an explicit ``proxy`` country code or through the geofence auto-proxy, so the
+    ``server_proxy`` policy from :func:`resolve_handler_proxy` applies here too: without the opt-in
+    the job must carry a full proxy URI or no proxy at all.
     """
     if not server_cdm_allowed(request, params.get("service")):
         raise APIError(
@@ -1206,6 +1209,9 @@ def enforce_download_gates(params: Dict[str, Any], request: Optional[web.Request
             "Download jobs license with the server CDM, which is not enabled for this key on this service.",
         )
 
+    if not server_proxy_allowed(request):
+        resolve_handler_proxy(params, params.get("service") or "", request)
+
     requested_cdm = params.get("cdm")
     if requested_cdm:
         allowed = (config.serve or {}).get("cdm_overrides")
@@ -1249,7 +1255,7 @@ async def download_handler(data: Dict[str, Any], request: Optional[web.Request]
             details={"service": normalized_service, "title_id": title_id},
         )
 
-    enforce_download_gates(data, request)
+    await asyncio.to_thread(enforce_download_gates, data, request)
 
     try:
         service_module = Services.load(normalized_service)
@@ -1278,6 +1284,9 @@ async def download_handler(data: Dict[str, Any], request: Optional[web.Request]
             **service_specific_defaults,
             **filtered_params,
         }
+        # The download worker reads this to decide whether dl may load the server's proxy
+        # providers. Always stamped server-side so a client-sent value cannot grant it.
+        params_with_defaults["server_proxy"] = server_proxy_allowed(request)
         job = manager.create_job(normalized_service, title_id, owner_key=caller_key(request), **params_with_defaults)
 
         return web.json_response(
@@ -1553,12 +1562,17 @@ async def retry_download_job_handler(job_id: str, request: Optional[web.Request]
                 f"Invalid or unavailable service: {job.service}",
                 details={"service": job.service},
             )
-        enforce_download_gates({**job.parameters, "service": job.service}, request)
+        await asyncio.to_thread(enforce_download_gates, {**job.parameters, "service": job.service}, request)
 
         await manager.start_workers()
 
         # Reuse the raw in-memory parameters; redaction only ever applies to serialized copies.
-        new_job = manager.create_job(job.service, job.title_id, owner_key=caller_key(request), **job.parameters)
+        new_job = manager.create_job(
+            job.service,
+            job.title_id,
+            owner_key=caller_key(request),
+            **{**job.parameters, "server_proxy": server_proxy_allowed(request)},
+        )
 
         return web.json_response(
             {
@@ -2020,7 +2034,7 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req
         )
 
     try:
-        proxy_param, proxy_providers = resolve_handler_proxy(data, normalized_service)
+        proxy_param, proxy_providers = await asyncio.to_thread(resolve_handler_proxy, data, normalized_service, request)
 
         import hashlib
         import uuid as uuid_mod
@@ -2520,31 +2534,55 @@ async def get_validated_session(session_id: str, request: Optional[web.Request])
     return session
 
 
-def resolve_handler_proxy(data: Dict[str, Any], normalized_service: str) -> tuple[Optional[str], list]:
-    """Get the proxy and initialise the proxy providers from API request data.
+def resolve_handler_proxy(
+    data: Dict[str, Any], normalized_service: str, request: Optional[web.Request] = None
+) -> tuple[Optional[str], list]:
+    """Get the proxy and the proxy providers for an API request.
 
-    Handles explicit proxy param, the `provider:country` format, and
-    client_region-based auto-proxy when server region differs.
+    ``server_proxy`` on the calling API key decides whether the server spends its own proxy
+    subscriptions on this client. Without it the proxy-provider list stays empty, so the server
+    rejects a country code in ``proxy``, and a client in another region must send a full proxy
+    URI or ``no_proxy``. The returned list reaches ``ContextData.proxy_providers``, so the same gate
+    also governs the geofence auto-proxy in ``Service.__init__``.
+
+    The client region is the client-reported ``client_region`` field: an identifier, not a
+    security boundary, because the spend gate is ``server_proxy`` on the API key. A client that
+    reports no region, and an unknown region on either side, never blocks the request.
     """
     proxy_param = data.get("proxy")
     no_proxy = data.get("no_proxy", False)
     proxy_providers: list = []
+    allowed = server_proxy_allowed(request)
 
-    if not no_proxy:
+    if allowed and not no_proxy:
         proxy_providers = initialize_proxy_providers()
 
     if proxy_param and not no_proxy:
         try:
             proxy_param = resolve_proxy(proxy_param, proxy_providers)
         except ValueError as e:
+            if allowed:
+                raise APIError(
+                    APIErrorCode.INVALID_PROXY,
+                    f"Proxy error: {e}",
+                    details={"proxy": data.get("proxy"), "service": normalized_service},
+                )
             raise APIError(
                 APIErrorCode.INVALID_PROXY,
-                f"Proxy error: {e}",
-                details={"proxy": data.get("proxy"), "service": normalized_service},
+                "This server does not resolve country codes into proxies. "
+                "Pass --proxy with a full proxy URI (http://, https:// or socks5://).",
+                details={"service": normalized_service},
             )
 
     client_region = data.get("client_region")
-    if not proxy_param and not no_proxy and client_region and proxy_providers:
+    if client_region is not None and not isinstance(client_region, str):
+        raise APIError(
+            APIErrorCode.INVALID_INPUT,
+            "client_region must be a string country code.",
+            details={"service": normalized_service},
+        )
+
+    if not proxy_param and not no_proxy and client_region and (proxy_providers or not allowed):
         try:
             from unshackle.core.utils.ip_info import get_ip_info
 
@@ -2554,7 +2592,16 @@ def resolve_handler_proxy(data: Dict[str, Any], normalized_service: str) -> tupl
             log.debug(f"Server region lookup failed: {e!r}")
             server_region = None
 
-        if server_region and server_region == client_region.lower():
+        in_client_region = bool(server_region) and server_region == client_region.lower()
+        if not allowed:
+            if server_region and not in_client_region:
+                raise APIError(
+                    APIErrorCode.INVALID_PROXY,
+                    "This server is in a different region from yours. Pass --proxy with a proxy "
+                    "in the region you want, or pass --no-proxy to use the server's own connection.",
+                    details={"service": normalized_service},
+                )
+        elif in_client_region:
             log.info(f"Server already in client region '{sanitize_log(client_region)}', no proxy needed")
         else:
             try:

unshackle/core/api/routes.py

@@ -368,7 +368,7 @@ async def search(request: web.Request) -> web.Response:
                 description: Profile to use for credentials and cookies (default - None)
               proxy:
                 type: string
-                description: Proxy URI or country code (default - None)
+                description: Full proxy URI, or a country code when the API key has server_proxy (default - None)
               no_proxy:
                 type: boolean
                 description: Force disable all proxy use (default - false)
@@ -773,7 +773,7 @@ async def download(request: web.Request) -> web.Response:
                 description: CDM device name on the server to license with, overriding the cdm config mapping. Requires serve.cdm_overrides to allow it (default - None)
               proxy:
                 type: string
-                description: Proxy URI or country code (default - None)
+                description: Full proxy URI, or a country code when the API key has server_proxy (default - None)
               no_proxy:
                 type: boolean
                 description: Force disable all proxy use (default - false)

unshackle/core/service.py

@@ -284,6 +284,11 @@ class Service(metaclass=ABCMeta):
                                     if proxy:
                                         self.log.info(f"Got Proxy from {proxy_provider.__class__.__name__}")
                                         break
+                                if not proxy:
+                                    self.log.warning(
+                                        f"No proxy available for {requested_proxy}. "
+                                        f"Pass --proxy with a proxy in {requested_proxy}, or the request can fail."
+                                    )
                         except Exception as e:
                             self.log.warning(f"Failed to check geofence: {e}")
                             current_region = None