diff --git a/app/db/repositories.py b/app/db/repositories.py index b239ff2..b703e40 100644 --- a/app/db/repositories.py +++ b/app/db/repositories.py @@ -23,6 +23,15 @@ class UserRepository: def list_enabled(self) -> list[SyncUser]: return list(self.session.scalars(select(SyncUser).where(SyncUser.enabled.is_(True)).order_by(SyncUser.id))) + def list_all(self) -> list[SyncUser]: + return list(self.session.scalars(select(SyncUser).order_by(SyncUser.name))) + + def update(self, user: SyncUser, **values) -> SyncUser: + for key, value in values.items(): + setattr(user, key, value) + self.session.commit() + return user + class ActivityRepository: def __init__(self, session: Session) -> None: diff --git a/app/web/forms.py b/app/web/forms.py new file mode 100644 index 0000000..6010dc4 --- /dev/null +++ b/app/web/forms.py @@ -0,0 +1,11 @@ +from dataclasses import dataclass + + +@dataclass(frozen=True) +class UserFormData: + name: str + mywhoosh_email: str + mywhoosh_password: str + garmin_email: str + garmin_password: str + enabled: bool diff --git a/app/web/routes.py b/app/web/routes.py index ecf1355..82e8f31 100644 --- a/app/web/routes.py +++ b/app/web/routes.py @@ -1,14 +1,29 @@ -from fastapi import APIRouter, Form, Request +from fastapi import APIRouter, Form, HTTPException, Request, status 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 +from app.db.models import SyncUser +from app.db.repositories import UserRepository +from app.security.credentials import CredentialCipher +from app.web.forms import UserFormData router = APIRouter() templates = Jinja2Templates(directory="app/web/templates") +def _get_user_or_404(repository: UserRepository, user_id: int) -> SyncUser: + user = repository.get(user_id) + if user is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") + return user + + +def _cipher(request: Request) -> CredentialCipher: + return CredentialCipher(request.app.state.settings.credential_encryption_key) + + @router.get("/login", response_class=HTMLResponse) def login_page(request: Request): return templates.TemplateResponse(request, "login.html", {"csrf_token": ensure_csrf_token(request)}) @@ -36,4 +51,136 @@ def login( @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": []}) + with request.app.state.session_factory() as session: + users = UserRepository(session).list_all() + return templates.TemplateResponse( + request, + "dashboard.html", + {"users": users, "csrf_token": ensure_csrf_token(request)}, + ) + + +@router.get("/users/new", response_class=HTMLResponse) +def new_user_page(request: Request): + require_admin(request) + return templates.TemplateResponse( + request, + "users/form.html", + { + "csrf_token": ensure_csrf_token(request), + "user": None, + "form_action": "/users", + "mywhoosh_email": "", + "garmin_email": "", + }, + ) + + +@router.post("/users") +def create_user( + request: Request, + csrf_token: str = Form(...), + name: str = Form(...), + mywhoosh_email: str = Form(...), + mywhoosh_password: str = Form(...), + garmin_email: str = Form(...), + garmin_password: str = Form(...), + enabled: str | None = Form(None), +): + require_admin(request) + validate_csrf(request, csrf_token) + form = UserFormData( + name=name, + mywhoosh_email=mywhoosh_email, + mywhoosh_password=mywhoosh_password, + garmin_email=garmin_email, + garmin_password=garmin_password, + enabled=enabled is not None, + ) + cipher = _cipher(request) + with request.app.state.session_factory() as session: + repository = UserRepository(session) + user = repository.create( + name=form.name.strip(), + enabled=form.enabled, + mywhoosh_email_enc=cipher.encrypt(form.mywhoosh_email.strip()), + mywhoosh_password_enc=cipher.encrypt(form.mywhoosh_password), + garmin_email_enc=cipher.encrypt(form.garmin_email.strip()), + garmin_password_enc=cipher.encrypt(form.garmin_password), + ) + user_id = user.id + return RedirectResponse(f"/users/{user_id}", status_code=303) + + +@router.get("/users/{user_id}", response_class=HTMLResponse) +def user_detail(request: Request, user_id: int): + require_admin(request) + with request.app.state.session_factory() as session: + user = _get_user_or_404(UserRepository(session), user_id) + return templates.TemplateResponse( + request, + "users/detail.html", + { + "csrf_token": ensure_csrf_token(request), + "user": user, + }, + ) + + +@router.get("/users/{user_id}/edit", response_class=HTMLResponse) +def edit_user_page(request: Request, user_id: int): + require_admin(request) + cipher = _cipher(request) + with request.app.state.session_factory() as session: + user = _get_user_or_404(UserRepository(session), user_id) + return templates.TemplateResponse( + request, + "users/form.html", + { + "csrf_token": ensure_csrf_token(request), + "user": user, + "form_action": f"/users/{user_id}", + "mywhoosh_email": cipher.decrypt(user.mywhoosh_email_enc), + "garmin_email": cipher.decrypt(user.garmin_email_enc), + }, + ) + + +@router.post("/users/{user_id}") +def update_user( + request: Request, + user_id: int, + csrf_token: str = Form(...), + name: str = Form(...), + mywhoosh_email: str = Form(...), + mywhoosh_password: str = Form(""), + garmin_email: str = Form(...), + garmin_password: str = Form(""), + enabled: str | None = Form(None), +): + require_admin(request) + validate_csrf(request, csrf_token) + form = UserFormData( + name=name, + mywhoosh_email=mywhoosh_email, + mywhoosh_password=mywhoosh_password, + garmin_email=garmin_email, + garmin_password=garmin_password, + enabled=enabled is not None, + ) + cipher = _cipher(request) + with request.app.state.session_factory() as session: + repository = UserRepository(session) + user = _get_user_or_404(repository, user_id) + values = { + "name": form.name.strip(), + "enabled": form.enabled, + "mywhoosh_email_enc": cipher.encrypt(form.mywhoosh_email.strip()), + "garmin_email_enc": cipher.encrypt(form.garmin_email.strip()), + } + if form.mywhoosh_password: + values["mywhoosh_password_enc"] = cipher.encrypt(form.mywhoosh_password) + if form.garmin_password: + values["garmin_password_enc"] = cipher.encrypt(form.garmin_password) + repository.update(user, **values) + return RedirectResponse(f"/users/{user_id}", status_code=303) diff --git a/app/web/templates/dashboard.html b/app/web/templates/dashboard.html index 421e1c8..320affb 100644 --- a/app/web/templates/dashboard.html +++ b/app/web/templates/dashboard.html @@ -4,9 +4,16 @@ {% block content %}

