-
-
Notifications
You must be signed in to change notification settings - Fork 132
Require API authentication when reachable over the network #110
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| """ | ||
| API Authentication | ||
| ------------------- | ||
| Guards the API behind a shared secret whenever the server is reachable over | ||
| the network. | ||
|
|
||
| Threat model: after a user has OAuth'd their Gmail account, the action and | ||
| status endpoints can read and destroy mail using the stored credentials. They | ||
| must therefore not be callable by anyone who can merely reach the port. | ||
|
|
||
| Policy: | ||
| - A loopback-only bind (the default) is treated as trusted single-user use and | ||
| needs no token. | ||
| - Any non-loopback bind requires a token. If ``API_TOKEN`` is not set, one is | ||
| generated at startup so the API is never left open by accident. | ||
|
|
||
| The same-origin web UI authenticates automatically: the root page sets the | ||
| token as a cookie, which the browser then sends with every ``/api`` request. | ||
| """ | ||
|
|
||
| import logging | ||
| import secrets | ||
|
|
||
| from fastapi import HTTPException, Request, status | ||
|
|
||
| from app.core import settings | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| # Token enforced for this process. Resolved once at startup; ``None`` means the | ||
| # server is loopback-only and auth is disabled. | ||
| _effective_token: str | None = None | ||
| # True when the active token was generated rather than supplied via API_TOKEN. | ||
| _token_was_generated = False | ||
|
|
||
|
|
||
| def resolve_effective_token() -> str | None: | ||
| """Decide whether API auth is required and which token enforces it. | ||
|
|
||
| Returns the active token (or ``None`` when auth is disabled). | ||
| """ | ||
| global _effective_token, _token_was_generated | ||
|
|
||
| if settings.api_token: | ||
| _effective_token = settings.api_token | ||
| _token_was_generated = False | ||
| elif not settings.is_loopback_host(): | ||
| # Reachable over the network but no token configured: refuse to run | ||
| # open. Mint a random one so the operator gets a working, secured app. | ||
| _effective_token = secrets.token_urlsafe(32) | ||
| _token_was_generated = True | ||
| else: | ||
| _effective_token = None | ||
| _token_was_generated = False | ||
|
|
||
| return _effective_token | ||
|
|
||
|
|
||
| def get_effective_token() -> str | None: | ||
| """Return the token currently enforced (``None`` if auth is disabled).""" | ||
| return _effective_token | ||
|
|
||
|
|
||
| def token_was_generated() -> bool: | ||
| """Return True if the active token was auto-generated at startup.""" | ||
| return _token_was_generated | ||
|
|
||
|
|
||
| def _extract_token(request: Request) -> str | None: | ||
| """Pull a presented token from the Authorization header or a cookie.""" | ||
| auth_header = request.headers.get("Authorization", "") | ||
| if auth_header.startswith("Bearer "): | ||
| return auth_header[len("Bearer "):].strip() | ||
|
|
||
| explicit = request.headers.get("X-API-Token") | ||
| if explicit: | ||
| return explicit.strip() | ||
|
|
||
| return request.cookies.get("api_token") | ||
|
|
||
|
|
||
| async def require_api_auth(request: Request) -> None: | ||
| """FastAPI dependency enforcing the API token when one is in effect.""" | ||
| token = _effective_token | ||
| if not token: | ||
| # Loopback-only deployment: not network-reachable, no token required. | ||
| return | ||
|
|
||
| presented = _extract_token(request) | ||
| if not presented or not secrets.compare_digest(presented, token): | ||
| raise HTTPException( | ||
| status_code=status.HTTP_401_UNAUTHORIZED, | ||
| detail="Missing or invalid API token", | ||
| headers={"WWW-Authenticate": "Bearer"}, | ||
| ) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -9,12 +9,17 @@ | |
| import time | ||
| from contextlib import asynccontextmanager | ||
|
|
||
| from fastapi import FastAPI, Request | ||
| from fastapi import Depends, FastAPI, Request | ||
| from fastapi.staticfiles import StaticFiles | ||
| from fastapi.templating import Jinja2Templates | ||
|
|
||
| from app.core import settings | ||
| from app.api import status_router, actions_router | ||
| from app.api.auth import ( | ||
| get_effective_token, | ||
| require_api_auth, | ||
| resolve_effective_token, | ||
| ) | ||
|
|
||
| templates = Jinja2Templates(directory="templates") | ||
|
|
||
|
|
@@ -125,6 +130,10 @@ async def lifespan(app: FastAPI): | |
| def create_app() -> FastAPI: | ||
| """Application factory - creates and configures the FastAPI app.""" | ||
|
|
||
| # Decide whether the API must require a token (non-loopback bind) and, if | ||
| # so, which token enforces it. Done before routers are mounted. | ||
| resolve_effective_token() | ||
|
|
||
| app = FastAPI( | ||
| title=settings.app_name, | ||
| description="Bulk unsubscribe and email management tool for Gmail", | ||
|
|
@@ -137,19 +146,34 @@ def create_app() -> FastAPI: | |
| # Mount static files | ||
| app.mount("/static", StaticFiles(directory="static"), name="static") | ||
|
|
||
| # Include API routers | ||
| app.include_router(status_router) | ||
| app.include_router(actions_router) | ||
| # Include API routers — every /api endpoint is guarded by the token check. | ||
| # When the server is loopback-only the dependency is a no-op, preserving the | ||
| # local single-user experience. | ||
| app.include_router(status_router, dependencies=[Depends(require_api_auth)]) | ||
| app.include_router(actions_router, dependencies=[Depends(require_api_auth)]) | ||
|
|
||
| # HTML routes | ||
| @app.get("/", include_in_schema=False) | ||
| async def root(request: Request): | ||
| """Serve the main HTML page.""" | ||
| return templates.TemplateResponse( | ||
| response = templates.TemplateResponse( | ||
| request, | ||
| "index.html", | ||
| {"cache_bust": STARTUP_CACHE_BUST, "version": settings.app_version}, | ||
| ) | ||
| # Hand the same-origin browser its API token via a cookie so the UI | ||
| # keeps working without any client-side changes. Sent only when auth is | ||
| # actually in effect. | ||
| token = get_effective_token() | ||
| if token: | ||
| response.set_cookie( | ||
| "api_token", | ||
| token, | ||
| httponly=True, | ||
| samesite="strict", | ||
| secure=request.url.scheme == "https", | ||
| ) | ||
|
Comment on lines
+159
to
+175
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Anonymous This hands the shared token to anyone who can load the root page. Because Only set this cookie after a real UI authentication step, or gate 🤖 Prompt for AI Agents |
||
| return response | ||
|
|
||
| return app | ||
|
|
||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🧩 Analysis chain
🏁 Script executed:
Repository: Gururagavendra/gmail-cleaner
Length of output: 16494
🏁 Script executed:
Repository: Gururagavendra/gmail-cleaner
Length of output: 6591
🌐 Web query:
How does Uvicorn interpret host="" (empty string) passed to uvicorn.run(host=...) bind address?💡 Result:
Uvicorn does not special-case an empty string host. If you pass host="" to uvicorn.run(host=...), Uvicorn will pass that value down to its socket-binding logic, which ultimately binds the listening socket(s) according to Python/asyncio semantics for host="". Concretely, Uvicorn’s config stores the host value and its bind_socket calls sock.bind((self.host, self.port)) (for the normal TCP case), without checking whether self.host is empty. Therefore, host="" is treated as “bind to all interfaces” by the underlying Python networking/asyncio behavior rather than being converted to 0.0.0.0/:: inside Uvicorn.[1][2] Notes/implications: 1) The “all interfaces” behavior comes from host="" meaning “bind to all interfaces” in the underlying socket/create_server implementation (and is consistent with asyncio’s documented behavior); some asyncio edge cases exist when host is provided as a sequence containing only "" (e.g., binding issues), but that’s about passing [''] to asyncio rather than passing "" directly to Uvicorn.[3] 2) Uvicorn’s own documented default host is 127.0.0.1, not empty string; use host="0.0.0.0" explicitly if you want an unambiguous IPv4-only bind, or host="::" for IPv6.[4][5] If your goal is “listen on all IPv4 interfaces” reliably, prefer uvicorn.run(..., host="0.0.0.0",...); if you want “all interfaces” (v4+v6), host="" (or None) will generally bind accordingly, but the exact address family behavior can depend on the underlying platform/asyncio behavior.[4][1][3]
Citations:
Don’t treat blank
HOST("") as loopback—this can disable API auth while binding to all interfaces.app/core/config.py’sis_loopback_host()returnsTruefor"", which makesresolve_effective_token()disable auth; thenmain.pypassessettings.hostdirectly intouvicorn.run(host=host). Uvicorn doesn’t special-casehost="", so it binds as “all interfaces” at the socket layer—re-exposing the API without a token.🛡️ Suggested hardening
🤖 Prompt for AI Agents