import asyncio
import base64
import os
import sys
import tempfile
from pathlib import Path

import pytest
from cryptography.exceptions import InvalidTag
from httpx import ASGITransport, AsyncClient
from sqlmodel import select

from beaver_gateway.app import AgentRegistry, McpRegistry
from beaver_gateway.frontends.api.frontend import build_app as build_api
from beaver_gateway.frontends.base import GatewayRuntime
from beaver_gateway.keystore import (
    SecretBox,
    SecretsDisabledError,
    SecretStore,
    UnknownSecretError,
    cli,
    serve,
)
from beaver_gateway.security.auth import TokenStore
from beaver_gateway.storage import Database
from beaver_gateway.storage.models import AuditLog, Secret

KEY = "11" * 32
HEADERS = {"Authorization": "Bearer tok"}


@pytest.fixture
async def db():
    root = Path(tempfile.mkdtemp(prefix="beaver-keys-"))
    database = Database(f"sqlite:///{root / 'k.db'}")
    await database.create_all()
    yield database
    await database.dispose()


@pytest.fixture
def store(db: Database) -> SecretStore:
    return SecretStore(db, SecretBox.from_hex(KEY))


def test_box_binds_the_value_to_its_name() -> None:
    box = SecretBox.from_hex(KEY)
    sealed = box.seal("A_KEY", b"hunter22")
    assert b"hunter22" not in sealed
    assert box.open("A_KEY", sealed) == b"hunter22"
    with pytest.raises(InvalidTag):
        box.open("OTHER", sealed)
    with pytest.raises(ValueError, match="32 bytes"):
        SecretBox.from_hex("abcd")


async def test_store_keeps_only_ciphertext_and_reveals_by_name(
    store: SecretStore, db: Database
) -> None:
    await store.put("NODEREAL_BSC", description="bsc rpc", data=b"nr-abcdef123")
    await store.put("GCP_SA", data=b'{"k": 1}', filename="../sa.json")
    async with db.session() as session:
        rows = (await session.exec(select(Secret))).all()
    assert all(b"nr-abcdef123" not in r.sealed for r in rows)
    assert {r.name: r.filename for r in rows} == {
        "GCP_SA": "sa.json",
        "NODEREAL_BSC": None,
    }

    await store.put("NODEREAL_BSC", description="  rpc key ")
    listed = {r.name: r for r in await store.rows()}
    assert listed["NODEREAL_BSC"].description == "rpc key"
    assert listed["NODEREAL_BSC"].used_at is None

    revealed = await store.reveal(["NODEREAL_BSC", "GCP_SA"])
    assert [(r.name, r.data) for r in revealed] == [
        ("NODEREAL_BSC", b"nr-abcdef123"),
        ("GCP_SA", b'{"k": 1}'),
    ]
    assert all(r.used_at is not None for r in await store.rows())
    with pytest.raises(UnknownSecretError):
        await store.reveal(["NODEREAL_BSC", "MISSING"])
    with pytest.raises(ValueError, match="invalid secret name"):
        await store.put("lower", data=b"x")
    with pytest.raises(ValueError, match="needs a value"):
        await store.put("EMPTY", description="no value")
    assert await store.delete("GCP_SA")
    assert not await store.delete("GCP_SA")


async def test_another_key_or_no_key_opens_nothing(
    store: SecretStore, db: Database
) -> None:
    await store.put("A_KEY", data=b"value-123")
    with pytest.raises(SecretsDisabledError, match="SECRETS_KEY changed"):
        await SecretStore(db, SecretBox.from_hex("22" * 32)).reveal(["A_KEY"])
    off = SecretStore(db, None)
    assert not off.enabled
    assert [r.name for r in await off.rows()] == ["A_KEY"]
    with pytest.raises(SecretsDisabledError):
        await off.reveal(["A_KEY"])
    with pytest.raises(SecretsDisabledError):
        await off.put("B_KEY", data=b"x")


