From 860ceca22d3d093fc1d0e955a7385f4f8b7da38d Mon Sep 17 00:00:00 2001 From: RuochenPan Date: Tue, 8 Sep 2026 08:54:14 +0800 Subject: [PATCH] fix: validate local permission support before saving config --- astrbot/dashboard/api/app.py | 5 +- astrbot/dashboard/services/config_service.py | 97 +++++++++++++++++++- tests/test_computer_config.py | 94 +++++++++++++++++++ tests/test_fastapi_v1_dashboard.py | 77 +++++++++++++++- tests/unit/test_config_profile_service.py | 15 ++- 5 files changed, 280 insertions(+), 8 deletions(-) diff --git a/astrbot/dashboard/api/app.py b/astrbot/dashboard/api/app.py index d4766ce4c..5eca6e2f8 100644 --- a/astrbot/dashboard/api/app.py +++ b/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), diff --git a/astrbot/dashboard/services/config_service.py b/astrbot/dashboard/services/config_service.py index 41efeb546..ebad00529 100644 --- a/astrbot/dashboard/services/config_service.py +++ b/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) diff --git a/tests/test_computer_config.py b/tests/test_computer_config.py index ea75e4b8e..387c2e8f2 100644 --- a/tests/test_computer_config.py +++ b/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 # ═══════════════════════════════════════════════════════════════ diff --git a/tests/test_fastapi_v1_dashboard.py b/tests/test_fastapi_v1_dashboard.py index 64a56cf4e..af76bdc82 100644 --- a/tests/test_fastapi_v1_dashboard.py +++ b/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, diff --git a/tests/unit/test_config_profile_service.py b/tests/unit/test_config_profile_service.py index 9d38beb72..7ac2d3e02 100644 --- a/tests/unit/test_config_profile_service.py +++ b/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")