1088 lines
32 KiB
Markdown
1088 lines
32 KiB
Markdown
# Foundation, Admin UI, and Data Layer Implementation Plan
|
|
|
|
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
|
|
|
**Goal:** Build the runnable FastAPI application shell with SQLite persistence, encrypted per-user credentials, single-password admin authentication, CSRF protection, local user CRUD, and Docker packaging.
|
|
|
|
**Architecture:** Use one FastAPI process with synchronous SQLAlchemy sessions for low-volume SQLite access, server-rendered Jinja2 templates, signed cookie sessions, and Fernet encryption for stored service credentials. Deployment secrets and configuration come only from environment variables; persistent state lives under `/data`.
|
|
|
|
**Tech Stack:** Python 3.12, FastAPI, Starlette sessions, SQLAlchemy 2.x, Pydantic Settings, cryptography/Fernet, Jinja2, python-multipart, pytest, Docker.
|
|
|
|
## Global Constraints
|
|
|
|
- The admin UI is intended for local-network use only.
|
|
- Use one admin password from `ADMIN_PASSWORD`; no username and no per-user web logins.
|
|
- Use SQLite for persistent application state.
|
|
- Store deployment secrets/configuration in environment variables, not in SQLite.
|
|
- Encrypt MyWhoosh and Garmin credentials before writing them to SQLite.
|
|
- Never return stored passwords to templates or API responses.
|
|
- Use `HttpOnly` and `SameSite=Lax` session cookies.
|
|
- Protect every state-changing web request with CSRF validation.
|
|
- Keep all persistent application data under `DATA_DIR`, defaulting to `/data`.
|
|
- Do not introduce Angular, React, Tailwind, Bootstrap, OAuth/OIDC, or an external queue in v1.
|
|
|
|
---
|
|
|
|
## File Structure
|
|
|
|
```text
|
|
pyproject.toml
|
|
Dockerfile
|
|
docker-compose.example.yml
|
|
.env.example
|
|
app/
|
|
__init__.py
|
|
main.py
|
|
config.py
|
|
auth/
|
|
__init__.py
|
|
admin.py
|
|
csrf.py
|
|
db/
|
|
__init__.py
|
|
models.py
|
|
session.py
|
|
repositories.py
|
|
security/
|
|
__init__.py
|
|
credentials.py
|
|
web/
|
|
__init__.py
|
|
routes.py
|
|
forms.py
|
|
templates/
|
|
base.html
|
|
login.html
|
|
dashboard.html
|
|
users/form.html
|
|
users/detail.html
|
|
static/
|
|
app.css
|
|
tests/
|
|
conftest.py
|
|
test_config.py
|
|
db/test_repositories.py
|
|
security/test_credentials.py
|
|
web/test_auth.py
|
|
web/test_users.py
|
|
```
|
|
|
|
## Task 1: Bootstrap configuration and application factory
|
|
|
|
**Files:**
|
|
- Create: `pyproject.toml`
|
|
- Create: `app/config.py`
|
|
- Create: `app/main.py`
|
|
- Create: `tests/test_config.py`
|
|
- Create: `tests/conftest.py`
|
|
|
|
**Interfaces:**
|
|
- Produces: `Settings`, `get_settings()`, `create_app(settings: Settings | None = None) -> FastAPI`.
|
|
- Later tasks consume `Settings.data_dir`, `Settings.database_url`, `Settings.admin_password`, `Settings.secret_key`, and `Settings.credential_encryption_key`.
|
|
|
|
- [ ] **Step 1: Write failing configuration tests**
|
|
|
|
```python
|
|
# tests/test_config.py
|
|
from pathlib import Path
|
|
|
|
from app.config import Settings
|
|
|
|
|
|
def test_settings_build_default_data_paths(tmp_path: Path) -> None:
|
|
settings = Settings(
|
|
ADMIN_PASSWORD="admin-secret",
|
|
SECRET_KEY="session-secret",
|
|
CREDENTIAL_ENCRYPTION_KEY="ZmFrZS1rZXktZm9yLXRlc3RzLW11c3QtYmUtNDQtY2hhcnM=",
|
|
DATA_DIR=str(tmp_path),
|
|
SYNC_INTERVAL_MINUTES=5,
|
|
)
|
|
|
|
assert settings.data_dir == tmp_path
|
|
assert settings.database_url == f"sqlite:///{tmp_path / 'app.db'}"
|
|
assert settings.tokens_dir == tmp_path / "tokens"
|
|
assert settings.activities_dir == tmp_path / "activities"
|
|
|
|
|
|
def test_sync_interval_must_be_positive(tmp_path: Path) -> None:
|
|
try:
|
|
Settings(
|
|
ADMIN_PASSWORD="admin-secret",
|
|
SECRET_KEY="session-secret",
|
|
CREDENTIAL_ENCRYPTION_KEY="ZmFrZS1rZXktZm9yLXRlc3RzLW11c3QtYmUtNDQtY2hhcnM=",
|
|
DATA_DIR=str(tmp_path),
|
|
SYNC_INTERVAL_MINUTES=0,
|
|
)
|
|
except ValueError:
|
|
return
|
|
raise AssertionError("Expected validation failure for non-positive interval")
|
|
```
|
|
|
|
- [ ] **Step 2: Run the tests and verify they fail**
|
|
|
|
Run: `pytest tests/test_config.py -v`
|
|
|
|
Expected: import/definition failure because `app.config.Settings` does not exist yet.
|
|
|
|
- [ ] **Step 3: Add project dependencies and implement `Settings`**
|
|
|
|
```toml
|
|
# pyproject.toml
|
|
[project]
|
|
name = "mywhoosh-garmin-sync"
|
|
version = "0.1.0"
|
|
requires-python = ">=3.12"
|
|
dependencies = [
|
|
"fastapi>=0.115,<1",
|
|
"uvicorn[standard]>=0.30,<1",
|
|
"sqlalchemy>=2.0,<3",
|
|
"pydantic-settings>=2.0,<3",
|
|
"cryptography>=43,<50",
|
|
"jinja2>=3.1,<4",
|
|
"python-multipart>=0.0.9,<1",
|
|
]
|
|
|
|
[project.optional-dependencies]
|
|
test = [
|
|
"pytest>=8,<9",
|
|
"httpx>=0.27,<1",
|
|
]
|
|
|
|
[tool.pytest.ini_options]
|
|
testpaths = ["tests"]
|
|
```
|
|
|
|
```python
|
|
# app/config.py
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
from pydantic import Field, PositiveInt, model_validator
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
model_config = SettingsConfigDict(env_file=".env", extra="ignore", case_sensitive=False)
|
|
|
|
admin_password: str = Field(alias="ADMIN_PASSWORD", min_length=1)
|
|
secret_key: str = Field(alias="SECRET_KEY", min_length=16)
|
|
credential_encryption_key: str = Field(alias="CREDENTIAL_ENCRYPTION_KEY", min_length=1)
|
|
data_dir: Path = Field(default=Path("/data"), alias="DATA_DIR")
|
|
database_url: str | None = Field(default=None, alias="DATABASE_URL")
|
|
sync_interval_minutes: PositiveInt = Field(default=5, alias="SYNC_INTERVAL_MINUTES")
|
|
|
|
@model_validator(mode="after")
|
|
def derive_paths(self) -> "Settings":
|
|
self.data_dir = self.data_dir.expanduser().resolve()
|
|
if self.database_url is None:
|
|
self.database_url = f"sqlite:///{self.data_dir / 'app.db'}"
|
|
return self
|
|
|
|
@property
|
|
def tokens_dir(self) -> Path:
|
|
return self.data_dir / "tokens"
|
|
|
|
@property
|
|
def activities_dir(self) -> Path:
|
|
return self.data_dir / "activities"
|
|
|
|
|
|
@lru_cache
|
|
def get_settings() -> Settings:
|
|
return Settings()
|
|
```
|
|
|
|
- [ ] **Step 4: Implement a minimal application factory**
|
|
|
|
```python
|
|
# app/main.py
|
|
from fastapi import FastAPI
|
|
|
|
from app.config import Settings, get_settings
|
|
|
|
|
|
def create_app(settings: Settings | None = None) -> FastAPI:
|
|
resolved = settings or get_settings()
|
|
resolved.data_dir.mkdir(parents=True, exist_ok=True)
|
|
resolved.tokens_dir.mkdir(parents=True, exist_ok=True)
|
|
resolved.activities_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
app = FastAPI(title="MyWhoosh Garmin Sync")
|
|
app.state.settings = resolved
|
|
|
|
@app.get("/healthz")
|
|
def healthz() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|
|
```
|
|
|
|
- [ ] **Step 5: Run tests and smoke-test the app factory**
|
|
|
|
Run: `pytest tests/test_config.py -v`
|
|
|
|
Expected: PASS.
|
|
|
|
Run: `python -c "from app.main import create_app; print(create_app)"`
|
|
|
|
Expected: prints the function object without configuration-time crashes.
|
|
|
|
- [ ] **Step 6: Commit**
|
|
|
|
```bash
|
|
git add pyproject.toml app/config.py app/main.py tests/test_config.py tests/conftest.py
|
|
git commit -m "feat: bootstrap FastAPI configuration"
|
|
```
|
|
|
|
## Task 2: Add SQLite models and repositories
|
|
|
|
**Files:**
|
|
- Create: `app/db/models.py`
|
|
- Create: `app/db/session.py`
|
|
- Create: `app/db/repositories.py`
|
|
- Create: `tests/db/test_repositories.py`
|
|
- Modify: `app/main.py`
|
|
|
|
**Interfaces:**
|
|
- Produces: `SyncUser`, `Activity`, `SyncRun`, `UserRepository`, `ActivityRepository`, `SyncRunRepository`, `create_db_engine()`, `create_session_factory()`.
|
|
- `Activity` must enforce unique `(user_id, mywhoosh_activity_id)`.
|
|
- To make resumability explicit, store both `status` and `last_completed_stage`; `status="failed"` does not erase the last durable stage.
|
|
|
|
- [ ] **Step 1: Write repository tests for isolated users and idempotent activities**
|
|
|
|
```python
|
|
# tests/db/test_repositories.py
|
|
from app.db.models import ActivityStatus, HealthState
|
|
|
|
|
|
def test_create_two_independent_users(db_session, user_repository) -> None:
|
|
first = user_repository.create(
|
|
name="Max",
|
|
enabled=True,
|
|
health_state=HealthState.HEALTHY,
|
|
mywhoosh_email_enc="mw-1",
|
|
mywhoosh_password_enc="mw-pw-1",
|
|
garmin_email_enc="g-1",
|
|
garmin_password_enc="g-pw-1",
|
|
)
|
|
second = user_repository.create(
|
|
name="Anna",
|
|
enabled=True,
|
|
health_state=HealthState.HEALTHY,
|
|
mywhoosh_email_enc="mw-2",
|
|
mywhoosh_password_enc="mw-pw-2",
|
|
garmin_email_enc="g-2",
|
|
garmin_password_enc="g-pw-2",
|
|
)
|
|
|
|
assert first.id != second.id
|
|
assert {u.name for u in user_repository.list_enabled()} == {"Max", "Anna"}
|
|
|
|
|
|
def test_activity_external_id_is_unique_per_user(user_repository, activity_repository) -> None:
|
|
user = user_repository.create(
|
|
name="Max",
|
|
enabled=True,
|
|
health_state=HealthState.HEALTHY,
|
|
mywhoosh_email_enc="a",
|
|
mywhoosh_password_enc="b",
|
|
garmin_email_enc="c",
|
|
garmin_password_enc="d",
|
|
)
|
|
created, inserted = activity_repository.get_or_create_discovered(
|
|
user_id=user.id,
|
|
mywhoosh_activity_id="mw-123",
|
|
activity_name="Morning Ride",
|
|
activity_timestamp=None,
|
|
)
|
|
same, inserted_again = activity_repository.get_or_create_discovered(
|
|
user_id=user.id,
|
|
mywhoosh_activity_id="mw-123",
|
|
activity_name="Morning Ride",
|
|
activity_timestamp=None,
|
|
)
|
|
|
|
assert inserted is True
|
|
assert inserted_again is False
|
|
assert created.id == same.id
|
|
assert same.status == ActivityStatus.DISCOVERED
|
|
```
|
|
|
|
- [ ] **Step 2: Run the repository tests and verify failure**
|
|
|
|
Run: `pytest tests/db/test_repositories.py -v`
|
|
|
|
Expected: imports fail because database modules are not implemented.
|
|
|
|
- [ ] **Step 3: Define enums and models**
|
|
|
|
```python
|
|
# app/db/models.py
|
|
from __future__ import annotations
|
|
|
|
import enum
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import Boolean, DateTime, Enum, ForeignKey, Integer, String, Text, UniqueConstraint
|
|
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
|
|
|
|
|
|
def utcnow() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
class HealthState(str, enum.Enum):
|
|
HEALTHY = "healthy"
|
|
SYNCING = "syncing"
|
|
DEGRADED = "degraded"
|
|
ACTION_REQUIRED = "action_required"
|
|
DISABLED = "disabled"
|
|
|
|
|
|
class ActivityStatus(str, enum.Enum):
|
|
DISCOVERED = "discovered"
|
|
DOWNLOADED = "downloaded"
|
|
CONVERTED = "converted"
|
|
IMPORTED = "imported"
|
|
DUPLICATE = "duplicate"
|
|
FAILED = "failed"
|
|
|
|
|
|
class SyncRunStatus(str, enum.Enum):
|
|
RUNNING = "running"
|
|
SUCCESS = "success"
|
|
PARTIAL = "partial"
|
|
FAILED = "failed"
|
|
|
|
|
|
class SyncUser(Base):
|
|
__tablename__ = "sync_users"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
name: Mapped[str] = mapped_column(String(120), nullable=False)
|
|
enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
|
|
health_state: Mapped[HealthState] = mapped_column(Enum(HealthState), nullable=False, default=HealthState.HEALTHY)
|
|
mywhoosh_state: Mapped[str] = mapped_column(String(32), nullable=False, default="unknown")
|
|
garmin_state: Mapped[str] = mapped_column(String(32), nullable=False, default="unknown")
|
|
action_reason: Mapped[str | None] = mapped_column(Text)
|
|
mywhoosh_email_enc: Mapped[str] = mapped_column(Text, nullable=False)
|
|
mywhoosh_password_enc: Mapped[str] = mapped_column(Text, nullable=False)
|
|
garmin_email_enc: Mapped[str] = mapped_column(Text, nullable=False)
|
|
garmin_password_enc: Mapped[str] = mapped_column(Text, nullable=False)
|
|
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
|
|
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow)
|
|
|
|
|
|
class Activity(Base):
|
|
__tablename__ = "activities"
|
|
__table_args__ = (UniqueConstraint("user_id", "mywhoosh_activity_id", name="uq_activity_user_mywhoosh"),)
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
user_id: Mapped[int] = mapped_column(ForeignKey("sync_users.id", ondelete="CASCADE"), nullable=False, index=True)
|
|
mywhoosh_activity_id: Mapped[str] = mapped_column(String(255), nullable=False)
|
|
activity_name: Mapped[str] = mapped_column(String(255), nullable=False)
|
|
activity_timestamp: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
|
|
source_fit_path: Mapped[str | None] = mapped_column(Text)
|
|
converted_fit_path: Mapped[str | None] = mapped_column(Text)
|
|
status: Mapped[ActivityStatus] = mapped_column(Enum(ActivityStatus), nullable=False, default=ActivityStatus.DISCOVERED)
|
|
last_completed_stage: Mapped[ActivityStatus] = mapped_column(Enum(ActivityStatus), nullable=False, default=ActivityStatus.DISCOVERED)
|
|
retryable: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
|
|
garmin_activity_id: Mapped[str | None] = mapped_column(String(255))
|
|
last_error: Mapped[str | None] = mapped_column(Text)
|
|
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
|
|
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow)
|
|
|
|
|
|
class SyncRun(Base):
|
|
__tablename__ = "sync_runs"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
user_id: Mapped[int] = mapped_column(ForeignKey("sync_users.id", ondelete="CASCADE"), nullable=False, index=True)
|
|
started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
|
|
finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
|
|
status: Mapped[SyncRunStatus] = mapped_column(Enum(SyncRunStatus), nullable=False, default=SyncRunStatus.RUNNING)
|
|
discovered_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
imported_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
skipped_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
failed_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
summary_error: Mapped[str | None] = mapped_column(Text)
|
|
```
|
|
|
|
- [ ] **Step 4: Implement session factory and focused repositories**
|
|
|
|
```python
|
|
# app/db/session.py
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.engine import Engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from app.db.models import Base
|
|
|
|
|
|
def create_db_engine(database_url: str) -> Engine:
|
|
connect_args = {"check_same_thread": False} if database_url.startswith("sqlite") else {}
|
|
return create_engine(database_url, connect_args=connect_args, future=True)
|
|
|
|
|
|
def create_session_factory(engine: Engine) -> sessionmaker[Session]:
|
|
return sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
|
|
|
|
|
|
def initialize_schema(engine: Engine) -> None:
|
|
Base.metadata.create_all(engine)
|
|
```
|
|
|
|
```python
|
|
# app/db/repositories.py
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.models import Activity, ActivityStatus, HealthState, SyncUser
|
|
|
|
|
|
class UserRepository:
|
|
def __init__(self, session: Session) -> None:
|
|
self.session = session
|
|
|
|
def create(self, **values) -> SyncUser:
|
|
user = SyncUser(**values)
|
|
self.session.add(user)
|
|
self.session.commit()
|
|
return user
|
|
|
|
def get(self, user_id: int) -> SyncUser | None:
|
|
return self.session.get(SyncUser, user_id)
|
|
|
|
def list_enabled(self) -> list[SyncUser]:
|
|
return list(self.session.scalars(select(SyncUser).where(SyncUser.enabled.is_(True)).order_by(SyncUser.id)))
|
|
|
|
|
|
class ActivityRepository:
|
|
def __init__(self, session: Session) -> None:
|
|
self.session = session
|
|
|
|
def get_or_create_discovered(
|
|
self,
|
|
*,
|
|
user_id: int,
|
|
mywhoosh_activity_id: str,
|
|
activity_name: str,
|
|
activity_timestamp: datetime | None,
|
|
) -> tuple[Activity, bool]:
|
|
existing = self.session.scalar(
|
|
select(Activity).where(
|
|
Activity.user_id == user_id,
|
|
Activity.mywhoosh_activity_id == mywhoosh_activity_id,
|
|
)
|
|
)
|
|
if existing is not None:
|
|
return existing, False
|
|
activity = Activity(
|
|
user_id=user_id,
|
|
mywhoosh_activity_id=mywhoosh_activity_id,
|
|
activity_name=activity_name,
|
|
activity_timestamp=activity_timestamp,
|
|
status=ActivityStatus.DISCOVERED,
|
|
last_completed_stage=ActivityStatus.DISCOVERED,
|
|
)
|
|
self.session.add(activity)
|
|
try:
|
|
self.session.commit()
|
|
except IntegrityError:
|
|
self.session.rollback()
|
|
existing = self.session.scalar(
|
|
select(Activity).where(
|
|
Activity.user_id == user_id,
|
|
Activity.mywhoosh_activity_id == mywhoosh_activity_id,
|
|
)
|
|
)
|
|
if existing is None:
|
|
raise
|
|
return existing, False
|
|
return activity, True
|
|
```
|
|
|
|
- [ ] **Step 5: Wire database initialization into `create_app` and add DB test fixtures**
|
|
|
|
```python
|
|
# add inside create_app in app/main.py
|
|
from app.db.session import create_db_engine, create_session_factory, initialize_schema
|
|
|
|
engine = create_db_engine(resolved.database_url)
|
|
initialize_schema(engine)
|
|
app.state.db_engine = engine
|
|
app.state.session_factory = create_session_factory(engine)
|
|
```
|
|
|
|
```python
|
|
# tests/conftest.py
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.db.models import Base
|
|
from app.db.repositories import ActivityRepository, UserRepository
|
|
|
|
|
|
@pytest.fixture
|
|
def db_session() -> Session:
|
|
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)
|
|
with factory() as session:
|
|
yield session
|
|
|
|
|
|
@pytest.fixture
|
|
def user_repository(db_session: Session) -> UserRepository:
|
|
return UserRepository(db_session)
|
|
|
|
|
|
@pytest.fixture
|
|
def activity_repository(db_session: Session) -> ActivityRepository:
|
|
return ActivityRepository(db_session)
|
|
```
|
|
|
|
- [ ] **Step 6: Run tests**
|
|
|
|
Run: `pytest tests/db/test_repositories.py -v`
|
|
|
|
Expected: PASS.
|
|
|
|
- [ ] **Step 7: Commit**
|
|
|
|
```bash
|
|
git add app/db app/main.py tests/conftest.py tests/db/test_repositories.py
|
|
git commit -m "feat: add SQLite persistence models"
|
|
```
|
|
|
|
## Task 3: Encrypt credentials at rest
|
|
|
|
**Files:**
|
|
- Create: `app/security/credentials.py`
|
|
- Create: `tests/security/test_credentials.py`
|
|
|
|
**Interfaces:**
|
|
- Produces: `CredentialCipher.encrypt(value: str) -> str`, `CredentialCipher.decrypt(token: str) -> str`.
|
|
- The constructor accepts exactly the value of `CREDENTIAL_ENCRYPTION_KEY`.
|
|
|
|
- [ ] **Step 1: Write failing encryption tests**
|
|
|
|
```python
|
|
# tests/security/test_credentials.py
|
|
from cryptography.fernet import Fernet
|
|
|
|
from app.security.credentials import CredentialCipher
|
|
|
|
|
|
def test_round_trip_and_ciphertext_does_not_contain_plaintext() -> None:
|
|
cipher = CredentialCipher(Fernet.generate_key().decode("ascii"))
|
|
encrypted = cipher.encrypt("secret-password")
|
|
|
|
assert "secret-password" not in encrypted
|
|
assert cipher.decrypt(encrypted) == "secret-password"
|
|
|
|
|
|
def test_empty_credentials_are_rejected() -> None:
|
|
cipher = CredentialCipher(Fernet.generate_key().decode("ascii"))
|
|
try:
|
|
cipher.encrypt("")
|
|
except ValueError:
|
|
return
|
|
raise AssertionError("empty secrets must be rejected")
|
|
```
|
|
|
|
- [ ] **Step 2: Run tests and verify failure**
|
|
|
|
Run: `pytest tests/security/test_credentials.py -v`
|
|
|
|
Expected: import failure.
|
|
|
|
- [ ] **Step 3: Implement the cipher**
|
|
|
|
```python
|
|
# app/security/credentials.py
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
|
|
|
|
class CredentialCipher:
|
|
def __init__(self, key: str) -> None:
|
|
try:
|
|
self._fernet = Fernet(key.encode("ascii"))
|
|
except Exception as exc:
|
|
raise ValueError("CREDENTIAL_ENCRYPTION_KEY must be a valid Fernet key") from exc
|
|
|
|
def encrypt(self, value: str) -> str:
|
|
if not value:
|
|
raise ValueError("credential value must not be empty")
|
|
return self._fernet.encrypt(value.encode("utf-8")).decode("ascii")
|
|
|
|
def decrypt(self, token: str) -> str:
|
|
try:
|
|
return self._fernet.decrypt(token.encode("ascii")).decode("utf-8")
|
|
except InvalidToken as exc:
|
|
raise ValueError("stored credential cannot be decrypted") from exc
|
|
```
|
|
|
|
- [ ] **Step 4: Run tests**
|
|
|
|
Run: `pytest tests/security/test_credentials.py -v`
|
|
|
|
Expected: PASS.
|
|
|
|
- [ ] **Step 5: Commit**
|
|
|
|
```bash
|
|
git add app/security/credentials.py tests/security/test_credentials.py
|
|
git commit -m "feat: encrypt stored service credentials"
|
|
```
|
|
|
|
## Task 4: Add admin login, signed session, and CSRF protection
|
|
|
|
**Files:**
|
|
- Create: `app/auth/admin.py`
|
|
- Create: `app/auth/csrf.py`
|
|
- Create: `app/web/routes.py`
|
|
- Create: `app/web/templates/base.html`
|
|
- Create: `app/web/templates/login.html`
|
|
- Create: `tests/web/test_auth.py`
|
|
- Modify: `app/main.py`
|
|
|
|
**Interfaces:**
|
|
- Produces: `require_admin(request)`, `ensure_csrf_token(request)`, `validate_csrf(request, submitted_token)`.
|
|
- Session key for authentication is `request.session["admin_authenticated"] is True`.
|
|
- Session key for CSRF is `request.session["csrf_token"]`.
|
|
|
|
- [ ] **Step 1: Write failing auth and CSRF tests**
|
|
|
|
```python
|
|
# tests/web/test_auth.py
|
|
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"] == "/"
|
|
```
|
|
|
|
- [ ] **Step 2: Run tests and verify failure**
|
|
|
|
Run: `pytest tests/web/test_auth.py -v`
|
|
|
|
Expected: route/import failures.
|
|
|
|
- [ ] **Step 3: Implement admin comparison and CSRF helpers**
|
|
|
|
```python
|
|
# app/auth/admin.py
|
|
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"})
|
|
```
|
|
|
|
```python
|
|
# app/auth/csrf.py
|
|
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")
|
|
```
|
|
|
|
- [ ] **Step 4: Add login/dashboard routes and templates**
|
|
|
|
```python
|
|
# app/web/routes.py
|
|
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": []})
|
|
```
|
|
|
|
- [ ] **Step 5: Install `SessionMiddleware` and include routes**
|
|
|
|
```python
|
|
# app/main.py additions
|
|
from starlette.middleware.sessions import SessionMiddleware
|
|
from app.web.routes import router as web_router
|
|
|
|
app.add_middleware(
|
|
SessionMiddleware,
|
|
secret_key=resolved.secret_key,
|
|
same_site="lax",
|
|
https_only=False,
|
|
)
|
|
app.include_router(web_router)
|
|
```
|
|
|
|
- [ ] **Step 6: Add the concrete FastAPI test client fixture and run auth tests**
|
|
|
|
```python
|
|
# tests/conftest.py additions
|
|
from cryptography.fernet import Fernet
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.config import Settings
|
|
from app.main import create_app
|
|
|
|
|
|
@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))
|
|
```
|
|
|
|
Run: `pytest tests/web/test_auth.py -v`
|
|
|
|
Expected: PASS.
|
|
|
|
- [ ] **Step 7: Commit**
|
|
|
|
```bash
|
|
git add app/auth app/web app/main.py tests/web/test_auth.py tests/conftest.py
|
|
git commit -m "feat: add local admin authentication"
|
|
```
|
|
|
|
## Task 5: Add user CRUD without exposing stored secrets
|
|
|
|
**Files:**
|
|
- Create: `app/web/forms.py`
|
|
- Create: `app/web/templates/users/form.html`
|
|
- Create: `app/web/templates/users/detail.html`
|
|
- Modify: `app/web/routes.py`
|
|
- Modify: `app/db/repositories.py`
|
|
- Create: `tests/web/test_users.py`
|
|
|
|
**Interfaces:**
|
|
- Produces routes: `GET /users/new`, `POST /users`, `GET /users/{id}`, `GET /users/{id}/edit`, `POST /users/{id}`.
|
|
- Empty password fields on edit preserve existing encrypted passwords.
|
|
- Templates receive only non-secret fields.
|
|
|
|
- [ ] **Step 1: Write failing user CRUD tests**
|
|
|
|
```python
|
|
# tests/web/test_users.py
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
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 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
|
|
```
|
|
|
|
- [ ] **Step 2: Run the test and verify failure**
|
|
|
|
Run: `pytest tests/web/test_users.py -v`
|
|
|
|
Expected: missing routes/form support.
|
|
|
|
- [ ] **Step 3: Extend repository update methods**
|
|
|
|
```python
|
|
# app/db/repositories.py additions
|
|
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
|
|
```
|
|
|
|
- [ ] **Step 4: Implement create/edit request handling with encrypted fields**
|
|
|
|
```python
|
|
# app/web/forms.py
|
|
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
|
|
```
|
|
|
|
In `app/web/routes.py`, construct `CredentialCipher(request.app.state.settings.credential_encryption_key)` and encrypt all four service values before repository writes. On edit, only replace an encrypted password if the submitted password is non-empty. Decrypt emails for display; never decrypt passwords for a template.
|
|
|
|
Use the exact update payload shape:
|
|
|
|
```python
|
|
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)
|
|
```
|
|
|
|
Every POST route must call `validate_csrf(request, csrf_token)` before changing state.
|
|
|
|
- [ ] **Step 5: Replace the hard-coded dashboard user list with repository data**
|
|
|
|
```python
|
|
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)},
|
|
)
|
|
```
|
|
|
|
- [ ] **Step 6: Run web tests**
|
|
|
|
Run: `pytest tests/web/test_users.py tests/web/test_auth.py -v`
|
|
|
|
Expected: PASS, including explicit assertion that passwords never occur in response HTML.
|
|
|
|
- [ ] **Step 7: Commit**
|
|
|
|
```bash
|
|
git add app/db/repositories.py app/web tests/web/test_users.py
|
|
git commit -m "feat: add encrypted sync user management"
|
|
```
|
|
|
|
## Task 6: Package the foundation as a local Docker service
|
|
|
|
**Files:**
|
|
- Create: `Dockerfile`
|
|
- Create: `docker-compose.example.yml`
|
|
- Create: `.env.example`
|
|
- Modify: `app/main.py`
|
|
|
|
**Interfaces:**
|
|
- Produces a container exposing FastAPI on port `8080` and persisting `/data`.
|
|
|
|
- [ ] **Step 1: Add deterministic startup command**
|
|
|
|
```dockerfile
|
|
# Dockerfile
|
|
FROM python:3.12-slim
|
|
WORKDIR /app
|
|
COPY pyproject.toml /app/
|
|
RUN pip install --no-cache-dir .
|
|
COPY app /app/app
|
|
RUN mkdir -p /data && chmod 700 /data
|
|
ENV DATA_DIR=/data
|
|
EXPOSE 8080
|
|
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"]
|
|
```
|
|
|
|
- [ ] **Step 2: Add example environment and compose file**
|
|
|
|
```env
|
|
# .env.example
|
|
ADMIN_PASSWORD=change-me
|
|
SECRET_KEY=replace-with-at-least-16-random-characters
|
|
CREDENTIAL_ENCRYPTION_KEY=replace-with-a-valid-fernet-key
|
|
SYNC_INTERVAL_MINUTES=5
|
|
DATA_DIR=/data
|
|
DATABASE_URL=sqlite:////data/app.db
|
|
```
|
|
|
|
```yaml
|
|
# docker-compose.example.yml
|
|
services:
|
|
sync:
|
|
build: .
|
|
env_file: .env
|
|
ports:
|
|
- "8080:8080"
|
|
volumes:
|
|
- ./data:/data
|
|
restart: unless-stopped
|
|
```
|
|
|
|
- [ ] **Step 3: Build the image**
|
|
|
|
Run: `docker build -t mywhoosh-garmin-sync:test .`
|
|
|
|
Expected: successful image build.
|
|
|
|
- [ ] **Step 4: Run a container smoke test**
|
|
|
|
Run with a real generated Fernet key and test-only secrets:
|
|
|
|
```bash
|
|
docker run --rm -d --name mywhoosh-garmin-sync-test \
|
|
-p 18080:8080 \
|
|
-e ADMIN_PASSWORD=admin-secret \
|
|
-e SECRET_KEY=0123456789abcdef0123456789abcdef \
|
|
-e CREDENTIAL_ENCRYPTION_KEY="$(python -c 'from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())')" \
|
|
-v "$PWD/.tmp-data:/data" \
|
|
mywhoosh-garmin-sync:test
|
|
```
|
|
|
|
Run: `curl -fsS http://127.0.0.1:18080/healthz`
|
|
|
|
Expected: `{"status":"ok"}`.
|
|
|
|
Then run: `docker stop mywhoosh-garmin-sync-test`
|
|
|
|
- [ ] **Step 5: Run the foundation regression suite**
|
|
|
|
Run: `pytest tests/test_config.py tests/db tests/security tests/web -v`
|
|
|
|
Expected: PASS.
|
|
|
|
- [ ] **Step 6: Commit**
|
|
|
|
```bash
|
|
git add Dockerfile docker-compose.example.yml .env.example app/main.py
|
|
git commit -m "build: package local admin service"
|
|
```
|