key-cape/scripts/provider-onboarding-contract.py

"""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)