auth für user

This commit is contained in:
Bastian Wagner
2026-08-15 21:34:40 +02:00
parent adfe14dfa8
commit f7b04337ce
11 changed files with 773 additions and 0 deletions

36
app/auth/account.py Normal file
View File

@@ -0,0 +1,36 @@
from sqlalchemy.orm import Session
from app.auth.admin import password_matches
from app.db.models import SyncUser
from app.db.repositories import UserRepository
from app.security.credentials import CredentialCipher
def authenticate_self_service(
session: Session, cipher: CredentialCipher, email: str, password: str
) -> SyncUser | None:
"""Matches submitted email/password against any user's stored MyWhoosh OR
Garmin credentials -- decrypted and compared locally rather than
verified against the live MyWhoosh/Garmin APIs, so logging into this app
never opens a redundant upstream session (which, for MyWhoosh, would
itself trigger the "already logged in from another device" conflict)."""
submitted_email = email.strip()
if not submitted_email or not password:
return None
for user in UserRepository(session).list_all():
if _credential_matches(cipher, user.mywhoosh_email_enc, user.mywhoosh_password_enc, submitted_email, password):
return user
if _credential_matches(cipher, user.garmin_email_enc, user.garmin_password_enc, submitted_email, password):
return user
return None
def _credential_matches(
cipher: CredentialCipher, email_enc: str, password_enc: str, submitted_email: str, submitted_password: str
) -> bool:
try:
stored_email = cipher.decrypt(email_enc)
stored_password = cipher.decrypt(password_enc)
except ValueError:
return False
return stored_email == submitted_email and password_matches(submitted_password, stored_password)

View File

