fix(proxy): keep service traffic on --proxy and warn about requests that bypass it

--proxy now overrides HTTP(S)_PROXY and the Windows system proxy while keeping the env CA bundle. Core fallback sessions use the run's proxy, and a warn-only audit hook names service code that sends a request or opens a raw socket outside the proxy; a session with proxy_exempt = True opts out.

docs/dev/creating-a-service.md

@@ -139,7 +139,7 @@ will use throughout the service:
 |---|---|
 | `self.config` | This service's `config.yaml` contents (a dict), or `None` if absent. |
 | `self.log` | A `logging.Logger` named after your class. |
-| `self.session` | A prepared `requests.Session` with config headers and a retry adapter (5 retries, backoff 0.2, retries on 429/500/502/503/504). |
+| `self.session` | A prepared `requests.Session` with config headers and a retry adapter (5 retries, backoff 0.2, retries on 429/500/502/503/504). Send every request through it: with a proxy active, a request from service code that does not go through the proxy logs a warning. A separate HTTP session that goes direct on purpose sets `proxy_exempt = True`. |
 | `self.cache` | A `Cacher` for arbitrary key/value data, for example authentication tokens. |
 | `self.title_cache` | A region/account-aware `TitleCacher` used by `get_titles_cached`. |
 | `self.cache_dir` | `config.directories.cache / <ClassName>`. |

docs/dev/service-architecture.md

