forked from Github-Mirrors/canaille
oidc implicit flow test
This commit is contained in:
parent
e5bd075f8f
commit
3e453810ec
4 changed files with 64 additions and 12 deletions
|
@ -136,3 +136,10 @@ def user(app, slapd_connection):
|
|||
)
|
||||
u.save(slapd_connection)
|
||||
return u
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def logged_user(user, testclient):
|
||||
with testclient.session_transaction() as sess:
|
||||
sess["user_dn"] = user.dn
|
||||
return user
|
||||
|
|
|
@ -42,10 +42,7 @@ def test_oauth_hybrid(testclient, slapd_connection, user, client):
|
|||
assert {"foo": "bar"} == res.json
|
||||
|
||||
|
||||
def test_oidc_hybrid(testclient, slapd_connection, user, client):
|
||||
with testclient.session_transaction() as sess:
|
||||
sess["user_dn"] = user.dn
|
||||
|
||||
def test_oidc_hybrid(testclient, slapd_connection, logged_user, client):
|
||||
res = testclient.get(
|
||||
"/oauth/authorize",
|
||||
params=dict(
|
||||
|
@ -73,8 +70,8 @@ def test_oidc_hybrid(testclient, slapd_connection, user, client):
|
|||
|
||||
id_token = params["id_token"][0]
|
||||
claims = jwt.decode(id_token, "secret-key")
|
||||
assert user.dn == claims['sub']
|
||||
assert user.sn == claims['name']
|
||||
assert logged_user.dn == claims['sub']
|
||||
assert logged_user.sn == claims['name']
|
||||
|
||||
res = testclient.get("/api/me", headers={"Authorization": f"Bearer {access_token}"})
|
||||
assert 200 == res.status_code
|
||||
|
|
|
@ -1,8 +1,9 @@
|
|||
from authlib.jose import jwt
|
||||
from urllib.parse import urlsplit, parse_qs
|
||||
from web.models import Token
|
||||
|
||||
|
||||
def test_success(testclient, slapd_connection, user, client):
|
||||
def test_oauth_implicit(testclient, slapd_connection, user, client):
|
||||
client.oauthGrantType = ["token"]
|
||||
client.oauthTokenEndpointAuthMethod = "none"
|
||||
|
||||
|
@ -32,8 +33,8 @@ def test_success(testclient, slapd_connection, user, client):
|
|||
|
||||
assert res.location.startswith(client.oauthRedirectURIs[0])
|
||||
params = parse_qs(urlsplit(res.location).fragment)
|
||||
access_token = params["access_token"][0]
|
||||
|
||||
access_token = params["access_token"][0]
|
||||
token = Token.get(access_token, conn=slapd_connection)
|
||||
assert token is not None
|
||||
|
||||
|
@ -44,3 +45,52 @@ def test_success(testclient, slapd_connection, user, client):
|
|||
client.oauthGrantType = ["code"]
|
||||
client.oauthTokenEndpointAuthMethod = "client_secret_basic"
|
||||
client.save(slapd_connection)
|
||||
|
||||
|
||||
def test_oidc_implicit(testclient, slapd_connection, user, client):
|
||||
client.oauthGrantType = ["token id_token"]
|
||||
client.oauthTokenEndpointAuthMethod = "none"
|
||||
|
||||
client.save(slapd_connection)
|
||||
|
||||
res = testclient.get(
|
||||
"/oauth/authorize",
|
||||
params=dict(
|
||||
response_type="id_token token",
|
||||
client_id=client.oauthClientID,
|
||||
scope="openid profile",
|
||||
nonce="somenonce",
|
||||
),
|
||||
)
|
||||
assert (200, "text/html") == (res.status_code, res.content_type)
|
||||
|
||||
res.form["login"] = user.name
|
||||
res.form["password"] = "correct horse battery staple"
|
||||
res = res.form.submit()
|
||||
assert 302 == res.status_code
|
||||
|
||||
res = res.follow()
|
||||
assert (200, "text/html") == (res.status_code, res.content_type), res.json
|
||||
|
||||
res = res.forms["accept"].submit()
|
||||
assert 302 == res.status_code
|
||||
|
||||
assert res.location.startswith(client.oauthRedirectURIs[0])
|
||||
params = parse_qs(urlsplit(res.location).fragment)
|
||||
|
||||
access_token = params["access_token"][0]
|
||||
token = Token.get(access_token, conn=slapd_connection)
|
||||
assert token is not None
|
||||
|
||||
id_token = params["id_token"][0]
|
||||
claims = jwt.decode(id_token, "secret-key")
|
||||
assert user.dn == claims['sub']
|
||||
assert user.sn == claims['name']
|
||||
|
||||
res = testclient.get("/api/me", headers={"Authorization": f"Bearer {access_token}"})
|
||||
assert (200, "application/json") == (res.status_code, res.content_type)
|
||||
assert {"foo": "bar"} == res.json
|
||||
|
||||
client.oauthGrantType = ["code"]
|
||||
client.oauthTokenEndpointAuthMethod = "client_secret_basic"
|
||||
client.save(slapd_connection)
|
||||
|
|
|
@ -103,15 +103,13 @@ class RefreshTokenGrant(_RefreshTokenGrant):
|
|||
|
||||
class OpenIDImplicitGrant(_OpenIDImplicitGrant):
|
||||
def exists_nonce(self, nonce, request):
|
||||
raise NotImplementedError()
|
||||
return exists_nonce(nonce, request)
|
||||
|
||||
def get_jwt_config(self, grant):
|
||||
raise NotImplementedError()
|
||||
def get_jwt_config(self, grant=None):
|
||||
return DUMMY_JWT_CONFIG
|
||||
|
||||
def generate_user_info(self, user, scope):
|
||||
raise NotImplementedError()
|
||||
user = User.get(user)
|
||||
return generate_user_info(user, scope)
|
||||
|
||||
|
||||
|
|
Loading…
Reference in a new issue