diff --git a/supervisor/api/__init__.py b/supervisor/api/__init__.py index 045bc595bb1..096be48bbc6 100644 --- a/supervisor/api/__init__.py +++ b/supervisor/api/__init__.py @@ -305,6 +305,9 @@ 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.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 dc4ed6f834a..6655f5580ef 100644 --- a/supervisor/api/const.py +++ b/supervisor/api/const.py @@ -44,6 +44,8 @@ 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 93afa0d82cb..c3f2f1c3ccf 100644 --- a/supervisor/api/middleware/security.py +++ b/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/.+" diff --git a/supervisor/api/os.py b/supervisor/api/os.py index 8e4188e0e4f..ac0a7508087 100644 --- a/supervisor/api/os.py +++ b/supervisor/api/os.py @@ -47,6 +47,8 @@ ATTR_DEV_PATH, ATTR_DEVICE, ATTR_DISKS, + ATTR_KEY, + ATTR_KEYS, ATTR_MODEL, ATTR_STATUS, ATTR_SYSTEM_HEALTH_LED, @@ -61,6 +63,10 @@ # .Firmware D-Bus interface first shipped in this OS Agent release. RPI_FIRMWARE_MIN_OS_AGENT_VERSION: AwesomeVersion = AwesomeVersion("1.9.0") +# 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)}) @@ -89,6 +95,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_KEY = vol.Schema({vol.Required(ATTR_KEY): ssh_auth_key}) # pylint: enable=no-value-for-parameter @@ -153,6 +188,32 @@ 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_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 69f5ce6835d..200f3944faf 100644 --- a/supervisor/dbus/agent/system.py +++ b/supervisor/dbus/agent/system.py @@ -20,3 +20,25 @@ 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") + + @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 7ff573351dd..37d59dbf04a 100644 --- a/supervisor/os/manager.py +++ b/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,58 @@ 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_add_ssh_authorized_key", + conditions=[JobCondition.HAOS], + on_condition=HassOSJobError, + 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, + internal=True, + ) + async def clear_ssh_authorized_keys(self) -> None: + """Remove all SSH authorized keys of root on the host.""" + _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 diff --git a/tests/api/middleware/test_security.py b/tests/api/middleware/test_security.py index e2bb15b438c..19af85084ae 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", {"admin"}), + ("post", "/os/ssh/authorized_keys", {"admin"}), + ("delete", "/os/ssh/authorized_keys", {"admin"}), ("post", "/addons/self/sys_options", set()), ("post", "/addons/abc123/sys_options", set()), ], diff --git a/tests/api/test_os.py b/tests/api/test_os.py index e0f48fbcb0b..218ee8f89b3 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.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 @@ -805,3 +805,243 @@ 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: + resp = await api_client.delete(f"{prefix}/os/ssh/authorized_keys") + assert resp.status == 200 + + assert system_service.ClearSSHAuthKeys.calls == [()] + start.assert_not_called() + + +@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], + 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", + ) + + resp = await api_client.delete(f"{prefix}/os/ssh/authorized_keys") + assert resp.status == expected_status + + if expected_status == 400: + result = await resp.json() + assert "Can't clear SSH authorized keys" in result["message"] + + +@pytest.mark.usefixtures("os_available") +async def test_api_os_ssh_authorized_keys_clear_error( + api_client_with_prefix: tuple[TestClient, str], + 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" + ) + + 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"] + + +@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 e747032fab1..526c98cd118 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 70a60c033a6..417cfd22a88 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 @@ def MigrateDockerStorageDriver(self, backend: "s") -> None: 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