@@ -114,7 +114,7 @@ up the objects your methods rely on:
 |---|---|
 | `self.config` | The per-service config slice (see [Request and HTTP session objects](#request-and-http-session-objects)). |
 | `self.log` | A `logging.Logger` named after your class. |
-| `self.session` | A ready-to-use HTTP session (headers, retries, and proxy applied). |
+| `self.session` | A ready-to-use HTTP session (headers, retries, and proxy applied). Send every request through it: with a proxy active, a request from service code that does not go through the proxy logs a warning. A separate HTTP session that goes direct on purpose sets `proxy_exempt = True`. |
 | `self.cache` | A generic key/value `Cacher` namespaced to the service. |
 | `self.title_cache` | A region- and account-aware `TitleCacher`. |
 | `self.cache_dir` | `config.directories.cache / <ServiceName>`. |

docs/guide/proxies-and-vpn.md

@@ -176,6 +176,32 @@ because geolocation data can be out of date.
 
 Over the REST API, the server returns this failure to the client as `INVALID_PROXY`.
 
+### System proxy and environment variables
+
+`--proxy` takes priority over the `HTTP_PROXY` and `HTTPS_PROXY` environment variables
+and over the Windows system proxy. unshackle still reads a `REQUESTS_CA_BUNDLE` or
+`CURL_CA_BUNDLE` certificate bundle, so a local debugging proxy with its own certificate
+authority continues to work. Without `--proxy`, unshackle uses the system proxy.
+
+### Requests outside the proxy
+
+With a proxy active, unshackle logs a warning when service code sends a request that does
+not go through the proxy:
+
+```text
+A request to api.example.com did not go through the proxy, so api.example.com sees your own IP (EXAMPLE/__init__.py:120)
+```
+
+The host in the message sees your own IP address, and a service can treat that as proxy
+detection. The request still goes: unshackle only warns. Report the message, with the
+file and line, to the service maintainer.
+
+A request to your own machine or to your local network gives no warning. This includes a
+private IP address such as `192.168.1.50`, and a `.local`, `.lan` or `.home.arpa` host name.
+
+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.
+
 ## Basic (static proxies)
 
 The `Basic` proxy provider is pure static configuration, with no accounts and no network
@@ -868,6 +894,19 @@ The returned proxy is HTTPS on **port 443**.
     resolver for the server with `get_remote_proxy()`, and it removes Control D from a bare
     query, because a loopback forwarder is of no use to a server on another machine.
 
+!!! note "Keeping service requests on the proxy (developers)"
+    `unshackle/core/utils/run_proxy.py` holds the proxy rules for service traffic:
+
+    - `pin(session)` stops environment and system proxies from overriding the proxy of a
+      requests HTTP session, and keeps the environment CA bundle. Call it on each requests
+      HTTP session that gets a proxy.
+    - `fallback_session()` gives a core helper with no HTTP session (a manifest parser, a DRM
+      class) a session on the proxy of the service that called it.
+    - An audit hook on `socket.getaddrinfo` and on `socket.connect` to a host name, and a
+      check in `RnetSession.request`, warn about requests from service code that do not use
+      the proxy. A service sets
+      `proxy_exempt = True` on an HTTP session that goes direct on purpose.
+
 !!! note "Implementing a new provider (developers)"
     Every proxy provider subclasses `unshackle.core.proxies.proxy.Proxy` and has three
     methods:

tests/core/test_run_proxy.py

@@ -0,0 +1,221 @@
+"""This file stands in for service code: its directory is the watched service directory."""
+
+import logging
+import os
+import sys
+
+import pytest
+import requests
+
+from unshackle.core.session import RnetSession
+from unshackle.core.utils import run_proxy
+
+PROXY = "https://u:p@proxy.example:89"
+
+
+class Svc:
+    def __init__(self, proxy=None):
+        self.session = requests.Session()
+        if proxy:
+            self.session.proxies["all"] = proxy
+        self.log = logging.getLogger("Svc")
+
+    def lookup(self, host: str) -> None:
+        lookup(host)
+
+    def fallback(self) -> requests.Session:
+        return run_proxy.fallback_session()
+
+
+@pytest.fixture(autouse=True)
+def watching(monkeypatch):
+    monkeypatch.setattr(run_proxy, "watched", {})
+    monkeypatch.setattr(run_proxy, "proxy_hosts", set())
+    monkeypatch.setattr(run_proxy, "warned", set())
+    run_proxy.watch(Svc, PROXY)
+
+
+def lookup(host: str) -> None:
+    sys.audit("socket.getaddrinfo", host, 443, 0, 0, 0)
+
+
+def run_as(file: str, source: str, out: dict | None = None, **names) -> None:
+    """Run ``source`` in a frame whose file name is ``file``; ``out`` receives the names it defines."""
+    namespace = {"sys": sys, **names}
+    exec(compile(source, file, "exec"), namespace)  # nosec B102 - sets the frame file name
+    if out is not None:
+        out.update(namespace)
+
+
+def leaks(caplog) -> list[str]:
+    return [r.getMessage() for r in caplog.records if "did not go through the proxy" in r.getMessage()]
+
+
+def test_direct_lookup_from_service_code_warns_once(caplog):
+    with caplog.at_level(logging.WARNING):
+        lookup("api.example")
+        lookup("api.example")
+    (message,) = leaks(caplog)
+    assert "A request to api.example did not go through the proxy" in message
+    assert f"core{os.sep}test_run_proxy.py:" in message
+
+
+@pytest.mark.parametrize(
+    "host",
+    [
+        "proxy.example",
+        "localhost",
+        "localhost.",
+        "localhost.localdomain",
+        "127.0.0.1",
+        "0.0.0.0",  # nosec B104 - a host name to classify, not a bind address
+        "::ffff:127.0.0.1",
+        "10.0.0.5",
+        "172.17.0.2",
+        "192.168.1.50",
+        "fe80::1",
+        "nas.local",
+        "node.lan",
+    ],
+)
+def test_proxy_and_local_hosts_do_not_warn(caplog, host):
+    with caplog.at_level(logging.WARNING):
+        lookup(host)
+    assert leaks(caplog) == []
+
+
+@pytest.mark.parametrize("host", ["8.8.8.8", "2606:4700:4700::1111", "api.example"])
+def test_public_hosts_warn(caplog, host):
+    with caplog.at_level(logging.WARNING):
+        lookup(host)
+    assert len(leaks(caplog)) == 1
+
+
+@pytest.mark.parametrize(("address", "warns"), [(("ws.example", 443), True), (("203.0.113.9", 443), False)])
+def test_raw_socket_connect_to_a_host_name_warns(caplog, address, warns):
+    """A connect to a host name looks it up in C, with no getaddrinfo event; an IP address had its lookup."""
+    with caplog.at_level(logging.WARNING):
+        sys.audit("socket.connect", object(), address)
+    assert bool(leaks(caplog)) is warns
+
+
+def test_core_frame_ends_the_walk(caplog):
+    with caplog.at_level(logging.WARNING):
+        source = "sys.audit('socket.getaddrinfo', 'vault.example', 443, 0, 0, 0)"
+        run_as(os.path.join(run_proxy.CORE_DIR, "core", "vaults.py"), source)
+    assert leaks(caplog) == []
+
+
+def test_socks_lookup_does_not_warn(caplog):
+    """A socks5:// proxy looks up the target host locally, but the request still goes through the proxy."""
+    with caplog.at_level(logging.WARNING):
+        run_as("/site-packages/socks.py", "sys.audit('socket.getaddrinfo', 'api.example', 443, 0, 0, 0)")
+    assert leaks(caplog) == []
+
+
+@pytest.mark.parametrize("exempt", [False, True])
+def test_an_exempt_http_session_does_not_warn(caplog, exempt):
+    """The requests frame between the lookup and the service code holds the HTTP session that sends it."""
+    session = requests.Session()
+    if exempt:
+        session.proxy_exempt = True  # type: ignore[attr-defined]
+    found: dict = {}
+    source = "def send(self, host):\n    sys.audit('socket.getaddrinfo', host, 443, 0, 0, 0)\n"
+    run_as("/site-packages/requests/sessions.py", source, out=found)
+    with caplog.at_level(logging.WARNING):
+        found["send"](session, "api.example")
+    assert bool(leaks(caplog)) is not exempt
+
+
+def test_service_instance_uses_the_proxy_of_its_own_run(caplog):
+    with caplog.at_level(logging.WARNING):
+        Svc().lookup("direct.example")  # a run with no proxy
+        first, second = Svc(PROXY), Svc(PROXY)
+        first.lookup("leak.example")
+        first.lookup("leak.example")
+        second.lookup("leak.example")  # a second run warns again
+    assert leaks(caplog) == [leaks(caplog)[0]] * 2
+    assert "leak.example" in leaks(caplog)[0]
+    assert {r.name for r in caplog.records if r.getMessage() in leaks(caplog)} == {"Svc"}
+
+
+def test_fallback_session_takes_the_proxy_of_the_calling_run():
+    session = Svc("http://own.example:1").fallback()
+    assert session.proxies == {"all": "http://own.example:1"}
+    assert session.trust_env is False
+    assert run_proxy.fallback_session().proxies == {"all": PROXY}  # this module, with one watched run
+
+
+def test_fallback_session_without_a_service_frame(monkeypatch):
+    def from_core():
+        source = "result.append(run_proxy.fallback_session())"
+        found: list = []
+        run_as(os.path.join(run_proxy.CORE_DIR, "core", "x.py"), source, run_proxy=run_proxy, result=found)
+        return found[0]
+
+    monkeypatch.setattr(Svc, "__module__", "elsewhere")  # no frame in this file is service code now
+    run_proxy.watched[next(iter(run_proxy.watched))].directory = "/nowhere/"
+    assert from_core().proxies == {"all": PROXY}  # one watched run, as in a CLI download thread
+    run_proxy.watch(Svc, "http://second.example:1")
+    run_proxy.watched["/nowhere/"] = run_proxy.watched.popitem()[1]
+    run_proxy.watched["/nowhere/"].proxies.append("http://second.example:1")
+    assert from_core().proxies == {}  # two runs: no way to tell which one
+
+
+def test_pin_keeps_the_env_ca_bundle(monkeypatch):
+    monkeypatch.setenv("REQUESTS_CA_BUNDLE", "/certs/mitm.pem")
+    session = requests.Session()
+    run_proxy.pin(session)
+    assert (session.trust_env, session.verify) == (False, "/certs/mitm.pem")
+    session = requests.Session()
+    session.verify = False
+    run_proxy.pin(session)
+    assert session.verify is False
+
+
+class Sent(Exception):
+    pass
+
+
+class NoNetwork:
+    def __getattr__(self, name):
+        raise Sent
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "exempt", "warns"), [({}, False, True), ({"proxy": PROXY}, False, False), ({}, True, False)]
+)
+def test_rnet_session_without_proxy_warns(caplog, kwargs, exempt, warns):
+    session = RnetSession(**kwargs)
+    session.proxy_exempt = exempt
+    assert dict(session.proxies) == ({"all": PROXY} if kwargs else {})
+    session.ensure_client = NoNetwork  # type: ignore[method-assign]
+    with caplog.at_level(logging.WARNING), pytest.raises(Sent):
+        session.get("https://api.example/x", max_retries=0)
+    assert bool(leaks(caplog)) is warns
+
+
+def test_a_failing_check_never_fails_the_request(monkeypatch):
+    def broken(*args):
+        raise RuntimeError("dictionary changed size during iteration")
+
+    monkeypatch.setattr(run_proxy, "check", broken)
+    session = RnetSession()
+    session.ensure_client = NoNetwork  # type: ignore[method-assign]
+    with pytest.raises(Sent):
+        session.get("https://api.example/x", max_retries=0)
+
+
+def test_core_parser_without_a_session_uses_the_run_proxy(monkeypatch):
+    from unshackle.core.manifests import DASH
+
+    seen: dict = {}
+
+    def get(self, url, **kwargs):
+        seen.update(self.proxies, trust_env=self.trust_env)
+        raise Sent
+
+    monkeypatch.setattr(requests.Session, "get", get)
+    with pytest.raises(Sent):
+        DASH.from_url("https://cdn.example/manifest.mpd")
+    assert seen == {"all": PROXY, "trust_env": False}