Dashboard

+

Add user

{% endblock %} diff --git a/app/web/templates/users/detail.html b/app/web/templates/users/detail.html new file mode 100644 index 0000000..3ebce2f --- /dev/null +++ b/app/web/templates/users/detail.html @@ -0,0 +1,31 @@ +{% extends "base.html" %} + +{% block title %}{{ user.name }} - MyWhoosh Garmin Sync{% endblock %} + +{% block content %} +

{{ user.name }}

+

Edit | Back to dashboard

+ +
+
Enabled
+
{{ "Yes" if user.enabled else "No" }}
+ +
Health state
+
{{ user.health_state.value }}
+ +
MyWhoosh state
+
{{ user.mywhoosh_state }}
+ +
Garmin state
+
{{ user.garmin_state }}
+ +
Action reason
+
{{ user.action_reason or "-" }}
+ +
Created at
+
{{ user.created_at }}
+ +
Updated at
+
{{ user.updated_at }}
+
+{% endblock %} diff --git a/app/web/templates/users/form.html b/app/web/templates/users/form.html new file mode 100644 index 0000000..4732ac1 --- /dev/null +++ b/app/web/templates/users/form.html @@ -0,0 +1,40 @@ +{% extends "base.html" %} + +{% block title %}{% if user %}Edit {{ user.name }}{% else %}New User{% endif %} - MyWhoosh Garmin Sync{% endblock %} + +{% block content %} +

