tegwick
4dffe2fe7a
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a1346b-72a4-7d40-b8bd-8b8ca346cfa2
252 lines
12 KiB
Python
252 lines
12 KiB
Python
"""Cutover state machine and credential-scoped HTTP regression checks."""
|
|
import copy
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
from types import SimpleNamespace
|
|
import pytest
|
|
|
|
from test_realm_bootstrap import snapshot, server, HELPERS
|
|
sys.path.insert(0, str(HELPERS))
|
|
import resolver_cutover as c
|
|
import pi_api as p
|
|
|
|
TARGET = dict(helper='update', pi_url='http://127.0.0.1:1234', lldap_url='http://127.0.0.1:1235',
|
|
keycape_url='http://127.0.0.1:1236', user='fixture-user', kubeconfig='/tmp/fixture',
|
|
context='fixture', server='https://127.0.0.1:6443', cluster_uid='fixture-uid')
|
|
|
|
|
|
class Client:
|
|
def __init__(self):
|
|
self.state = snapshot(); self.calls = []; self.reads = 0; self.drift = False
|
|
self.bad_bind = False; self.post_failure = False; self.users = [{'username':'fixture-user','resolver':c.realm.RESOLVER}]
|
|
def snapshot(self):
|
|
self.reads += 1
|
|
if self.drift and self.reads == 2: self.state['resolver']['data']['TIMEOUT'] = '99'
|
|
return copy.deepcopy(self.state)
|
|
def call(self, path, payload=None):
|
|
self.calls.append((path, copy.deepcopy(payload)))
|
|
if path == '/resolver/test': return not self.bad_bind
|
|
if path.startswith('/user/'): return self.users
|
|
if path == '/resolver/' + c.realm.RESOLVER:
|
|
if self.post_failure: raise c.Refused()
|
|
return 1
|
|
raise AssertionError()
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch):
|
|
monkeypatch.setattr(c, 'login', lambda *a, **kw: None)
|
|
return Client()
|
|
|
|
|
|
def apply(client, target=TARGET, **kw):
|
|
options = dict(replacement='synthetic-new', expected_digest=c.digest(client.state, target))
|
|
options.update(kw)
|
|
return c.run(client, target, 'apply', **options)
|
|
|
|
|
|
def writes(client): return [x for x in client.calls if x[0] == '/resolver/' + c.realm.RESOLVER]
|
|
|
|
|
|
def test_minimal_update_and_no_fabricated_acceptance(client):
|
|
before=copy.deepcopy(client.state)
|
|
report=apply(client)
|
|
assert writes(client)==[('/resolver/'+c.realm.RESOLVER, {'type':'ldapresolver','BINDPW':'synthetic-new'})]
|
|
assert client.state==before and report['result']=='configured'
|
|
assert report['incident_acceptance']=='unverified' and not report['mfa_verified']
|
|
tests=[payload for path,payload in client.calls if path=='/resolver/test']
|
|
assert [x['BINDPW'] for x in tests]==['synthetic-new','__CENSORED__']
|
|
assert tests[0]['SIZELIMIT']=='700'
|
|
assert 'synthetic-new' not in json.dumps(report)
|
|
|
|
|
|
def test_preview_without_candidate_or_write(client):
|
|
report=c.run(client,TARGET,'preview')
|
|
assert report['result']=='planned' and not client.calls
|
|
|
|
|
|
@pytest.mark.parametrize('change', ['missing','core','tuning','realm','uncensored'])
|
|
def test_incomplete_or_conflicting_setup_refused(client,change):
|
|
if change=='missing':client.state['resolver']=None
|
|
elif change=='core':client.state['resolver']['data']['LDAPURI']='ldap://other'
|
|
elif change=='tuning':del client.state['resolver']['data']['TIMEOUT']
|
|
elif change=='realm':client.state['realms']['coulomb']['resolver']=[]
|
|
else:client.state['resolver']['data']['BINDPW']='synthetic-uncensored'
|
|
with pytest.raises(c.Refused):c.run(client,TARGET,'apply',replacement='synthetic-new',expected_digest='0'*64)
|
|
assert not writes(client)
|
|
|
|
|
|
@pytest.mark.parametrize('failure',['stale','drift','candidate'])
|
|
def test_prewrite_failures_never_mutate(client,failure):
|
|
client.drift=failure=='drift';client.bad_bind=failure=='candidate'
|
|
opts={'expected_digest':'0'*64} if failure=='stale' else {}
|
|
with pytest.raises(c.Refused):apply(client,**opts)
|
|
assert not writes(client)
|
|
|
|
|
|
def test_ambiguous_write_is_not_retried_and_receipt_remains_unverified(client):
|
|
client.post_failure=True;report={}
|
|
with pytest.raises(c.Refused):apply(client,report=report)
|
|
assert report['result']=='unverified' and report['write_attempted']
|
|
assert report['configuration_writes']==0 and len(writes(client))==1
|
|
|
|
|
|
@pytest.mark.parametrize('predecessor,result',[(None,'incomplete'),('synthetic-old','verified')])
|
|
def test_full_proofs_and_honest_predecessor_result(client,predecessor,result):
|
|
target=dict(TARGET,helper='reconcile');gates=[]
|
|
report=apply(client,target,predecessor=predecessor,gate=lambda:gates.append(1),mfa_proof=lambda:{'result':'PASS'})
|
|
assert report['result']==result and report['mfa_verified'] and len(gates)==2
|
|
assert report['supplied_predecessor_denied']==bool(predecessor)
|
|
assert report['incident_acceptance']=='unverified'
|
|
|
|
|
|
@pytest.mark.parametrize('users',[[],[{'username':'other','resolver':c.realm.RESOLVER}],
|
|
[{'username':'fixture-user','resolver':'wrong'}],[{'username':'fixture-user','resolver':c.realm.RESOLVER}]*2])
|
|
def test_exact_user_before_mfa(client,users):
|
|
client.users=users
|
|
with pytest.raises(c.Refused):apply(client,dict(TARGET,helper='reconcile'),gate=lambda:None,mfa_proof=lambda:pytest.fail('OTP too early'))
|
|
|
|
|
|
@pytest.mark.parametrize('status',[0,200,302,400,500])
|
|
def test_predecessor_network_or_unexpected_result_not_denial(monkeypatch,status):
|
|
monkeypatch.setattr(c,'request',lambda *a,**kw:(status,{'token':'synthetic'}))
|
|
with pytest.raises(c.Refused):c.login(TARGET['lldap_url'],'old',denied=True)
|
|
|
|
|
|
@pytest.mark.parametrize('profile,path,payload',[
|
|
('lldap-auth','/auth/simple/login',{'username':'admin','password':'synthetic'}),
|
|
('keycape-discovery','/.well-known/openid-configuration',None)])
|
|
def test_named_profile_actual_http(server,profile,path,payload):
|
|
base,state=server
|
|
assert p.request(base+path,payload,profile=profile)[0]==200
|
|
assert len(state['requests'])==1
|
|
assert state['requests'][0][2]==(json.dumps(payload).encode() if payload else b'')
|
|
assert 'Authorization' not in state['requests'][0][1]
|
|
|
|
|
|
@pytest.mark.parametrize('profile,path,payload,token,method',[
|
|
('lldap-auth','/other',{'username':'admin','password':'synthetic'},None,None),
|
|
('lldap-auth','/auth/simple/login',{'username':'pi-admin','password':'synthetic'},None,None),
|
|
('lldap-auth','/auth/simple/login',{'username':'admin','password':'synthetic'},'pi-token',None),
|
|
('lldap-auth','/auth/simple/login?x=1',{'username':'admin','password':'synthetic'},None,None),
|
|
('lldap-auth','/auth/simple/login',{'username':'admin','password':'synthetic'},None,'PUT'),
|
|
('keycape-discovery','/.well-known/openid-configuration',{'password':'synthetic'},None,None),
|
|
('keycape-discovery','/.well-known/openid-configuration',None,'pi-token',None),
|
|
('unknown','/auth',{},None,None)])
|
|
def test_profile_refusal_before_socket(server,profile,path,payload,token,method):
|
|
base,state=server
|
|
assert p.request(base+path,payload,token,profile=profile,method=method)==(0,None)
|
|
assert not state['requests']
|
|
|
|
|
|
@pytest.mark.parametrize('profile',['lldap-auth','keycape-discovery'])
|
|
@pytest.mark.parametrize('origin',['http://lldap.coulomb.social','https://evil.invalid','https://pink.coulomb.social','http://localhost:1234'])
|
|
def test_origin_scope(profile,origin):
|
|
with pytest.raises(Exception):p.service_endpoint(origin,profile)
|
|
|
|
|
|
@pytest.mark.parametrize('status',[301,302,307,308,500])
|
|
def test_profile_no_redirect(server,status):
|
|
base,state=server;state['status']=status
|
|
assert p.request(base+'/auth/simple/login',{'username':'admin','password':'synthetic'},profile='lldap-auth')==(status,None)
|
|
assert len(state['requests'])==1
|
|
|
|
|
|
@pytest.mark.parametrize('bad',['server','uid','generation','replicas','deletion'])
|
|
def test_cluster_pin_and_current_generation(monkeypatch,bad):
|
|
calls=[]
|
|
def command(argv,**kw):
|
|
calls.append(argv)
|
|
if 'view' in argv:out='https://wrong:6443' if bad=='server' else TARGET['server']
|
|
elif 'namespace' in argv:out='wrong' if bad=='uid' else TARGET['cluster_uid']
|
|
else:out={'generation':'4|3|1|1|1|1|1|','replicas':'4|4|1|2|1|1|1|','deletion':'4|4|1|1|1|1|1|now'}.get(bad,'4|4|1|1|1|1|1|')
|
|
return SimpleNamespace(returncode=0,stdout=out)
|
|
monkeypatch.setattr(c.subprocess,'run',command)
|
|
with pytest.raises(c.Refused):c.cluster_ready(TARGET)
|
|
assert all(argv[1:5]==['--kubeconfig','/tmp/fixture','--context','fixture'] for argv in calls)
|
|
assert not any('secret' in argv or '--raw' in argv for argv in calls)
|
|
|
|
|
|
@pytest.mark.parametrize('script',['update-lldap-resolver-live.sh','reconcile-lldap-resolver-live.sh'])
|
|
def test_unattended_and_unsafe_legacy_argument_refuse_without_echo(script):
|
|
result=subprocess.run(['bash','-x',str(HELPERS/script),'--note','synthetic-private-note'],capture_output=True,text=True)
|
|
assert result.returncode!=0 and 'synthetic-private-note' not in result.stdout+result.stderr
|
|
assert 'Traceback' not in result.stderr and 'password:' not in result.stderr
|
|
|
|
|
|
@pytest.mark.parametrize('broken',['issuer','endpoint','redirect','malformed'])
|
|
def test_health_semantics(monkeypatch,broken):
|
|
issuer='https://kc.coulomb.social'
|
|
body={'issuer':issuer,**{k:issuer+'/'+k for k in ('authorization_endpoint','token_endpoint','jwks_uri')}}
|
|
if broken=='issuer':body['issuer']='https://other.invalid'
|
|
if broken=='endpoint':body['token_endpoint']='https://kc.coulomb.social.evil.invalid/token'
|
|
if broken=='malformed':body=[]
|
|
monkeypatch.setattr(c,'request',lambda url,**kw:(401,None) if url.endswith('/token/') else (302 if broken=='redirect' else 200,body))
|
|
with pytest.raises(c.Refused):c.health(TARGET)
|
|
|
|
|
|
@pytest.mark.parametrize('script',['update-lldap-resolver-live.sh','reconcile-lldap-resolver-live.sh'])
|
|
def test_pty_tracing_never_discloses_password(server,tmp_path,script):
|
|
import pty
|
|
import select
|
|
import signal
|
|
import time
|
|
if script.startswith('reconcile'):
|
|
# Pin validation must happen before any credential prompt.
|
|
argv=['--preview','--server','https://bad.invalid']
|
|
else:
|
|
argv=['--preview','--pi-url',server[0]]
|
|
server[1]['response']={'result':{'status':True,'value':{'role':'admin','token':'fixture-token'}}}
|
|
pid,fd=pty.fork()
|
|
if pid==0:
|
|
os.environ['TMPDIR']=str(tmp_path)
|
|
os.execv('/bin/bash',['bash','-x',str(HELPERS/script),*argv])
|
|
output=b'';sent=False;status=None
|
|
try:
|
|
deadline=time.monotonic()+10
|
|
while time.monotonic()<deadline:
|
|
if select.select([fd],[],[],.1)[0]:
|
|
try:chunk=os.read(fd,65536)
|
|
except OSError:break
|
|
if not chunk:break
|
|
output+=chunk
|
|
if b'pi-admin password:' in output and not sent:
|
|
os.write(fd,b'fixture-private-terminal-value\n');sent=True
|
|
done,value=os.waitpid(pid,os.WNOHANG)
|
|
if done:status=value;break
|
|
if status is None:
|
|
done,value=os.waitpid(pid,os.WNOHANG)
|
|
if not done:
|
|
os.kill(pid,signal.SIGKILL);os.waitpid(pid,0);pytest.fail('attended helper timed out')
|
|
status=value
|
|
assert os.waitstatus_to_exitcode(status)!=0
|
|
assert sent == script.startswith('update')
|
|
assert b'fixture-private-terminal-value' not in output and b'Traceback' not in output
|
|
assert not list(tmp_path.iterdir())
|
|
finally:os.close(fd)
|
|
|
|
|
|
def test_postwrite_drift_stops_before_success(client,monkeypatch):
|
|
original=client.call
|
|
def change(path,payload=None):
|
|
value=original(path,payload)
|
|
if path=='/resolver/'+c.realm.RESOLVER:client.state['resolver']['data']['SIZELIMIT']='500'
|
|
return value
|
|
monkeypatch.setattr(client,'call',change);report={}
|
|
with pytest.raises(c.Refused):apply(client,report=report)
|
|
assert report['write_attempted'] and report['configuration_writes']==1
|
|
assert report['result']=='unverified' and report['phase']=='stored-bind-verification'
|
|
|
|
|
|
def test_mfa_failure_never_reports_success(client):
|
|
report={}
|
|
with pytest.raises(c.Refused):apply(client,dict(TARGET,helper='reconcile'),gate=lambda:None,mfa_proof=lambda:{'result':'FAIL'},report=report)
|
|
assert report['phase']=='token-backed-mfa' and not report['mfa_verified']
|
|
|
|
|
|
def test_censored_sentinel_is_not_a_replacement(client):
|
|
with pytest.raises(c.Refused):apply(client,replacement='__CENSORED__')
|
|
assert not client.calls
|