From bb8c1382fd73be1889131844c31ef59e5f180407 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Antonio=20V=C3=A1zquez=20Blanco?= Date: Fri, 16 Oct 2026 15:05:07 +0200 Subject: [PATCH] Add install_application_update and keep_credentials Add install_application_update(), which updates the application when the device is out of date and waits until it runs the new version. The update restarts the device's server and ends the session, so the client logs in again with credentials it keeps in memory when created with keep_credentials=True. Logging out, changing the password, a failed login and closing the client drop them. update_application() now waits up to 15 minutes for the install. It also treats the restart as success: the dropped connection on non-Pro firmware before 2.5.1, and the 502 on 2.5.1 and newer. --- docs/api.md | 31 +++- docs/connecting.md | 10 + nanokvm/client.py | 5 + nanokvm/components/services.py | 146 +++++++++++++-- nanokvm/components/session.py | 50 ++++- tests/test_application_updates.py | 294 +++++++++++++++++++++++++++++- tests/test_kept_credentials.py | 165 +++++++++++++++++ 7 files changed, 680 insertions(+), 21 deletions(-) create mode 100644 tests/test_kept_credentials.py diff --git a/docs/api.md b/docs/api.md index bdd2ba5..107e2ba 100644 --- a/docs/api.md +++ b/docs/api.md @@ -184,7 +184,7 @@ Autostart scripts run when the device boots. | Methods | Requires | |---|---| -| `get_application_version`, `update_application` | all | +| `get_application_version`, `update_application`, `install_application_update` | all, see below | | `get_preview_status`, `set_preview_state` | non-Pro 2.2.5 / Pro | | `update_application_offline` | non-Pro 2.3.1 | | `get_update_server`, `set_update_server` | non-Pro 2.5.1 | @@ -194,8 +194,35 @@ Autostart scripts run when the device boots. | `assistant_install`, `assistant_start` | Pro 1.1.4 | | `kvmadmin_install`, `kvmadmin_uninstall`, `kvmadmin_start`, `kvmadmin_stop`, `kvmadmin_status` | Pro 1.1.5 | +`install_application_update()` updates the application when the device is +out of date, and waits until it runs the new version, which it returns. The +update restarts the device's server and ends the session, so the client logs +in again with credentials kept by `keep_credentials=True` (see +[Connecting](connecting.md#keeping-credentials)): + +```python +async with NanoKVMClient(url, keep_credentials=True) as client: + await client.authenticate("username", "password") + version = await client.install_application_update() +``` + +- Without kept credentials it raises `NanoKVMNotAuthenticatedError` before + contacting the device. It raises `NanoKVMError` when the device cannot reach + its update server, and `TimeoutError` when the new version is not running + within `timeout` (default 10 minutes). +- `get_application_version()` returns the `current` and `latest` versions. + `latest` is empty when the device cannot reach its update server. The device + asks the server while answering, so the call waits up to 60 seconds. +- `update_application()` answers once the update is installed, which can take + minutes; it waits up to 15 minutes. It does not wait for the restart. Non-Pro + firmware before 2.5.1 restarts before it answers, so the call returns when + the connection drops, without confirming the update; the same applies to + `update_application_offline()`. `install_application_update()` confirms it. +- With `set_preview_state(True)`, the device installs preview releases. + `update_application_offline("nanokvm_2.5.1.tar.gz")` installs an official -package from a local file. +package from a local file. Pass `sha256=` to check the file first; non-Pro +2.5.1 and newer also check it on the device. ### Accounts diff --git a/docs/connecting.md b/docs/connecting.md index ff8b523..31b1273 100644 --- a/docs/connecting.md +++ b/docs/connecting.md @@ -39,6 +39,16 @@ plain text only after the device rejects the credentials. Pass `get_account()` returns the username and, when the firmware provides it, the role (older firmware returns `None`). +### Keeping credentials + +The client does not keep the password. With `keep_credentials=True`, it keeps +the username and password of a successful login in memory, so that it can log +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()`. + +### Sessions and logout + Starting another login or logging out invalidates operations and streams still running on the old session; they raise `NanoKVMNotAuthenticatedError`. On current non-Pro firmware, logging out revokes **every** session on the device, diff --git a/nanokvm/client.py b/nanokvm/client.py index dbcac99..351b964 100644 --- a/nanokvm/client.py +++ b/nanokvm/client.py @@ -82,6 +82,7 @@ def __init__( ssl_ca_cert: str | None = None, ssl_fingerprint: str | None = None, use_password_obfuscation: bool | None = None, + keep_credentials: bool = False, ) -> None: """ Initialize the NanoKVM client. @@ -103,6 +104,9 @@ def __init__( None = auto-detect (try obfuscated first, fall back to plain text). True = always use obfuscated passwords (older NanoKVM versions). False = always use plain text passwords (newer HTTPS-enabled versions). + keep_credentials: Keep the username and password of a successful + login in memory, to log in again after the device restarts, as + ``install_application_update()`` needs (default: False). """ super().__init__( SessionController( @@ -114,6 +118,7 @@ def __init__( ssl_ca_cert=ssl_ca_cert, ssl_fingerprint=ssl_fingerprint, use_password_obfuscation=use_password_obfuscation, + keep_credentials=keep_credentials, ) ) diff --git a/nanokvm/components/services.py b/nanokvm/components/services.py index 9f3ae93..ed2283e 100644 --- a/nanokvm/components/services.py +++ b/nanokvm/components/services.py @@ -3,7 +3,9 @@ from __future__ import annotations import asyncio +import contextlib import hashlib +import math from os import PathLike from pathlib import Path import re @@ -16,6 +18,11 @@ require_application_version, require_hardware, ) +from ..exceptions import ( + NanoKVMError, + NanoKVMInvalidResponseError, + NanoKVMNotAuthenticatedError, +) from ..models.common import ( GetPreviewRsp, GetTailscaleStatusRsp, @@ -40,6 +47,13 @@ _OFFLINE_UPDATE_PACKAGE_RE = re.compile(r"^nanokvm_[0-9]+\.[0-9]+\.[0-9]+\.tar\.gz$") _OFFLINE_UPDATE_TIMEOUT_SECONDS = 15 * 60 +# The device asks its update server while answering; the web UI waits 60 s. +_VERSION_CHECK_TIMEOUT_SECONDS = 60.0 +# The device downloads and installs before answering; the web UI waits 15 min. +_UPDATE_TIMEOUT_SECONDS = 15 * 60.0 +# A restarting server can drop the connection, time out, or answer with a page +# that is not the API's JSON. +_RESTARTING_ERRORS = (aiohttp.ClientError, TimeoutError, NanoKVMInvalidResponseError) def _calculate_file_sha256(file_path: Path) -> str: @@ -60,12 +74,18 @@ async def get_tailscale_status(self) -> GetTailscaleStatusRsp: response_model=GetTailscaleStatusRsp, ) - async def get_application_version(self) -> GetVersionRsp: - """Get current and latest application versions.""" + async def get_application_version( + self, *, timeout: float = _VERSION_CHECK_TIMEOUT_SECONDS + ) -> GetVersionRsp: + """Get current and latest application versions. + + ``latest`` is empty when the device cannot reach its update server. + """ return await self._session.api_request_json( hdrs.METH_GET, "/application/version", response_model=GetVersionRsp, + timeout=aiohttp.ClientTimeout(total=timeout), ) @require_application_version(non_pro="2.2.5") @@ -86,11 +106,114 @@ async def set_preview_state(self, enable: bool) -> None: data=SetPreviewReq(enable=enable), ) - async def update_application(self) -> None: - """Trigger the application update process.""" - await self._session.api_request_json(hdrs.METH_POST, "/application/update") + async def update_application( + self, *, timeout: float = _UPDATE_TIMEOUT_SECONDS + ) -> None: + """Install the latest application version. + + The device answers once the update is installed, then restarts its + server, which ends the session. Non-Pro firmware before 2.5.1 restarts + before its answer is sent, so it drops the connection instead. + """ + drops_connection = await self._restarts_before_answering() + try: + await self._session.api_request_json( + hdrs.METH_POST, + "/application/update", + timeout=aiohttp.ClientTimeout(total=timeout), + ) + except aiohttp.ServerDisconnectedError: + if not drops_connection: + raise + except aiohttp.ClientResponseError as err: + if not self._is_restart_gateway_error(err): + raise self._session.forget_versions() + async def install_application_update( + self, *, timeout: float = 10 * 60, poll_interval: float = 2.0 + ) -> str: + """Update the application if needed, and wait until the device runs it. + + Returns the application version the device runs afterwards. The + restart ends the session, so the client logs in again with the + credentials it keeps with ``keep_credentials=True``. ``timeout`` bounds + the wait for the restart, in seconds. + """ + for value in (timeout, poll_interval): + if not math.isfinite(value) or value <= 0: + raise ValueError("timeout and poll_interval must be positive seconds") + session = self._session + if session._credentials is None: + raise NanoKVMNotAuthenticatedError( + "install_application_update needs a client created with " + "keep_credentials=True that has authenticated" + ) + + versions = await self.get_application_version() + if not versions.latest: + raise NanoKVMError("The device could not reach its update server") + if _version_at_least(versions.current, versions.latest): + return versions.current + + await self.update_application() + return await self._wait_for_new_application( + versions.current, timeout, poll_interval + ) + + async def _wait_for_new_application( + self, previous: str, timeout: float, poll_interval: float + ) -> str: + """Wait until the device answers with another application version.""" + session = self._session + try: + async with asyncio.timeout(timeout): + while True: + await asyncio.sleep(poll_interval) + try: + await session.detect_versions() + except NanoKVMNotAuthenticatedError: + # The restarted server rejects the old session. Log in + # again; a server still starting is retried next time. + with contextlib.suppress(*_RESTARTING_ERRORS): + await session.reauthenticate() + continue + except _RESTARTING_ERRORS: + continue + version = session._application_version + if version is not None and version != previous: + return version + except TimeoutError: + raise TimeoutError( + f"The device did not run a new application version within " + f"{timeout:g} seconds" + ) from None + + async def _restarts_before_answering(self) -> bool: + """Whether an update restarts the server before its answer is sent. + + Before 2.5.1, the non-Pro update handler runs the restart itself, which + stops the server while the answer is still buffered. + """ + session = self._session + if await session.application_version_at_least(non_pro="2.5.1"): + return False + return session.is_hardware_family(HWFamily.NON_PRO) + + def _is_restart_gateway_error(self, err: aiohttp.ClientResponseError) -> bool: + """Whether a 502 came from the device restarting after an update. + + On non-Pro 2.5.1 and newer, the device can stop its old HTTP service + before the proxy has forwarded the successful response. Its official + UI treats that gateway response as successful too. + """ + version = self._session._application_version + return ( + err.status == 502 + and version is not None + and _version_at_least(version, "2.5.1") + ) + @require_hardware(HWFamily.NON_PRO, min_version="2.3.1") async def update_application_offline( self, @@ -122,6 +245,7 @@ async def update_application_offline( if checksum and supports_remote_checksum else {} ) + drops_connection = await self._restarts_before_answering() try: await self._session.upload_file( "/application/update/offline", @@ -129,15 +253,11 @@ async def update_application_offline( headers=headers, timeout=aiohttp.ClientTimeout(total=_OFFLINE_UPDATE_TIMEOUT_SECONDS), ) + except aiohttp.ServerDisconnectedError: + if not drops_connection: + raise except aiohttp.ClientResponseError as err: - # The device can stop its old HTTP service before the proxy has - # forwarded the successful response. Its official UI treats that - # restart-related gateway response as successful too. - if ( - err.status != 502 - or self._session._application_version is None - or not _version_at_least(self._session._application_version, "2.5.1") - ): + if not self._is_restart_gateway_error(err): raise self._session.forget_versions() diff --git a/nanokvm/components/session.py b/nanokvm/components/session.py index 035f4e4..23aa4ba 100644 --- a/nanokvm/components/session.py +++ b/nanokvm/components/session.py @@ -5,6 +5,7 @@ import asyncio from collections.abc import AsyncGenerator import contextlib +from dataclasses import dataclass, field from http.cookies import Morsel, SimpleCookie import json import logging @@ -61,6 +62,14 @@ _CURRENT_PASSWORD_MIN_NON_PRO_VERSION = (2, 5, 1) +@dataclass(frozen=True) +class _Credentials: + """Login credentials kept in memory to log in again.""" + + username: str + password: str = field(repr=False) + + class SessionController: """Own authentication, shared state, HTTP and WebSocket resources.""" @@ -75,6 +84,7 @@ def __init__( ssl_ca_cert: str | None = None, ssl_fingerprint: str | None = None, use_password_obfuscation: bool | None = None, + keep_credentials: bool = False, ) -> None: """Initialize the shared session, transport and device state.""" self.url = yarl.URL(url) @@ -94,6 +104,8 @@ def __init__( self._ssl_ca_cert = ssl_ca_cert self._ssl_fingerprint = ssl_fingerprint self._use_password_obfuscation = use_password_obfuscation + self._keep_credentials = keep_credentials + self._credentials: _Credentials | None = None self._ssl_config: ssl.SSLContext | Fingerprint | bool | None = None self._hw_version: HWVersion | None = None self._application_version: str | None = None @@ -127,6 +139,7 @@ async def enter(self) -> None: async def exit(self) -> None: """Close the client's WebSocket and owned HTTP resources.""" + self._credentials = None await self.close_ws() if self._ws_session is not None: @@ -377,7 +390,35 @@ async def do_authenticate( raise async def authenticate(self, username: str, password: str) -> int: - """Authenticate and store the session token.""" + """Authenticate and store the session token. + + With ``keep_credentials``, a successful login also keeps the + credentials for ``reauthenticate()``. + """ + self._credentials = None + generation = await self._login(username, password) + if self._keep_credentials: + self._credentials = _Credentials(username, password) + return generation + + async def reauthenticate(self) -> int: + """Log in again with the credentials kept from the last login. + + The credentials stay kept when the device cannot be reached, and are + dropped when it rejects them. + """ + credentials = self._credentials + if credentials is None: + raise NanoKVMNotAuthenticatedError("No credentials are kept to log in") + try: + return await self._login(credentials.username, credentials.password) + except NanoKVMAuthenticationFailure: + if self._credentials is credentials: + self._credentials = None + raise + + 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() _LOGGER.debug("Attempting authentication for user: %s", username) @@ -417,6 +458,7 @@ def check_session_generation(self, generation: int) -> None: async def logout(self) -> None: """Log out and clear the session token.""" + self._credentials = None generation = self._session_generation try: if self._token and self._token != "disabled": @@ -505,8 +547,12 @@ async def change_password( *, current_password: str | None = None, ) -> None: - """Change the KVM password for the authenticated account.""" + """Change the KVM password for the authenticated account. + + Kept credentials are dropped; authenticate again to keep the new ones. + """ generation = self._session_generation + self._credentials = None if await self.uses_current_password_contract(): if not current_password or not current_password.strip(): raise ValueError( diff --git a/tests/test_application_updates.py b/tests/test_application_updates.py index fb33c48..310b3d0 100644 --- a/tests/test_application_updates.py +++ b/tests/test_application_updates.py @@ -1,18 +1,63 @@ -"""Tests for offline and custom-server application updates.""" +"""Tests for application version checks and updates.""" +from collections.abc import AsyncIterator import hashlib import logging from pathlib import Path +from typing import Any import aiohttp from aioresponses import aioresponses import pytest import yarl -from nanokvm.client import NanoKVMClient, NanoKVMNotSupportedError -from nanokvm.models import GetUpdateServerRsp, HWVersion, SetUpdateServerReq +from nanokvm.client import ( + NanoKVMApiError, + NanoKVMAuthenticationFailure, + NanoKVMClient, + NanoKVMError, + NanoKVMNotAuthenticatedError, + NanoKVMNotSupportedError, +) +from nanokvm.models import ( + ApiResponseCode, + GetUpdateServerRsp, + HWVersion, + SetUpdateServerReq, +) + +from .common import BASE_URL, info_payload, mark_detected + +_LOGIN_URL = f"{BASE_URL}auth/login" +_VERSION_URL = f"{BASE_URL}application/version" +_UPDATE_URL = f"{BASE_URL}application/update" +_INFO_URL = f"{BASE_URL}vm/info" + + +def _ok(data: object = None) -> dict[str, object]: + return {"code": 0, "msg": "success", "data": data} + + +def _versions(current: str, latest: str) -> dict[str, object]: + return _ok({"current": current, "latest": latest}) + + +def _calls(mocked: aioresponses, method: str, url: str) -> list[Any]: + calls: list[Any] = mocked.requests.get((method, yarl.URL(url)), []) + return calls + -from .common import BASE_URL, mark_detected +@pytest.fixture +async def kept_client(mocked: aioresponses) -> AsyncIterator[NanoKVMClient]: + """A logged-in client on 2.4.3 that kept its credentials.""" + async with NanoKVMClient( + BASE_URL, keep_credentials=True, use_password_obfuscation=False + ) as client: + mocked.post(_LOGIN_URL, payload=_ok({"token": "first-token"})) + mocked.get(f"{BASE_URL}vm/hardware", payload=_ok({"version": "PCIE"})) + await client.authenticate("synthetic-user", "synthetic-password") + mark_detected(client, HWVersion.PCIE, "2.4.3") + yield client async def test_update_application_offline_uploads_package_and_checksum( @@ -296,3 +341,244 @@ async def test_update_server_methods_reject_unsupported_devices_before_io( await getattr(client, method_name)(*args) assert not mocked.requests + + +async def test_version_check_waits_for_the_update_server( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """The device asks its update server while answering, so allow it 60 s.""" + mocked.get(_VERSION_URL, payload=_versions("2.4.3", "2.5.2")) + mocked.get(_VERSION_URL, payload=_versions("2.4.3", "")) + + assert (await client.get_application_version()).latest == "2.5.2" + assert (await client.get_application_version(timeout=5)).latest == "" + + calls = _calls(mocked, "GET", _VERSION_URL) + assert [call.kwargs["timeout"].total for call in calls] == [60, 5] + + +async def test_update_waits_for_the_installation( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """The device installs before answering, so allow it 15 minutes.""" + mark_detected(client, application_version="2.4.3") + mocked.post(_UPDATE_URL, payload=_ok()) + mocked.post(_UPDATE_URL, payload=_ok()) + + await client.update_application() + mark_detected(client, application_version="2.4.3") + await client.update_application(timeout=30) + + calls = _calls(mocked, "POST", _UPDATE_URL) + assert [call.kwargs["timeout"].total for call in calls] == [15 * 60, 30] + assert client.application_version is None + + +@pytest.mark.parametrize(("version", "accepted"), [("2.5.1", True), ("2.5.0", False)]) +async def test_update_accepts_the_restart_502_from_2_5_1( + client: NanoKVMClient, mocked: aioresponses, version: str, accepted: bool +) -> None: + """A proxy 502 during the restart counts as success where it can happen.""" + mark_detected(client, application_version=version) + mocked.post(_UPDATE_URL, status=502) + + if accepted: + await client.update_application() + else: + with pytest.raises(aiohttp.ClientResponseError): + await client.update_application() + + +async def test_install_update_waits_for_the_restart_and_logs_in_again( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """The client follows the restart and logs in to the new server.""" + mocked.get(_VERSION_URL, payload=_versions("2.4.3", "2.5.2")) + # 2.4.3 restarts before answering; the restarted server has a new secret. + mocked.post(_UPDATE_URL, exception=aiohttp.ServerDisconnectedError()) + mocked.get(_INFO_URL, payload=info_payload("2.4.3")) + mocked.get(_INFO_URL, exception=aiohttp.ClientConnectionError()) + mocked.get(_INFO_URL, status=401) + mocked.post(_LOGIN_URL, exception=aiohttp.ClientConnectionError()) + mocked.post(_LOGIN_URL, payload=_ok({"token": "second-token"})) + mocked.get(_INFO_URL, payload=info_payload("2.5.2", image="v1.4.2")) + + version = await kept_client.install_application_update(poll_interval=0.001) + + assert version == "2.5.2" + assert kept_client.token == "second-token" + assert kept_client.application_version == "2.5.2" + assert kept_client.image_version == "v1.4.2" + assert len(_calls(mocked, "POST", _LOGIN_URL)) == 3 + + +async def test_install_update_retries_pages_served_while_starting( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """A starting server can answer with a page instead of the API's JSON.""" + page = "Starting" + mocked.get(_VERSION_URL, payload=_versions("2.4.3", "2.5.2")) + mocked.post(_UPDATE_URL, payload=_ok()) + 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.post(_LOGIN_URL, payload=_ok({"token": "second-token"})) + mocked.get(_INFO_URL, payload=info_payload("2.5.2")) + + assert await kept_client.install_application_update(poll_interval=0.001) == "2.5.2" + assert kept_client.token == "second-token" + + +async def test_install_update_does_nothing_when_up_to_date( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """An up-to-date device is not updated.""" + mocked.get(_VERSION_URL, payload=_versions("2.5.2", "2.5.2")) + + assert await kept_client.install_application_update() == "2.5.2" + assert not _calls(mocked, "POST", _UPDATE_URL) + + +async def test_install_update_needs_the_update_server( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """Without a latest version, nothing is installed.""" + mocked.get(_VERSION_URL, payload=_versions("2.4.3", "")) + + with pytest.raises(NanoKVMError, match="update server"): + await kept_client.install_application_update() + assert not _calls(mocked, "POST", _UPDATE_URL) + + +async def test_install_update_needs_kept_credentials( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Without kept credentials, the device is left untouched.""" + with pytest.raises(NanoKVMNotAuthenticatedError, match="keep_credentials"): + await client.install_application_update() + assert not mocked.requests + + +@pytest.mark.parametrize( + ("timeout", "poll_interval"), + [(0, 1), (-1, 1), (float("nan"), 1), (60, 0), (60, float("inf"))], +) +async def test_install_update_rejects_invalid_timing( + kept_client: NanoKVMClient, + mocked: aioresponses, + timeout: float, + poll_interval: float, +) -> None: + """Invalid timing is rejected before any request.""" + with pytest.raises(ValueError): + await kept_client.install_application_update( + timeout=timeout, poll_interval=poll_interval + ) + assert not _calls(mocked, "GET", _VERSION_URL) + + +async def test_install_update_reports_a_failed_update( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """A failure reported by the device is raised without waiting.""" + mocked.get(_VERSION_URL, payload=_versions("2.4.3", "2.5.2")) + mocked.post(_UPDATE_URL, payload={"code": -1, "msg": "update failed"}) + + with pytest.raises(NanoKVMApiError): + await kept_client.install_application_update() + assert not _calls(mocked, "GET", _INFO_URL) + + +async def test_install_update_stops_when_the_login_is_rejected( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """Rejected credentials after the restart end the wait at once.""" + mocked.get(_VERSION_URL, payload=_versions("2.4.3", "2.5.2")) + mocked.post(_UPDATE_URL, payload=_ok()) + mocked.get(_INFO_URL, status=401) + mocked.post( + _LOGIN_URL, + payload={ + "code": ApiResponseCode.INVALID_USERNAME_OR_PASSWORD.value, + "msg": "invalid username or password", + }, + ) + + with pytest.raises(NanoKVMAuthenticationFailure): + await kept_client.install_application_update(poll_interval=0.001) + + +async def test_install_update_times_out_when_the_version_stays( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """A device that keeps the old version ends the wait with a clear error.""" + mocked.get(_VERSION_URL, payload=_versions("2.4.3", "2.5.2")) + mocked.post(_UPDATE_URL, payload=_ok()) + mocked.get(_INFO_URL, payload=info_payload("2.4.3"), repeat=True) + + with pytest.raises(TimeoutError, match="within 0.05 seconds"): + await kept_client.install_application_update(timeout=0.05, poll_interval=0.01) + + +@pytest.mark.parametrize( + ("hw_version", "version", "accepted"), + [ + (HWVersion.PCIE, "2.4.3", True), + (HWVersion.PCIE, "2.5.0", True), + (HWVersion.PCIE, "2.5.1", False), + (HWVersion.PRO, "1.2.14", False), + ], +) +async def test_update_accepts_the_dropped_connection_before_2_5_1( + client: NanoKVMClient, + mocked: aioresponses, + hw_version: HWVersion, + version: str, + accepted: bool, +) -> None: + """Older non-Pro firmware restarts before its answer is sent.""" + mark_detected(client, hw_version, version) + mocked.post(_UPDATE_URL, exception=aiohttp.ServerDisconnectedError()) + + if accepted: + await client.update_application() + else: + with pytest.raises(aiohttp.ServerDisconnectedError): + await client.update_application() + + +async def test_update_detects_the_firmware_before_the_restart( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """The firmware is read first: the device cannot answer after the update.""" + mark_detected(client, HWVersion.PCIE, application_version=None) + mocked.get(_INFO_URL, payload=info_payload("2.4.3")) + mocked.post(_UPDATE_URL, exception=aiohttp.ServerDisconnectedError()) + + await client.update_application() + + assert len(_calls(mocked, "GET", _INFO_URL)) == 1 + + +@pytest.mark.parametrize(("version", "accepted"), [("2.4.3", True), ("2.5.1", False)]) +async def test_offline_update_accepts_the_dropped_connection_before_2_5_1( + tmp_path: Path, + client: NanoKVMClient, + mocked: aioresponses, + version: str, + accepted: bool, +) -> None: + """The offline update handler restarted early before 2.5.1 too.""" + package = tmp_path / "nanokvm_2.5.2.tar.gz" + package.write_bytes(b"synthetic update package") + mark_detected(client, HWVersion.PCIE, version) + mocked.post( + f"{BASE_URL}application/update/offline", + exception=aiohttp.ServerDisconnectedError(), + ) + + if accepted: + await client.update_application_offline(package) + else: + with pytest.raises(aiohttp.ServerDisconnectedError): + await client.update_application_offline(package) diff --git a/tests/test_kept_credentials.py b/tests/test_kept_credentials.py new file mode 100644 index 0000000..c9d86c4 --- /dev/null +++ b/tests/test_kept_credentials.py @@ -0,0 +1,165 @@ +"""Tests for credentials kept in memory to log in again.""" + +from collections.abc import AsyncIterator + +import aiohttp +from aioresponses import aioresponses +import pytest +import yarl + +from nanokvm.client import ( + NanoKVMAuthenticationFailure, + NanoKVMClient, + NanoKVMNotAuthenticatedError, +) +from nanokvm.components.session import _Credentials +from nanokvm.models import ApiResponseCode, HWVersion + +from .common import BASE_URL, mark_detected + +_LOGIN_URL = f"{BASE_URL}auth/login" +_HARDWARE_URL = f"{BASE_URL}vm/hardware" +_USERNAME = "synthetic-user" +_PASSWORD = "synthetic-password" + + +def _ok(data: object = None) -> dict[str, object]: + return {"code": 0, "msg": "success", "data": data} + + +_REJECTED = { + "code": ApiResponseCode.INVALID_USERNAME_OR_PASSWORD.value, + "msg": "invalid username or password", + "data": None, +} + + +def _mock_login(mocked: aioresponses, token: str) -> None: + mocked.post(_LOGIN_URL, payload=_ok({"token": token})) + + +def _login_bodies(mocked: aioresponses) -> list[object]: + return [ + call.kwargs["json"] + for call in mocked.requests.get(("POST", yarl.URL(_LOGIN_URL)), []) + ] + + +@pytest.fixture +async def kept_client(mocked: aioresponses) -> AsyncIterator[NanoKVMClient]: + """A client that kept the credentials of its login.""" + async with NanoKVMClient( + BASE_URL, keep_credentials=True, use_password_obfuscation=False + ) as client: + _mock_login(mocked, "first-token") + mocked.get(_HARDWARE_URL, payload=_ok({"version": "PCIE"})) + await client.authenticate(_USERNAME, _PASSWORD) + yield client + + +async def test_credentials_are_not_kept_by_default(mocked: aioresponses) -> None: + """Without the opt-in, the client never keeps the password.""" + async with NanoKVMClient(BASE_URL, use_password_obfuscation=False) as client: + _mock_login(mocked, "token") + mocked.get(_HARDWARE_URL, payload=_ok({"version": "PCIE"})) + await client.authenticate(_USERNAME, _PASSWORD) + + assert client._session._credentials is None + with pytest.raises(NanoKVMNotAuthenticatedError, match="No credentials"): + await client._session.reauthenticate() + + +async def test_reauthenticate_logs_in_with_kept_credentials( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """A kept login is replayed and the new token replaces the old one.""" + _mock_login(mocked, "second-token") + + await kept_client._session.reauthenticate() + + assert kept_client.token == "second-token" + assert _login_bodies(mocked) == [ + {"username": _USERNAME, "password": _PASSWORD}, + {"username": _USERNAME, "password": _PASSWORD}, + ] + + +def test_kept_password_is_hidden_from_repr() -> None: + """Debug output never shows the kept password.""" + text = repr(_Credentials(_USERNAME, _PASSWORD)) + + assert _USERNAME in text + assert _PASSWORD not in text + + +async def test_unreachable_device_keeps_credentials( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """A login that cannot reach the device can be retried later.""" + mocked.post(_LOGIN_URL, exception=aiohttp.ClientConnectionError()) + with pytest.raises(aiohttp.ClientConnectionError): + await kept_client._session.reauthenticate() + + _mock_login(mocked, "second-token") + await kept_client._session.reauthenticate() + assert kept_client.token == "second-token" + + +async def test_rejected_credentials_are_dropped( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """Credentials the device rejects are not tried again.""" + mocked.post(_LOGIN_URL, payload=_REJECTED) + + with pytest.raises(NanoKVMAuthenticationFailure): + await kept_client._session.reauthenticate() + + assert kept_client._session._credentials is None + + +async def test_failed_login_drops_the_previous_credentials( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """A failed switch to another account keeps neither account.""" + mocked.post(_LOGIN_URL, payload=_REJECTED) + + with pytest.raises(NanoKVMAuthenticationFailure): + await kept_client.authenticate("other-user", "wrong-password") + + assert kept_client._session._credentials is None + + +async def test_logout_drops_credentials( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """Logging out forgets the kept credentials.""" + mocked.post(f"{BASE_URL}auth/logout", payload=_ok()) + + await kept_client.logout() + + assert kept_client._session._credentials is None + + +async def test_password_change_drops_credentials( + kept_client: NanoKVMClient, mocked: aioresponses +) -> None: + """The old password is not kept after it changes.""" + mark_detected(kept_client, HWVersion.PCIE, "2.4.3") + mocked.post(f"{BASE_URL}auth/password", payload=_ok()) + + await kept_client.change_password(_USERNAME, "new-synthetic-password") + + assert kept_client._session._credentials is None + + +async def test_closing_the_client_drops_credentials(mocked: aioresponses) -> None: + """Credentials do not outlive the client.""" + async with NanoKVMClient( + BASE_URL, keep_credentials=True, use_password_obfuscation=False + ) as client: + _mock_login(mocked, "token") + mocked.get(_HARDWARE_URL, payload=_ok({"version": "PCIE"})) + await client.authenticate(_USERNAME, _PASSWORD) + assert client._session._credentials is not None + + assert client._session._credentials is None