Require admin for blueprint ws commands (#161008)

homeassistant/components/blueprint/websocket_api.py

@@ -64,6 +64,7 @@ def _ws_with_blueprint_domain(
     return with_domain_blueprints
 
 
+@websocket_api.require_admin
 @websocket_api.websocket_command(
     {
         vol.Required("type"): "blueprint/list",
@@ -97,6 +98,7 @@ async def ws_list_blueprints(
     connection.send_result(msg["id"], results)
 
 
+@websocket_api.require_admin
 @websocket_api.websocket_command(
     {
         vol.Required("type"): "blueprint/import",
@@ -150,6 +152,7 @@ async def ws_import_blueprint(
     )
 
 
+@websocket_api.require_admin
 @websocket_api.websocket_command(
     {
         vol.Required("type"): "blueprint/save",
@@ -206,6 +209,7 @@ async def ws_save_blueprint(
     )
 
 
+@websocket_api.require_admin
 @websocket_api.websocket_command(
     {
         vol.Required("type"): "blueprint/delete",
@@ -233,6 +237,7 @@ async def ws_delete_blueprint(
     )
 
 
+@websocket_api.require_admin
 @websocket_api.websocket_command(
     {
         vol.Required("type"): "blueprint/substitute",

tests/components/blueprint/test_websocket_api.py

@@ -11,6 +11,7 @@ from homeassistant.core import HomeAssistant
 from homeassistant.setup import async_setup_component
 from homeassistant.util.yaml import UndefinedSubstitution, parse_yaml
 
+from tests.common import MockUser
 from tests.test_util.aiohttp import AiohttpClientMocker
 from tests.typing import WebSocketGenerator
 
@@ -103,6 +104,51 @@ async def test_list_blueprints_non_existing_domain(
     assert blueprints == {}
 
 
+@pytest.mark.parametrize(
+    "message",
+    [
+        {"type": "blueprint/list", "domain": "automation"},
+        {"type": "blueprint/import", "url": "https://example.com/blueprint.yaml"},
+        {
+            "type": "blueprint/save",
+            "path": "test_save",
+            "yaml": "raw_data",
+            "domain": "automation",
+        },
+        {
+            "type": "blueprint/delete",
+            "path": "test_delete",
+            "domain": "automation",
+        },
+        {
+            "type": "blueprint/substitute",
+            "domain": "automation",
+            "path": "test_event_service.yaml",
+            "input": {
+                "trigger_event": "test_event",
+                "service_to_call": "test.automation",
+                "a_number": 5,
+            },
+        },
+    ],
+)
+async def test_blueprint_ws_command_requires_admin(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    hass_admin_user: MockUser,
+    message: dict[str, Any],
+) -> None:
+    """Test that blueprint websocket commands require admin."""
+    hass_admin_user.groups = []  # Remove admin privileges
+    client = await hass_ws_client(hass)
+    await client.send_json_auto_id(message)
+
+    msg = await client.receive_json()
+
+    assert not msg["success"]
+    assert msg["error"]["code"] == "unauthorized"
+
+
 async def test_import_blueprint(
     hass: HomeAssistant,
     aioclient_mock: AiohttpClientMocker,