tegwick
19a1c75d30
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a12d13-27f2-7163-b8c0-f682af02772c
172 lines
7.2 KiB
Python
172 lines
7.2 KiB
Python
"""Bootstrap tests use synthetic credentials and fake kubectl, never a cluster."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PI = ROOT / 'sso-mfa/k8s/privacyidea'
|
|
PASSWORD = 'SyntheticAdminCredential12345678901234567890'
|
|
|
|
|
|
@pytest.fixture
|
|
def bootstrap(tmp_path):
|
|
scripts = tmp_path / 'privacyidea'; scripts.mkdir()
|
|
for name in ('bootstrap-admin.sh', 'bootstrap-admin.py', 'bootstrap-admin-remote.py'):
|
|
if (PI / name).exists(): shutil.copyfile(PI / name, scripts / name)
|
|
private = tmp_path / 'private'; private.mkdir(mode=0o700)
|
|
component = private / 'privacyidea'; component.mkdir(mode=0o700)
|
|
file = component / 'secrets.env'; file.write_text('PI_ADMIN_PASSWORD=' + PASSWORD + '\n'); file.chmod(0o600)
|
|
binary = tmp_path / 'bin'; binary.mkdir()
|
|
(binary / 'kubectl').write_text('''#!/usr/bin/python3
|
|
import json,os,sys
|
|
args=sys.argv[1:]
|
|
raw=sys.stdin.read() if '-i' in args or args[-1:]==['-'] else ''
|
|
with open(os.environ['CALLS'],'a') as f:f.write(json.dumps({'args':args,'stdin':raw})+'\\n')
|
|
if args[0]=='get':
|
|
print(os.environ.get('PODS','privacyidea-disposable'))
|
|
elif 'python3' in args:
|
|
if os.environ.get('FAIL'):
|
|
print(os.environ['PASSWORD'],file=sys.stderr);sys.exit(1)
|
|
print(os.environ.get('REMOTE_RESULT','{"result":"verified","mfa_verified":false}'))
|
|
else:
|
|
print(os.environ['PASSWORD'])
|
|
if os.environ.get('FAIL'):sys.exit(1)
|
|
''')
|
|
(binary / 'curl').write_text('#!/bin/sh\necho CURL_FAILED\nexit 1\n')
|
|
(binary / 'openssl').write_text('#!/bin/sh\nprintf "%s" "$PASSWORD"\n')
|
|
for command in binary.iterdir(): command.chmod(0o700)
|
|
env={'PATH':str(binary)+':'+os.environ['PATH'],'HOME':str(tmp_path),'KUBECONFIG':str(tmp_path/'unused'),
|
|
'CALLS':str(tmp_path/'calls'),'PASSWORD':PASSWORD}
|
|
return scripts,private,file,env
|
|
|
|
|
|
def invoke(bootstrap, **extra):
|
|
scripts,private,_,env=bootstrap
|
|
return subprocess.run(['bash','-x',str(scripts/'bootstrap-admin.sh'),str(private)],
|
|
env={**env,**extra},capture_output=True,text=True,timeout=10)
|
|
|
|
|
|
def calls(bootstrap):
|
|
p=Path(bootstrap[3]['CALLS'])
|
|
return [json.loads(line) for line in p.read_text().splitlines()] if p.exists() else []
|
|
|
|
|
|
def test_bootstrap_has_no_secret_output_or_secret_command_arguments(bootstrap):
|
|
result=invoke(bootstrap)
|
|
assert result.returncode==0
|
|
assert bool(PASSWORD in result.stdout+result.stderr) is False
|
|
observed=calls(bootstrap)
|
|
assert bool(PASSWORD in json.dumps([c['args'] for c in observed])) is False
|
|
remote=next(c for c in observed if 'python3' in c['args'])
|
|
assert json.loads(remote['stdin'])=={'password':PASSWORD}
|
|
assert 'MFA' in result.stdout
|
|
|
|
|
|
def test_bootstrap_does_not_create_legacy_trigger_account_or_secret(bootstrap):
|
|
result=invoke(bootstrap)
|
|
assert result.returncode==0
|
|
commands=[c['args'] for c in calls(bootstrap)]
|
|
assert all('trigger-admin' not in ' '.join(c) for c in commands)
|
|
assert all(c[0] not in ('create','apply','delete') for c in commands)
|
|
|
|
|
|
def test_failed_admin_operation_is_not_reported_as_complete(bootstrap):
|
|
result=invoke(bootstrap,FAIL='1')
|
|
assert result.returncode!=0
|
|
assert bool(PASSWORD in result.stdout+result.stderr) is False
|
|
assert 'bootstrap complete' not in result.stdout.lower()
|
|
|
|
|
|
@pytest.mark.parametrize('pods',['','privacyidea-one\nprivacyidea-two','unexpected/shape'])
|
|
def test_missing_or_ambiguous_pod_refused(bootstrap,pods):
|
|
result=invoke(bootstrap,PODS=pods)
|
|
assert result.returncode!=0
|
|
assert all(c['args'][0]=='get' for c in calls(bootstrap))
|
|
|
|
|
|
@pytest.mark.parametrize('defect',['missing','duplicate','public-file','public-directory','symlink','hardlink','executable-value'])
|
|
def test_invalid_credential_input_never_reaches_cluster(bootstrap,defect):
|
|
_,private,file,_=bootstrap
|
|
marker=private/'executed'
|
|
if defect=='missing':file.unlink()
|
|
elif defect=='duplicate':file.write_text(file.read_text()*2)
|
|
elif defect=='public-file':file.chmod(0o644)
|
|
elif defect=='public-directory':file.parent.chmod(0o755)
|
|
elif defect=='symlink':
|
|
target=file.with_name('other');file.rename(target);file.symlink_to(target)
|
|
elif defect=='hardlink':os.link(file,file.with_name('alias'))
|
|
else:file.write_text('PI_ADMIN_PASSWORD=$(touch '+str(marker)+'; echo '+PASSWORD+')\n')
|
|
result=invoke(bootstrap)
|
|
assert result.returncode!=0
|
|
assert not marker.exists()
|
|
assert not calls(bootstrap)
|
|
assert bool(PASSWORD in result.stdout+result.stderr) is False
|
|
|
|
|
|
@pytest.mark.parametrize('response',['not-json','{"result":"verified","mfa_verified":0}','{"result":"verified","mfa_verified":true}', '{"result":"verified","mfa_verified":false,"password":"extra"}'])
|
|
def test_malformed_or_unexpected_remote_receipt_refused(bootstrap,response):
|
|
result=invoke(bootstrap,REMOTE_RESULT=response)
|
|
assert result.returncode!=0
|
|
assert response not in result.stdout+result.stderr
|
|
|
|
|
|
@pytest.fixture
|
|
def remote():
|
|
import importlib.util
|
|
spec=importlib.util.spec_from_file_location('admin_bootstrap_remote',PI/'bootstrap-admin-remote.py')
|
|
module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def policy():
|
|
return {'name':'pi-admin-all-rights','scope':'admin','adminuser':['pi-admin'],
|
|
'action':{'*':True},'active':True,'priority':1,'check_all_resolvers':False,
|
|
'user_case_insensitive':False}
|
|
|
|
|
|
def test_new_account_and_scoped_policy_requested_together(remote):
|
|
effects=[]
|
|
def insert(password,**flags): effects.append(flags)
|
|
result=remote.ensure(PASSWORD,[],[],lambda *_:bool(effects),insert)
|
|
assert effects==[{'account':True,'policy':True}]
|
|
assert result=={'result':'created','mfa_verified':False}
|
|
|
|
|
|
def test_existing_matching_account_is_verified_without_mutation(remote):
|
|
effects=[]
|
|
result=remote.ensure(PASSWORD,['pi-admin'],[policy()],lambda *_:True,lambda *a,**kw:effects.append(kw))
|
|
assert result=={'result':'verified','mfa_verified':False}
|
|
assert effects==[]
|
|
|
|
|
|
def test_existing_mismatching_account_is_never_reset(remote):
|
|
effects=[]
|
|
with pytest.raises(remote.Refused):
|
|
remote.ensure(PASSWORD,['pi-admin'],[policy()],lambda *_:False,lambda *a,**kw:effects.append(kw))
|
|
assert effects==[]
|
|
|
|
|
|
@pytest.mark.parametrize('change',[{'adminuser':[]},{'scope':'user'},{'action':{'triggerchallenge':True}},
|
|
{'active':False},{'conditions':[('restriction',)]},{'user_case_insensitive':True}])
|
|
def test_existing_unexpected_policy_is_never_overwritten(remote,change):
|
|
effects=[]
|
|
with pytest.raises(remote.Refused):
|
|
remote.ensure(PASSWORD,['pi-admin'],[{**policy(),**change}],lambda *_:True,lambda *a,**kw:effects.append(kw))
|
|
assert effects==[]
|
|
|
|
|
|
@pytest.mark.parametrize('admins,policies',[(['other-admin'],[]),(['pi-admin'],[{'name':'other','scope':'admin'}])])
|
|
def test_policy_migration_requires_owner_review(remote,admins,policies):
|
|
effects=[]
|
|
with pytest.raises(remote.Refused):
|
|
remote.ensure(PASSWORD,admins,policies,lambda *_:True,lambda *a,**kw:effects.append(kw))
|
|
assert effects==[]
|
|
|
|
|
|
def test_failed_post_creation_verification_is_failure(remote):
|
|
with pytest.raises(remote.Refused):
|
|
remote.ensure(PASSWORD,[],[],lambda *_:False,lambda *a,**kw:None)
|