tests/core/test_service_proxy_env.py

@@ -0,0 +1,39 @@
+from types import SimpleNamespace
+
+import requests
+from requests.utils import select_proxy
+
+from unshackle.core import service as service_mod
+from unshackle.core.service import Service
+from unshackle.core.utils import run_proxy
+
+PROXY = "https://u:p@chosen.example:89"
+URL = "https://api.example.com/x"
+
+
+class Dummy(Service):
+    def get_titles(self):  # type: ignore[override]
+        raise NotImplementedError
+
+    def get_tracks(self, title):  # type: ignore[override]
+        raise NotImplementedError
+
+    def get_chapters(self, title):  # type: ignore[override]
+        raise NotImplementedError
+
+
+def test_chosen_proxy_beats_env_proxy_and_is_watched(monkeypatch):
+    monkeypatch.setenv("HTTPS_PROXY", "http://system.example:8080")
+    monkeypatch.setattr(service_mod, "verify_proxy_exit", lambda session: {"country": "fr"})
+    monkeypatch.setattr(run_proxy, "watched", {})
+    monkeypatch.setattr(run_proxy, "proxy_hosts", set())
+    ctx = SimpleNamespace(
+        obj=SimpleNamespace(config={}, proxy_providers=[]),
+        parent=SimpleNamespace(params={"proxy": PROXY}),
+    )
+    session = Dummy(ctx).session  # type: ignore[arg-type]
+    assert isinstance(session, requests.Session)
+    settings = session.merge_environment_settings(URL, {}, None, None, None)
+    assert select_proxy(URL, settings["proxies"]) == PROXY
+    (watch,) = run_proxy.watched.values()
+    assert (watch.cls, watch.proxies) == (Dummy, [PROXY])

tests/downloaders/test_session_spec.py

@@ -191,6 +191,19 @@ def test_requests_spec_carries_verify() -> None:
     assert rebuilt.verify is False
 
 
+def test_requests_spec_carries_trust_env() -> None:
+    """A proxied session ignores env proxies, so a child must not let HTTPS_PROXY win again."""
+    rs = Session()
+    rs.trust_env = False
+
+    spec = build_session_spec(rs)
+    assert spec is not None
+    rebuilt = rebuild_session(pickle.loads(pickle.dumps(spec)), 1)
+    assert rebuilt is not None
+
+    assert rebuilt.trust_env is False
+
+
 def test_requests_spec_carries_params() -> None:
     """Services hold an auth token in session.params; without it every segment request is unauthed."""
     rs = Session()

tests/remote/unit/test_remote_service_helpers.py

