UmbraAudio/infra/remote_credential_broker/tests/test_broker.py

from __future__ import annotations

import json
import sys
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from remote_credential_broker.broker import BrokerState, RemoteCredentialBroker
from remote_credential_broker.config import BrokerConfig, RemoteProfile
from remote_credential_broker.executor import ExecutionResult, SecretRedactor
from remote_credential_broker.keychain import CredentialError, CredentialVault, validate_profile


class FakeStore:
    service = "test-service"

    def __init__(self) -> None:
        self.values: dict[str, str] = {}

    def set(self, profile: str, secret: str, ttl_seconds: int | None = None) -> None:
        self.values[profile] = secret

    def get(self, profile: str) -> str:
        return self.values[profile]

    def delete(self, profile: str) -> None:
        self.values.pop(profile, None)

    def exists(self, profile: str) -> bool:
        return profile in self.values


class FakeExecutor:
    def __init__(self) -> None:
        self.calls: list[tuple[str, tuple[str, ...], bool]] = []

    def run(
        self,
        profile: RemoteProfile,
        command: list[str] | tuple[str, ...],
        *,
        sudo: bool = False,
        timeout_seconds: int = 120,
    ) -> ExecutionResult:
        self.calls.append((profile.name, tuple(command), sudo))
        return ExecutionResult(0, "ok\n", "", 1)


class BrokerTests(unittest.TestCase):
    def setUp(self) -> None:
        self.store = FakeStore()
        self.state_path = Path("/tmp/remote-broker-test-state.json")
        self.state_path.unlink(missing_ok=True)
        self.state = BrokerState(self.state_path)
        self.executor = FakeExecutor()
        self.profile = RemoteProfile(
            name="oldbox-umbra",
            host="oldbox.bdmurphy.me",
            username="umbra",
            ssh_credential="oldbox-umbra-ssh",
            sudo_credential="oldbox-umbra-sudo",
            allowed_sudo_commands=(("/usr/local/sbin/umbracontrol-backup",),),
        )
        config = BrokerConfig(profiles={self.profile.name: self.profile})
        self.broker = RemoteCredentialBroker(
            self.store,
            config,
            self.state,
            lambda profile, purpose: "popup-secret",
            self.executor,
        )

    def tearDown(self) -> None:
        self.state_path.unlink(missing_ok=True)

    def test_credential_popup_is_not_returned(self) -> None:
        result = self.broker.call_tool(
            "credential_set",
            {"profile": "oldbox-umbra-ssh", "purpose": "SSH for oldbox"},
        )
        encoded = json.dumps(result)
        self.assertNotIn("popup-secret", encoded)
        payload = json.loads(result["content"][0]["text"])
        self.assertTrue(payload["stored"])
        self.assertEqual(self.store.get("oldbox-umbra-ssh"), "popup-secret")

    def test_status_never_returns_secret(self) -> None:
        self.broker.call_tool(
            "credential_set",
            {"profile": "oldbox-umbra-ssh", "purpose": "SSH for oldbox"},
        )
        result = self.broker.call_tool("credential_status", {"profile": "oldbox-umbra-ssh"})
        self.assertNotIn("popup-secret", json.dumps(result))
        payload = json.loads(result["content"][0]["text"])
        self.assertTrue(payload["configured"])

    def test_sudo_requires_exact_allowlist(self) -> None:
        result = self.broker.call_tool(
            "remote_sudo_exec",
            {"profile": "oldbox-umbra", "command": ["/bin/sh", "-c", "id"]},
        )
        self.assertTrue(result["isError"])
        self.assertIn("allowlist", result["content"][0]["text"])
        self.assertEqual(self.executor.calls, [])

    def test_allowed_sudo_is_forwarded(self) -> None:
        result = self.broker.call_tool(
            "remote_sudo_exec",
            {"profile": "oldbox-umbra", "command": ["/usr/local/sbin/umbracontrol-backup"]},
        )
        self.assertFalse(result["isError"])
        self.assertEqual(self.executor.calls[0][2], True)

    def test_json_rpc_tools_list_and_initialize(self) -> None:
        response = self.broker.handle_json_rpc({"jsonrpc": "2.0", "id": 1, "method": "initialize"})
        assert response is not None
        self.assertEqual(response["result"]["serverInfo"]["name"], "umbratrans-remote-broker")
        response = self.broker.handle_json_rpc({"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
        assert response is not None
        names = {tool["name"] for tool in response["result"]["tools"]}
        self.assertIn("credential_set", names)
        self.assertIn("remote_sudo_exec", names)

    def test_redactor(self) -> None:
        redactor = SecretRedactor()
        redactor.add("secret-value")
        self.assertEqual(redactor.redact("x secret-value y"), "x <redacted> y")
    def test_temporary_profile_is_in_memory_and_readable(self) -> None:
        persistent = FakeStore()
        vault = CredentialVault(persistent)
        vault.set("temporary-profile", "temporary-value", ttl_seconds=60)
        self.assertEqual(vault.get("temporary-profile"), "temporary-value")
        self.assertTrue(vault.exists("temporary-profile"))
        self.assertNotIn("temporary-profile", persistent.values)
        vault.delete("temporary-profile")
        self.assertFalse(vault.exists("temporary-profile"))


class ValidationTests(unittest.TestCase):
    def test_profile_rejects_shell_metacharacters(self) -> None:
        with self.assertRaises(CredentialError):
            validate_profile("bad profile;rm")

    def test_profile_accepts_normal_name(self) -> None:
        self.assertEqual(validate_profile("oldbox-umbra_ssh.1"), "oldbox-umbra_ssh.1")


if __name__ == "__main__":
    unittest.main()