diff --git a/app/auth.py b/app/auth.py new file mode 100644 index 00000000..88cfd426 --- /dev/null +++ b/app/auth.py @@ -0,0 +1,188 @@ +from __future__ import annotations + +import hashlib +import hmac +import time +from dataclasses import dataclass + +from fastapi import HTTPException, Request, Response + +from app.config import Settings + +USER_COOKIE_NAME = "mrwk_user" +ADMIN_COOKIE_NAME = "mrwk_admin" +OAUTH_STATE_COOKIE_NAME = "mrwk_oauth_state" +USER_COOKIE_MAX_AGE_SECONDS = 604_800 +ADMIN_COOKIE_MAX_AGE_SECONDS = 86_400 +OAUTH_STATE_MAX_AGE_SECONDS = 600 +CSRF_MAX_AGE_SECONDS = 3_600 + + +def oauth_configured(settings: Settings) -> bool: + return bool( + settings.github_oauth_client_id + and settings.github_oauth_client_secret + and settings.cookie_secret + ) + + +def safe_next_path(next_path: str | None) -> str: + if ( + not next_path + or not next_path.startswith("/") + or next_path.startswith("//") + or len(next_path) > 2048 + or "\\" in next_path + or any(ord(char) < 32 or 127 <= ord(char) < 160 for char in next_path) + ): + return "/me" + return next_path + + +def signed_value(value: str, secret: str) -> str: + timestamp = str(int(time.time())) + body = f"{value}|{timestamp}" + signature = hmac.new(secret.encode(), body.encode(), hashlib.sha256).hexdigest() + return f"{body}|{signature}" + + +def verified_value(token: str | None, secret: str, max_age_seconds: int) -> str | None: + if not token or not secret: + return None + try: + value, timestamp, signature = token.rsplit("|", 2) + age = int(time.time()) - int(timestamp) + except ValueError: + return None + if age < 0 or age > max_age_seconds: + return None + expected = hmac.new( + secret.encode(), f"{value}|{timestamp}".encode(), hashlib.sha256 + ).hexdigest() + if not hmac.compare_digest(signature, expected): + return None + return value + + +def csrf_token(action: str, login: str, secret: str) -> str: + return signed_value(f"{action}:{login}", secret) + + +def verify_csrf_token( + token: str | None, + *, + action: str, + login: str, + secret: str, + max_age_seconds: int = CSRF_MAX_AGE_SECONDS, +) -> bool: + expected = f"{action}:{login}" + return verified_value(token, secret, max_age_seconds) == expected + + +@dataclass(frozen=True) +class AuthContext: + settings: Settings + + def oauth_configured(self) -> bool: + return oauth_configured(self.settings) + + def signed_value(self, value: str) -> str: + return signed_value(value, self.settings.cookie_secret) + + def github_login_from_request(self, request: Request) -> str | None: + login = verified_value( + request.cookies.get(USER_COOKIE_NAME), + self.settings.cookie_secret, + USER_COOKIE_MAX_AGE_SECONDS, + ) + return login.lower() if login else None + + def admin_login_from_request(self, request: Request) -> str | None: + token = request.headers.get("x-mergework-admin-token", "") + if self.settings.admin_token and hmac.compare_digest(token, self.settings.admin_token): + return "api-token" + login = verified_value( + request.cookies.get(ADMIN_COOKIE_NAME), + self.settings.cookie_secret, + ADMIN_COOKIE_MAX_AGE_SECONDS, + ) + if login and login.lower() in self.settings.admin_logins: + return login.lower() + return None + + def require_github_login(self, request: Request) -> str: + login = self.github_login_from_request(request) + if login is None: + raise HTTPException(status_code=401, detail="github login required") + return login + + def require_admin(self, request: Request) -> str: + login = self.admin_login_from_request(request) + if login is None: + raise HTTPException(status_code=401, detail="admin authentication required") + return login + + def require_admin_token(self, request: Request) -> str: + token = request.headers.get("x-mergework-admin-token", "") + if self.settings.admin_token and hmac.compare_digest(token, self.settings.admin_token): + return "api-token" + raise HTTPException(status_code=401, detail="admin token required") + + def csrf_token(self, action: str, login: str) -> str: + return csrf_token(action, login, self.settings.cookie_secret) + + def verify_csrf_token(self, token: str | None, *, action: str, login: str) -> bool: + return verify_csrf_token( + token, + action=action, + login=login, + secret=self.settings.cookie_secret, + ) + + def verify_oauth_state(self, cookie_state: str | None, state: str) -> str: + if not cookie_state or not hmac.compare_digest(cookie_state, state): + raise HTTPException(status_code=401, detail="invalid OAuth state") + state_value = verified_value( + state, self.settings.cookie_secret, OAUTH_STATE_MAX_AGE_SECONDS + ) + if state_value is None: + raise HTTPException(status_code=401, detail="expired OAuth state") + try: + _, next_path = state_value.split(",", 1) + except ValueError as exc: + raise HTTPException(status_code=401, detail="invalid OAuth state") from exc + return safe_next_path(next_path) + + def set_oauth_state_cookie(self, response: Response, state: str) -> None: + response.set_cookie( + OAUTH_STATE_COOKIE_NAME, + state, + httponly=True, + secure=True, + samesite="lax", + max_age=OAUTH_STATE_MAX_AGE_SECONDS, + ) + + def set_login_cookies(self, response: Response, login: str) -> None: + response.set_cookie( + USER_COOKIE_NAME, + self.signed_value(login), + httponly=True, + secure=True, + samesite="lax", + max_age=USER_COOKIE_MAX_AGE_SECONDS, + ) + if login in self.settings.admin_logins: + response.set_cookie( + ADMIN_COOKIE_NAME, + self.signed_value(login), + httponly=True, + secure=True, + samesite="lax", + max_age=ADMIN_COOKIE_MAX_AGE_SECONDS, + ) + + def clear_session_cookies(self, response: Response) -> None: + response.delete_cookie(USER_COOKIE_NAME) + response.delete_cookie(ADMIN_COOKIE_NAME) diff --git a/app/main.py b/app/main.py index 034b80f0..8c8cefff 100644 --- a/app/main.py +++ b/app/main.py @@ -1,11 +1,8 @@ from __future__ import annotations -import hashlib -import hmac import json import re import secrets -import time from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Annotated, Any @@ -20,7 +17,8 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session -from app.config import Settings, get_settings +from app.auth import AuthContext, safe_next_path +from app.config import get_settings from app.db import create_schema, session_scope from app.ledger.reconciliation import payout_reconciliation_summary, reconcile_accepted_payouts from app.ledger.service import ( @@ -264,27 +262,6 @@ def _proof_hashes_by_sequence(session: Session, sequences: list[int]) -> dict[in return {int(sequence): str(proof_hash) for sequence, proof_hash in rows} -def _oauth_configured(settings: Settings) -> bool: - return bool( - settings.github_oauth_client_id - and settings.github_oauth_client_secret - and settings.cookie_secret - ) - - -def _safe_next_path(next_path: str | None) -> str: - if ( - not next_path - or not next_path.startswith("/") - or next_path.startswith("//") - or len(next_path) > 2048 - or "\\" in next_path - or any(ord(char) < 32 or 127 <= ord(char) < 160 for char in next_path) - ): - return "/me" - return next_path - - def _normalized_account(account: str) -> str: if not account or not account.strip(): raise HTTPException(status_code=400, detail="account must not be empty") @@ -363,31 +340,6 @@ def _proof_hash_from_path(proof_hash: str) -> str: return clean -def _signed_value(value: str, secret: str) -> str: - timestamp = str(int(time.time())) - body = f"{value}|{timestamp}" - signature = hmac.new(secret.encode(), body.encode(), hashlib.sha256).hexdigest() - return f"{body}|{signature}" - - -def _verified_value(token: str | None, secret: str, max_age_seconds: int) -> str | None: - if not token or not secret: - return None - try: - value, timestamp, signature = token.rsplit("|", 2) - age = int(time.time()) - int(timestamp) - except ValueError: - return None - if age < 0 or age > max_age_seconds: - return None - expected = hmac.new( - secret.encode(), f"{value}|{timestamp}".encode(), hashlib.sha256 - ).hexdigest() - if not hmac.compare_digest(signature, expected): - return None - return value - - async def _json_object(request: Request) -> dict[str, Any]: try: data = await request.json() @@ -445,19 +397,9 @@ def _optional_int(data: dict[str, Any], field: str, default: int) -> int: return _parse_int(value, field) -def _csrf_token(action: str, login: str, secret: str) -> str: - return _signed_value(f"{action}:{login}", secret) - - -def _verify_csrf_token( - token: str | None, *, action: str, login: str, secret: str, max_age_seconds: int = 3_600 -) -> bool: - expected = f"{action}:{login}" - return _verified_value(token, secret, max_age_seconds) == expected - - def create_app(database_url: str | None = None, webhook_secret: str | None = None) -> FastAPI: settings = get_settings() + auth = AuthContext(settings) db_url = database_url or settings.database_url secret = webhook_secret if webhook_secret is not None else settings.github_webhook_secret create_schema(db_url) @@ -509,37 +451,6 @@ async def add_security_headers(request: Request, call_next: Any) -> Any: if static_dir.exists(): app.mount("/static", StaticFiles(directory=str(static_dir)), name="static") - def admin_login_from_request(request: Request) -> str | None: - token = request.headers.get("x-mergework-admin-token", "") - if settings.admin_token and hmac.compare_digest(token, settings.admin_token): - return "api-token" - login = _verified_value(request.cookies.get("mrwk_admin"), settings.cookie_secret, 86_400) - if login and login.lower() in settings.admin_logins: - return login.lower() - return None - - def github_login_from_request(request: Request) -> str | None: - login = _verified_value(request.cookies.get("mrwk_user"), settings.cookie_secret, 604_800) - return login.lower() if login else None - - def require_github_login(request: Request) -> str: - login = github_login_from_request(request) - if login is None: - raise HTTPException(status_code=401, detail="github login required") - return login - - def require_admin(request: Request) -> str: - login = admin_login_from_request(request) - if login is None: - raise HTTPException(status_code=401, detail="admin authentication required") - return login - - def require_admin_token(request: Request) -> str: - token = request.headers.get("x-mergework-admin-token", "") - if settings.admin_token and hmac.compare_digest(token, settings.admin_token): - return "api-token" - raise HTTPException(status_code=401, detail="admin token required") - def attempt_submitter_account(data: dict[str, Any], github_login: str) -> str: submitter_account = f"github:{github_login}" if data.get("submitter_account") is None: @@ -623,7 +534,7 @@ def api_bounties_summary( def api_admin_webhook_events( status: str | None = Query(None), limit: Annotated[int, Query(ge=1, le=200)] = 50, - admin_login: str = Depends(require_admin_token), + admin_login: str = Depends(auth.require_admin_token), ) -> list[dict[str, Any]]: del admin_login normalized_status = status.strip().lower() if status is not None else None @@ -651,7 +562,7 @@ def api_admin_webhook_events( @app.post("/api/v1/bounties") async def api_create_bounty( - request: Request, admin_login: str = Depends(require_admin_token) + request: Request, admin_login: str = Depends(auth.require_admin_token) ) -> dict[str, Any]: data = await _json_object(request) with session_scope(db_url) as session: @@ -709,7 +620,7 @@ def api_bounty_attempts(bounty_id: int, include_expired: bool = Query(False)) -> async def api_create_bounty_attempt( bounty_id: int, request: Request, - github_login: str = Depends(require_github_login), + github_login: str = Depends(auth.require_github_login), ) -> JSONResponse: bounty_id = _positive_bounty_id(bounty_id) data = await _json_object(request) @@ -807,7 +718,7 @@ async def api_create_bounty_attempt( async def api_release_bounty_attempt( attempt_id: int, request: Request, - github_login: str = Depends(require_github_login), + github_login: str = Depends(auth.require_github_login), ) -> dict[str, Any]: if attempt_id <= 0: raise HTTPException(status_code=400, detail="attempt id must be positive") @@ -838,7 +749,7 @@ async def api_release_bounty_attempt( @app.get("/api/v1/reconciliation/payouts") def api_payout_reconciliation( - admin_login: str = Depends(require_admin_token), + admin_login: str = Depends(auth.require_admin_token), ) -> dict[str, Any]: with session_scope(db_url) as session: checks = reconcile_accepted_payouts(session) @@ -852,7 +763,7 @@ def api_payout_reconciliation( async def api_pay_bounty( bounty_id: int, request: Request, - admin_login: str = Depends(require_admin_token), + admin_login: str = Depends(auth.require_admin_token), ) -> Any: bounty_id = _positive_bounty_id(bounty_id) data = await _json_object(request) @@ -922,7 +833,7 @@ async def api_pay_bounty( async def api_close_bounty( bounty_id: int, request: Request, - admin_login: str = Depends(require_admin_token), + admin_login: str = Depends(auth.require_admin_token), ) -> dict[str, Any]: bounty_id = _positive_bounty_id(bounty_id) data = await _json_object(request) @@ -985,7 +896,7 @@ def api_account_accepted_work(account: str) -> dict[str, Any]: @app.get("/api/v1/auth/me") def api_auth_me(request: Request) -> dict[str, Any]: - login = github_login_from_request(request) + login = auth.github_login_from_request(request) return {"authenticated": login is not None, "github_login": login} @app.post("/api/v1/wallets/register") @@ -1021,7 +932,7 @@ def api_wallet(address: str) -> dict[str, Any]: @app.post("/api/v1/wallets/link-github") async def api_link_wallet_github( - request: Request, github_login: str = Depends(require_github_login) + request: Request, github_login: str = Depends(auth.require_github_login) ) -> dict[str, Any]: data = await _json_object(request) with session_scope(db_url) as session: @@ -1039,7 +950,7 @@ async def api_link_wallet_github( @app.post("/api/v1/github/claim") async def api_github_claim( - request: Request, github_login: str = Depends(require_github_login) + request: Request, github_login: str = Depends(auth.require_github_login) ) -> dict[str, Any]: data = await _json_object(request) with session_scope(db_url) as session: @@ -1284,11 +1195,11 @@ def docs_page(request: Request) -> HTMLResponse: @app.get("/auth/github/login") def auth_github_login(next_path: str | None = Query(None, alias="next")) -> RedirectResponse: - if not _oauth_configured(settings): + if not auth.oauth_configured(): raise HTTPException(status_code=503, detail="GitHub OAuth is not configured") - safe_next = _safe_next_path(next_path) + safe_next = safe_next_path(next_path) state_value = f"{secrets.token_urlsafe(24)},{safe_next}" - state = _signed_value(state_value, settings.cookie_secret) + state = auth.signed_value(state_value) query = urlencode( { "client_id": settings.github_oauth_client_id, @@ -1300,26 +1211,15 @@ def auth_github_login(next_path: str | None = Query(None, alias="next")) -> Redi response = RedirectResponse( f"https://github.com/login/oauth/authorize?{query}", status_code=302 ) - response.set_cookie( - "mrwk_oauth_state", state, httponly=True, secure=True, samesite="lax", max_age=600 - ) + auth.set_oauth_state_cookie(response, state) return response @app.get("/auth/github/callback") async def auth_github_callback(request: Request, code: str, state: str) -> RedirectResponse: - if not _oauth_configured(settings): + if not auth.oauth_configured(): raise HTTPException(status_code=503, detail="GitHub OAuth is not configured") cookie_state = request.cookies.get("mrwk_oauth_state") - if not cookie_state or not hmac.compare_digest(cookie_state, state): - raise HTTPException(status_code=401, detail="invalid OAuth state") - state_value = _verified_value(state, settings.cookie_secret, 600) - if state_value is None: - raise HTTPException(status_code=401, detail="expired OAuth state") - try: - _, next_path = state_value.split(",", 1) - except ValueError as exc: - raise HTTPException(status_code=401, detail="invalid OAuth state") from exc - next_path = _safe_next_path(next_path) + next_path = auth.verify_oauth_state(cookie_state, state) async with httpx.AsyncClient(timeout=10) as client: token_response = await client.post( "https://github.com/login/oauth/access_token", @@ -1347,23 +1247,7 @@ async def auth_github_callback(request: Request, code: str, state: str) -> Redir if not login: raise HTTPException(status_code=401, detail="GitHub OAuth user lookup failed") response = RedirectResponse(next_path, status_code=302) - response.set_cookie( - "mrwk_user", - _signed_value(login, settings.cookie_secret), - httponly=True, - secure=True, - samesite="lax", - max_age=604_800, - ) - if login in settings.admin_logins: - response.set_cookie( - "mrwk_admin", - _signed_value(login, settings.cookie_secret), - httponly=True, - secure=True, - samesite="lax", - max_age=86_400, - ) + auth.set_login_cookies(response, login) response.delete_cookie("mrwk_oauth_state") return response @@ -1379,13 +1263,12 @@ async def admin_callback(request: Request) -> RedirectResponse: @app.post("/auth/logout") def auth_logout() -> RedirectResponse: response = RedirectResponse("/", status_code=303) - response.delete_cookie("mrwk_user") - response.delete_cookie("mrwk_admin") + auth.clear_session_cookies(response) return response @app.get("/me", response_class=HTMLResponse) def me_page(request: Request) -> HTMLResponse: - login = github_login_from_request(request) + login = auth.github_login_from_request(request) github_balance_mrwk = "0" linked_wallet_address = "" if login: @@ -1407,8 +1290,7 @@ def me_page(request: Request) -> HTMLResponse: @app.post("/admin/logout") def admin_logout() -> RedirectResponse: response = RedirectResponse("/", status_code=303) - response.delete_cookie("mrwk_admin") - response.delete_cookie("mrwk_user") + auth.clear_session_cookies(response) return response @app.get("/admin", response_class=HTMLResponse) @@ -1417,9 +1299,9 @@ def admin_page( webhook_status: str | None = Query(None), webhook_limit: Annotated[int, Query(ge=1, le=100)] = 25, ) -> Any: - login = admin_login_from_request(request) + login = auth.admin_login_from_request(request) if login is None: - if _oauth_configured(settings): + if auth.oauth_configured(): return RedirectResponse("/auth/github/login?next=/admin", status_code=302) raise HTTPException(status_code=503, detail="GitHub OAuth is not configured") normalized_status = webhook_status.strip().lower() if webhook_status is not None else "" @@ -1437,7 +1319,7 @@ def admin_page( "admin.html", { "login": login, - "csrf_token": _csrf_token("admin-bounty", login, settings.cookie_secret), + "csrf_token": auth.csrf_token("admin-bounty", login), "webhook_events": webhook_events, "webhook_limit": webhook_limit, "webhook_limit_options": [10, 25, 50, 100], @@ -1456,14 +1338,13 @@ def admin_create_bounty( max_awards: int = Form(1), acceptance: str = Form(...), csrf_token: str | None = Form(None), - admin_login: str = Depends(require_admin), + admin_login: str = Depends(auth.require_admin), ) -> RedirectResponse: del request - if admin_login != "api-token" and not _verify_csrf_token( + if admin_login != "api-token" and not auth.verify_csrf_token( csrf_token, action="admin-bounty", login=admin_login, - secret=settings.cookie_secret, ): raise HTTPException(status_code=403, detail="invalid CSRF token") with session_scope(db_url) as session: diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 00000000..db052bac --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +import pytest +from fastapi import HTTPException, Request + +from app.auth import AuthContext, safe_next_path, signed_value +from app.config import Settings + + +def _settings() -> Settings: + return Settings( + database_url="sqlite:///./mergework.sqlite3", + public_base_url="https://mrwk.ltclab.site", + github_webhook_secret="webhook-secret", + github_oauth_client_id="client-id", + github_oauth_client_secret="client-secret", + admin_logins=("alice",), + admin_token="admin-token", + cookie_secret="cookie-secret", + github_accepted_labelers=("alice",), + ) + + +def _request(*, cookie: str = "", headers: list[tuple[bytes, bytes]] | None = None) -> Request: + raw_headers = list(headers or []) + if cookie: + raw_headers.append((b"cookie", cookie.encode())) + return Request({"type": "http", "method": "GET", "path": "/", "headers": raw_headers}) + + +def test_safe_next_path_keeps_only_internal_paths() -> None: + assert safe_next_path("/wallets?tab=linked") == "/wallets?tab=linked" + assert safe_next_path("https://evil.example/me") == "/me" + assert safe_next_path("//evil.example/me") == "/me" + assert safe_next_path("/admin\\evil") == "/me" + + +def test_auth_context_reads_signed_user_and_admin_cookies() -> None: + auth = AuthContext(_settings()) + user_cookie = signed_value("ALICE", "cookie-secret") + admin_cookie = signed_value("alice", "cookie-secret") + request = _request(cookie=f"mrwk_user={user_cookie}; mrwk_admin={admin_cookie}") + + assert auth.github_login_from_request(request) == "alice" + assert auth.admin_login_from_request(request) == "alice" + assert auth.require_github_login(request) == "alice" + assert auth.require_admin(request) == "alice" + + +def test_auth_context_keeps_admin_token_separate_from_cookie_admin() -> None: + auth = AuthContext(_settings()) + token_request = _request(headers=[(b"x-mergework-admin-token", b"admin-token")]) + user_cookie = signed_value("bob", "cookie-secret") + non_admin_cookie_request = _request(cookie=f"mrwk_admin={user_cookie}") + + assert auth.require_admin_token(token_request) == "api-token" + assert auth.admin_login_from_request(token_request) == "api-token" + assert auth.admin_login_from_request(non_admin_cookie_request) is None + with pytest.raises(HTTPException) as exc_info: + auth.require_admin(non_admin_cookie_request) + assert exc_info.value.status_code == 401 + + +def test_auth_context_verifies_oauth_state_and_next_path() -> None: + auth = AuthContext(_settings()) + valid_state = auth.signed_value("nonce,/wallets") + external_next_state = auth.signed_value("nonce,https://evil.example") + malformed_state = auth.signed_value("missing-comma") + + assert auth.verify_oauth_state(valid_state, valid_state) == "/wallets" + assert auth.verify_oauth_state(external_next_state, external_next_state) == "/me" + with pytest.raises(HTTPException) as mismatch: + auth.verify_oauth_state(valid_state, auth.signed_value("other,/wallets")) + assert mismatch.value.status_code == 401 + assert mismatch.value.detail == "invalid OAuth state" + with pytest.raises(HTTPException) as malformed: + auth.verify_oauth_state(malformed_state, malformed_state) + assert malformed.value.status_code == 401 + assert malformed.value.detail == "invalid OAuth state" + + +def test_csrf_uses_same_signed_cookie_boundary() -> None: + auth = AuthContext(_settings()) + token = auth.csrf_token("admin-bounty", "alice") + + assert auth.verify_csrf_token(token, action="admin-bounty", login="alice") + assert not auth.verify_csrf_token(token, action="admin-bounty", login="bob") diff --git a/tests/test_bounty_attempts.py b/tests/test_bounty_attempts.py index a649033c..5fbbc605 100644 --- a/tests/test_bounty_attempts.py +++ b/tests/test_bounty_attempts.py @@ -5,16 +5,17 @@ from fastapi.testclient import TestClient from sqlalchemy import select +from app.auth import signed_value from app.db import create_schema, session_scope from app.ledger.service import close_bounty, create_bounty, ensure_genesis, pay_bounty -from app.main import _signed_value, create_app +from app.main import create_app from app.models import BountyAttempt, LedgerEntry COOKIE_SECRET = "test-cookie-secret" def _set_login(client: TestClient, login: str) -> None: - client.cookies.set("mrwk_user", _signed_value(login, COOKIE_SECRET)) + client.cookies.set("mrwk_user", signed_value(login, COOKIE_SECRET)) def test_bounty_attempts_register_list_duplicate_and_release(sqlite_url: str, monkeypatch) -> None: diff --git a/tests/test_security.py b/tests/test_security.py index 23b5b5dd..8a6a0984 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -10,6 +10,7 @@ from sqlalchemy import select import app.ledger.service as ledger_service +from app.auth import safe_next_path, signed_value from app.db import create_schema, session_scope from app.ledger.service import ( TREASURY_ACCOUNT, @@ -26,7 +27,7 @@ register_wallet, validate_public_url, ) -from app.main import _safe_next_path, _signed_value, create_app +from app.main import create_app from app.models import Bounty, LedgerEntry, Proof, Submission, WebhookEvent from app.webhooks.github import handle_github_webhook @@ -111,7 +112,7 @@ def test_admin_bounty_form_requires_csrf_for_cookie_auth( create_app(database_url=sqlite_url, webhook_secret="secret"), base_url="https://testserver", ) - client.cookies.set("mrwk_admin", _signed_value("alice", "test-cookie-secret")) + client.cookies.set("mrwk_admin", signed_value("alice", "test-cookie-secret")) page = client.get("/admin") token_match = re.search(r'name="csrf_token" value="([^"]+)"', page.text) @@ -170,7 +171,7 @@ def test_admin_page_renders_safe_webhook_events_for_cookie_admin( ) unauthenticated = client.get("/admin", follow_redirects=False) - client.cookies.set("mrwk_admin", _signed_value("alice", "test-cookie-secret")) + client.cookies.set("mrwk_admin", signed_value("alice", "test-cookie-secret")) all_events = client.get("/admin?webhook_limit=10") filtered = client.get("/admin?webhook_status= missing_submitter &webhook_limit=10") limited = client.get("/admin?webhook_limit=1") @@ -203,7 +204,7 @@ def test_admin_bounty_api_requires_admin_token_not_cookie_auth( create_app(database_url=sqlite_url, webhook_secret="secret"), base_url="https://testserver", ) - client.cookies.set("mrwk_admin", _signed_value("alice", "test-cookie-secret")) + client.cookies.set("mrwk_admin", signed_value("alice", "test-cookie-secret")) cookie_only = client.post("/api/v1/bounties", json=_admin_bounty_form_data()) token_auth = client.post( @@ -424,7 +425,7 @@ def test_admin_payout_api_requires_admin_token_not_cookie_auth( bounty_id = bounty.id wallet = register_wallet(session, public_key_hex="11" * 32, label="Contributor") wallet_address = wallet.address - client.cookies.set("mrwk_admin", _signed_value("alice", "test-cookie-secret")) + client.cookies.set("mrwk_admin", signed_value("alice", "test-cookie-secret")) payload = { "to_account": wallet_address, @@ -833,7 +834,7 @@ def test_admin_bounty_api_accepts_multi_award_count( def test_oauth_next_path_rejects_external_or_headerlike_paths( next_path: str | None, expected: str ) -> None: - assert _safe_next_path(next_path) == expected + assert safe_next_path(next_path) == expected def test_amount_parser_rejects_non_finite_values() -> None: diff --git a/tests/test_wallet_api.py b/tests/test_wallet_api.py index 302956e3..9d47b4f0 100644 --- a/tests/test_wallet_api.py +++ b/tests/test_wallet_api.py @@ -8,6 +8,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from fastapi.testclient import TestClient +from app.auth import safe_next_path, signed_value, verified_value from app.db import create_schema, session_scope from app.ledger.service import ( TREASURY_ACCOUNT, @@ -19,7 +20,7 @@ wallet_claim_payload, wallet_link_payload, ) -from app.main import _safe_next_path, _signed_value, _verified_value, create_app +from app.main import create_app from app.wallets import address_from_public_key_hex, canonical_wallet_json @@ -332,7 +333,7 @@ def test_wallet_api_malformed_link_and_claim_requests_return_4xx( create_app(database_url=sqlite_url, webhook_secret="secret"), base_url="https://testserver", ) - client.cookies.set("mrwk_user", _signed_value("alice", "test-cookie-secret")) + client.cookies.set("mrwk_user", signed_value("alice", "test-cookie-secret")) client.post("/api/v1/wallets/register", json={"public_key_hex": public_hex}) missing_signature = client.post( @@ -391,7 +392,7 @@ def test_github_session_can_link_and_claim_wallet(sqlite_url: str, monkeypatch) create_app(database_url=sqlite_url, webhook_secret="secret"), base_url="https://testserver", ) - client.cookies.set("mrwk_user", _signed_value("alice", "test-cookie-secret")) + client.cookies.set("mrwk_user", signed_value("alice", "test-cookie-secret")) _register_wallet(client, public_hex) with session_scope(sqlite_url) as session: ensure_genesis(session) @@ -449,7 +450,7 @@ def test_github_login_redirects_when_oauth_is_configured(sqlite_url: str, monkey ("//evil.example/path", "/\\evil.example/path", "/me\nLocation:https://evil.example"), ) def test_oauth_next_path_rejects_redirect_ambiguity(next_path: str) -> None: - assert _safe_next_path(next_path) == "/me" + assert safe_next_path(next_path) == "/me" def test_github_login_stores_safe_default_for_backslash_next(sqlite_url: str, monkeypatch) -> None: @@ -463,7 +464,7 @@ def test_github_login_stores_safe_default_for_backslash_next(sqlite_url: str, mo assert response.status_code == 302 query = parse_qs(urlparse(response.headers["location"]).query) - state_value = _verified_value(query["state"][0], "test-cookie-secret", 600) + state_value = verified_value(query["state"][0], "test-cookie-secret", 600) assert state_value is not None _nonce, next_path = state_value.split(",", 1) assert next_path == "/me" @@ -521,7 +522,7 @@ def test_me_page_shows_signed_in_github_claim_balance(sqlite_url: str, monkeypat create_app(database_url=sqlite_url, webhook_secret="secret"), base_url="https://testserver", ) - client.cookies.set("mrwk_user", _signed_value("alice", "test-cookie-secret")) + client.cookies.set("mrwk_user", signed_value("alice", "test-cookie-secret")) me = client.get("/me").text @@ -536,7 +537,7 @@ def test_wallet_pages_do_not_require_manual_nonce(sqlite_url: str, monkeypatch) create_app(database_url=sqlite_url, webhook_secret="secret"), base_url="https://testserver", ) - client.cookies.set("mrwk_user", _signed_value("alice", "test-cookie-secret")) + client.cookies.set("mrwk_user", signed_value("alice", "test-cookie-secret")) transfer = client.get("/transfer").text me = client.get("/me").text @@ -615,7 +616,7 @@ def test_me_page_prefills_claim_address_for_linked_wallet(sqlite_url: str, monke create_app(database_url=sqlite_url, webhook_secret="secret"), base_url="https://testserver", ) - client.cookies.set("mrwk_user", _signed_value("alice", "test-cookie-secret")) + client.cookies.set("mrwk_user", signed_value("alice", "test-cookie-secret")) me = client.get("/me").text