From 0aa19052d00b2e7d0d6edd5905314484480e7ea2 Mon Sep 17 00:00:00 2001
From: Frederick Muriuki Muriithi
Date: Thu, 9 May 2024 08:06:38 +0300
Subject: Register the RefreshTokenGrant with the server

Register the RefreshTokenGrant with the server to enable refreshing of
the tokens.
---
 .../oauth2/grants/refresh_token_grant.py           | 51 ++++++++++++++++++++++
 gn_auth/auth/authentication/oauth2/server.py       |  2 +
 2 files changed, 53 insertions(+)
 create mode 100644 gn_auth/auth/authentication/oauth2/grants/refresh_token_grant.py

diff --git a/gn_auth/auth/authentication/oauth2/grants/refresh_token_grant.py b/gn_auth/auth/authentication/oauth2/grants/refresh_token_grant.py
new file mode 100644
index 0000000..fd6804d
--- /dev/null
+++ b/gn_auth/auth/authentication/oauth2/grants/refresh_token_grant.py
@@ -0,0 +1,51 @@
+"""RefreshTokenGrant: Useful for refreshing the tokens."""
+from pymonad.maybe import Nothing
+from flask import current_app as app
+from authlib.oauth2.rfc6749 import grants
+
+from gn_auth.auth.errors import NotFoundError
+from gn_auth.auth.db.sqlite3 import connection
+from gn_auth.auth.authentication.users import user_by_id
+from gn_auth.auth.authentication.oauth2.models.jwtrefreshtoken import (
+    load_refresh_token,
+    revoke_refresh_token,
+    is_refresh_token_valid)
+
+
+class RefreshTokenGrant(grants.RefreshTokenGrant):
+    """Useful for refreshing tokens"""
+    INCLUDE_NEW_REFRESH_TOKEN = True
+    TOKEN_ENDPOINT_AUTH_METHODS = ['client_secret_basic', 'client_secret_post']
+    DEFAULT_EXPIRES_IN = 432000 # 5 days
+
+    def authenticate_refresh_token(self, refresh_token):
+        """
+        Check that the refresh token is good.
+
+        Maybe also check that token has not been used before: if it has, revoke
+        any new tokens and refresh tokens issued from that.
+        """
+        with connection(app.config["AUTH_DB"]) as conn:
+            return load_refresh_token(
+                conn, refresh_token
+            ).then(
+                lambda _tok: (
+                    _tok if is_refresh_token_valid(_tok, self.request.client)
+                    else Nothing)
+            ).maybe(None, lambda _tok: _tok)
+
+    def authenticate_user(self, credential):
+        """Check that user is valid for given token."""
+        with connection(app.config["AUTH_DB"]) as conn:
+            try:
+                return user_by_id(conn, credential.user.user_id)
+            except NotFoundError as _nfe:
+                return None
+
+        return None
+
+    def revoke_old_credential(self, credential):
+        """Revoke any old refresh token after issuing new refresh token."""
+        with connection(app.config["AUTH_DB"]) as conn:
+            if credential.parent_of is not None:
+                revoke_refresh_token(conn, credential)
diff --git a/gn_auth/auth/authentication/oauth2/server.py b/gn_auth/auth/authentication/oauth2/server.py
index 70a0e51..8b65aa9 100644
--- a/gn_auth/auth/authentication/oauth2/server.py
+++ b/gn_auth/auth/authentication/oauth2/server.py
@@ -16,6 +16,7 @@ from .models.oauth2token import OAuth2Token, save_token
 from .models.jwtrefreshtoken import JWTRefreshToken, save_refresh_token
 
 from .grants.password_grant import PasswordGrant
+from .grants.refresh_token_grant import RefreshTokenGrant
 from .grants.authorisation_code_grant import AuthorisationCodeGrant
 from .grants.jwt_bearer_grant import JWTBearerGrant, JWTBearerTokenGenerator
 
@@ -83,6 +84,7 @@ def setup_oauth2_server(app: Flask) -> None:
     server.register_token_generator(
         "urn:ietf:params:oauth:grant-type:jwt-bearer",
         JWTBearerTokenGenerator(app.config["SSL_PRIVATE_KEY"]))
+    server.register_grant(RefreshTokenGrant)
 
     # register endpoints
     server.register_endpoint(RevocationEndpoint)
-- 
cgit v1.2.3