feat: support locally auth

server/.env.example

@@ -23,8 +23,14 @@ X_GITHUB_APP_ID=github_app_id
 X_GITHUB_APPS_CLIENT_ID=github_apps_client_id
 X_GITHUB_APPS_CLIENT_SECRET=github_apps_client_secret
 
+PETERCAT_AUTH0_ENABLED=False
+# OPTIONAL - Local Authorization Configures
+
+PETERCAT_LOCAL_UID="petercat|001"
+PETERCAT_LOCAL_UNAME="petercat"
 
 # OPTIONAL - AUTH0 Configures
+
 API_IDENTIFIER=api_identifier
 AUTH0_DOMAIN=auth0_domain
 AUTH0_CLIENT_ID=auth0_client_id

server/.env.local.example

@@ -23,6 +23,12 @@ X_GITHUB_APP_ID=github_app_id
 X_GITHUB_APPS_CLIENT_ID=github_apps_client_id
 X_GITHUB_APPS_CLIENT_SECRET=github_apps_client_secret
 
+# OPTIONAL - Local Authorization Configures
+PETERCAT_LOCAL_UID="petercat|001"
+PETERCAT_LOCAL_UNAME="petercat"
+
+# OPTIONAL - SKIP AUTH0 Authorization
+PETERCAT_AUTH0_ENABLED=True
 
 # OPTIONAL - AUTH0 Configures
 API_IDENTIFIER=api_identifier

server/auth/clients/__init__.py

@@ -0,0 +1,12 @@
+from auth.clients.auth0 import Auth0Client
+from auth.clients.base import BaseAuthClient
+from auth.clients.local import LocalClient
+
+from petercat_utils import get_env_variable
+
+PETERCAT_AUTH0_ENABLED = get_env_variable("PETERCAT_AUTH0_ENABLED", "True") == "True"
+
+def get_auth_client() -> BaseAuthClient:
+  if PETERCAT_AUTH0_ENABLED:
+    return Auth0Client()
+  return LocalClient()

server/auth/clients/auth0.py

@@ -0,0 +1,92 @@
+import httpx
+import secrets
+
+from fastapi import Request
+from auth.clients.base import BaseAuthClient
+from petercat_utils import get_env_variable
+from starlette.config import Config
+from authlib.integrations.starlette_client import OAuth
+
+CLIENT_ID = get_env_variable("AUTH0_CLIENT_ID")
+CLIENT_SECRET = get_env_variable("AUTH0_CLIENT_SECRET")
+AUTH0_DOMAIN = get_env_variable("AUTH0_DOMAIN")
+API_AUDIENCE = get_env_variable("API_IDENTIFIER")
+API_URL = get_env_variable("API_URL")
+
+CALLBACK_URL = f"{API_URL}/api/auth/callback"
+
+config = Config(
+    environ={
+        "AUTH0_CLIENT_ID": CLIENT_ID,
+        "AUTH0_CLIENT_SECRET": CLIENT_SECRET,
+    }
+)
+
+class Auth0Client(BaseAuthClient):
+  _client: OAuth
+
+  def __init__(self):
+    self._client = OAuth(config)
+    self._client.register(
+        name="auth0",
+        server_metadata_url=f"https://{AUTH0_DOMAIN}/.well-known/openid-configuration",
+        client_kwargs={"scope": "openid email profile"},
+    )
+
+  async def login(self, request):
+    return await self._client.auth0.authorize_redirect(
+        request, redirect_uri=CALLBACK_URL
+    )
+
+  async def get_oauth_token(self):
+    url = f'https://{AUTH0_DOMAIN}/oauth/token'
+    headers = {"content-type": "application/x-www-form-urlencoded"}
+    data = {
+        'client_id': CLIENT_ID,
+        'client_secret': CLIENT_SECRET,
+        'audience': API_AUDIENCE,
+        'grant_type': 'client_credentials'
+    }
+    async with httpx.AsyncClient() as client:
+        response = await client.post(url, data=data, headers=headers)
+        return response.json()['access_token']
+
+  async def get_user_info(self, request: Request) -> dict:
+    auth0_token = await self._client.auth0.authorize_access_token(request)
+    access_token = auth0_token["access_token"]
+    userinfo_url = f"https://{AUTH0_DOMAIN}/userinfo"
+    headers = {"authorization": f"Bearer {access_token}"}
+    async with httpx.AsyncClient() as client:
+        user_info_response = await client.get(userinfo_url, headers=headers)
+        if user_info_response.status_code == 200:
+            user_info = user_info_response.json()
+            data = {
+                "id": user_info["sub"],
+                "nickname": user_info.get("nickname"),
+                "name": user_info.get("name"),
+                "picture": user_info.get("picture"),
+                "sub": user_info["sub"],
+                "sid": secrets.token_urlsafe(32),
+                "agreement_accepted": user_info.get("agreement_accepted"),
+            }
+            return data
+        else:
+            return None
+
+  async def get_access_token(self, user_id: str, provider="github"):
+    token = await self.get_oauth_token()
+    user_accesstoken_url = f"https://{AUTH0_DOMAIN}/api/v2/users/{user_id}"
+
+    async with httpx.AsyncClient() as client:
+        headers = {"authorization": f"Bearer {token}"}
+        user_info_response = await client.get(user_accesstoken_url, headers=headers)
+        user = user_info_response.json()
+        identity = next(
+            (
+                identity
+                for identity in user["identities"]
+                if identity["provider"] == provider
+            ),
+            None,
+        )
+        return identity["access_token"]

