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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
|
import pytest
from flask import json
from authlib.jose import jwt
from authlib.oauth2.rfc7591 import (
ClientRegistrationEndpoint as _ClientRegistrationEndpoint,
)
from tests.util import read_file_path
from .models import Client
from .models import db
class ClientRegistrationEndpoint(_ClientRegistrationEndpoint):
software_statement_alg_values_supported = ["RS256"]
def authenticate_token(self, request):
auth_header = request.headers.get("Authorization")
if auth_header:
request.user_id = 1
return auth_header
def resolve_public_key(self, request):
return read_file_path("rsa_public.pem")
def save_client(self, client_info, client_metadata, request):
client = Client(user_id=request.user_id, **client_info)
client.set_client_metadata(client_metadata)
db.session.add(client)
db.session.commit()
return client
@pytest.fixture
def metadata():
return {}
@pytest.fixture(autouse=True)
def server(server, app, metadata):
class MyClientRegistration(ClientRegistrationEndpoint):
def get_server_metadata(test_client):
return metadata
server.register_endpoint(MyClientRegistration)
@app.route("/create_client", methods=["POST"])
def create_client():
return server.create_endpoint_response("client_registration")
return server
def test_access_denied(test_client):
rv = test_client.post("/create_client", json={})
resp = json.loads(rv.data)
assert resp["error"] == "access_denied"
def test_invalid_request(test_client):
headers = {"Authorization": "bearer abc"}
rv = test_client.post("/create_client", json={}, headers=headers)
resp = json.loads(rv.data)
assert resp["error"] == "invalid_request"
def test_create_client(test_client):
headers = {"Authorization": "bearer abc"}
body = {"client_name": "Authlib"}
rv = test_client.post("/create_client", json=body, headers=headers)
resp = json.loads(rv.data)
assert "client_id" in resp
assert resp["client_name"] == "Authlib"
def test_software_statement(test_client):
payload = {"software_id": "uuid-123", "client_name": "Authlib"}
s = jwt.encode({"alg": "RS256"}, payload, read_file_path("rsa_private.pem"))
body = {
"software_statement": s.decode("utf-8"),
}
headers = {"Authorization": "bearer abc"}
rv = test_client.post("/create_client", json=body, headers=headers)
resp = json.loads(rv.data)
assert "client_id" in resp
assert resp["client_name"] == "Authlib"
def test_no_public_key(test_client, server):
class ClientRegistrationEndpoint2(ClientRegistrationEndpoint):
def get_server_metadata(test_client):
return None
def resolve_public_key(self, request):
return None
payload = {"software_id": "uuid-123", "client_name": "Authlib"}
s = jwt.encode({"alg": "RS256"}, payload, read_file_path("rsa_private.pem"))
body = {
"software_statement": s.decode("utf-8"),
}
server._endpoints[ClientRegistrationEndpoint.ENDPOINT_NAME] = [
ClientRegistrationEndpoint2(server)
]
headers = {"Authorization": "bearer abc"}
rv = test_client.post("/create_client", json=body, headers=headers)
resp = json.loads(rv.data)
assert resp["error"] in "unapproved_software_statement"
def test_scopes_supported(test_client, metadata):
metadata["scopes_supported"] = ["profile", "email"]
headers = {"Authorization": "bearer abc"}
body = {"scope": "profile email", "client_name": "Authlib"}
rv = test_client.post("/create_client", json=body, headers=headers)
resp = json.loads(rv.data)
assert "client_id" in resp
assert resp["client_name"] == "Authlib"
body = {"scope": "profile email address", "client_name": "Authlib"}
rv = test_client.post("/create_client", json=body, headers=headers)
resp = json.loads(rv.data)
assert resp["error"] in "invalid_client_metadata"
def test_response_types_supported(test_client, metadata):
metadata["response_types_supported"] = ["code", "code id_token"]
headers = {"Authorization": "bearer abc"}
body = {"response_types": ["code"], "client_name": "Authlib"}
rv = test_client.post("/create_client", json=body, headers=headers)
resp = json.loads(rv.data)
assert "client_id" in resp
assert resp["client_name"] == "Authlib"
# The items order should not matter
# Extension response types MAY contain a space-delimited (%x20) list of
# values, where the order of values does not matter (e.g., response
# type "a b" is the same as "b a").
headers = {"Authorization": "bearer abc"}
body = {"response_types": ["id_token code"], "client_name": "Authlib"}
rv = test_client.post("/create_client", json=body, headers=headers)
resp = json.loads(rv.data)
assert "client_id" in resp
assert resp["client_name"] == "Authlib"
# https://www.rfc-editor.org/rfc/rfc7591.html#section-2
# If omitted, the default is that the client will use only the "code"
# response type.
body = {"client_name": "Authlib"}
rv = test_client.post("/create_client", json=body, headers=headers)
resp = json.loads(rv.data)
assert "client_id" in resp
assert resp["client_name"] == "Authlib"
body = {"response_types": ["code", "token"], "client_name": "Authlib"}
rv = test_client.post("/create_client", json=body, headers=headers)
resp = json.loads(rv.data)
assert resp["error"] in "invalid_client_metadata"
def test_grant_types_supported(test_client, metadata):
metadata["grant_types_supported"] = ["authorization_code", "password"]
headers = {"Authorization": "bearer abc"}
body = {"grant_types": ["password"], "client_name": "Authlib"}
rv = test_client.post("/create_client", json=body, headers=headers)
resp = json.loads(rv.data)
assert "client_id" in resp
assert resp["client_name"] == "Authlib"
# https://www.rfc-editor.org/rfc/rfc7591.html#section-2
# If omitted, the default behavior is that the client will use only
# the "authorization_code" Grant Type.
body = {"client_name": "Authlib"}
rv = test_client.post("/create_client", json=body, headers=headers)
resp = json.loads(rv.data)
assert "client_id" in resp
assert resp["client_name"] == "Authlib"
body = {"grant_types": ["client_credentials"], "client_name": "Authlib"}
rv = test_client.post("/create_client", json=body, headers=headers)
resp = json.loads(rv.data)
assert resp["error"] in "invalid_client_metadata"
def test_token_endpoint_auth_methods_supported(test_client, metadata):
metadata["token_endpoint_auth_methods_supported"] = ["client_secret_basic"]
headers = {"Authorization": "bearer abc"}
body = {
"token_endpoint_auth_method": "client_secret_basic",
"client_name": "Authlib",
}
rv = test_client.post("/create_client", json=body, headers=headers)
resp = json.loads(rv.data)
assert "client_id" in resp
assert resp["client_name"] == "Authlib"
body = {"token_endpoint_auth_method": "none", "client_name": "Authlib"}
rv = test_client.post("/create_client", json=body, headers=headers)
resp = json.loads(rv.data)
assert resp["error"] in "invalid_client_metadata"
|