@@ -10,3 +10,10 @@ def password_matches(submitted: str, configured: str) -> bool:
def require_admin(request: Request) -> None: def require_admin(request: Request) -> None:
if request.session.get("admin_authenticated") is not True: if request.session.get("admin_authenticated") is not True:
raise HTTPException(status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/login"}) raise HTTPException(status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/login"})
def require_self_service(request: Request) -> int:
user_id = request.session.get("self_service_user_id")
if not isinstance(user_id, int):
raise HTTPException(status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/account-login"})
return user_id

View File

@@ -14,6 +14,7 @@ from app.notifications.emailer import EmailNotifier
from app.security.credentials import CredentialCipher from app.security.credentials import CredentialCipher
from app.sync.manager import SyncManager from app.sync.manager import SyncManager
from app.sync.scheduler import SyncScheduler from app.sync.scheduler import SyncScheduler
from app.web.account import router as account_router
from app.web.operations import router as operations_router from app.web.operations import router as operations_router
from app.web.routes import router as web_router from app.web.routes import router as web_router
@@ -78,6 +79,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
) )
app.include_router(web_router) app.include_router(web_router)
app.include_router(operations_router) app.include_router(operations_router)
app.include_router(account_router)
app.mount( app.mount(
"/static", "/static",
StaticFiles(directory=str(Path(__file__).resolve().parent / "web" / "static")), StaticFiles(directory=str(Path(__file__).resolve().parent / "web" / "static")),

153
app/web/account.py Normal file
View File

@@ -0,0 +1,153 @@
from fastapi import APIRouter, Form, HTTPException, Request, status
from fastapi.responses import HTMLResponse, RedirectResponse
from app.auth.account import authenticate_self_service
from app.auth.admin import require_self_service
from app.auth.csrf import ensure_csrf_token, validate_csrf
from app.db.repositories import ActivityRepository, SyncRunRepository, UserRepository
from app.security.credentials import CredentialCipher
from app.sync.manager import SyncAlreadyRunning
from app.web.operations import _normalize_outcome
from app.web.routes import templates
router = APIRouter()
def _cipher(request: Request) -> CredentialCipher:
return CredentialCipher(request.app.state.settings.credential_encryption_key)
def _require_non_empty(value: str, field_name: str) -> None:
if not value.strip():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"{field_name} must not be empty",
)
@router.get("/account-login", response_class=HTMLResponse)
def account_login_page(request: Request):
return templates.TemplateResponse(request, "account_login.html", {"csrf_token": ensure_csrf_token(request)})
@router.post("/account-login")
def account_login(
request: Request,
csrf_token: str = Form(...),
email: str = Form(...),
password: str = Form(...),
):
validate_csrf(request, csrf_token)
cipher = _cipher(request)
with request.app.state.session_factory() as session:
user = authenticate_self_service(session, cipher, email, password)
if user is None:
return templates.TemplateResponse(
request,
"account_login.html",
{"csrf_token": ensure_csrf_token(request), "error": "Invalid email or password"},
status_code=401,
)
request.session["self_service_user_id"] = user.id
return RedirectResponse("/account", status_code=303)
@router.post("/account-logout")
def account_logout(request: Request, csrf_token: str = Form(...)):
validate_csrf(request, csrf_token)
request.session.pop("self_service_user_id", None)
return RedirectResponse("/account-login", status_code=303)
def _get_own_user_or_reauth(request: Request, repository: UserRepository, user_id: int):
user = repository.get(user_id)
if user is None:
request.session.pop("self_service_user_id", None)
raise HTTPException(status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/account-login"})
return user
@router.get("/account", response_class=HTMLResponse)
def account_detail(request: Request):
user_id = require_self_service(request)
with request.app.state.session_factory() as session:
user = _get_own_user_or_reauth(request, UserRepository(session), user_id)
activities = ActivityRepository(session).list_pending_for_user(user_id)
recent_runs = SyncRunRepository(session).list_recent_for_user(user_id, limit=10)
return templates.TemplateResponse(
request,
"account/detail.html",
{
"csrf_token": ensure_csrf_token(request),
"user": user,
"activities": activities,
"recent_runs": recent_runs,
},
)
@router.post("/account/sync", response_class=HTMLResponse)
async def account_sync(request: Request, csrf_token: str = Form(...)):
user_id = require_self_service(request)
validate_csrf(request, csrf_token)
try:
outcome = await request.app.state.sync_manager.sync_user(user_id)
except SyncAlreadyRunning:
return HTMLResponse("Sync already running for this user", status_code=409)
return templates.TemplateResponse(
request, "fragments/sync_result.html", {"outcomes": [_normalize_outcome(outcome)]}
)
@router.get("/account/edit", response_class=HTMLResponse)
def account_edit_page(request: Request):
user_id = require_self_service(request)
cipher = _cipher(request)
with request.app.state.session_factory() as session:
user = _get_own_user_or_reauth(request, UserRepository(session), user_id)
return templates.TemplateResponse(
request,
"account/form.html",
{
"csrf_token": ensure_csrf_token(request),
"user": user,
"mywhoosh_email": cipher.decrypt(user.mywhoosh_email_enc),
"garmin_email": cipher.decrypt(user.garmin_email_enc),
},
)
@router.post("/account/edit")
def account_update(
request: Request,
csrf_token: str = Form(...),
mywhoosh_email: str = Form(""),
mywhoosh_password: str = Form(""),
garmin_email: str = Form(""),
garmin_password: str = Form(""),
notify_email_enabled: str | None = Form(None),
notification_email: str = Form(""),
):
user_id = require_self_service(request)
validate_csrf(request, csrf_token)
_require_non_empty(mywhoosh_email, "mywhoosh_email")
_require_non_empty(garmin_email, "garmin_email")
notify_enabled = notify_email_enabled is not None
if notify_enabled:
_require_non_empty(notification_email, "notification_email")
cipher = _cipher(request)
with request.app.state.session_factory() as session:
repository = UserRepository(session)
user = _get_own_user_or_reauth(request, repository, user_id)
values = {
"mywhoosh_email_enc": cipher.encrypt(mywhoosh_email.strip()),
"garmin_email_enc": cipher.encrypt(garmin_email.strip()),
"notify_email_enabled": notify_enabled,
"notification_email": notification_email.strip() or None,
}
if mywhoosh_password:
values["mywhoosh_password_enc"] = cipher.encrypt(mywhoosh_password)
if garmin_password:
values["garmin_password_enc"] = cipher.encrypt(garmin_password)
repository.update(user, **values)
return RedirectResponse("/account", status_code=303)

View File

@@ -0,0 +1,80 @@
{% extends "base.html" %}
{% block title %}My Account - MyWhoosh Garmin Sync{% endblock %}
{% block content %}
<h1>{{ user.name }}</h1>
<div class="page-actions">
<a class="btn secondary" href="/account/edit">Edit</a>
<form method="post" action="/account/sync" class="inline-form">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<button type="submit">Sync now</button>
</form>
<form method="post" action="/account-logout" class="inline-form">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<button type="submit" class="secondary">Log out</button>
</form>
</div>
<div class="card">
<dl class="info-grid">
<dt>Status</dt>
<dd><span class="badge badge-{{ user.health_state.value }}">{{ user.health_state.value.replace("_", " ") }}</span></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>
</dl>
</div>
{% if user.action_reason == "mywhoosh_device_conflict" %}
<h2>MyWhoosh device conflict</h2>
<div class="card">
<p>MyWhoosh reports this account is already logged in on another device. Log out of MyWhoosh there (app or website), then retry the sync.</p>
</div>
{% endif %}
<h2>Recent sync runs</h2>
<div class="card">
{% if recent_runs %}
<table>
<thead>
<tr>
<th>Started</th>
<th>Finished</th>
<th>Status</th>
<th>Discovered</th>
<th>Imported</th>
<th>Skipped</th>
<th>Failed</th>
</tr>
</thead>
<tbody>
{% for run in recent_runs %}
<tr>
<td>{{ run.started_at }}</td>
<td>{{ run.finished_at or "-" }}</td>
<td><span class="badge badge-{{ run.status.value }}">{{ run.status.value }}</span></td>
<td>{{ run.discovered_count }}</td>
<td>{{ run.imported_count }}</td>
<td>{{ run.skipped_count }}</td>
<td>{{ run.failed_count }}</td>
</tr>
{% if run.summary_error %}
<tr>
<td colspan="7" class="summary-error">{{ run.summary_error }}</td>
</tr>
{% endif %}
{% endfor %}
</tbody>
</table>
{% else %}
<p class="empty-state">No sync runs yet.</p>
{% endif %}
</div>
{% endblock %}

View File

@@ -0,0 +1,39 @@
{% extends "base.html" %}
{% block title %}Edit My Account - MyWhoosh Garmin Sync{% endblock %}
{% block content %}
<h1>Edit My Account</h1>
<div class="card">
<form method="post" action="/account/edit" class="stacked-form">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<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">
<p class="hint">Leave blank to keep the existing password.</p>
<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">
<p class="hint">Leave blank to keep the existing password.</p>
<label for="notify_email_enabled">
<input type="checkbox" id="notify_email_enabled" name="notify_email_enabled"
{% if user.notify_email_enabled %}checked{% endif %}>
Email me when this account needs attention
</label>
<label for="notification_email">Notification email</label>
<input type="email" id="notification_email" name="notification_email"
value="{{ user.notification_email or '' }}">
<button type="submit">Save</button>
</form>
</div>
<p><a href="/account">Back to my account</a></p>
{% endblock %}

View File

@@ -0,0 +1,22 @@
{% extends "base.html" %}
{% block title %}Account Login - MyWhoosh Garmin Sync{% endblock %}
{% block content %}
<h1>Account Login</h1>
<div class="card">
{% if error %}
<p class="error">{{ error }}</p>
{% endif %}
<form method="post" action="/account-login" class="stacked-form">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<label for="email">MyWhoosh or Garmin email</label>
<input type="email" id="email" name="email" required autofocus>
<label for="password">MyWhoosh or Garmin password</label>
<input type="password" id="password" name="password" required>
<button type="submit">Log in</button>
</form>
<p class="hint">Use the email and password for either your MyWhoosh or your Garmin account.</p>
</div>
<p><a href="/login">Admin login</a></p>
{% endblock %}

View File

@@ -10,8 +10,13 @@
<header class="topbar"> <header class="topbar">
<span class="brand">MyWhoosh &rarr; Garmin Sync</span> <span class="brand">MyWhoosh &rarr; Garmin Sync</span>
<nav> <nav>
{% if request.session.get('admin_authenticated') %}
<a href="/">Dashboard</a> <a href="/">Dashboard</a>
<a href="/system">System</a> <a href="/system">System</a>
{% endif %}
{% if request.session.get('self_service_user_id') %}
<a href="/account">My Account</a>
{% endif %}
</nav> </nav>
</header> </header>
<main class="container"> <main class="container">

View File

@@ -15,4 +15,5 @@
<button type="submit">Log in</button> <button type="submit">Log in</button>
</form> </form>
</div> </div>
<p><a href="/account-login">Log in with your MyWhoosh or Garmin account instead</a></p>
{% endblock %} {% endblock %}

108
tests/auth/test_account.py Normal file
View File

@@ -0,0 +1,108 @@
from cryptography.fernet import Fernet
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from app.auth.account import authenticate_self_service
from app.db.models import Base
from app.db.repositories import UserRepository
from app.security.credentials import CredentialCipher
def _make_session_and_cipher():
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
Base.metadata.create_all(engine)
factory = sessionmaker(bind=engine, expire_on_commit=False)
cipher = CredentialCipher(Fernet.generate_key().decode("ascii"))
return factory(), cipher
def _seed_user(session, cipher, **overrides):
values = dict(
name="Max",
enabled=True,
mywhoosh_email_enc=cipher.encrypt("max@mywhoosh.example"),
mywhoosh_password_enc=cipher.encrypt("mw-secret"),
garmin_email_enc=cipher.encrypt("max@garmin.example"),
garmin_password_enc=cipher.encrypt("garmin-secret"),
)
values.update(overrides)
return UserRepository(session).create(**values)
def test_authenticates_with_mywhoosh_credentials() -> None:
session, cipher = _make_session_and_cipher()
user = _seed_user(session, cipher)
result = authenticate_self_service(session, cipher, "max@mywhoosh.example", "mw-secret")
assert result is not None
assert result.id == user.id
def test_authenticates_with_garmin_credentials() -> None:
session, cipher = _make_session_and_cipher()
user = _seed_user(session, cipher)
result = authenticate_self_service(session, cipher, "max@garmin.example", "garmin-secret")
assert result is not None
assert result.id == user.id
def test_rejects_wrong_password() -> None:
session, cipher = _make_session_and_cipher()
_seed_user(session, cipher)
assert authenticate_self_service(session, cipher, "max@mywhoosh.example", "wrong") is None
def test_rejects_unknown_email() -> None:
session, cipher = _make_session_and_cipher()
_seed_user(session, cipher)
assert authenticate_self_service(session, cipher, "nobody@example.com", "mw-secret") is None
def test_rejects_mixed_email_and_password_from_different_accounts() -> None:
"""A MyWhoosh email paired with the Garmin password (or vice versa) for
the same user must not authenticate -- each pair is checked together."""
session, cipher = _make_session_and_cipher()
_seed_user(session, cipher)
assert authenticate_self_service(session, cipher, "max@mywhoosh.example", "garmin-secret") is None
assert authenticate_self_service(session, cipher, "max@garmin.example", "mw-secret") is None
def test_rejects_empty_password() -> None:
session, cipher = _make_session_and_cipher()
_seed_user(session, cipher)
assert authenticate_self_service(session, cipher, "max@mywhoosh.example", "") is None
def test_picks_correct_user_among_several() -> None:
session, cipher = _make_session_and_cipher()
_seed_user(
session,
cipher,
name="Anna",
mywhoosh_email_enc=cipher.encrypt("anna@mywhoosh.example"),
mywhoosh_password_enc=cipher.encrypt("anna-secret"),
garmin_email_enc=cipher.encrypt("anna@garmin.example"),
garmin_password_enc=cipher.encrypt("anna-garmin-secret"),
)
bob = _seed_user(
session,
cipher,
name="Bob",
mywhoosh_email_enc=cipher.encrypt("bob@mywhoosh.example"),
mywhoosh_password_enc=cipher.encrypt("bob-secret"),
garmin_email_enc=cipher.encrypt("bob@garmin.example"),
garmin_password_enc=cipher.encrypt("bob-garmin-secret"),
)
result = authenticate_self_service(session, cipher, "bob@mywhoosh.example", "bob-secret")
assert result is not None
assert result.id == bob.id

View File

@@ -0,0 +1,320 @@
from fastapi.testclient import TestClient
from app.db.repositories import UserRepository
from app.security.credentials import CredentialCipher
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 admin_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 create_user_via_admin(client: TestClient, **overrides) -> int:
admin_login(client)
page = client.get("/users/new")
csrf = extract_csrf(page.text)
payload = {
"csrf_token": csrf,
"name": "Max",
"mywhoosh_email": "max@mywhoosh.example",
"mywhoosh_password": "mw-secret",
"garmin_email": "max@garmin.example",
"garmin_password": "garmin-secret",
"enabled": "on",
}
payload.update(overrides)
response = client.post("/users", data=payload, follow_redirects=False)
assert response.status_code == 303
user_id = int(response.headers["location"].rsplit("/", 1)[-1])
client.cookies.clear()
return user_id
def account_login(client: TestClient, *, email: str, password: str):
page = client.get("/account-login")
csrf = extract_csrf(page.text)
return client.post(
"/account-login",
data={"csrf_token": csrf, "email": email, "password": password},
follow_redirects=False,
)
def test_login_with_mywhoosh_credentials_succeeds(client: TestClient) -> None:
create_user_via_admin(client)
response = account_login(client, email="max@mywhoosh.example", password="mw-secret")
assert response.status_code == 303
assert response.headers["location"] == "/account"
def test_login_with_garmin_credentials_succeeds(client: TestClient) -> None:
create_user_via_admin(client)
response = account_login(client, email="max@garmin.example", password="garmin-secret")
assert response.status_code == 303
assert response.headers["location"] == "/account"
def test_login_with_wrong_password_is_rejected(client: TestClient) -> None:
create_user_via_admin(client)
response = account_login(client, email="max@mywhoosh.example", password="wrong")
assert response.status_code == 401
assert "Invalid email or password" in response.text
def test_login_never_makes_the_stored_password_appear_in_response(client: TestClient) -> None:
create_user_via_admin(client)
response = account_login(client, email="max@mywhoosh.example", password="wrong")
assert "mw-secret" not in response.text
def test_account_detail_requires_login(client: TestClient) -> None:
response = client.get("/account", follow_redirects=False)
assert response.status_code == 303
assert response.headers["location"] == "/account-login"
def test_account_detail_shows_own_status_only(client: TestClient) -> None:
create_user_via_admin(client)
account_login(client, email="max@mywhoosh.example", password="mw-secret")
response = client.get("/account")
assert response.status_code == 200
assert "Max" in response.text
assert "mw-secret" not in response.text
assert "garmin-secret" not in response.text
def test_account_edit_page_prefills_emails_not_passwords(client: TestClient) -> None:
create_user_via_admin(client)
account_login(client, email="max@mywhoosh.example", password="mw-secret")
response = client.get("/account/edit")
assert response.status_code == 200
assert "max@mywhoosh.example" in response.text
assert "max@garmin.example" in response.text
assert "mw-secret" not in response.text
assert "garmin-secret" not in response.text
def test_account_update_blank_password_preserves_existing_password(client: TestClient) -> None:
user_id = create_user_via_admin(client)
account_login(client, email="max@mywhoosh.example", password="mw-secret")
edit_page = client.get("/account/edit")
csrf = extract_csrf(edit_page.text)
response = client.post(
"/account/edit",
data={
"csrf_token": csrf,
"mywhoosh_email": "max@mywhoosh.example",
"mywhoosh_password": "",
"garmin_email": "max@garmin.example",
"garmin_password": "",
},
follow_redirects=False,
)
assert response.status_code == 303
with client.app.state.session_factory() as session:
user = UserRepository(session).get(user_id)
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_account_update_can_set_new_password(client: TestClient) -> None:
user_id = create_user_via_admin(client)
account_login(client, email="max@mywhoosh.example", password="mw-secret")
edit_page = client.get("/account/edit")
csrf = extract_csrf(edit_page.text)
response = client.post(
"/account/edit",
data={
"csrf_token": csrf,
"mywhoosh_email": "max@mywhoosh.example",
"mywhoosh_password": "new-mw-secret",
"garmin_email": "max@garmin.example",
"garmin_password": "",
},
follow_redirects=False,
)
assert response.status_code == 303
with client.app.state.session_factory() as session:
user = UserRepository(session).get(user_id)
cipher = CredentialCipher(client.app.state.settings.credential_encryption_key)
assert cipher.decrypt(user.mywhoosh_password_enc) == "new-mw-secret"
def test_account_update_cannot_change_name_or_enabled(client: TestClient) -> None:
"""Self-service editing must not expose name/enabled -- those stay
administrative decisions, not something the account owner can flip."""
user_id = create_user_via_admin(client)
account_login(client, email="max@mywhoosh.example", password="mw-secret")
edit_page = client.get("/account/edit")
csrf = extract_csrf(edit_page.text)
client.post(
"/account/edit",
data={
"csrf_token": csrf,
"mywhoosh_email": "max@mywhoosh.example",
"mywhoosh_password": "",
"garmin_email": "max@garmin.example",
"garmin_password": "",
},
follow_redirects=False,
)
with client.app.state.session_factory() as session:
user = UserRepository(session).get(user_id)
assert user.name == "Max"
assert user.enabled is True
def test_account_update_persists_notification_preferences(client: TestClient) -> None:
user_id = create_user_via_admin(client)
account_login(client, email="max@mywhoosh.example", password="mw-secret")
edit_page = client.get("/account/edit")
csrf = extract_csrf(edit_page.text)
response = client.post(
"/account/edit",
data={
"csrf_token": csrf,
"mywhoosh_email": "max@mywhoosh.example",
"mywhoosh_password": "",
"garmin_email": "max@garmin.example",
"garmin_password": "",
"notify_email_enabled": "on",
"notification_email": "alerts@example.com",
},
follow_redirects=False,
)
assert response.status_code == 303
with client.app.state.session_factory() as session:
user = UserRepository(session).get(user_id)
assert user.notify_email_enabled is True
assert user.notification_email == "alerts@example.com"
def test_account_edit_rejects_invalid_csrf(client: TestClient) -> None:
create_user_via_admin(client)
account_login(client, email="max@mywhoosh.example", password="mw-secret")
response = client.post(
"/account/edit",
data={
"csrf_token": "invalid-token",
"mywhoosh_email": "max@mywhoosh.example",
"mywhoosh_password": "",
"garmin_email": "max@garmin.example",
"garmin_password": "",
},
)
assert response.status_code == 403
def test_logout_clears_session(client: TestClient) -> None:
create_user_via_admin(client)
account_login(client, email="max@mywhoosh.example", password="mw-secret")
page = client.get("/account")
csrf = extract_csrf(page.text)
response = client.post("/account-logout", data={"csrf_token": csrf}, follow_redirects=False)
assert response.status_code == 303
response = client.get("/account", follow_redirects=False)
assert response.status_code == 303
assert response.headers["location"] == "/account-login"
def test_cannot_view_other_users_account(client: TestClient) -> None:
"""Each self-service session is bound to the user_id captured at login;
another user created afterwards must not be reachable from it."""
create_user_via_admin(client, name="Max")
account_login(client, email="max@mywhoosh.example", password="mw-secret")
with client.app.state.session_factory() as session:
UserRepository(session).create(
name="Other",
enabled=True,
mywhoosh_email_enc="unused",
mywhoosh_password_enc="unused",
garmin_email_enc="unused",
garmin_password_enc="unused",
)
response = client.get("/account")
assert response.status_code == 200
assert "Max" in response.text
assert "Other" not in response.text
def test_account_sync_triggers_own_user_only(app, client: TestClient, fake_sync_manager) -> None:
user_id = create_user_via_admin(client)
account_login(client, email="max@mywhoosh.example", password="mw-secret")
app.state.sync_manager = fake_sync_manager
page = client.get("/account")
csrf = extract_csrf(page.text)
response = client.post("/account/sync", data={"csrf_token": csrf})
assert response.status_code == 200
assert fake_sync_manager.user_calls == [user_id]
def test_account_sync_reports_already_running(app, client: TestClient, fake_sync_manager) -> None:
create_user_via_admin(client)
account_login(client, email="max@mywhoosh.example", password="mw-secret")
app.state.sync_manager = fake_sync_manager
fake_sync_manager.raise_already_running = True
page = client.get("/account")
csrf = extract_csrf(page.text)
response = client.post("/account/sync", data={"csrf_token": csrf})
assert response.status_code == 409
assert "already running" in response.text.lower()
def test_account_sync_requires_login(client: TestClient) -> None:
response = client.post("/account/sync", data={"csrf_token": "whatever"}, follow_redirects=False)
assert response.status_code == 303
assert response.headers["location"] == "/account-login"
def test_account_sync_rejects_invalid_csrf(app, client: TestClient, fake_sync_manager) -> None:
create_user_via_admin(client)
account_login(client, email="max@mywhoosh.example", password="mw-secret")
app.state.sync_manager = fake_sync_manager
response = client.post("/account/sync", data={"csrf_token": "invalid-token"})
assert response.status_code == 403
assert fake_sync_manager.user_calls == []