@pytest.fixture
async def socket_path(store: SecretStore):
    path = Path(tempfile.mkdtemp(prefix="bk-", dir="/tmp")) / "run" / "gw.sock"  # noqa: S108
    used: list[list[str]] = []

    async def on_use(names):
        used.append(list(names))

    task = asyncio.create_task(serve(store, path, on_use=on_use))
    for _ in range(100):
        if path.exists():
            break
        await asyncio.sleep(0.01)
    yield path, used
    task.cancel()


async def test_socket_is_root_only_and_answers_ls_and_get(
    store: SecretStore, socket_path
) -> None:
    path, used = socket_path
    assert path.stat().st_mode & 0o777 == 0o600
    assert path.parent.stat().st_mode & 0o777 == 0o700
    await store.put("API_TOKEN", description="the api", data=b"tok-123456")

    ls = await asyncio.to_thread(cli.request, {"op": "ls"}, path)
    assert ls["secrets"] == [
        {"name": "API_TOKEN", "description": "the api", "filename": None}
    ]
    got = await asyncio.to_thread(
        cli.request, {"op": "get", "names": ["API_TOKEN"]}, path
    )
    assert base64.b64decode(got["secrets"]["API_TOKEN"]["data"]) == b"tok-123456"
    assert used == [["API_TOKEN"]]
    with pytest.raises(cli.SecretError, match="unknown secret: NOPE"):
        await asyncio.to_thread(cli.request, {"op": "get", "names": ["NOPE"]}, path)
    with pytest.raises(cli.SecretError, match="invalid secret name"):
        await asyncio.to_thread(cli.request, {"op": "get", "names": ["../x"]}, path)


async def test_run_hands_values_and_files_and_masks_the_output(
    store: SecretStore, socket_path, monkeypatch, capfd
) -> None:
    path, _ = socket_path
    await store.put("API_TOKEN", data=b"tok/123456")
    await store.put("SA_JSON", data=b'{"private": "pem"}', filename="sa.json")
    monkeypatch.setattr(
        cli, "ask", lambda argv: cli.request({"op": "get", "names": argv[1:]}, path)
    )
    shm = Path(tempfile.mkdtemp(prefix="bk-shm-"))
    stale = shm / f"{cli.PREFIX}999999-old"
    stale.mkdir()
    script = (
        "import os, sys, pathlib;"
        "p = pathlib.Path(os.environ['SA_JSON']);"
        "print(p.read_text(), oct(p.stat().st_mode & 0o777), p.parent.parent.name.startswith('beaver-secret-'));"
        "print('url', 'https://x/?key=' + os.environ['API_TOKEN'].replace('/', '%2F'));"
        "sys.stderr.write('boom ' + os.environ['API_TOKEN'] + '\\n');"
        "print(p, flush=True); sys.exit(3)"
    )
    code = await asyncio.to_thread(
        cli.run, ["API_TOKEN", "SA_JSON"], [sys.executable, "-c", script], root=shm
    )
    out, err = capfd.readouterr()
    assert code == 3
    lines = out.splitlines()
    assert lines[0] == '{"private": "pem"} 0o600 True'
    assert lines[1] == "url https://x/?key=<secret:API_TOKEN>"
    assert "tok/123456" not in out + err
    assert err == "boom <secret:API_TOKEN>\n"
    assert not Path(lines[2]).exists()
    assert list(shm.iterdir()) == []


def test_masker_never_splits_a_value_across_chunks() -> None:
    masker = cli.Masker(cli.needles_for({"K": b"abcdef123"}))
    out = b"".join(
        [
            masker.feed(b"one abc"),
            masker.feed(b"def1"),
            masker.feed(b"23 two\nthr"),
            masker.flush(),
        ]
    )
    assert out == b"one <secret:K> two\nthr"
    assert cli.Masker({}).feed(b"as is") == b"as is"
    blob = b"x" * (cli.HOLD + 10) + b"abcdef"
    masker = cli.Masker(cli.needles_for({"K": b"abcdef123"}))
    head = masker.feed(blob)
    assert (
        head + masker.feed(b"123") + masker.flush()
        == b"x" * (cli.HOLD + 10) + b"<secret:K>"
    )
    assert cli.needles_for({"S": b"abc"}) == {}