server/auth/clients/base.py

@@ -0,0 +1,51 @@
+import secrets
+
+from abc import abstractmethod
+from fastapi import Request
+from utils.random_str import random_str
+from petercat_utils import get_client
+
+
+class BaseAuthClient:
+  def __init__(self):
+    pass
+
+  def generateAnonymousUser(self, clientId: str) -> tuple[str, dict]:
+    token = f"client|{clientId}"
+    seed = clientId[:4]
+    random_name = f"{seed}_{random_str(4)}"
+    data = {
+        "id": token,
+        "sub": token,
+        "nickname": random_name,
+        "name": random_name,
+        "picture": f"https://picsum.photos/seed/{seed}/100/100",
+        "sid": secrets.token_urlsafe(32),
+        "agreement_accepted": False,
+    }
+
+    return token, data
+
+  async def anonymouseLogin(self, request: Request) -> dict:
+    clientId = request.query_params.get("clientId") or random_str()
+    token, data = self.generateAnonymousUser(clientId = clientId)
+    supabase = get_client()
+    supabase.table("profiles").upsert(data).execute()
+    request.session["user"] = data
+    return data
+
+  @abstractmethod
+  async def login(self, request: Request):
+    pass
+
+  @abstractmethod
+  async def get_oauth_token(self) -> str:
+    pass
+
+  @abstractmethod
+  async def get_user_info(self, request: Request) -> dict:
+    pass
+
+  @abstractmethod
+  async def get_access_token(self, user_id: str, provider="github") -> str:
+    pass

server/auth/clients/local.py

@@ -0,0 +1,38 @@
+import secrets
+from fastapi import Request
+from fastapi.responses import RedirectResponse
+from petercat_utils import get_client, get_env_variable
+from auth.clients.base import BaseAuthClient
+
+PETERCAT_LOCAL_UID = get_env_variable("PETERCAT_LOCAL_UID")
+PETERCAT_LOCAL_UNAME = get_env_variable("PETERCAT_LOCAL_UNAME")
+WEB_URL = get_env_variable("WEB_URL")
+WEB_LOGIN_SUCCESS_URL = f"{WEB_URL}/user/login"
+
+class LocalClient(BaseAuthClient):
+  def __init__(self):
+    pass
+
+  async def login(self, request: Request):
+    data = await self.get_user_info()
+    supabase = get_client()
+    supabase.table("profiles").upsert(data).execute()
+    request.session["user"] = data
+
+    return RedirectResponse(url=f"{WEB_LOGIN_SUCCESS_URL}", status_code=302)
+
+  async def get_user_info(user_id):
+    token = PETERCAT_LOCAL_UID
+    username = PETERCAT_LOCAL_UNAME
+    seed = token[:4]
+
+    return {
+      "id": token,
+      "sub": token,
+      "nickname": username,
+      "name": username,
+      "picture": f"https://picsum.photos/seed/{seed}/100/100",
+      "sid": secrets.token_urlsafe(32),
+      "agreement_accepted": False,
+    }
+

server/auth/get_user_info.py