@@ -606,6 +606,7 @@ def test_the_client_session_downloads_through_the_proxy_it_sends(
     with pytest.raises(_SentCreate) as exc:
         svc.authenticate()
     assert dict(svc._session.proxies) == expected
+    assert svc._session.trust_env is no_proxy  # a proxied session ignores HTTP(S)_PROXY
     assert exc.value.args[0].get("proxy") == (None if no_proxy else "http://proxy.example:8080")
 
 

unshackle/core/downloaders/requests.py

@@ -28,6 +28,7 @@ from rich import filesize
 
 from unshackle.core.constants import DOWNLOAD_CANCELLED, DownloadCancelled
 from unshackle.core.utilities import get_debug_logger, get_extension
+from unshackle.core.utils import run_proxy
 
 MAX_ATTEMPTS = 5
 RETRY_WAIT = 2
@@ -1215,6 +1216,7 @@ def build_session_spec(session: Optional[Any]) -> Optional[dict[str, Any]]:
             "headers": dict(session.headers),
             "cookies": session.cookies,  # RequestsCookieJar pickles cleanly
             "proxies": dict(session.proxies),
+            "trust_env": session.trust_env,
             "params": copy(session.params),
             "verify": session.verify,
             "cert": session.cert,
@@ -1260,6 +1262,7 @@ def rebuild_session(spec: dict[str, Any], max_workers: int) -> Optional[Any]:
             rs.cookies.update(spec["cookies"])
         if spec.get("proxies"):
             rs.proxies.update(spec["proxies"])
+        rs.trust_env = spec.get("trust_env", True)
         if spec.get("params"):
             rs.params = spec["params"]
         rs.verify = spec.get("verify", True)
@@ -1649,6 +1652,7 @@ def requests(
             session.cookies.update(cookies)
         if proxy:
             session.proxies.update({"all": proxy})
+            run_proxy.pin(session)
         adapter = HTTPAdapter(pool_connections=max_workers, pool_maxsize=max_workers, pool_block=True)
         session.mount("https://", adapter)
         session.mount("http://", adapter)

unshackle/core/drm/clearkey.py

@@ -14,6 +14,7 @@ from requests import Session
 
 from unshackle.core.session import RnetSession
 from unshackle.core.utilities import log_event
+from unshackle.core.utils import run_proxy
 
 log = logging.getLogger(__name__)
 
@@ -124,7 +125,7 @@ class ClearKey:
             raise ValueError("No URI in M3U Key, unable to get Key.")
 
         if not session:
-            session = Session()
+            session = run_proxy.fallback_session()
 
         if not session.headers.get("User-Agent"):
             # commonly needed default for HLS playlists

unshackle/core/drm/clearkey_cenc.py

@@ -17,6 +17,7 @@ from unshackle.core import binaries
 from unshackle.core.config import config
 from unshackle.core.console import console
 from unshackle.core.utilities import log_event
+from unshackle.core.utils import run_proxy
 
 
 def b64url_encode_nopad(data: bytes) -> str:
@@ -114,7 +115,7 @@ class ClearKeyCENC:
 
         if response is None and self.laurl:
             if not session:
-                session = Session()
+                session = run_proxy.fallback_session()
                 session.headers.update(config.headers)
             r = session.post(self.laurl, data=challenge, headers={"Content-Type": "application/json"})
             r.raise_for_status()

unshackle/core/drm/playready.py

@@ -27,6 +27,7 @@ from unshackle.core.constants import AnyTrack
 from unshackle.core.drm import key_args as drm_key_args
 from unshackle.core.session import utf8_text
 from unshackle.core.utilities import get_boxes, log_event
+from unshackle.core.utils import run_proxy
 from unshackle.core.utils.subprocess import ffprobe
 
 
@@ -172,7 +173,7 @@ class PlayReady:
     @classmethod
     def from_track(cls, track: AnyTrack, session: Optional[Session] = None) -> PlayReady:
         if not session:
-            session = Session()
+            session = run_proxy.fallback_session()
             session.headers.update(config.headers)
 
         kid: Optional[UUID] = None

unshackle/core/drm/widevine.py

@@ -24,6 +24,7 @@ from unshackle.core.constants import AnyTrack
 from unshackle.core.drm import key_args as drm_key_args
 from unshackle.core.session import utf8_text
 from unshackle.core.utilities import get_boxes, log_event
+from unshackle.core.utils import run_proxy
 from unshackle.core.utils.subprocess import ffprobe
 
 
@@ -86,7 +87,7 @@ class Widevine:
             KIDNotFound: If the KID was not found within the data or PSSH.
         """
         if not session:
-            session = Session()
+            session = run_proxy.fallback_session()
             session.headers.update(config.headers)
 
         kid: Optional[UUID] = None

unshackle/core/import_service.py

@@ -30,6 +30,7 @@ from unshackle.core.titles import Episode, Movies, Series, Title_T, Titles_T, re
 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.utils import run_proxy
 
 log = logging.getLogger("import")
 E = TypeVar("E", bound=Enum)
@@ -299,6 +300,7 @@ class ImportService:
 
         if proxy:
             session.proxies.update({"all": proxy})
+            run_proxy.pin(session)
             if not explicit:
                 log.info(f"No --proxy given; using export region '{region}' via your proxy provider.")
         return session

unshackle/core/manifests/dash.py

@@ -40,6 +40,7 @@ from unshackle.core.tracks.track import (
     parse_sidx,
 )
 from unshackle.core.utilities import is_close_match, log_event, try_ensure_utf8
+from unshackle.core.utils import run_proxy
 from unshackle.core.utils.redact import safe_display_url
 from unshackle.core.utils.xml import load_xml
 
@@ -125,7 +126,7 @@ class DASH:
             raise TypeError(f"Expected url to be a {str}, not {url!r}")
 
         if not session:
-            session = Session()
+            session = run_proxy.fallback_session()
         elif not isinstance(session, (Session, RnetSession)):
             raise TypeError(f"Expected session to be a {Session} or {RnetSession}, not {session!r}")
 
@@ -481,8 +482,6 @@ class DASH:
                     seen_segments.add(seg)
                     segments.append(seg)
             segment_durations.extend(p_durations)
-            # ponytail: a sample comes from the first content period only; map the window across
-            # periods if a manifest with a short first period needs it
             if config.sample and segments:
                 break
 

unshackle/core/manifests/hls.py

@@ -41,6 +41,7 @@ from unshackle.core.session import RnetResponse, RnetSession, utf8_text
 from unshackle.core.tracks import Audio, DownloadContext, Subtitle, Tracks, Video, resume
 from unshackle.core.tracks.track import DRM_PREFERENCE_TYPES, assert_fragments_decrypted
 from unshackle.core.utilities import get_extension, is_close_match, log_event, try_ensure_utf8
+from unshackle.core.utils import run_proxy
 from unshackle.core.utils.redact import safe_display_url
 from unshackle.core.utils.subprocess import log_tool_run
 
@@ -63,7 +64,7 @@ class HLS:
             raise ValueError("Expected the M3U(8) manifest to be a Variant Playlist.")
 
         self.manifest = manifest
-        self.session = session or Session()
+        self.session = session or run_proxy.fallback_session()
         self.url = url
         self.raw_text = raw_text
 
@@ -75,7 +76,7 @@ class HLS:
             raise TypeError(f"Expected url to be a {str}, not {url!r}")
 
         if not session:
-            session = Session()
+            session = run_proxy.fallback_session()
         elif not isinstance(session, (Session, RnetSession)):
             raise TypeError(f"Expected session to be a {Session} or {RnetSession}, not {session!r}")
 
@@ -388,7 +389,7 @@ class HLS:
     ) -> Optional[tuple[int, int, Video.Codec]]:
         """Probe the first TS segment of a variant playlist to extract resolution and codec."""
         if not session:
-            session = Session()
+            session = run_proxy.fallback_session()
 
         res = session.get(variant_url)
         variant = m3u8.loads(utf8_text(res), uri=variant_url)
@@ -1372,7 +1373,7 @@ class HLS:
             value = getattr(data, "value", None)
             if not value and data.uri:
                 if not session:
-                    session = Session()
+                    session = run_proxy.fallback_session()
                 value = utf8_text(session.get(urljoin(manifest.base_uri or "", data.uri)))
 
             if not value:
@@ -1561,7 +1562,7 @@ class HLS:
         if not isinstance(session, (Session, RnetSession, type(None))):
             raise TypeError(f"Expected session to be a {Session} or {RnetSession}, not {type(session)}")
         if not session:
-            session = Session()
+            session = run_proxy.fallback_session()
 
         # TODO: Add support for 'SAMPLE-AES', 'AES-CTR', 'AES-CBC', 'ClearKey'
         if key.method == "AES-128":
@@ -1609,7 +1610,7 @@ class HLS:
         try:
             return HLS.get_drm(key, session)
         except Widevine.Exceptions.KIDNotFound:
-            kid = HLS.get_track_kid_from_init(master, track, probe_session or session or Session(), init_section)
+            kid = HLS.get_track_kid_from_init(master, track, probe_session or session or run_proxy.fallback_session(), init_section)
             if not kid:
                 raise
             return HLS.get_drm(key, session, kid=kid)

unshackle/core/manifests/ism.py

@@ -36,6 +36,7 @@ from unshackle.core.session import RnetSession
 from unshackle.core.tracks import Audio, DownloadContext, Subtitle, Track, Tracks, Video, resume
 from unshackle.core.tracks.track import assert_fragments_decrypted
 from unshackle.core.utilities import log_event, try_ensure_utf8
+from unshackle.core.utils import run_proxy
 from unshackle.core.utils.redact import safe_display_url
 from unshackle.core.utils.xml import load_xml
 
@@ -60,7 +61,7 @@ class ISM:
         if not url:
             raise requests.URLRequired("ISM manifest URL must be provided")
         if not session:
-            session = Session()
+            session = run_proxy.fallback_session()
         elif not isinstance(session, (Session, RnetSession)):
             raise TypeError(f"Expected session to be a {Session} or {RnetSession}, not {session!r}")
         res = session.get(url, **kwargs)

unshackle/core/manifests/m3u8.py

@@ -10,6 +10,7 @@ from requests import Session
 from unshackle.core.manifests.hls import HLS
 from unshackle.core.session import RnetSession
 from unshackle.core.tracks import Tracks
+from unshackle.core.utils import run_proxy
 
 
 def parse(
@@ -22,7 +23,7 @@ def parse(
     """Parse a variant playlist to ``Tracks`` with basic information, defer DRM loading."""
     tracks = HLS(master, session=session, url=url).to_tracks(language)
 
-    bool(master.session_keys or HLS.parse_session_data_keys(master, session or Session()))
+    bool(master.session_keys or HLS.parse_session_data_keys(master, session or run_proxy.fallback_session()))
 
     if True:
         for t in tracks.videos + tracks.audio:

unshackle/core/proxies/gluetun.py

@@ -14,6 +14,7 @@ import requests
 from unshackle.core import binaries
 from unshackle.core.proxies.proxy import Proxy
 from unshackle.core.utilities import get_country_code, get_country_name, get_debug_logger, log_event
+from unshackle.core.utils import run_proxy
 from unshackle.core.utils.ip_info import get_ip_info
 
 gluetun_instances: list["Gluetun"] = []
@@ -1128,6 +1129,7 @@ class Gluetun(Proxy):
         session = requests.Session()
         try:
             session.proxies = {"http": proxy_url, "https": proxy_url}
+            run_proxy.pin(session)
 
             for attempt in range(max_retries):
                 try:

unshackle/core/remote_service.py

@@ -48,6 +48,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.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
 from unshackle.core.vault import Vault
@@ -1074,6 +1075,7 @@ class RemoteService:
         local_proxy = None if no_proxy else (proxy or create_data.get("proxy"))
         if local_proxy:
             self._session.proxies["all"] = local_proxy
+            run_proxy.pin(self._session)
 
         if not no_proxy and not proxy and client_region:
             create_data["client_region"] = client_region

unshackle/core/service.py

@@ -42,6 +42,7 @@ from unshackle.core.titles import Title_T, Titles_T, remap_titles
 from unshackle.core.tracks import Chapters, Tracks
 from unshackle.core.tracks.video import Video
 from unshackle.core.utilities import declared_kwargs
+from unshackle.core.utils import run_proxy
 from unshackle.core.utils.ip_info import get_ip_info, verify_proxy_exit
 from unshackle.core.utils.redact import mask_proxy
 
@@ -299,6 +300,8 @@ class Service(metaclass=ABCMeta):
 
             if proxy:
                 self.session.proxies.update({"all": proxy})
+                if isinstance(self.session, requests.Session):
+                    run_proxy.pin(self.session)
                 # Don't set Proxy-Authorization manually: both rnet (Proxy.all) and
                 # requests authenticate from the credentials embedded in the proxy URL.
                 # A manual header here was malformed (no "Basic " scheme) and broke
@@ -322,6 +325,8 @@ class Service(metaclass=ABCMeta):
                     self.log.debug(f"Failed to get cached IP info: {e}")
                     self.current_region = None
 
+        run_proxy.watch(type(self), run_proxy.proxy_of(self.session))
+
     @classmethod
     def is_region_allowed(cls, region: Optional[str]) -> bool:
         """True when an exit IP in `region` (two-letter region code, any case) can use the service.

unshackle/core/session.py

@@ -6,6 +6,7 @@ import http
 import logging
 import math
 import random
+import sys
 import time
 from collections.abc import Iterator, MutableMapping
 from datetime import datetime, timezone
@@ -19,6 +20,7 @@ from requests import HTTPError, Request, Response
 from requests.structures import CaseInsensitiveDict
 
 from unshackle.core.config import config
+from unshackle.core.utils import run_proxy
 
 # rnet's only distinguishing role is TLS fingerprinting; retry, backoff,
 # pooling, and timeouts must match the plain-requests path. Both RnetSession
@@ -584,10 +586,7 @@ class RnetSession:
         ):
             if key in session_kwargs:
                 client_kwargs[key] = session_kwargs.pop(key)
-        if "proxy" in session_kwargs:
-            proxy_url = session_kwargs.pop("proxy")
-            if proxy_url:
-                client_kwargs["proxies"] = [rnet.Proxy.all(proxy_url)]
+        proxy_url = session_kwargs.pop("proxy", None)
 
         client_kwargs["cookie_store"] = True
 
@@ -602,6 +601,7 @@ class RnetSession:
         self.headers = RnetSessionHeaders(self)
         self.cookies = RnetCookieAdapter(None, self)
         self.proxies = RnetProxyDict(self)
+        self.proxy_exempt = False  # True: this HTTP session goes direct on purpose, so the proxy check skips it
 
         if "headers" in session_kwargs:
             self.headers.update(session_kwargs.pop("headers"))
@@ -609,6 +609,8 @@ class RnetSession:
             self.cookies.update(session_kwargs.pop("cookies"))
         if "proxies" in session_kwargs:
             self.proxies.update(session_kwargs.pop("proxies"))
+        if proxy_url:
+            self.proxies.update({"all": proxy_url})
 
     @property
     def impersonate_name(self) -> Optional[str]:
@@ -778,6 +780,11 @@ class RnetSession:
         url = self.build_url(url, kwargs.pop("params", None))
 
         parsed_origin = urlparse(url)
+        if run_proxy.watched and not self.proxy_exempt and not self._client_kwargs.get("proxies"):
+            try:
+                run_proxy.check(parsed_origin.hostname, sys._getframe(1))
+            except Exception:  # nosec B110 - the proxy check only warns, it must never fail a request
+                pass
         if parsed_origin.scheme and parsed_origin.netloc:
             self._origins.add(f"{parsed_origin.scheme}://{parsed_origin.netloc}")
 

unshackle/core/tracks/attachment.py

@@ -13,6 +13,7 @@ import requests
 from unshackle.core.config import config
 from unshackle.core.constants import DOWNLOAD_LICENCE_ONLY
 from unshackle.core.session import RnetSession
+from unshackle.core.utils import run_proxy
 
 AnySession = Union[requests.Session, RnetSession]
 
@@ -110,7 +111,7 @@ class Attachment:
 
         from unshackle.core.tracks.track import direct_session
 
-        session = session or self.session or requests.Session()
+        session = session or self.session or run_proxy.fallback_session()
         if no_proxy_download:
             if any(session.proxies.values()):
                 session = direct_session(session)

unshackle/core/tracks/track.py

@@ -34,6 +34,7 @@ from unshackle.core.events import events
 from unshackle.core.session import MAX_BACKOFF, RnetSession, utf8_text
 from unshackle.core.tracks import resume
 from unshackle.core.utilities import get_boxes, log_event, try_ensure_utf8
+from unshackle.core.utils import run_proxy
 from unshackle.core.utils.subprocess import ffprobe
 
 DRM_PREFERENCE_TYPES: dict[str, Union[type[Widevine], type[PlayReady]]] = {
@@ -48,8 +49,7 @@ def direct_session(session: Union[Session, "RnetSession"], proxy: Optional[str]
     """requests.Session with copied headers/cookies and no proxy, or only ``proxy`` when given."""
     new = Session()
     if proxy:
-        # requests lets HTTP(S)_PROXY override session proxies, so the env must be ignored for a chosen proxy
-        new.trust_env = False
+        run_proxy.pin(new)
         new.proxies = {"http": proxy, "https": proxy}
     headers = getattr(session, "headers", None)
     if headers is not None:
@@ -418,7 +418,7 @@ class DownloadContext:
         """Return the HTTP session, or a new ``Session`` if none was set."""
         session = self.session
         if not session:
-            session = Session()
+            session = run_proxy.fallback_session()
         elif not isinstance(session, (Session, RnetSession)):
             raise TypeError(f"Expected session to be a {Session} or {RnetSession}, not {session!r}")
         return session
@@ -1143,7 +1143,7 @@ class Track:
             from pywidevine.cdm import Cdm as WidevineCdm
             from pywidevine.pssh import PSSH as WV_PSSH
 
-            session = getattr(self, "session", None) or Session()
+            session = getattr(self, "session", None) or run_proxy.fallback_session()
 
             response = session.get(self.url)
             playlist = m3u8.loads(utf8_text(response), self.url)
@@ -1220,7 +1220,7 @@ class Track:
             url = self.url
 
         if not session:
-            session = Session()
+            session = run_proxy.fallback_session()
 
         content_length = maximum_size
 

unshackle/core/utils/ip_info.py

@@ -7,6 +7,7 @@ from typing import Any, Callable, Optional
 import requests
 
 from unshackle.core.cacher import Cacher
+from unshackle.core.utils import run_proxy
 
 CACHE_KEY = "ip_info_v4"
 CACHE_TTL = 86400  # 24 hours
@@ -113,7 +114,7 @@ def lookup_session(source: Optional[requests.Session]) -> requests.Session:
     if proxies:
         proxy = proxies.get("all") or proxies.get("https") or proxies.get("http")
         if proxy:
-            sess.trust_env = False
+            run_proxy.pin(sess)
             sess.proxies.update({"http": proxy, "https": proxy})
     return sess
 

unshackle/core/utils/run_proxy.py

@@ -0,0 +1,204 @@
+"""Keep service traffic on the proxy of the run, and warn about service requests that do not use it.
+
+Nothing here blocks a request: a request that does not go through the proxy only logs a warning.
+A service marks an HTTP session that goes direct on purpose with ``session.proxy_exempt = True``.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import logging
+import os
+import sys
+from dataclasses import dataclass, field
+from types import FrameType
+from typing import Any, Optional
+from urllib.parse import urlsplit
+
+import requests
+
+CORE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(__file__))) + os.sep  # the unshackle package
+SESSION_FILE = os.path.join(CORE_DIR, "core", "session.py")
+
+
+@dataclass
+class Watch:
+    cls: type
+    directory: str
+    proxies: list[str] = field(default_factory=list)  # every proxy a run of this service used, newest last
+
+
+watched: dict[str, Watch] = {}  # service directory -> Watch
+proxy_hosts: set[str] = set()
+warned: set[tuple[str, str, int]] = set()
+hook_installed = False
+
+
+def pin(session: requests.Session) -> None:
+    """Stop HTTP(S)_PROXY and the system proxy from overriding ``session.proxies``.
+
+    requests merges the environment proxies over the proxies of the HTTP session. Turning ``trust_env`` off also drops
+    the environment CA bundle, so this copies ``REQUESTS_CA_BUNDLE`` or ``CURL_CA_BUNDLE`` into ``verify``.
+    """
+    if session.verify is True:
+        session.verify = os.environ.get("REQUESTS_CA_BUNDLE") or os.environ.get("CURL_CA_BUNDLE") or True
+    session.trust_env = False
+
+
+def proxy_of(session: Any) -> Optional[str]:
+    proxies = getattr(session, "proxies", None) or {}
+    return proxies.get("all") or proxies.get("https") or proxies.get("http")
+
+
+def hostname(uri: str) -> str:
+    return (urlsplit(uri).hostname or "").lower()
+
+
+def service_dir(cls: type) -> Optional[str]:
+    file = getattr(sys.modules.get(cls.__module__), "__file__", None)
+    return os.path.dirname(file) + os.sep if file else None
+
+
+def watch(cls: type, proxy: Optional[str]) -> None:
+    """Watch the requests of the service class ``cls`` in a run with the proxy ``proxy``."""
+    global hook_installed
+    directory = service_dir(cls)
+    if not proxy or not directory:
+        return
+    entry = watched.get(directory)
+    if entry is None:
+        entry = watched[directory] = Watch(cls, directory)
+    if proxy not in entry.proxies:
+        entry.proxies.append(proxy)
+    proxy_hosts.add(hostname(proxy))
+    if not hook_installed:
+        sys.addaudithook(audit)
+        hook_installed = True
+
+
+def find_service(frame: Optional[FrameType], through_core: bool) -> Optional[tuple[FrameType, Watch]]:
+    """Return the innermost frame of watched service code and its watch.
+
+    Without ``through_core`` the walk ends at the first unshackle core frame: core sends some requests
+    direct on purpose (key vaults, metadata, CDM APIs). A frame of a SOCKS module also ends it, because
+    a ``socks5://`` proxy looks up the target host locally and still sends the request through the proxy.
+    """
+    entries = tuple(watched.values())
+    while frame is not None:
+        file = frame.f_code.co_filename
+        for entry in entries:
+            if file.startswith(entry.directory):
+                return frame, entry
+        if not through_core and (file.endswith("socks.py") or (file.startswith(CORE_DIR) and file != SESSION_FILE)):
+            return None
+        frame = frame.f_back
+    return None
+
+
+def instance_of(frame: Optional[FrameType], entry: Watch) -> Any:
+    """The service instance nearest to ``frame``, which knows the proxy of its own run.
+
+    The walk goes on past helper functions of the service, which have no ``self``.
+    """
+    while frame is not None:
+        inst = frame.f_locals.get("self")
+        if isinstance(inst, entry.cls) and getattr(inst, "session", None) is not None:
+            return inst
+        frame = frame.f_back
+    return None
+
+
+def run_proxy_for(frame: Optional[FrameType]) -> Optional[str]:
+    """The proxy of the run of the service code that called into core at ``frame``.
+
+    A download thread has no service frame. With one watched service, as in a CLI run, its proxy applies.
+    """
+    found = find_service(frame, through_core=True) if watched else None
+    if not found:
+        entries = tuple(watched.values())
+        return entries[0].proxies[-1] if len(entries) == 1 and len(entries[0].proxies) == 1 else None
+    service_frame, entry = found
+    inst = instance_of(service_frame, entry)
+    if inst is not None:
+        return proxy_of(getattr(inst, "session", None))
+    return entry.proxies[-1] if entry.proxies else None
+
+
+def fallback_session() -> requests.Session:
+    """A new requests session for a core helper that got no session, on the proxy of the calling service."""
+    session = requests.Session()
+    proxy = run_proxy_for(sys._getframe(1))
+    if proxy:
+        session.proxies.update({"all": proxy})
+        pin(session)
+    return session
+
+
+def is_local(host: str) -> bool:
+    """True for this machine and the local network, which never see the internet IP of the user."""
+    if host in ("localhost", "localhost.", "localhost.localdomain"):
+        return True
+    if host.endswith((".localhost", ".local", ".lan", ".home.arpa")):
+        return True
+    try:
+        ip = ipaddress.ip_address(host)
+    except ValueError:
+        return False
+    if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped:
+        ip = ip.ipv4_mapped
+    return ip.is_private or ip.is_link_local or ip.is_unspecified
+
+
+def audit(event: str, args: tuple) -> None:
+    """Python raises ``socket.getaddrinfo`` for each connection that requests or urllib opens.
+
+    A raw ``socket.connect`` to a host name looks up the name in C with no ``socket.getaddrinfo`` event.
+    A connect to an IP address follows a lookup that this hook already saw, so only a host name is checked.
+    """
+    try:
+        if not watched:
+            return
+        if event == "socket.getaddrinfo":
+            check(args[0], sys._getframe(1))
+        elif event == "socket.connect" and isinstance(args[1], tuple) and isinstance(args[1][0], str):
+            try:
+                ipaddress.ip_address(args[1][0].split("%")[0])
+            except ValueError:
+                check(args[1][0], sys._getframe(1))
+    except Exception:  # nosec B110 - an audit hook that raises stops the lookup of the caller
+        pass
+
+
+def check(host: Any, frame: Optional[FrameType]) -> None:
+    """Warn when the innermost unshackle frame that sends a request to ``host`` is service code.
+
+    A method of a service instance knows the proxy of its own run, so a run with no proxy never warns.
+    """
+    if isinstance(host, bytes):
+        host = host.decode(errors="replace")
+    host = str(host or "").lower().strip("[]")
+    if not host or is_local(host) or host in proxy_hosts:
+        return
+    found = find_service(frame, through_core=False)
+    if not found:
+        return
+    service_frame, entry = found
+    while frame is not None and frame is not service_frame:
+        # the requests frames between the lookup and the service code hold the HTTP session that sends it
+        if getattr(frame.f_locals.get("self"), "proxy_exempt", False) is True:
+            return
+        frame = frame.f_back
+    inst = instance_of(service_frame, entry)
+    if inst is not None and not proxy_of(getattr(inst, "session", None)):
+        return
+    key = (entry.cls.__name__, host, id(inst))
+    if key in warned:
+        return
+    warned.add(key)
+    # a service instance logs through self.log, which a remote session mirrors to its client
+    log = getattr(inst, "log", None) or logging.getLogger(entry.cls.__name__)
+    where = os.path.relpath(service_frame.f_code.co_filename, os.path.dirname(entry.directory[:-1]))
+    log.warning(
+        f"A request to {host} did not go through the proxy, so {host} sees your own IP "
+        f"({where}:{service_frame.f_lineno})"
+    )