menzeljandClaude Opus 5
60a7ccff93
The repo had no tests, no lint config, and a CI that went straight from push
to docker push. That is the reason F1 could ship: authorization lives in the
routers, each of 171 routes independently picks require_admin or
get_current_user, and nothing checked the choice was right.
670 tests, no Docker daemon needed. The app is driven through TestClient
without entering it as a context manager, which skips the lifespan — no
background loops, no socket — and conftest points DATA_DIR/STACKS_DIR at a
temp directory before anything is imported.
test_route_authorization.py is the load-bearing one. Rather than 171 implied
decisions it states the policy once — every route requires admin unless it is
listed in USER_READABLE or PUBLIC — and fails on any route that disagrees. A
new route defaults to admin, which is the safe direction; what it catches is a
route written with get_current_user that nobody weighed against "can this
return a credential". Writing the allowlist meant auditing all 53 user-readable
routes, which turned up one more leak: GET /api/templates/{id} returns a
template's env, and "save stack as template" snapshots the stack's real .env
into it. Now admin-only; the listing stays open.
test_agent_authorization.py pins the same invariant on the agent, where the
whole access model is one shared token declared per route and a single
forgotten Depends(verify_token) would hand over the host.
Both were checked by reintroducing the bug: re-opening /api/files/read fails
three tests with actionable messages, dropping a token guard fails two.
test_bundled_templates.py covers the 83 templates — parse, image per service,
.env.example in sync with what compose reads, every bind-mounted file actually
shipped, and no working default password. It found one on its first run:
authentik shipped PG_PASS=change-me and AUTHENTIK_SECRET_KEY=change-me against
a compose that marks both required, so the stack would have come up with a
known password instead of refusing to start. Fixed.
The rest ports the ad-hoc harnesses from 0.44.0 into permanent tests (crypto
round-trip incl. plaintext passthrough and key-loss handling, the browse
sandbox) and covers compose_service's slug/status/file handling and
secret_service's name validation.
ruff is configured as a floor, not a style bar: F, E9 and B only. Import
sorting is deliberately out — it is style, and enabling it would rewrite the
imports of nine files that have nothing else wrong. The 12 findings it did have
are fixed here (unused imports, an unused local, four raise-without-from that
were swallowing exception context).
CI now runs check (ruff, pytest, tsc) and only builds if it passes.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Dk43rmEeRfYi5wsLDbmfyG
114 lines
4.0 KiB
Python
114 lines
4.0 KiB
Python
"""Encryption of the secrets that have to live in the database.
|
|
|
|
Backup-destination credentials can't be files on disk — background jobs need
|
|
them — so they are encrypted at rest with a key derived from ``SECRET_KEY``.
|
|
The behaviour that matters beyond the round-trip: rows written before this
|
|
existed are plaintext and must keep working, and a changed ``SECRET_KEY`` must
|
|
produce a clear error rather than silent garbage.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def crypto():
|
|
from services import crypto_service
|
|
|
|
return crypto_service
|
|
|
|
|
|
def test_round_trip(crypto):
|
|
payload = json.dumps({"host": "nas.lan", "password": "hunter2"})
|
|
encrypted = crypto.encrypt(payload)
|
|
assert encrypted.startswith(crypto.PREFIX)
|
|
assert "hunter2" not in encrypted
|
|
assert crypto.decrypt(encrypted) == payload
|
|
|
|
|
|
def test_encrypting_twice_is_a_no_op(crypto):
|
|
"""Guards the migration, which may run over a mix of both forms."""
|
|
once = crypto.encrypt("secret")
|
|
assert crypto.encrypt(once) == once
|
|
|
|
|
|
def test_plaintext_rows_are_passed_through(crypto):
|
|
"""Rows written before encryption existed must not break on read."""
|
|
legacy = '{"host": "old.example"}'
|
|
assert crypto.is_encrypted(legacy) is False
|
|
assert crypto.decrypt(legacy) == legacy
|
|
|
|
|
|
def test_empty_values(crypto):
|
|
assert crypto.decrypt(crypto.encrypt("")) == ""
|
|
assert crypto.decrypt("") == ""
|
|
|
|
|
|
def test_a_changed_secret_key_raises_a_useful_error(crypto, monkeypatch):
|
|
from config import settings
|
|
|
|
encrypted = crypto.encrypt("secret")
|
|
monkeypatch.setattr(settings, "SECRET_KEY", "a-completely-different-key", raising=False)
|
|
with pytest.raises(crypto.DecryptError) as excinfo:
|
|
crypto.decrypt(encrypted)
|
|
assert "SECRET_KEY" in str(excinfo.value)
|
|
|
|
|
|
def test_destination_config_round_trips_through_the_service(monkeypatch):
|
|
"""The service-level pair the routers actually use."""
|
|
from models.backup_destination import BackupDestination
|
|
from services import backup_destination_service as svc
|
|
|
|
config = {"host": "nas.lan", "username": "backup", "password": "hunter2"}
|
|
stored = svc.dump_config(config)
|
|
assert "hunter2" not in stored
|
|
dest = BackupDestination(name="nas", type="sftp", config=stored)
|
|
assert svc.parse_config(dest) == config
|
|
|
|
|
|
def test_undecryptable_destination_reads_as_unconfigured(monkeypatch):
|
|
"""A lost key must not take the destinations list down with a 500.
|
|
|
|
The destination shows up empty (and the reason is logged) so the rest of the
|
|
UI keeps working and the user can re-enter the credentials.
|
|
"""
|
|
from config import settings
|
|
from models.backup_destination import BackupDestination
|
|
from services import backup_destination_service as svc
|
|
|
|
dest = BackupDestination(name="nas", type="sftp", config=svc.dump_config({"host": "x"}))
|
|
monkeypatch.setattr(settings, "SECRET_KEY", "yet-another-key", raising=False)
|
|
assert svc.parse_config(dest) == {}
|
|
|
|
|
|
def test_migration_encrypts_plaintext_rows_and_is_idempotent(db):
|
|
from sqlmodel import Session, select
|
|
|
|
from database import engine
|
|
from models.backup_destination import BackupDestination
|
|
from services import backup_destination_service as svc
|
|
from services import crypto_service
|
|
|
|
with Session(engine) as session:
|
|
session.add(
|
|
BackupDestination(
|
|
name="legacy-plaintext",
|
|
type="sftp",
|
|
config='{"host": "legacy.example", "password": "in-the-clear"}',
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
assert svc.migrate_plaintext_configs(session) >= 1
|
|
row = session.exec(
|
|
select(BackupDestination).where(BackupDestination.name == "legacy-plaintext")
|
|
).one()
|
|
assert crypto_service.is_encrypted(row.config)
|
|
assert "in-the-clear" not in row.config
|
|
assert svc.parse_config(row)["password"] == "in-the-clear"
|
|
|
|
# Second pass finds nothing left to do.
|
|
assert svc.migrate_plaintext_configs(session) == 0
|