"""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)