92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Dependency-free safety checks for the hass_mcp YAML overlay."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import pathlib
|
|
import sys
|
|
import types
|
|
|
|
|
|
def module(name: str, **attributes):
|
|
value = types.ModuleType(name)
|
|
for key, item in attributes.items():
|
|
setattr(value, key, item)
|
|
sys.modules[name] = value
|
|
return value
|
|
|
|
|
|
class ToolError(Exception):
|
|
pass
|
|
|
|
|
|
def decorator(**_kwargs):
|
|
return lambda function: function
|
|
|
|
|
|
module("homeassistant")
|
|
module("homeassistant.core", HomeAssistant=object)
|
|
module("homeassistant.util", slugify=lambda value: str(value).lower().replace(" ", "_"))
|
|
module("guarded")
|
|
module("guarded.tools")
|
|
module("guarded.identity", user_context=lambda: None)
|
|
module("guarded.protocol", ToolError=ToolError, internal_error=lambda message, error: RuntimeError(f"{message}: {error}"))
|
|
module(
|
|
"guarded.registry",
|
|
LIMIT_FIELD={"type": "integer"},
|
|
OFFSET_FIELD={"type": "integer"},
|
|
paginate=lambda items, limit, offset: {"items": items[offset : offset + limit]},
|
|
schema=lambda **kwargs: kwargs,
|
|
tool=decorator,
|
|
)
|
|
|
|
source = pathlib.Path(__file__).parents[1] / "platform/mcp/patches/hass_mcp/yaml_config.py"
|
|
spec = importlib.util.spec_from_file_location("guarded.tools.yaml_config", source)
|
|
assert spec and spec.loader
|
|
guard = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = guard
|
|
spec.loader.exec_module(guard)
|
|
|
|
assert set(guard._KINDS) == {"automation", "script", "scene", "configuration"}
|
|
assert all("secret" not in filename for filename, _, _ in guard._KINDS.values())
|
|
assert guard._redact_line("api_key: abc") == "api_key: <redacted>"
|
|
assert guard._redact_line("token: abc") == "token: <redacted>"
|
|
assert guard._redact_line("value: !secret private_name") == "value: !secret <redacted-reference>"
|
|
assert guard._redact_line("alias: Safe automation") == "alias: Safe automation"
|
|
|
|
for sensitive in ("token: abc", "password: abc", "value: !secret private_name"):
|
|
try:
|
|
guard._reject_sensitive_replacement(sensitive)
|
|
except ToolError:
|
|
pass
|
|
else:
|
|
raise AssertionError(f"sensitive replacement was accepted: {sensitive}")
|
|
|
|
change = guard._change_record("automation", "update", "before", {"id": "demo"})
|
|
preview = guard._preview(change)
|
|
assert preview["changed"] is False
|
|
assert preview["confirmation_required"] is True
|
|
guard._consume_ticket(change, preview["approval_ticket"])
|
|
try:
|
|
guard._consume_ticket(change, preview["approval_ticket"])
|
|
except ToolError:
|
|
pass
|
|
else:
|
|
raise AssertionError("one-time approval ticket was reusable")
|
|
|
|
other = guard._change_record("automation", "update", "different", {"id": "demo"})
|
|
ticket = guard._preview(change)["approval_ticket"]
|
|
try:
|
|
guard._consume_ticket(other, ticket)
|
|
except ToolError:
|
|
pass
|
|
else:
|
|
raise AssertionError("ticket accepted a different current file fingerprint")
|
|
|
|
diff = guard._source_diff("automations.yaml", "a\nb\n", "a\nc\n")
|
|
assert any("-b" in line for line in diff)
|
|
assert any("+c" in line for line in diff)
|
|
|
|
print("hass_mcp_yaml_guard_tests=ok")
|