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

@@ -47,6 +47,7 @@
     ATTR_DEV_PATH,
     ATTR_DEVICE,
     ATTR_DISKS,
+    ATTR_KEYS,
     ATTR_MODEL,
     ATTR_STATUS,
     ATTR_SYSTEM_HEALTH_LED,
@@ -89,6 +90,35 @@
         vol.Optional(ATTR_SWAPPINESS): vol.All(int, vol.Range(min=0, max=200)),
     }
 )
+
+# dropbear, which consumes authorized_keys on Home Assistant OS, ignores
+# lines longer than 3000 bytes
+SSH_AUTH_KEY_MAX_LENGTH = 3000
+
+RE_SSH_KEY_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f]")
+
+
+def ssh_auth_key(value: Any) -> str:
+    """Run a basic sanity check on an SSH authorized key entry.
+
+    Proper key validation is done by OS Agent; reject only what could write
+    more than one authorized_keys line per key (control characters) or
+    produce a line dropbear ignores (too long).
+    """
+    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")
+
+    return key
+
+
+SCHEMA_SSH_AUTHORIZED_KEYS = vol.Schema({vol.Required(ATTR_KEYS): [ssh_auth_key]})
 # pylint: enable=no-value-for-parameter
 
 
@@ -153,6 +183,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 validates the key since 1.10.0; older releases write the
+        string verbatim to the authorized_keys file.
+        """
+        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,16 @@
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
+# SSH service on Home Assistant OS consuming /root/.ssh/authorized_keys
+DROPBEAR_SERVICE = "dropbear.service"
+
+# OS Agent releases before this return the os.Remove error when clearing an
+# already absent authorized_keys file (inverted error check)
+CLEAR_SSH_AUTH_KEYS_FIXED_VERSION = AwesomeVersion("1.10.0")
+CLEAR_SSH_AUTH_KEYS_MISSING_FILE_ERROR = (
+    "remove /root/.ssh/authorized_keys: no such file or directory"
+)
+
 
 @dataclass(slots=True, frozen=True)
 class SlotStatus:
@@ -501,3 +513,55 @@ 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 validates each key since 1.10.0 and only offers clear and
+        append operations, so the replacement is not atomic: if an append is
+        rejected or fails, keys added before it remain in place.
+        """
+        _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:
+            # On affected OS Agent releases the missing-file error is the
+            # empty state clearing aims for, so treat it as success there.
+            if (
+                self.sys_dbus.agent.version >= CLEAR_SSH_AUTH_KEYS_FIXED_VERSION
+                or CLEAR_SSH_AUTH_KEYS_MISSING_FILE_ERROR 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()),
     ],