tegwick
aa709fb854
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c
170 lines
12 KiB
Python
170 lines
12 KiB
Python
"""Isolated provider contract test. Never loads production config or credentials."""
|
|
import contextlib,io,json,logging,os,tempfile,time,base64,hmac,hashlib,struct,urllib.parse
|
|
from pathlib import Path
|
|
result={"success":False};phase="isolation"
|
|
def check(name,condition):
|
|
result[name]=bool(condition)
|
|
if not condition:raise ValueError(name)
|
|
def run():
|
|
global phase
|
|
for k in list(os.environ):
|
|
if k.startswith("PRIVACYIDEA_"):del os.environ[k]
|
|
logging.disable(logging.CRITICAL)
|
|
with tempfile.TemporaryDirectory(prefix="provider-contract-") as d:
|
|
root=Path(d);(root/'enckey').write_bytes(os.urandom(96))
|
|
cfg=root/'fixture.cfg';cfg.write_text("SQLALCHEMY_DATABASE_URI='sqlite:///:memory:'\nSECRET_KEY='isolated-fixture-only'\nPI_PEPPER='isolated-fixture-only'\nPI_NO_RESPONSE_SIGN=True\nPI_AUDIT_NO_SIGN=True\nPI_LOGFILE="+repr(str(root/'log'))+"\nPI_ENCFILE="+repr(str(root/'enckey'))+"\nPI_TRUSTED_JWT=[]\n")
|
|
with cfg.open('a') as f:f.write('PI_AUDIT_SQL_URI='+repr('sqlite:///'+str(root/'audit.sqlite'))+'\n')
|
|
from privacyidea.app import create_app
|
|
from privacyidea.models import db
|
|
from privacyidea.lib.policy import set_policy,enable_policy
|
|
from privacyidea.lib.resolver import save_resolver
|
|
from privacyidea.lib.realm import set_realm
|
|
from privacyidea.lib.auth import create_db_admin
|
|
from privacyidea.lib.resolvers.PasswdIdResolver import crypt_ctx
|
|
app=create_app(config_name="testing",config_file=str(cfg),silent=True)
|
|
phase="fixture_database"
|
|
with app.app_context():
|
|
check('database_isolated',str(db.engine.url)=='sqlite:///:memory:');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)
|
|
passwd=root/'users';passwd.write_text('alice:'+crypt_ctx.hash('fixture-password',scheme='sha512_crypt')+':1001:1001:Fixture:/tmp:/bin/false\n')
|
|
save_resolver({'resolver':'fixture-users','type':'passwdresolver','fileName':str(passwd)})
|
|
set_realm('fixture',[{'name':'fixture-users'}])
|
|
create_db_admin('fixture-reader',password='fixture-service-password')
|
|
set_policy(name='fixture-admin-baseline',scope='admin',action='*',adminuser=['*','!fixture-reader'])
|
|
set_policy(name='fixture-reader',scope='admin',action='tokenlist',adminuser='fixture-reader',realm='fixture')
|
|
set_policy(name='fixture-user',scope='user',action='enrollTOTP,delete,disable',realm='fixture')
|
|
set_policy(name='fixture-confirmation',scope='enrollment',action='verify_enrollment=totp',realm='fixture')
|
|
set_policy(name='fixture-passthru',scope='authentication',action='passthru',realm='fixture')
|
|
client=app.test_client()
|
|
def req(method,path,data=None,token=None):
|
|
headers={'Authorization':token} if token else {}
|
|
r=client.open(path,method=method,data=data,headers=headers)
|
|
return r.status_code,r.get_json()
|
|
def login(user,password,realm=None):
|
|
data={'username':user,'password':password}
|
|
if realm:data['realm']=realm
|
|
code,body=req('POST','/auth',data)
|
|
check('login_'+user,code==200 and body['result']['status'])
|
|
return body['result']['value']['token']
|
|
phase='password_login';user=login('alice','fixture-password','fixture');reader=login('fixture-reader','fixture-service-password')
|
|
phase='password_passthru'
|
|
code,body=req('POST','/validate/check',{'user':'alice','realm':'fixture','pass':'fixture-password'},reader)
|
|
check('password_success_is_not_otp_evidence',code==200 and body['result']['value'] is True and not body.get('detail',{}).get('serial'))
|
|
phase='pending_enrollment'
|
|
def enroll():
|
|
code,body=req('POST','/token/init',{'type':'totp','genkey':'1'},user)
|
|
check('possession_confirmation_required',code==200 and body['result']['status'] and body['detail'].get('rollout_state')=='verify')
|
|
return body['detail']
|
|
detail=enroll();serial=detail['serial']
|
|
code,body=req('GET','/token/?user=alice&realm=fixture&active=True',token=reader)
|
|
tokens=body['result']['value']['tokens']
|
|
check('pending_token_still_active',code==200 and len(tokens)==1 and tokens[0]['active'] is True and tokens[0]['rollout_state']=='verify')
|
|
phase='cancel_enrollment';code,body=req('DELETE','/token/'+serial,token=user)
|
|
check('pending_enrollment_cancelled',code==200 and body['result']['value']==1)
|
|
phase='confirm_enrollment';detail=enroll();serial=detail['serial']
|
|
uri=detail['googleurl']['value'];seed=urllib.parse.parse_qs(urllib.parse.urlsplit(uri).query)['secret'][0]
|
|
key=base64.b32decode(seed+'='*((-len(seed))%8))
|
|
def otp():
|
|
digest=hmac.new(key,struct.pack('>Q',int(time.time())//30),hashlib.sha1).digest();offset=digest[-1]&15
|
|
return str((struct.unpack('>I',digest[offset:offset+4])[0]&0x7fffffff)%1000000).zfill(6)
|
|
code,body=req('POST','/token/init',{'serial':serial,'type':'totp','verify':otp()},user)
|
|
result['confirmation_response']={'http':code,'status':body.get('result',{}).get('status'),'value_type':type(body.get('result',{}).get('value')).__name__,'error_code':body.get('result',{}).get('error',{}).get('code')}
|
|
check('possession_confirmed',code==200 and body['result']['status'] and body['result']['value'] is True)
|
|
code,body=req('GET','/token/?user=alice&realm=fixture&active=True',token=reader)
|
|
check('confirmed_token_enrolled',code==200 and body['result']['value']['tokens'][0]['rollout_state']=='enrolled')
|
|
phase='provider_scope_withdrawal'
|
|
with app.app_context():enable_policy('fixture-reader',False)
|
|
code,body=req('GET','/token/?tokenrealm=fixture',token=reader)
|
|
check('revoked_listing_is_empty_success',code==200 and body['result']['value']['count']==0)
|
|
with app.app_context():enable_policy('fixture-reader',True)
|
|
code,body=req('GET','/token/?tokenrealm=fixture',token=reader)
|
|
check('provider_permission_recovery',code==200 and body['result']['value']['count']==1)
|
|
phase='expiry'
|
|
with app.app_context():set_policy(name='fixture-short-session',scope='webui',action='jwt_validity=2',user='alice',realm='fixture')
|
|
short=login('alice','fixture-password','fixture')
|
|
claims=json.loads(base64.urlsafe_b64decode(short.split('.')[1]+'==='))
|
|
check('provider_issued_short_expiry',0<claims['exp']-time.time()<4)
|
|
time.sleep(3)
|
|
code,_=req('GET','/token/',token=short)
|
|
check('expired_provider_jwt_rejected',code==401)
|
|
fresh=login('alice','fixture-password','fixture')
|
|
code,_=req('GET','/token/',token=fresh)
|
|
check('new_provider_session_recovers_after_expiry',code==200)
|
|
code,_=req('GET','/token/',token=short)
|
|
check('expired_predecessor_stays_rejected',code==401)
|
|
phase='platform_admin_recovery'
|
|
from factor_recovery import ProviderStore,recover,RecoveryError
|
|
with app.app_context():
|
|
store=ProviderStore(app,realm='fixture')
|
|
request=dict(user='alice',serial=serial,realm='fixture',actor='fixture-platform-operator',reference='fixture-recovery-001')
|
|
preview=recover(store,request)
|
|
check('recovery_preview_preserves_factor',preview['active'] is True and preview['changes_applied'] is False)
|
|
approved=dict(request,apply=True,identity_verified=True,expected_version=preview['version'])
|
|
applied=recover(store,approved)
|
|
check('platform_recovery_disabled_factor',applied['status']=='recovered' and store.snapshot('alice',serial)['active'] is False)
|
|
check('platform_recovery_audit_durable',store.receipt(request['reference'])['complete'] is True)
|
|
check('platform_recovery_replay_no_mutation',recover(store,approved)['replayed'] is True)
|
|
phase='authenticated_recovery_service'
|
|
from recovery_service import RecoveryService,verify_token
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from werkzeug.test import Client
|
|
from werkzeug.wrappers import Response
|
|
import jwt
|
|
signing=rsa.generate_private_key(public_exponent=65537,key_size=2048)
|
|
now=int(time.time())
|
|
operator_claims=dict(iss='https://fixture.invalid',aud='fixture-portal',sub='fixture-operator',
|
|
iat=now,exp=now+300,principal_type='human',roles=['platform-operator'],assurance=dict(level='aal2',mfa=True,at=now))
|
|
def signed(**changes):return jwt.encode(dict(operator_claims,**changes),signing,algorithm='RS256')
|
|
with app.app_context():
|
|
service=RecoveryService(store,lambda token:verify_token(token,signing.public_key(),'https://fixture.invalid','fixture-portal'),b'fixture-ticket-key')
|
|
service_client=Client(service,Response)
|
|
from privacyidea.lib.token import enable_token
|
|
enable_token(serial,enable=True)
|
|
def operation(body,token=None):
|
|
r=service_client.post('/recover',json=body,headers={'Authorization':'Bearer '+(token or signed())})
|
|
return r.status_code,r.json
|
|
preview_body=dict(action='preview',user='alice',reference='fixture-browser-recovery')
|
|
for name,token in [('tenant_admin',signed(roles=['tenant-admin'])),('wrong_audience',signed(aud='other')),('stale_mfa',signed(assurance=dict(level='aal2',mfa=True,at=now-301))),('unsigned','invalid')]:
|
|
status,_=operation(preview_body,token)
|
|
check('service_denies_'+name,status==403 and store.snapshot('alice',serial)['active'])
|
|
status,_=operation(dict(preview_body,user='missing-fixture-user'))
|
|
check('service_unknown_user_not_realm_wide',status==409 and store.snapshot('alice',serial)['active'])
|
|
from privacyidea.models import TokenOwner
|
|
from privacyidea.lib.token import get_tokens
|
|
factor_id=get_tokens(serial=serial)[0].token.id
|
|
other_owner=TokenOwner(token_id=factor_id,user_id='second-fixture-owner',resolver='fixture-users',realmname='fixture')
|
|
other_owner.save()
|
|
status,shared=operation(preview_body)
|
|
check('shared_identity_factor_denied',status==409 and shared['failure']=='shared_factor_not_supported')
|
|
db.session.delete(other_owner);db.session.commit()
|
|
status,preview=operation(preview_body)
|
|
check('service_preview_owned_factor',status==200 and len(preview['factors'])==1)
|
|
confirmation=preview['factors'][0]['confirmation']
|
|
apply_body=dict(action='apply',confirmation=confirmation,identity_verified=True)
|
|
status,_=operation(dict(apply_body,confirmation=confirmation+'x'))
|
|
check('service_rejects_changed_confirmation',status==409 and store.snapshot('alice',serial)['active'])
|
|
status,receipt=operation(apply_body)
|
|
check('service_audited_recovery',status==200 and receipt['changes_applied'] and store.receipt('fixture-browser-recovery')['complete'])
|
|
status,receipt=operation(apply_body)
|
|
check('service_safe_retry',status==200 and receipt['replayed'])
|
|
status,receipt=operation(dict(action='status',reference='fixture-browser-recovery'))
|
|
check('service_support_reference_readback',status==200 and receipt['status']=='recovered')
|
|
phase='replacement_possession'
|
|
detail=enroll();serial=detail['serial']
|
|
seed=urllib.parse.parse_qs(urllib.parse.urlsplit(detail['googleurl']['value']).query)['secret'][0]
|
|
key=base64.b32decode(seed+'='*((-len(seed))%8))
|
|
code,body=req('POST','/token/init',{'serial':serial,'type':'totp','verify':otp()},user)
|
|
check('replacement_possession_confirmed',code==200 and body['result']['value'] is True)
|
|
digest=hmac.new(key,struct.pack('>Q',int(time.time())//30+1),hashlib.sha1).digest();offset=digest[-1]&15
|
|
next_code=str((struct.unpack('>I',digest[offset:offset+4])[0]&0x7fffffff)%1000000).zfill(6)
|
|
code,body=req('POST','/validate/check',{'user':'alice','realm':'fixture','pass':next_code},reader)
|
|
check('replacement_authentication_succeeds',code==200 and body['result']['value'] is True and body['detail']['serial']==serial)
|
|
phase='finished';result['success']=True
|
|
with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()):
|
|
try:run()
|
|
except Exception as e:
|
|
import traceback
|
|
result.update(phase=phase,failure_type=type(e).__name__,frames=[{"function":f.name,"line":f.lineno} for f in traceback.extract_tb(e.__traceback__)[-3:]])
|
|
print(json.dumps(result));raise SystemExit(0 if result['success'] else 1)
|