feat: add local admin authentication

This commit is contained in:
Bastian Wagner
2026-08-15 09:33:27 +02:00
parent 93232a809e
commit da6b94ca2f
10 changed files with 173 additions and 0 deletions

12
app/auth/admin.py Normal file
View File

@@ -0,0 +1,12 @@
import hmac
from fastapi import HTTPException, Request, status
def password_matches(submitted: str, configured: str) -> bool:
return hmac.compare_digest(submitted.encode("utf-8"), configured.encode("utf-8"))
def require_admin(request: Request) -> None:
if request.session.get("admin_authenticated") is not True:
raise HTTPException(status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/login"})

18
app/auth/csrf.py Normal file
View File

@@ -0,0 +1,18 @@
import hmac
import secrets
from fastapi import HTTPException, Request, status
def ensure_csrf_token(request: Request) -> str:
token = request.session.get("csrf_token")
if not isinstance(token, str):
token = secrets.token_urlsafe(32)
request.session["csrf_token"] = token
return token
def validate_csrf(request: Request, submitted_token: str) -> None:
expected = request.session.get("csrf_token")
if not isinstance(expected, str) or not hmac.compare_digest(expected, submitted_token):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Invalid CSRF token")

View File

@@ -1,7 +1,9 @@
from fastapi import FastAPI
from starlette.middleware.sessions import SessionMiddleware
from app.config import Settings, get_settings
from app.db.session import create_db_engine, create_session_factory, initialize_schema
from app.web.routes import router as web_router
def create_app(settings: Settings | None = None) -> FastAPI:
@@ -18,6 +20,14 @@ def create_app(settings: Settings | None = None) -> FastAPI:
app.state.db_engine = engine
app.state.session_factory = create_session_factory(engine)
app.add_middleware(
SessionMiddleware,
secret_key=resolved.secret_key,
same_site="lax",
https_only=False,
)
app.include_router(web_router)
@app.get("/healthz")
def healthz() -> dict[str, str]:
return {"status": "ok"}

39
app/web/routes.py Normal file
View File

@@ -0,0 +1,39 @@
from fastapi import APIRouter, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from app.auth.admin import password_matches, require_admin
from app.auth.csrf import ensure_csrf_token, validate_csrf
router = APIRouter()
templates = Jinja2Templates(directory="app/web/templates")
@router.get("/login", response_class=HTMLResponse)
def login_page(request: Request):
return templates.TemplateResponse(request, "login.html", {"csrf_token": ensure_csrf_token(request)})
@router.post("/login")
def login(
request: Request,
password: str = Form(...),
csrf_token: str = Form(...),
):
validate_csrf(request, csrf_token)
settings = request.app.state.settings
if not password_matches(password, settings.admin_password):
return templates.TemplateResponse(
request,
"login.html",
{"csrf_token": ensure_csrf_token(request), "error": "Invalid password"},
status_code=401,
)
request.session["admin_authenticated"] = True
return RedirectResponse("/", status_code=303)
@router.get("/", response_class=HTMLResponse)
def dashboard(request: Request):
require_admin(request)
return templates.TemplateResponse(request, "dashboard.html", {"csrf_token": ensure_csrf_token(request), "users": []})

View File

@@ -0,0 +1,10 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>{% block title %}MyWhoosh Garmin Sync{% endblock %}</title>
</head>
<body>
{% block content %}{% endblock %}
</body>
</html>

View File

@@ -0,0 +1,12 @@
{% extends "base.html" %}
{% block title %}Dashboard - MyWhoosh Garmin Sync{% endblock %}
{% block content %}
<h1>Dashboard</h1>
<ul>
{% for user in users %}
<li>{{ user }}</li>
{% endfor %}
</ul>
{% endblock %}

View File

@@ -0,0 +1,16 @@
{% extends "base.html" %}
{% block title %}Login - MyWhoosh Garmin Sync{% endblock %}
{% block content %}
<h1>Admin Login</h1>
{% if error %}
<p class="error">{{ error }}</p>
{% endif %}
<form method="post" action="/login">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<label for="password">Password</label>
<input type="password" id="password" name="password" required autofocus>
<button type="submit">Log in</button>
</form>
{% endblock %}

View File

@@ -14,6 +14,7 @@ dependencies = [
"cryptography>=43,<50",
"jinja2>=3.1,<4",
"python-multipart>=0.0.9,<1",
"itsdangerous>=2.1,<3",
]
[project.optional-dependencies]

View File

@@ -1,10 +1,16 @@
from pathlib import Path
import pytest
from cryptography.fernet import Fernet
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool
from app.config import Settings
from app.db.models import Base
from app.db.repositories import ActivityRepository, UserRepository
from app.main import create_app
@pytest.fixture
@@ -28,3 +34,16 @@ def user_repository(db_session: Session) -> UserRepository:
@pytest.fixture
def activity_repository(db_session: Session) -> ActivityRepository:
return ActivityRepository(db_session)
@pytest.fixture
def client(tmp_path: Path) -> TestClient:
settings = Settings(
ADMIN_PASSWORD="admin-secret",
SECRET_KEY="0123456789abcdef0123456789abcdef",
CREDENTIAL_ENCRYPTION_KEY=Fernet.generate_key().decode("ascii"),
DATA_DIR=str(tmp_path),
DATABASE_URL=f"sqlite:///{tmp_path / 'app.db'}",
SYNC_INTERVAL_MINUTES=5,
)
return TestClient(create_app(settings))

36
tests/web/test_auth.py Normal file
View File

@@ -0,0 +1,36 @@
from fastapi.testclient import TestClient
def test_dashboard_redirects_when_not_logged_in(client: TestClient) -> None:
response = client.get("/", follow_redirects=False)
assert response.status_code == 303
assert response.headers["location"] == "/login"
def extract_csrf(html: str) -> str:
marker = 'name="csrf_token" value="'
start = html.index(marker) + len(marker)
return html[start:html.index('"', start)]
def test_login_rejects_wrong_password(client: TestClient) -> None:
login_page = client.get("/login")
csrf = extract_csrf(login_page.text)
response = client.post(
"/login",
data={"password": "wrong", "csrf_token": csrf},
follow_redirects=False,
)
assert response.status_code == 401
def test_login_accepts_configured_password(client: TestClient) -> None:
login_page = client.get("/login")
csrf = extract_csrf(login_page.text)
response = client.post(
"/login",
data={"password": "admin-secret", "csrf_token": csrf},
follow_redirects=False,
)
assert response.status_code == 303
assert response.headers["location"] == "/"