From b06de6fdd309388cd67464e9a42a0e2ecf9e9674 Mon Sep 17 00:00:00 2001
From: Andreas Wrede <andreas@wrede.ca>
Date: Fri, 8 May 2026 13:53:57 -0400
Subject: [PATCH] fix: remove dead helper, add state logging, add
 integration-style oauth tests

- Remove unused `_gitea_cfg_url` module-level helper from http.py
- Add logger.warning on invalid/expired state in oauth_gitea_callback
- Add test_callback_invalid_state_rejects and test_full_oauth_flow_chain to tests/test_oauth.py (21 tests total, all passing)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 hbd/server/http.py  |  5 +---
 tests/test_oauth.py | 63 +++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 64 insertions(+), 4 deletions(-)

diff --git a/hbd/server/http.py b/hbd/server/http.py
index d22b38d..2102e9f 100644
--- a/hbd/server/http.py
+++ b/hbd/server/http.py
@@ -35,10 +35,6 @@ def _render_template(html_str: str, **context) -> str:
 SESSION_COOKIE = "hbd_session"
 
 
-def _gitea_cfg_url(config: dict) -> str:
-    return config.get("oauth", {}).get("gitea", {}).get("url", "")
-
-
 def _get_token(request) -> str:
     """Extract session token from Bearer header, X-Auth-Token header, or cookie."""
     auth = request.headers.get("Authorization", "")
@@ -935,6 +931,7 @@ async def start(
         if not code or not state:
             return web.Response(status=400, text="Missing code or state")
         if not oauth_mod.validate_state(state):
+            logger.warning("OAuth: invalid or expired state token from %s", request.remote)
             raise web.HTTPFound("/login?error=1")
         redirect_uri = f"{request.url.origin()}/login/oauth/gitea/callback"
         try:
diff --git a/tests/test_oauth.py b/tests/test_oauth.py
index 13ac90d..2767d25 100644
--- a/tests/test_oauth.py
+++ b/tests/test_oauth.py
@@ -259,3 +259,66 @@ async def test_fetch_user_raises_on_error_status():
     )):
         with pytest.raises(oauth.OAuthError):
             await oauth.fetch_user(CFG_ON, "tok123")
+
+
+# ---------------------------------------------------------------------------
+# Integration-style tests: callback logic chain
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_callback_invalid_state_rejects():
+    """Verify validate_state returns False for unknown state tokens."""
+    fake_state = "this-is-not-a-real-state"
+    assert oauth.validate_state(fake_state) is False
+
+
+@pytest.mark.asyncio
+async def test_full_oauth_flow_chain():
+    """Integration-style test: state → exchange → fetch → provision chain."""
+    redirect_uri = "https://hbd.example.com/login/oauth/gitea/callback"
+
+    # Step 1: create a state token
+    state = oauth.make_state()
+    assert oauth.validate_state(state) is True  # consumed; replay would return False
+
+    # Step 2: exchange code → token (mocked)
+    mock_token_response = AsyncMock()
+    mock_token_response.status = 200
+    mock_token_response.json = AsyncMock(return_value={"access_token": "flow_token"})
+
+    mock_user_response = AsyncMock()
+    mock_user_response.status = 200
+    mock_user_response.json = AsyncMock(return_value={
+        "login": "flowuser",
+        "full_name": "Flow User",
+        "avatar_url": "https://git.example.com/avatars/flow.png",
+    })
+
+    mock_session = MagicMock()
+    mock_session.post = MagicMock(return_value=AsyncMock(
+        __aenter__=AsyncMock(return_value=mock_token_response),
+        __aexit__=AsyncMock(return_value=False),
+    ))
+    mock_session.get = MagicMock(return_value=AsyncMock(
+        __aenter__=AsyncMock(return_value=mock_user_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, "authcode", redirect_uri)
+        profile = await oauth.fetch_user(CFG_ON, token)
+
+    assert token == "flow_token"
+    assert profile["login"] == "flowuser"
+
+    # Step 3: provision user
+    _reset_users()
+    user = users_mod.provision_oauth_user(
+        profile["login"], profile["full_name"], profile["avatar_url"]
+    )
+    assert user.username == "flowuser"
+    assert user.check_password("anything") is False