105 lines
3 KiB
Python
105 lines
3 KiB
Python
|
import json
|
||
|
|
from types import SimpleNamespace
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from secrets_engine import cli
|
||
|
|
from secrets_engine.config import Config
|
||
|
|
from secrets_engine.errors import BackendError, DecisionError
|
||
|
|
from secrets_engine.openbao import OpenBaoClient, accessor_fingerprint
|
||
|
|
|
||
|
|
|
||
|
|
ACCESSOR = "test-known-accessor-value"
|
||
|
|
|
||
|
|
|
||
|
|
def _config(tmp_path):
|
||
|
|
return Config(
|
||
|
|
catalog_dir=tmp_path,
|
||
|
|
policy_dir=tmp_path,
|
||
|
|
evidence_dir=tmp_path / "evidence",
|
||
|
|
hub_url="",
|
||
|
|
bao_addr="http://127.0.0.1:8200",
|
||
|
|
topic_id="test-topic",
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _accessor_file(tmp_path, value=ACCESSOR, mode=0o600):
|
||
|
|
path = tmp_path / "accessor.handle"
|
||
|
|
path.write_text(value, encoding="utf-8")
|
||
|
|
path.chmod(mode)
|
||
|
|
return path
|
||
|
|
|
||
|
|
|
||
|
|
def test_session_revoke_uses_fingerprint_only(tmp_path, monkeypatch):
|
||
|
|
seen = {}
|
||
|
|
|
||
|
|
class _Client:
|
||
|
|
def revoke_accessor(self, accessor):
|
||
|
|
seen["accessor"] = accessor
|
||
|
|
|
||
|
|
monkeypatch.setattr(
|
||
|
|
cli, "_open_backend", lambda *_args, **_kwargs: _ctx(_Client())
|
||
|
|
)
|
||
|
|
args = SimpleNamespace(
|
||
|
|
accessor_file=str(_accessor_file(tmp_path)),
|
||
|
|
stage="test",
|
||
|
|
bootstrap_token_file=None,
|
||
|
|
auth="auto",
|
||
|
|
)
|
||
|
|
rc = cli.cmd_session_revoke(_config(tmp_path), args)
|
||
|
|
assert rc == 0
|
||
|
|
assert seen["accessor"] == ACCESSOR
|
||
|
|
records = [
|
||
|
|
json.loads(line)
|
||
|
|
for line in next((tmp_path / "evidence").glob("evidence-*.jsonl")).read_text().splitlines()
|
||
|
|
]
|
||
|
|
dumped = json.dumps(records)
|
||
|
|
assert ACCESSOR not in dumped
|
||
|
|
assert records[-1]["detail"]["session_handle"] == accessor_fingerprint(ACCESSOR)
|
||
|
|
assert records[-1]["result"] == "revoked"
|
||
|
|
|
||
|
|
|
||
|
|
class _ctx:
|
||
|
|
def __init__(self, client):
|
||
|
|
self.client = client
|
||
|
|
|
||
|
|
def __enter__(self):
|
||
|
|
return self.client
|
||
|
|
|
||
|
|
def __exit__(self, *_args):
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def test_session_revoke_production_fails_closed_before_backend(tmp_path, monkeypatch):
|
||
|
|
monkeypatch.delenv("SECRETS_ENGINE_UNSAFE_DEMO", raising=False)
|
||
|
|
monkeypatch.setattr(
|
||
|
|
cli.OpenBaoClient,
|
||
|
|
"resolve",
|
||
|
|
lambda *_args, **_kwargs: pytest.fail("backend must not be reached"),
|
||
|
|
)
|
||
|
|
args = SimpleNamespace(
|
||
|
|
accessor_file=str(_accessor_file(tmp_path)),
|
||
|
|
stage="prod",
|
||
|
|
bootstrap_token_file=None,
|
||
|
|
auth="auto",
|
||
|
|
)
|
||
|
|
cfg = Config(
|
||
|
|
catalog_dir=tmp_path,
|
||
|
|
policy_dir=tmp_path,
|
||
|
|
evidence_dir=tmp_path / "evidence",
|
||
|
|
hub_url="http://127.0.0.1:8000",
|
||
|
|
bao_addr="http://127.0.0.1:8200",
|
||
|
|
topic_id="test-topic",
|
||
|
|
)
|
||
|
|
with pytest.raises(DecisionError, match="production action 'session-revoke'"):
|
||
|
|
cli.cmd_session_revoke(cfg, args)
|
||
|
|
|
||
|
|
|
||
|
|
def test_revoke_accessor_rejects_blank(monkeypatch):
|
||
|
|
client = OpenBaoClient(addr="http://example.invalid", token="t", bao_bin="bao")
|
||
|
|
monkeypatch.setattr(
|
||
|
|
client, "_run_ok", lambda *_args, **_kwargs: pytest.fail("must not call bao")
|
||
|
|
)
|
||
|
|
with pytest.raises(BackendError, match="missing or invalid"):
|
||
|
|
client.revoke_accessor(" ")
|