feat: add encrypted sync user management

Adds full CRUD for sync users (create/list/detail/edit) behind admin
auth and CSRF protection. Passwords are encrypted at rest and never
decrypted into a template context; only emails may be decrypted for
display on the edit form. Blank password fields on edit preserve the
existing encrypted password.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Bastian Wagner
2026-08-15 09:41:21 +02:00
parent da6b94ca2f
commit b39842fe2c
7 changed files with 484 additions and 3 deletions

View File

@@ -23,6 +23,15 @@ class UserRepository:
def list_enabled(self) -> list[SyncUser]: def list_enabled(self) -> list[SyncUser]:
return list(self.session.scalars(select(SyncUser).where(SyncUser.enabled.is_(True)).order_by(SyncUser.id))) 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: class ActivityRepository:
def __init__(self, session: Session) -> None: def __init__(self, session: Session) -> None:

11
app/web/forms.py Normal file
View File

@@ -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

View File

@@ -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.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates from fastapi.templating import Jinja2Templates
from app.auth.admin import password_matches, require_admin from app.auth.admin import password_matches, require_admin
from app.auth.csrf import ensure_csrf_token, validate_csrf 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() router = APIRouter()
templates = Jinja2Templates(directory="app/web/templates") 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) @router.get("/login", response_class=HTMLResponse)
def login_page(request: Request): def login_page(request: Request):
return templates.TemplateResponse(request, "login.html", {"csrf_token": ensure_csrf_token(request)}) return templates.TemplateResponse(request, "login.html", {"csrf_token": ensure_csrf_token(request)})
@@ -36,4 +51,136 @@ def login(
@router.get("/", response_class=HTMLResponse) @router.get("/", response_class=HTMLResponse)
def dashboard(request: Request): def dashboard(request: Request):
require_admin(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)

View File

@@ -4,9 +4,16 @@
{% block content %} {% block content %}
<h1>Dashboard</h1> <h1>Dashboard</h1>
<p><a href="/users/new">Add user</a></p>
<ul> <ul>
{% for user in users %} {% for user in users %}
<li>{{ user }}</li> <li>
<a href="/users/{{ user.id }}">{{ user.name }}</a>
&mdash; {{ "enabled" if user.enabled else "disabled" }}
&mdash; {{ user.health_state.value }}
</li>
{% else %}
<li>No users yet.</li>
{% endfor %} {% endfor %}
</ul> </ul>
{% endblock %} {% endblock %}

View File

@@ -0,0 +1,31 @@
{% extends "base.html" %}
{% block title %}{{ user.name }} - MyWhoosh Garmin Sync{% endblock %}
{% block content %}
<h1>{{ user.name }}</h1>
<p><a href="/users/{{ user.id }}/edit">Edit</a> | <a href="/">Back to dashboard</a></p>
<dl>
<dt>Enabled</dt>
<dd>{{ "Yes" if user.enabled else "No" }}</dd>
<dt>Health state</dt>
<dd>{{ user.health_state.value }}</dd>
<dt>MyWhoosh state</dt>
<dd>{{ user.mywhoosh_state }}</dd>
<dt>Garmin state</dt>
<dd>{{ user.garmin_state }}</dd>
<dt>Action reason</dt>
<dd>{{ user.action_reason or "-" }}</dd>
<dt>Created at</dt>
<dd>{{ user.created_at }}</dd>
<dt>Updated at</dt>
<dd>{{ user.updated_at }}</dd>
</dl>
{% endblock %}

View File

@@ -0,0 +1,40 @@
{% extends "base.html" %}
{% block title %}{% if user %}Edit {{ user.name }}{% else %}New User{% endif %} - MyWhoosh Garmin Sync{% endblock %}
{% block content %}
<h1>{% if user %}Edit User{% else %}New User{% endif %}</h1>
<form method="post" action="{{ form_action }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<label for="name">Name</label>
<input type="text" id="name" name="name" value="{{ user.name if user else '' }}" required>
<label for="mywhoosh_email">MyWhoosh Email</label>
<input type="email" id="mywhoosh_email" name="mywhoosh_email" value="{{ mywhoosh_email }}" required>
<label for="mywhoosh_password">MyWhoosh Password</label>
<input type="password" id="mywhoosh_password" name="mywhoosh_password" autocomplete="new-password"
{% if not user %}required{% endif %}>
{% if user %}
<p class="hint">Leave blank to keep the existing password.</p>
{% endif %}
<label for="garmin_email">Garmin Email</label>
<input type="email" id="garmin_email" name="garmin_email" value="{{ garmin_email }}" required>
<label for="garmin_password">Garmin Password</label>
<input type="password" id="garmin_password" name="garmin_password" autocomplete="new-password"
{% if not user %}required{% endif %}>
{% if user %}
<p class="hint">Leave blank to keep the existing password.</p>
{% endif %}
<label for="enabled">
<input type="checkbox" id="enabled" name="enabled" {% if not user or user.enabled %}checked{% endif %}>
Enabled
</label>
<button type="submit">{% if user %}Save{% else %}Create{% endif %}</button>
</form>
{% endblock %}

236
tests/web/test_users.py Normal file
View File

@@ -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