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>
This commit is contained in:
2026-05-08 13:53:57 -04:00
parent 940d0af35e
commit b06de6fdd3
2 changed files with 64 additions and 4 deletions
+1 -4
View File
@@ -35,10 +35,6 @@ def _render_template(html_str: str, **context) -> str:
SESSION_COOKIE = "hbd_session" SESSION_COOKIE = "hbd_session"
def _gitea_cfg_url(config: dict) -> str:
return config.get("oauth", {}).get("gitea", {}).get("url", "")
def _get_token(request) -> str: def _get_token(request) -> str:
"""Extract session token from Bearer header, X-Auth-Token header, or cookie.""" """Extract session token from Bearer header, X-Auth-Token header, or cookie."""
auth = request.headers.get("Authorization", "") auth = request.headers.get("Authorization", "")
@@ -935,6 +931,7 @@ async def start(
if not code or not state: if not code or not state:
return web.Response(status=400, text="Missing code or state") return web.Response(status=400, text="Missing code or state")
if not oauth_mod.validate_state(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") raise web.HTTPFound("/login?error=1")
redirect_uri = f"{request.url.origin()}/login/oauth/gitea/callback" redirect_uri = f"{request.url.origin()}/login/oauth/gitea/callback"
try: try:
+63
View File
@@ -259,3 +259,66 @@ async def test_fetch_user_raises_on_error_status():
)): )):
with pytest.raises(oauth.OAuthError): with pytest.raises(oauth.OAuthError):
await oauth.fetch_user(CFG_ON, "tok123") 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