Merge pull request #79 from UpstreamDataInc/dev_api_tokens

.djlintrc

@@ -0,0 +1,3 @@
+{
+    "ignore": "H030,H031"
+}

conftest.py

@@ -49,6 +49,9 @@ async def async_client(test_app):
         response = await client.post("/login", data=login_data, follow_redirects=True)
         assert response.status_code == 200
 
+        data = response.json()
+        client.cookies.set("session_id", data["access_token"])
+
         yield client
 
 

goosebit/__init__.py

@@ -1,3 +1,4 @@
+import importlib.metadata
 from contextlib import asynccontextmanager
 from typing import Annotated
 
@@ -9,12 +10,7 @@
 
 from goosebit import api, db, realtime, ui, updater
 from goosebit.api.telemetry import metrics
-from goosebit.auth import (
-    authenticate_user,
-    auto_redirect,
-    create_session,
-    get_current_user,
-)
+from goosebit.auth import get_user_from_request, login_user, redirect_if_authenticated
 from goosebit.ui.nav import nav
 from goosebit.ui.static import static
 from goosebit.ui.templates import templates
@@ -28,7 +24,24 @@ async def lifespan(_: FastAPI):
     await db.close()
 
 
-app = FastAPI(lifespan=lifespan)
+app = FastAPI(
+    title="gooseBit",
+    summary="A simplistic, opinionated remote update server implementing hawkBit™'s DDI API.",
+    version=importlib.metadata.version("goosebit"),
+    lifespan=lifespan,
+    license_info={
+        "name": "Apache 2.0",
+        "identifier": "Apache-2.0",
+    },
+    swagger_ui_parameters={"operationsSorter": "alpha"},
+    openapi_tags=[
+        {
+            "name": "login",
+            "description": "API authentication.  "
+            "Can be used in the `authorization` header, in the format `{token_type} {access_token}`.",
+        }
+    ],
+)
 app.include_router(updater.router)
 app.include_router(ui.router)
 app.include_router(api.router)
@@ -39,7 +52,7 @@ async def lifespan(_: FastAPI):
 
 @app.middleware("http")
 async def attach_user(request: Request, call_next):
-    request.scope["user"] = get_current_user(request)
+    request.scope["user"] = await get_user_from_request(request)
     return await call_next(request)
 
 
@@ -54,20 +67,18 @@ def root_redirect(request: Request):
     return RedirectResponse(request.url_for("ui_root"))
 
 
-@app.get("/login", dependencies=[Depends(auto_redirect)], include_in_schema=False)
-async def login_ui(request: Request):
+@app.get("/login", include_in_schema=False, dependencies=[Depends(redirect_if_authenticated)])
+async def login_get(request: Request):
     return templates.TemplateResponse(request, "login.html.jinja")
 
 
