net-kingdom/tests/mfa_state_native_fixture.py

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