Support SSL pinning (#14)

Author: puddly

README.md

@@ -105,7 +105,25 @@ async with NanoKVMClient(
     await client.authenticate("username", "password")
 ```
 
-#### Option 2: Use custom CA certificate (recommended)
+#### Option 2: Certificate pinning (recommended for self-signed)
+
+NanoKVM devices generate self-signed certificates for `localhost` with no CA to verify against. Certificate pinning verifies the server's certificate fingerprint directly instead of relying on CA-based trust.
+
+```python
+from nanokvm.utils import async_fetch_remote_fingerprint
+
+# First, fetch the fingerprint (trust-on-first-use)
+fingerprint = await async_fetch_remote_fingerprint("https://kvm.local/api/")
+
+# Then connect with the pinned fingerprint
+async with NanoKVMClient(
+    "https://kvm.local/api/",
+    ssl_fingerprint=fingerprint,
+) as client:
+    await client.authenticate("username", "password")
+```
+
+#### Option 3: Use custom CA certificate
 
 ```python
 async with NanoKVMClient(

nanokvm/client.py

@@ -16,8 +16,8 @@
     BodyPartReader,
     ClientResponse,
     ClientSession,
+    Fingerprint,
     MultipartReader,
-    TCPConnector,
     hdrs,
 )
 from PIL import Image
@@ -123,40 +123,51 @@ def __init__(
         *,
         token: str | None = None,
         request_timeout: int = 10,
+        session: ClientSession | None = None,
         verify_ssl: bool = True,
         ssl_ca_cert: str | None = None,
+        ssl_fingerprint: str | None = None,
         use_password_obfuscation: bool | None = None,
     ) -> None:
         """
         Initialize the NanoKVM client.
 
         Args:
             url: Base URL of the NanoKVM API (e.g., "https://kvm.local/api/")
+            session: aiohttp ClientSession to use for requests.
             token: Optional pre-existing authentication token
             request_timeout: Request timeout in seconds (default: 10)
             verify_ssl: Enable SSL certificate verification (default: True).
                 Set to False to disable verification for self-signed certificates.
             ssl_ca_cert: Path to custom CA certificate bundle file for SSL verification.
                 Useful for self-signed certificates or private CAs.
+            ssl_fingerprint: SHA-256 fingerprint of the server's TLS certificate
+                as a hex string. When set, the client will verify the server's
+                certificate fingerprint instead of performing CA-based verification.
+                Use `async_fetch_remote_fingerprint()` to retrieve this value.
             use_password_obfuscation: Control password obfuscation mode (default: None).
                 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).
         """
         self.url = yarl.URL(url)
-        self._session: ClientSession | None = None
+        self._session: ClientSession | None = session
+        self._external_session_provided = session is not None
         self._token = token
         self._request_timeout = request_timeout
         self._ws: aiohttp.ClientWebSocketResponse | None = None
         self._verify_ssl = verify_ssl
         self._ssl_ca_cert = ssl_ca_cert
+        self._ssl_fingerprint = ssl_fingerprint
         self._use_password_obfuscation = use_password_obfuscation
+        self._ssl_config: ssl.SSLContext | Fingerprint | bool | None = None
 
-    def _create_ssl_context(self) -> ssl.SSLContext | bool:
+    def _create_ssl_context(self) -> ssl.SSLContext | Fingerprint | bool:
         """
         Create and configure SSL context based on initialization parameters.
 
         Returns:
+            Fingerprint: Certificate fingerprint pinning (when ssl_fingerprint set)
             ssl.SSLContext: Configured SSL context for custom certificates
             True: Use default SSL verification (aiohttp default)
             False: Disable SSL verification
@@ -166,6 +177,10 @@ def _create_ssl_context(self) -> ssl.SSLContext | bool:
             ssl.SSLError: If the CA certificate is invalid.
         """
 
+        if self._ssl_fingerprint:
+            _LOGGER.debug("Using certificate fingerprint pinning")
+            return Fingerprint(bytes.fromhex(self._ssl_fingerprint.replace(":", "")))
+
         if not self._verify_ssl:
             _LOGGER.warning(
                 "SSL verification is disabled. This is insecure and should only be "
@@ -188,16 +203,10 @@ def token(self) -> str | None:
 
     async def __aenter__(self) -> NanoKVMClient:
         """Async context manager entry."""
+        if self._session is None and not self._external_session_provided:
+            self._session = ClientSession()
 
-        ssl_config = await asyncio.to_thread(self._create_ssl_context)
-        connector = TCPConnector(ssl=ssl_config)
-        self._session = ClientSession(connector=connector)
-
-        _LOGGER.debug(
-            "Created client session with SSL verification: %s",
-            "disabled" if ssl_config is False else "enabled",
-        )
-
+        self._ssl_config = await asyncio.to_thread(self._create_ssl_context)
         return self
 
     async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
@@ -206,8 +215,9 @@ async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
         if self._ws is not None and not self._ws.closed:
             await self._ws.close()
             self._ws = None
+
         # Close HTTP session
-        if self._session is not None:
+        if self._session is not None and not self._external_session_provided:
             await self._session.close()
             self._session = None
 
@@ -221,16 +231,15 @@ async def _request(
         **kwargs: Any,
     ) -> AsyncIterator[ClientResponse]:
         """Make an API request."""
-        assert self._session is not None, (
-            "Client session not initialized. "
-            "Use as context manager: 'async with NanoKVMClient(url) as client:'"
-        )
         cookies = {}
         if authenticate:
             if not self._token:
                 raise NanoKVMNotAuthenticatedError("Client is not authenticated")
             cookies["nano-kvm-token"] = self._token
 
+        assert self._session is not None
+        assert self._ssl_config is not None
+
         async with self._session.request(
             method,
             self.url / path.lstrip("/"),
@@ -240,6 +249,7 @@ async def _request(
             cookies=cookies,
             timeout=aiohttp.ClientTimeout(total=self._request_timeout),
             raise_for_status=True,
+            ssl=self._ssl_config,
             **kwargs,
         ) as response:
             yield response
@@ -278,7 +288,7 @@ async def _api_request_json(
         async with self._request(
             method,
             path,
-            json=(data.dict() if data is not None else None),
+            json=(data.model_dump() if data is not None else None),
             **kwargs,
         ) as response:
             try:
@@ -795,21 +805,20 @@ async def set_mouse_jiggler_state(
     async def _get_ws(self) -> aiohttp.ClientWebSocketResponse:
         """Get or create WebSocket connection for mouse events."""
         if self._ws is None or self._ws.closed:
-            assert self._session is not None, (
-                "Client session not initialized. "
-                "Use as context manager: 'async with NanoKVMClient(url) as client:'"
-            )
-
             if not self._token:
                 raise NanoKVMNotAuthenticatedError("Client is not authenticated")
 
             # WebSocket URL uses ws:// or wss:// scheme
             scheme = "ws" if self.url.scheme == "http" else "wss"
             ws_url = self.url.with_scheme(scheme) / "ws"
 
+            assert self._session is not None
+            assert self._ssl_config is not None
+
             self._ws = await self._session.ws_connect(
                 str(ws_url),
                 headers={"Cookie": f"nano-kvm-token={self._token}"},
+                ssl=self._ssl_config,
             )
         return self._ws
 

nanokvm/utils.py

@@ -1,6 +1,8 @@
+import asyncio
 import base64
 import hashlib
 import os
+import ssl
 import urllib.parse
 
 from cryptography.hazmat.primitives import padding
@@ -52,3 +54,34 @@ def obfuscate_password(password: str) -> str:
     )
 
     return urllib.parse.quote(base64.b64encode(password_enc).decode("utf-8"), safe="")
+
+
+async def async_fetch_remote_fingerprint(
+    url: str, *, timeout: float | None = 10.0
+) -> str:
+    """Retrieve the SHA-256 fingerprint of the remote server's TLS certificate.
+
+    Connects to the server with verification disabled to grab the raw certificate,
+    then returns its SHA-256 hash as an uppercase hex string.
+
+    This is useful for establishing an initial trust-on-first-use pin with
+    `NanoKVMClient(url, ssl_fingerprint=...)`.
+    """
+    parsed_url = urllib.parse.urlparse(url)
+    hostname = parsed_url.hostname
+    port = parsed_url.port or 443
+
+    ssl_ctx = await asyncio.to_thread(ssl.create_default_context)
+    ssl_ctx.check_hostname = False
+    ssl_ctx.verify_mode = ssl.CERT_NONE
+
+    async with asyncio.timeout(timeout):
+        _, writer = await asyncio.open_connection(hostname, port, ssl=ssl_ctx)
+
+    try:
+        ssl_obj = writer.get_extra_info("ssl_object")
+        der_cert = ssl_obj.getpeercert(binary_form=True)
+        return hashlib.sha256(der_cert).hexdigest().upper()
+    finally:
+        writer.close()
+        await writer.wait_closed()

pyproject.toml

@@ -29,6 +29,7 @@ exclude = ["tests", "tests.*"]
 testing = [
     "tomli",
     "coverage[toml]",
+    "cryptography",
     "pytest",
     "pytest-xdist",
     "pytest-asyncio",