diff --git a/hbd/server/oauth.py b/hbd/server/oauth.py index 99427b1..aabb0a9 100644 --- a/hbd/server/oauth.py +++ b/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", ""), + } diff --git a/tests/test_oauth.py b/tests/test_oauth.py index a43d88c..b4e4e6e 100644 --- a/tests/test_oauth.py +++ b/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", + }