"""Disposable real-provider acceptance, invoked in a separate Python process.""" import contextlib import importlib.util import io import json import logging import os from pathlib import Path import sys import tempfile import threading import time def run(): for name in list(os.environ): if name.startswith('PRIVACYIDEA_'): del os.environ[name] logging.disable(logging.CRITICAL) root_repo = Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location('mfa_state', root_repo/'sso-mfa/k8s/privacyidea/mfa_state.py') m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m) 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 from privacyidea.lib.token import init_token from privacyidea.lib.user import User from werkzeug.serving import make_server result = {} with tempfile.TemporaryDirectory(prefix='mfa-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='disposable-native-fixture', PI_PEPPER='disposable-native-pepper', PI_NO_RESPONSE_SIGN=True, PI_AUDIT_NO_SIGN=True, PI_LOGFILE=str(root/'log'), PI_ENCFILE=str(root/'enckey'), 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) passwd = root/'users' password = crypt_ctx.hash('fixture-user-password', scheme='sha512_crypt') passwd.write_text('alice:'+password+':1001:1001:Fixture:/tmp:/bin/false\nbob:'+password+':1002:1002:Fixture:/tmp:/bin/false\n') save_resolver({'resolver':'fixture-users','type':'passwdresolver','fileName':str(passwd)}) set_realm('coulomb',[{'name':'fixture-users'}]) create_db_admin('fixture-reader', password='fixture-reader-password') set_policy(name='fixture-baseline',scope='admin',action='*',adminuser=['*','!fixture-reader']) set_policy(name='fixture-reader',scope='admin',action='tokenlist',adminuser='fixture-reader',realm='coulomb') probe = init_token({'serial':m.PROBE,'type':'hotp','genkey':1,'realm':'coulomb'}) probe.token.active = False; probe.token.save() for i in range(18): init_token({'serial':'FIXTURE'+str(i),'type':'hotp','genkey':1},user=User('alice','coulomb')) pending = init_token({'serial':'FIXTUREPENDING','type':'hotp','genkey':1},user=User('alice','coulomb')) pending.token.rollout_state = 'verify'; pending.token.save() inactive = init_token({'serial':'FIXTUREINACTIVE','type':'hotp','genkey':1},user=User('alice','coulomb')) inactive.token.active = False; inactive.token.save() native = app.test_client() def login(user, password, realm=None): args={'username':user,'password':password} if realm: args['realm']=realm response=native.post('/auth',json=args) assert response.status_code==200 and response.get_json()['result']['status'] return response.get_json()['result']['value']['token'] token=login('fixture-reader','fixture-reader-password') tokenfile=root/'token';tokenfile.write_text(token);tokenfile.chmod(0o600) assert m.credential(tokenfile)==token server=make_server('127.0.0.1',0,app); thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start() try: client=m.Client(f'http://127.0.0.1:{server.server_port}',token) counts=m.diagnose(client,'alice')['factors'] assert counts=={'total':20,'active_confirmed':18,'active_pending':1,'inactive':1},counts result['native_counts_beyond_default_page']=True assert m.diagnose(client,'bob')['factors']['total']==0 result['native_empty_with_visibility']=True def refused(fn): try: fn() except m.Refused: return True return False with app.app_context(): enable_policy('fixture-reader',False) raw=native.get('/token/?user=alice&realm=coulomb',headers={'Authorization':token}) assert raw.status_code==200 and raw.get_json()['result']['value']['count']==0 assert refused(lambda:m.diagnose(client,'alice')) result['native_withdrawal_empty_success_refused']=True with app.app_context(): enable_policy('fixture-reader',True) assert m.diagnose(client,'alice')['factors']['total']==20 result['native_permission_recovery']=True usertoken=login('alice','fixture-user-password','coulomb') assert refused(lambda:m.diagnose(m.Client(client.base,usertoken),'bob')) result['native_self_service_cannot_prove_other_user_visibility']=True assert refused(lambda:m.diagnose(m.Client(client.base,token[:-5]+'XXXXX'),'alice')) result['native_tampered_signature_refused']=True with app.app_context(): set_policy(name='fixture-short-session',scope='webui',action='jwt_validity=2') short=login('fixture-reader','fixture-reader-password') import base64 claims=json.loads(base64.urlsafe_b64decode(short.split('.')[1]+'===')) remaining=claims['exp']-time.time();assert 0 < remaining < 3 time.sleep(remaining+0.2) assert refused(lambda:m.diagnose(m.Client(client.base,short),'alice')) result['native_expired_provider_jwt_refused']=True finally: server.shutdown();thread.join();server.server_close() 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: # Fixture-only, sanitized: no database/HTTP body/credential traceback. 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, 'counts':exc.args[0] if exc.args and isinstance(exc.args[0],dict) and set(exc.args[0])=={'total','active_confirmed','active_pending','inactive'} else None})) raise SystemExit(1)