andreas/heartbeat
Public Access
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 44 Wiki Activity

feat: generic build_auth_url/exchange_code/fetch_user for multi-provider OAuth2

Browse Source 
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Andreas Wrede
2026-05-09 08:38:11 -04:00
parent f24500a6b5
commit 87aeec5999
2 changed files with 325 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hbd/server/oauth.py
+56 -44
View File
@@ -1,17 +1,30 @@
"""Gitea OAuth2 support.
"""OAuth2 provider support.
Config shape (in ~/.hb.yaml):
oauth:
gitea:
url: https://git.example.com
my-gitea: # route slug → /login/oauth/my-gitea
type: gitea # "gitea" | "github" | "nextcloud"
# omit type to default to "gitea"
url: https://git.example.com # required for gitea and nextcloud
client_id: <client-id>
client_secret: <client-secret>
label: "Work Gitea" # optional display name on login button
logo: https://example.com/logo.png # optional logo URL
github:
type: github
client_id: <client-id>
client_secret: <client-secret>
Register a Gitea OAuth2 application at:
Gitea → Settings → Applications → OAuth2
Set the redirect URI to:
https://<hbd-host>/login/oauth/gitea/callback
nextcloud:
type: nextcloud
url: https://cloud.example.com
client_id: <client-id>
client_secret: <client-secret>
Register the OAuth app with each provider and set the redirect URI to:
https://<hbd-host>/login/oauth/<name>/callback
"""
import logging
@@ -155,44 +168,32 @@ def get_providers(config: dict) -> list[ResolvedProvider]:
return result
def _gitea_cfg(config: dict) -> dict:
"""Return the gitea sub-dict or {} if absent/incomplete."""
return config.get("oauth", {}).get("gitea", {})
def is_enabled(config: dict) -> bool:
"""Return True when all three required Gitea OAuth keys are present."""
g = _gitea_cfg(config)
return bool(g.get("url") and g.get("client_id") and g.get("client_secret"))
"""Return True when at least one OAuth provider is fully configured."""
return bool(get_providers(config))
def authorization_url(config: dict, state: str, redirect_uri: str) -> str:
"""Return the Gitea OAuth2 authorization URL to redirect the browser to."""
g = _gitea_cfg(config)
if not (g.get("url") and g.get("client_id") and g.get("client_secret")):
raise OAuthError("Gitea OAuth2 is not configured")
params = urllib.parse.urlencode({
"client_id": g["client_id"],
def build_auth_url(provider: ResolvedProvider, state: str, redirect_uri: str) -> str:
"""Return the provider's OAuth2 authorization URL to redirect the browser to."""
params: dict = {
"client_id": provider.client_id,
"redirect_uri": redirect_uri,
"response_type": "code",
"scope": "user:email",
"state": state,
})
return f"{g['url'].rstrip('/')}/login/oauth/authorize?{params}"
}
if provider.scope:
params["scope"] = provider.scope
return f"{provider.authorize_url}?{urllib.parse.urlencode(params)}"
async def exchange_code(config: dict, code: str, redirect_uri: str) -> str:
"""Exchange an authorization *code* for a Gitea access token.
async def exchange_code(provider: ResolvedProvider, code: str, redirect_uri: str) -> str:
"""Exchange an authorization *code* for an access token.
Returns the access token string. Raises OAuthError on any failure.
"""
g = _gitea_cfg(config)
if not (g.get("url") and g.get("client_id") and g.get("client_secret")):
raise OAuthError("Gitea OAuth2 is not configured")
url = f"{g['url'].rstrip('/')}/login/oauth/access_token"
payload = {
"client_id": g["client_id"],
"client_secret": g["client_secret"],
"client_id": provider.client_id,
"client_secret": provider.client_secret,
"code": code,
"grant_type": "authorization_code",
"redirect_uri": redirect_uri,
@@ -200,7 +201,11 @@ async def exchange_code(config: dict, code: str, redirect_uri: str) -> str:
timeout = aiohttp.ClientTimeout(total=10)
try:
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(url, json=payload, headers={"Accept": "application/json"}) as resp:
async with session.post(
provider.token_url,
json=payload,
headers={"Accept": "application/json"},
) as resp:
if resp.status != 200:
text = await resp.text()
raise OAuthError(f"Token exchange failed ({resp.status}): {text}")
@@ -213,28 +218,35 @@ async def exchange_code(config: dict, code: str, redirect_uri: str) -> str:
return token
async def fetch_user(config: dict, token: str) -> dict:
"""Fetch the authenticated user's profile from Gitea.
async def fetch_user(provider: ResolvedProvider, token: str) -> dict:
"""Fetch the authenticated user's profile from the provider.
Returns a dict with keys: login, full_name, avatar_url.
Raises OAuthError on any failure.
"""
g = _gitea_cfg(config)
if not (g.get("url") and g.get("client_id") and g.get("client_secret")):
raise OAuthError("Gitea OAuth2 is not configured")
url = f"{g['url'].rstrip('/')}/api/v1/user"
timeout = aiohttp.ClientTimeout(total=10)
try:
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(url, headers={"Authorization": f"token {token}"}) as resp:
async with session.get(
provider.profile_url,
headers={
"Authorization": f"Bearer {token}",
"Accept": "application/json",
},
) as resp:
if resp.status != 200:
text = await resp.text()
raise OAuthError(f"User fetch failed ({resp.status}): {text}")
data = await resp.json()
except aiohttp.ClientError as exc:
raise OAuthError(f"User fetch network error: {exc}") from exc
for key in provider.profile_data_path:
data = data.get(key, {})
avatar_field = provider.field_map.get("avatar")
return {
"login": data.get("login", ""),
"full_name": data.get("full_name", ""),
"avatar_url": data.get("avatar_url", ""),
"login": data.get(provider.field_map["username"], ""),
"full_name": data.get(provider.field_map["full_name"], ""),
"avatar_url": data.get(avatar_field, "") if avatar_field else "",
}