security: preserve realm configuration and remove shell credential evaluation

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a1346b-72a4-7d40-b8bd-8b8ca346cfa2

tests/test_realm_bootstrap.py

@@ -0,0 +1,207 @@
+"""Configuration-preserving realm repair, private input and real HTTP boundaries."""
+import copy
+import importlib
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import threading
+from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
+from urllib.parse import urlsplit
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+HELPERS=ROOT/'sso-mfa/k8s/privacyidea'
+sys.path.insert(0,str(HELPERS))
+r=importlib.import_module('realm_bootstrap')
+p=importlib.import_module('pi_api')
+
+
+def snapshot():
+    return {'resolver':{'type':'ldapresolver','data':dict(r.CORE,**{**r.TUNING,'SIZELIMIT':'700'},BINDPW='__CENSORED__',TLS_VERIFY='True')},
+            'realms':{'coulomb':{'resolver':[{'name':r.RESOLVER,'priority':5}], 'default':False},
+                      'other':{'resolver':[], 'default':True}}}
+
+
+class Fake:
+    def __init__(self,state): self.state=copy.deepcopy(state);self.calls=[];self.fail=None;self.drift=None;self.reads=0
+    def snapshot(self):
+        self.reads+=1
+        if self.drift and self.reads==2:self.state['realms']['other']['default']=False
+        return copy.deepcopy(self.state)
+    def call(self,path,payload):
+        self.calls.append((path,copy.deepcopy(payload)))
+        if self.fail==path:raise r.Refused()
+        if path=='/resolver/'+r.RESOLVER:
+            if self.state['resolver'] is None:self.state['resolver']={'type':'ldapresolver','data':{}}
+            self.state['resolver']['data'].update({k:('__CENSORED__' if k=='BINDPW' else v) for k,v in payload.items() if k!='type'})
+            return 1
+        if path=='/realm/coulomb':
+            self.state['realms']['coulomb']={'resolver':[{'name':r.RESOLVER,'priority':1}],'default':False}
+            return {'added':[r.RESOLVER],'failed':[]}
+        if path=='/defaultrealm/coulomb':self.state['realms']['coulomb']['default']=True;return 1
+        raise AssertionError('unexpected write')
+
+
+def test_existing_configuration_is_noop_without_reading_bind_password():
+    client=Fake(snapshot());before=copy.deepcopy(client.state)
+    def forbidden():raise AssertionError('existing bind must not be requested')
+    report=r.repair(client,forbidden,apply=True)
+    assert report['configuration_writes']==0 and client.state==before and not client.calls
+
+
+def test_missing_tuning_is_minimal_and_preserves_nondefault_values():
+    client=Fake(snapshot());del client.state['resolver']['data']['TIMEOUT'];before=copy.deepcopy(client.state)
+    report=r.repair(client,lambda:None,apply=True)
+    assert report['configuration_writes']==1
+    assert client.calls==[('/resolver/'+r.RESOLVER,{'TIMEOUT':'5','type':'ldapresolver'})]
+    before['resolver']['data']['TIMEOUT']='5';assert client.state==before
+
+
+def test_missing_objects_are_created_and_other_default_is_preserved():
+    state=snapshot();state['resolver']=None;del state['realms']['coulomb'];client=Fake(state)
+    result=r.repair(client,lambda:'fixture-bind',apply=True)
+    assert result['actions']==['create-resolver','create-realm']
+    assert client.state['realms']['other']==state['realms']['other']
+    assert all(not route.startswith('/policy') for route,_ in client.calls)
+
+
+def test_first_realm_gets_explicit_initial_default():
+    client=Fake({'resolver':None,'realms':{}})
+    result=r.repair(client,lambda:'fixture-bind',apply=True)
+    assert result['configuration_writes']==3 and client.state['realms']['coulomb']['default'] is True
+
+
+def test_preview_has_no_configuration_writes_or_bind_prompt():
+    client=Fake({'resolver':None,'realms':{}})
+    report=r.repair(client,lambda:(_ for _ in ()).throw(AssertionError()),apply=False)
+    assert report['result']=='planned' and not client.calls
+
+
+@pytest.mark.parametrize('field,value',[('type','other'),('LDAPURI','ldap://other'),('TIMEOUT','invalid')])
+def test_conflicting_profile_refused_before_any_write(field,value):
+    state=snapshot()
+    if field=='type':state['resolver']['type']=value
+    else:state['resolver']['data'][field]=value
+    client=Fake(state)
+    with pytest.raises(r.Refused):r.repair(client,lambda:'fixture-bind',apply=True)
+    assert not client.calls
+
+
+def test_existing_wrong_realm_binding_refused_before_resolver_write():
+    state=snapshot();state['realms']['coulomb']['resolver']=[];del state['resolver']['data']['TIMEOUT'];client=Fake(state)
+    with pytest.raises(r.Refused):r.repair(client,lambda:None,apply=True)
+    assert not client.calls
+
+
+def test_preflight_drift_refused():
+    client=Fake(snapshot());del client.state['resolver']['data']['TIMEOUT'];client.drift=True
+    with pytest.raises(r.Refused):r.repair(client,lambda:None,apply=True)
+    assert not client.calls
+
+
+def test_partial_failure_stops_without_policy_or_default_writes():
+    client=Fake({'resolver':None,'realms':{}});client.fail='/realm/coulomb'
+    with pytest.raises(r.Refused):r.repair(client,lambda:'fixture-bind',apply=True)
+    assert len(client.calls)==2 and client.state['resolver'] is not None
+
+
+@pytest.fixture
+def server():
+    state={'requests':[],'response':{'result':{'status':True,'value':{}}},'status':200}
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self,*_):pass
+        def do_GET(self):self.respond()
+        def do_POST(self):self.respond()
+        def respond(self):
+            body=self.rfile.read(int(self.headers.get('Content-Length',0)))
+            state['requests'].append((self.path,dict(self.headers),body))
+            self.send_response(state['status']);self.send_header('Location','/leak');self.end_headers()
+            self.wfile.write(state.get('raw',json.dumps(state['response']).encode()))
+    s=ThreadingHTTPServer(('127.0.0.1',0),Handler);t=threading.Thread(target=s.serve_forever,daemon=True);t.start()
+    try:yield f'http://127.0.0.1:{s.server_port}',state
+    finally:s.shutdown();s.server_close();t.join()
+
+
+@pytest.mark.parametrize('status',[301,302,307,308,401,403,500])
+def test_transport_does_not_follow_or_render_provider_failure(server,status):
+    url,state=server;state['status']=status;state['raw']=b'private-provider-error'
+    code,body=p.request(url+'/auth',{'password':'synthetic-password'},'synthetic-token')
+    assert code==status and body is None and len(state['requests'])==1
+
+
+def test_transport_ignores_proxy_and_preserves_bodyless_get(server,monkeypatch):
+    url,state=server
+    for name in ['http_proxy','HTTP_PROXY']:monkeypatch.setenv(name,'http://127.0.0.1:1')
+    for name in ['no_proxy','NO_PROXY']:monkeypatch.setenv(name,'')
+    assert p.request(url+'/realm/',token='fixture-token')[0]==200
+    assert not state['requests'][0][2] and 'Content-Type' not in state['requests'][0][1]
+
+
+@pytest.mark.parametrize('raw',[b'x'*(p.MAX_BODY+1),b'{"result":{"status":true,"status":false}}',b'bad-json'],ids=['oversized','duplicate','invalid'])
+def test_transport_invalid_response_is_not_forwarded(server,raw):
+    url,state=server;state['raw']=raw
+    assert p.request(url+'/realm/')[1] is None
+
+
+def test_transport_refuses_unapproved_destination():
+    assert p.request('https://other.invalid/auth',{'password':'synthetic'})==(0,None)
+
+
+@pytest.fixture
+def private_root(tmp_path):
+    for component,key in [('privacyidea','PI_ADMIN_PASSWORD'),('lldap','LLDAP_LDAP_USER_PASS')]:
+        directory=tmp_path/component;directory.mkdir(mode=0o700)
+        f=directory/'secrets.env';f.write_text(key+'=literal-$(never-execute)-quote\'"\\end\n');f.chmod(0o600)
+    return tmp_path
+
+
+def test_private_env_is_literal_not_shell(private_root):
+    a,b=r.private_inputs(private_root);assert a==b=="literal-$(never-execute)-quote'\"\\end"
+
+
+@pytest.mark.parametrize('kind',['symlink','hardlink','fifo','public-file','public-dir','duplicate','invalid'])
+def test_private_input_rejected_before_network(private_root,server,kind):
+    f=private_root/'privacyidea/secrets.env'
+    if kind=='symlink':f.rename(f.with_suffix('.real'));f.symlink_to(f.with_suffix('.real'))
+    elif kind=='hardlink':os.link(f,f.with_suffix('.link'))
+    elif kind=='fifo':f.unlink();os.mkfifo(f,0o600)
+    elif kind=='public-file':f.chmod(0o644)
+    elif kind=='public-dir':f.parent.chmod(0o755)
+    elif kind=='duplicate':f.write_text('PI_ADMIN_PASSWORD=one\nPI_ADMIN_PASSWORD=two\n')
+    elif kind=='invalid':f.write_text('touch forbidden-marker\n')
+    url,state=server
+    proc=subprocess.run(['bash','-x',str(HELPERS/'bootstrap-realm.sh'),str(private_root),url,'--apply'],capture_output=True,text=True,timeout=10)
+    assert proc.returncode==1 and not proc.stdout and not state['requests']
+    assert 'literal-' not in proc.stderr and str(f) not in proc.stderr
+
+
+def test_attended_wrapper_refuses_without_a_terminal():
+    proc=subprocess.run(['bash','-x',str(HELPERS/'repair-realm-live.sh')],stdin=subprocess.DEVNULL,capture_output=True,text=True)
+    assert proc.returncode==1 and not proc.stdout and '"phase": "input"' in proc.stderr
+
+
+def test_real_provider_merge_and_policy_preservation():
+    pytest.importorskip('privacyidea')
+    proc=subprocess.run([sys.executable,str(ROOT/'tests/realm_bootstrap_native_fixture.py')],capture_output=True,text=True,timeout=120)
+    assert proc.returncode==0,proc.stdout
+    result=json.loads(proc.stdout);assert len(result)==6 and all(v is True for v in result.values())
+
+
+@pytest.mark.parametrize('role',[None,'user'])
+def test_login_must_confirm_admin_role(server,role):
+    url,state=server;state['response']={'result':{'status':True,'value':{'token':'fixture-token','role':role}}}
+    with pytest.raises(r.Refused):r.Client(url,'fixture-password')
+    assert len(state['requests'])==1 and state['requests'][0][0]=='/auth'
+
+
+def test_default_readback_refuses_unexpected_other_realm():
+    class Concurrent(Fake):
+        def call(self,path,payload):
+            value=super().call(path,payload)
+            if path=='/defaultrealm/coulomb':self.state['realms']['unreviewed']={'default':False,'resolver':[]}
+            return value
+    client=Concurrent({'resolver':None,'realms':{}})
+    with pytest.raises(r.Refused):r.repair(client,lambda:'fixture-bind',apply=True)
+    assert len(client.calls)==3