Rearrange the files and create some necessary ones

Author: ArtyomVancyan

.env

@@ -1,8 +1,8 @@
-GITHUB_CLIENT_ID=eccd08d6736b7999a32a
-GITHUB_CLIENT_SECRET=642999c1c5f2b3df8b877afdc78252ef5b594d31
-GITHUB_REDIRECT_URL=http://127.0.0.1:8000/auth/callback
-MAIN_PAGE_REDIRECT_URL=http://127.0.0.1:8000/
+CLIENT_ID=eccd08d6736b7999a32a
+CLIENT_SECRET=642999c1c5f2b3df8b877afdc78252ef5b594d31
+CALLBACK_URL=http://127.0.0.1:8000/auth/callback
+REDIRECT_URL=http://127.0.0.1:8000/
 
-JWT_SECRET_KEY=secret
+JWT_SECRET=secret
 JWT_ALGORITHM=HS256
-JWT_TOKEN_EXPIRES=300
+JWT_EXPIRES=5

demo/config.py

@@ -8,7 +8,7 @@
 from starlette.requests import Request
 from starlette.status import HTTP_403_FORBIDDEN
 
-from .config import SECRET_KEY, ALGORITHM
+from fastapi_oauth2.config import JWT_SECRET, JWT_ALGORITHM
 
 
 class OAuth2PasswordBearerCookie(OAuth2):
@@ -19,38 +19,17 @@ def __init__(
             scopes: dict = None,
             auto_error: bool = True,
     ):
-        if not scopes:
-            scopes = {}
-        flows = OAuthFlowsModel(password={"tokenUrl": tokenUrl, "scopes": scopes})
+        flows = OAuthFlowsModel(password={"tokenUrl": tokenUrl, "scopes": scopes or {}})
         super().__init__(flows=flows, scheme_name=scheme_name, auto_error=auto_error)
 
     async def __call__(self, request: Request) -> Optional[str]:
-        header_authorization: str = request.headers.get("Authorization")
-        cookie_authorization: str = request.cookies.get("Authorization")
+        scheme, param = get_authorization_scheme_param(request.headers.get("Authorization"))
+        authorization = scheme.lower() == "bearer"
+        if not authorization:
+            scheme, param = get_authorization_scheme_param(request.cookies.get("Authorization"))
+            authorization = scheme.lower() == "bearer"
 
