Add API to manage SSH authorized keys on Home Assistant OS

supervisor/api/__init__.py

@@ -305,6 +305,7 @@ def _register_os(self, app: web.Application) -> None:
                 web.get("/os/datadisk/list", api_os.list_data),
                 web.post("/os/datadisk/wipe", api_os.wipe_data),
                 web.post("/os/boot-slot", api_os.set_boot_slot),
+                web.post("/os/ssh/authorized_keys", api_os.ssh_authorized_keys),
             ]
         )
 

supervisor/api/const.py

@@ -44,6 +44,7 @@
 ATTR_IS_ACTIVE = "is_active"
 ATTR_IS_OWNER = "is_owner"
 ATTR_JOBS = "jobs"
+ATTR_KEYS = "keys"
 ATTR_LLMNR = "llmnr"
 ATTR_LLMNR_HOSTNAME = "llmnr_hostname"
 ATTR_LOCAL_ONLY = "local_only"

supervisor/api/middleware/security.py

@@ -146,7 +146,7 @@ class _AppSecurityPatterns:
             r"|/multicast/.+"
             r"|/network/.+"
             r"|/observer/.+"
-            r"|/os/(?!datadisk/wipe).+"
+            r"|/os/(?!datadisk/wipe|ssh/authorized_keys).+"
             r"|/refresh_updates"
             r"|/resolution/.+"
             r"|/security/.+"
@@ -226,7 +226,7 @@ class _AppSecurityPatterns:
             r"|/multicast/.+"
             r"|/network/.+"
             r"|/observer/.+"
-            r"|/os/(?!datadisk/wipe).+"
+            r"|/os/(?!datadisk/wipe|ssh/authorized_keys).+"
             r"|/reload_updates"
             r"|/resolution/.+"
             r"|/security/.+"

supervisor/api/os.py

@@ -1,10 +1,12 @@
 """Init file for Supervisor HassOS RESTful API."""
 
 import asyncio
+import base64
+import binascii
 from collections.abc import Awaitable
 import logging
 import re
-from typing import Any
+from typing import Any, Final
 
 from aiohttp import web
 from awesomeversion import AwesomeVersion
@@ -47,6 +49,7 @@
     ATTR_DEV_PATH,
     ATTR_DEVICE,
     ATTR_DISKS,
+    ATTR_KEYS,
     ATTR_MODEL,
     ATTR_STATUS,
     ATTR_SYSTEM_HEALTH_LED,
@@ -89,6 +92,77 @@
         vol.Optional(ATTR_SWAPPINESS): vol.All(int, vol.Range(min=0, max=200)),
     }
 )
+
+# Plain OpenSSH public key types accepted for root's authorized_keys.
+# Certificates and authorized_keys options are deliberately not supported.
+SSH_AUTH_KEY_TYPES: Final = frozenset(
+    {
+        "ssh-ed25519",
+        "ssh-rsa",
+        "ecdsa-sha2-nistp256",
+        "ecdsa-sha2-nistp384",
+        "ecdsa-sha2-nistp521",
+        "sk-ssh-ed25519@openssh.com",
+        "sk-ecdsa-sha2-nistp256@openssh.com",
+    }
+)
+
+# dropbear, which consumes authorized_keys on Home Assistant OS, ignores
+# lines longer than 3000 bytes (and OS Agent rejects them)
+SSH_AUTH_KEY_MAX_LENGTH: Final = 3000
+
+RE_SSH_KEY_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f]")
+
+
+def ssh_public_key(value: Any) -> str:
+    """Validate an OpenSSH public key in authorized_keys format.
+
+    OS Agent appends the string verbatim to root's authorized_keys, so only
+    a strictly well-formed plain public key (type, base64 blob, optional
+    comment) may pass — no options, no control characters.
+    """
+    if not isinstance(value, str):
+        raise vol.Invalid("SSH public key must be a string")
+
+    key = value.strip()
+    # dropbear and OS Agent limit the line length in bytes, not characters
+    if not key or len(key.encode()) > SSH_AUTH_KEY_MAX_LENGTH:
+        raise vol.Invalid("SSH public key is empty or too long")
+    if RE_SSH_KEY_CONTROL_CHARS.search(key):
+        raise vol.Invalid("SSH public key contains control characters")
+
+    parts = key.split(maxsplit=2)
+    if len(parts) < 2:
+        raise vol.Invalid(
+            "SSH public key must be in '<type> <base64-key> [comment]' format"
+        )
+    key_type, key_data = parts[0], parts[1]
+
+    if key_type not in SSH_AUTH_KEY_TYPES:
+        raise vol.Invalid(f"Unsupported SSH public key type: {key_type}")
+
+    try:
+        blob = base64.b64decode(key_data, validate=True)
+    except binascii.Error:
+        raise vol.Invalid("SSH public key data is not valid base64") from None
+
+    # The decoded blob embeds the algorithm name; require it to match the
+    # declared type so arbitrary data can't be smuggled into the file.
+    embedded_len = int.from_bytes(blob[:4], "big") if len(blob) >= 4 else -1
+    if (
+        embedded_len < 0
+        or 4 + embedded_len > len(blob)
+        or blob[4 : 4 + embedded_len] != key_type.encode()
+    ):
+        raise vol.Invalid("SSH public key data does not match its declared type")
+
+    canonical = f"{key_type} {key_data}"
+    if len(parts) == 3:
+        canonical += f" {parts[2]}"
+    return canonical
+
+
+SCHEMA_SSH_AUTHORIZED_KEYS = vol.Schema({vol.Required(ATTR_KEYS): [ssh_public_key]})
 # pylint: enable=no-value-for-parameter
 
 
