feat: multi-forge OAuth providers — Gitea + GitHub adapter pattern
- providers.py: abstract OAuthProvider with GiteaProvider + GitHubProvider - Each provider handles authorize, exchange_code, get_user, list_repositories - Provider registry: get_providers() and get_provider(name) - GitHub config: github_client_id, github_client_secret in config.py - Ready for multi-forge OAuth callback routing
This commit is contained in:
@@ -0,0 +1,259 @@
|
||||
"""Multi-forge OAuth providers — Gitea + GitHub."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from abc import ABC, abstractmethod
|
||||
from urllib.parse import urlencode
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class OAuthProvider(ABC):
|
||||
"""Abstract OAuth2 provider interface."""
|
||||
|
||||
name: str = ""
|
||||
icon: str = "🔗"
|
||||
|
||||
@abstractmethod
|
||||
def is_enabled(self) -> bool:
|
||||
"""Whether this provider is configured."""
|
||||
|
||||
@abstractmethod
|
||||
def get_authorize_url(self, state: str) -> str:
|
||||
"""Build the authorization URL."""
|
||||
|
||||
@abstractmethod
|
||||
async def exchange_code(self, code: str) -> dict | None:
|
||||
"""Exchange authorization code for access token."""
|
||||
|
||||
@abstractmethod
|
||||
async def get_user(self, access_token: str) -> dict | None:
|
||||
"""Fetch user profile from the provider."""
|
||||
|
||||
@abstractmethod
|
||||
async def list_repositories(self, access_token: str) -> list[dict]:
|
||||
"""List all repositories accessible to this user."""
|
||||
|
||||
|
||||
class GiteaProvider(OAuthProvider):
|
||||
"""Gitea OAuth2 provider."""
|
||||
|
||||
name = "gitea"
|
||||
icon = "🔗"
|
||||
|
||||
def __init__(self, base_url: str, client_id: str, client_secret: str, redirect_uri: str):
|
||||
self.base = base_url.rstrip("/")
|
||||
self.client_id = client_id
|
||||
self.client_secret = client_secret
|
||||
self.redirect_uri = redirect_uri
|
||||
|
||||
def is_enabled(self) -> bool:
|
||||
return bool(self.client_id and self.client_secret)
|
||||
|
||||
def get_authorize_url(self, state: str) -> str:
|
||||
return (
|
||||
f"{self.base}/login/oauth/authorize?"
|
||||
+ urlencode({
|
||||
"client_id": self.client_id,
|
||||
"redirect_uri": self.redirect_uri,
|
||||
"response_type": "code",
|
||||
"state": state,
|
||||
})
|
||||
)
|
||||
|
||||
async def exchange_code(self, code: str) -> dict | None:
|
||||
url = f"{self.base}/login/oauth/access_token"
|
||||
data = {
|
||||
"client_id": self.client_id,
|
||||
"client_secret": self.client_secret,
|
||||
"code": code,
|
||||
"grant_type": "authorization_code",
|
||||
"redirect_uri": self.redirect_uri,
|
||||
}
|
||||
async with httpx.AsyncClient(timeout=15) as client:
|
||||
r = await client.post(url, json=data, headers={"Accept": "application/json"})
|
||||
if r.status_code != 200:
|
||||
logger.error("Gitea token exchange failed: %s", r.text)
|
||||
return None
|
||||
return r.json()
|
||||
|
||||
async def get_user(self, access_token: str) -> dict | None:
|
||||
url = f"{self.base}/api/v1/user"
|
||||
async with httpx.AsyncClient(timeout=15) as client:
|
||||
r = await client.get(url, headers={"Authorization": f"token {access_token}"})
|
||||
if r.status_code != 200:
|
||||
return None
|
||||
data = r.json()
|
||||
return {
|
||||
"login": data.get("login") or data.get("username", ""),
|
||||
"full_name": data.get("full_name", ""),
|
||||
"email": data.get("email", ""),
|
||||
"avatar_url": data.get("avatar_url", ""),
|
||||
"provider_id": str(data.get("id", "")),
|
||||
}
|
||||
|
||||
async def list_repositories(self, access_token: str) -> list[dict]:
|
||||
repos = []
|
||||
async with httpx.AsyncClient(timeout=30) as client:
|
||||
for page in range(1, 6):
|
||||
r = await client.get(
|
||||
f"{self.base}/api/v1/user/repos",
|
||||
headers={"Authorization": f"token {access_token}"},
|
||||
params={"page": page, "limit": 50},
|
||||
)
|
||||
if r.status_code != 200:
|
||||
break
|
||||
data = r.json()
|
||||
if not data:
|
||||
break
|
||||
for repo in data:
|
||||
repos.append({
|
||||
"id": str(repo.get("id", "")),
|
||||
"name": repo.get("name", ""),
|
||||
"full_name": repo.get("full_name", ""),
|
||||
"description": repo.get("description", ""),
|
||||
"html_url": repo.get("html_url", ""),
|
||||
"clone_url": repo.get("clone_url", ""),
|
||||
"default_branch": repo.get("default_branch", "main"),
|
||||
"language": repo.get("language", ""),
|
||||
"updated_at": repo.get("updated_at", ""),
|
||||
"private": repo.get("private", False),
|
||||
"forge": "gitea",
|
||||
})
|
||||
return repos
|
||||
|
||||
|
||||
class GitHubProvider(OAuthProvider):
|
||||
"""GitHub OAuth2 provider."""
|
||||
|
||||
name = "github"
|
||||
icon = "🐙"
|
||||
|
||||
def __init__(self, client_id: str, client_secret: str, redirect_uri: str):
|
||||
self.client_id = client_id
|
||||
self.client_secret = client_secret
|
||||
self.redirect_uri = redirect_uri
|
||||
self.authorize_url = "https://github.com/login/oauth/authorize"
|
||||
self.token_url = "https://github.com/login/oauth/access_token"
|
||||
self.api_url = "https://api.github.com"
|
||||
|
||||
def is_enabled(self) -> bool:
|
||||
return bool(self.client_id and self.client_secret)
|
||||
|
||||
def get_authorize_url(self, state: str) -> str:
|
||||
return (
|
||||
f"{self.authorize_url}?"
|
||||
+ urlencode({
|
||||
"client_id": self.client_id,
|
||||
"redirect_uri": self.redirect_uri,
|
||||
"scope": "repo,user",
|
||||
"state": state,
|
||||
})
|
||||
)
|
||||
|
||||
async def exchange_code(self, code: str) -> dict | None:
|
||||
async with httpx.AsyncClient(timeout=15) as client:
|
||||
r = await client.post(
|
||||
self.token_url,
|
||||
data={
|
||||
"client_id": self.client_id,
|
||||
"client_secret": self.client_secret,
|
||||
"code": code,
|
||||
"redirect_uri": self.redirect_uri,
|
||||
},
|
||||
headers={"Accept": "application/json"},
|
||||
)
|
||||
if r.status_code != 200:
|
||||
logger.error("GitHub token exchange failed: %s", r.text)
|
||||
return None
|
||||
data = r.json()
|
||||
if "access_token" not in data:
|
||||
return None
|
||||
return data
|
||||
|
||||
async def get_user(self, access_token: str) -> dict | None:
|
||||
url = f"{self.api_url}/user"
|
||||
async with httpx.AsyncClient(timeout=15) as client:
|
||||
r = await client.get(
|
||||
url,
|
||||
headers={"Authorization": f"Bearer {access_token}", "Accept": "application/vnd.github.v3+json"},
|
||||
)
|
||||
if r.status_code != 200:
|
||||
return None
|
||||
data = r.json()
|
||||
return {
|
||||
"login": data.get("login", ""),
|
||||
"full_name": data.get("name", "") or data.get("login", ""),
|
||||
"email": data.get("email", ""),
|
||||
"avatar_url": data.get("avatar_url", ""),
|
||||
"provider_id": str(data.get("id", "")),
|
||||
}
|
||||
|
||||
async def list_repositories(self, access_token: str) -> list[dict]:
|
||||
repos = []
|
||||
async with httpx.AsyncClient(timeout=30) as client:
|
||||
for page in range(1, 6):
|
||||
r = await client.get(
|
||||
f"{self.api_url}/user/repos",
|
||||
headers={"Authorization": f"Bearer {access_token}", "Accept": "application/vnd.github.v3+json"},
|
||||
params={"page": page, "per_page": 50, "sort": "updated"},
|
||||
)
|
||||
if r.status_code != 200:
|
||||
break
|
||||
data = r.json()
|
||||
if not data:
|
||||
break
|
||||
for repo in data:
|
||||
repos.append({
|
||||
"id": str(repo.get("id", "")),
|
||||
"name": repo.get("name", ""),
|
||||
"full_name": repo.get("full_name", ""),
|
||||
"description": repo.get("description", ""),
|
||||
"html_url": repo.get("html_url", ""),
|
||||
"clone_url": repo.get("clone_url", ""),
|
||||
"default_branch": repo.get("default_branch", "main"),
|
||||
"language": repo.get("language", ""),
|
||||
"updated_at": repo.get("updated_at", ""),
|
||||
"private": repo.get("private", False),
|
||||
"forge": "github",
|
||||
})
|
||||
return repos
|
||||
|
||||
|
||||
# ═══════════ Provider registry ═══════════
|
||||
|
||||
def get_providers() -> list[OAuthProvider]:
|
||||
"""Return all configured OAuth providers."""
|
||||
from app.config import settings
|
||||
|
||||
providers: list[OAuthProvider] = []
|
||||
|
||||
gitea = GiteaProvider(
|
||||
base_url=settings.gitea_url,
|
||||
client_id=settings.gitea_oauth_client_id,
|
||||
client_secret=settings.gitea_oauth_client_secret,
|
||||
redirect_uri=settings.oauth_redirect_uri,
|
||||
)
|
||||
if gitea.is_enabled():
|
||||
providers.append(gitea)
|
||||
|
||||
github = GitHubProvider(
|
||||
client_id=settings.github_client_id or "",
|
||||
client_secret=settings.github_client_secret or "",
|
||||
redirect_uri=settings.oauth_redirect_uri or "",
|
||||
)
|
||||
if github.is_enabled():
|
||||
providers.append(github)
|
||||
|
||||
return providers
|
||||
|
||||
|
||||
def get_provider(name: str) -> OAuthProvider | None:
|
||||
"""Get a specific provider by name."""
|
||||
for p in get_providers():
|
||||
if p.name == name:
|
||||
return p
|
||||
return None
|
||||
Reference in New Issue
Block a user