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..88d58ec --- /dev/null +++ b/tests/test_no_token_in_logs.py @@ -0,0 +1,244 @@ +"""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, and check _describe itself. +""" +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, _describe + +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_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) + + 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() + + 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( + "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.""" + + 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_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") + if not all(results): + raise SystemExit(1) + + +if __name__ == "__main__": + asyncio.run(main())