add new token endpoint

Author: GalMunGral

backend/app/routers/keycloak.py

@@ -11,8 +11,8 @@
     retreive_refresh_token,
 )
 from app.models.tokens import TokenDB
-from app.models.users import UserDB, UserLogin
-from app.routers.utils import save_refresh_token
+from app.models.users import UserLogin
+from app.routers.utils import save_refresh_token, get_token
 from fastapi import APIRouter, HTTPException, Security
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from jose import ExpiredSignatureError, JWTError, jwt
@@ -107,56 +107,10 @@ async def loginPost(userIn: UserLogin):
 async def auth(code: str) -> RedirectResponse:
     """Redirect endpoint Keycloak redirects to after login."""
     logger.info("In /api/v2/auth")
-    # get token from Keycloak
-    token_body = keycloak_openid.token(
-        grant_type="authorization_code",
-        code=code,
-        redirect_uri=settings.auth_redirect_uri,
-    )
 
+    token_body = await get_token(code)
     access_token = token_body["access_token"]
 
-    # create user in db if it doesn't already exist; get the user_id
-    userinfo = keycloak_openid.userinfo(access_token)
-    keycloak_id = userinfo["sub"]
-    given_name = userinfo.get("given_name", " ")
-    family_name = userinfo.get("family_name", " ")
-    email = userinfo["email"]
-
-    # check if this is the 1st user, make it admin
-    count = await UserDB.count()
-
-    if count == 0:
-        user = UserDB(
-            email=email,
-            first_name=given_name,
-            last_name=family_name,
-            hashed_password="",
-            keycloak_id=keycloak_id,
-            admin=True,
-        )
-    else:
-        user = UserDB(
-            email=email,
-            first_name=given_name,
-            last_name=family_name,
-            hashed_password="",
-            keycloak_id=keycloak_id,
-            admin=False,
-        )
-    matched_user = await UserDB.find_one(UserDB.email == email)
-    if matched_user is None:
-        await user.insert()
-
-    # store/update refresh token and link to that userid
-    token_exist = await TokenDB.find_one(TokenDB.email == email)
-    if token_exist is not None:
-        token_exist.refresh_token = token_body["refresh_token"]
-        await token_exist.save()
-    else:
-        token_created = TokenDB(email=email, refresh_token=token_body["refresh_token"])
-        await token_created.insert()
-
     # redirect to frontend
     auth_url = f"{settings.frontend_url}/auth"
 
@@ -168,6 +122,11 @@ async def auth(code: str) -> RedirectResponse:
     return response
 
 
+@router.get("/token")
+async def token(code: str):
+    return await get_token(code)
+
+
 @router.get("/refresh_token")
 async def refresh_token(
     credentials: HTTPAuthorizationCredentials = Security(security),

backend/app/routers/utils.py

@@ -1,8 +1,11 @@
 import mimetypes
 from typing import Optional
 
+from app.config import settings
+from app.keycloak_auth import keycloak_openid
 from app.models.files import ContentType
 from app.models.tokens import TokenDB
+from app.models.users import UserDB
 
 
 def get_content_type(
@@ -35,3 +38,58 @@ async def save_refresh_token(refresh_token: str, email: str):
     else:
         token_created = TokenDB(email=email, refresh_token=refresh_token)
         await token_created.insert()
+
+
+async def get_token(code: str):
+    # get token from Keycloak
+    token_body = keycloak_openid.token(
+        grant_type="authorization_code",
+        code=code,
+        redirect_uri=settings.auth_redirect_uri,
+    )
+
+    access_token = token_body["access_token"]
+    refresh_token = token_body["refresh_token"]
+
+    # create user in db if it doesn't already exist; get the user_id
+    userinfo = keycloak_openid.userinfo(access_token)
+    keycloak_id = userinfo["sub"]
+    given_name = userinfo.get("given_name", " ")
+    family_name = userinfo.get("family_name", " ")
+    email = userinfo["email"]
+
+    # check if this is the 1st user, make it admin
+    count = await UserDB.count()
+
+    if count == 0:
+        user = UserDB(
+            email=email,
+            first_name=given_name,
+            last_name=family_name,
+            hashed_password="",
+            keycloak_id=keycloak_id,
+            admin=True,
+        )
+    else:
+        user = UserDB(
+            email=email,
+            first_name=given_name,
+            last_name=family_name,
+            hashed_password="",
+            keycloak_id=keycloak_id,
+            admin=False,
+        )
+    matched_user = await UserDB.find_one(UserDB.email == email)
+    if matched_user is None:
+        await user.insert()
+
+    # store/update refresh token and link to that userid
+    token_exist = await TokenDB.find_one(TokenDB.email == email)
+    if token_exist is not None:
+        token_exist.refresh_token = refresh_token
+        await token_exist.save()
+    else:
+        token_created = TokenDB(email=email, refresh_token=refresh_token)
+        await token_created.insert()
+
+    return token_body

frontend/src/openapi/v2/services/AuthService.ts

@@ -88,6 +88,27 @@ export class AuthService {
         });
     }
 
+    /**
+     * Token
+     * @param code
+     * @returns any Successful Response
+     * @throws ApiError
+     */
+    public static tokenApiV2AuthTokenGet(
+        code: string,
+    ): CancelablePromise<any> {
+        return __request({
+            method: 'GET',
+            path: `/api/v2/auth/token`,
+            query: {
+                'code': code,
+            },
+            errors: {
+                422: `Validation Error`,
+            },
+        });
+    }
+
     /**
      * Refresh Token
      * @returns any Successful Response

openapi.json

@@ -12058,6 +12058,46 @@
         }
       }
     },
+    "/api/v2/auth/token": {
+      "get": {
+        "tags": [
+          "auth"
+        ],
+        "summary": "Token",
+        "operationId": "token_api_v2_auth_token_get",
+        "parameters": [
+          {
+            "required": true,
+            "schema": {
+              "title": "Code",
+              "type": "string"
+            },
+            "name": "code",
+            "in": "query"
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Successful Response",
+            "content": {
+              "application/json": {
+                "schema": {}
+              }
+            }
+          },
+          "422": {
+            "description": "Validation Error",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/HTTPValidationError"
+                }
+              }
+            }
+          }
+        }
+      }
+    },
     "/api/v2/auth/refresh_token": {
       "get": {
         "tags": [