72 lines
3.4 KiB
Python
72 lines
3.4 KiB
Python
|
from types import SimpleNamespace
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from secrets_engine import cli
|
||
|
|
from secrets_engine.engine_auth import require_backend_execution_ready
|
||
|
|
from secrets_engine.errors import BackendError
|
||
|
|
|
||
|
|
|
||
|
|
def cfg(contract=None):
|
||
|
|
return SimpleNamespace(openbao_jwt_login_file=contract)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('provider,contract', [('service-jwt', None), ('auto', '/non-secret/login.yaml')])
|
||
|
|
def test_login_only_refused_before_claim_consume_or_backend(provider, contract, monkeypatch):
|
||
|
|
events = []
|
||
|
|
for name in ['authorize_action', 'require_production_consume', 'login_service_jwt']:
|
||
|
|
monkeypatch.setattr(cli, name, lambda *a, _name=name, **k: events.append(_name))
|
||
|
|
monkeypatch.setattr(cli.OpenBaoClient, 'resolve', lambda *a, **k: events.append('backend'))
|
||
|
|
args = SimpleNamespace(auth=provider, bootstrap_token_file=None)
|
||
|
|
with pytest.raises(BackendError, match='Approval was not consumed'):
|
||
|
|
cli._require_lane_approval(cfg(contract), SimpleNamespace(stage='prod'), 'apply', auth_args=args)
|
||
|
|
assert events == []
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('provider,bootstrap', [('env', None), ('bootstrap', '/private/explicit-token')])
|
||
|
|
def test_named_legacy_provider_retained(provider, bootstrap):
|
||
|
|
args = SimpleNamespace(auth=provider, bootstrap_token_file=bootstrap)
|
||
|
|
result = require_backend_execution_ready(cfg('/published/login.yaml'), args, 'apply')
|
||
|
|
assert result.provider == provider
|
||
|
|
assert result.break_glass is (provider == 'bootstrap')
|
||
|
|
|
||
|
|
|
||
|
|
def test_auto_with_contract_and_bootstrap_never_falls_back():
|
||
|
|
with pytest.raises(BackendError, match='no fallback'):
|
||
|
|
require_backend_execution_ready(cfg('/published/login.yaml'),
|
||
|
|
SimpleNamespace(auth='auto', bootstrap_token_file='/private/file'), 'apply')
|
||
|
|
|
||
|
|
|
||
|
|
def test_readiness_cannot_be_self_asserted():
|
||
|
|
c = cfg('/published/login.yaml')
|
||
|
|
c.execution_entitlement_file = '/pretend/approved.json'
|
||
|
|
c.service_jwt_execution_ready = True
|
||
|
|
with pytest.raises(BackendError, match='no reviewed task-scoped'):
|
||
|
|
require_backend_execution_ready(c, SimpleNamespace(auth='auto', bootstrap_token_file=None), 'apply')
|
||
|
|
|
||
|
|
|
||
|
|
def test_self_probe_auth_selection_remains_service_jwt():
|
||
|
|
from secrets_engine.engine_auth import select_engine_auth
|
||
|
|
assert select_engine_auth(cfg('/published/login.yaml'),
|
||
|
|
SimpleNamespace(auth='auto', bootstrap_token_file=None)).provider == 'service-jwt'
|
||
|
|
|
||
|
|
|
||
|
|
def test_real_apply_handler_refuses_before_consumption(tmp_path, monkeypatch):
|
||
|
|
import copy
|
||
|
|
from secrets_engine.catalog import validate_entry
|
||
|
|
from secrets_engine.config import Config
|
||
|
|
from tests.test_catalog import VALID
|
||
|
|
entry = validate_entry(copy.deepcopy(VALID))
|
||
|
|
configuration = 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='fixture', openbao_jwt_login_file=tmp_path / 'login-contract.yaml')
|
||
|
|
monkeypatch.setattr(cli, '_resolve_lane_and_decision', lambda *a: entry)
|
||
|
|
calls = []
|
||
|
|
for name in ['authorize_action', 'require_production_consume', 'login_service_jwt', 'apply_plan']:
|
||
|
|
monkeypatch.setattr(cli, name, lambda *a, _name=name, **k: calls.append(_name))
|
||
|
|
args = SimpleNamespace(ref=entry.id, stage=entry.stage, dry_run=False,
|
||
|
|
auth='auto', bootstrap_token_file=None)
|
||
|
|
with pytest.raises(BackendError, match='Approval was not consumed'):
|
||
|
|
cli.cmd_apply(configuration, args)
|
||
|
|
assert calls == []
|