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]