-        header_scheme, header_param = get_authorization_scheme_param(
-            header_authorization
-        )
-        cookie_scheme, cookie_param = get_authorization_scheme_param(
-            cookie_authorization
-        )
-
-        if header_scheme.lower() == "bearer":
-            authorization = True
-            scheme = header_scheme
-            param = header_param
-
-        elif cookie_scheme.lower() == "bearer":
-            authorization = True
-            scheme = cookie_scheme
-            param = cookie_param
-
-        else:
-            authorization = False
-            scheme = ""
-            param = ""
-
-        if not authorization or scheme.lower() != "bearer":
+        if not authorization:
             if self.auto_error:
                 raise HTTPException(
                     status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
@@ -65,7 +44,7 @@ async def __call__(self, request: Request) -> Optional[str]:
 
 async def get_current_user(token: str = Depends(oauth2_scheme)):
     try:
-        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
     except JWTError:
         raise HTTPException(
             status_code=HTTP_403_FORBIDDEN, detail="Could not validate credentials"

demo/dependencies.py

@@ -1,28 +1,10 @@
-from datetime import timedelta
-
 from fastapi import APIRouter
 from fastapi import Depends
-from fastapi.responses import RedirectResponse
 from starlette.requests import Request
 
-from fastapi_oauth2.github import GitHubSSO
-from .config import (
-    CLIENT_ID,
-    CLIENT_SECRET,
-    redirect_url,
-    ACCESS_TOKEN_EXPIRE_MINUTES,
-    redirect_url_main_page,
-)
 from .dependencies import get_current_user
-from .utils import create_access_token
 
 router = APIRouter()
-sso = GitHubSSO(
-    client_id=CLIENT_ID,
-    client_secret=CLIENT_SECRET,
-    redirect_uri=redirect_url,
-    allow_insecure_http=True,
-)
 
 
 @router.get("/user")
@@ -33,33 +15,3 @@ def user(current_user=Depends(get_current_user)):
 @router.post("/token")
 def token(request: Request):
     return request.cookies.get("Authorization")
-
-
-@router.get("/auth/login")
-async def auth_init():
-    return await sso.get_login_redirect()
-
-
-@router.get("/auth/callback")
-async def auth_callback(request: Request):
-    user = await sso.verify_and_process(request)
-    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
-    access_token = create_access_token(
-        data=dict(user), expires_delta=access_token_expires
-    )
-    response = RedirectResponse(redirect_url_main_page)
-    response.set_cookie(
-        "Authorization",
-        value=f"Bearer {access_token}",
-        httponly=sso.allow_insecure_http,
-        max_age=ACCESS_TOKEN_EXPIRE_MINUTES * 60,
-        expires=ACCESS_TOKEN_EXPIRE_MINUTES * 60,
-    )
-    return response
-
-
-@router.get("/auth/logout")
-async def auth_logout():
-    response = RedirectResponse(redirect_url_main_page)
-    response.delete_cookie("Authorization")
-    return response

demo/router.py

@@ -21,7 +21,7 @@ class SSOBase:
 
     client_id: str = None
     client_secret: str = None
-    redirect_uri: Optional[str] = None
+    callback_url: Optional[str] = None
     allow_insecure_http: bool = False
     scope: Optional[List[str]] = None
     state: Optional[str] = None
@@ -36,13 +36,13 @@ def __init__(
             self,
             client_id: str,
             client_secret: str,
-            redirect_uri: Optional[str] = None,
+            callback_url: Optional[str] = None,
             allow_insecure_http: bool = False,
             scope: Optional[List[str]] = None,
     ):
         self.client_id = client_id
         self.client_secret = client_secret
-        self.redirect_uri = redirect_uri
+        self.callback_url = callback_url
         self.allow_insecure_http = allow_insecure_http
         if allow_insecure_http:
             os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1"
@@ -71,9 +71,9 @@ async def get_login_url(
     ) -> Any:
         self.state = state
         params = params or {}
-        redirect_uri = redirect_uri or self.redirect_uri
+        redirect_uri = redirect_uri or self.callback_url
         if redirect_uri is None:
-            raise ValueError("redirect_uri must be provided, either at construction or request time")
+            raise ValueError("callback_url must be provided, either at construction or request time")
         return self.oauth_client.prepare_request_uri(
             self.authorization_endpoint, redirect_uri=redirect_uri, state=state, scope=self.scope, **params
         )
@@ -113,7 +113,7 @@ async def verify_and_process(
         token_url, headers, content = self.oauth_client.prepare_token_request(
             self.token_endpoint,
             authorization_response=current_url,
-            redirect_url=redirect_uri or self.redirect_uri or current_path,
+            redirect_url=redirect_uri or self.callback_url or current_path,
             code=request.query_params.get("code"),
             **params,
         )

demo/utils.py

@@ -0,0 +1,14 @@
+import os
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+CLIENT_ID = os.getenv("CLIENT_ID")
+CLIENT_SECRET = os.getenv("CLIENT_SECRET")
+CALLBACK_URL = os.getenv("CALLBACK_URL")
+REDIRECT_URL = os.getenv("REDIRECT_URL")
+
+JWT_SECRET = os.getenv("JWT_SECRET")
+JWT_ALGORITHM = os.getenv("JWT_ALGORITHM")
+JWT_EXPIRES = int(os.getenv("JWT_EXPIRES"))

fastapi_oauth2/base.py

@@ -0,0 +1,53 @@
+from datetime import timedelta
+
+from fastapi import APIRouter
+from fastapi.responses import RedirectResponse
+from starlette.requests import Request
+
+from fastapi_oauth2.github import GitHubSSO
+from .config import (
+    CLIENT_ID,
+    CLIENT_SECRET,
+    CALLBACK_URL,
+    JWT_EXPIRES,
+    REDIRECT_URL,
+)
+from .utils import create_access_token
+
+router = APIRouter()
+sso = GitHubSSO(
+    client_id=CLIENT_ID,
+    client_secret=CLIENT_SECRET,
+    callback_url=CALLBACK_URL,
+    allow_insecure_http=True,
+)
+
+
+@router.get("/auth/login")
+async def login():
+    return await sso.get_login_redirect()
+
+
+@router.get("/auth/callback")
+async def callback(request: Request):
+    user = await sso.verify_and_process(request)
+    expires_delta = timedelta(minutes=JWT_EXPIRES)
+    access_token = create_access_token(
+        data=dict(user), expires_delta=expires_delta
+    )
+    response = RedirectResponse(REDIRECT_URL)
+    response.set_cookie(
+        "Authorization",
+        value=f"Bearer {access_token}",
+        httponly=sso.allow_insecure_http,
+        max_age=JWT_EXPIRES * 60,
+        expires=JWT_EXPIRES * 60,
+    )
+    return response
+
+
+@router.get("/auth/logout")
+async def logout():
+    response = RedirectResponse(REDIRECT_URL)
+    response.delete_cookie("Authorization")
+    return response

fastapi_oauth2/config.py

@@ -0,0 +1,11 @@
+from datetime import datetime, timedelta
+from typing import Optional
+
+from jose import jwt
+
+from .config import JWT_SECRET, JWT_ALGORITHM
+
+
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
+    expire = datetime.utcnow() + expires_delta if expires_delta else timedelta(minutes=15)
+    return jwt.encode({**data, "exp": expire}, JWT_SECRET, algorithm=JWT_ALGORITHM)

fastapi_oauth2/router.py

@@ -1,14 +1,15 @@
 import json
 
-import jwt
 from fastapi import FastAPI, Request, APIRouter
 from fastapi.responses import HTMLResponse
+from fastapi.security.utils import get_authorization_scheme_param
 from fastapi.templating import Jinja2Templates
 from starlette.authentication import AuthenticationBackend
 from starlette.middleware.authentication import AuthenticationMiddleware
 
-from demo.config import SECRET_KEY, ALGORITHM
-from demo.router import router as auth_router
+from demo.dependencies import get_current_user
+from demo.router import router as demo_router
+from fastapi_oauth2.router import router as oauth2_router
 
 router = APIRouter()
 templates = Jinja2Templates(directory="templates")
@@ -21,20 +22,19 @@ async def root(request: Request):
 
 app = FastAPI()
 app.include_router(router)
-app.include_router(auth_router)
+app.include_router(demo_router)
+app.include_router(oauth2_router)
 
 
 class BearerTokenAuthBackend(AuthenticationBackend):
     async def authenticate(self, request):
         authorization = request.cookies.get("Authorization")
+        scheme, param = get_authorization_scheme_param(authorization)
 
-        if not authorization:
+        if not scheme or not param:
             return "", None
 
-        access_token = authorization.split(" ")[1]
-        user = jwt.decode(access_token, SECRET_KEY, algorithms=[ALGORITHM])
-
-        return authorization, user
+        return authorization, await get_current_user(param)
 
 
 @app.on_event('startup')