oauth2.py
3.8 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
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
from authlib.integrations.flask_oauth2 import (
AuthorizationServer, ResourceProtector)
from authlib.integrations.sqla_oauth2 import (
create_query_client_func,
create_save_token_func,
create_bearer_token_validator,
)
from authlib.oauth2.rfc6749.grants import (
AuthorizationCodeGrant as _AuthorizationCodeGrant,
)
from authlib.oidc.core.grants import (
OpenIDCode as _OpenIDCode,
OpenIDImplicitGrant as _OpenIDImplicitGrant,
OpenIDHybridGrant as _OpenIDHybridGrant,
)
from authlib.oidc.core import UserInfo
from werkzeug.security import gen_salt
from .models import db, User
from .models import OAuth2Client, OAuth2AuthorizationCode, OAuth2Token
DUMMY_JWT_CONFIG = {
'key': 'secret-key',
'alg': 'HS256',
'iss': 'https://authlib.org',
'exp': 7200,
}
def exists_nonce(nonce, req):
exists = OAuth2AuthorizationCode.query.filter_by(
client_id=req.client_id, nonce=nonce
).first()
return bool(exists)
def generate_user_info(user, scope):
return UserInfo(sub=str(user.id), name=user.username, role=user.role, company=user.company)
def create_authorization_code(client, grant_user, request):
code = gen_salt(48)
nonce = request.data.get('nonce')
item = OAuth2AuthorizationCode(
code=code,
client_id=client.client_id,
redirect_uri=request.redirect_uri,
scope=request.scope,
user_id=grant_user.id,
nonce=nonce,
)
db.session.add(item)
db.session.commit()
return code
class AuthorizationCodeGrant(_AuthorizationCodeGrant):
def create_authorization_code(self, client, grant_user, request):
return create_authorization_code(client, grant_user, request)
def parse_authorization_code(self, code, client):
item = OAuth2AuthorizationCode.query.filter_by(
code=code, client_id=client.client_id).first()
if item and not item.is_expired():
return item
def delete_authorization_code(self, authorization_code):
db.session.delete(authorization_code)
db.session.commit()
def authenticate_user(self, authorization_code):
return User.query.get(authorization_code.user_id)
class OpenIDCode(_OpenIDCode):
def exists_nonce(self, nonce, request):
return exists_nonce(nonce, request)
def get_jwt_config(self, grant):
return DUMMY_JWT_CONFIG
def generate_user_info(self, user, scope):
return generate_user_info(user, scope)
class ImplicitGrant(_OpenIDImplicitGrant):
def exists_nonce(self, nonce, request):
return exists_nonce(nonce, request)
def get_jwt_config(self, grant):
return DUMMY_JWT_CONFIG
def generate_user_info(self, user, scope):
return generate_user_info(user, scope)
class HybridGrant(_OpenIDHybridGrant):
def create_authorization_code(self, client, grant_user, request):
return create_authorization_code(client, grant_user, request)
def exists_nonce(self, nonce, request):
return exists_nonce(nonce, request)
def get_jwt_config(self):
return DUMMY_JWT_CONFIG
def generate_user_info(self, user, scope):
return generate_user_info(user, scope)
authorization = AuthorizationServer()
require_oauth = ResourceProtector()
def config_oauth(app):
query_client = create_query_client_func(db.session, OAuth2Client)
save_token = create_save_token_func(db.session, OAuth2Token)
authorization.init_app(
app,
query_client=query_client,
save_token=save_token
)
# support all openid grants
authorization.register_grant(AuthorizationCodeGrant, [
OpenIDCode(require_nonce=True),
])
authorization.register_grant(ImplicitGrant)
authorization.register_grant(HybridGrant)
# protect resource
bearer_cls = create_bearer_token_validator(db.session, OAuth2Token)
require_oauth.register_token_validator(bearer_cls())