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()