HA token requests never follow a redirect (P1 from the #76 review)

carwatch/mercedesme.py

@@ -30,7 +30,7 @@
 import urllib.parse
 import urllib.request
 
-from carwatch import cloudcar
+from carwatch import cloudcar, tokenhttp
 
 
 # Hostnames that name a private mesh/LAN rather than the public internet.
@@ -146,16 +146,22 @@ def _get(path: str, token: str, timeout: float = 8.0):
         base + path,
         headers={"Authorization": f"Bearer {token}",
                  "Content-Type": "application/json"})
-    with urllib.request.urlopen(req, timeout=timeout) as r:
+    # ...and never follow a redirect with it either (carwatch/tokenhttp.py).
+    with tokenhttp.urlopen(req, timeout=timeout) as r:
         return json.loads(r.read().decode())
 
 
 def _post(path: str, token: str, payload: dict, timeout: float = 15.0):
+    # Same two rules as _get: a private target only, and no redirects. _post had
+    # neither, so a command could carry the token anywhere _ha_url() pointed.
+    base = _ha_url()
+    if not _is_private_ha(base):
+        raise ValueError("refusing to send HA token to a non-private host")
     req = urllib.request.Request(
-        _ha_url() + path, data=json.dumps(payload).encode(),
+        base + path, data=json.dumps(payload).encode(),
         headers={"Authorization": f"Bearer {token}",
                  "Content-Type": "application/json"}, method="POST")
-    with urllib.request.urlopen(req, timeout=timeout) as r:
+    with tokenhttp.urlopen(req, timeout=timeout) as r:
         return r.status
 
 

carwatch/radiation.py

@@ -46,6 +46,7 @@
 import urllib.request
 from datetime import datetime
 
+from carwatch import tokenhttp
 from carwatch.config import load_raw
 
 DEFAULT_STALE_S = 60
@@ -111,7 +112,9 @@ def _expand(p: str) -> str:
 
 def _get_json(url: str, headers: dict | None = None):
     req = urllib.request.Request(url, headers=headers or {})
-    with urllib.request.urlopen(req, timeout=_HTTP_TIMEOUT) as r:
+    # A request carrying a token never follows a redirect (carwatch/tokenhttp.py).
+    opener = tokenhttp.urlopen if "Authorization" in (headers or {}) else urllib.request.urlopen
+    with opener(req, timeout=_HTTP_TIMEOUT) as r:
         return r.read()
 
 
@@ -223,6 +226,10 @@ def read_ha(c: dict, now: float) -> dict:
         try:
             states[key] = json.loads(_get_json(f"{url}/api/states/{eid}", hdr).decode())
         except urllib.error.HTTPError as e:
+            if 300 <= e.code < 400:
+                raise SourceError(f"Home Assistant answered a redirect (HTTP {e.code}); the token "
+                                  "is never sent on after a redirect, so set radiation.ha.url "
+                                  "to the final address")
             if e.code in (401, 403):
                 raise SourceError(f"Home Assistant rejected the token (HTTP {e.code})")
             if e.code == 404:

carwatch/tokenhttp.py

@@ -0,0 +1,31 @@
+"""HTTP for requests that carry a secret token: redirects are refused, never followed.
+
+urllib's default redirect handler builds the follow-up request with the
+original headers (CPython drops only the body headers), so a token request
+that gets a 30x hands the token to whatever host the Location names. For the
+Home Assistant token that host has never passed `_is_private_ha`: a private HA
+or the proxy in front of it that redirects to a public URL would leak a
+long-lived token with full control of the house (Codex review of CarWatch #76).
+
+So a token request answers a redirect with an HTTPError carrying the 30x code
+and the target, and the caller reports it. Point the configured URL at the
+final address instead of relying on a redirect.
+"""
+import urllib.error
+import urllib.request
+
+
+class _RefuseRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        raise urllib.error.HTTPError(
+            req.full_url, code,
+            f"redirect to {newurl} refused: this request carries a token", headers, fp)
+
+
+# build_opener drops its default HTTPRedirectHandler when given a subclass of it.
+_OPENER = urllib.request.build_opener(_RefuseRedirect)
+
+
+def urlopen(req, timeout: float):
+    """urllib.request.urlopen for token-bearing requests: same API, no redirects."""
+    return _OPENER.open(req, timeout=timeout)

