From 210b49fb039976732a007bcab346ccf73bb6b3db Mon Sep 17 00:00:00 2001 From: Stefan Agner Date: Thu, 27 Aug 2026 10:12:44 +0200 Subject: [PATCH] Add API to manage SSH authorized keys on Home Assistant OS (#7039) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Add API to manage SSH authorized keys on Home Assistant OS The OS Agent has long exposed AddSSHAuthKey and ClearSSHAuthKeys on its io.hass.os System D-Bus object, but the Supervisor never wrapped them, so there was no way to manage root's SSH authorized keys through the Supervisor API. Add POST /os/ssh/authorized_keys, which replaces the configured keys with the submitted list (an empty list just clears them). Since OS Agent writes each key verbatim to /root/.ssh/authorized_keys as root, the endpoint validates strictly before anything is written: plain public keys only (no options, no certificates), a key type allowlist matching what dropbear on Home Assistant OS can verify, no control characters (one submitted key can never write more than one line), the base64 blob must embed the declared key type, and entries are capped at dropbear's 3000 byte per-line limit. The endpoint is admin-only for add-on tokens. Replacement clears the existing keys and then adds each key. OS Agent releases up to 1.10.x return an error when clearing an already absent file (inverted error check, since fixed), which is the state of every first-time user, so this specific error is treated as the empty state it reports. dropbear on Home Assistant OS is gated by ConditionFileNotEmpty on the authorized_keys file, which systemd only evaluates when the unit starts, so the service is started after a non-empty key set is written. A running dropbear re-reads the file on every authentication attempt and needs no restart. * Delegate SSH key validation to OS Agent Review discussion questioned the full key validation (type allowlist, base64 blob checks, canonicalization): OS Agent 1.10.0 validates submitted keys itself and treats clearing an already absent authorized_keys file as success, so the Supervisor can rely on it instead of duplicating the logic. Require OS Agent 1.10.0 or newer and reject requests on older releases with 404, like the Raspberry Pi firmware endpoints do; this also makes the missing-file compatibility shim for the clear call unnecessary. A key rejected by OS Agent surfaces as an error response including its validation message. The Supervisor keeps only a basic per-key sanity check that runs before anything is written: no control characters (one submitted key can never write more than one authorized_keys line) and at most 3000 bytes (dropbear ignores longer lines, which would leave a key that passes but never works). * Support OS Agent releases before 1.10.0 Requiring OS Agent 1.10.0 would keep the feature unavailable until the next OS update reaches users, while Supervisor updates roll out independently. Drop the version requirement and accept that validation is lossy on older releases: the Supervisor sanity check still prevents writing more than one line per key or lines dropbear ignores, but proper key validation only happens on OS Agent 1.10.0 or newer. This brings back the need to tolerate the error OS Agent releases before 1.10.0 return when clearing an already absent authorized_keys file (inverted error check): match the complete os.Remove error message for the authorized_keys path and treat it as the empty state clearing aims for, on affected versions only. * Split SSH authorized keys API into add and clear endpoints Review feedback preferred endpoints mapping 1:1 onto the OS Agent D-Bus methods over a single replace-the-set API, whose POST semantics were also questioned (an idempotent full replacement would be PUT). POST /os/ssh/authorized_keys now takes a single key ({"key": "..."}) and appends it via AddSSHAuthKey; DELETE /os/ssh/authorized_keys removes all keys via ClearSSHAuthKeys. Each call maps to exactly one OS Agent operation, so no request can partially succeed. Clients that want to replace the configured set clear and re-add; a GET (which needs an OS Agent extension first) and an idempotent PUT can be added later. The per-key sanity check, the dropbear service start after adding a key, and the tolerance for the missing-file clear error of OS Agent releases before 1.10.0 carry over unchanged. * Add endpoint to list SSH authorized keys With only add and clear operations the authorized_keys file is write-only for API consumers: a user cannot audit which keys grant access to the box, or whether any exist at all — including keys that were imported from USB or written by add-ons. OS Agent 1.11.0 added a ListSSHAuthKeys D-Bus method. Expose it as GET /os/ssh/authorized_keys, returning the configured entries verbatim. The endpoint requires OS Agent 1.11.0 and returns 404 on older releases, like the Raspberry Pi firmware endpoints do; add and clear keep working on all OS Agent releases. * Restrict SSH authorized keys endpoints to Home Assistant Core Review decision: manipulating root's SSH access is not a capability add-ons should have, even with the admin role, so move the endpoints from admin-only to the core_only middleware pattern. Only requests authenticated with the Home Assistant Core token pass; add-on tokens of any role, the CLI plugin, and the observer are rejected. This also means the host shell (ha CLI) cannot use the endpoints for now — the restriction can be opened up later. The exclusion from the manager role allowlist is kept: if the path is ever removed from core_only again, it falls back to admin-only rather than becoming manager-accessible. * Stop dropbear after clearing SSH authorized keys Clearing all authorized keys is a revocation, but without stopping dropbear the listener keeps running until reboot and established sessions survive, as only a service stop terminates them. Stop the service after a successful clear, mirroring the USB config import (haos-config), which also stops dropbear when the imported authorized_keys file is removed. Stopping an inactive unit is a no-op. * Serialize SSH authorized keys jobs on a common lock The add and clear jobs each perform a file operation followed by a service operation, and nothing prevented them from running concurrently. An interleaving like clear-file, add-key, start-dropbear, stop-dropbear lets both requests succeed while the final service state does not match the final key state (key configured, dropbear stopped). Make OSManager a JobGroup and run both jobs with GROUP_QUEUE concurrency, so each file-and-service operation completes before the next starts. The regression test fails without the shared lock. --- supervisor/api/__init__.py | 3 + supervisor/api/const.py | 2 + supervisor/api/middleware/security.py | 6 +- supervisor/api/os.py | 61 +++++ supervisor/dbus/agent/system.py | 22 ++ supervisor/os/manager.py | 89 +++++++- tests/api/middleware/test_security.py | 10 + tests/api/test_os.py | 275 ++++++++++++++++++++++- tests/dbus/agent/test_system.py | 41 ++++ tests/dbus_service_mocks/agent_system.py | 22 ++ tests/os/test_manager.py | 53 +++++ 11 files changed, 577 insertions(+), 7 deletions(-) diff --git a/supervisor/api/__init__.py b/supervisor/api/__init__.py index 529c3353c..6edc2e4c3 100644 --- a/supervisor/api/__init__.py +++ b/supervisor/api/__init__.py @@ -314,6 +314,9 @@ class RestAPI(CoreSysAttributes): 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.get("/os/ssh/authorized_keys", api_os.ssh_authorized_keys_list), + web.post("/os/ssh/authorized_keys", api_os.ssh_authorized_keys_add), + web.delete("/os/ssh/authorized_keys", api_os.ssh_authorized_keys_clear), ] ) diff --git a/supervisor/api/const.py b/supervisor/api/const.py index dc4ed6f83..6655f5580 100644 --- a/supervisor/api/const.py +++ b/supervisor/api/const.py @@ -44,6 +44,8 @@ ATTR_IDENTIFIERS = "identifiers" ATTR_IS_ACTIVE = "is_active" ATTR_IS_OWNER = "is_owner" ATTR_JOBS = "jobs" +ATTR_KEY = "key" +ATTR_KEYS = "keys" ATTR_LLMNR = "llmnr" ATTR_LLMNR_HOSTNAME = "llmnr_hostname" ATTR_LOCAL_ONLY = "local_only" diff --git a/supervisor/api/middleware/security.py b/supervisor/api/middleware/security.py index 54d2b21f3..8368c7b87 100644 --- a/supervisor/api/middleware/security.py +++ b/supervisor/api/middleware/security.py @@ -109,6 +109,7 @@ _V1_PATTERNS: Final = _AppSecurityPatterns( core_only=re.compile( r"^(?:" r"/addons/" + RE_SLUG + r"/sys_options" + r"|/os/ssh/authorized_keys" r")$" ), role_access={ @@ -150,7 +151,7 @@ _V1_PATTERNS: Final = _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/.+" @@ -190,6 +191,7 @@ _V2_PATTERNS: Final = _AppSecurityPatterns( core_only=re.compile( r"^/v2(?:" r"/apps/" + RE_SLUG + r"/sys_options" + r"|/os/ssh/authorized_keys" r")$" ), role_access={ @@ -230,7 +232,7 @@ _V2_PATTERNS: Final = _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/.+" diff --git a/supervisor/api/os.py b/supervisor/api/os.py index e5b4796ec..1b2e99679 100644 --- a/supervisor/api/os.py +++ b/supervisor/api/os.py @@ -49,6 +49,8 @@ from .const import ( ATTR_DEV_PATH, ATTR_DEVICE, ATTR_DISKS, + ATTR_KEY, + ATTR_KEYS, ATTR_MODEL, ATTR_STATUS, ATTR_SYSTEM_HEALTH_LED, @@ -69,6 +71,10 @@ CORE_VERSION_PENDING_MIN_VERSION: AwesomeVersion = AwesomeVersion( "2026.8.0.dev202607250310" ) +# Listing SSH authorized keys requires the ListSSHAuthKeys D-Bus method +# first shipped in this OS Agent release. +SSH_KEYS_LIST_MIN_OS_AGENT_VERSION: AwesomeVersion = AwesomeVersion("1.11.0") + # pylint: disable=no-value-for-parameter SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): version_tag}) SCHEMA_SET_BOOT_SLOT = vol.Schema({vol.Required(ATTR_BOOT_SLOT): vol.Coerce(BootSlot)}) @@ -97,6 +103,35 @@ SCHEMA_SWAP_OPTIONS = vol.Schema( 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_KEY = vol.Schema({vol.Required(ATTR_KEY): ssh_auth_key}) # pylint: enable=no-value-for-parameter @@ -173,6 +208,32 @@ class APIOS(CoreSysAttributes): 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_list(self, request: web.Request) -> dict[str, Any]: + """Return root's SSH authorized keys on the host.""" + if ( + not self.sys_dbus.agent.is_connected + or self.sys_dbus.agent.version < SSH_KEYS_LIST_MIN_OS_AGENT_VERSION + ): + raise APINotFound( + f"OS Agent {SSH_KEYS_LIST_MIN_OS_AGENT_VERSION} or newer required " + "to list SSH authorized keys", + _LOGGER.debug, + ) + + return {ATTR_KEYS: await self.sys_dbus.agent.system.list_ssh_auth_keys()} + + @api_process + async def ssh_authorized_keys_add(self, request: web.Request) -> None: + """Add an SSH authorized key for root on the host.""" + body = await api_validate(SCHEMA_SSH_AUTHORIZED_KEY, request) + await asyncio.shield(self.sys_os.add_ssh_authorized_key(body[ATTR_KEY])) + + @api_process + def ssh_authorized_keys_clear(self, request: web.Request) -> Awaitable[None]: + """Remove all SSH authorized keys of root on the host.""" + return asyncio.shield(self.sys_os.clear_ssh_authorized_keys()) + @api_process async def list_data(self, request: web.Request) -> dict[str, Any]: """Return possible data targets.""" diff --git a/supervisor/dbus/agent/system.py b/supervisor/dbus/agent/system.py index 69f5ce683..200f3944f 100644 --- a/supervisor/dbus/agent/system.py +++ b/supervisor/dbus/agent/system.py @@ -20,3 +20,25 @@ class System(DBusInterface): 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") + + @dbus_connected + async def list_ssh_auth_keys(self) -> list[str]: + """Return root's SSH authorized keys on the host. + + Requires OS Agent 1.11.0 or newer. + """ + return await self.connected_dbus.System.call("list_ssh_auth_keys") diff --git a/supervisor/os/manager.py b/supervisor/os/manager.py index 7ff573351..0138b202a 100644 --- a/supervisor/os/manager.py +++ b/supervisor/os/manager.py @@ -9,24 +9,37 @@ import aiohttp from awesomeversion import AwesomeVersion, AwesomeVersionException from cpe import CPE -from ..coresys import CoreSys, CoreSysAttributes +from ..coresys import CoreSys from ..dbus.agent.boards.const import BOARD_NAME_SUPERVISED from ..dbus.rauc import RaucState, SlotStatusDataType from ..exceptions import ( DBusError, DBusNotConnectedError, + HassOSError, HassOSJobError, HassOSSlotNotFound, HassOSSlotUpdateError, HassOSUpdateError, + HostError, ) from ..jobs.const import JobConcurrency, JobCondition from ..jobs.decorator import Job +from ..jobs.job_group import JobGroup from ..resolution.const import ContextType, IssueType, SuggestionType from .data_disk import DataDisk _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: @@ -80,12 +93,12 @@ class SlotStatus: ) -class OSManager(CoreSysAttributes): +class OSManager(JobGroup): """OS interface inside supervisor.""" def __init__(self, coresys: CoreSys): """Initialize HassOS handler.""" - self.coresys: CoreSys = coresys + super().__init__(coresys, "os_manager") self._datadisk: DataDisk = DataDisk(coresys) self._available: bool = False self._version: AwesomeVersion | None = None @@ -501,3 +514,73 @@ class OSManager(CoreSysAttributes): _LOGGER.info("Rebooting into new boot slot now") await self.sys_host.control.reboot() + + @Job( + name="os_manager_add_ssh_authorized_key", + conditions=[JobCondition.HAOS], + on_condition=HassOSJobError, + concurrency=JobConcurrency.GROUP_QUEUE, + internal=True, + ) + async def add_ssh_authorized_key(self, key: str) -> None: + """Add an SSH authorized key for root on the host and start dropbear. + + OS Agent validates the key since 1.10.0; older releases append it to + the authorized_keys file as submitted. + """ + _LOGGER.info("Adding SSH authorized key on host") + 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 + + # 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 key written, but can't start dropbear: {err!s}", + _LOGGER.error, + ) from err + + @Job( + name="os_manager_clear_ssh_authorized_keys", + conditions=[JobCondition.HAOS], + on_condition=HassOSJobError, + concurrency=JobConcurrency.GROUP_QUEUE, + internal=True, + ) + async def clear_ssh_authorized_keys(self) -> None: + """Remove all SSH authorized keys of root on the host and stop dropbear.""" + _LOGGER.info("Clearing SSH authorized keys on host") + 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 + + # Mirror the USB config import (haos-config), which stops dropbear + # when the imported authorized_keys file is removed. Clearing all + # keys is a revocation, so also terminate established sessions, + # which survive until the service stops. Stopping an inactive unit + # is a no-op. + try: + await self.sys_host.services.stop(DROPBEAR_SERVICE) + except (HostError, DBusError) as err: + raise HassOSError( + f"SSH authorized keys cleared, but can't stop dropbear: {err!s}", + _LOGGER.error, + ) from err diff --git a/tests/api/middleware/test_security.py b/tests/api/middleware/test_security.py index 13a3aee9e..c28395e48 100644 --- a/tests/api/middleware/test_security.py +++ b/tests/api/middleware/test_security.py @@ -215,6 +215,9 @@ def _versioned_path(prefix: str, path: str) -> str: ("post", "/addons/abc123/restart", {"admin", "manager"}), ("post", "/addons/abc123/security", {"admin"}), ("post", "/os/datadisk/wipe", {"admin"}), + ("get", "/os/ssh/authorized_keys", set()), + ("post", "/os/ssh/authorized_keys", set()), + ("delete", "/os/ssh/authorized_keys", set()), ("post", "/addons/self/sys_options", set()), ("post", "/addons/abc123/sys_options", set()), ], @@ -260,6 +263,13 @@ async def test_home_assistant_paths( ) assert resp.status == 200 + for method in ("get", "post", "delete"): + resp = await getattr(client, method)( + _versioned_path(prefix, "/os/ssh/authorized_keys"), + headers={"Authorization": "Bearer abc123"}, + ) + assert resp.status == 200 + @pytest.mark.usefixtures("plugin_tokens") async def test_blacklist( diff --git a/tests/api/test_os.py b/tests/api/test_os.py index db14a21b1..801db1b66 100644 --- a/tests/api/test_os.py +++ b/tests/api/test_os.py @@ -1,6 +1,6 @@ """Test OS API.""" -from unittest.mock import Mock, PropertyMock, patch +from unittest.mock import AsyncMock, Mock, PropertyMock, patch from aiohttp.test_utils import TestClient from awesomeversion import AwesomeVersion @@ -12,7 +12,7 @@ from supervisor.coresys import CoreSys from supervisor.dbus.agent import OSAgent from supervisor.dbus.agent.boards import BoardManager from supervisor.dbus.agent.boards.interface import BoardProxy -from supervisor.exceptions import DBusError as SupervisorDBusError +from supervisor.exceptions import DBusError as SupervisorDBusError, HostError from supervisor.host.control import SystemControl from supervisor.os.manager import OSManager from supervisor.resolution.const import ContextType, IssueType, SuggestionType @@ -819,3 +819,274 @@ async def test_api_board_raspberrypi_firmware_unavailable_on_board( resp = await api_client.post(f"{prefix}/os/boards/raspberrypi/firmware/update") assert resp.status == 404 + + +TEST_SSH_KEY_ED25519 = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIDXD8u9KB94/l1YukYflKOsO7KzoSEQD4dNNlWY9zaQP test@example.com" + + +@pytest.mark.parametrize("os_agent_version", ["1.11.0"], indirect=True) +@pytest.mark.usefixtures("os_available", "os_agent_version") +async def test_api_os_ssh_authorized_keys_list( + api_client_with_prefix: tuple[TestClient, str], + os_agent_services: dict[str, DBusServiceMock], +): + """Test listing the SSH authorized keys.""" + api_client, prefix = api_client_with_prefix + system_service: SystemService = os_agent_services["agent_system"] + system_service.response_list_ssh_auth_keys = [ + TEST_SSH_KEY_ED25519, + "ssh-rsa AAAA imported@usb", + ] + + resp = await api_client.get(f"{prefix}/os/ssh/authorized_keys") + assert resp.status == 200 + result = await resp.json() + assert result["data"]["keys"] == [ + TEST_SSH_KEY_ED25519, + "ssh-rsa AAAA imported@usb", + ] + + +@pytest.mark.parametrize("os_agent_version", ["1.10.0"], indirect=True) +@pytest.mark.usefixtures("os_available", "os_agent_version") +async def test_api_os_ssh_authorized_keys_list_requires_os_agent_version( + api_client_with_prefix: tuple[TestClient, str], + os_agent_services: dict[str, DBusServiceMock], +): + """Test 404 is returned on an OS Agent without ListSSHAuthKeys.""" + api_client, prefix = api_client_with_prefix + + resp = await api_client.get(f"{prefix}/os/ssh/authorized_keys") + assert resp.status == 404 + result = await resp.json() + assert "OS Agent 1.11.0 or newer required" in result["message"] + + +@pytest.mark.usefixtures("os_available") +async def test_api_os_ssh_authorized_keys_add( + api_client_with_prefix: tuple[TestClient, str], + coresys: CoreSys, + os_agent_services: dict[str, DBusServiceMock], +): + """Test adding an SSH authorized key.""" + api_client, prefix = api_client_with_prefix + system_service: SystemService = os_agent_services["agent_system"] + system_service.AddSSHAuthKey.calls.clear() + system_service.ClearSSHAuthKeys.calls.clear() + + with patch.object(coresys.host.services, "start", new=AsyncMock()) as start: + resp = await api_client.post( + f"{prefix}/os/ssh/authorized_keys", + # Trailing newline from a pasted key is stripped before writing + json={"key": TEST_SSH_KEY_ED25519 + "\n"}, + ) + assert resp.status == 200 + + assert system_service.AddSSHAuthKey.calls == [(TEST_SSH_KEY_ED25519,)] + assert system_service.ClearSSHAuthKeys.calls == [] + # dropbear only starts if authorized_keys is non-empty when the unit + # starts, so a stopped service must be started after adding a key + start.assert_called_once_with("dropbear.service") + + +@pytest.mark.parametrize( + "body", + [ + {}, + {"key": [TEST_SSH_KEY_ED25519]}, + {"key": 42}, + {"key": ""}, + # Newline injection must not smuggle extra authorized_keys lines + {"key": f"{TEST_SSH_KEY_ED25519}\nssh-rsa evil"}, + {"key": TEST_SSH_KEY_ED25519.replace(" test@", "\x1b test@")}, + # dropbear ignores authorized_keys lines longer than 3000 bytes + {"key": f"{TEST_SSH_KEY_ED25519} {'a' * 3000}"}, + ], + ids=[ + "missing key", + "key is a list", + "key not a string", + "empty key", + "newline injection", + "control character", + "oversized key", + ], +) +@pytest.mark.usefixtures("os_available") +async def test_api_os_ssh_authorized_keys_add_invalid( + api_client_with_prefix: tuple[TestClient, str], + os_agent_services: dict[str, DBusServiceMock], + body: dict, +): + """Test malformed bodies are rejected before touching the host.""" + api_client, prefix = api_client_with_prefix + system_service: SystemService = os_agent_services["agent_system"] + system_service.AddSSHAuthKey.calls.clear() + + resp = await api_client.post(f"{prefix}/os/ssh/authorized_keys", json=body) + assert resp.status == 400 + + assert system_service.AddSSHAuthKey.calls == [] + + +@pytest.mark.usefixtures("os_available") +async def test_api_os_ssh_authorized_keys_add_rejected_key( + api_client_with_prefix: tuple[TestClient, str], + coresys: CoreSys, + os_agent_services: dict[str, DBusServiceMock], +): + """Test a key rejected by OS Agent validation is reported.""" + api_client, prefix = api_client_with_prefix + system_service: SystemService = os_agent_services["agent_system"] + system_service.response_add_ssh_auth_key = DBusError( + ErrorType.FAILED, "invalid SSH authorized key: ssh: no key found" + ) + + with patch.object(coresys.host.services, "start", new=AsyncMock()) as start: + resp = await api_client.post( + f"{prefix}/os/ssh/authorized_keys", json={"key": TEST_SSH_KEY_ED25519} + ) + assert resp.status == 400 + result = await resp.json() + assert "Can't add SSH authorized key" in result["message"] + assert "invalid SSH authorized key" in result["message"] + start.assert_not_called() + + +@pytest.mark.usefixtures("os_available") +async def test_api_os_ssh_authorized_keys_add_dropbear_start_error( + api_client_with_prefix: tuple[TestClient, str], + coresys: CoreSys, + os_agent_services: dict[str, DBusServiceMock], +): + """Test a dropbear start failure is reported after the key was written.""" + api_client, prefix = api_client_with_prefix + + with patch.object( + coresys.host.services, "start", new=AsyncMock(side_effect=HostError("boom")) + ): + resp = await api_client.post( + f"{prefix}/os/ssh/authorized_keys", json={"key": TEST_SSH_KEY_ED25519} + ) + assert resp.status == 400 + result = await resp.json() + assert "can't start dropbear" in result["message"] + + +@pytest.mark.usefixtures("os_available") +async def test_api_os_ssh_authorized_keys_clear( + api_client_with_prefix: tuple[TestClient, str], + coresys: CoreSys, + os_agent_services: dict[str, DBusServiceMock], +): + """Test clearing the SSH authorized keys.""" + api_client, prefix = api_client_with_prefix + system_service: SystemService = os_agent_services["agent_system"] + system_service.ClearSSHAuthKeys.calls.clear() + + with ( + patch.object(coresys.host.services, "start", new=AsyncMock()) as start, + patch.object(coresys.host.services, "stop", new=AsyncMock()) as stop, + ): + resp = await api_client.delete(f"{prefix}/os/ssh/authorized_keys") + assert resp.status == 200 + + assert system_service.ClearSSHAuthKeys.calls == [()] + start.assert_not_called() + # Established sessions only end when the service stops (revocation) + stop.assert_called_once_with("dropbear.service") + + +@pytest.mark.parametrize( + ("os_agent_version", "expected_status"), + [("1.9.0", 200), ("1.10.0", 400)], + indirect=["os_agent_version"], +) +@pytest.mark.usefixtures("os_available", "os_agent_version") +async def test_api_os_ssh_authorized_keys_clear_old_os_agent_missing_file( + api_client_with_prefix: tuple[TestClient, str], + coresys: CoreSys, + os_agent_services: dict[str, DBusServiceMock], + expected_status: int, +): + """Test the missing-file clear error is only tolerated on affected OS Agents. + + OS Agent before 1.10.0 returns an error when the file is already absent + (inverted error check); on 1.10.0 or newer the same error is genuine. + """ + api_client, prefix = api_client_with_prefix + system_service: SystemService = os_agent_services["agent_system"] + system_service.response_clear_ssh_auth_keys = DBusError( + ErrorType.FAILED, + "remove /root/.ssh/authorized_keys: no such file or directory", + ) + + with patch.object(coresys.host.services, "stop", new=AsyncMock()) as stop: + resp = await api_client.delete(f"{prefix}/os/ssh/authorized_keys") + assert resp.status == expected_status + + if expected_status == 200: + stop.assert_called_once_with("dropbear.service") + else: + result = await resp.json() + assert "Can't clear SSH authorized keys" in result["message"] + stop.assert_not_called() + + +@pytest.mark.usefixtures("os_available") +async def test_api_os_ssh_authorized_keys_clear_error( + api_client_with_prefix: tuple[TestClient, str], + coresys: CoreSys, + os_agent_services: dict[str, DBusServiceMock], +): + """Test a genuine clear failure is reported.""" + api_client, prefix = api_client_with_prefix + system_service: SystemService = os_agent_services["agent_system"] + system_service.response_clear_ssh_auth_keys = DBusError( + ErrorType.FAILED, "remove /root/.ssh/authorized_keys: permission denied" + ) + + with patch.object(coresys.host.services, "stop", new=AsyncMock()) as stop: + resp = await api_client.delete(f"{prefix}/os/ssh/authorized_keys") + assert resp.status == 400 + result = await resp.json() + assert "Can't clear SSH authorized keys" in result["message"] + stop.assert_not_called() + + +@pytest.mark.usefixtures("os_available") +async def test_api_os_ssh_authorized_keys_dropbear_stop_error( + api_client_with_prefix: tuple[TestClient, str], + coresys: CoreSys, + os_agent_services: dict[str, DBusServiceMock], +): + """Test a dropbear stop failure is reported after the keys were cleared.""" + api_client, prefix = api_client_with_prefix + + with patch.object( + coresys.host.services, "stop", new=AsyncMock(side_effect=HostError("boom")) + ): + resp = await api_client.delete(f"{prefix}/os/ssh/authorized_keys") + assert resp.status == 400 + result = await resp.json() + assert "can't stop dropbear" in result["message"] + + +@pytest.mark.parametrize( + ("method", "body"), + [("post", {"key": TEST_SSH_KEY_ED25519}), ("delete", None)], + ids=["add", "clear"], +) +async def test_api_os_ssh_authorized_keys_no_os( + api_client_with_prefix: tuple[TestClient, str], + method: str, + body: dict | None, +): + """Test SSH authorized keys endpoints require Home Assistant OS.""" + api_client, prefix = api_client_with_prefix + resp = await getattr(api_client, method)( + f"{prefix}/os/ssh/authorized_keys", json=body + ) + assert resp.status == 400 + result = await resp.json() + assert "no Home Assistant OS available" in result["message"] diff --git a/tests/dbus/agent/test_system.py b/tests/dbus/agent/test_system.py index e747032fa..526c98cd1 100644 --- a/tests/dbus/agent/test_system.py +++ b/tests/dbus/agent/test_system.py @@ -32,3 +32,44 @@ async def test_dbus_osagent_system_wipe( assert await os_agent.system.schedule_wipe_device() is True assert system_service.ScheduleWipeDevice.calls == [()] + + +async def test_dbus_osagent_system_ssh_auth_keys( + system_service: SystemService, dbus_session_bus: MessageBus +): + """Test add and clear of SSH authorized keys on host.""" + system_service.AddSSHAuthKey.calls.clear() + system_service.ClearSSHAuthKeys.calls.clear() + os_agent = OSAgent() + key = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIDXD8u9KB94/l1YukYflKOsO7KzoSEQD4dNNlWY9zaQP test@example.com" + + with pytest.raises(DBusNotConnectedError): + await os_agent.system.add_ssh_auth_key(key) + + with pytest.raises(DBusNotConnectedError): + await os_agent.system.clear_ssh_auth_keys() + + await os_agent.connect(dbus_session_bus) + + await os_agent.system.clear_ssh_auth_keys() + await os_agent.system.add_ssh_auth_key(key) + + assert system_service.ClearSSHAuthKeys.calls == [()] + assert system_service.AddSSHAuthKey.calls == [(key,)] + + +async def test_dbus_osagent_system_list_ssh_auth_keys( + system_service: SystemService, dbus_session_bus: MessageBus +): + """Test listing SSH authorized keys on host.""" + system_service.response_list_ssh_auth_keys = ["ssh-ed25519 AAAA test@example.com"] + os_agent = OSAgent() + + with pytest.raises(DBusNotConnectedError): + await os_agent.system.list_ssh_auth_keys() + + await os_agent.connect(dbus_session_bus) + + assert await os_agent.system.list_ssh_auth_keys() == [ + "ssh-ed25519 AAAA test@example.com" + ] diff --git a/tests/dbus_service_mocks/agent_system.py b/tests/dbus_service_mocks/agent_system.py index 70a60c033..417cfd22a 100644 --- a/tests/dbus_service_mocks/agent_system.py +++ b/tests/dbus_service_mocks/agent_system.py @@ -22,6 +22,9 @@ class System(DBusServiceMock): interface = "io.hass.os.System" response_schedule_wipe_device: bool | DBusError = True response_migrate_docker_storage_driver: None | DBusError = None + response_add_ssh_auth_key: None | DBusError = None + response_clear_ssh_auth_keys: None | DBusError = None + response_list_ssh_auth_keys: list[str] | DBusError = [] @dbus_method() def ScheduleWipeDevice(self) -> "b": @@ -40,3 +43,22 @@ class System(DBusServiceMock): ErrorType.FAILED, f"unsupported driver: {backend} (only 'overlayfs' is currently supported)", ) + + @dbus_method() + def AddSSHAuthKey(self, key: "s") -> None: + """Add SSH authorized key.""" + if isinstance(self.response_add_ssh_auth_key, DBusError): + raise self.response_add_ssh_auth_key # pylint: disable=raising-bad-type + + @dbus_method() + def ClearSSHAuthKeys(self) -> None: + """Clear SSH authorized keys.""" + if isinstance(self.response_clear_ssh_auth_keys, DBusError): + raise self.response_clear_ssh_auth_keys # pylint: disable=raising-bad-type + + @dbus_method() + def ListSSHAuthKeys(self) -> "as": + """List SSH authorized keys.""" + if isinstance(self.response_list_ssh_auth_keys, DBusError): + raise self.response_list_ssh_auth_keys # pylint: disable=raising-bad-type + return self.response_list_ssh_auth_keys diff --git a/tests/os/test_manager.py b/tests/os/test_manager.py index ca4f1ccf8..69191bb72 100644 --- a/tests/os/test_manager.py +++ b/tests/os/test_manager.py @@ -1,5 +1,6 @@ """Test Home Assistant OS functionality.""" +import asyncio from pathlib import Path from unittest.mock import AsyncMock, PropertyMock, call, patch @@ -454,3 +455,55 @@ async def test_config_sync_service_name( await coresys.os.config_sync() restart.assert_called_once_with(expected_service) + + +@pytest.mark.usefixtures("os_available") +async def test_ssh_authorized_keys_jobs_serialized(coresys: CoreSys): + """Test concurrent SSH key add and clear do not interleave. + + Interleaved file and service operations could end with the service + state not matching the key state (e.g. a key configured but dropbear + stopped). + """ + events: list[str] = [] + + def record(name: str): + async def _record(*args): + events.append(f"{name}-begin") + await asyncio.sleep(0.01) + events.append(f"{name}-end") + + return _record + + with ( + patch.object( + type(coresys.dbus.agent.system), "add_ssh_auth_key", new=record("add-key") + ), + patch.object( + type(coresys.dbus.agent.system), + "clear_ssh_auth_keys", + new=record("clear-keys"), + ), + patch.object( + coresys.host.services, "start", new=AsyncMock(side_effect=record("start")) + ), + patch.object( + coresys.host.services, "stop", new=AsyncMock(side_effect=record("stop")) + ), + ): + await asyncio.gather( + coresys.os.add_ssh_authorized_key("ssh-ed25519 AAAA test@example.com"), + coresys.os.clear_ssh_authorized_keys(), + ) + + # Each operation's file change and service action must be contiguous + assert events == [ + "add-key-begin", + "add-key-end", + "start-begin", + "start-end", + "clear-keys-begin", + "clear-keys-end", + "stop-begin", + "stop-end", + ]