tegwick
4dffe2fe7a
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a1346b-72a4-7d40-b8bd-8b8ca346cfa2
141 lines
8 KiB
Python
141 lines
8 KiB
Python
"""Local LLDAP + privacyIDEA 3.12.2 bind/cutover/HOTP proof; synthetic data only."""
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
|
|
|
|
def provider_run(lldap_url, ldap_url, replacement):
|
|
for name in list(os.environ):
|
|
if name.startswith('PRIVACYIDEA_'):del os.environ[name]
|
|
logging.disable(logging.CRITICAL)
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'sso-mfa/k8s/privacyidea'))
|
|
import realm_bootstrap as r
|
|
import resolver_cutover as cut
|
|
from privacyidea.lib.token import init_token
|
|
from privacyidea.lib.user import User
|
|
r.CORE = dict(LDAPURI=ldap_url, LDAPBASE="dc=example,dc=test", BINDDN="uid=admin,ou=people,dc=example,dc=test")
|
|
from privacyidea.app import create_app
|
|
from privacyidea.models import db
|
|
from privacyidea.lib.auth import create_db_admin
|
|
from privacyidea.lib.policy import PolicyClass,set_policy
|
|
from privacyidea.lib.resolver import get_resolver_config,save_resolver
|
|
from privacyidea.lib.realm import set_realm,set_default_realm
|
|
from werkzeug.serving import make_server
|
|
with tempfile.TemporaryDirectory(prefix='realm-native-') as directory:
|
|
root=Path(directory);(root/'enckey').write_bytes(os.urandom(96))
|
|
cfg=root/'fixture.cfg'
|
|
config=dict(SQLALCHEMY_DATABASE_URI='sqlite:///'+str(root/'db.sqlite'),SECRET_KEY='fixture-secret-key',
|
|
PI_PEPPER='fixture-pepper',PI_ENCFILE=str(root/'enckey'),PI_NO_RESPONSE_SIGN=True,
|
|
PI_AUDIT_NO_SIGN=True,PI_LOGFILE=str(root/'log'),PI_TRUSTED_JWT=[],
|
|
PI_AUDIT_SQL_URI='sqlite:///'+str(root/'audit.sqlite'))
|
|
cfg.write_text('\n'.join(k+'='+repr(v) for k,v in config.items()))
|
|
app=create_app(config_name='testing',config_file=str(cfg),silent=True)
|
|
with app.app_context():
|
|
assert str(db.engine.url)==config['SQLALCHEMY_DATABASE_URI'];db.create_all()
|
|
from privacyidea.lib.audit import getAudit
|
|
from privacyidea.lib.auditmodules.sqlaudit import LogEntry
|
|
LogEntry.__table__.create(getAudit(app.config).engine,checkfirst=True)
|
|
create_db_admin('pi-admin',password='fixture-admin-password')
|
|
policies=PolicyClass().policies
|
|
server=make_server('127.0.0.1',0,app);thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
|
|
try:
|
|
client=r.Client(f'http://127.0.0.1:{server.server_port}','fixture-admin-password')
|
|
r.repair(client,lambda:'synthetic-invalid-predecessor',apply=True)
|
|
with app.app_context():
|
|
save_resolver({'resolver':r.RESOLVER,'type':'ldapresolver','TIMEOUT':'7','SIZELIMIT':'777','CACHE_TIMEOUT':'0'})
|
|
set_realm('other',[]);set_default_realm('other')
|
|
set_policy(name='mfa-passthru-phase1',scope='authentication',action='passthru',realm='coulomb',active=False)
|
|
policies=PolicyClass().policies
|
|
target=dict(helper='update',pi_url=client.url,lldap_url=lldap_url,user='admin')
|
|
before=client.snapshot()
|
|
try:cut.native_bind(client,before)
|
|
except r.Refused:pass
|
|
else:raise AssertionError('old bind unexpectedly works')
|
|
preview=cut.run(client,target,'preview')
|
|
assert client.snapshot()==before
|
|
report=cut.run(client,target,'apply',replacement=replacement,expected_digest=preview['snapshot_sha256'])
|
|
assert report['result']=='configured' and report['configuration_writes']==1
|
|
assert cut.public_snapshot(client.snapshot())==cut.public_snapshot(before)
|
|
with app.app_context():
|
|
assert get_resolver_config(r.RESOLVER)['BINDPW']==replacement
|
|
assert PolicyClass().policies==policies
|
|
token=init_token({'type':'hotp','genkey':1,'pin':''},user=User('admin',r.REALM,r.RESOLVER))
|
|
otp=token.get_otp()[2]
|
|
target['helper']='reconcile'
|
|
# Fixture has actual PI/LDAP/MFA; cluster and KeyCape gates are explicitly substituted.
|
|
report=cut.run(client,target,'check',replacement=replacement,predecessor='synthetic-invalid-predecessor',
|
|
gate=lambda:None,mfa_proof=lambda:cut.verify(client.url,'admin',r.REALM,r.RESOLVER,
|
|
'fixture-admin-password',lambda:otp))
|
|
assert report['result']=='verified' and report['mfa_verified'] and report['configuration_writes']==0
|
|
return {'native_candidate_and_stored_ldap_bind':True,'native_old_bind_refused':True,
|
|
'native_minimal_password_update':True,'native_custom_config_and_realms_preserved':True,
|
|
'native_policies_preserved':True,'native_exact_user_and_hotp':True,
|
|
'synthetic_wrong_predecessor_http_denied':True,'cluster_and_keycape_gates_substituted':True,
|
|
'production_acceptance':False}
|
|
finally:server.shutdown();thread.join();server.server_close()
|
|
|
|
|
|
|
|
IMAGE = 'sha256:2a8454b668c1aba7157e832eab0e242e1e7eb5fb7591d7e7774ba05286511ca8'
|
|
|
|
|
|
def run():
|
|
import re
|
|
import secrets
|
|
import subprocess
|
|
import time
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'sso-mfa/k8s/privacyidea'))
|
|
from resolver_cutover import login
|
|
def command(args):
|
|
result=subprocess.run(['docker',*args],capture_output=True,text=True,timeout=60)
|
|
if result.returncode:raise RuntimeError('fixture command failed')
|
|
return result.stdout.strip()
|
|
contexts=json.loads(command(['context','inspect']))
|
|
# DOCKER_CONTEXT takes precedence over DOCKER_HOST. Reject any remote
|
|
# configured endpoint, including conflicting environment selections.
|
|
hosts=[contexts[0]['Endpoints']['docker']['Host']]
|
|
if os.environ.get('DOCKER_HOST'):hosts.append(os.environ['DOCKER_HOST'])
|
|
assert all(host.startswith('unix://') for host in hosts)
|
|
assert command(['image','inspect',IMAGE,'--format','{{.Id}}'])==IMAGE
|
|
container=None
|
|
try:
|
|
with tempfile.TemporaryDirectory(prefix='resolver-lldap-native-') as directory:
|
|
root=Path(directory);password=secrets.token_hex(24)
|
|
source=root/'container.env'
|
|
fd=os.open(source,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600)
|
|
with os.fdopen(fd,'w') as stream:
|
|
stream.write('LLDAP_LDAP_USER_PASS='+password+'\nLLDAP_JWT_SECRET='+secrets.token_hex(32)+'\nLLDAP_LDAP_BASE_DN=dc=example,dc=test\n')
|
|
container=command(['run','--detach','--rm','--pull=never','--memory=256m','--cpus=0.5','--pids-limit=128',
|
|
'--publish','127.0.0.1::17170','--publish','127.0.0.1::3890',
|
|
'--tmpfs','/data:rw,nosuid,nodev,size=32m,uid=1000,gid=1000','--env-file',str(source),IMAGE])
|
|
assert re.fullmatch(r'[0-9a-f]{64}',container)
|
|
ports=json.loads(command(['inspect',container,'--format','{{json .NetworkSettings.Ports}}']))
|
|
assert all(ports[p][0]['HostIp']=='127.0.0.1' for p in ('17170/tcp','3890/tcp'))
|
|
base='http://127.0.0.1:'+ports['17170/tcp'][0]['HostPort']
|
|
ldap='ldap://127.0.0.1:'+ports['3890/tcp'][0]['HostPort']
|
|
for _ in range(80):
|
|
try:login(base,password);break
|
|
except Exception:time.sleep(.25)
|
|
else:raise RuntimeError('fixture startup failed')
|
|
result=provider_run(base,ldap,password)
|
|
finally:
|
|
if container and re.fullmatch(r'[0-9a-f]{64}',container):command(['rm','--force',container])
|
|
result['fixture_removed']=True
|
|
return result
|
|
|
|
|
|
if __name__=='__main__':
|
|
try:
|
|
with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()):result=run()
|
|
print(json.dumps(result,sort_keys=True))
|
|
except Exception as exc:
|
|
import traceback
|
|
frame=traceback.extract_tb(exc.__traceback__)[-1]
|
|
print(json.dumps({'result':'failed','type':type(exc).__name__,'file':Path(frame.filename).name,'line':frame.lineno}))
|
|
raise SystemExit(1)
|