<s> import time <EOL> import pytest <EOL> from oic . extension . token import JWTToken <EOL> from oic import rndstr <EOL> from oic . oauth2 import AuthorizationRequest <EOL> from oic . oic import OpenIDRequest <EOL> from oic . utils . keyio import KeyBundle <EOL> from oic . utils . keyio import KeyJar <EOL> from oic . utils . sdb import SessionDB <EOL> from oic . utils . sdb import AuthnEvent <EOL> from oic . utils . sdb import ExpiredToken <EOL> from oic . utils . sdb import AccessCodeUsed <EOL> __author__ = '<STR_LIT>' <EOL> def _eq ( l1 , l2 ) : <EOL> return set ( l1 ) == set ( l2 ) <EOL> AREQ = AuthorizationRequest ( response_type = "<STR_LIT:code>" , client_id = "<STR_LIT>" , <EOL> redirect_uri = "<STR_LIT>" , <EOL> scope = [ "<STR_LIT>" ] , state = "<STR_LIT>" ) <EOL> AREQN = AuthorizationRequest ( response_type = "<STR_LIT:code>" , client_id = "<STR_LIT>" , <EOL> redirect_uri = "<STR_LIT>" , <EOL> scope = [ "<STR_LIT>" ] , state = "<STR_LIT>" , <EOL> nonce = "<STR_LIT>" ) <EOL> AREQO = AuthorizationRequest ( response_type = "<STR_LIT:code>" , client_id = "<STR_LIT>" , <EOL> redirect_uri = "<STR_LIT>" , <EOL> scope = [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> prompt = "<STR_LIT>" , state = "<STR_LIT>" ) <EOL> OIDR = OpenIDRequest ( response_type = "<STR_LIT:code>" , client_id = "<STR_LIT>" , <EOL> redirect_uri = "<STR_LIT>" , scope = [ "<STR_LIT>" ] , <EOL> state = "<STR_LIT>" ) <EOL> JWKS = { "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:d>" : "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" : <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> "<STR_LIT:e>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:true>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:n>" : <EOL> "<STR_LIT>" , <EOL> "<STR_LIT:p>" : <EOL> "<STR_LIT>" , <EOL> "<STR_LIT:q>" : <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , { <EOL> "<STR_LIT:k>" : <EOL> b"<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } ] } <EOL> SESSION_INFO = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT:code>' ] , <EOL> '<STR_LIT>' : '<STR_LIT:{}>' <EOL> } <EOL> class TestToken ( object ) : <EOL> @ pytest . fixture ( autouse = True ) <EOL> def create_token ( self ) : <EOL> kb = KeyBundle ( JWKS [ "<STR_LIT>" ] ) <EOL> kj = KeyJar ( ) <EOL> kj . issuer_keys [ '<STR_LIT>' ] = [ kb ] <EOL> self . access_token = JWTToken ( <EOL> '<STR_LIT:T>' , keyjar = kj , lt_pattern = { '<STR_LIT:code>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> } , <EOL> iss = '<STR_LIT>' , sign_alg = '<STR_LIT>' ) <EOL> def test_create ( self ) : <EOL> sid = rndstr ( <NUM_LIT:32> ) <EOL> session_info = SESSION_INFO <EOL> _jwt = self . access_token ( sid , sinfo = session_info , kid = '<STR_LIT>' ) <EOL> assert _jwt <EOL> assert len ( _jwt . split ( '<STR_LIT:.>' ) ) == <NUM_LIT:3> <EOL> def test_create_with_aud ( self ) : <EOL> sid = rndstr ( <NUM_LIT:32> ) <EOL> session_info = SESSION_INFO <EOL> _jwt = self . access_token ( sid , sinfo = session_info , kid = '<STR_LIT>' , <EOL> aud = [ '<STR_LIT>' ] ) <EOL> assert _jwt <EOL> class TestToken2 ( object ) : <EOL> @ pytest . fixture ( autouse = True ) <EOL> def create_token ( self ) : <EOL> kb = KeyBundle ( JWKS [ "<STR_LIT>" ] ) <EOL> kj = KeyJar ( ) <EOL> kj . issuer_keys [ '<STR_LIT>' ] = [ kb ] <EOL> self . access_token = JWTToken ( <EOL> '<STR_LIT:T>' , keyjar = kj , iss = '<STR_LIT>' , sign_alg = '<STR_LIT>' ) <EOL> def test_create ( self ) : <EOL> sid = rndstr ( <NUM_LIT:32> ) <EOL> session_info = SESSION_INFO <EOL> _jwt = self . access_token ( sid , sinfo = session_info , kid = '<STR_LIT>' , <EOL> lifetime = <NUM_LIT> ) <EOL> assert _jwt <EOL> assert len ( _jwt . split ( '<STR_LIT:.>' ) ) == <NUM_LIT:3> <EOL> def test_create_with_aud ( self ) : <EOL> sid = rndstr ( <NUM_LIT:32> ) <EOL> session_info = SESSION_INFO <EOL> _jwt = self . access_token ( <EOL> sid , sinfo = session_info , kid = '<STR_LIT>' , <EOL> aud = [ '<STR_LIT>' ] , lifetime = <NUM_LIT> ) <EOL> assert _jwt <EOL> info = self . access_token . get_info ( _jwt ) <EOL> assert info [ '<STR_LIT>' ] - info [ '<STR_LIT>' ] == <NUM_LIT> <EOL> class TestEncToken ( object ) : <EOL> @ pytest . fixture ( autouse = True ) <EOL> def create_token ( self ) : <EOL> kb = KeyBundle ( JWKS [ "<STR_LIT>" ] ) <EOL> kj = KeyJar ( ) <EOL> kj . issuer_keys [ '<STR_LIT>' ] = [ kb ] <EOL> self . access_token = JWTToken ( '<STR_LIT:T>' , keyjar = kj , <EOL> lt_pattern = { '<STR_LIT:code>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> } , <EOL> iss = '<STR_LIT>' , encrypt = True ) <EOL> def test_enc_create ( self ) : <EOL> sid = rndstr ( <NUM_LIT:32> ) <EOL> session_info = SESSION_INFO <EOL> _jwe = self . access_token ( sid , sinfo = session_info , kid = '<STR_LIT>' ) <EOL> assert _jwe <EOL> assert len ( _jwe . split ( '<STR_LIT:.>' ) ) == <NUM_LIT:5> <EOL> def test_parse_enc ( self ) : <EOL> sid = rndstr ( <NUM_LIT:32> ) <EOL> session_info = SESSION_INFO <EOL> _jwe = self . access_token ( sid , sinfo = session_info , kid = '<STR_LIT>' ) <EOL> _info = self . access_token . get_info ( _jwe ) <EOL> assert _info <EOL> class TestSessionDB ( object ) : <EOL> @ pytest . fixture ( autouse = True ) <EOL> def create_sdb ( self ) : <EOL> kb = KeyBundle ( JWKS [ "<STR_LIT>" ] ) <EOL> kj = KeyJar ( ) <EOL> kj . issuer_keys [ '<STR_LIT>' ] = [ kb ] <EOL> self . sdb = SessionDB ( <EOL> "<STR_LIT>" , <EOL> token_factory = JWTToken ( '<STR_LIT:T>' , keyjar = kj , <EOL> lt_pattern = { '<STR_LIT:code>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> } , <EOL> iss = '<STR_LIT>' , <EOL> sign_alg = '<STR_LIT>' ) , <EOL> refresh_token_factory = JWTToken ( <EOL> '<STR_LIT:R>' , keyjar = kj , lt_pattern = { '<STR_LIT>' : <NUM_LIT> * <NUM_LIT> } , <EOL> iss = '<STR_LIT>' ) <EOL> ) <EOL> def test_create_authz_session ( self ) : <EOL> ae = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae , AREQ ) <EOL> self . sdb . do_sub ( sid , "<STR_LIT>" ) <EOL> info = self . sdb [ sid ] <EOL> assert info [ "<STR_LIT>" ] == "<STR_LIT>" <EOL> def test_create_authz_session_without_nonce ( self ) : <EOL> ae = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae , AREQ ) <EOL> info = self . sdb [ sid ] <EOL> assert info [ "<STR_LIT>" ] == "<STR_LIT>" <EOL> def test_create_authz_session_with_nonce ( self ) : <EOL> ae = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae , AREQN ) <EOL> info = self . sdb [ sid ] <EOL> assert info [ "<STR_LIT>" ] == "<STR_LIT>" <EOL> def test_create_authz_session_with_id_token ( self ) : <EOL> ae = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae , AREQN , id_token = "<STR_LIT>" ) <EOL> info = self . sdb [ sid ] <EOL> assert info [ "<STR_LIT>" ] == "<STR_LIT>" <EOL> def test_create_authz_session_with_oidreq ( self ) : <EOL> ae = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae , AREQN , oidreq = OIDR ) <EOL> info = self . sdb [ sid ] <EOL> assert "<STR_LIT>" not in info <EOL> assert "<STR_LIT>" in info <EOL> def test_create_authz_session_with_sector_id ( self ) : <EOL> ae = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae , AREQN , oidreq = OIDR ) <EOL> self . sdb . do_sub ( sid , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> info_1 = self . sdb [ sid ] . copy ( ) <EOL> assert "<STR_LIT>" not in info_1 <EOL> assert "<STR_LIT>" in info_1 <EOL> assert info_1 [ "<STR_LIT>" ] != "<STR_LIT>" <EOL> self . sdb . do_sub ( sid , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> info_2 = self . sdb [ sid ] <EOL> assert info_2 [ "<STR_LIT>" ] != "<STR_LIT>" <EOL> assert info_2 [ "<STR_LIT>" ] != info_1 [ "<STR_LIT>" ] <EOL> def test_upgrade_to_token ( self ) : <EOL> ae1 = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae1 , AREQ ) <EOL> self . sdb [ sid ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> grant = self . sdb [ sid ] [ "<STR_LIT:code>" ] <EOL> _dict = self . sdb . upgrade_to_token ( grant ) <EOL> assert _eq ( list ( _dict . keys ( ) ) , <EOL> [ '<STR_LIT>' , '<STR_LIT:code>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:state>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> with pytest . raises ( AccessCodeUsed ) : <EOL> self . sdb . upgrade_to_token ( grant ) <EOL> self . sdb . upgrade_to_token ( _dict [ "<STR_LIT>" ] ) <EOL> def test_upgrade_to_token_refresh ( self ) : <EOL> ae1 = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae1 , AREQO ) <EOL> self . sdb . do_sub ( sid , ae1 . salt ) <EOL> grant = self . sdb [ sid ] [ "<STR_LIT:code>" ] <EOL> _dict = self . sdb . upgrade_to_token ( grant , issue_refresh = True ) <EOL> assert _eq ( _dict . keys ( ) , <EOL> [ '<STR_LIT>' , '<STR_LIT:code>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT:state>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> with pytest . raises ( AccessCodeUsed ) : <EOL> self . sdb . upgrade_to_token ( grant ) <EOL> self . sdb . upgrade_to_token ( _dict [ "<STR_LIT>" ] ) <EOL> def test_upgrade_to_token_with_id_token_and_oidreq ( self ) : <EOL> ae2 = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae2 , AREQ ) <EOL> self . sdb [ sid ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> grant = self . sdb [ sid ] [ "<STR_LIT:code>" ] <EOL> _dict = self . sdb . upgrade_to_token ( grant , id_token = "<STR_LIT>" , <EOL> oidreq = OIDR ) <EOL> assert _eq ( list ( _dict . keys ( ) ) , <EOL> [ '<STR_LIT>' , '<STR_LIT:code>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT:state>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> assert _dict [ "<STR_LIT>" ] == "<STR_LIT>" <EOL> assert isinstance ( _dict [ "<STR_LIT>" ] , OpenIDRequest ) <EOL> def test_refresh_token ( self ) : <EOL> ae = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae , AREQ ) <EOL> self . sdb [ sid ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> grant = self . sdb [ sid ] [ "<STR_LIT:code>" ] <EOL> dict1 = self . sdb . upgrade_to_token ( grant , issue_refresh = True ) . copy ( ) <EOL> rtoken = dict1 [ "<STR_LIT>" ] <EOL> dict2 = self . sdb . refresh_token ( rtoken , AREQ [ '<STR_LIT>' ] ) <EOL> assert dict1 [ "<STR_LIT>" ] != dict2 [ "<STR_LIT>" ] <EOL> with pytest . raises ( ExpiredToken ) : <EOL> self . sdb . refresh_token ( dict2 [ "<STR_LIT>" ] , AREQ [ '<STR_LIT>' ] ) <EOL> def test_refresh_token_cleared_session ( self ) : <EOL> ae = AuthnEvent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> sid = self . sdb . create_authz_session ( ae , AREQ ) <EOL> self . sdb [ sid ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> grant = self . sdb [ sid ] [ '<STR_LIT:code>' ] <EOL> dict1 = self . sdb . upgrade_to_token ( grant , issue_refresh = True ) <EOL> ac1 = dict1 [ '<STR_LIT>' ] <EOL> rtoken = dict1 [ '<STR_LIT>' ] <EOL> dict2 = self . sdb . refresh_token ( rtoken , AREQ [ '<STR_LIT>' ] ) <EOL> assert ac1 != dict2 [ "<STR_LIT>" ] <EOL> def test_is_valid ( self ) : <EOL> ae1 = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae1 , AREQ ) <EOL> self . sdb [ sid ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> grant = self . sdb [ sid ] [ "<STR_LIT:code>" ] <EOL> assert self . sdb . is_valid ( grant ) <EOL> sinfo = self . sdb . upgrade_to_token ( grant , issue_refresh = True ) <EOL> assert not self . sdb . is_valid ( grant ) <EOL> access_token = sinfo [ "<STR_LIT>" ] <EOL> assert self . sdb . access_token . valid ( access_token ) <EOL> refresh_token = sinfo [ "<STR_LIT>" ] <EOL> sinfo = self . sdb . refresh_token ( refresh_token , AREQ [ '<STR_LIT>' ] ) <EOL> access_token2 = sinfo [ "<STR_LIT>" ] <EOL> assert self . sdb . is_valid ( access_token2 ) <EOL> try : <EOL> self . sdb . is_valid ( access_token ) <EOL> except KeyError : <EOL> pass <EOL> def test_valid_grant ( self ) : <EOL> ae = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae , AREQ ) <EOL> grant = self . sdb [ sid ] [ "<STR_LIT:code>" ] <EOL> assert self . sdb . is_valid ( grant ) <EOL> def test_revoke_token ( self ) : <EOL> ae1 = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae1 , AREQ ) <EOL> self . sdb [ sid ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> grant = self . sdb [ sid ] [ "<STR_LIT:code>" ] <EOL> tokens = self . sdb . upgrade_to_token ( grant , issue_refresh = True ) <EOL> access_token = tokens [ "<STR_LIT>" ] <EOL> refresh_token = tokens [ "<STR_LIT>" ] <EOL> assert self . sdb . is_valid ( access_token ) <EOL> self . sdb . revoke_token ( access_token ) <EOL> assert not self . sdb . is_valid ( access_token ) <EOL> sinfo = self . sdb . refresh_token ( refresh_token , AREQ [ '<STR_LIT>' ] ) <EOL> access_token = sinfo [ "<STR_LIT>" ] <EOL> assert self . sdb . is_valid ( access_token ) <EOL> self . sdb . revoke_refresh_token ( refresh_token ) <EOL> assert not self . sdb . is_valid ( refresh_token ) <EOL> try : <EOL> self . sdb . refresh_token ( refresh_token , AREQ [ '<STR_LIT>' ] ) <EOL> except ExpiredToken : <EOL> pass <EOL> assert self . sdb . is_valid ( access_token ) <EOL> ae2 = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae2 , AREQ ) <EOL> grant = self . sdb [ sid ] [ "<STR_LIT:code>" ] <EOL> self . sdb . revoke_token ( grant ) <EOL> assert not self . sdb . is_valid ( grant ) <EOL> def test_sub_to_authn_event ( self ) : <EOL> ae = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" , time_stamp = time . time ( ) ) <EOL> sid = self . sdb . create_authz_session ( ae , AREQ ) <EOL> sub = self . sdb . do_sub ( sid , "<STR_LIT>" ) <EOL> sids = self . sdb . get_sids_by_sub ( sub ) <EOL> ae = self . sdb [ sids [ <NUM_LIT:0> ] ] [ "<STR_LIT>" ] <EOL> assert ae . valid ( ) <EOL> def test_do_sub_deterministic ( self ) : <EOL> ae = AuthnEvent ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> sid = self . sdb . create_authz_session ( ae , AREQ ) <EOL> self . sdb . do_sub ( sid , "<STR_LIT>" ) <EOL> info = self . sdb [ sid ] <EOL> assert info [ "<STR_LIT>" ] == '<STR_LIT>' <EOL> self . sdb . do_sub ( sid , "<STR_LIT>" , <EOL> sector_id = '<STR_LIT>' , <EOL> subject_type = "<STR_LIT>" ) <EOL> info2 = self . sdb [ sid ] <EOL> assert info2 [ "<STR_LIT>" ] == '<STR_LIT>' <EOL> self . sdb . do_sub ( sid , "<STR_LIT>" , <EOL> sector_id = '<STR_LIT>' , <EOL> subject_type = "<STR_LIT>" ) <EOL> info2 = self . sdb [ sid ] <EOL> assert info2 [ "<STR_LIT>" ] == '<STR_LIT>' </s>
