Pyra/tests/security/test_plugin_isolation.py

"""Security tests: plugins cannot access the vault."""
import json
from pathlib import Path

import pytest

from pyra.security.boundaries import VaultAccessError
from pyra.plugins.loader import load_plugin_by_name


def _make_plugin(plugins_dir: Path, name: str, code: str) -> Path:
    d = plugins_dir / name
    d.mkdir(parents=True)
    (d / "manifest.json").write_text(json.dumps({"name": name, "version": "1.0.0"}))
    (d / "plugin.py").write_text(code)
    return d


# ── vault access via on_load ───────────────────────────────────────────────────

def test_plugin_cannot_receive_vault_path_via_vault_reader(tmp_pyra_home, tmp_path):
    """vault_reader returns None for any key not in the vault — plugins can't fish for paths."""
    plugins_dir = tmp_path / "plugins"
    plugins_dir.mkdir()
    code = """\
from pyra.plugins.base import BasePlugin

class _P(BasePlugin):
    name = "vault_fisher"
    description = "tries to get vault contents"
    version = "1.0.0"
    found = None

    def on_load(self, vault_reader):
        # Plugin can only call vault_reader with a string key, gets None back if key absent
        self.found = vault_reader("plugin:vault_fisher:secret")

def get_plugin():
    return _P()
"""
    _make_plugin(plugins_dir, "vault_fisher", code)
    plugin = load_plugin_by_name("vault_fisher", plugins_dir)
    assert plugin is not None
    # vault_reader returns None because the key doesn't exist — no vault data exposed
    assert plugin.found is None  # type: ignore[attr-defined]


def test_plugin_symlink_in_plugins_dir_is_blocked(tmp_pyra_home, tmp_path):
    """A plugin directory that is a symlink pointing inside the vault is blocked."""
    plugins_dir = tmp_path / "plugins"
    plugins_dir.mkdir()

    # Create a symlink from plugins/evil -> vault/
    evil_link = plugins_dir / "evil"
    evil_link.symlink_to(tmp_pyra_home / "vault")

    # Loading should fail because assert_safe_path blocks vault-pointing paths
    result = load_plugin_by_name("evil", plugins_dir)
    assert result is None


def test_plugin_on_load_receives_vault_reader_callable(tmp_pyra_home, tmp_path):
    """on_load receives vault_reader callable. Plugin can only access keys it knows the name of —
    the trust model relies on naming convention (plugin:name:key), not code-level sandboxing."""
    plugins_dir = tmp_path / "plugins"
    plugins_dir.mkdir()

    code = """\
from pyra.plugins.base import BasePlugin

class _P(BasePlugin):
    name = "vault_test"
    description = "tests vault_reader"
    version = "1.0.0"
    got_none = None

    def on_load(self, vault_reader):
        # Asking for a key that doesn't exist returns None
        self.got_none = vault_reader("plugin:vault_test:nonexistent_key")

def get_plugin():
    return _P()
"""
    _make_plugin(plugins_dir, "vault_test", code)
    plugin = load_plugin_by_name("vault_test", plugins_dir)
    assert plugin is not None

    # Call on_load manually (normally done by registry.load_all)
    from pyra.vault.reader import get_key
    plugin.on_load(get_key)

    # Non-existent key returns None — plugin gets no data
    assert plugin.got_none is None  # type: ignore[attr-defined]


def test_assert_safe_path_blocks_vault_directory(tmp_pyra_home):
    """Core invariant: assert_safe_path always blocks paths inside vault/."""
    from pyra.security.boundaries import assert_safe_path
    vault_path = tmp_pyra_home / "vault" / "secrets" / "api_keys.json"
    with pytest.raises(VaultAccessError):
        assert_safe_path(vault_path)


def test_plugin_load_does_not_grant_vault_path_access(tmp_pyra_home, tmp_path):
    """A plugin that calls open() on the vault path directly gets a file not found or
    permission error — but assert_safe_path isn't called inside plugin code by the core.
    This test verifies the loader path itself goes through assert_safe_path."""
    plugins_dir = tmp_path / "plugins"
    plugins_dir.mkdir()

    # Plugin dir is clean (not pointing at vault) — load should succeed
    code = """\
from pyra.plugins.base import BasePlugin

class _P(BasePlugin):
    name = "normal_plugin"
    description = "Normal plugin"
    version = "1.0.0"

def get_plugin():
    return _P()
"""
    _make_plugin(plugins_dir, "normal_plugin", code)
    plugin = load_plugin_by_name("normal_plugin", plugins_dir)
    assert plugin is not None
    assert plugin.name == "normal_plugin"