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