From e1faac805b14c1b12134e760f0a49b870ef63c69 Mon Sep 17 00:00:00 2001 From: Petrus Pennanen Date: Fri, 9 Oct 2026 16:26:12 +0300 Subject: [PATCH] HA token requests never follow a redirect Codex's review of #76 (P1): urllib's default redirect handler carries every header except the body ones to the next request, Authorization included. So if the private Home Assistant, or the proxy in front of it, answered 30x to another host, CarWatch sent the long-lived HA token to a host that never passed _is_private_ha. Measured: Python 3.11, 3.12 (CI) and 3.13 forward it; 3.14 already drops it on a cross-origin redirect. - carwatch/tokenhttp.py: an opener whose redirect handler refuses (HTTPError with the 30x code and the target) instead of following. - radiation.read_ha uses it for every token request, and reports a redirect as such ("set radiation.ha.url to the final address"). - The same bug existed in mercedesme._get and _post, which send the same token: both use tokenhttp now. _post also lacked the private-host check that _get has; it has it now. - Tests: test_tokenhttp.py with two local servers (HOME redirects, OTHER logs every Authorization it sees). A negative control shows plain urllib carrying the token to OTHER; tokenhttp, radiation and both Mercedes calls must refuse, and OTHER must see nothing. Reverting either module's switch makes its tests fail. Full suite on 3.12: 235 tests OK. Not changed: room.py sends the GroupMind key to https://groupmind.one, which does not redirect; lower risk, separate decision. Co-Authored-By: Claude Opus 5.5 --- carwatch/mercedesme.py | 14 +++- carwatch/radiation.py | 9 ++- carwatch/tokenhttp.py | 31 ++++++++ tests/test_radiation.py | 17 +++++ tests/test_tokenhttp.py | 165 ++++++++++++++++++++++++++++++++++++++++ 5 files changed, 231 insertions(+), 5 deletions(-) create mode 100644 carwatch/tokenhttp.py create mode 100644 tests/test_tokenhttp.py diff --git a/carwatch/mercedesme.py b/carwatch/mercedesme.py index e9bd622..b1ceb12 100644 --- a/carwatch/mercedesme.py +++ b/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 diff --git a/carwatch/radiation.py b/carwatch/radiation.py index 7014a7d..2bf8cef 100644 --- a/carwatch/radiation.py +++ b/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: diff --git a/carwatch/tokenhttp.py b/carwatch/tokenhttp.py new file mode 100644 index 0000000..d3470fc --- /dev/null +++ b/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) diff --git a/tests/test_radiation.py b/tests/test_radiation.py index e940f70..16fee58 100644 --- a/tests/test_radiation.py +++ b/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) diff --git a/tests/test_tokenhttp.py b/tests/test_tokenhttp.py new file mode 100644 index 0000000..e67f2e4 --- /dev/null +++ b/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()