feat: add authorization_url, exchange_code, fetch_user to oauth module

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

hbd/server/oauth.py

@@ -17,6 +17,9 @@ Set the redirect URI to:
 import logging
 import secrets
 import time
+import urllib.parse
+
+import aiohttp
 
 logger = logging.getLogger(__name__)
 
@@ -63,3 +66,71 @@ 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"))
+
+
+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)
+    params = urllib.parse.urlencode({
+        "client_id": g["client_id"],
+        "redirect_uri": redirect_uri,
+        "response_type": "code",
+        "scope": "user:email",
+        "state": state,
+    })
+    return f"{g['url'].rstrip('/')}/login/oauth/authorize?{params}"
+
+
+async def exchange_code(config: dict, code: str, redirect_uri: str) -> str:
+    """Exchange an authorization *code* for a Gitea access token.
+
+    Returns the access token string.  Raises OAuthError on any failure.
+    """
+    g = _gitea_cfg(config)
+    url = f"{g['url'].rstrip('/')}/login/oauth/access_token"
+    payload = {
+        "client_id": g["client_id"],
+        "client_secret": g["client_secret"],
+        "code": code,
+        "grant_type": "authorization_code",
+        "redirect_uri": redirect_uri,
+    }
+    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:
+                if resp.status != 200:
+                    text = await resp.text()
+                    raise OAuthError(f"Token exchange failed ({resp.status}): {text}")
+                data = await resp.json()
+    except aiohttp.ClientError as exc:
+        raise OAuthError(f"Token exchange network error: {exc}") from exc
+    token = data.get("access_token")
+    if not token:
+        raise OAuthError(f"No access_token in response: {data}")
+    return token
+
+
+async def fetch_user(config: dict, token: str) -> dict:
+    """Fetch the authenticated user's profile from Gitea.
+
+    Returns a dict with keys: login, full_name, avatar_url.
+    Raises OAuthError on any failure.
+    """
+    g = _gitea_cfg(config)
+    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:
+                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
+    return {
+        "login": data.get("login", ""),
+        "full_name": data.get("full_name", ""),
+        "avatar_url": data.get("avatar_url", ""),
+    }

tests/test_oauth.py

@@ -130,3 +130,93 @@ def test_provision_oauth_user_survives_config_reload():
     # Reload with empty config — OAuth user should survive
     users_mod.load_users({})
     assert "oauthonly" in users_mod.users
+
+
+from unittest.mock import AsyncMock, MagicMock, patch
+from urllib.parse import urlparse, parse_qs
+
+
+def test_authorization_url_shape():
+    state = "teststate"
+    redirect_uri = "https://hbd.example.com/login/oauth/gitea/callback"
+    url = oauth.authorization_url(CFG_ON, state, redirect_uri)
+    parsed = urlparse(url)
+    qs = parse_qs(parsed.query)
+    assert parsed.scheme == "https"
+    assert parsed.netloc == "git.example.com"
+    assert parsed.path == "/login/oauth/authorize"
+    assert qs["client_id"] == ["cid"]
+    assert qs["state"] == ["teststate"]
+    assert qs["redirect_uri"] == [redirect_uri]
+    assert qs["scope"] == ["user:email"]
+    assert qs["response_type"] == ["code"]
+
+
+@pytest.mark.asyncio
+async def test_exchange_code_returns_token():
+    redirect_uri = "https://hbd.example.com/login/oauth/gitea/callback"
+    mock_response = AsyncMock()
+    mock_response.status = 200
+    mock_response.json = AsyncMock(return_value={"access_token": "tok123"})
+
+    mock_session = MagicMock()
+    mock_session.post = MagicMock(return_value=AsyncMock(
+        __aenter__=AsyncMock(return_value=mock_response),
+        __aexit__=AsyncMock(return_value=False),
+    ))
+
+    with patch("hbd.server.oauth.aiohttp.ClientSession", return_value=AsyncMock(
+        __aenter__=AsyncMock(return_value=mock_session),
+        __aexit__=AsyncMock(return_value=False),
+    )):
+        token = await oauth.exchange_code(CFG_ON, "mycode", redirect_uri)
+    assert token == "tok123"
+
+
+@pytest.mark.asyncio
+async def test_exchange_code_raises_on_error_status():
+    redirect_uri = "https://hbd.example.com/login/oauth/gitea/callback"
+    mock_response = AsyncMock()
+    mock_response.status = 401
+    mock_response.text = AsyncMock(return_value="unauthorized")
+
+    mock_session = MagicMock()
+    mock_session.post = MagicMock(return_value=AsyncMock(
+        __aenter__=AsyncMock(return_value=mock_response),
+        __aexit__=AsyncMock(return_value=False),
+    ))
+
+    with patch("hbd.server.oauth.aiohttp.ClientSession", return_value=AsyncMock(
+        __aenter__=AsyncMock(return_value=mock_session),
+        __aexit__=AsyncMock(return_value=False),
+    )):
+        with pytest.raises(oauth.OAuthError):
+            await oauth.exchange_code(CFG_ON, "badcode", redirect_uri)
+
+
+@pytest.mark.asyncio
+async def test_fetch_user_returns_profile():
+    mock_response = AsyncMock()
+    mock_response.status = 200
+    mock_response.json = AsyncMock(return_value={
+        "login": "alice",
+        "full_name": "Alice Smith",
+        "avatar_url": "https://git.example.com/avatars/alice.png",
+    })
+
+    mock_session = MagicMock()
+    mock_session.get = MagicMock(return_value=AsyncMock(
+        __aenter__=AsyncMock(return_value=mock_response),
+        __aexit__=AsyncMock(return_value=False),
+    ))
+
+    with patch("hbd.server.oauth.aiohttp.ClientSession", return_value=AsyncMock(
+        __aenter__=AsyncMock(return_value=mock_session),
+        __aexit__=AsyncMock(return_value=False),
+    )):
+        profile = await oauth.fetch_user(CFG_ON, "tok123")
+    assert profile == {
+        "login": "alice",
+        "full_name": "Alice Smith",
+        "avatar_url": "https://git.example.com/avatars/alice.png",
+    }