Home-Assistant/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2026-09-30 12:58:49 +01:00
Code Activity
Files
2d153ac49713457ecefed88db7d38e7b7ae820fd
supervisor/tests/api/test_os.py
T
Stefan Agner 210b49fb03 Add API to manage SSH authorized keys on Home Assistant OS (#7039) 
* 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.
2026-08-27 10:12:44 +02:00

1093 lines
38 KiB
Python
Raw Blame History

"""Test OS API."""
from unittest.mock import AsyncMock, Mock, PropertyMock, patch
from aiohttp.test_utils import TestClient
from awesomeversion import AwesomeVersion
from dbus_fast import DBusError, ErrorType
import pytest
from supervisor.const import CoreState
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, HostError
from supervisor.host.control import SystemControl
from supervisor.os.manager import OSManager
from supervisor.resolution.const import ContextType, IssueType, SuggestionType
from supervisor.resolution.data import Issue, Suggestion
from tests.common import mock_dbus_services
from tests.dbus_service_mocks.agent_boards import Boards as BoardsService
from tests.dbus_service_mocks.agent_boards_green import Green as GreenService
from tests.dbus_service_mocks.agent_boards_rpi_firmware import (
RPiFirmware as RPiFirmwareService,
)
from tests.dbus_service_mocks.agent_boards_yellow import Yellow as YellowService
from tests.dbus_service_mocks.agent_datadisk import DataDisk as DataDiskService
from tests.dbus_service_mocks.agent_swap import Swap as SwapService
from tests.dbus_service_mocks.agent_system import System as SystemService
from tests.dbus_service_mocks.base import DBusServiceMock
from tests.dbus_service_mocks.rauc import Rauc as RaucService
@pytest.fixture(name="boards_service")
async def fixture_boards_service(
os_agent_services: dict[str, DBusServiceMock],
) -> BoardsService:
"""Return mock Boards service."""
return os_agent_services["agent_boards"]
@pytest.fixture
async def os_agent_version(request: pytest.FixtureRequest) -> None:
"""Mock OS Agent version."""
version = (
AwesomeVersion(request.param)
if hasattr(request, "param")
else AwesomeVersion("1.9.0")
)
with patch.object(OSAgent, "version", new=PropertyMock(return_value=version)):
yield
async def test_api_os_info(api_client_with_prefix: tuple[TestClient, str]):
"""Test os info api."""
api_client, prefix = api_client_with_prefix
resp = await api_client.get(f"{prefix}/os/info")
result = await resp.json()
for attr in (
"version",
"version_latest",
"version_pending",
"update_available",
"board",
"boot",
"data_disk",
"boot_slots",
):
assert attr in result["data"]
async def test_api_os_info_with_agent(
api_client_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
"""Test os info api for data disk."""
api_client, prefix = api_client_with_prefix
resp = await api_client.get(f"{prefix}/os/info")
result = await resp.json()
assert result["data"]["data_disk"] == "BJTD4R-0x97cde291"
async def test_api_os_info_boot_slots(
api_client_with_prefix: tuple[TestClient, str], coresys: CoreSys, os_available
):
"""Test os info api for boot slots."""
api_client, prefix = api_client_with_prefix
await coresys.os.load()
resp = await api_client.get(f"{prefix}/os/info")
result = await resp.json()
assert result["data"]["boot_slots"] == {
"A": {
"state": "inactive",
"status": "good",
"version": "9.0.dev20220818",
},
"B": {"state": "booted", "status": "good", "version": "9.0.dev20220824"},
}
@pytest.mark.parametrize(
"new_disk",
["/dev/sdaaaa", "/dev/mmcblk1", "Generic-Flash-Disk-61BCDDB6"],
ids=["non-existent", "unavailable drive by path", "unavailable drive by id"],
)
async def test_api_os_datadisk_move_fail(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
new_disk: str,
os_available,
):
"""Test datadisk move to non-existent or invalid devices."""
api_client, prefix = api_client_with_prefix
resp = await api_client.post(
f"{prefix}/os/datadisk/move", json={"device": new_disk}
)
result = await resp.json()
assert result["message"] == f"'{new_disk}' not a valid data disk target!"
async def test_api_os_datadisk_list(
api_client_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
"""Test datadisk list function."""
api_client, prefix = api_client_with_prefix
resp = await api_client.get(f"{prefix}/os/datadisk/list")
result = await resp.json()
assert result["data"]["devices"] == ["SSK-SSK-Storage-DF56419883D56"]
assert result["data"]["disks"] == [
{
"vendor": "SSK",
"model": "SSK Storage",
"serial": "DF56419883D56",
"id": "SSK-SSK-Storage-DF56419883D56",
"size": 250059350016,
"dev_path": "/dev/sda",
"name": "SSK SSK Storage (DF56419883D56)",
}
]
@pytest.mark.parametrize(
"new_disk",
["SSK-SSK-Storage-DF56419883D56", "/dev/sda"],
ids=["by drive id", "by device path"],
)
async def test_api_os_datadisk_migrate(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
os_agent_services: dict[str, DBusServiceMock],
new_disk: str,
os_available,
):
"""Test migrating datadisk."""
api_client, prefix = api_client_with_prefix
datadisk_service: DataDiskService = os_agent_services["agent_datadisk"]
datadisk_service.ChangeDevice.calls.clear()
with patch.object(SystemControl, "reboot") as reboot:
resp = await api_client.post(
f"{prefix}/os/datadisk/move", json={"device": new_disk}
)
assert resp.status == 200
assert datadisk_service.ChangeDevice.calls == [("/dev/sda",)]
reboot.assert_called_once()
async def test_api_os_datadisk_wipe(
api_client_with_prefix: tuple[TestClient, str],
os_agent_services: dict[str, DBusServiceMock],
os_available,
):
"""Test datadisk wipe."""
api_client, prefix = api_client_with_prefix
system_service: SystemService = os_agent_services["agent_system"]
system_service.ScheduleWipeDevice.calls.clear()
with patch.object(SystemControl, "reboot") as reboot:
resp = await api_client.post(f"{prefix}/os/datadisk/wipe")
assert resp.status == 200
assert system_service.ScheduleWipeDevice.calls == [()]
reboot.assert_called_once()
async def test_api_set_boot_slot(
api_client_with_prefix: tuple[TestClient, str],
all_dbus_services: dict[str, DBusServiceMock],
coresys: CoreSys,
os_available,
):
"""Test changing the boot slot via API."""
api_client, prefix = api_client_with_prefix
rauc_service: RaucService = all_dbus_services["rauc"]
rauc_service.Mark.calls.clear()
await coresys.os.load()
with patch.object(SystemControl, "reboot") as reboot:
resp = await api_client.post(f"{prefix}/os/boot-slot", json={"boot_slot": "A"})
assert resp.status == 200
reboot.assert_called_once()
assert rauc_service.Mark.calls == [("active", "kernel.0")]
async def test_api_set_boot_slot_invalid(
api_client_with_prefix: tuple[TestClient, str],
):
"""Test invalid calls to set boot slot."""
api_client, prefix = api_client_with_prefix
resp = await api_client.post(f"{prefix}/os/boot-slot", json={"boot_slot": "C"})
assert resp.status == 400
result = await resp.json()
assert "expected BootSlot or one of 'A', 'B'" in result["message"]
resp = await api_client.post(f"{prefix}/os/boot-slot", json={"boot_slot": "A"})
assert resp.status == 400
result = await resp.json()
assert "no Home Assistant OS available" in result["message"]
async def test_api_set_boot_slot_error(
api_client_with_prefix: tuple[TestClient, str],
all_dbus_services: dict[str, DBusServiceMock],
coresys: CoreSys,
capture_exception: Mock,
os_available,
):
"""Test changing the boot slot via API."""
api_client, prefix = api_client_with_prefix
rauc_service: RaucService = all_dbus_services["rauc"]
rauc_service.response_mark = DBusError(ErrorType.FAILED, "fail")
await coresys.os.load()
resp = await api_client.post(f"{prefix}/os/boot-slot", json={"boot_slot": "A"})
assert resp.status == 400
result = await resp.json()
assert result["message"] == "Can't mark A as active!"
capture_exception.assert_called_once()
async def test_api_os_update_no_auto_reboot_creates_issue(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
tmp_supervisor_data,
path_extern: None,
supervisor_internet,
os_available,
):
"""Successful OS update does not auto-reboot and creates a reboot issue."""
api_client, prefix = api_client_with_prefix
await coresys.core.set_state(CoreState.RUNNING)
async def fake_download(url, raucb) -> None:
raucb.touch()
with (
patch.object(
coresys.os,
"_get_download_url",
return_value="https://example.invalid/update.raucb",
),
patch.object(coresys.os, "_download_raucb", side_effect=fake_download),
patch.object(coresys.host.control, "reboot") as reboot,
):
resp = await api_client.post(f"{prefix}/os/update", json={"version": "13.0"})
assert resp.status == 200
reboot.assert_not_called()
assert (
Issue(IssueType.REBOOT_REQUIRED, ContextType.SYSTEM)
in coresys.resolution.issues
)
assert (
Suggestion(SuggestionType.EXECUTE_REBOOT, ContextType.SYSTEM)
in coresys.resolution.suggestions
)
resp = await api_client.get(f"{prefix}/os/info")
result = await resp.json()
assert result["data"]["version_pending"] == "13.0"
# Installed update is reported as current version so Core releases
# unaware of version_pending don't offer the update again
assert result["data"]["version"] == "13.0"
# Same for Core versions predating version_pending support
coresys.homeassistant.version = AwesomeVersion("2026.7.3")
resp = await api_client.get(f"{prefix}/os/info")
result = await resp.json()
assert result["data"]["version_pending"] == "13.0"
assert result["data"]["version"] == "13.0"
# Core versions consuming version_pending get the real current version
coresys.homeassistant.version = AwesomeVersion("2026.8.0.dev202607250310")
resp = await api_client.get(f"{prefix}/os/info")
result = await resp.json()
assert result["data"]["version_pending"] == "13.0"
assert result["data"]["version"] == "16.2"
async def test_api_os_update_failure_no_reboot_no_issue(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
tmp_supervisor_data,
path_extern: None,
supervisor_internet,
os_available,
):
"""Failed OS update does not auto-reboot and does not create reboot issue."""
api_client, prefix = api_client_with_prefix
await coresys.core.set_state(CoreState.RUNNING)
async def fake_download(url, raucb) -> None:
raucb.touch()
with (
patch.object(
coresys.os,
"_get_download_url",
return_value="https://example.invalid/update.raucb",
),
patch.object(coresys.os, "_download_raucb", side_effect=fake_download),
patch.object(
coresys.dbus.rauc,
"install",
side_effect=SupervisorDBusError("fail"),
),
patch.object(coresys.host.control, "reboot") as reboot,
):
resp = await api_client.post(f"{prefix}/os/update", json={"version": "13.0"})
assert resp.status == 400
result = await resp.json()
assert result["message"] == "Rauc communication error"
reboot.assert_not_called()
assert (
Issue(IssueType.REBOOT_REQUIRED, ContextType.SYSTEM)
not in coresys.resolution.issues
)
assert (
Suggestion(SuggestionType.EXECUTE_REBOOT, ContextType.SYSTEM)
not in coresys.resolution.suggestions
)
async def test_api_board_yellow_info(
api_client_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
"""Test yellow board info."""
api_client, prefix = api_client_with_prefix
resp = await api_client.get(f"{prefix}/os/boards/yellow")
assert resp.status == 200
result = await resp.json()
assert result["data"]["disk_led"] is True
assert result["data"]["heartbeat_led"] is True
assert result["data"]["power_led"] is True
assert (await api_client.get(f"{prefix}/os/boards/green")).status == 400
assert (await api_client.get(f"{prefix}/os/boards/supervised")).status == 400
assert (await api_client.get(f"{prefix}/os/boards/not-real")).status == 400
async def test_api_board_yellow_options(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
os_agent_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]],
):
"""Test yellow board options."""
api_client, prefix = api_client_with_prefix
yellow_service: YellowService = os_agent_services["agent_boards_yellow"]
assert coresys.dbus.agent.board.yellow.disk_led is True
assert coresys.dbus.agent.board.yellow.heartbeat_led is True
assert coresys.dbus.agent.board.yellow.power_led is True
assert len(coresys.resolution.issues) == 0
with patch.object(BoardProxy, "save_data") as save_data:
resp = await api_client.post(
f"{prefix}/os/boards/yellow",
json={"disk_led": False, "heartbeat_led": False, "power_led": False},
)
assert resp.status == 200
save_data.assert_called_once()
await yellow_service.ping()
assert coresys.dbus.agent.board.yellow.disk_led is False
assert coresys.dbus.agent.board.yellow.heartbeat_led is False
assert coresys.dbus.agent.board.yellow.power_led is False
assert (
Issue(IssueType.REBOOT_REQUIRED, ContextType.SYSTEM)
in coresys.resolution.issues
)
assert (
Suggestion(SuggestionType.EXECUTE_REBOOT, ContextType.SYSTEM)
in coresys.resolution.suggestions
)
async def test_api_board_green_info(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
boards_service: BoardsService,
):
"""Test green board info."""
api_client, prefix = api_client_with_prefix
await mock_dbus_services({"agent_boards_green": None}, coresys.dbus.bus)
boards_service.board = "Green"
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
resp = await api_client.get(f"{prefix}/os/boards/green")
assert resp.status == 200
result = await resp.json()
assert result["data"]["activity_led"] is True
assert result["data"]["power_led"] is True
assert result["data"]["system_health_led"] is True
assert (await api_client.get(f"{prefix}/os/boards/yellow")).status == 400
assert (await api_client.get(f"{prefix}/os/boards/supervised")).status == 400
assert (await api_client.get(f"{prefix}/os/boards/not-real")).status == 400
async def test_api_board_green_options(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
boards_service: BoardsService,
):
"""Test yellow board options."""
api_client, prefix = api_client_with_prefix
green_service: GreenService = (
await mock_dbus_services({"agent_boards_green": None}, coresys.dbus.bus)
)["agent_boards_green"]
boards_service.board = "Green"
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
assert coresys.dbus.agent.board.green.activity_led is True
assert coresys.dbus.agent.board.green.power_led is True
assert coresys.dbus.agent.board.green.user_led is True
assert len(coresys.resolution.issues) == 0
with patch.object(BoardProxy, "save_data") as save_data:
resp = await api_client.post(
f"{prefix}/os/boards/green",
json={
"activity_led": False,
"power_led": False,
"system_health_led": False,
},
)
assert resp.status == 200
save_data.assert_called_once()
await green_service.ping()
assert coresys.dbus.agent.board.green.activity_led is False
assert coresys.dbus.agent.board.green.power_led is False
assert coresys.dbus.agent.board.green.user_led is False
assert len(coresys.resolution.issues) == 0
async def test_api_board_supervised_info(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
boards_service: BoardsService,
):
"""Test supervised board info."""
api_client, prefix = api_client_with_prefix
await mock_dbus_services({"agent_boards_supervised": None}, coresys.dbus.bus)
boards_service.board = "Supervised"
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
with patch("supervisor.os.manager.CPE.get_product", return_value=["not-hassos"]):
await coresys.os.load()
assert (await api_client.get(f"{prefix}/os/boards/supervised")).status == 200
assert (
await api_client.post(f"{prefix}/os/boards/supervised", json={})
).status == 405
assert (await api_client.get(f"{prefix}/os/boards/yellow")).status == 400
assert (await api_client.get(f"{prefix}/os/boards/not-real")).status == 400
async def test_api_board_other_info(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
boards_service: BoardsService,
):
"""Test info for other board without dbus object."""
api_client, prefix = api_client_with_prefix
boards_service.board = "not-real"
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
with patch.object(OSManager, "board", new=PropertyMock(return_value="not-real")):
assert (await api_client.get(f"{prefix}/os/boards/not-real")).status == 200
assert (
await api_client.post(f"{prefix}/os/boards/not-real", json={})
).status == 405
assert (await api_client.get(f"{prefix}/os/boards/yellow")).status == 400
assert (await api_client.get(f"{prefix}/os/boards/supervised")).status == 400
@pytest.mark.parametrize("os_available", ["15.0"], indirect=True)
async def test_api_config_swap_info(
api_client_with_prefix: tuple[TestClient, str], coresys: CoreSys, os_available
):
"""Test swap info."""
api_client, prefix = api_client_with_prefix
await coresys.dbus.agent.swap.connect(coresys.dbus.bus)
resp = await api_client.get(f"{prefix}/os/config/swap")
assert resp.status == 200
result = await resp.json()
assert result["data"]["swap_size"] == "1M"
assert result["data"]["swappiness"] == 1
@pytest.mark.parametrize("os_available", ["15.0"], indirect=True)
async def test_api_config_swap_options(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
os_agent_services: dict[str, DBusServiceMock],
os_available,
):
"""Test swap setting."""
api_client, prefix = api_client_with_prefix
swap_service: SwapService = os_agent_services["agent_swap"]
await coresys.dbus.agent.swap.connect(coresys.dbus.bus)
assert coresys.dbus.agent.swap.swap_size == "1M"
assert coresys.dbus.agent.swap.swappiness == 1
resp = await api_client.post(
f"{prefix}/os/config/swap",
json={
"swap_size": "2M",
"swappiness": 10,
},
)
assert resp.status == 200
await swap_service.ping()
assert coresys.dbus.agent.swap.swap_size == "2M"
assert coresys.dbus.agent.swap.swappiness == 10
assert (
Issue(IssueType.REBOOT_REQUIRED, ContextType.SYSTEM)
in coresys.resolution.issues
)
assert (
Suggestion(SuggestionType.EXECUTE_REBOOT, ContextType.SYSTEM)
in coresys.resolution.suggestions
)
# test setting only the swap size
resp = await api_client.post(
f"{prefix}/os/config/swap",
json={
"swap_size": "10M",
},
)
assert resp.status == 200
await swap_service.ping()
assert coresys.dbus.agent.swap.swap_size == "10M"
assert coresys.dbus.agent.swap.swappiness == 10
# test setting only the swappiness
resp = await api_client.post(
f"{prefix}/os/config/swap",
json={
"swappiness": 100,
},
)
assert resp.status == 200
await swap_service.ping()
assert coresys.dbus.agent.swap.swap_size == "10M"
assert coresys.dbus.agent.swap.swappiness == 100
@pytest.mark.parametrize("os_available", ["15.0"], indirect=True)
async def test_api_config_swap_options_no_reboot(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
os_agent_services: dict[str, DBusServiceMock],
os_available,
):
"""Test no resolution is shown when setting are submitted empty or unchanged."""
api_client, prefix = api_client_with_prefix
await coresys.dbus.agent.swap.connect(coresys.dbus.bus)
# empty options
resp = await api_client.post(
f"{prefix}/os/config/swap",
json={},
)
assert resp.status == 200
assert (
Issue(IssueType.REBOOT_REQUIRED, ContextType.SYSTEM)
not in coresys.resolution.issues
)
assert (
Suggestion(SuggestionType.EXECUTE_REBOOT, ContextType.SYSTEM)
not in coresys.resolution.suggestions
)
# no change
resp = await api_client.post(
f"{prefix}/os/config/swap",
json={
"swappiness": coresys.dbus.agent.swap.swappiness,
"swap_size": coresys.dbus.agent.swap.swap_size,
},
)
assert resp.status == 200
assert (
Issue(IssueType.REBOOT_REQUIRED, ContextType.SYSTEM)
not in coresys.resolution.issues
)
assert (
Suggestion(SuggestionType.EXECUTE_REBOOT, ContextType.SYSTEM)
not in coresys.resolution.suggestions
)
async def test_api_config_swap_not_os(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
os_agent_services: dict[str, DBusServiceMock],
):
"""Test 404 is returned for swap endpoints if not running on HAOS."""
api_client, prefix = api_client_with_prefix
await coresys.dbus.agent.swap.connect(coresys.dbus.bus)
resp = await api_client.get(f"{prefix}/os/config/swap")
assert resp.status == 404
resp = await api_client.post(
f"{prefix}/os/config/swap",
json={
"swap_size": "2M",
"swappiness": 10,
},
)
assert resp.status == 404
@pytest.mark.parametrize("os_available", ["14.2"], indirect=True)
async def test_api_config_swap_old_os(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
os_agent_services: dict[str, DBusServiceMock],
os_available,
):
"""Test 404 is returned for swap endpoints if OS is older than 15.0."""
api_client, prefix = api_client_with_prefix
await coresys.dbus.agent.swap.connect(coresys.dbus.bus)
resp = await api_client.get(f"{prefix}/os/config/swap")
assert resp.status == 404
resp = await api_client.post(
f"{prefix}/os/config/swap",
json={
"swap_size": "2M",
"swappiness": 10,
},
)
assert resp.status == 404
@pytest.mark.usefixtures("os_agent_version")
async def test_api_board_raspberrypi_info(
api_client_with_prefix: tuple[TestClient, str],
os_agent_services: dict[str, DBusServiceMock],
os_available,
):
"""Test Raspberry Pi firmware info endpoint."""
api_client, prefix = api_client_with_prefix
resp = await api_client.get(f"{prefix}/os/boards/raspberrypi/firmware")
assert resp.status == 200
result = await resp.json()
assert result["data"] == {
"current_version": "1618412973",
"latest_version": "1700000000",
"update_available": True,
"update_blocked": False,
"update_pending": False,
"blocked_reason": None,
}
@pytest.mark.usefixtures("os_agent_version")
async def test_api_board_raspberrypi_info_blocked_creates_issue(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
os_agent_services: dict[str, DBusServiceMock],
os_available,
):
"""GET on raspberrypi info while blocked raises a repair issue."""
api_client, prefix = api_client_with_prefix
rpi_service: RPiFirmwareService = os_agent_services["agent_boards_rpi_firmware"]
rpi_service.set_state(update_blocked=True, blocked_reason="unsupported_boot_device")
await coresys.dbus.agent.board.rpi_firmware.update()
resp = await api_client.get(f"{prefix}/os/boards/raspberrypi/firmware")
assert resp.status == 200
assert (
Issue(IssueType.RPI_FIRMWARE_UPDATE_BLOCKED, ContextType.SYSTEM)
in coresys.resolution.issues
)
# Clearing the blocked state dismisses the issue on the next info call.
rpi_service.set_state(update_blocked=False, blocked_reason="")
await coresys.dbus.agent.board.rpi_firmware.update()
resp = await api_client.get(f"{prefix}/os/boards/raspberrypi/firmware")
assert resp.status == 200
assert (
Issue(IssueType.RPI_FIRMWARE_UPDATE_BLOCKED, ContextType.SYSTEM)
not in coresys.resolution.issues
)
@pytest.mark.usefixtures("os_agent_version")
async def test_api_board_raspberrypi_update(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
os_agent_services: dict[str, DBusServiceMock],
os_available,
):
"""Successful firmware update creates REBOOT_REQUIRED issue."""
api_client, prefix = api_client_with_prefix
rpi_service: RPiFirmwareService = os_agent_services["agent_boards_rpi_firmware"]
assert rpi_service.update_called is False
resp = await api_client.post(f"{prefix}/os/boards/raspberrypi/firmware/update")
assert resp.status == 200
await rpi_service.ping()
assert rpi_service.update_called is True
assert (
Issue(IssueType.REBOOT_REQUIRED, ContextType.SYSTEM)
in coresys.resolution.issues
)
assert (
Suggestion(SuggestionType.EXECUTE_REBOOT, ContextType.SYSTEM)
in coresys.resolution.suggestions
)
@pytest.mark.usefixtures("os_agent_version")
async def test_api_board_raspberrypi_update_blocked(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
os_agent_services: dict[str, DBusServiceMock],
os_available,
):
"""POST to raspberrypi update on a blocked device returns an error and surfaces the repair issue."""
api_client, prefix = api_client_with_prefix
rpi_service: RPiFirmwareService = os_agent_services["agent_boards_rpi_firmware"]
rpi_service.set_state(update_blocked=True, blocked_reason="unsupported_boot_device")
await coresys.dbus.agent.board.rpi_firmware.update()
resp = await api_client.post(f"{prefix}/os/boards/raspberrypi/firmware/update")
assert resp.status == 400
assert rpi_service.update_called is False
assert (
Issue(IssueType.RPI_FIRMWARE_UPDATE_BLOCKED, ContextType.SYSTEM)
in coresys.resolution.issues
)
# No reboot issue should be raised when the update was rejected.
assert (
Issue(IssueType.REBOOT_REQUIRED, ContextType.SYSTEM)
not in coresys.resolution.issues
)
@pytest.mark.parametrize("os_agent_version", ["1.8.0"], indirect=True)
async def test_api_board_raspberrypi_requires_os_agent_version(
api_client_with_prefix: tuple[TestClient, str],
os_agent_services: dict[str, DBusServiceMock],
os_available,
os_agent_version, # pylint: disable=redefined-outer-name
):
"""Test 404 is returned for raspberrypi endpoints on an OS Agent older than 1.9.0."""
api_client, prefix = api_client_with_prefix
resp = await api_client.get(f"{prefix}/os/boards/raspberrypi/firmware")
assert resp.status == 404
resp = await api_client.post(f"{prefix}/os/boards/raspberrypi/firmware/update")
assert resp.status == 404
@pytest.mark.usefixtures("os_agent_version")
async def test_api_board_raspberrypi_firmware_unavailable_on_board(
api_client_with_prefix: tuple[TestClient, str],
os_agent_services: dict[str, DBusServiceMock],
os_available,
):
"""Test 404 is returned for raspberrypi endpoints on a board without firmware."""
api_client, prefix = api_client_with_prefix
with patch.object(
BoardManager, "has_rpi_firmware", new=PropertyMock(return_value=False)
):
resp = await api_client.get(f"{prefix}/os/boards/raspberrypi/firmware")
assert resp.status == 404
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"]
View Git Blame Copy Permalink