"""Tests for ``muse auth key list`` / ``muse auth key delete`` (musehub#221 follow-up).

These commands expose the pre-existing hub key-management endpoints
(``GET``/``DELETE /api/auth/keys/{handle}[/{key_id}]``) as a proper CLI
resource, following the same ``<noun> <verb>`` convention as
``muse hub label``/``muse hub webhook`` -- and give a durable way to clean
up leftover pre-hub-scoping keys after ``muse migrate hub-scoping``, since
the migration's own best-effort deregistration can fail (e.g. the old
key's key_id lookup gap fixed alongside this) and identity.toml no longer
has the old fingerprint on hand once migration has completed.
"""

from __future__ import annotations

import json
import pathlib

import pytest

from tests.cli_test_helper import CliRunner
from muse.core import keypair as kp_module
from muse.core import identity as id_module
from muse.core.paths import muse_dir

runner = CliRunner()

_HUB = "https://localhost:1337"
_MNEMONIC = (
    "abandon abandon abandon abandon abandon abandon abandon abandon "
    "abandon abandon abandon about"
)


@pytest.fixture()
def isolated(monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path) -> pathlib.Path:
    fake_home = tmp_path / "home"
    fake_home.mkdir(parents=True, exist_ok=True)
    fake_muse = muse_dir(fake_home)
    fake_muse.mkdir(parents=True, exist_ok=True)

    monkeypatch.setattr(pathlib.Path, "home", staticmethod(lambda: fake_home))
    monkeypatch.setattr(kp_module, "_KEYS_DIR", fake_muse / "keys")
    monkeypatch.setattr(id_module, "_IDENTITY_DIR", fake_muse)
    monkeypatch.setattr(id_module, "_IDENTITY_FILE", fake_muse / "identity.toml")
    monkeypatch.setattr("muse.cli.commands.auth._stderr_isatty", lambda: False)

    _kc: dict[str, str] = {}
    monkeypatch.setattr("muse.core.keychain.is_available", lambda: True)
    monkeypatch.setattr("muse.core.keychain.load", lambda: _kc.get("mnemonic"))
    monkeypatch.setattr("muse.core.keychain.store", lambda m: _kc.__setitem__("mnemonic", m))
    monkeypatch.setattr("muse.core.keychain.delete", lambda: _kc.pop("mnemonic", None))

    import muse.cli.commands.auth as _auth_mod
    _challenge = {"challenge_token": "deadbeef" * 8, "is_new_key": True, "algorithm": "ed25519"}
    _verify = {"handle": "gabriel", "identity_id": "id-123", "is_new_identity": False, "auth_method": "ed25519"}
    monkeypatch.setattr(
        _auth_mod, "_json_post_raw",
        lambda base, path, payload, extra_headers=None: _challenge if "challenge" in path else _verify,
    )
    monkeypatch.setattr(_auth_mod, "_hub_delete", lambda url, auth_header, ssl_ctx=None: None)
    return fake_home


@pytest.fixture()
def fixed_mnemonic(monkeypatch: pytest.MonkeyPatch) -> str:
    from muse.core import bip39 as bip39_mod
    monkeypatch.setattr(bip39_mod, "generate_mnemonic", lambda **kw: _MNEMONIC)
    return _MNEMONIC


def _keygen_and_register() -> None:
    result = runner.invoke(None, ["auth", "keygen", "--hub", _HUB, "--json"])
    assert result.exit_code == 0, f"keygen failed: {result.output}"
    result = runner.invoke(None, ["auth", "register", "--hub", _HUB, "--handle", "gabriel", "--json"])
    assert result.exit_code == 0, f"register failed: {result.output}"


def _current_fingerprint() -> str:
    from muse.core.identity import load_identity
    entry = load_identity(_HUB)
    assert entry is not None
    return str(entry["fingerprint"])