@@ -1,80 +1,14 @@
 from fastapi import Request
-import httpx
-import secrets
+from auth.clients import get_auth_client
 from core.models.user import User
 
-from utils.random_str import random_str
-
-from .get_oauth_token import get_oauth_token
-from petercat_utils import get_client, get_env_variable
-
+from petercat_utils import get_env_variable
 
 AUTH0_DOMAIN = get_env_variable("AUTH0_DOMAIN")
 
-
-async def getUserInfoByToken(token):
-    userinfo_url = f"https://{AUTH0_DOMAIN}/userinfo"
-    headers = {"authorization": f"Bearer {token}"}
-    async with httpx.AsyncClient() as client:
-        user_info_response = await client.get(userinfo_url, headers=headers)
-        if user_info_response.status_code == 200:
-            user_info = user_info_response.json()
-            data = {
-                "id": user_info["sub"],
-                "nickname": user_info.get("nickname"),
-                "name": user_info.get("name"),
-                "picture": user_info.get("picture"),
-                "avatar": user_info.get("picture"),
-                "sub": user_info["sub"],
-                "sid": secrets.token_urlsafe(32),
-                "agreement_accepted": user_info.get("agreement_accepted"),
-            }
-            return data
-        else:
-            return None
-
-
 async def getUserAccessToken(user_id: str, provider="github"):
-    token = await get_oauth_token()
-    user_accesstoken_url = f"https://{AUTH0_DOMAIN}/api/v2/users/{user_id}"
-
-    async with httpx.AsyncClient() as client:
-        headers = {"authorization": f"Bearer {token}"}
-        user_info_response = await client.get(user_accesstoken_url, headers=headers)
-        user = user_info_response.json()
-        identity = next(
-            (
-                identity
-                for identity in user["identities"]
-                if identity["provider"] == provider
-            ),
-            None,
-        )
-        return identity["access_token"]
-
-
-async def generateAnonymousUser(clientId: str):
-    token = f"client|{clientId}"
-    seed = clientId[:4]
-    random_name = f"{seed}_{random_str(4)}"
-    data = {
-        "id": token,
-        "sub": token,
-        "nickname": random_name,
-        "name": random_name,
-        "picture": f"https://picsum.photos/seed/{seed}/100/100",
-        "sid": secrets.token_urlsafe(32),
-        "agreement_accepted": False,
-    }
-
-    return token, data
-
-
-async def getAnonymousUserInfoByToken(token: str):
-    supabase = get_client()
-    rows = supabase.table("profiles").select("*").eq("id", token).execute()
-    return rows.data[0] if (len(rows.data) > 0) else None
-
+    auth_client = get_auth_client()
+    return await auth_client.get_access_token(user_id=user_id, provider=provider)
 
 async def get_user_id(request: Request):
     user_info = request.session.get("user")

server/auth/rate_limit.py

@@ -1,22 +1,24 @@
-from typing import Annotated
-from fastapi import Cookie, HTTPException
+from typing import Optional
+from fastapi import Depends, HTTPException
 from datetime import datetime, timedelta
 
+from auth.clients import get_auth_client
+from auth.clients.base import BaseAuthClient
 from petercat_utils import get_client, get_env_variable
 
-from auth.get_user_info import getUserInfoByToken
+from auth.get_user_info import get_user_id
 
 RATE_LIMIT_ENABLED = get_env_variable("RATE_LIMIT_ENABLED", "False") == 'True'
 RATE_LIMIT_REQUESTS = get_env_variable("RATE_LIMIT_REQUESTS") or 100
 RATE_LIMIT_DURATION = timedelta(minutes=int(get_env_variable("RATE_LIMIT_DURATION") or 1))
 
-async def verify_rate_limit(petercat_user_token: Annotated[str | None, Cookie()] = None):
+async def verify_rate_limit(user_id: Optional[str] = Depends(get_user_id), auth_client: BaseAuthClient = Depends(get_auth_client)):
     if not RATE_LIMIT_ENABLED:
         return
 
-    if not petercat_user_token:
+    if not user_id:
         raise HTTPException(status_code=403, detail="Must Login")
-    user = await getUserInfoByToken(petercat_user_token)
+    user = await auth_client.get_user_info(user_id)
 
     if user is None:
         raise HTTPException(