diff options
| -rw-r--r-- | gn_auth/auth/authentication/oauth2/grants/jwt_bearer_grant.py | 89 | ||||
| -rw-r--r-- | gn_auth/auth/authentication/oauth2/server.py | 6 | ||||
| -rw-r--r-- | gn_auth/auth/authentication/users.py | 4 |
3 files changed, 99 insertions, 0 deletions
diff --git a/gn_auth/auth/authentication/oauth2/grants/jwt_bearer_grant.py b/gn_auth/auth/authentication/oauth2/grants/jwt_bearer_grant.py new file mode 100644 index 0000000..492009c --- /dev/null +++ b/gn_auth/auth/authentication/oauth2/grants/jwt_bearer_grant.py @@ -0,0 +1,89 @@ +"""JWT as Authorisation Grant""" +import uuid +from urllib.parse import urlparse +from datetime import datetime, timedelta + +from flask import request, current_app as app + +from authlib.jose import jwt +from authlib.jose.rfc7517.key_set import KeySet +from authlib.oauth2.rfc7523.jwt_bearer import JWTBearerGrant as _JWTBearerGrant +from authlib.oauth2.rfc7523.token import ( + JWTBearerTokenGenerator as _JWTBearerTokenGenerator) + +from gn_auth.auth.authentication.users import user_by_id +from gn_auth.auth.db.sqlite3 import connection, with_db_connection +from gn_auth.auth.authentication.oauth2.models.oauth2client import client +from gn_auth.auth.authentication.oauth2.grants.authorisation_code_grant import AuthorisationCodeGrant + + +class JWTBearerTokenGenerator(_JWTBearerTokenGenerator): + """ + A JSON Web Token formatted bearer token generator for jwt-bearer grant type. + """ + + DEFAULT_EXPIRES_IN = 300 + + def get_token_data(self, grant_type, client, expires_in=300, 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"])} + + +class JWTBearerGrant(_JWTBearerGrant, AuthorisationCodeGrant): + """Implement JWT as Authorisation Grant.""" + + + def create_authorization_response(self, redirect_uri: str, grant_user): + resp = super().create_authorization_response(redirect_uri, grant_user) + headers = dict(resp[2]) + location = urlparse(headers["Location"]) + query = { + key.strip(): value.strip() for key, value in + (item.split("=") for item in + (param.strip() for param in location.query.split("&")))} + parsed_redirect = urlparse(redirect_uri) + issued = datetime.now() + jwtkey = app.config["JWT_PRIVATE_KEY"] + jwttoken = jwt.encode( + {"alg": "RS256", "typ": "jwt", "kid": jwtkey.kid}, + { + "iss": str(self.client.client_id), + "sub": str(grant_user.user_id), + "aud": f"{parsed_redirect.scheme}://{parsed_redirect.netloc}", + "exp": (issued + timedelta(minutes=5)), + "nbf": int(issued.timestamp()), + "iat": int(issued.timestamp()), + "jti": str(uuid.uuid4()), + "code": query["code"]}, + jwtkey).decode("utf8") + return (302, "", [("Location", f"{location.geturl()}&jwt={jwttoken}")]) + + + 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["JWT_PUBLIC_KEY"] + + + 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!!! diff --git a/gn_auth/auth/authentication/oauth2/server.py b/gn_auth/auth/authentication/oauth2/server.py index 0bce3d3..09a3449 100644 --- a/gn_auth/auth/authentication/oauth2/server.py +++ b/gn_auth/auth/authentication/oauth2/server.py @@ -14,6 +14,7 @@ from .models.oauth2token import OAuth2Token, save_token from .grants.password_grant import PasswordGrant from .grants.authorisation_code_grant import AuthorisationCodeGrant +from .grants.jwt_bearer_grant import JWTBearerGrant, JWTBearerTokenGenerator from .endpoints.revocation import RevocationEndpoint from .endpoints.introspection import IntrospectionEndpoint @@ -59,6 +60,11 @@ def setup_oauth2_server(app: Flask) -> None: # below server.register_grant(AuthorisationCodeGrant) + server.register_grant(JWTBearerGrant) + server.register_token_generator( + "urn:ietf:params:oauth:grant-type:jwt-bearer", + JWTBearerTokenGenerator(app.config["JWT_PRIVATE_KEY"])) + # register endpoints server.register_endpoint(RevocationEndpoint) server.register_endpoint(IntrospectionEndpoint) diff --git a/gn_auth/auth/authentication/users.py b/gn_auth/auth/authentication/users.py index cb6a68e..1f364b8 100644 --- a/gn_auth/auth/authentication/users.py +++ b/gn_auth/auth/authentication/users.py @@ -17,6 +17,10 @@ class User: email: str name: str + def get_user_id(self): + """Get the user's ID.""" + return self.user_id + DUMMY_USER = User(user_id=UUID("a391cf60-e8b7-4294-bd22-ddbbda4b3530"), email="gn3@dummy.user", |