diff --git a/.gitignore b/.gitignore index 74dec1c..14c2cab 100644 --- a/.gitignore +++ b/.gitignore @@ -73,4 +73,5 @@ ENV/ .vscode .DS_Store -priv/ \ No newline at end of file +priv/ +uv.lock diff --git a/README.md b/README.md index dff25ef..fc0f05a 100644 --- a/README.md +++ b/README.md @@ -1,222 +1,45 @@ # python-nanokvm -Async Python client for [NanoKVM](https://github.com/sipeed/NanoKVM). +Async Python client for the [Sipeed NanoKVM](https://github.com/sipeed/NanoKVM) +and [NanoKVM Pro](https://github.com/sipeed/NanoKVM-Pro). -## Basic usage - -```python -from nanokvm.client import NanoKVMClient -from nanokvm.models import GpioType, MouseButton - -async with NanoKVMClient("https://kvm.local/api/") as client: - await client.authenticate("username", "password") - - info = await client.get_info() - hardware = await client.get_hardware() - gpio = await client.get_gpio() - - await client.paste_text("Hello\nworld!") - await client.mouse_click(MouseButton.LEFT, 0.5, 0.5) - await client.mouse_move_abs(0.25, 0.75) - await client.mouse_scroll(0, -3) - await client.push_button(GpioType.POWER, duration_ms=1000) - - async for frame in client.mjpeg_stream(): - print(frame) -``` - -## Authentication and permissions - -Authentication tries the obfuscated password format first, then plain text only -after the credentials are rejected. Pass `use_password_obfuscation=False` or -`True` to force either mode. - -`get_account()` returns the username and, when provided by the firmware, its -role. Older firmware may return `None` for `role`. Permission failures raise -`NanoKVMPermissionError` without ending the session: - -```python -from nanokvm.client import NanoKVMPermissionError - -account = await client.get_account() -try: - await client.get_images() -except NanoKVMPermissionError as error: - print(error.status, error.method, error.path) -``` - -Non-Pro firmware 2.5.1 and newer has user accounts with the roles `admin` and -`user`. An admin can manage them: - -```python -from nanokvm.models import UserRole - -await client.create_user("guest", "password", UserRole.USER) -users = await client.get_users() -await client.update_user("guest", enabled=False) -await client.set_user_password("guest", "new-password") -await client.delete_user("guest") +```bash +pip install nanokvm ``` -Updating, deleting or resetting the password of a user ends that user's -sessions. The device owner account cannot be disabled, demoted or deleted. - -On non-Pro firmware 2.5.1 and newer, changing a password requires the current -password and the username must match the authenticated account: +## Usage ```python -await client.change_password( - "username", - "new-password", - current_password="current-password", -) -``` - -Older non-Pro firmware and NanoKVM Pro use the original two-argument call. - -A successful password change clears the local session. Starting another login -or logging out invalidates in-flight authenticated operations and streams. -Superseded operations raise `NanoKVMNotAuthenticatedError`. - -On current non-Pro firmware, logging out revokes every session on the device, -not only the caller's. Other clients and open web UI tabs have to log in again. - -When an external `aiohttp.ClientSession` is supplied, it remains owned by the -caller. WebSockets retain its connector, headers, proxy cookies, basic auth, -proxy settings, and tracing, while isolating the device session cookie. Custom -session subclasses and middleware apply only to HTTP requests. - -`NanoKVMApiError` includes the numeric API code. Its original `msg` and `data` -remain available as attributes but are omitted from automatic diagnostics. - -## Hardware and firmware support - -Each method knows which hardware and application version it needs. The client -detects both and raises `NanoKVMNotSupportedError` before sending a request the -device cannot serve: - -```python -from nanokvm.client import NanoKVMNotSupportedError - -try: - await client.get_swap_size() -except NanoKVMNotSupportedError as error: - print(error) # get_swap_size requires hardware: non-Pro (detected: Pro) -``` - -Hardware is detected at login, or on the first call when the client was created -from a stored `token`. Application and image versions are detected on the first -version-checked call and kept until the session resets (logout, a new login or -an expired session). After a firmware update on a client that stays logged in, -call `await client.detect_versions()` to refresh them. Version strings that are -not plain numbers, such as development builds, are treated as supported. +import asyncio -The HDMI controls (`get_hdmi_state`, `reset_hdmi`, `enable_hdmi`, -`disable_hdmi`, `set_hdmi_idle_timeout`) are available on PCIe hardware only. - -## Images - -`upload_image` sends a local image to the device and reports progress: - -```python -def on_progress(progress): - print(f"{progress.percentage:.0f}%") - -await client.upload_image("debian.iso", progress_callback=on_progress) -await client.mount_image("/data/debian.iso", cdrom=True) -``` - -- Non-Pro devices need application 2.3.1 or newer and accept `.iso` files with - ASCII names. Pass `sha256=` (2.5.0 or newer) to have the device verify the - upload. -- NanoKVM Pro accepts `.iso` and `.img` files. -- An existing image with the same name raises `FileExistsError` unless - `overwrite=True`. On NanoKVM Pro, overwriting deletes the existing image - before the upload starts, so a failed upload leaves neither copy. - -The device can also download an image itself: - -```python -await client.download_image("https://example.com/debian.iso") -async for status in client.watch_image_download(): - print(status.status, status.percentage) -``` - -`download_image` accepts `sha256=` and `cancel_image_download()` stops a -download; both need a non-Pro device on 2.5.0 or newer. - -## Other non-Pro features - -- **Application updates:** `update_application_offline("nanokvm_2.5.1.tar.gz")` - installs an official package from a local file (2.3.1 or newer); - `get_update_server` and `set_update_server` manage a custom update server - (2.5.1 or newer). -- **MCP and coordinated control** (2.5.0 or newer): `get_mcp_config`, - `set_mcp_enabled`, `regenerate_mcp_api_key`, `get_ai_control_status` and - `set_ai_control_mode`. -- **Input region** (2.5.1 or newer): `get_input_region`, `set_input_region` and - `get_input_resolution`. - -## SSH - -SSH host-key verification is strict by default. Add the device key to the -system known-hosts file or pass a separate file with `known_hosts`. Password -authentication does not use an SSH agent or private-key discovery unless -`allow_agent=True` or `look_for_keys=True` is set. -Pass `connect_timeout`, `banner_timeout`, or `auth_timeout` to adjust the -connection phases. - -```python -from pathlib import Path - -from nanokvm.ssh_client import NanoKVMSSH - -async with NanoKVMSSH( - "kvm.local", - known_hosts=Path.home() / ".ssh" / "known_hosts", -) as ssh: - await ssh.authenticate("password") - - uptime = await ssh.run_command("cat /proc/uptime") - disk = await ssh.run_command("df -h /") -``` - -Set `allow_unknown_host_key=True` only for controlled testing; it disables -host-key verification. Authentication failures raise -`NanoKVMSSHAuthenticationError`, connection and host-key failures raise -`NanoKVMSSHConnectionError`, and command failures raise -`NanoKVMSSHCommandError`. +from nanokvm.client import NanoKVMClient +from nanokvm.models import GpioType, MouseButton -## HTTPS -Certificates issued by a public or locally trusted CA work without extra -configuration: +async def main() -> None: + async with NanoKVMClient("https://kvm.local/api/") as client: + await client.authenticate("username", "password") -```python -client = NanoKVMClient("https://kvm.local/api/") -``` + await client.push_button(GpioType.POWER, duration_ms=1000) + await client.paste_text("Hello\nworld!") + await client.mouse_click(MouseButton.LEFT, 0.5, 0.5) -For a self-signed certificate, certificate pinning is recommended: + async for frame in client.mjpeg_stream(): + frame.save("screen.jpg") + break -```python -from nanokvm.utils import async_fetch_remote_fingerprint -fingerprint = await async_fetch_remote_fingerprint("https://kvm.local/api/") -client = NanoKVMClient( - "https://kvm.local/api/", - ssl_fingerprint=fingerprint, -) +asyncio.run(main()) ``` -A custom CA certificate can be supplied instead: - -```python -client = NanoKVMClient("https://kvm.local/api/", ssl_ca_cert="/path/to/ca.pem") -``` +The client detects the hardware and firmware, and raises +`NanoKVMNotSupportedError` instead of calling an endpoint the device lacks. -Certificate verification can be disabled for testing, but this does not verify -the device identity and should not be used in production: +## Documentation -```python -client = NanoKVMClient("https://kvm.local/api/", verify_ssl=False) -``` +- [Connecting and authentication](docs/connecting.md): HTTPS and self-signed + certificates, sessions, passwords and user accounts +- [API reference](docs/api.md): every method and the hardware and firmware it + needs +- [Errors](docs/errors.md) +- [SSH](docs/ssh.md) diff --git a/docs/api.md b/docs/api.md new file mode 100644 index 0000000..b2ca85e --- /dev/null +++ b/docs/api.md @@ -0,0 +1,181 @@ +# API reference + +## Hardware and firmware support + +There are two hardware families: + +- **non-Pro:** the original NanoKVM (hardware Alpha and Beta) and the NanoKVM PCIe +- **Pro:** NanoKVM Pro + +Many endpoints exist on only one family or need a minimum application version. +Each method declares what it needs. The client checks it and raises +`NanoKVMNotSupportedError` without contacting the device: + +```python +from nanokvm.client import NanoKVMNotSupportedError + +try: + await client.get_swap_size() +except NanoKVMNotSupportedError as error: + print(error) # get_swap_size requires hardware: non-Pro (detected: Pro) +``` + +The hardware is detected at login, or on the first call when the client +starts from a stored `token`. Firmware versions are detected on the first call +that needs them, and kept until the session resets. After updating the firmware +on a client that stays logged in, call `await client.detect_versions()`. +Version strings that are not plain numbers, such as development builds, count +as supported. + +`client.hw_version`, `client.application_version` and `client.image_version` +expose what was detected. + +## Methods + +Every method below is a coroutine on `NanoKVMClient`. The requirements column +uses these abbreviations: + +- **all:** every device +- **non-Pro 2.3.1:** non-Pro with application 2.3.1 or newer, and not the Pro +- **non-Pro 2.2.6 / Pro:** non-Pro 2.2.6 or newer, or any Pro +- **PCIe:** NanoKVM PCIe only + +Request and response models live in `nanokvm.models`. + +### Device and system + +| Methods | Requires | +|---|---| +| `get_info`, `get_hardware`, `get_gpio`, `push_button` | all | +| `get_oled_info`, `set_oled_sleep` | all | +| `get_virtual_device_status`, `update_virtual_device` | all | +| `get_hostname`, `set_hostname`, `get_web_title`, `set_web_title` | non-Pro 2.2.6 / Pro | +| `get_ssh_state`, `enable_ssh`, `disable_ssh` | non-Pro 2.1.6 / Pro | +| `get_mdns_state`, `enable_mdns`, `disable_mdns` | non-Pro 2.2.2 / Pro | +| `reboot_system` | non-Pro 2.2.2 / Pro | +| `set_tls` | non-Pro 2.2.7 | +| `get_swap_size`, `set_swap_size` | non-Pro 2.2.6 | +| `get_memory_limit`, `set_memory_limit` | non-Pro | +| `switch_to_pikvm`, `get_led_strip`, `set_led_strip` | Pro | +| `get_lcd_time_format`, `set_lcd_time_format` | Pro 1.1.6 | +| `get_timezone`, `set_timezone`, `get_time_status`, `sync_time` | Pro 1.1.6 | +| `get_time` | Pro 1.2.6 | +| `get_low_power`, `set_low_power`, `refresh_virtual_device` | Pro 1.2.7 | +| `get_menubar_config`, `set_menubar_config` | Pro 1.2.10 | + +`set_tls` restarts the device's web server, which then answers only on the +other scheme (`http` or `https`), within about half a minute. + +### Keyboard and mouse + +| Methods | Requires | +|---|---| +| `paste_text` | all | +| `mouse_move_abs`, `mouse_move_rel`, `mouse_down`, `mouse_up`, `mouse_click`, `mouse_scroll` | all | +| `get_hid_mode`, `set_hid_mode` | non-Pro 2.2.5 / Pro | +| `reset_hid` | non-Pro / Pro 1.1.6 | +| `get_mouse_jiggler_state`, `set_mouse_jiggler_state` | non-Pro 2.2.6 / Pro | +| `get_shortcuts`, `add_shortcut`, `delete_shortcut` | non-Pro 2.3.2 / Pro 1.2.8 | +| `get_leader_key`, `set_leader_key` | non-Pro 2.3.4 / Pro 1.2.12 | +| `get_keyboard_led_status` | non-Pro 2.5.0 | + +`paste_text` accepts only the characters in `nanokvm.client.PASTE_CHAR_MAP`. + +### Video and streaming + +| Methods | Requires | +|---|---| +| `mjpeg_stream`, `mjpeg_stream_bytes`, `open_ws` | all | +| `set_screen`, `set_frame_detect`, `pause_frame_detect` | non-Pro | +| `get_input_region`, `set_input_region`, `get_input_resolution` | non-Pro 2.5.1 | +| `reset_hdmi` | PCIe 2.1.5 | +| `get_hdmi_state`, `enable_hdmi`, `disable_hdmi` | PCIe 2.2.8 | +| `set_hdmi_idle_timeout` | PCIe 2.5.0 | +| `set_stream_mode`, `set_stream_quality`, `set_gop` | Pro | +| `get_hdmi_capture`, `set_hdmi_capture`, `get_hdmi_passthrough`, `set_hdmi_passthrough` | Pro | +| `get_edid`, `switch_edid` | Pro | +| `get_custom_edid_list`, `upload_edid`, `delete_edid` | Pro 1.2.3 | +| `set_rate_control_mode` | Pro 1.2.6 | +| `set_fps` | Pro 1.2.8 | + +`mjpeg_stream()` yields decoded `PIL.Image` frames; `mjpeg_stream_bytes()` +yields the raw JPEG bytes. `open_ws(path)` opens a new WebSocket on a path +relative to the API base, such as `stream/h264/webrtc`, for protocols the +client does not wrap. The caller closes it before the client exits. + +### Images and scripts + +| Methods | Requires | +|---|---| +| `get_images`, `get_mounted_image`, `mount_image`, `upload_image` | all, see below | +| `delete_image` | non-Pro 2.3.0 / Pro 1.1.6 | +| `download_image`, `get_image_download_status`, `is_image_download_enabled`, `watch_image_download` | non-Pro 2.1.6 / Pro | +| `cancel_image_download` | non-Pro 2.5.0 | +| `get_cdrom_status` | non-Pro 2.2.1 | +| `get_scripts`, `upload_script`, `run_script`, `delete_script` | all | +| `get_autostart_scripts`, `get_autostart_script`, `upload_autostart_script`, `delete_autostart_script` | non-Pro 2.3.1 | + +Upload an image with progress reporting, then mount it: + +```python +def on_progress(progress): + print(f"{progress.percentage:.0f}%") + +await client.upload_image("debian.iso", progress_callback=on_progress) +await client.mount_image("/data/debian.iso", cdrom=True) +``` + +- Non-Pro devices need application 2.3.1 or newer and accept `.iso` files with + ASCII names. On 2.5.0 or newer, pass `sha256=` to have the device verify the + upload. +- The Pro accepts `.iso` and `.img` files. +- An existing image with the same name raises `FileExistsError` unless + `overwrite=True`. On the Pro, overwriting deletes the old image before the + upload starts, so a failed upload leaves neither copy. + +Or have the device download the image itself (`sha256=` needs non-Pro 2.5.0): + +```python +await client.download_image("https://example.com/debian.iso") +async for status in client.watch_image_download(): + print(status.status, status.percentage) +``` + +Autostart scripts run when the device boots. + +### Network + +| Methods | Requires | +|---|---| +| `get_wifi_status`, `connect_wifi_no_auth` | all | +| `send_wake_on_lan`, `get_wol_macs`, `delete_wol_mac` | all | +| `set_wol_mac_name` | non-Pro 2.2.6 / Pro | +| `connect_wifi`, `disconnect_wifi` | non-Pro 2.3.1 / Pro | +| `verify_ap_login` | non-Pro 2.3.6 / Pro 1.2.14 | +| `get_dns`, `set_dns` | non-Pro 2.4.1 | +| `scan_wifi` | Pro 1.1.7 | +| `get_static_ip`, `set_static_ip` | Pro 1.2.3 | + +### Updates and extensions + +| Methods | Requires | +|---|---| +| `get_application_version`, `update_application` | all | +| `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 | +| `get_tailscale_status`, `tailscale_install`, `tailscale_uninstall`, `tailscale_up`, `tailscale_down`, `tailscale_login`, `tailscale_logout`, `tailscale_start`, `tailscale_stop`, `tailscale_restart` | non-Pro 2.1.6 / Pro | +| `get_mcp_config`, `set_mcp_enabled`, `regenerate_mcp_api_key` | non-Pro 2.5.0 | +| `get_ai_control_status`, `set_ai_control_mode` | non-Pro 2.5.0 | +| `assistant_install`, `assistant_start` | Pro 1.1.4 | +| `kvmadmin_install`, `kvmadmin_uninstall`, `kvmadmin_start`, `kvmadmin_stop`, `kvmadmin_status` | Pro 1.1.5 | + +`update_application_offline("nanokvm_2.5.1.tar.gz")` installs an official +package from a local file. + +### Accounts + +| Methods | Requires | +|---|---| +| `authenticate`, `logout`, `get_account`, `change_password`, `is_password_updated` | all | +| `get_users`, `create_user`, `update_user`, `delete_user`, `set_user_password` | non-Pro 2.5.1 | diff --git a/docs/connecting.md b/docs/connecting.md new file mode 100644 index 0000000..ff8b523 --- /dev/null +++ b/docs/connecting.md @@ -0,0 +1,77 @@ +# Connecting and authentication + +The client URL points at the device's `/api/` path, such as +`https://kvm.local/api/`. Mouse coordinates are fractions of the screen, from +`0.0` to `1.0`. + +## HTTPS + +Certificates from a public or locally trusted CA work as is. For the device's +self-signed certificate, pin its fingerprint: + +```python +from nanokvm.utils import async_fetch_remote_fingerprint + +fingerprint = await async_fetch_remote_fingerprint("https://kvm.local/api/") +client = NanoKVMClient("https://kvm.local/api/", ssl_fingerprint=fingerprint) +``` + +Or trust a custom CA with `ssl_ca_cert="/path/to/ca.pem"`. `verify_ssl=False` +turns verification off; it does not check the device identity, so use it only +for testing. + +## Sessions + +- **Stored token:** pass `token=` to reuse a session instead of logging in. + `client.token` holds the current one. +- **Your own `aiohttp.ClientSession`:** pass `session=`. It stays yours to close. + WebSockets reuse its connector, headers, proxy settings, cookies, basic auth + and tracing, while the device session cookie stays isolated. Session + subclasses and middleware apply to HTTP requests only. +- **Timeout:** `request_timeout` (seconds, default 10) applies to each request. + +## Authentication and accounts + +`authenticate()` tries the obfuscated password format first and falls back to +plain text only after the device rejects the credentials. Pass +`use_password_obfuscation=True` or `False` to force one. + +`get_account()` returns the username and, when the firmware provides it, the +role (older firmware returns `None`). + +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, +including open web UI tabs. + +### Changing the password + +```python +await client.change_password( + "username", + "new-password", + current_password="current-password", +) +``` + +Non-Pro firmware 2.5.1 and newer requires `current_password`, and the username +must match the logged-in account. Older non-Pro firmware and the Pro take only +the username and new password. A successful change clears the local session. + +### User accounts + +Non-Pro firmware 2.5.1 and newer has user accounts with the roles `admin` and +`user`. An admin can manage them: + +```python +from nanokvm.models import UserRole + +await client.create_user("guest", "password", UserRole.USER) +users = await client.get_users() +await client.update_user("guest", enabled=False) +await client.set_user_password("guest", "new-password") +await client.delete_user("guest") +``` + +Updating, deleting or resetting the password of a user ends that user's +sessions. The device owner cannot be disabled, demoted or deleted. diff --git a/docs/errors.md b/docs/errors.md new file mode 100644 index 0000000..f6b1030 --- /dev/null +++ b/docs/errors.md @@ -0,0 +1,17 @@ +# Errors + +All exceptions derive from `NanoKVMError` and are importable from +`nanokvm.client`. + +| Exception | Raised when | +|---|---| +| `NanoKVMNotSupportedError` | The detected hardware or firmware lacks the endpoint; nothing was sent | +| `NanoKVMAuthenticationFailure` | Login was rejected | +| `NanoKVMNotAuthenticatedError` | There is no session, or a newer login or logout replaced it | +| `NanoKVMPermissionError` | The account lacks permission; the session stays valid | +| `NanoKVMApiError` | The device answered with an error code | +| `NanoKVMInvalidResponseError` | The response could not be parsed | + +`NanoKVMPermissionError` carries `status`, `method` and `path`. +`NanoKVMApiError` carries the numeric `code`; the device's `msg` and `data` are +available as attributes but left out of the exception message. diff --git a/docs/ssh.md b/docs/ssh.md new file mode 100644 index 0000000..8dfd0da --- /dev/null +++ b/docs/ssh.md @@ -0,0 +1,26 @@ +# SSH + +`NanoKVMSSH` runs commands on the device over SSH: + +```python +from pathlib import Path + +from nanokvm.ssh_client import NanoKVMSSH + +async with NanoKVMSSH( + "kvm.local", + known_hosts=Path.home() / ".ssh" / "known_hosts", +) as ssh: + await ssh.authenticate("password") + uptime = await ssh.run_command("cat /proc/uptime") +``` + +- Host keys are verified strictly: add the device key to the system + known-hosts file or pass `known_hosts=`. `allow_unknown_host_key=True` + disables the check; use it only for controlled testing. +- Password login does not use an SSH agent or look for private keys unless you + pass `allow_agent=True` or `look_for_keys=True`. +- `connect_timeout`, `banner_timeout` and `auth_timeout` tune each connection + phase. +- Failures raise `NanoKVMSSHAuthenticationError`, `NanoKVMSSHConnectionError` + (including host-key errors) or `NanoKVMSSHCommandError`. diff --git a/nanokvm/components/storage.py b/nanokvm/components/storage.py index 8d66332..ec0b7cb 100644 --- a/nanokvm/components/storage.py +++ b/nanokvm/components/storage.py @@ -11,13 +11,15 @@ import re import stat from typing import Any +from urllib.parse import quote import aiohttp from aiohttp import hdrs from aiohttp.payload import Payload +from pydantic import RootModel from ..compatibility import require_application_version, require_hardware -from ..exceptions import NanoKVMNotSupportedError +from ..exceptions import NanoKVMApiError, NanoKVMNotSupportedError from ..models.common import ( DeleteImageReq, DeleteScriptReq, @@ -36,7 +38,7 @@ StatusImageRsp, UploadScriptRsp, ) -from ..models.non_pro import GetCdRomRsp +from ..models.non_pro import GetAutostartRsp, GetCdRomRsp, UploadAutostartReq from ..utils import _validate_sha256 from .session import Controller @@ -110,6 +112,17 @@ async def write(self, writer: Any) -> None: raise OSError("image changed while it was being uploaded") +class _AutostartContent(RootModel[str]): + pass + + +def _autostart_path(name: str) -> str: + """Return the route for one autostart script, rejecting path components.""" + if not name or name in {".", ".."} or "/" in name: + raise ValueError(f"Invalid autostart script name: {name!r}") + return f"/vm/autostart/{quote(name, safe='')}" + + class StorageController(Controller): """Implement storage operations behind the public client facade.""" @@ -146,6 +159,46 @@ async def delete_script(self, name: str) -> None: data=DeleteScriptReq(name=name), ) + @require_hardware(HWFamily.NON_PRO, min_version="2.3.1") + async def get_autostart_scripts(self) -> GetAutostartRsp: + """List the scripts the device runs at boot.""" + try: + return await self._session.api_request_json( + hdrs.METH_GET, + "/vm/autostart", + response_model=GetAutostartRsp, + ) + except NanoKVMApiError as err: + # The firmware creates the directory on first upload and reports + # the failed walk of a missing one as code -1. + if err.code != -1: + raise + return GetAutostartRsp() + + @require_hardware(HWFamily.NON_PRO, min_version="2.3.1") + async def get_autostart_script(self, name: str) -> str: + """Get the content of an autostart script.""" + content = await self._session.api_request_json( + hdrs.METH_GET, + _autostart_path(name), + response_model=_AutostartContent, + ) + return content.root + + @require_hardware(HWFamily.NON_PRO, min_version="2.3.1") + async def upload_autostart_script(self, name: str, content: str) -> None: + """Create or replace an autostart script.""" + await self._session.api_request_json( + hdrs.METH_POST, + _autostart_path(name), + data=UploadAutostartReq(content=content), + ) + + @require_hardware(HWFamily.NON_PRO, min_version="2.3.1") + async def delete_autostart_script(self, name: str) -> None: + """Delete an autostart script.""" + await self._session.api_request_json(hdrs.METH_DELETE, _autostart_path(name)) + async def get_images(self) -> GetImagesRsp: """Get the list of available image files.""" return await self._session.api_request_json( diff --git a/nanokvm/components/system.py b/nanokvm/components/system.py index 3117228..fec126f 100644 --- a/nanokvm/components/system.py +++ b/nanokvm/components/system.py @@ -2,10 +2,13 @@ from __future__ import annotations -from aiohttp import hdrs +import contextlib +from datetime import UTC, datetime + +from aiohttp import ServerDisconnectedError, hdrs from ..compatibility import require_application_version, require_hardware -from ..exceptions import NanoKVMApiError +from ..exceptions import NanoKVMApiError, NanoKVMInvalidResponseError from ..models.common import ( GetGpioRsp, GetHardwareRsp, @@ -30,6 +33,7 @@ GetSwapSizeRsp, SetMemoryLimitReq, SetSwapSizeReq, + SetTlsReq, ) from ..models.pro import ( DiskType, @@ -135,6 +139,23 @@ async def disable_mdns(self) -> None: """Disable mDNS.""" await self._session.api_request_json(hdrs.METH_POST, "/vm/mdns/disable") + @require_hardware(HWFamily.NON_PRO, min_version="2.2.7") + async def set_tls(self, enabled: bool) -> None: + """Serve the web interface over HTTPS or plain HTTP. + + Enabling creates a self-signed certificate. The device restarts its + web server, so this client's URL stops working; reach the device again + on the other scheme once it is back, which takes about half a minute. + """ + # The firmware restarts the server before its answer goes out; every + # failure before that point answers with an error code. + with contextlib.suppress(ServerDisconnectedError): + await self._session.api_request_json( + hdrs.METH_POST, + "/vm/tls", + data=SetTlsReq(enabled=enabled), + ) + async def get_oled_info(self) -> GetOLEDRsp: """Get OLED information.""" return await self._session.api_request_json( @@ -359,6 +380,21 @@ async def get_time_status(self) -> GetTimeStatusRsp: response_model=GetTimeStatusRsp, ) + @require_hardware(HWFamily.PRO, min_version="1.2.6") + async def get_time(self) -> datetime: + """Get the device clock, as an aware UTC datetime.""" + # The endpoint is an endless event stream of epoch milliseconds, one + # per second; the first event is enough. + async with self._session.request(hdrs.METH_GET, "/vm/time") as response: + async for line in response.content: + if line.startswith(b"data:"): + try: + millis = int(line[5:]) + except ValueError: + break + return datetime.fromtimestamp(millis / 1000, UTC) + raise NanoKVMInvalidResponseError("Time stream ended without a timestamp") + @require_hardware(HWFamily.PRO, min_version="1.1.6") async def sync_time(self) -> None: """Synchronize time.""" diff --git a/nanokvm/components/video.py b/nanokvm/components/video.py index 8a1bb9c..6754c02 100644 --- a/nanokvm/components/video.py +++ b/nanokvm/components/video.py @@ -4,7 +4,7 @@ from os import PathLike -from aiohttp import hdrs +from aiohttp import ClientTimeout, hdrs from ..compatibility import require_hardware from ..models.common import HWFamily, HWVersion @@ -16,9 +16,11 @@ ManualRegion, OriginalResolution, ScreenSettingType, + SetFrameDetectReq, SetHdmiIdleTimeoutReq, SetInputRegionReq, SetScreenReq, + StopFrameDetectReq, ) from ..models.pro import ( DeleteEdidReq, @@ -224,6 +226,31 @@ async def delete_edid(self, edid: str) -> None: data=DeleteEdidReq(edid=edid), ) + @require_hardware(HWFamily.NON_PRO) + async def set_frame_detect(self, enabled: bool) -> None: + """Skip MJPEG frames that did not change, or send every frame.""" + await self._session.api_request_json( + hdrs.METH_POST, + "/stream/mjpeg/detect", + data=SetFrameDetectReq(enabled=enabled), + ) + + @require_hardware(HWFamily.NON_PRO) + async def pause_frame_detect(self, duration: int = 10) -> None: + """Send every MJPEG frame for ``duration`` seconds, then skip again. + + The device answers only once the pause is over, and turns frame + detection on at its end even if it was off before. + """ + if duration < 1: + raise ValueError("duration must be at least 1 second") + await self._session.api_request_json( + hdrs.METH_POST, + "/stream/mjpeg/detect/stop", + data=StopFrameDetectReq(duration=duration), + timeout=ClientTimeout(total=duration + self._session._request_timeout), + ) + @require_hardware(HWFamily.PRO, min_version="1.2.6") async def set_rate_control_mode(self, mode: RateControlMode) -> None: """Set the stream rate control mode (CBR/VBR).""" diff --git a/nanokvm/models/non_pro.py b/nanokvm/models/non_pro.py index bbb59d2..e930df3 100644 --- a/nanokvm/models/non_pro.py +++ b/nanokvm/models/non_pro.py @@ -14,6 +14,7 @@ "CreateUserReq", "DNSInfo", "DNSMode", + "GetAutostartRsp", "GetCdRomRsp", "GetDNSRsp", "GetHdmiStateRsp", @@ -31,15 +32,19 @@ "SetAIControlModeReq", "SetAIControlModeRsp", "SetDNSReq", + "SetFrameDetectReq", "SetHdmiIdleTimeoutReq", "SetInputRegionReq", "SetMCPConfigReq", "SetMemoryLimitReq", "SetScreenReq", "SetSwapSizeReq", + "SetTlsReq", "SetUpdateServerReq", "SetUserPasswordReq", + "StopFrameDetectReq", "UpdateUserReq", + "UploadAutostartReq", "UserInfo", "UserRole", ] @@ -89,6 +94,18 @@ class SetScreenReq(BaseModel): value: int +class SetTlsReq(BaseModel): + enabled: bool + + +class SetFrameDetectReq(BaseModel): + enabled: bool + + +class StopFrameDetectReq(BaseModel): + duration: int + + class GetMCPConfigRsp(BaseModel): """MCP configuration and current coordinated-control state.""" @@ -222,6 +239,19 @@ class SetSwapSizeReq(BaseModel): size: int +class GetAutostartRsp(BaseModel): + files: list[str] = Field(default_factory=list) + + @field_validator("files", mode="before") + @classmethod + def _normalize_files(cls, value: Any) -> Any: + return [] if value is None else value + + +class UploadAutostartReq(BaseModel): + content: str + + class GetUpdateServerRsp(BaseModel): enabled: bool url: str = Field(repr=False) diff --git a/tests/test_storage.py b/tests/test_storage.py index fd5ce8b..5694cf3 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -150,3 +150,29 @@ async def test_mount_image_read_only_requires_pro_1_2_1( with pytest.raises(NanoKVMNotSupportedError, match="1.2.1"): await client.mount_image("/data/test.img", read_only=True) assert not mocked.requests + + +async def test_autostart_scripts(client: NanoKVMClient, mocked: aioresponses) -> None: + """Autostart treats a missing directory as empty and guards script names.""" + mark_detected(client, HWVersion.PCIE, "2.5.2") + mocked.get( + f"{BASE_URL}vm/autostart", + payload={"code": -1, "msg": "get autostart directory fail", "data": None}, + ) + mocked.get( + f"{BASE_URL}vm/autostart/boot.sh", + payload={"code": 0, "msg": "success", "data": "#!/bin/sh\n"}, + ) + mocked.post( + f"{BASE_URL}vm/autostart/boot.sh", + payload={"code": 0, "msg": "success", "data": "boot.sh"}, + ) + + assert (await client.get_autostart_scripts()).files == [] + assert await client.get_autostart_script("boot.sh") == "#!/bin/sh\n" + await client.upload_autostart_script("boot.sh", "true\n") + calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}vm/autostart/boot.sh"))] + assert calls[0].kwargs["json"] == {"content": "true\n"} + for name in ("", "..", "../kvm/server.yaml"): + with pytest.raises(ValueError): + await client.delete_autostart_script(name) diff --git a/tests/test_system.py b/tests/test_system.py index 35144f8..95a2d38 100644 --- a/tests/test_system.py +++ b/tests/test_system.py @@ -1,5 +1,8 @@ """Tests for system, virtual device and LED operations.""" +from datetime import UTC, datetime + +import aiohttp from aioresponses import aioresponses import pytest import yarl @@ -172,3 +175,29 @@ async def test_set_led_strip_partial_preserves_current_config( "ver": 6, "brightness": 50, } + + +async def test_get_time_reads_first_stream_event( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test get_time parses the first event of the Pro clock stream.""" + mark_detected(client, HWVersion.PRO, "1.2.15") + mocked.get( + f"{BASE_URL}vm/time", + body="event:message\ndata:1791944949000\n\nevent:message\ndata:1\n\n", + ) + + assert await client.get_time() == datetime(2026, 10, 14, 2, 29, 9, tzinfo=UTC) + + +async def test_set_tls_treats_restart_disconnect_as_success( + client: NanoKVMClient, mocked: aioresponses +) -> None: + """Test set_tls succeeds when the device restarts before answering.""" + mark_detected(client, HWVersion.PCIE, "2.5.2") + mocked.post(f"{BASE_URL}vm/tls", exception=aiohttp.ServerDisconnectedError()) + + await client.set_tls(True) + + calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}vm/tls"))] + assert calls[0].kwargs["json"] == {"enabled": True} diff --git a/tests/test_video.py b/tests/test_video.py index d9f709a..3c72be7 100644 --- a/tests/test_video.py +++ b/tests/test_video.py @@ -1,12 +1,13 @@ """Tests for video stream settings.""" from aioresponses import aioresponses +import pytest import yarl from nanokvm.client import NanoKVMClient from nanokvm.models import HWVersion, StreamMode -from .common import BASE_URL +from .common import BASE_URL, mark_detected async def test_set_stream_mode_accepts_existing_string_values( @@ -24,3 +25,17 @@ async def test_set_stream_mode_accepts_existing_string_values( calls = mocked.requests[("POST", yarl.URL(f"{BASE_URL}stream/mode"))] assert calls[0].kwargs.get("json") == {"mode": "h264-direct"} + + +async def test_pause_frame_detect(client: NanoKVMClient, mocked: aioresponses) -> None: + """Test the pause sends its duration and rejects non-positive ones.""" + mark_detected(client, HWVersion.PCIE) + url = f"{BASE_URL}stream/mjpeg/detect/stop" + mocked.post(url, payload={"code": 0, "msg": "success", "data": None}) + + await client.pause_frame_detect(3) + + calls = mocked.requests[("POST", yarl.URL(url))] + assert calls[0].kwargs["json"] == {"duration": 3} + with pytest.raises(ValueError): + await client.pause_frame_detect(0)