diff --git a/app/auth/providers.py b/app/auth/providers.py new file mode 100644 index 0000000..9cd457d --- /dev/null +++ b/app/auth/providers.py @@ -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