Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-04-02 08:26:41 +01:00
Code Activity

Verify Proxmox permissions when creating snapshots (#166547)

Browse Source
This commit is contained in:
Tom
2026-03-26 17:21:30 +01:00
committed by GitHub
parent 5620cfbfd8
commit 0a9d4ef138
6 changed files with 71 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
homeassistant/components/proxmoxve/button.py
View File

@@ -23,7 +23,7 @@ from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util import dt as dt_util
from .const import DOMAIN
from .const import DOMAIN, ProxmoxPermission
from .coordinator import ProxmoxConfigEntry, ProxmoxCoordinator, ProxmoxNodeData
from .entity import ProxmoxContainerEntity, ProxmoxNodeEntity, ProxmoxVMEntity
from .helpers import is_granted
@@ -34,6 +34,8 @@ class ProxmoxNodeButtonNodeEntityDescription(ButtonEntityDescription):
"""Class to hold Proxmox node button description."""
press_action: Callable[[ProxmoxCoordinator, str], None]
permission: ProxmoxPermission = ProxmoxPermission.POWER
permission_raise: str = "no_permission_node_power"
@dataclass(frozen=True, kw_only=True)
@@ -41,6 +43,8 @@ class ProxmoxVMButtonEntityDescription(ButtonEntityDescription):
"""Class to hold Proxmox VM button description."""
press_action: Callable[[ProxmoxCoordinator, str, int], None]
permission: ProxmoxPermission = ProxmoxPermission.POWER
permission_raise: str = "no_permission_vm_lxc_power"
@dataclass(frozen=True, kw_only=True)
@@ -48,6 +52,8 @@ class ProxmoxContainerButtonEntityDescription(ButtonEntityDescription):
"""Class to hold Proxmox container button description."""
press_action: Callable[[ProxmoxCoordinator, str, int], None]
permission: ProxmoxPermission = ProxmoxPermission.POWER
permission_raise: str = "no_permission_vm_lxc_power"
NODE_BUTTONS: tuple[ProxmoxNodeButtonNodeEntityDescription, ...] = (
@@ -156,6 +162,8 @@ VM_BUTTONS: tuple[ProxmoxVMButtonEntityDescription, ...] = (
)
)
),
permission=ProxmoxPermission.SNAPSHOT,
permission_raise="no_permission_snapshot",
entity_category=EntityCategory.CONFIG,
),
)
@@ -199,6 +207,8 @@ CONTAINER_BUTTONS: tuple[ProxmoxContainerButtonEntityDescription, ...] = (
)
)
),
permission=ProxmoxPermission.SNAPSHOT,
permission_raise="no_permission_snapshot",
entity_category=EntityCategory.CONFIG,
),
)
@@ -315,10 +325,15 @@ class ProxmoxNodeButtonEntity(ProxmoxNodeEntity, ProxmoxBaseButton):
async def _async_press_call(self) -> None:
"""Execute the node button action via executor."""
node_id = self._node_data.node["node"]
if not is_granted(self.coordinator.permissions, p_type="nodes", p_id=node_id):
if not is_granted(
self.coordinator.permissions,
p_type="nodes",
p_id=node_id,
permission=self.entity_description.permission,
):
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="no_permission_node_power",
translation_key=self.entity_description.permission_raise,
)
await self.hass.async_add_executor_job(
self.entity_description.press_action,
@@ -335,10 +350,15 @@ class ProxmoxVMButtonEntity(ProxmoxVMEntity, ProxmoxBaseButton):
async def _async_press_call(self) -> None:
"""Execute the VM button action via executor."""
vmid = self.vm_data["vmid"]
if not is_granted(self.coordinator.permissions, p_type="vms", p_id=vmid):
if not is_granted(
self.coordinator.permissions,
p_type="vms",
p_id=vmid,
permission=self.entity_description.permission,
):
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="no_permission_vm_lxc_power",
translation_key=self.entity_description.permission_raise,
)
await self.hass.async_add_executor_job(
self.entity_description.press_action,
@@ -357,10 +377,15 @@ class ProxmoxContainerButtonEntity(ProxmoxContainerEntity, ProxmoxBaseButton):
"""Execute the container button action via executor."""
vmid = self.container_data["vmid"]
# Container power actions fall under vms
if not is_granted(self.coordinator.permissions, p_type="vms", p_id=vmid):
if not is_granted(
self.coordinator.permissions,
p_type="vms",
p_id=vmid,
permission=self.entity_description.permission,
):
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="no_permission_vm_lxc_power",
translation_key=self.entity_description.permission_raise,
)
await self.hass.async_add_executor_job(
self.entity_description.press_action,

9
homeassistant/components/proxmoxve/const.py
View File

@@ -1,5 +1,7 @@
"""Constants for ProxmoxVE."""
from enum import StrEnum
DOMAIN = "proxmoxve"
CONF_AUTH_METHOD = "auth_method"
CONF_REALM = "realm"
@@ -33,4 +35,9 @@ TYPE_VM = 0
TYPE_CONTAINER = 1
UPDATE_INTERVAL = 60
PERM_POWER = "VM.PowerMgmt"
class ProxmoxPermission(StrEnum):
"""Proxmox permissions."""
POWER = "VM.PowerMgmt"
SNAPSHOT = "VM.Snapshot"

4
homeassistant/components/proxmoxve/helpers.py
View File

@@ -1,13 +1,13 @@
"""Helpers for Proxmox VE."""
from .const import PERM_POWER
from .const import ProxmoxPermission
def is_granted(
permissions: dict[str, dict[str, int]],
p_type: str = "vms",
p_id: str | int | None = None, # can be str for nodes
permission: str = PERM_POWER,
permission: ProxmoxPermission = ProxmoxPermission.POWER,
) -> bool:
"""Validate user permissions for the given type and permission."""
paths = [f"/{p_type}/{p_id}", f"/{p_type}", "/"]

3
homeassistant/components/proxmoxve/strings.json
View File

@@ -315,6 +315,9 @@
"no_permission_node_power": {
"message": "The configured Proxmox VE user does not have permission to manage the power state of nodes. Please grant the user the 'VM.PowerMgmt' permission and try again."
},
"no_permission_snapshot": {
"message": "The configured Proxmox VE user does not have permission to create snapshots of VMs and containers. Please grant the user the 'VM.Snapshot' permission and try again."
},
"no_permission_vm_lxc_power": {
"message": "The configured Proxmox VE user does not have permission to manage the power state of VMs and containers. Please grant the user the 'VM.PowerMgmt' permission and try again."
},

15
tests/components/proxmoxve/__init__.py
View File

@@ -31,9 +31,20 @@ POWER_PERMISSIONS = {
"/vms/101": {"VM.PowerMgmt": 0},
}
SNAPSHOT_PERMISSIONS = {
"/vms": {"VM.Snapshot": 1},
"/vms/101": {"VM.Snapshot": 0},
}
MERGED_PERMISSIONS = {
key: {**AUDIT_PERMISSIONS.get(key, {}), **POWER_PERMISSIONS.get(key, {})}
for key in set(AUDIT_PERMISSIONS) | set(POWER_PERMISSIONS)
key: {
**AUDIT_PERMISSIONS.get(key, {}),
**POWER_PERMISSIONS.get(key, {}),
**SNAPSHOT_PERMISSIONS.get(key, {}),
}
for key in set(AUDIT_PERMISSIONS)
| set(POWER_PERMISSIONS)
| set(SNAPSHOT_PERMISSIONS)
}

15
tests/components/proxmoxve/test_button.py
View File

@@ -370,6 +370,7 @@ async def test_container_buttons_exceptions(
("button.pve1_start_all", "no_permission_node_power"),
("button.ct_nginx_start", "no_permission_vm_lxc_power"),
("button.vm_web_start", "no_permission_vm_lxc_power"),
("button.vm_web_create_snapshot", "no_permission_snapshot"),
],
)
async def test_node_buttons_permission_denied_for_auditor_role(
@@ -394,19 +395,29 @@ async def test_node_buttons_permission_denied_for_auditor_role(
assert exc_info.value.translation_key == translation_key
@pytest.mark.parametrize(
("entity_id", "translation_key"),
[
("button.vm_db_start", "no_permission_vm_lxc_power"),
("button.vm_db_create_snapshot", "no_permission_snapshot"),
],
)
async def test_vm_buttons_denied_for_specific_vm(
hass: HomeAssistant,
mock_proxmox_client: MagicMock,
mock_config_entry: MockConfigEntry,
entity_id: str,
translation_key: str,
) -> None:
"""Test that button only works on actual permissions."""
await setup_integration(hass, mock_config_entry)
mock_proxmox_client._node_mock.qemu(101)
with pytest.raises(ServiceValidationError):
with pytest.raises(ServiceValidationError) as exc_info:
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: "button.vm_db_start"},
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert exc_info.value.translation_key == translation_key