tests/test_radiation.py

@@ -53,6 +53,12 @@ class H(BaseHTTPRequestHandler):
         def do_GET(self):
             _Fake.seen_auth.append(self.headers.get("Authorization"))
             code, body = _Fake.routes.get(self.path, (404, {"message": "Entity not found."}))
+            if 300 <= code < 400:   # {"location": url} -> a redirect
+                self.send_response(code)
+                self.send_header("Location", body["location"])
+                self.send_header("Content-Length", "0")
+                self.end_headers()
+                return
             data = body if isinstance(body, bytes) else json.dumps(body).encode()
             self.send_response(code)
             self.send_header("Content-Type", "application/json")
@@ -291,6 +297,17 @@ def test_401(self):
         self.assertIn("rejected the token (HTTP 401)", d["error"])
         self.assertNotIn(self.TOKEN, json.dumps(d))
 
+    def test_redirect_is_refused_and_the_token_goes_nowhere(self):
+        # Codex review of #76: urllib would follow a 30x with the Authorization header.
+        # The target is a dead port: following it would read "unreachable", not "redirect".
+        _Fake.routes = {p: (302, {"location": _dead_url() + p}) for p in _ha_routes()}
+        d = radiation.status(self._cfg(), now=NOW)
+        self.assertEqual(d["state"], "error", d)
+        self.assertIn("redirect (HTTP 302)", d["error"])
+        self.assertNotIn("unreachable", d["error"])
+        self.assertNotIn(self.TOKEN, json.dumps(d))
+        self.assertEqual(len(_Fake.seen_auth), 1)   # one request to the private HA, none after
+
     def test_unreachable(self):
         t0 = time.time()
         d = radiation.status(self._cfg(url=_dead_url()), now=NOW)

tests/test_tokenhttp.py

