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
|
"""JWT as Authorisation Grant"""
import uuid
from dataclasses import asdict
from flask import current_app as app
from authlib.common.security import generate_token
from authlib.oauth2.rfc7523.jwt_bearer import JWTBearerGrant as _JWTBearerGrant
from authlib.oauth2.rfc7523.token import (
JWTBearerTokenGenerator as _JWTBearerTokenGenerator)
from gn_auth.auth.db.sqlite3 import with_db_connection
from gn_auth.auth.authentication.users import user_by_id
from gn_auth.auth.authorisation.roles.models import user_roles
def convert_uuids_to_string(srcdict: dict) -> dict:
"""
Convert *ALL* UUID objects in a dict to strings.
`json.dumps` does not encode UUID objects by default.
"""
def uuid2str(key, value):
if isinstance(value, dict):
return (key, convert_uuids_to_string(value))
if isinstance(value, uuid.UUID):
return (key, str(value))
return (key, value)
return dict(tuple(uuid2str(_key, _val) for _key, _val in srcdict.items()))
class JWTBearerTokenGenerator(_JWTBearerTokenGenerator):
"""
A JSON Web Token formatted bearer token generator for jwt-bearer grant type.
"""
DEFAULT_EXPIRES_IN = 300
def get_token_data(#pylint: disable=[too-many-arguments]
self, grant_type, client, expires_in=None, user=None, scope=None
):
"""Post process data to prevent JSON serialization problems."""
tokendata = super().get_token_data(
grant_type, client, expires_in, user, scope)
return {
**{
key: str(value) if key.endswith("_id") else value
for key, value in tokendata.items()
},
"sub": str(tokendata["sub"]),
"jti": str(uuid.uuid4()),
"gn:auth:user:roles": tuple(convert_uuids_to_string({
**item,
"roles": tuple(convert_uuids_to_string(asdict(role))
for role in item["roles"])
}) for item in with_db_connection(
lambda conn: user_roles(conn, user)))
}
def __call__(# pylint: disable=[too-many-arguments]
self, grant_type, client, user=None, scope=None, expires_in=None,
include_refresh_token=True
):
# there is absolutely no refresh token in JWT format
"""
The default generator does not provide refresh tokens with JWT. It goes
so far as to state "there is absolutely no refresh token in JWT format".
This shim allows us to have a refresh token. We should probably look for
a supported way of using JWTs with refresh capability.
"""
token = self.generate(grant_type, client, user, scope, expires_in)
if include_refresh_token:
return {
**token,
"refresh_token": generate_token(length=42)
}
return token
class JWTBearerGrant(_JWTBearerGrant):
"""Implement JWT as Authorisation Grant."""
TOKEN_ENDPOINT_AUTH_METHODS = ["client_secret_post", "client_secret_jwt"]
CLAIMS_OPTIONS = {
**_JWTBearerGrant.CLAIMS_OPTIONS,
"jti": {"essential": True}
}
def resolve_issuer_client(self, issuer):
"""Fetch client via "iss" in assertion claims."""
return with_db_connection(
lambda conn: self.server.query_client(issuer))
def resolve_client_key(self, client, headers, payload):
"""Resolve client key to decode assertion data."""
return app.config["SSL_PUBLIC_KEYS"].get(headers["kid"])
def authenticate_user(self, subject):
"""Authenticate user with the given assertion claims."""
return with_db_connection(lambda conn: user_by_id(conn, subject))
def has_granted_permission(self, client, user):
"""
Check if the client has permission to access the given user's resource.
"""
return True # TODO: Check this!!!
def create_token_response(self):
"""If valid and authorized, the authorization server issues an access
token.
"""
token = self.generate_token(
scope=self.request.scope,
user=self.request.user,
include_refresh_token=self.request.client.check_grant_type(
"refresh_token")
)
app.logger.debug('Issue token %r to %r', token, self.request.client)
self.save_token(token)
return 200, token, self.TOKEN_RESPONSE_HEADER
|
