-
Notifications
You must be signed in to change notification settings - Fork 28
Expand file tree
/
Copy pathtest_client_55_token_exchange.py
More file actions
92 lines (74 loc) · 3.23 KB
/
Copy pathtest_client_55_token_exchange.py
File metadata and controls
92 lines (74 loc) · 3.23 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
import os
from cryptojwt.key_jar import init_key_jar
import pytest
from idpyoidc.client.entity import Entity
from idpyoidc.message import Message
from idpyoidc.message.oauth2 import AccessTokenResponse
from idpyoidc.message.oauth2 import AuthorizationResponse
from idpyoidc.message.oidc import IdToken
from tests.test_client_21_oidc_service import make_keyjar
KEYSPEC = [
{"type": "RSA", "use": ["sig"]},
{"type": "EC", "crv": "P-256", "use": ["sig"]},
]
_dirname = os.path.dirname(os.path.abspath(__file__))
ISS = "https://example.com"
ISS_KEY = init_key_jar(
public_path="{}/pub_iss.jwks".format(_dirname),
private_path="{}/priv_iss.jwks".format(_dirname),
key_defs=KEYSPEC,
issuer_id=ISS,
read_only=False,
)
ISS_KEY.import_jwks_as_json(open("{}/pub_client.jwks".format(_dirname)).read(), "client_id")
def create_jws(val):
lifetime = 3600
idts = IdToken(**val)
return idts.to_jwt(
key=ISS_KEY.get_signing_key("ec", issuer_id=ISS), algorithm="ES256", lifetime=lifetime
)
class TestUserInfo(object):
@pytest.fixture(autouse=True)
def create_request(self):
self._iss = ISS
client_config = {
"client_id": "client_id",
"client_secret": "a longesh password",
"redirect_uris": ["https://example.com/cli/authz_cb"],
"issuer": self._iss,
"requests_dir": "requests",
"base_url": "https://example.com/cli/",
}
entity = Entity(keyjar=make_keyjar(), config=client_config,
services={
"discovery": {
"class":
"idpyoidc.client.oauth2.server_metadata.ServerMetadata"},
"authorization": {
"class": "idpyoidc.client.oauth2.authorization.Authorization"},
"access_token": {
"class": "idpyoidc.client.oauth2.access_token.AccessToken"},
"token_exchange": {
"class":
"idpyoidc.client.oauth2.token_exchange.TokenExchange"
},
}
)
entity.client_get("service_context").issuer = "https://example.com"
self.service = entity.client_get("service", "token_exchange")
_state_interface = self.service.client_get("service_context").state
# Add history
auth_response = AuthorizationResponse(code="access_code").to_json()
_state_interface.store_item(auth_response, "auth_response", "abcde")
idtval = {"nonce": "KUEYfRM2VzKDaaKD", "sub": "diana", "iss": ISS, "aud": "client_id"}
idt = create_jws(idtval)
ver_idt = IdToken().from_jwt(idt, make_keyjar())
token_response = AccessTokenResponse(
access_token="access_token", id_token=idt, __verified_id_token=ver_idt
).to_json()
_state_interface.store_item(token_response, "token_response", "abcde")
def test_construct(self):
_req = self.service.construct(state="abcde")
assert isinstance(_req, Message)
assert len(_req) == 2
assert "subject_token" in _req