fix: validate local permission support before saving config

astrbot/dashboard/api/app.py

@@ -100,8 +100,9 @@ def create_dashboard_asgi_app(
     app.state.jwt_secret = jwt_secret
     app.state.dashboard_static_folder = static_folder
     log_broker = getattr(core_lifecycle, "log_broker", None) or LogBroker()
+    stats = StatService(db, core_lifecycle, core_lifecycle.astrbot_config)
     app.state.services = SimpleNamespace(
-        config_profiles=ConfigProfileService(core_lifecycle, db),
+        config_profiles=ConfigProfileService(core_lifecycle, db, runtime=stats.runtime),
         config_display=ConfigDisplayService(core_lifecycle),
         config_files=ConfigFileService(core_lifecycle),
         config_routes=ConfigRoutingService(core_lifecycle),
@@ -129,7 +130,7 @@ def create_dashboard_asgi_app(
         open_api=OpenApiService(db, core_lifecycle),
         sessions=SessionManagementService(core_lifecycle, db),
         skills=SkillsService(core_lifecycle),
-        stats=StatService(db, core_lifecycle, core_lifecycle.astrbot_config),
+        stats=stats,
         subagents=SubAgentService(core_lifecycle),
         t2i=T2iService(core_lifecycle),
         tools=ToolsService(core_lifecycle),

astrbot/dashboard/services/config_service.py

@@ -203,7 +203,27 @@ def sanitize_filename(name: str) -> str:
     return _sanitize_filename(name)
 
 
-def validate_config(data, schema: dict, is_core: bool) -> tuple[list[str], dict]:
+def validate_config(
+    data,
+    schema: dict,
+    is_core: bool,
+    *,
+    runtime: dict | None = None,
+    current_config: dict | None = None,
+) -> tuple[list[str], dict]:
+    """Validate configuration values and normalize linked Local permissions.
+
+    Args:
+        data: Submitted configuration, normalized in place.
+        schema: Configuration metadata used for validation.
+        is_core: Whether this is a core configuration rather than a plugin.
+        runtime: Startup runtime snapshot for platform-specific validation.
+        current_config: Existing configuration whose unchanged Local policies
+            may be retained when saving unrelated settings.
+
+    Returns:
+        Validation errors and the normalized configuration.
+    """
     errors = []
 
     def validate(data: dict, metadata: dict = schema, path="") -> None:
@@ -308,6 +328,7 @@ def validate_config(data, schema: dict, is_core: bool) -> tuple[list[str], dict]
             if isinstance(provider_settings, dict)
             else {}
         )
+        submitted_permissions = copy.deepcopy(permissions)
         if not isinstance(permissions, dict):
             errors.append("Local computer permissions must be an object.")
         else:
@@ -335,6 +356,50 @@ def validate_config(data, schema: dict, is_core: bool) -> tuple[list[str], dict]
                     policy["allow_network"] = False
                 elif policy.get("allow_execution", role == "admin") is False:
                     policy["allow_network"] = False
+
+        if (
+            not errors
+            and runtime is not None
+            and isinstance(provider_settings, dict)
+            and provider_settings.get("computer_use_runtime") == "local"
+            and runtime["sandbox"]["status"] != "detected"
+        ):
+            old_settings = (current_config or {}).get("provider_settings", {})
+            old_permissions = old_settings.get("computer_use_local_permissions", {})
+            was_local = old_settings.get("computer_use_runtime") == "local"
+            defaults = DEFAULT_CONFIG["provider_settings"][
+                "computer_use_local_permissions"
+            ]
+            for role in ("member", "admin"):
+                # Keep unchanged legacy policies, but check both roles when
+                # activating Local access or creating a profile.
+                if was_local and submitted_permissions.get(
+                    role, {}
+                ) == old_permissions.get(role, {}):
+                    continue
+                policy = {**defaults[role], **permissions.get(role, {})}
+                scope = policy["filesystem_scope"]
+                if scope == "none":
+                    continue
+                unsupported = runtime["sandbox"]["status"] == "unsupported"
+                if not (
+                    (unsupported and scope == "workspace")
+                    or (
+                        policy["allow_execution"]
+                        and (scope == "workspace" or not policy["allow_network"])
+                    )
+                ):
+                    continue
+                if unsupported:
+                    reason = f"Local isolation is not supported on {runtime['os']}."
+                else:
+                    dependency = (
+                        "Seatbelt (/usr/bin/sandbox-exec)"
+                        if runtime["sandbox"]["backend"] == "seatbelt"
+                        else "bubblewrap (bwrap)"
+                    )
+                    reason = f"Missing {dependency}; restricted Local execution is unavailable."
+                errors.append(f"Local permission {role}: {reason}")
     else:
         validate(data, schema)
 
@@ -475,7 +540,20 @@ def save_config(
     post_config: dict,
     config: AstrBotConfig,
     is_core: bool = False,
+    *,
+    runtime: dict | None = None,
 ) -> None:
+    """Validate and persist a dashboard configuration update.
+
+    Args:
+        post_config: Submitted configuration to validate and save.
+        config: Existing configuration and persistence target.
+        is_core: Whether this is a core configuration rather than a plugin.
+        runtime: Startup runtime snapshot supplied by the profile service.
+
+    Raises:
+        ValueError: If configuration validation fails.
+    """
     if is_core:
         post_config["agent_runner"] = normalize_agent_runner(
             post_config.get("agent_runner")
@@ -487,6 +565,8 @@ def save_config(
                 post_config,
                 CONFIG_METADATA_2,
                 is_core,
+                runtime=runtime,
+                current_config=dict(config),
             )
         else:
             errors, post_config = validate_config(
@@ -511,10 +591,13 @@ class ConfigProfileService:
         self,
         core_lifecycle: AstrBotCoreLifecycle,
         db: BaseDatabase | None = None,
+        *,
+        runtime: dict,
     ) -> None:
         self.core_lifecycle = core_lifecycle
         self.acm = core_lifecycle.astrbot_config_mgr
         self.db = db
+        self.runtime = runtime
 
     def get_profile_schema(self) -> dict:
         return {
@@ -576,6 +659,7 @@ class ConfigProfileService:
 
         Raises:
             ApiError: If caller attempts to define administrator IDs without scope.
+            ValueError: If configuration validation fails.
         """
         if (
             not allow_admin_id_change
@@ -592,6 +676,11 @@ class ConfigProfileService:
             profile_config["agent_runner"] = normalize_agent_runner(
                 profile_config["agent_runner"]
             )
+        errors, profile_config = validate_config(
+            profile_config, CONFIG_METADATA_2, is_core=True, runtime=self.runtime
+        )
+        if errors:
+            raise ValueError(f"Configuration validation failed: {errors}")
         conf_id = await self.acm.create_conf(
             name=name,
             config=profile_config,
@@ -658,7 +747,7 @@ class ConfigProfileService:
 
         Raises:
             ApiError: If admin IDs change without permission or TOTP is invalid.
-            ValueError: If the requested config profile does not exist.
+            ValueError: If the profile does not exist or validation fails.
         """
         if config_id not in self.acm.confs:
             raise ValueError(f"Config file {config_id} does not exist")
@@ -695,7 +784,9 @@ class ConfigProfileService:
             _set_nested_value(config, ("dashboard", "totp", "recovery_code_hash"), "")
 
         set_pending_totp_secret(None)
-        save_config(config, self.acm.confs[config_id], is_core=True)
+        save_config(
+            config, self.acm.confs[config_id], is_core=True, runtime=self.runtime
+        )
         if protected_2fa_changed and self.db is not None:
             await revoke_user_trusted_devices(self.db)
         await self.core_lifecycle.reload_pipeline_scheduler(config_id)

tests/test_computer_config.py

@@ -90,6 +90,100 @@ def test_local_permission_validation_does_not_inject_missing_policies():
     assert saved == original
 
 
+@pytest.mark.parametrize("role", ["member", "admin"])
+@pytest.mark.parametrize("status", ["detected", "missing", "unsupported"])
+@pytest.mark.parametrize(
+    ("scope", "execution", "network", "denied_on"),
+    [
+        ("none", False, False, ()),
+        ("workspace", False, False, ("unsupported",)),
+        ("host", False, False, ()),
+        ("workspace", True, False, ("missing", "unsupported")),
+        ("workspace", True, True, ("missing", "unsupported")),
+        ("host", True, False, ("missing", "unsupported")),
+        ("host", True, True, ()),
+    ],
+)
+def test_local_permission_platform_matrix(
+    role, status, scope, execution, network, denied_on
+):
+    permissions = {r: {"filesystem_scope": "none"} for r in ("member", "admin")}
+    permissions[role] = {
+        "filesystem_scope": scope,
+        "allow_execution": execution,
+        "allow_network": network,
+    }
+    errors, _ = validate_config(
+        {
+            "provider_settings": {
+                "computer_use_runtime": "local",
+                "computer_use_local_permissions": permissions,
+            }
+        },
+        CONFIG_METADATA_2,
+        is_core=True,
+        runtime={
+            "os": "windows" if status == "unsupported" else "linux",
+            "sandbox": {"status": status, "backend": "bubblewrap"},
+        },
+    )
+
+    assert bool(errors) == (status in denied_on)
+    if errors:
+        assert len(errors) == 1
+        assert errors[0].startswith(f"Local permission {role}:")
+
+
+@pytest.mark.parametrize(
+    ("old_mode", "new_mode", "role", "change", "rejected"),
+    [
+        ("local", "local", "member", {}, False),
+        ("local", "local", "admin", {"filesystem_scope": "host"}, False),
+        ("local", "local", "member", {"allow_execution": True}, True),
+        ("local", "local", "member", None, True),
+        ("none", "local", "member", {}, True),
+        ("local", "none", "member", {}, False),
+        ("local", "sandbox", "member", {}, False),
+        ("local", "local", "member", {"filesystem_scope": "none"}, False),
+    ],
+)
+def test_windows_legacy_permissions(old_mode, new_mode, role, change, rejected):
+    old = {
+        "provider_settings": {
+            "computer_use_runtime": old_mode,
+            "computer_use_local_permissions": {
+                r: {
+                    "filesystem_scope": "workspace",
+                    "allow_execution": False,
+                    "allow_network": True,
+                }
+                for r in ("member", "admin")
+            },
+        }
+    }
+    payload = deepcopy(old)
+    settings = payload["provider_settings"]
+    settings.update(computer_use_runtime=new_mode, default_provider_id="new-provider")
+    permissions = settings["computer_use_local_permissions"]
+    if change is None:
+        del permissions[role]
+    else:
+        permissions[role].update(change)
+
+    errors, normalized = validate_config(
+        payload,
+        CONFIG_METADATA_2,
+        is_core=True,
+        current_config=old,
+        runtime={"os": "windows", "sandbox": {"status": "unsupported"}},
+    )
+
+    assert bool(errors) == rejected
+    if not rejected:
+        saved = normalized["provider_settings"]["computer_use_local_permissions"]
+        assert all(not policy["allow_network"] for policy in saved.values())
+
+
 # ═══════════════════════════════════════════════════════════════
 # _discover_bay_credentials
 # ═══════════════════════════════════════════════════════════════

tests/test_fastapi_v1_dashboard.py

@@ -1668,7 +1668,9 @@ async def test_v1_system_config_update_preserves_independent_bot_provider_sectio
     fake_core_lifecycle,
     monkeypatch: pytest.MonkeyPatch,
 ):
-    def fake_save_config(post_config: dict, config: FakeAstrBotConfig, is_core=False):
+    def fake_save_config(
+        post_config: dict, config: FakeAstrBotConfig, is_core=False, *, runtime=None
+    ):
         config.save_config(post_config)
 
     monkeypatch.setattr(config_service, "save_config", fake_save_config)
@@ -1704,6 +1706,79 @@ async def test_v1_system_config_update_preserves_independent_bot_provider_sectio
     assert fake_core_lifecycle.reloaded_config_ids == ["default"]
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("scope", ["workspace", "host"])
+@pytest.mark.parametrize(
+    ("system", "backend", "status", "reason"),
+    [
+        ("windows", None, "unsupported", "windows"),
+        ("linux", "bubblewrap", "missing", "bwrap"),
+        ("darwin", "seatbelt", "missing", "sandbox-exec"),
+    ],
+)
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/api/v1/system-config",
+        "/api/v1/config-profiles/default",
+        "/api/config/astrbot/update",
+    ],
+)
+async def test_config_api_validates_local_permissions(
+    asgi_app,
+    asgi_client,
+    fake_core_lifecycle,
+    path,
+    scope,
+    system,
+    backend,
+    status,
+    reason,
+):
+    runtime = asgi_app.state.services.stats.runtime
+    assert asgi_app.state.services.config_profiles.runtime is runtime
+    runtime.update({"os": system, "sandbox": {"backend": backend, "status": status}})
+    original = copy.deepcopy(fake_core_lifecycle.astrbot_config)
+    payload = copy.deepcopy(original)
+    payload["agent_runner"] = {"runner_type": "local"}
+    payload["provider_settings"] = {
+        "computer_use_runtime": "local",
+        "computer_use_local_permissions": {
+            "member": {
+                "filesystem_scope": scope,
+                "allow_execution": True,
+                "allow_network": True,
+            },
+            "admin": {"filesystem_scope": "none"},
+        },
+    }
+    legacy = path.startswith("/api/config/")
+    response = await asgi_client.request(
+        "POST" if legacy else "PUT",
+        path,
+        headers=_jwt_headers(),
+        json={"conf_id": "default", "config": payload} if legacy else payload,
+    )
+
+    assert response.status_code == (400 if scope == "workspace" and not legacy else 200)
+    if scope == "workspace":
+        assert response.json()["status"] == "error"
+        assert "Local permission member:" in response.json()["message"]
+        assert reason in response.json()["message"]
+        assert fake_core_lifecycle.astrbot_config == original
+        assert fake_core_lifecycle.reloaded_config_ids == []
+    else:
+        assert response.json()["status"] == "ok"
+        payload["provider_settings"]["computer_use_local_permissions"]["admin"].update(
+            allow_execution=False, allow_network=False
+        )
+        assert (
+            fake_core_lifecycle.astrbot_config["provider_settings"]
+            == payload["provider_settings"]
+        )
+        assert fake_core_lifecycle.reloaded_config_ids == ["default"]
+
+
 @pytest.mark.asyncio
 async def test_v1_system_config_returns_system_metadata(
     asgi_client: httpx.AsyncClient,

tests/unit/test_config_profile_service.py

@@ -15,7 +15,8 @@ def test_get_system_config_includes_effective_server_time() -> None:
             astrbot_config_mgr=SimpleNamespace(
                 confs={"default": {"timezone": "Asia/Shanghai"}}
             )
-        )
+        ),
+        runtime={"os": "windows", "sandbox": {"status": "unsupported"}},
     )
 
     with patch("astrbot.dashboard.services.config_service.datetime") as mock_datetime:
@@ -40,7 +41,17 @@ async def test_profile_mutations_await_config_manager() -> None:
         reload_pipeline_scheduler=AsyncMock(),
         pipeline_scheduler_mapping={"profile-id": object()},
     )
-    service = ConfigProfileService(lifecycle)
+    service = ConfigProfileService(
+        lifecycle, runtime={"os": "windows", "sandbox": {"status": "unsupported"}}
+    )
+
+    with pytest.raises(ValueError, match="Local permission member:") as exc:
+        await service.create_profile(
+            "Invalid", {"provider_settings": {"computer_use_runtime": "local"}}
+        )
+    assert "Local permission admin:" in str(exc.value)
+    config_manager.create_conf.assert_not_awaited()
+    lifecycle.reload_pipeline_scheduler.assert_not_awaited()
 
     result = await service.create_profile("Profile", {"timezone": "UTC"})
     await service.rename_profile("profile-id", "Renamed")