Verify reader scope before accepting absence of enrolled factors

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c

scripts/provider-onboarding-contract.py

@@ -0,0 +1,94 @@
+"""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")
+  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()
+   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)
+  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)