@@ -0,0 +1,165 @@
+"""carwatch.tokenhttp: a request that carries a token never follows a redirect.
+
+Two local servers. HOME answers /redirect with a 302 to OTHER (urllib follows a 302 even
+for a POST, turning it into a GET and keeping the headers), which records every Authorization header it receives. The negative
+control first shows that plain urllib DOES carry the token to OTHER on this
+Python, so the remaining tests check a real hazard, not a vacuous one. Then
+tokenhttp, Home Assistant's radiation reader and the Mercedes module's _get /
+_post must all refuse the redirect, and OTHER must never see the token."""
+
+import json
+import socket
+import socketserver
+import threading
+import unittest
+import urllib.error
+import urllib.request
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from unittest import mock
+
+from carwatch import mercedesme, tokenhttp
+
+TOKEN = "test-token-not-a-secret-redirect-0123"
+
+
+class _Other(BaseHTTPRequestHandler):
+    seen: list = []
+
+    def _answer(self):
+        n = int(self.headers.get("Content-Length") or 0)
+        if n:
+            self.rfile.read(n)
+        _Other.seen.append(self.headers.get("Authorization"))
+        body = b'{"ok": true}'
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    do_GET = do_POST = _answer
+
+    def log_message(self, *a):
+        pass
+
+
+class _Home(BaseHTTPRequestHandler):
+    other = ""
+
+    def _answer(self, code):
+        n = int(self.headers.get("Content-Length") or 0)
+        if n:
+            self.rfile.read(n)
+        if self.path.startswith("/redirect"):
+            self.send_response(code)
+            self.send_header("Location", _Home.other + "/landed")
+            self.send_header("Content-Length", "0")
+            self.end_headers()
+            return
+        body = b'{"ok": true}'
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_GET(self):
+        self._answer(302)
+
+    def do_POST(self):
+        self._answer(302)
+
+    def log_message(self, *a):
+        pass
+
+
+class _S(ThreadingHTTPServer):
+    def server_bind(self):   # skip getfqdn(), see test_webchat_ask
+        socketserver.TCPServer.server_bind(self)
+        self.server_name, self.server_port = "localhost", self.server_address[1]
+
+
+def _serve(handler):
+    s = socket.socket(); s.bind(("127.0.0.1", 0)); port = s.getsockname()[1]; s.close()
+    srv = _S(("127.0.0.1", port), handler)
+    threading.Thread(target=srv.serve_forever, daemon=True).start()
+    return srv, f"http://127.0.0.1:{port}"
+
+
+def setUpModule():
+    global HOME_SRV, OTHER_SRV, HOME, OTHER
+    OTHER_SRV, OTHER = _serve(_Other)
+    HOME_SRV, HOME = _serve(_Home)
+    _Home.other = OTHER
+
+
+def tearDownModule():
+    HOME_SRV.shutdown()
+    OTHER_SRV.shutdown()
+
+
+def _req(path, method="GET"):
+    data = json.dumps({"vin": "TEST"}).encode() if method == "POST" else None
+    return urllib.request.Request(HOME + path, data=data, method=method,
+                                  headers={"Authorization": f"Bearer {TOKEN}",
+                                           "Content-Type": "application/json"})
+
+
+class TestTokenHttp(unittest.TestCase):
+    def setUp(self):
+        _Other.seen = []
+
+    def test_negative_control_plain_urllib_carries_the_token(self):
+        # Measured 9 Oct 2026: Python 3.11, 3.12 (CI) and 3.13 carry it; 3.14 drops it itself.
+        for method in ("GET", "POST"):
+            _Other.seen = []
+            with urllib.request.urlopen(_req("/redirect", method), timeout=5) as r:
+                r.read()
+            if f"Bearer {TOKEN}" not in _Other.seen:
+                self.skipTest("this Python drops Authorization on redirects by itself")
+            self.assertIn(f"Bearer {TOKEN}", _Other.seen, method)
+
+    def test_get_redirect_is_refused(self):
+        with self.assertRaises(urllib.error.HTTPError) as e:
+            tokenhttp.urlopen(_req("/redirect"), timeout=5)
+        self.assertEqual(e.exception.code, 302)
+        self.assertIn("refused", str(e.exception))
+        self.assertEqual(_Other.seen, [])
+
+    def test_post_redirect_is_refused(self):
+        with self.assertRaises(urllib.error.HTTPError) as e:
+            tokenhttp.urlopen(_req("/redirect", "POST"), timeout=5)
+        self.assertEqual(e.exception.code, 302)
+        self.assertEqual(_Other.seen, [])
+
+    def test_a_plain_answer_still_works(self):
+        with tokenhttp.urlopen(_req("/ok"), timeout=5) as r:
+            self.assertEqual(json.loads(r.read()), {"ok": True})
+
+
+class TestMercedesTokenRequests(unittest.TestCase):
+    def setUp(self):
+        _Other.seen = []
+        p = mock.patch.object(mercedesme, "_ha_url", lambda: HOME)
+        p.start(); self.addCleanup(p.stop)
+
+    def test_get_does_not_follow_a_redirect(self):
+        with self.assertRaises(urllib.error.HTTPError):
+            mercedesme._get("/redirect", TOKEN, timeout=5)
+        self.assertEqual(_Other.seen, [])
+
+    def test_post_does_not_follow_a_redirect(self):
+        with self.assertRaises(urllib.error.HTTPError):
+            mercedesme._post("/redirect", TOKEN, {"vin": "TEST"}, timeout=5)
+        self.assertEqual(_Other.seen, [])
+
+    def test_post_refuses_a_public_host_before_connecting(self):
+        with mock.patch.object(mercedesme, "_ha_url", lambda: "http://8.8.8.8:8123"), \
+             mock.patch.object(tokenhttp, "urlopen") as opened:
+            with self.assertRaises(ValueError):
+                mercedesme._post("/api/services/x", TOKEN, {"vin": "TEST"})
+            opened.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()