From a2a707017089b0518046c2cad473e10f240b448c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pierre-Etienne=20Cheri=C3=A8re?= Date: Sun, 11 Oct 2026 20:18:12 -0700 Subject: [PATCH 1/2] fix: don't log the access token on stream errors repr(aiohttp.ClientResponseError) includes the request headers, so the "Client error" and "Authentication failed" log lines contained "Authorization: Bearer ". Log the exception type and str(error) instead, which keeps the status, message and URL. --- teslemetry_stream/stream.py | 24 +++- tests/test_no_token_in_logs.py | 202 +++++++++++++++++++++++++++++++++ 2 files changed, 222 insertions(+), 4 deletions(-) create mode 100644 tests/test_no_token_in_logs.py diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index 4c05de4..8dbcdd0 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -377,18 +377,18 @@ async def __anext__(self) -> dict[str, Any]: # A rejected token is a definitive answer, not a transient # blip - retrying it can never succeed, and doing so masks # a bad credential as an indefinitely quiet stream. - LOGGER.error("Authentication failed, not retrying: %s", repr(error)) + LOGGER.error("Authentication failed, not retrying: %s", _describe(error)) self.active = False self._connection_failed() raise TeslemetryStreamAuthenticationError() from error - LOGGER.warning("Client error: %s", repr(error)) + LOGGER.warning("Client error: %s", _describe(error)) self._connection_failed() delay = min(2**self.retries, 600) LOGGER.debug("Reconnecting in %s seconds", delay) await asyncio.sleep(delay) self.retries += 1 except Exception as error: - LOGGER.error("Unexpected error: %s", repr(error)) + LOGGER.error("Unexpected error: %s", _describe(error)) self._connection_failed() LOGGER.debug("Reconnecting in %s seconds", 1) await asyncio.sleep(1) @@ -481,7 +481,9 @@ def _mark_connected(value: bool) -> None: self._dispatch(event) except BaseException as error: if not connected_once: - LOGGER.warning("Listen task ended before its first connect: %r", error) + LOGGER.warning( + "Listen task ended before its first connect: %s", _describe(error) + ) raise if not connected_once: LOGGER.warning("Listen task ended before its first connect") @@ -646,6 +648,20 @@ def _parse_created_at(created_at: str) -> int: return int(dt.timestamp()) * 1000 + dt.microsecond // 1000 +def _describe(error: BaseException) -> str: + """ + Describe an exception for the log without its repr. + + The repr of an aiohttp.ClientResponseError embeds its RequestInfo, including the + request headers - the Authorization header carries the access token. str() of the + aiohttp errors has the status, message and URL only. + + :param error: The exception to describe. + :return: ": ". + """ + return f"{type(error).__name__}: {error}" + + def _now() -> str: """Current time in the same format the stream sends `createdAt` in.""" return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" diff --git a/tests/test_no_token_in_logs.py b/tests/test_no_token_in_logs.py new file mode 100644 index 0000000..5ee0367 --- /dev/null +++ b/tests/test_no_token_in_logs.py @@ -0,0 +1,202 @@ +"""Regression tests: the access token must never reach the log. + +aiohttp.ClientResponseError's repr embeds its RequestInfo, and with it the +request headers - including "Authorization: Bearer ". Logging +repr(error) on a stream error (e.g. a 5xx from the SSE endpoint) wrote the +token into every Home Assistant log that shipped it anywhere. These tests +fail any log record that contains the token, on the retry path (5xx), the +auth-failure path (401) and the listen-task path. +""" +from __future__ import annotations + +import asyncio +import contextlib +import logging +from typing import Any + +import aiohttp +from multidict import CIMultiDict, CIMultiDictProxy +from yarl import URL + +from teslemetry_stream.exception import TeslemetryStreamAuthenticationError +from teslemetry_stream.stream import LOGGER, TeslemetryStream + +TOKEN = "s3cr3t-access-token-value" +REQUEST_INFO = aiohttp.RequestInfo( + url=URL("https://fake.teslemetry.com/sse"), + method="GET", + headers=CIMultiDictProxy(CIMultiDict({"Authorization": f"Bearer {TOKEN}"})), + real_url=URL("https://fake.teslemetry.com/sse"), +) + + +class FakeContent: + """Async-iterable response body that blocks until failed or cancelled.""" + + def __init__(self) -> None: + self._blocker: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + def __aiter__(self) -> FakeContent: + return self + + async def __anext__(self) -> bytes: + await self._blocker + raise AssertionError("unreachable - blocker only resolves via an exception") + + def fail(self, exc: BaseException) -> None: + if not self._blocker.done(): + self._blocker.set_exception(exc) + + +class FakeResponse: + def __init__(self) -> None: + self.url = "https://fake.teslemetry.com/sse" + self.status = 200 + self.content = FakeContent() + self.closed = False + + def close(self) -> None: + self.closed = True + + +class FakeSession: + """Each `get()` raises or returns the next queued result.""" + + def __init__(self, get_results: list[Any]) -> None: + self.calls = 0 + self._get_results = list(get_results) + self.responses: list[FakeResponse] = [] + + async def get(self, url: str, **kwargs: Any) -> FakeResponse: + self.calls += 1 + result = self._get_results.pop(0) + if isinstance(result, BaseException): + raise result + self.responses.append(result) + return result + + +def make_stream(session: FakeSession) -> TeslemetryStream: + return TeslemetryStream( + session=session, # type: ignore[arg-type] + access_token=TOKEN, + server="api.teslemetry.com", + manual=True, + ) + + +def response_error(status: int, message: str) -> aiohttp.ClientResponseError: + return aiohttp.ClientResponseError( + request_info=REQUEST_INFO, history=(), status=status, message=message + ) + + +def check(label: str, ok: bool, detail: str = "") -> bool: + print(f"{label:<72} {'PASS' if ok else 'FAIL'}{' ' + detail if detail else ''}") + return ok + + +def leaked(records: list[logging.LogRecord]) -> list[str]: + return [r.getMessage() for r in records if TOKEN in r.getMessage()] + + +async def drain_cancelled(task: asyncio.Task[Any]) -> None: + with contextlib.suppress(asyncio.CancelledError): + await task + + +async def test_fixture_would_leak_through_repr(results: list[bool]) -> None: + """Guard: the fixture really carries the token in its repr, so the + other tests would catch a regression back to repr(error).""" + results.append( + check( + "repr(ClientResponseError) contains the token", + TOKEN in repr(response_error(500, "x")), + ) + ) + + +async def test_server_error_does_not_log_token(results: list[bool]) -> None: + session = FakeSession([FakeResponse(), FakeResponse()]) + stream = make_stream(session) + + with CaptureLogs() as records: + task = asyncio.create_task(stream.listen()) + await asyncio.sleep(0) + await asyncio.sleep(0) + session.responses[0].content.fail(response_error(502, "Bad Gateway")) + # retries starts at 0, so the first backoff delay is 2**0 == 1 second. + await asyncio.sleep(1.2) + + warnings = [r for r in records if "Client error" in r.getMessage()] + results.append( + check("a 502 logs a Client error warning", len(warnings) == 1, f"got {warnings}") + ) + results.append( + check( + "the warning keeps the status and message", + bool(warnings) and "502" in warnings[0].getMessage() + and "Bad Gateway" in warnings[0].getMessage(), + ) + ) + results.append(check("no log record contains the token (502)", not leaked(records))) + + stream.close() + await drain_cancelled(task) + + +async def test_auth_failure_does_not_log_token(results: list[bool]) -> None: + session = FakeSession([response_error(401, "Unauthorized")]) + stream = make_stream(session) + + with CaptureLogs() as records, contextlib.suppress(TeslemetryStreamAuthenticationError): + await stream.listen() + + results.append( + check( + "a 401 logs the authentication failure", + any("Authentication failed" in r.getMessage() for r in records), + ) + ) + results.append(check("no log record contains the token (401)", not leaked(records))) + + +class CaptureLogs: + """Minimal stand-in for pytest's caplog fixture, scoped to LOGGER.""" + + def __enter__(self) -> list[logging.LogRecord]: + self._records: list[logging.LogRecord] = [] + self._handler = _ListHandler(self._records) + self._prev_level = LOGGER.level + LOGGER.addHandler(self._handler) + LOGGER.setLevel(logging.DEBUG) + return self._records + + def __exit__(self, *exc_info: Any) -> None: + LOGGER.removeHandler(self._handler) + LOGGER.setLevel(self._prev_level) + + +class _ListHandler(logging.Handler): + def __init__(self, records: list[logging.LogRecord]) -> None: + super().__init__() + self._records = records + + def emit(self, record: logging.LogRecord) -> None: + self._records.append(record) + + +async def main() -> None: + results: list[bool] = [] + await test_fixture_would_leak_through_repr(results) + await test_server_error_does_not_log_token(results) + await test_auth_failure_does_not_log_token(results) + + print("-" * 72) + print("ALL PASS" if all(results) else "FAILURES PRESENT") + if not all(results): + raise SystemExit(1) + + +if __name__ == "__main__": + asyncio.run(main()) From 251cc667095b6daf8e744a2423b6bce49f7be1d5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pierre-Etienne=20Cheri=C3=A8re?= Date: Sun, 11 Oct 2026 20:56:29 -0700 Subject: [PATCH 2/2] test: cover _describe and the listen-task warning, check 401 log content The 401 test now also checks the log keeps the status and message, and a new test reaches the "ended before its first connect" warning with a token-bearing error (it fails if that line goes back to repr). --- tests/test_no_token_in_logs.py | 50 +++++++++++++++++++++++++++++++--- 1 file changed, 46 insertions(+), 4 deletions(-) diff --git a/tests/test_no_token_in_logs.py b/tests/test_no_token_in_logs.py index 5ee0367..88d58ec 100644 --- a/tests/test_no_token_in_logs.py +++ b/tests/test_no_token_in_logs.py @@ -5,7 +5,7 @@ repr(error) on a stream error (e.g. a 5xx from the SSE endpoint) wrote the token into every Home Assistant log that shipped it anywhere. These tests fail any log record that contains the token, on the retry path (5xx), the -auth-failure path (401) and the listen-task path. +auth-failure path (401) and the listen-task path, and check _describe itself. """ from __future__ import annotations @@ -19,7 +19,7 @@ from yarl import URL from teslemetry_stream.exception import TeslemetryStreamAuthenticationError -from teslemetry_stream.stream import LOGGER, TeslemetryStream +from teslemetry_stream.stream import LOGGER, TeslemetryStream, _describe TOKEN = "s3cr3t-access-token-value" REQUEST_INFO = aiohttp.RequestInfo( @@ -116,6 +116,19 @@ async def test_fixture_would_leak_through_repr(results: list[bool]) -> None: ) +async def test_describe_omits_headers(results: list[bool]) -> None: + described = _describe(response_error(400, "Bad Request")) + results.append(check("_describe omits the token", TOKEN not in described, described)) + results.append( + check( + "_describe keeps the type, status and message", + described.startswith("ClientResponseError: 400") + and "Bad Request" in described, + described, + ) + ) + + async def test_server_error_does_not_log_token(results: list[bool]) -> None: session = FakeSession([FakeResponse(), FakeResponse()]) stream = make_stream(session) @@ -152,15 +165,42 @@ async def test_auth_failure_does_not_log_token(results: list[bool]) -> None: with CaptureLogs() as records, contextlib.suppress(TeslemetryStreamAuthenticationError): await stream.listen() + auth_logs = [r.getMessage() for r in records if "Authentication failed" in r.getMessage()] + results.append(check("a 401 logs the authentication failure", len(auth_logs) == 1)) results.append( check( - "a 401 logs the authentication failure", - any("Authentication failed" in r.getMessage() for r in records), + "the authentication log keeps the status and message", + bool(auth_logs) and "401" in auth_logs[0] and "Unauthorized" in auth_logs[0], + f"got {auth_logs}", ) ) results.append(check("no log record contains the token (401)", not leaked(records))) +async def test_listen_ended_before_connect_does_not_log_token(results: list[bool]) -> None: + """A connection listener that raises while the first connect is failing + ends the listen task, which logs the escaping error.""" + session = FakeSession([response_error(502, "Bad Gateway")]) + stream = make_stream(session) + + def failing_listener(connected: bool) -> None: + raise response_error(500, "listener failed") + + stream.async_add_connection_listener(failing_listener) + + with CaptureLogs() as records, contextlib.suppress(aiohttp.ClientResponseError): + await stream.listen() + + ended = [r.getMessage() for r in records if "ended before its first connect" in r.getMessage()] + results.append( + check("the listen task logs ending before its first connect", len(ended) == 1) + ) + results.append( + check("no log record contains the token (listen task)", not leaked(records)) + ) + stream.close() + + class CaptureLogs: """Minimal stand-in for pytest's caplog fixture, scoped to LOGGER.""" @@ -189,8 +229,10 @@ def emit(self, record: logging.LogRecord) -> None: async def main() -> None: results: list[bool] = [] await test_fixture_would_leak_through_repr(results) + await test_describe_omits_headers(results) await test_server_error_does_not_log_token(results) await test_auth_failure_does_not_log_token(results) + await test_listen_ended_before_connect_does_not_log_token(results) print("-" * 72) print("ALL PASS" if all(results) else "FAILURES PRESENT")