{% if user %}Edit User{% else %}New User{% endif %}

+
+ + + + + + + + + + + {% if user %} +

Leave blank to keep the existing password.

+ {% endif %} + + + + + + + {% if user %} +

Leave blank to keep the existing password.

+ {% endif %} + + + + +
+{% endblock %} diff --git a/tests/web/test_users.py b/tests/web/test_users.py new file mode 100644 index 0000000..82f72b6 --- /dev/null +++ b/tests/web/test_users.py @@ -0,0 +1,236 @@ +from fastapi.testclient import TestClient + +from app.db.repositories import UserRepository +from app.security.credentials import CredentialCipher + + +def login(client: TestClient) -> None: + page = client.get("/login") + csrf = extract_csrf(page.text) + response = client.post( + "/login", + data={"password": "admin-secret", "csrf_token": csrf}, + follow_redirects=False, + ) + assert response.status_code == 303 + + +def extract_csrf(html: str) -> str: + marker = 'name="csrf_token" value="' + start = html.index(marker) + len(marker) + end = html.index('"', start) + return html[start:end] + + +def create_user_via_http(client: TestClient) -> int: + page = client.get("/users/new") + csrf = extract_csrf(page.text) + response = client.post( + "/users", + data={ + "csrf_token": csrf, + "name": "Max", + "mywhoosh_email": "max@example.com", + "mywhoosh_password": "mw-secret", + "garmin_email": "max-garmin@example.com", + "garmin_password": "garmin-secret", + "enabled": "on", + }, + follow_redirects=False, + ) + assert response.status_code == 303 + location = response.headers["location"] + return int(location.rsplit("/", 1)[-1]) + + +def test_create_user_encrypts_credentials_and_never_renders_them(client: TestClient) -> None: + login(client) + page = client.get("/users/new") + assert page.status_code == 200 + csrf = extract_csrf(page.text) + response = client.post( + "/users", + data={ + "csrf_token": csrf, + "name": "Max", + "mywhoosh_email": "max@example.com", + "mywhoosh_password": "mw-secret", + "garmin_email": "max-garmin@example.com", + "garmin_password": "garmin-secret", + "enabled": "on", + }, + follow_redirects=True, + ) + assert response.status_code == 200 + assert "mw-secret" not in response.text + assert "garmin-secret" not in response.text + + +def test_create_user_persists_encrypted_credentials(client: TestClient) -> None: + login(client) + user_id = create_user_via_http(client) + + with client.app.state.session_factory() as session: + user = UserRepository(session).get(user_id) + assert user is not None + assert user.mywhoosh_email_enc != "max@example.com" + assert user.mywhoosh_password_enc != "mw-secret" + + cipher = CredentialCipher(client.app.state.settings.credential_encryption_key) + assert cipher.decrypt(user.mywhoosh_email_enc) == "max@example.com" + assert cipher.decrypt(user.mywhoosh_password_enc) == "mw-secret" + assert cipher.decrypt(user.garmin_email_enc) == "max-garmin@example.com" + assert cipher.decrypt(user.garmin_password_enc) == "garmin-secret" + + +def test_edit_page_never_renders_passwords(client: TestClient) -> None: + login(client) + user_id = create_user_via_http(client) + + page = client.get(f"/users/{user_id}/edit") + assert page.status_code == 200 + assert "mw-secret" not in page.text + assert "garmin-secret" not in page.text + # emails may be shown for display + assert "max@example.com" in page.text + assert "max-garmin@example.com" in page.text + + +def test_edit_user_blank_password_preserves_existing_password(client: TestClient) -> None: + login(client) + user_id = create_user_via_http(client) + + edit_page = client.get(f"/users/{user_id}/edit") + csrf = extract_csrf(edit_page.text) + response = client.post( + f"/users/{user_id}", + data={ + "csrf_token": csrf, + "name": "Max Updated", + "mywhoosh_email": "max@example.com", + "mywhoosh_password": "", + "garmin_email": "max-garmin@example.com", + "garmin_password": "", + "enabled": "on", + }, + follow_redirects=True, + ) + assert response.status_code == 200 + assert "mw-secret" not in response.text + assert "garmin-secret" not in response.text + + with client.app.state.session_factory() as session: + user = UserRepository(session).get(user_id) + assert user is not None + assert user.name == "Max Updated" + cipher = CredentialCipher(client.app.state.settings.credential_encryption_key) + assert cipher.decrypt(user.mywhoosh_password_enc) == "mw-secret" + assert cipher.decrypt(user.garmin_password_enc) == "garmin-secret" + + +def test_edit_user_replaces_password_when_submitted(client: TestClient) -> None: + login(client) + user_id = create_user_via_http(client) + + edit_page = client.get(f"/users/{user_id}/edit") + csrf = extract_csrf(edit_page.text) + response = client.post( + f"/users/{user_id}", + data={ + "csrf_token": csrf, + "name": "Max", + "mywhoosh_email": "max@example.com", + "mywhoosh_password": "new-mw-secret", + "garmin_email": "max-garmin@example.com", + "garmin_password": "", + "enabled": "on", + }, + follow_redirects=True, + ) + assert response.status_code == 200 + assert "new-mw-secret" not in response.text + + with client.app.state.session_factory() as session: + user = UserRepository(session).get(user_id) + assert user is not None + cipher = CredentialCipher(client.app.state.settings.credential_encryption_key) + assert cipher.decrypt(user.mywhoosh_password_enc) == "new-mw-secret" + assert cipher.decrypt(user.garmin_password_enc) == "garmin-secret" + + +def test_user_detail_page_shows_no_secrets(client: TestClient) -> None: + login(client) + user_id = create_user_via_http(client) + + response = client.get(f"/users/{user_id}") + assert response.status_code == 200 + assert "mw-secret" not in response.text + assert "garmin-secret" not in response.text + assert "max@example.com" not in response.text + assert "max-garmin@example.com" not in response.text + assert "Max" in response.text + + +def test_unknown_user_returns_404_for_detail(client: TestClient) -> None: + login(client) + response = client.get("/users/999999") + assert response.status_code == 404 + + +def test_unknown_user_returns_404_for_edit(client: TestClient) -> None: + login(client) + response = client.get("/users/999999/edit") + assert response.status_code == 404 + + +def test_unknown_user_returns_404_for_update(client: TestClient) -> None: + login(client) + page = client.get("/users/new") + csrf = extract_csrf(page.text) + response = client.post( + "/users/999999", + data={ + "csrf_token": csrf, + "name": "Nobody", + "mywhoosh_email": "a@example.com", + "mywhoosh_password": "", + "garmin_email": "b@example.com", + "garmin_password": "", + "enabled": "on", + }, + ) + assert response.status_code == 404 + + +def test_create_user_requires_admin(client: TestClient) -> None: + response = client.get("/users/new", follow_redirects=False) + assert response.status_code == 303 + assert response.headers["location"] == "/login" + + +def test_create_user_rejects_invalid_csrf(client: TestClient) -> None: + login(client) + response = client.post( + "/users", + data={ + "csrf_token": "invalid-token", + "name": "Max", + "mywhoosh_email": "max@example.com", + "mywhoosh_password": "mw-secret", + "garmin_email": "max-garmin@example.com", + "garmin_password": "garmin-secret", + "enabled": "on", + }, + ) + assert response.status_code == 403 + + +def test_dashboard_lists_created_user_without_secrets(client: TestClient) -> None: + login(client) + create_user_via_http(client) + + response = client.get("/") + assert response.status_code == 200 + assert "Max" in response.text + assert "mw-secret" not in response.text + assert "garmin-secret" not in response.text