@@ -153,6 +227,12 @@ async def set_boot_slot(self, request: web.Request) -> None:
         body = await api_validate(SCHEMA_SET_BOOT_SLOT, request)
         await asyncio.shield(self.sys_os.set_boot_slot(body[ATTR_BOOT_SLOT]))
 
+    @api_process
+    async def ssh_authorized_keys(self, request: web.Request) -> None:
+        """Replace root's SSH authorized keys on the host."""
+        body = await api_validate(SCHEMA_SSH_AUTHORIZED_KEYS, request)
+        await asyncio.shield(self.sys_os.set_ssh_authorized_keys(body[ATTR_KEYS]))
+
     @api_process
     async def list_data(self, request: web.Request) -> dict[str, Any]:
         """Return possible data targets."""

supervisor/dbus/agent/system.py

@@ -20,3 +20,17 @@ async def schedule_wipe_device(self) -> bool:
     async def migrate_docker_storage_driver(self, backend: str) -> None:
         """Migrate Docker storage driver."""
         await self.connected_dbus.System.call("migrate_docker_storage_driver", backend)
+
+    @dbus_connected
+    async def add_ssh_auth_key(self, key: str) -> None:
+        """Append a public key to root's SSH authorized keys on the host.
+
+        OS Agent writes the string verbatim to the authorized_keys file, so
+        callers must validate it first.
+        """
+        await self.connected_dbus.System.call("add_ssh_auth_key", key)
+
+    @dbus_connected
+    async def clear_ssh_auth_keys(self) -> None:
+        """Remove all of root's SSH authorized keys on the host."""
+        await self.connected_dbus.System.call("clear_ssh_auth_keys")

supervisor/os/manager.py

@@ -15,10 +15,12 @@
 from ..exceptions import (
     DBusError,
     DBusNotConnectedError,
+    HassOSError,
     HassOSJobError,
     HassOSSlotNotFound,
     HassOSSlotUpdateError,
     HassOSUpdateError,
+    HostError,
 )
 from ..jobs.const import JobConcurrency, JobCondition
 from ..jobs.decorator import Job
@@ -27,6 +29,9 @@
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
+# SSH service on Home Assistant OS consuming /root/.ssh/authorized_keys
+DROPBEAR_SERVICE = "dropbear.service"
+
 
 @dataclass(slots=True, frozen=True)
 class SlotStatus:
@@ -501,3 +506,54 @@ async def set_boot_slot(self, boot_name: str) -> None:
 
         _LOGGER.info("Rebooting into new boot slot now")
         await self.sys_host.control.reboot()
+
+    @Job(
+        name="os_manager_set_ssh_authorized_keys",
+        conditions=[JobCondition.HAOS],
+        on_condition=HassOSJobError,
+        concurrency=JobConcurrency.REJECT,
+        internal=True,
+    )
+    async def set_ssh_authorized_keys(self, keys: list[str]) -> None:
+        """Replace root's SSH authorized keys on the host and start dropbear.
+
+        OS Agent only offers clear and append operations, so the replacement
+        is not atomic: if an append fails, keys added before it remain in
+        place. Callers must validate the keys beforehand.
+        """
+        _LOGGER.info("Replacing SSH authorized keys on host (%d keys)", len(keys))
+        try:
+            await self.sys_dbus.agent.system.clear_ssh_auth_keys()
+        except DBusError as err:
+            # OS Agent up to 1.10.x returns the os.Remove error when the
+            # authorized_keys file is already absent (inverted error check,
+            # fixed since). That is the empty state clearing aims for, so
+            # treat it as success.
+            if "no such file or directory" not in str(err):
+                raise HassOSError(
+                    f"Can't clear SSH authorized keys: {err!s}", _LOGGER.error
+                ) from err
+
+        for key in keys:
+            try:
+                await self.sys_dbus.agent.system.add_ssh_auth_key(key)
+            except DBusError as err:
+                raise HassOSError(
+                    f"Can't add SSH authorized key: {err!s}", _LOGGER.error
+                ) from err
+
+        if not keys:
+            return
+
+        # dropbear on Home Assistant OS is gated by
+        # ConditionFileNotEmpty=/root/.ssh/authorized_keys, which systemd only
+        # evaluates when the unit starts. A running dropbear re-reads the file
+        # on every authentication attempt and starting an active unit is a
+        # no-op, so only the stopped service needs this.
+        try:
+            await self.sys_host.services.start(DROPBEAR_SERVICE)
+        except (HostError, DBusError) as err:
+            raise HassOSError(
+                f"SSH authorized keys written, but can't start dropbear: {err!s}",
+                _LOGGER.error,
+            ) from err

tests/api/middleware/test_security.py

@@ -215,6 +215,7 @@ def _versioned_path(prefix: str, path: str) -> str:
         ("post", "/addons/abc123/restart", {"admin", "manager"}),
         ("post", "/addons/abc123/security", {"admin"}),
         ("post", "/os/datadisk/wipe", {"admin"}),
+        ("post", "/os/ssh/authorized_keys", {"admin"}),
         ("post", "/addons/self/sys_options", set()),
         ("post", "/addons/abc123/sys_options", set()),
     ],