fix: seed standalone MCP RBAC config #32
@@ -480,7 +480,11 @@ def _clamp_heartbeat(value: object) -> int:
|
||||
def load_config(config_path: Optional[str] = None) -> WorkspaceConfig:
|
||||
"""Load config from WORKSPACE_CONFIG_PATH or the given path."""
|
||||
if config_path is None:
|
||||
config_path = os.environ.get("WORKSPACE_CONFIG_PATH", "/configs")
|
||||
config_path = os.environ.get("WORKSPACE_CONFIG_PATH")
|
||||
if config_path is None:
|
||||
from molecule_runtime.configs_dir import resolve as resolve_configs_dir
|
||||
|
||||
config_path = str(resolve_configs_dir())
|
||||
|
||||
config_file = Path(config_path) / "config.yaml"
|
||||
if not config_file.exists():
|
||||
|
||||
@@ -49,6 +49,7 @@ from __future__ import annotations
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import molecule_runtime.configs_dir as configs_dir
|
||||
import molecule_runtime.mcp_heartbeat as mcp_heartbeat
|
||||
@@ -80,6 +81,24 @@ _read_token_file = mcp_workspace_resolver.read_token_file
|
||||
_start_inbox_pollers = mcp_inbox_pollers.start_inbox_pollers
|
||||
|
||||
|
||||
def _runtime_config_dir() -> Path:
|
||||
"""Return the config directory load_config() will use."""
|
||||
explicit = os.environ.get("WORKSPACE_CONFIG_PATH", "").strip()
|
||||
if explicit:
|
||||
path = Path(explicit)
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
return path
|
||||
return configs_dir.resolve()
|
||||
|
||||
|
||||
def _ensure_default_config() -> None:
|
||||
"""Seed standalone MCP's default RBAC config when none exists."""
|
||||
config_file = _runtime_config_dir() / "config.yaml"
|
||||
if config_file.exists():
|
||||
return
|
||||
config_file.write_text("rbac:\n roles:\n - operator\n", encoding="utf-8")
|
||||
|
||||
|
||||
def _effective_platform_url(default_platform_url: str, workspace_platform_url: str) -> str:
|
||||
"""Return the tenant URL a workspace should use for platform calls."""
|
||||
return (workspace_platform_url or default_platform_url).strip().rstrip("/")
|
||||
@@ -161,6 +180,8 @@ def main() -> None:
|
||||
)
|
||||
sys.exit(2)
|
||||
|
||||
_ensure_default_config()
|
||||
|
||||
platform_url = os.environ["PLATFORM_URL"].strip().rstrip("/")
|
||||
|
||||
# In multi-workspace mode the FIRST entry is treated as the
|
||||
|
||||
@@ -32,6 +32,8 @@ def _isolate_env(monkeypatch):
|
||||
"WORKSPACE_ID",
|
||||
"MOLECULE_WORKSPACE_TOKEN",
|
||||
"PLATFORM_URL",
|
||||
"WORKSPACE_CONFIG_PATH",
|
||||
"CONFIGS_DIR",
|
||||
):
|
||||
monkeypatch.delenv(var, raising=False)
|
||||
|
||||
@@ -155,6 +157,28 @@ class TestResolveWorkspaces:
|
||||
assert out == []
|
||||
assert any("MOLECULE_WORKSPACE_TOKEN" in e for e in errors)
|
||||
|
||||
def test_mcp_startup_creates_default_operator_rbac_config(self, monkeypatch, tmp_path):
|
||||
# External operators running molecule-mcp on a laptop normally do not
|
||||
# have /configs/config.yaml. Startup must seed a minimal config so the
|
||||
# MCP RBAC gate sees the documented operator default instead of the
|
||||
# fail-secure read-only fallback.
|
||||
monkeypatch.setenv("CONFIGS_DIR", str(tmp_path))
|
||||
mcp_cli = _import_mcp_cli()
|
||||
|
||||
cfg = tmp_path / "config.yaml"
|
||||
assert not cfg.exists()
|
||||
mcp_cli._ensure_default_config()
|
||||
|
||||
assert cfg.read_text(encoding="utf-8") == "rbac:\n roles:\n - operator\n"
|
||||
|
||||
import molecule_runtime.builtin_tools.audit as audit_mod
|
||||
|
||||
audit_mod._load_workspace_config.cache_clear()
|
||||
roles, custom = audit_mod.get_workspace_roles()
|
||||
assert roles == ["operator"]
|
||||
assert custom == {}
|
||||
audit_mod._load_workspace_config.cache_clear()
|
||||
|
||||
|
||||
class TestPlatformAuthRegistry:
|
||||
"""The token registry is what wires per-workspace heartbeats /
|
||||
|
||||
Reference in New Issue
Block a user