300 lines
11 KiB
Python
300 lines
11 KiB
Python
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)
|