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", + ]