From 235b543c867a2e899255fa41744724f6b838ca8c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Antonio=20V=C3=A1zquez=20Blanco?= Date: Wed, 28 Oct 2026 11:05:50 +0100 Subject: [PATCH] Log in again when the device rejects the session With keep_credentials=True, a request that gets a 401 now logs in again with the kept credentials and is sent once more. The device rejects it in its auth middleware before running it, so nothing runs twice. This covers API calls, the start of a stream and WebSocket handshakes, but not uploads, whose body cannot be sent twice. The renewal keeps the session generation, since the account is the same, so operations in progress carry on. Calls that lose the session together share one login, and a login or logout by the caller still wins. Rejected credentials are dropped and raise NanoKVMAuthenticationFailure; an unreachable device raises its error and the next call tries again. --- docs/connecting.md | 10 ++ nanokvm/components/session.py | 201 ++++++++++++++++--------- tests/test_application_updates.py | 3 + tests/test_kept_credentials.py | 235 +++++++++++++++++++++++++++++- 4 files changed, 382 insertions(+), 67 deletions(-) diff --git a/docs/connecting.md b/docs/connecting.md index 31b1273..932e831 100644 --- a/docs/connecting.md +++ b/docs/connecting.md @@ -47,6 +47,16 @@ in again after the device restarts, as `install_application_update()` needs. Logging out, changing the password, a failed login and closing the client drop them. They are never written to disk or shown in `repr()`. +With kept credentials, the client also logs in again by itself when the device +rejects the session, for example after its server restarted or someone logged +out, and sends the rejected request once more. The device rejects such a +request before running it, so nothing runs twice. This covers API calls, the +start of a stream and WebSocket handshakes, but not uploads, whose body cannot +be sent twice. Calls that lose the session together share one login. If the +device rejects the credentials, they are dropped and the call raises +`NanoKVMAuthenticationFailure`; if it cannot be reached, the error is raised +and the next call tries again. + ### Sessions and logout Starting another login or logging out invalidates operations and streams still diff --git a/nanokvm/components/session.py b/nanokvm/components/session.py index 28b5f32..1963fc0 100644 --- a/nanokvm/components/session.py +++ b/nanokvm/components/session.py @@ -106,6 +106,7 @@ def __init__( self._ws_heartbeat: asyncio.Task[None] | None = None self._ws_session: ClientSession | None = None self._ws_lock = asyncio.Lock() + self._renew_lock = asyncio.Lock() self._mouse_buttons = 0 self._mouse_mode: Literal["relative", "absolute"] = "relative" self._mouse_abs_position: tuple[int, int] = (0, 0) @@ -170,56 +171,73 @@ async def request( expected_generation: int | None = None, **kwargs: Any, ) -> AsyncGenerator[ClientResponse, None]: - """Make an authenticated or unauthenticated API request.""" + """Make an authenticated or unauthenticated API request. + + With kept credentials, a request rejected with 401 logs in again and is + sent once more. The device rejects it before running it, so this is + safe; only a ``data`` body (an upload) cannot be sent twice. + """ if expected_generation is not None: self.check_session_generation(expected_generation) generation = self._session_generation - cookies = {} - if authenticate: - if not self._token: - raise NanoKVMNotAuthenticatedError("Client is not authenticated") - cookies[_SESSION_COOKIE_NAME] = self._token - - if self._http_session is None or self._ssl_config is None: - raise RuntimeError(_NOT_STARTED) - - self.clear_session_cookies() + renew = authenticate and "data" not in kwargs request_headers = { hdrs.ACCEPT: "application/json", **kwargs.pop("headers", {}), } - async with self._http_session.request( - method, - self.url / path.lstrip("/"), - headers=request_headers, - cookies=cookies, - timeout=timeout or ClientTimeout(total=self._request_timeout), - raise_for_status=False, - ssl=self._ssl_config, - **kwargs, - ) as response: - # The explicit token owns the session. Do not leave response cookies - # available to another client sharing this HTTP session. - self.clear_session_cookies() - if authenticate: - self.check_session_generation(generation) - if authenticate and response.status == 401: - await self.clear_local_session(expected_generation=generation) - raise NanoKVMNotAuthenticatedError( - "NanoKVM session is no longer authenticated" - ) - if authenticate and response.status == 403: - raise NanoKVMPermissionError( - status=response.status, - method=response.method, - path=f"/{path.lstrip('/')}", - ) - response.raise_for_status() - yield response + while True: + cookies = {} + token = self._token if authenticate: - self.check_session_generation(generation) + if not token: + raise NanoKVMNotAuthenticatedError("Client is not authenticated") + cookies[_SESSION_COOKIE_NAME] = token + + if self._http_session is None or self._ssl_config is None: + raise RuntimeError(_NOT_STARTED) + + self.clear_session_cookies() + + async with self._http_session.request( + method, + self.url / path.lstrip("/"), + headers=request_headers, + cookies=cookies, + timeout=timeout or ClientTimeout(total=self._request_timeout), + raise_for_status=False, + ssl=self._ssl_config, + **kwargs, + ) as response: + # The explicit token owns the session. Do not leave response + # cookies available to another client sharing this HTTP session. + self.clear_session_cookies() + if authenticate: + self.check_session_generation(generation) + if authenticate and response.status == 401: + if not renew or self._credentials is None: + await self.clear_local_session(expected_generation=generation) + raise NanoKVMNotAuthenticatedError( + "NanoKVM session is no longer authenticated" + ) + else: + if authenticate and response.status == 403: + raise NanoKVMPermissionError( + status=response.status, + method=response.method, + path=f"/{path.lstrip('/')}", + ) + response.raise_for_status() + yield response + if authenticate: + self.check_session_generation(generation) + return + + # Log in again outside the rejected response, then send once more. + assert token is not None + renew = False + await self._renew_token(generation, token) async def read_json_response(self, response: ClientResponse) -> Any: """Read a JSON response and normalize decoding failures.""" @@ -430,6 +448,43 @@ async def _login(self, username: str, password: str) -> int: """Replace the session with a new login.""" # A failed identity switch must never leave the previous account usable. generation = await self.clear_local_session() + await self._request_token(username, password, generation) + return generation + + async def _renew_token(self, generation: int, rejected_token: str) -> None: + """Log in again with the kept credentials after a 401 on ``rejected_token``. + + The account stays the same, so the session keeps its generation and + operations in progress go on with the new token. Concurrent callers + share one login. Rejected credentials are dropped and end the session. + """ + async with self._renew_lock: + self.check_session_generation(generation) + if self._token != rejected_token: + return + credentials = self._credentials + if credentials is None: + await self.clear_local_session(expected_generation=generation) + raise NanoKVMNotAuthenticatedError( + "NanoKVM session is no longer authenticated" + ) + _LOGGER.debug("Session rejected; logging in again") + try: + await self._request_token( + credentials.username, credentials.password, generation + ) + except NanoKVMAuthenticationFailure: + if self._credentials is credentials: + self._credentials = None + await self.clear_local_session(expected_generation=generation) + raise + # A lost session often means the server restarted, maybe updated. + self.forget_versions() + + async def _request_token( + self, username: str, password: str, generation: int + ) -> None: + """Log in within session ``generation`` and store the new token.""" _LOGGER.debug("Attempting authentication for user: %s", username) # Unknown mode: try obfuscated first, then plain text, and remember @@ -455,7 +510,6 @@ async def _login(self, username: str, password: str) -> int: break self.check_session_generation(generation) - return generation def check_session_generation(self, generation: int) -> None: """Reject an operation superseded by an authentication transition.""" @@ -730,25 +784,24 @@ async def _send_heartbeats(self, ws: ClientWebSocketResponse) -> None: async def get_ws(self) -> ClientWebSocketResponse: """Get or create the WebSocket shared by keyboard and mouse events.""" - generation = self._session_generation - try: - async with self._ws_lock: - if self._ws is not None and not self._ws.closed: - return self._ws - - stale_ws = self._detach_ws() - if stale_ws is not None: - await stale_ws.close() - - generation = self._session_generation - ws = await self._connect_ws("ws") - self._ws = ws - self._ws_heartbeat = asyncio.create_task(self._send_heartbeats(ws)) - return ws - except ClientResponseError as err: - # _clear_local_session acquires _ws_lock, so do this after the - # lock above has been released to avoid a self-deadlock. - await self._raise_ws_error(err, "ws", generation) + return await self._handshake("ws", self._get_or_connect_ws) + + async def _get_or_connect_ws(self) -> ClientWebSocketResponse: + """Return the open shared WebSocket, or connect a new one.""" + # Clearing the session acquires _ws_lock, so _handshake maps a failed + # handshake only after this lock has been released. + async with self._ws_lock: + if self._ws is not None and not self._ws.closed: + return self._ws + + stale_ws = self._detach_ws() + if stale_ws is not None: + await stale_ws.close() + + ws = await self._connect_ws("ws") + self._ws = ws + self._ws_heartbeat = asyncio.create_task(self._send_heartbeats(ws)) + return ws async def send_ws( self, send: Callable[[ClientWebSocketResponse], Awaitable[None]] @@ -780,11 +833,29 @@ async def open_ws(self, path: str, **kwargs: Any) -> ClientWebSocketResponse: the session and rejected. The caller closes the WebSocket, and must do so before the client exits. Logging out does not close it. """ - generation = self._session_generation - try: - return await self._connect_ws(path, **kwargs) - except ClientResponseError as err: - await self._raise_ws_error(err, path, generation) + return await self._handshake(path, lambda: self._connect_ws(path, **kwargs)) + + async def _handshake( + self, path: str, connect: Callable[[], Awaitable[ClientWebSocketResponse]] + ) -> ClientWebSocketResponse: + """Run a WebSocket handshake, logging in again once after a 401.""" + renew = True + while True: + generation = self._session_generation + token = self._token + try: + return await connect() + except ClientResponseError as err: + if ( + renew + and err.status == 401 + and token + and self._credentials is not None + ): + renew = False + await self._renew_token(generation, token) + continue + await self._raise_ws_error(err, path, generation) async def _connect_ws(self, path: str, **kwargs: Any) -> ClientWebSocketResponse: """Connect a WebSocket on ``path`` carrying this session's token.""" diff --git a/tests/test_application_updates.py b/tests/test_application_updates.py index 310b3d0..c55e5b5 100644 --- a/tests/test_application_updates.py +++ b/tests/test_application_updates.py @@ -400,6 +400,8 @@ async def test_install_update_waits_for_the_restart_and_logs_in_again( mocked.get(_INFO_URL, exception=aiohttp.ClientConnectionError()) mocked.get(_INFO_URL, status=401) mocked.post(_LOGIN_URL, exception=aiohttp.ClientConnectionError()) + # The next poll still carries the rejected token. + mocked.get(_INFO_URL, status=401) mocked.post(_LOGIN_URL, payload=_ok({"token": "second-token"})) mocked.get(_INFO_URL, payload=info_payload("2.5.2", image="v1.4.2")) @@ -422,6 +424,7 @@ async def test_install_update_retries_pages_served_while_starting( mocked.get(_INFO_URL, body=page, content_type="text/html") mocked.get(_INFO_URL, status=401) mocked.post(_LOGIN_URL, body=page, content_type="text/html") + mocked.get(_INFO_URL, status=401) mocked.post(_LOGIN_URL, payload=_ok({"token": "second-token"})) mocked.get(_INFO_URL, payload=info_payload("2.5.2")) diff --git a/tests/test_kept_credentials.py b/tests/test_kept_credentials.py index c9d86c4..a04c35a 100644 --- a/tests/test_kept_credentials.py +++ b/tests/test_kept_credentials.py @@ -1,8 +1,13 @@ """Tests for credentials kept in memory to log in again.""" -from collections.abc import AsyncIterator +import asyncio +from collections.abc import AsyncGenerator, AsyncIterator +from contextlib import asynccontextmanager +from pathlib import Path +from typing import Any import aiohttp +from aiohttp import web from aioresponses import aioresponses import pytest import yarl @@ -13,12 +18,13 @@ NanoKVMNotAuthenticatedError, ) from nanokvm.components.session import _Credentials -from nanokvm.models import ApiResponseCode, HWVersion +from nanokvm.models import ApiResponseCode, GpioType, HWVersion from .common import BASE_URL, mark_detected _LOGIN_URL = f"{BASE_URL}auth/login" _HARDWARE_URL = f"{BASE_URL}vm/hardware" +_GPIO_URL = f"{BASE_URL}vm/gpio" _USERNAME = "synthetic-user" _PASSWORD = "synthetic-password" @@ -163,3 +169,228 @@ async def test_closing_the_client_drops_credentials(mocked: aioresponses) -> Non assert client._session._credentials is not None assert client._session._credentials is None + + +def _tokens(mocked: aioresponses, method: str, url: str) -> list[str]: + return [ + call.kwargs["cookies"]["nano-kvm-token"] + for call in mocked.requests.get((method, yarl.URL(url)), []) + ] + + +async def test_a_rejected_request_logs_in_again_and_is_sent_once_more( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """A lost session is renewed and the request repeated with the new token.""" + mark_detected(kept_client, HWVersion.PCIE, "2.4.3") + generation = kept_client._session._session_generation + mocked.post(_GPIO_URL, status=401) + _mock_login(mocked, "second-token") + mocked.post(_GPIO_URL, payload=_ok()) + + await kept_client.push_button(GpioType.POWER, 800) + + assert kept_client.token == "second-token" + assert _tokens(mocked, "POST", _GPIO_URL) == ["first-token", "second-token"] + calls = mocked.requests[("POST", yarl.URL(_GPIO_URL))] + assert [call.kwargs["json"] for call in calls] == [ + {"type": "power", "duration": 800} + ] * 2 + assert len(_login_bodies(mocked)) == 2 + # Same account: operations in progress are not invalidated. + assert kept_client._session._session_generation == generation + # The server may have restarted on a new version: detect it again. + assert kept_client.application_version is None + + +async def test_a_second_rejection_is_not_retried( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """A request rejected again after the login ends the session.""" + mocked.post(_GPIO_URL, status=401) + _mock_login(mocked, "second-token") + mocked.post(_GPIO_URL, status=401) + + with pytest.raises(NanoKVMNotAuthenticatedError): + await kept_client.push_button(GpioType.POWER, 800) + + assert len(_login_bodies(mocked)) == 2 + assert kept_client.token is None + + +async def test_without_kept_credentials_a_rejection_ends_the_session( + mocked: aioresponses, +) -> None: + """Clients that do not keep credentials behave as before.""" + async with NanoKVMClient(BASE_URL, token="first-token") as client: + mocked.post(_GPIO_URL, status=401) + + with pytest.raises(NanoKVMNotAuthenticatedError): + await client.push_button(GpioType.POWER, 800) + + assert client.token is None + assert not _login_bodies(mocked) + + +async def test_rejected_credentials_end_the_session_when_renewing( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """Credentials the device no longer accepts are dropped, not retried.""" + mocked.post(_GPIO_URL, status=401) + mocked.post(_LOGIN_URL, payload=_REJECTED) + + with pytest.raises(NanoKVMAuthenticationFailure): + await kept_client.push_button(GpioType.POWER, 800) + + assert kept_client._session._credentials is None + assert kept_client.token is None + + +async def test_unreachable_device_while_renewing_keeps_the_credentials( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """The next call renews the session once the device answers again.""" + mocked.post(_GPIO_URL, status=401) + mocked.post(_LOGIN_URL, exception=aiohttp.ClientConnectionError()) + + with pytest.raises(aiohttp.ClientConnectionError): + await kept_client.push_button(GpioType.POWER, 800) + assert kept_client._session._credentials is not None + + mocked.post(_GPIO_URL, status=401) + _mock_login(mocked, "second-token") + mocked.post(_GPIO_URL, payload=_ok()) + await kept_client.push_button(GpioType.POWER, 800) + + assert kept_client.token == "second-token" + + +async def test_uploads_are_not_sent_twice( + kept_client: NanoKVMClient, mocked: aioresponses, tmp_path: Path +) -> None: + """A multipart body cannot be repeated, so an upload ends the session.""" + mark_detected(kept_client, HWVersion.PCIE, "2.5.2") + package = tmp_path / "nanokvm_2.5.2.tar.gz" + package.write_bytes(b"synthetic package") + mocked.post(f"{BASE_URL}application/update/offline", status=401) + + with pytest.raises(NanoKVMNotAuthenticatedError): + await kept_client.update_application_offline(package) + + assert len(_login_bodies(mocked)) == 1 + + +async def test_a_stream_logs_in_again( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """Starting the MJPEG stream on a lost session renews it.""" + stream_url = f"{BASE_URL}stream/mjpeg" + mocked.get(stream_url, status=401) + _mock_login(mocked, "second-token") + mocked.get( + stream_url, + headers={"Content-Type": "multipart/x-mixed-replace; boundary=frame"}, + body=b"--frame\r\nContent-Type: image/jpeg\r\n\r\n\xff\xd8one\r\n--frame--\r\n", + ) + + frames = [frame async for frame in kept_client.mjpeg_stream_bytes()] + + assert frames == [b"\xff\xd8one"] + assert _tokens(mocked, "GET", stream_url) == ["first-token", "second-token"] + + +@asynccontextmanager +async def _device( + login_delay: float = 0, +) -> AsyncGenerator[tuple[NanoKVMClient, dict[str, Any]], None]: + """A logged-in client and a device accepting only ``state["valid"]`` tokens.""" + state: dict[str, Any] = {"valid": set(), "logins": []} + + def rejected(request: web.Request) -> web.Response | None: + if request.cookies.get("nano-kvm-token") in state["valid"]: + return None + return web.json_response("unauthorized", status=401) + + async def login(request: web.Request) -> web.Response: + username = (await request.json())["username"] + state["logins"].append(username) + await asyncio.sleep(login_delay) + token = f"{username}-{len(state['logins'])}" + state["valid"].add(token) + return web.json_response(_ok({"token": token})) + + async def hardware(request: web.Request) -> web.Response: + return web.json_response(_ok({"version": "PCIE"})) + + async def gpio(request: web.Request) -> web.Response: + return rejected(request) or web.json_response(_ok({"pwr": True, "hdd": False})) + + async def socket(request: web.Request) -> web.StreamResponse: + if (response := rejected(request)) is not None: + return response + ws = web.WebSocketResponse() + await ws.prepare(request) + async for _ in ws: + pass + return ws + + application = web.Application() + application.router.add_post("/api/auth/login", login) + application.router.add_get("/api/vm/hardware", hardware) + application.router.add_get("/api/vm/gpio", gpio) + application.router.add_get("/api/ws", socket) + runner = web.AppRunner(application) + await runner.setup() + try: + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + async with NanoKVMClient( + f"http://127.0.0.1:{runner.addresses[0][1]}/api/", + keep_credentials=True, + use_password_obfuscation=False, + ) as client: + await client.authenticate(_USERNAME, _PASSWORD) + yield client, state + finally: + await runner.cleanup() + + +async def test_concurrent_rejections_share_one_login() -> None: + """Calls that lose the same session wait for a single new login.""" + async with _device(login_delay=0.05) as (client, state): + state["valid"].clear() + + results = await asyncio.gather(*(client.get_gpio() for _ in range(3))) + + assert all(result.pwr for result in results) + assert state["logins"] == [_USERNAME, _USERNAME] + + +async def test_a_new_login_wins_over_a_renewal_in_progress() -> None: + """Renewing never replaces a session the caller started meanwhile.""" + async with _device(login_delay=0.1) as (client, state): + state["valid"].clear() + + renewing = asyncio.create_task(client.get_gpio()) + await asyncio.sleep(0.02) + await client.authenticate("other-user", "other-password") + + with pytest.raises(NanoKVMNotAuthenticatedError, match="Session changed"): + await renewing + assert client.token == "other-user-3" + + +@pytest.mark.parametrize("shared", [False, True]) +async def test_a_websocket_handshake_logs_in_again(shared: bool) -> None: + """The keyboard/mouse socket and open_ws() renew a lost session.""" + async with _device() as (client, state): + state["valid"].clear() + + if shared: + await client._session.get_ws() + else: + ws = await client.open_ws("ws") + await ws.close() + + assert client.token == f"{_USERNAME}-2" + assert state["logins"] == [_USERNAME, _USERNAME]