-@app.post("/login", include_in_schema=False, dependencies=[Depends(authenticate_user)])
-async def login(request: Request, form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
-    resp = RedirectResponse(request.url_for("ui_root"), status_code=302)
-    resp.set_cookie(key="session_id", value=create_session(form_data.username))
-    return resp
+@app.post("/login", tags=["login"])
+async def login_post(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
+    return {"access_token": login_user(form_data.username, form_data.password), "token_type": "bearer"}
 
 
 @app.get("/logout", include_in_schema=False)
 async def logout(request: Request):
-    resp = RedirectResponse(request.url_for("login_ui"), status_code=302)
+    resp = RedirectResponse(request.url_for("login_get"), status_code=302)
     resp.delete_cookie(key="session_id")
     return resp

goosebit/api/routes.py

@@ -1,9 +1,9 @@
 from fastapi import APIRouter, Depends
 
-from goosebit.auth import Authentication
+from goosebit.auth import validate_current_user
 
 from . import telemetry, v1
 
-router = APIRouter(prefix="/api", dependencies=[Depends(Authentication())])
+router = APIRouter(prefix="/api", dependencies=[Depends(validate_current_user)])
 router.include_router(telemetry.router)
 router.include_router(v1.router)

goosebit/auth/__init__.py

@@ -1,115 +1,118 @@
+from __future__ import annotations
+
 import logging
+from typing import Annotated
 
 from argon2.exceptions import VerifyMismatchError
 from fastapi import Depends, HTTPException
 from fastapi.requests import HTTPConnection, Request
-from fastapi.security import SecurityScopes
+from fastapi.security import OAuth2PasswordBearer, SecurityScopes
 from joserfc import jwt
 from joserfc.errors import BadSignatureError
 
-from goosebit.settings import PWD_CXT, SECRET, USERS
+from goosebit.settings import PWD_CXT, USERS, config
+from goosebit.settings.schema import User
 
 logger = logging.getLogger(__name__)
 
 
-class Authentication:
-    def __init__(self, redirect: bool = False):
-        if redirect:
-            self.status_code = 302
-        else:
-            self.status_code = 401
+oauth2_auth = OAuth2PasswordBearer(tokenUrl="login", auto_error=False)
 
-    def __call__(self, connection: HTTPConnection):
-        session_id = connection.cookies.get("session_id")
-        headers = {"location": str(connection.url_for("login"))}
-        if session_id is None:
-            raise HTTPException(
-                status_code=self.status_code,
-                headers=headers,
-                detail="Invalid session ID",
-            )
-        user = get_user_from_session(session_id)
-        if user is None:
-            raise HTTPException(
-                status_code=self.status_code,
-                headers=headers,
-                detail="Invalid username",
-            )
-        return user
 
+async def session_auth(connection: HTTPConnection) -> str:
+    return connection.cookies.get("session_id")
 
-async def authenticate_user(request: Request):
-    form_data = await request.form()
-    username = form_data.get("username")
-    password = form_data.get("password")
+
+def create_token(username: str) -> str:
+    return jwt.encode(header={"alg": "HS256"}, claims={"username": username}, key=config.secret_key)
+
+
+def get_user_from_token(token: str) -> User | None:
+    if token is None:
+        return
+    try:
+        token_data = jwt.decode(token, config.secret_key)
+        username = token_data.claims["username"]
+        return USERS.get(username)
+    except (BadSignatureError, LookupError, ValueError):
+        pass
+
+
+def login_user(username: str, password: str) -> str:
     user = USERS.get(username)
     if user is None:
         raise HTTPException(
-            status_code=302,
-            headers={"location": str(request.url_for("login"))},
-            detail="Invalid credentials",
+            status_code=401,
+            detail="Invalid username or password",
+            headers={"WWW-Authenticate": "Bearer"},
         )
     try:
-        if not PWD_CXT.verify(user.hashed_pwd, password):
-            raise HTTPException(
-                status_code=302,
-                headers={"location": str(request.url_for("login"))},
-                detail="Invalid credentials",
-            )
+        PWD_CXT.verify(user.hashed_pwd, password)
     except VerifyMismatchError:
         raise HTTPException(
-            status_code=302,
-            headers={"location": str(request.url_for("login"))},
-            detail="Invalid credentials",
+            status_code=401,
+            detail="Invalid username or password",
+            headers={"WWW-Authenticate": "Bearer"},
         )
+    return create_token(user.username)
+
+
+def get_current_user(
+    session_token: Annotated[str, Depends(session_auth)] = None,
+    oauth2_token: Annotated[str, Depends(oauth2_auth)] = None,
+) -> User:
+    session_user = get_user_from_token(session_token)
+    oauth2_user = get_user_from_token(oauth2_token)
+    user = session_user or oauth2_user
     return user
 
 
-def create_session(username: str) -> str:
-    return jwt.encode(header={"alg": "HS256"}, claims={"username": username}, key=SECRET)
+# using | Request because oauth2_auth.__call__ expects is
+async def get_user_from_request(connection: HTTPConnection | Request) -> User:
+    token = await session_auth(connection) or await oauth2_auth(connection)
+    return get_user_from_token(token)
 
 
-def auto_redirect(request: Request):
-    session_id = request.cookies.get("session_id")
-    if get_user_from_session(session_id) is None:
-        return request
-    raise HTTPException(
-        status_code=302,
-        headers={"location": str(request.url_for("ui_root"))},
-        detail="Already logged in",
-    )
+def redirect_if_unauthenticated(connection: HTTPConnection, user: Annotated[User, Depends(get_current_user)]):
+    if user is None:
+        raise HTTPException(
+            status_code=302,
+            headers={"location": str(connection.url_for("login_get"))},
+            detail="Invalid username",
+        )
 
 
-def get_user_from_session(session_id: str):
-    if session_id is None:
-        return
-    try:
-        session_data = jwt.decode(session_id, SECRET)
-        return session_data.claims["username"]
-    except (BadSignatureError, LookupError, ValueError):
-        pass
+def redirect_if_authenticated(connection: HTTPConnection, user: Annotated[User, Depends(get_current_user)]):
+    if user is not None:
+        raise HTTPException(
+            status_code=302,
+            headers={"location": str(connection.url_for("ui_root"))},
+            detail="Already logged in",
+        )
 
 
-def get_current_user(request: Request):
-    session_id = request.cookies.get("session_id")
-    user = get_user_from_session(session_id)
+def validate_current_user(user: Annotated[User, Depends(get_current_user)]):
     if user is None:
-        return None
-    return USERS[user]
+        raise HTTPException(
+            status_code=401,
+            detail="Not authenticated",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    return user
 
 
 def validate_user_permissions(
     connection: HTTPConnection,
     security: SecurityScopes,
-    username: str = Depends(Authentication()),
+    user: User = Depends(get_current_user),
 ) -> HTTPConnection:
-    user = USERS[username]
     if security.scopes is None:
         return connection
     for scope in security.scopes:
         if scope not in user.permissions:
-            logger.warning(f"User {username} does not have permission {scope}")
+            logger.warning(f"User {user.username} does not have permission {scope}")
             raise HTTPException(
                 status_code=403,
                 detail="Not enough permissions",
+                headers={"WWW-Authenticate": "Bearer"},
             )

goosebit/realtime/routes.py

@@ -1,12 +1,12 @@
 from fastapi import APIRouter, Depends
 
-from goosebit.auth import Authentication
+from goosebit.auth import validate_current_user
 
 from . import logs
 
 router = APIRouter(
     prefix="/realtime",
-    dependencies=[Depends(Authentication())],
+    dependencies=[Depends(validate_current_user)],
     tags=["realtime"],
 )
 

goosebit/settings/__init__.py

@@ -1,4 +1,4 @@
-from .const import BASE_DIR, DB_MIGRATIONS_LOC, PWD_CXT, SECRET  # noqa: F401
+from .const import BASE_DIR, DB_MIGRATIONS_LOC, PWD_CXT  # noqa: F401
 from .schema import GooseBitSettings
 
 config = GooseBitSettings()

goosebit/settings/const.py

@@ -1,13 +1,10 @@
-import secrets
 from pathlib import Path
 
 from argon2 import PasswordHasher
-from joserfc.jwk import OctKey
 
 BASE_DIR = Path(__file__).resolve().parent.parent
 DB_MIGRATIONS_LOC = BASE_DIR.joinpath("migrations")
 
-SECRET = OctKey.import_key(secrets.token_hex(16))
 PWD_CXT = PasswordHasher()
 
 LOGGING_DEFAULT = {

goosebit/settings/schema.py

@@ -1,6 +1,8 @@
+import secrets
 from pathlib import Path
 from typing import Annotated, Iterable
 
+from joserfc.rfc7518.oct_key import OctKey
 from pydantic import BaseModel, BeforeValidator, Field
 from pydantic_settings import (
     BaseSettings,
@@ -49,6 +51,8 @@ class GooseBitSettings(BaseSettings):
     poll_time_updating: str = "00:00:05"
     poll_time_registration: str = "00:00:10"
 
+    secret_key: Annotated[OctKey, BeforeValidator(OctKey.import_key)] = secrets.token_hex(16)
+
     users: list[User] = []
 
     db_uri: str = f"sqlite:///{BASE_DIR.joinpath('db.sqlite3')}"

goosebit/ui/routes.py

@@ -4,7 +4,7 @@
 from fastapi.responses import RedirectResponse
 from fastapi.security import OAuth2PasswordBearer
 
-from goosebit.auth import Authentication, validate_user_permissions
+from goosebit.auth import redirect_if_unauthenticated, validate_user_permissions
 from goosebit.models import Firmware, Rollout
 from goosebit.permissions import Permissions
 from goosebit.settings import config
@@ -16,7 +16,7 @@
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")
 
-router = APIRouter(prefix="/ui", dependencies=[Depends(Authentication(redirect=True))], include_in_schema=False)
+router = APIRouter(prefix="/ui", dependencies=[Depends(redirect_if_unauthenticated)], include_in_schema=False)
 router.include_router(bff.router)
 
 

goosebit/ui/static/js/login.js

@@ -0,0 +1,23 @@
+loginForm = document.getElementById("login_form");
+
+async function login() {
+    const formData = new FormData(loginForm);
+
+    try {
+        const response = await fetch("/login", {
+            method: "POST",
+            body: formData,
+        });
+        tokenData = await response.json();
+        document.cookie = `session_id=${tokenData.access_token}; path=/`;
+        location.reload();
+    } catch (e) {
+        // handle form errors later
+        console.error(e);
+    }
+}
+
+loginForm.addEventListener("submit", (event) => {
+    event.preventDefault();
+    login();
+});