def test_cli_rejects_bad_shapes(capsys) -> None:
    assert cli.main(["run", "A_KEY"]) == 2
    assert cli.main(["run", "bad-name", "--", "true"]) == 2
    assert cli.main(["--root", "get", "../etc"]) == 2
    assert cli.main(["--root", "rm"]) == 2
    assert "usage" in capsys.readouterr().err


class _Runtime:
    def __init__(self, db: Database, store: SecretStore | None) -> None:
        self.tokens = TokenStore(db, bootstrap={"t": "tok"})
        self.tokens.grant("weak", "weak", scope="api")
        self.runtime = GatewayRuntime(
            agents=AgentRegistry([]),
            mcps=McpRegistry([]),
            backends={},
            token_store=self.tokens,
            db=db,
            conversations=object(),
            bus=object(),
            secrets=store,
        )
        self.http = AsyncClient(
            transport=ASGITransport(app=build_api(self.runtime)), base_url="http://api"
        )


async def test_api_writes_secrets_but_never_shows_a_value(
    store: SecretStore, db: Database
) -> None:
    api = _Runtime(db, store).http
    assert (
        await api.get("/secrets", headers={"Authorization": "Bearer weak"})
    ).status_code == 403
    put = await api.put(
        "/secrets/NODEREAL_BSC",
        json={"description": "bsc", "value": "nr-secret-1"},
        headers=HEADERS,
    )
    assert put.status_code == 200, put.text
    upload = await api.put(
        "/secrets/SA_JSON",
        json={"file": {"name": "sa.json", "content": base64.b64encode(b"{}").decode()}},
        headers=HEADERS,
    )
    assert upload.json()["secret"]["kind"] == "file"
    rename = await api.put(
        "/secrets/NODEREAL_BSC", json={"description": "rpc"}, headers=HEADERS
    )
    assert rename.json()["secret"]["size"] == len("nr-secret-1")
    listed = await api.get("/secrets", headers=HEADERS)
    assert listed.json()["enabled"] is True
    assert "nr-secret-1" not in listed.text
    assert [(s["name"], s["description"]) for s in listed.json()["secrets"]] == [
        ("NODEREAL_BSC", "rpc"),
        ("SA_JSON", ""),
    ]
    assert (await store.reveal(["NODEREAL_BSC"]))[0].data == b"nr-secret-1"
    bad = await api.put("/secrets/bad name", json={"value": "x"}, headers=HEADERS)
    assert bad.status_code == 400
    both = await api.put(
        "/secrets/X_KEY",
        json={"value": "x", "file": {"name": "a", "content": ""}},
        headers=HEADERS,
    )
    assert both.status_code == 400
    assert (await api.delete("/secrets/SA_JSON", headers=HEADERS)).status_code == 200
    assert (await api.delete("/secrets/SA_JSON", headers=HEADERS)).status_code == 404
    async with db.session() as session:
        audit = (await session.exec(select(AuditLog))).all()
    assert [a.kind for a in audit] == [
        "secret_set",
        "secret_set",
        "secret_set",
        "secret_delete",
    ]
    assert all("nr-secret-1" not in a.detail_json for a in audit)


async def test_api_without_a_key_lists_but_refuses_writes(db: Database) -> None:
    api = _Runtime(db, SecretStore(db, None)).http
    assert (await api.get("/secrets", headers=HEADERS)).json() == {
        "enabled": False,
        "secrets": [],
    }
    put = await api.put("/secrets/A_KEY", json={"value": "x"}, headers=HEADERS)
    assert put.status_code == 503


def test_cli_runs_with_isolated_stdlib_python() -> None:
    source = Path(cli.__file__).read_text(encoding="utf-8")
    imports = {
        line.split()[1].split(".")[0]
        for line in source.splitlines()
        if line.startswith(("import ", "from "))
    }
    assert imports <= set(sys.stdlib_module_names) | {"__future__"}
    assert os.access(cli.__file__, os.R_OK)