class TestKeyList:
    def test_list_returns_keys_from_hub(
        self, isolated: pathlib.Path, fixed_mnemonic: str, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        _keygen_and_register()
        current_fp = _current_fingerprint()

        seen = {}

        def _fake_get(url: str, auth_header: str, ssl_ctx=None) -> dict:
            seen["url"] = url
            seen["auth_header"] = auth_header
            return {"keys": [
                {"key_id": "sha256:" + "1" * 64, "algorithm": "ed25519", "fingerprint": current_fp,
                 "label": "", "created_at": "2026-01-01T00:00:00Z", "last_used_at": None},
            ]}

        monkeypatch.setattr("muse.cli.commands.auth._hub_get", _fake_get)

        result = runner.invoke(None, ["auth", "key", "list", "--hub", _HUB, "--json"])
        assert result.exit_code == 0, result.output
        data = json.loads(result.output)
        assert len(data["keys"]) == 1
        assert data["keys"][0]["fingerprint"] == current_fp
        assert "gabriel" in seen["url"]
        assert seen["auth_header"].startswith("MSign ")

    def test_list_no_hub_errors(self, isolated: pathlib.Path, fixed_mnemonic: str) -> None:
        result = runner.invoke(None, ["auth", "key", "list", "--json"])
        assert result.exit_code != 0

    def test_list_no_identity_errors(self, isolated: pathlib.Path, fixed_mnemonic: str) -> None:
        result = runner.invoke(None, ["auth", "key", "list", "--hub", _HUB, "--json"])
        assert result.exit_code != 0


class TestKeyDelete:
    def test_delete_looks_up_key_id_by_fingerprint_and_revokes(
        self, isolated: pathlib.Path, fixed_mnemonic: str, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        _keygen_and_register()
        current_fp = _current_fingerprint()
        old_fp = "sha256:" + "a" * 64  # a different, "old" key -- not the active one
        real_key_id = "sha256:" + "7" * 64

        list_seen = {}
        delete_seen = {}

        def _fake_get(url: str, auth_header: str, ssl_ctx=None) -> dict:
            list_seen["url"] = url
            list_seen["auth_header"] = auth_header
            return {"keys": [
                {"key_id": real_key_id, "fingerprint": old_fp},
                {"key_id": "sha256:" + "2" * 64, "fingerprint": current_fp},
            ]}

        def _fake_delete(url: str, auth_header: str, ssl_ctx=None) -> None:
            delete_seen["url"] = url
            delete_seen["auth_header"] = auth_header

        monkeypatch.setattr("muse.cli.commands.auth._hub_get", _fake_get)
        monkeypatch.setattr("muse.cli.commands.auth._hub_delete", _fake_delete)

        result = runner.invoke(
            None, ["auth", "key", "delete", "--hub", _HUB, "--fingerprint", old_fp, "--json"]
        )
        assert result.exit_code == 0, result.output
        data = json.loads(result.output)
        assert data["revoked"] is True
        assert data["key_id"] == real_key_id

        import urllib.parse
        assert urllib.parse.quote(real_key_id) in delete_seen["url"]
        assert "gabriel" in delete_seen["url"]
        assert delete_seen["auth_header"].startswith("MSign ")

    def test_delete_refuses_current_active_key(
        self, isolated: pathlib.Path, fixed_mnemonic: str, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """Deleting the fingerprint that's currently active in identity.toml would
        strand the account with no working key from this client -- refuse outright,
        never even contact the hub."""
        _keygen_and_register()
        current_fp = _current_fingerprint()

        get_called = []
        monkeypatch.setattr(
            "muse.cli.commands.auth._hub_get",
            lambda *a, **kw: get_called.append(1) or {"keys": []},
        )

        result = runner.invoke(
            None, ["auth", "key", "delete", "--hub", _HUB, "--fingerprint", current_fp, "--json"]
        )
        assert result.exit_code != 0
        assert get_called == []

    def test_delete_fingerprint_not_found_on_hub_errors(
        self, isolated: pathlib.Path, fixed_mnemonic: str, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        _keygen_and_register()
        current_fp = _current_fingerprint()
        delete_called = []

        monkeypatch.setattr(
            "muse.cli.commands.auth._hub_get",
            lambda *a, **kw: {"keys": [{"key_id": "sha256:" + "3" * 64, "fingerprint": current_fp}]},
        )
        monkeypatch.setattr(
            "muse.cli.commands.auth._hub_delete",
            lambda *a, **kw: delete_called.append(1),
        )

        result = runner.invoke(
            None, ["auth", "key", "delete", "--hub", _HUB, "--fingerprint", "sha256:" + "f" * 64, "--json"]
        )
        assert result.exit_code != 0
        assert delete_called == []

    def test_delete_no_identity_errors(self, isolated: pathlib.Path, fixed_mnemonic: str) -> None:
        result = runner.invoke(
            None, ["auth", "key", "delete", "--hub", _HUB, "--fingerprint", "sha256:" + "a" * 64, "--json"]
        )
        assert result.exit_code != 0


# ---------------------------------------------------------------------------
# Security
# ---------------------------------------------------------------------------


class TestKeySecurity:
    """Mirrors test_auth_rotate.py's TestRotateSecurity for the key resource:
    no secret leakage, no terminal-injection via hub-controlled strings, and
    a bounded response size against a malicious/misbehaving hub."""

    def test_mnemonic_not_in_list_json_output(
        self, isolated: pathlib.Path, fixed_mnemonic: str, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        _keygen_and_register()
        monkeypatch.setattr("muse.cli.commands.auth._hub_get", lambda *a, **kw: {"keys": []})
        result = runner.invoke(None, ["auth", "key", "list", "--hub", _HUB, "--json"])
        assert result.exit_code == 0, result.output
        assert _MNEMONIC not in (result.output or "")

    def test_mnemonic_not_in_delete_json_output(
        self, isolated: pathlib.Path, fixed_mnemonic: str, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        _keygen_and_register()
        old_fp = "sha256:" + "a" * 64
        monkeypatch.setattr(
            "muse.cli.commands.auth._hub_get",
            lambda *a, **kw: {"keys": [{"key_id": "sha256:" + "7" * 64, "fingerprint": old_fp}]},
        )
        monkeypatch.setattr("muse.cli.commands.auth._hub_delete", lambda *a, **kw: None)
        result = runner.invoke(
            None, ["auth", "key", "delete", "--hub", _HUB, "--fingerprint", old_fp, "--json"]
        )
        assert result.exit_code == 0, result.output
        assert _MNEMONIC not in (result.output or "")

    def test_no_pem_written_during_list_or_delete(
        self, isolated: pathlib.Path, fixed_mnemonic: str, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """Keys are derived from the mnemonic at sign time -- neither command
        should ever write PEM material to disk."""
        _keygen_and_register()
        old_fp = "sha256:" + "a" * 64
        monkeypatch.setattr(
            "muse.cli.commands.auth._hub_get",
            lambda *a, **kw: {"keys": [{"key_id": "sha256:" + "7" * 64, "fingerprint": old_fp}]},
        )
        monkeypatch.setattr("muse.cli.commands.auth._hub_delete", lambda *a, **kw: None)

        runner.invoke(None, ["auth", "key", "list", "--hub", _HUB, "--json"])
        runner.invoke(None, ["auth", "key", "delete", "--hub", _HUB, "--fingerprint", old_fp, "--json"])

        keys_dir = muse_dir(isolated) / "keys"
        pem_files = list(keys_dir.glob("**/*.pem")) if keys_dir.exists() else []
        assert pem_files == [], f"PEM files must not be written by key list/delete: {pem_files}"

    def test_malicious_hub_response_size_is_bounded(
        self, isolated: pathlib.Path, fixed_mnemonic: str, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """A hub (compromised, buggy, or MITM'd past TLS) that returns an
        oversized response body must not be processed -- _hub_get enforces
        the same 1 MiB bound as the rest of the HTTP layer (_json_post_raw)."""
        _keygen_and_register()

        import urllib.error

        def _oversized_get(url, auth_header, ssl_ctx=None):
            # Exercise the real _hub_get body-size check via urlopen, not a
            # hand-rolled bypass -- monkeypatch urlopen's read() to return an
            # oversized body and confirm _hub_get itself raises.
            import io
            import muse.cli.commands.auth as _auth_mod

            class _FakeResp:
                def read(self, n):
                    return b"x" * (_auth_mod._MAX_RESPONSE_BYTES + 1)
                def __enter__(self):
                    return self
                def __exit__(self, *a):
                    return False

            monkeypatch.setattr(
                "muse.cli.commands.auth.urllib.request.urlopen",
                lambda req, timeout=10, context=None: _FakeResp(),
            )
            from muse.cli.commands.auth import _hub_get as _real_hub_get
            return _real_hub_get(url, auth_header, ssl_ctx)

        monkeypatch.setattr("muse.cli.commands.auth._hub_get", _oversized_get)

        result = runner.invoke(None, ["auth", "key", "list", "--hub", _HUB, "--json"])
        assert result.exit_code != 0, "oversized hub response must not be processed as valid JSON"

    def test_ansi_escape_in_hub_label_is_stripped_from_terminal_output(
        self, isolated: pathlib.Path, fixed_mnemonic: str, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """A hub-controlled field (label) containing an ANSI/OSC escape sequence
        must never reach the terminal unsanitized in non-JSON output -- a
        compromised or malicious hub could otherwise inject terminal escapes
        into the operator's shell via a key label."""
        _keygen_and_register()
        current_fp = _current_fingerprint()
        malicious_label = "\x1b]0;pwned\x07evil-label"

        monkeypatch.setattr(
            "muse.cli.commands.auth._hub_get",
            lambda *a, **kw: {"keys": [
                {"key_id": "sha256:" + "9" * 64, "fingerprint": current_fp, "label": malicious_label},
            ]},
        )

        result = runner.invoke(None, ["auth", "key", "list", "--hub", _HUB])
        assert result.exit_code == 0, result.output
        assert "\x1b" not in (result.output or "")
        assert "\x07" not in (result.output or "")


# ---------------------------------------------------------------------------
# Performance
# ---------------------------------------------------------------------------


class TestKeyPerformance:
    """Mirrors test_auth_rotate.py's TestRotatePerformance: derivation +
    request-building overhead is negligible with the hub stubbed."""

    def test_list_completes_under_500ms(
        self, isolated: pathlib.Path, fixed_mnemonic: str, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        import time

        _keygen_and_register()
        monkeypatch.setattr("muse.cli.commands.auth._hub_get", lambda *a, **kw: {"keys": []})

        start = time.perf_counter()
        result = runner.invoke(None, ["auth", "key", "list", "--hub", _HUB, "--json"])
        elapsed_ms = (time.perf_counter() - start) * 1000

        assert result.exit_code == 0, result.output
        assert elapsed_ms < 500, f"key list took {elapsed_ms:.1f} ms — expected under 500 ms."

    def test_delete_completes_under_500ms(
        self, isolated: pathlib.Path, fixed_mnemonic: str, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        import time

        _keygen_and_register()
        old_fp = "sha256:" + "a" * 64
        monkeypatch.setattr(
            "muse.cli.commands.auth._hub_get",
            lambda *a, **kw: {"keys": [{"key_id": "sha256:" + "7" * 64, "fingerprint": old_fp}]},
        )
        monkeypatch.setattr("muse.cli.commands.auth._hub_delete", lambda *a, **kw: None)

        start = time.perf_counter()
        result = runner.invoke(
            None, ["auth", "key", "delete", "--hub", _HUB, "--fingerprint", old_fp, "--json"]
        )
        elapsed_ms = (time.perf_counter() - start) * 1000

        assert result.exit_code == 0, result.output
        assert elapsed_ms < 500, f"key delete took {elapsed_ms:.1f} ms — expected under 500 ms."


# ---------------------------------------------------------------------------
# Stress
# ---------------------------------------------------------------------------


class TestKeyStress:
    """Mirrors test_auth_rotate.py's TestRotateStress: repeated calls remain
    correct -- no state leakage or drift across repeated list/delete cycles."""

    def test_repeated_list_calls_are_stable(
        self, isolated: pathlib.Path, fixed_mnemonic: str, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        _keygen_and_register()
        fp = _current_fingerprint()
        monkeypatch.setattr(
            "muse.cli.commands.auth._hub_get",
            lambda *a, **kw: {"keys": [{"key_id": "sha256:" + "1" * 64, "fingerprint": fp}]},
        )

        results = []
        for _ in range(20):
            r = runner.invoke(None, ["auth", "key", "list", "--hub", _HUB, "--json"])
            assert r.exit_code == 0, r.output
            results.append(json.loads(r.output))

        # Compare only "keys" -- the envelope's timestamp/duration_ms legitimately
        # differ per call.
        assert all(r["keys"] == results[0]["keys"] for r in results), (
            "repeated list calls must return the same key set"
        )

    def test_ten_sequential_deletes_of_distinct_fingerprints(
        self, isolated: pathlib.Path, fixed_mnemonic: str, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """10 distinct orphaned keys, deleted one at a time, each correctly
        matched to its own key_id -- guards against any accidental caching or
        off-by-one in the fingerprint-to-key_id lookup across repeated calls."""
        _keygen_and_register()
        current_fp = _current_fingerprint()

        orphans = [
            {"key_id": f"sha256:{i:064d}", "fingerprint": f"sha256:{'a' * 63}{i}"}
            for i in range(10)
        ]
        all_keys = orphans + [{"key_id": "sha256:" + "9" * 64, "fingerprint": current_fp}]

        monkeypatch.setattr("muse.cli.commands.auth._hub_get", lambda *a, **kw: {"keys": all_keys})
        deleted_key_ids = []
        monkeypatch.setattr(
            "muse.cli.commands.auth._hub_delete",
            lambda url, auth_header, ssl_ctx=None: deleted_key_ids.append(url),
        )

        for orphan in orphans:
            result = runner.invoke(
                None,
                ["auth", "key", "delete", "--hub", _HUB, "--fingerprint", orphan["fingerprint"], "--json"],
            )
            assert result.exit_code == 0, result.output
            data = json.loads(result.output)
            assert data["key_id"] == orphan["key_id"]

        assert len(deleted_key_ids) == 10
        assert len(set(deleted_key_ids)) == 10, "each delete must target a distinct key_id"