diff --git a/docs/api.md b/docs/api.md index b2ca85e..8eb86be 100644 --- a/docs/api.md +++ b/docs/api.md @@ -160,7 +160,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 | @@ -170,8 +170,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..135b8ad 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,7 @@ require_application_version, require_hardware, ) +from ..exceptions import NanoKVMError, NanoKVMNotAuthenticatedError from ..models.common import ( GetPreviewRsp, GetTailscaleStatusRsp, @@ -40,6 +43,10 @@ _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 def _calculate_file_sha256(file_path: Path) -> str: @@ -60,12 +67,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 +99,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(aiohttp.ClientError, TimeoutError): + await session.reauthenticate() + continue + except (aiohttp.ClientError, TimeoutError): + continue # Restarting. + 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 +238,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 +246,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..f3bbd8d 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,227 @@ 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_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