-
-
Notifications
You must be signed in to change notification settings - Fork 6
Optional authentication token for API and Web UI #93
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,39 @@ | ||
| """ | ||
| Auth API routes for the optional shared auth token. | ||
| """ | ||
|
|
||
| from core.settings import settings | ||
| from litestar import Router, get | ||
|
|
||
| from ..guards.auth import require_auth | ||
|
|
||
|
|
||
| @get("/config", tags=["Auth"], summary="Auth Configuration") | ||
| async def get_auth_config() -> dict[str, bool]: | ||
| """ | ||
| Whether the API currently requires authentication. | ||
|
|
||
| Always public, since the Web UI needs to know whether to show the | ||
| login page before it has (or needs) a token of its own. | ||
|
|
||
| Returns: | ||
| dict: {"enabled": bool} | ||
| """ | ||
| return {"enabled": bool(settings.auth_token)} | ||
|
|
||
|
|
||
| @get("/verify", tags=["Auth"], summary="Verify Token", guards=[require_auth]) | ||
| async def verify_token() -> dict[str, bool]: | ||
| """ | ||
| Confirm a bearer token is valid. | ||
|
|
||
| Only reached if the guard already accepted the Authorization header; | ||
| a missing/invalid token never reaches this handler. | ||
|
|
||
| Returns: | ||
| dict: {"valid": true} | ||
| """ | ||
| return {"valid": True} | ||
|
|
||
|
|
||
| auth_router = Router(path="/auth", route_handlers=[get_auth_config, verify_token]) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,26 @@ | ||
| """ | ||
| Guard enforcing the optional shared auth token. | ||
| """ | ||
|
|
||
| import hmac | ||
|
|
||
| from core.domain.entities.exceptions import UnauthorizedError | ||
| from core.settings import settings | ||
| from litestar.connection import ASGIConnection | ||
| from litestar.handlers.base import BaseRouteHandler | ||
|
|
||
|
|
||
| def require_auth(connection: ASGIConnection, _: BaseRouteHandler) -> None: | ||
| """Reject the request unless it carries the configured auth token. | ||
|
|
||
| A no-op when SHELLY_AUTH_TOKEN is unset, preserving the zero-config | ||
| default of no authentication. | ||
| """ | ||
| token = settings.auth_token | ||
| if not token: | ||
| return | ||
|
|
||
| header = connection.headers.get("authorization", "") | ||
| presented = header.removeprefix("Bearer ") if header.startswith("Bearer ") else "" | ||
| if not presented or not hmac.compare_digest(presented, token): | ||
|
Owner
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. hmac.compare_digest on str raises TypeError when either side contains non-ASCII, so a garbage Authorization header becomes a logged 500 instead of a 401. Can we compare presented.encode() against token.encode()? |
||
| raise UnauthorizedError() | ||
|
Owner
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. Nothing in core raises this, bearer transport is purely an API concern, so I'd rather keep it out of the domain layer. If the guard raises Litestar's NotAuthorizedException instead, the existing HTTPException handler already produces the same envelope, and both the core exception and the new EXCEPTION_HANDLERS entry can go. |
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,80 @@ | ||
| import core.settings | ||
| from api.controllers.auth import auth_router | ||
| from api.presentation.handlers import EXCEPTION_HANDLERS | ||
| from litestar.testing import create_test_client | ||
|
|
||
|
|
||
| class TestAuthController: | ||
|
|
||
| def test_config_reports_disabled_when_no_token_is_set(self, monkeypatch): | ||
| monkeypatch.setattr(core.settings.settings, "auth_token", None) | ||
|
|
||
| with create_test_client(route_handlers=[auth_router]) as client: | ||
| response = client.get("/auth/config") | ||
|
|
||
| assert response.status_code == 200 | ||
| assert response.json() == {"enabled": False} | ||
|
|
||
| def test_config_reports_enabled_when_a_token_is_set(self, monkeypatch): | ||
| monkeypatch.setattr(core.settings.settings, "auth_token", "secret123") | ||
|
|
||
| with create_test_client(route_handlers=[auth_router]) as client: | ||
| response = client.get("/auth/config") | ||
|
|
||
| assert response.status_code == 200 | ||
| assert response.json() == {"enabled": True} | ||
|
|
||
| def test_config_needs_no_header_even_when_a_token_is_set(self, monkeypatch): | ||
| monkeypatch.setattr(core.settings.settings, "auth_token", "secret123") | ||
|
|
||
| with create_test_client(route_handlers=[auth_router]) as client: | ||
| response = client.get("/auth/config") | ||
|
|
||
| assert response.status_code == 200 | ||
|
|
||
| def test_verify_passes_through_when_auth_is_disabled(self, monkeypatch): | ||
| monkeypatch.setattr(core.settings.settings, "auth_token", None) | ||
|
|
||
| with create_test_client(route_handlers=[auth_router]) as client: | ||
| response = client.get("/auth/verify") | ||
|
|
||
| assert response.status_code == 200 | ||
| assert response.json() == {"valid": True} | ||
|
|
||
| def test_verify_rejects_a_missing_token(self, monkeypatch): | ||
| monkeypatch.setattr(core.settings.settings, "auth_token", "secret123") | ||
|
|
||
| with create_test_client( | ||
| route_handlers=[auth_router], exception_handlers=EXCEPTION_HANDLERS | ||
| ) as client: | ||
| response = client.get("/auth/verify") | ||
|
|
||
| assert response.status_code == 401 | ||
| assert response.json()["error"] == "Unauthorized" | ||
|
|
||
| def test_verify_rejects_the_wrong_token(self, monkeypatch): | ||
| monkeypatch.setattr(core.settings.settings, "auth_token", "secret123") | ||
|
|
||
| with create_test_client( | ||
| route_handlers=[auth_router], exception_handlers=EXCEPTION_HANDLERS | ||
| ) as client: | ||
| response = client.get( | ||
| "/auth/verify", headers={"Authorization": "Bearer wrong"} | ||
| ) | ||
|
|
||
| assert response.status_code == 401 | ||
|
|
||
| def test_verify_accepts_the_correct_token(self, monkeypatch): | ||
| monkeypatch.setattr(core.settings.settings, "auth_token", "secret123") | ||
|
|
||
| with create_test_client(route_handlers=[auth_router]) as client: | ||
| response = client.get( | ||
| "/auth/verify", headers={"Authorization": "Bearer secret123"} | ||
| ) | ||
|
|
||
| assert response.status_code == 200 | ||
| assert response.json() == {"valid": True} | ||
|
|
||
| def test_it_is_mounted_on_the_app(self, app): | ||
| assert any(route.path == "/api/auth/config" for route in app.routes) | ||
| assert any(route.path == "/api/auth/verify" for route in app.routes) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| import core.settings | ||
| import pytest | ||
| from api.guards.auth import require_auth | ||
| from core.domain.entities.exceptions import UnauthorizedError | ||
| from litestar.testing import RequestFactory | ||
|
|
||
|
|
||
| def _connection(authorization: str | None = None): | ||
| headers = {"Authorization": authorization} if authorization else {} | ||
| return RequestFactory().get("/", headers=headers) | ||
|
|
||
|
|
||
| class TestRequireAuth: | ||
|
|
||
| def test_it_allows_the_request_when_auth_is_disabled(self, monkeypatch): | ||
| monkeypatch.setattr(core.settings.settings, "auth_token", None) | ||
|
|
||
| require_auth(_connection(), None) | ||
|
|
||
| def test_it_rejects_a_missing_header_when_auth_is_enabled(self, monkeypatch): | ||
| monkeypatch.setattr(core.settings.settings, "auth_token", "secret123") | ||
|
|
||
| with pytest.raises(UnauthorizedError): | ||
| require_auth(_connection(), None) | ||
|
|
||
| def test_it_rejects_a_malformed_header(self, monkeypatch): | ||
| monkeypatch.setattr(core.settings.settings, "auth_token", "secret123") | ||
|
|
||
| with pytest.raises(UnauthorizedError): | ||
| require_auth(_connection(authorization="secret123"), None) | ||
|
|
||
| def test_it_rejects_the_wrong_token(self, monkeypatch): | ||
| monkeypatch.setattr(core.settings.settings, "auth_token", "secret123") | ||
|
|
||
| with pytest.raises(UnauthorizedError): | ||
| require_auth(_connection(authorization="Bearer wrong"), None) | ||
|
|
||
| def test_it_allows_the_correct_token(self, monkeypatch): | ||
| monkeypatch.setattr(core.settings.settings, "auth_token", "secret123") | ||
|
|
||
| require_auth(_connection(authorization="Bearer secret123"), None) |
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.
/docs and /docs/openapi.json also stay public, can we list them here? Worth also recommending a long random token, since there's no rate limiting on /auth/verify.