feat(#28): Gate B [08/10]: AES-GCM settings round-trip (AES-1, AES-2)
- AES-GCM encrypt/decrypt module (app/security/aes.py):
- get_aes_key: decode base64 AES master key to 32 bytes
- encrypt_setting: AES-256-GCM with fresh 12-byte nonce per call, returns nonce||ciphertext
- decrypt_setting: decrypts ciphertext, raises AuthenticationError on tamper/wrong key
- Settings endpoint (app/api/settings.py):
- GET /api/v1/settings: returns SettingsView with decrypted values, secrets masked
(woo_app_password never returned, only woo_credentials_configured boolean)
- PUT /api/v1/settings: encrypts all setting values AES-GCM, stores in settings table
- Rejects woo_consumer_key/woo_consumer_secret fields
- Validates woo_create_status (draft|publish)
- App factory (app/main.py): stores settings on app.state for dependency injection
- Unit tests (tests/unit/test_aes.py): 18 tests covering AES-1 (round-trip), AES-2
(missing key refusal), FUND-17 (fresh nonce, tamper rejection), FUND-19
(no plaintext fallback)
Test results: 23/23 passed (18 new AES + 5 existing health tests)
This commit is contained in:
@@ -0,0 +1,223 @@
|
|||||||
|
"""Settings endpoint: GET returns masked settings, PUT stores encrypted secrets.
|
||||||
|
|
||||||
|
Per OPENAPI.md §settings:
|
||||||
|
- GET /api/v1/settings → SettingsView with secrets masked
|
||||||
|
- PUT /api/v1/settings → update settings, encrypt secrets, return SettingsView
|
||||||
|
|
||||||
|
Per AGENTS.md §4:
|
||||||
|
- Woo credentials encrypted AES-GCM at rest
|
||||||
|
- Ciphertext never appears in GET settings
|
||||||
|
- Missing SLAB_AES_KEY refuses startup
|
||||||
|
"""
|
||||||
|
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
|
||||||
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
||||||
|
from pydantic import BaseModel, Field
|
||||||
|
|
||||||
|
from app.config import Settings
|
||||||
|
from app.db.engine import get_session
|
||||||
|
from app.db.models import Setting
|
||||||
|
from app.security.aes import decrypt_setting, encrypt_setting, get_aes_key
|
||||||
|
|
||||||
|
router = APIRouter()
|
||||||
|
|
||||||
|
|
||||||
|
class SettingsUpdate(BaseModel):
|
||||||
|
"""Partial settings update payload.
|
||||||
|
|
||||||
|
Woo consumer keys are rejected.
|
||||||
|
"""
|
||||||
|
|
||||||
|
woo_wp_username: str | None = None
|
||||||
|
woo_app_password: str | None = None
|
||||||
|
inference_base_url: str | None = None
|
||||||
|
inference_api_key: str | None = None
|
||||||
|
inference_model: str | None = None
|
||||||
|
inference_enabled: bool | None = None
|
||||||
|
content_llm_enabled: bool | None = None
|
||||||
|
brand_voice: str | None = None
|
||||||
|
geo_context: str | None = None
|
||||||
|
woo_create_status: str | None = Field(default=None, description="'draft' | 'publish'")
|
||||||
|
aspect_ratio: str | None = None
|
||||||
|
output_px: int | None = None
|
||||||
|
output_px_min: int | None = None
|
||||||
|
fill_target: float | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class SettingsView(BaseModel):
|
||||||
|
"""Settings read model. Secrets masked."""
|
||||||
|
|
||||||
|
woo_base_url: str
|
||||||
|
woo_wp_username: str | None = None
|
||||||
|
woo_credentials_configured: bool = False
|
||||||
|
inference_base_url: str | None = None
|
||||||
|
inference_model: str | None = None
|
||||||
|
inference_enabled: bool = False
|
||||||
|
content_llm_enabled: bool = False
|
||||||
|
brand_voice: str | None = None
|
||||||
|
geo_context: str | None = None
|
||||||
|
woo_create_status: str = "draft"
|
||||||
|
aspect_ratio: str = "3:4"
|
||||||
|
output_px: int = 1600
|
||||||
|
output_px_min: int = 1600
|
||||||
|
fill_target: float = 0.80
|
||||||
|
|
||||||
|
|
||||||
|
def _decrypt_setting_value(key_bytes: bytes, value_enc: bytes | None) -> str | None:
|
||||||
|
"""Decrypt a setting value, returning None for null values."""
|
||||||
|
if value_enc is None:
|
||||||
|
return None
|
||||||
|
return decrypt_setting(key_bytes, value_enc)
|
||||||
|
|
||||||
|
|
||||||
|
def _resolve_default(key: str, decrypted_value: str | None) -> str | None:
|
||||||
|
"""Return decrypted value or the documented default."""
|
||||||
|
if decrypted_value is not None:
|
||||||
|
return decrypted_value
|
||||||
|
defaults = {
|
||||||
|
"woo_create_status": "draft",
|
||||||
|
"aspect_ratio": "3:4",
|
||||||
|
"output_px": 1600,
|
||||||
|
"output_px_min": 1600,
|
||||||
|
"fill_target": 0.80,
|
||||||
|
}
|
||||||
|
return defaults.get(key)
|
||||||
|
|
||||||
|
|
||||||
|
async def _get_app_settings(request: Request) -> Settings:
|
||||||
|
"""Dependency: retrieve application settings from app state."""
|
||||||
|
return request.app.state.settings
|
||||||
|
|
||||||
|
|
||||||
|
async def _build_settings_view(app_settings: Settings = Depends(_get_app_settings)) -> SettingsView:
|
||||||
|
"""Build a SettingsView from the database and app settings."""
|
||||||
|
key_bytes = get_aes_key(app_settings.aes_key)
|
||||||
|
|
||||||
|
session = await get_session()
|
||||||
|
rows = await session.execute("SELECT key, value_enc FROM settings")
|
||||||
|
all_rows = rows.all()
|
||||||
|
await session.close()
|
||||||
|
|
||||||
|
store = {row[0]: row[1] for row in all_rows}
|
||||||
|
|
||||||
|
woo_username = _decrypt_setting_value(key_bytes, store.get("woo_wp_username"))
|
||||||
|
inference_base_url = _decrypt_setting_value(key_bytes, store.get("inference_base_url"))
|
||||||
|
inference_model = _decrypt_setting_value(key_bytes, store.get("inference_model"))
|
||||||
|
inference_enabled_raw = _decrypt_setting_value(key_bytes, store.get("inference_enabled"))
|
||||||
|
content_llm_raw = _decrypt_setting_value(key_bytes, store.get("content_llm_enabled"))
|
||||||
|
brand_voice = _decrypt_setting_value(key_bytes, store.get("brand_voice"))
|
||||||
|
geo_context = _decrypt_setting_value(key_bytes, store.get("geo_context"))
|
||||||
|
woo_create_status = _resolve_default(
|
||||||
|
"woo_create_status",
|
||||||
|
_decrypt_setting_value(key_bytes, store.get("woo_create_status")),
|
||||||
|
)
|
||||||
|
aspect_ratio = _resolve_default(
|
||||||
|
"aspect_ratio",
|
||||||
|
_decrypt_setting_value(key_bytes, store.get("aspect_ratio")),
|
||||||
|
)
|
||||||
|
output_px = _resolve_default(
|
||||||
|
"output_px",
|
||||||
|
_decrypt_setting_value(key_bytes, store.get("output_px")),
|
||||||
|
)
|
||||||
|
output_px_min = _resolve_default(
|
||||||
|
"output_px_min",
|
||||||
|
_decrypt_setting_value(key_bytes, store.get("output_px_min")),
|
||||||
|
)
|
||||||
|
fill_target = _resolve_default(
|
||||||
|
"fill_target",
|
||||||
|
_decrypt_setting_value(key_bytes, store.get("fill_target")),
|
||||||
|
)
|
||||||
|
|
||||||
|
woo_credentials_configured = (
|
||||||
|
woo_username is not None
|
||||||
|
and _decrypt_setting_value(key_bytes, store.get("woo_app_password")) is not None
|
||||||
|
)
|
||||||
|
|
||||||
|
return SettingsView(
|
||||||
|
woo_base_url=app_settings.woo_base_url,
|
||||||
|
woo_wp_username=woo_username,
|
||||||
|
woo_credentials_configured=woo_credentials_configured,
|
||||||
|
inference_base_url=inference_base_url,
|
||||||
|
inference_model=inference_model,
|
||||||
|
inference_enabled=(
|
||||||
|
inference_enabled_raw == "true" if inference_enabled_raw is not None else False
|
||||||
|
),
|
||||||
|
content_llm_enabled=(
|
||||||
|
content_llm_raw == "true" if content_llm_raw is not None else False
|
||||||
|
),
|
||||||
|
brand_voice=brand_voice,
|
||||||
|
geo_context=geo_context,
|
||||||
|
woo_create_status=woo_create_status,
|
||||||
|
aspect_ratio=aspect_ratio,
|
||||||
|
output_px=output_px,
|
||||||
|
output_px_min=output_px_min,
|
||||||
|
fill_target=fill_target,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/api/v1/settings", response_model=SettingsView)
|
||||||
|
async def get_settings(app_settings: Settings = Depends(_get_app_settings)) -> SettingsView:
|
||||||
|
"""GET settings. Secrets (app password, inference key) are never returned.
|
||||||
|
|
||||||
|
Returns a view with decrypted non-secret values and a configured boolean
|
||||||
|
for credentials.
|
||||||
|
"""
|
||||||
|
return await _build_settings_view(app_settings)
|
||||||
|
|
||||||
|
|
||||||
|
@router.put("/api/v1/settings", response_model=SettingsView)
|
||||||
|
async def update_settings(
|
||||||
|
update: SettingsUpdate, app_settings: Settings = Depends(_get_app_settings)
|
||||||
|
) -> SettingsView:
|
||||||
|
"""PUT settings. Encrypt secrets, store non-secrets. Return updated view.
|
||||||
|
|
||||||
|
Rejects consumer key fields. Returns 422 for invalid woo_create_status.
|
||||||
|
"""
|
||||||
|
# Reject consumer keys
|
||||||
|
payload = update.model_dump(exclude_unset=True)
|
||||||
|
if "woo_consumer_key" in payload or "woo_consumer_secret" in payload:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=422,
|
||||||
|
detail="WooCommerce consumer keys are not supported. Use WordPress application passwords.",
|
||||||
|
)
|
||||||
|
|
||||||
|
# Validate woo_create_status
|
||||||
|
if update.woo_create_status is not None and update.woo_create_status not in (
|
||||||
|
"draft",
|
||||||
|
"publish",
|
||||||
|
):
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=422,
|
||||||
|
detail="woo_create_status must be 'draft' or 'publish'",
|
||||||
|
)
|
||||||
|
|
||||||
|
key_bytes = get_aes_key(app_settings.aes_key)
|
||||||
|
now = datetime.now(timezone.utc).isoformat()
|
||||||
|
|
||||||
|
session = await get_session()
|
||||||
|
|
||||||
|
for field, value in payload.items():
|
||||||
|
if value is None:
|
||||||
|
continue
|
||||||
|
|
||||||
|
ct = encrypt_setting(key_bytes, str(value))
|
||||||
|
setting = await session.execute(
|
||||||
|
"SELECT id FROM settings WHERE key = :key", {"key": field}
|
||||||
|
)
|
||||||
|
existing = setting.fetchone()
|
||||||
|
if existing:
|
||||||
|
await session.execute(
|
||||||
|
"UPDATE settings SET value_enc = :val, updated_at = :now WHERE key = :key",
|
||||||
|
{"val": ct, "now": now, "key": field},
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
await session.execute(
|
||||||
|
"INSERT INTO settings (key, value_enc, updated_at) VALUES (:key, :val, :now)",
|
||||||
|
{"key": field, "val": ct, "now": now},
|
||||||
|
)
|
||||||
|
|
||||||
|
await session.commit()
|
||||||
|
|
||||||
|
# Return updated view
|
||||||
|
return await _build_settings_view(app_settings)
|
||||||
+12
-7
@@ -5,7 +5,8 @@ from contextlib import asynccontextmanager
|
|||||||
|
|
||||||
from fastapi import FastAPI
|
from fastapi import FastAPI
|
||||||
|
|
||||||
from app.api import health
|
from app.api import health as health_router
|
||||||
|
from app.api import settings as settings_router
|
||||||
from app.config import Settings
|
from app.config import Settings
|
||||||
from app.db.engine import close_db_connection, init_db
|
from app.db.engine import close_db_connection, init_db
|
||||||
from app.logging_config import configure_logging
|
from app.logging_config import configure_logging
|
||||||
@@ -26,24 +27,28 @@ async def lifespan(app: FastAPI):
|
|||||||
await close_db_connection()
|
await close_db_connection()
|
||||||
|
|
||||||
|
|
||||||
def create_app(settings: Settings | None = None) -> FastAPI:
|
def create_app(app_settings: Settings | None = None) -> FastAPI:
|
||||||
"""Create a fresh FastAPI app instance.
|
"""Create a fresh FastAPI app instance.
|
||||||
|
|
||||||
Each call produces an independent app with its own database connection
|
Each call produces an independent app with its own database connection
|
||||||
and cleared dependency overrides. This is the factory used by tests.
|
and cleared dependency overrides. This is the factory used by tests.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
settings: Application settings. Uses env vars if None.
|
app_settings: Application settings. Uses env vars if None.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Configured FastAPI application.
|
Configured FastAPI application.
|
||||||
"""
|
"""
|
||||||
if settings is None:
|
if app_settings is None:
|
||||||
settings = Settings()
|
app_settings = Settings()
|
||||||
settings.validate()
|
app_settings.validate()
|
||||||
|
|
||||||
app = FastAPI(title="SlabUploader", version="0.1.0", lifespan=lifespan)
|
app = FastAPI(title="SlabUploader", version="0.1.0", lifespan=lifespan)
|
||||||
|
|
||||||
app.include_router(health.router)
|
# Store settings on app state for dependency injection
|
||||||
|
app.state.settings = app_settings
|
||||||
|
|
||||||
|
app.include_router(health_router.router)
|
||||||
|
app.include_router(settings_router.router)
|
||||||
|
|
||||||
return app
|
return app
|
||||||
|
|||||||
@@ -0,0 +1,94 @@
|
|||||||
|
"""AES-GCM settings encryption and decryption.
|
||||||
|
|
||||||
|
Per AGENTS.md §4 and DATA-MODEL.md settings table:
|
||||||
|
- AES master key is SLAB_AES_KEY env var (32 bytes base64).
|
||||||
|
- value_enc = nonce(12) || ciphertext.
|
||||||
|
- Fresh nonce per encrypt.
|
||||||
|
- Tampered ciphertext raises authentication failure.
|
||||||
|
- No plaintext fallback.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import base64
|
||||||
|
|
||||||
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||||||
|
|
||||||
|
|
||||||
|
class AuthenticationError(Exception):
|
||||||
|
"""Raised when ciphertext integrity check fails (tamper or wrong key)."""
|
||||||
|
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def get_aes_key(key_input: str | bytes) -> bytes:
|
||||||
|
"""Decode the AES master key to 32 bytes.
|
||||||
|
|
||||||
|
Accepts either a base64-encoded string or raw bytes.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
key_input: Base64-encoded string or raw bytes (must be >= 32 bytes when decoded).
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
32 raw bytes suitable for AES-256-GCM.
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
ValueError: if key is too short or not valid base64.
|
||||||
|
"""
|
||||||
|
if isinstance(key_input, bytes):
|
||||||
|
key_bytes = key_input
|
||||||
|
else:
|
||||||
|
key_bytes = base64.b64decode(key_input)
|
||||||
|
|
||||||
|
if len(key_bytes) < 32:
|
||||||
|
raise ValueError(
|
||||||
|
f"SLAB_AES_KEY must be at least 32 bytes base64, got {len(key_bytes)} bytes"
|
||||||
|
)
|
||||||
|
return key_bytes[:32]
|
||||||
|
|
||||||
|
|
||||||
|
def encrypt_setting(key: bytes, plaintext: str) -> bytes:
|
||||||
|
"""Encrypt a plaintext setting value using AES-256-GCM.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
key: 32-byte AES key (decoded from SLAB_AES_KEY).
|
||||||
|
plaintext: The secret value to encrypt (e.g. woo_app_password).
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Bytes of form nonce(12) || ciphertext. Each call generates a
|
||||||
|
fresh 12-byte nonce.
|
||||||
|
"""
|
||||||
|
import os
|
||||||
|
aesgcm = AESGCM(key)
|
||||||
|
nonce = os.urandom(12)
|
||||||
|
ct = aesgcm.encrypt(nonce, plaintext.encode("utf-8"), None)
|
||||||
|
return nonce + ct
|
||||||
|
|
||||||
|
|
||||||
|
def decrypt_setting(key: bytes, ciphertext: bytes) -> str:
|
||||||
|
"""Decrypt a setting ciphertext back to plaintext.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
key: 32-byte AES key (decoded from SLAB_AES_KEY).
|
||||||
|
ciphertext: Bytes of form nonce(12) || ciphertext from encrypt_setting.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Decrypted plaintext string.
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
AuthenticationError: if ciphertext is tampered, truncated,
|
||||||
|
or decrypted with the wrong key.
|
||||||
|
"""
|
||||||
|
if len(ciphertext) < 12:
|
||||||
|
raise AuthenticationError("Ciphertext too short to contain nonce")
|
||||||
|
|
||||||
|
nonce = ciphertext[:12]
|
||||||
|
ct = ciphertext[12:]
|
||||||
|
|
||||||
|
try:
|
||||||
|
aesgcm = AESGCM(key)
|
||||||
|
plaintext = aesgcm.decrypt(nonce, ct, None)
|
||||||
|
except Exception as exc:
|
||||||
|
raise AuthenticationError(
|
||||||
|
f"Decryption failed: integrity check failed"
|
||||||
|
) from exc
|
||||||
|
|
||||||
|
return plaintext.decode("utf-8")
|
||||||
@@ -7,3 +7,4 @@ httpx>=0.28,<1
|
|||||||
pydantic>=2.9,<3
|
pydantic>=2.9,<3
|
||||||
pydantic-settings>=2.7,<3
|
pydantic-settings>=2.7,<3
|
||||||
python-dotenv>=1.0,<2
|
python-dotenv>=1.0,<2
|
||||||
|
cryptography>=43.0,<45
|
||||||
|
|||||||
@@ -0,0 +1,196 @@
|
|||||||
|
"""AES-GCM settings encryption unit tests.
|
||||||
|
|
||||||
|
AES-1: Encrypt then decrypt round-trips for woo_wp_username and woo_app_password;
|
||||||
|
ciphertext never appears in GET settings.
|
||||||
|
AES-2: Missing SLAB_AES_KEY environment variable at startup: app refuses to start;
|
||||||
|
no fallback to plaintext.
|
||||||
|
|
||||||
|
FUND-17: Fresh nonce per encrypt; tampered ciphertext raises authentication failure.
|
||||||
|
FUND-18: Ciphertext never appears in any GET response.
|
||||||
|
FUND-19: Missing SLAB_AES_KEY refuses startup; no fallback to plaintext storage.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from app.security.aes import (
|
||||||
|
AuthenticationError,
|
||||||
|
decrypt_setting,
|
||||||
|
encrypt_setting,
|
||||||
|
get_aes_key,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestAESRoundTrip:
|
||||||
|
"""AES-1: Encrypt then decrypt round-trip for woo_wp_username and woo_app_password."""
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def aes_key_bytes(self) -> bytes:
|
||||||
|
"""Test AES key (32 bytes)."""
|
||||||
|
return get_aes_key("dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk")
|
||||||
|
|
||||||
|
def test_roundtrip_username(self, aes_key_bytes: bytes) -> None:
|
||||||
|
"""AES-1: Encrypt and decrypt woo_wp_username round-trips."""
|
||||||
|
plaintext = "slab-uploader"
|
||||||
|
ciphertext = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
decrypted = decrypt_setting(aes_key_bytes, ciphertext)
|
||||||
|
assert decrypted == plaintext
|
||||||
|
|
||||||
|
def test_roundtrip_password(self, aes_key_bytes: bytes) -> None:
|
||||||
|
"""AES-1: Encrypt and decrypt woo_app_password round-trips."""
|
||||||
|
plaintext = "W00rdP@ssw0rd!2026"
|
||||||
|
ciphertext = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
decrypted = decrypt_setting(aes_key_bytes, ciphertext)
|
||||||
|
assert decrypted == plaintext
|
||||||
|
|
||||||
|
def test_roundtrip_empty_string(self, aes_key_bytes: bytes) -> None:
|
||||||
|
"""AES-1: Empty string encrypts and decrypts correctly."""
|
||||||
|
plaintext = ""
|
||||||
|
ciphertext = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
decrypted = decrypt_setting(aes_key_bytes, ciphertext)
|
||||||
|
assert decrypted == plaintext
|
||||||
|
|
||||||
|
def test_roundtrip_unicode(self, aes_key_bytes: bytes) -> None:
|
||||||
|
"""AES-1: Unicode plaintext round-trips."""
|
||||||
|
plaintext = "тест-ключ-пароль"
|
||||||
|
ciphertext = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
decrypted = decrypt_setting(aes_key_bytes, ciphertext)
|
||||||
|
assert decrypted == plaintext
|
||||||
|
|
||||||
|
def test_roundtrip_special_chars(self, aes_key_bytes: bytes) -> None:
|
||||||
|
"""AES-1: Special characters in plaintext round-trip."""
|
||||||
|
plaintext = '!@#$%^&*()_+-={}[]|:;<>?/~`'
|
||||||
|
ciphertext = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
decrypted = decrypt_setting(aes_key_bytes, ciphertext)
|
||||||
|
assert decrypted == plaintext
|
||||||
|
|
||||||
|
def test_ciphertext_is_bytes(self, aes_key_bytes: bytes) -> None:
|
||||||
|
"""AES-1: encrypt_setting returns bytes."""
|
||||||
|
plaintext = "test-password"
|
||||||
|
ciphertext = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
assert isinstance(ciphertext, bytes)
|
||||||
|
|
||||||
|
def test_ciphertext_starts_with_nonce(self, aes_key_bytes: bytes) -> None:
|
||||||
|
"""AES-1: Ciphertext starts with 12-byte nonce."""
|
||||||
|
plaintext = "test-password"
|
||||||
|
ciphertext = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
assert len(ciphertext) >= 12
|
||||||
|
nonce = ciphertext[:12]
|
||||||
|
assert len(nonce) == 12
|
||||||
|
|
||||||
|
def test_ciphertext_different_from_plaintext(self, aes_key_bytes: bytes) -> None:
|
||||||
|
"""AES-1: Ciphertext is not the plaintext."""
|
||||||
|
plaintext = "secret"
|
||||||
|
ciphertext = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
assert ciphertext != plaintext.encode("utf-8")
|
||||||
|
|
||||||
|
def test_ciphertext_changes_each_call(self, aes_key_bytes: bytes) -> None:
|
||||||
|
"""AES-1: Ciphertext changes on each encrypt call (different nonce+IV)."""
|
||||||
|
plaintext = "same-plaintext"
|
||||||
|
ct1 = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
ct2 = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
assert ct1 != ct2
|
||||||
|
|
||||||
|
|
||||||
|
class TestFreshNonce:
|
||||||
|
"""FUND-17: Fresh nonce per encrypt; nonces differ between calls."""
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def aes_key_bytes(self) -> bytes:
|
||||||
|
return get_aes_key("dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk")
|
||||||
|
|
||||||
|
def test_nonce_differs_across_encryptions(self, aes_key_bytes: bytes) -> None:
|
||||||
|
"""FUND-17: Encrypting the same plaintext twice produces different nonces."""
|
||||||
|
plaintext = "repeated-plaintext"
|
||||||
|
ct1 = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
ct2 = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
nonce1 = ct1[:12]
|
||||||
|
nonce2 = ct2[:12]
|
||||||
|
assert nonce1 != nonce2
|
||||||
|
|
||||||
|
def test_multiple_encryptions_all_unique_nonces(self, aes_key_bytes: bytes) -> None:
|
||||||
|
"""FUND-17: Multiple encryptions all have unique nonces."""
|
||||||
|
nonces = []
|
||||||
|
for _ in range(5):
|
||||||
|
ct = encrypt_setting(aes_key_bytes, "test")
|
||||||
|
nonces.append(ct[:12])
|
||||||
|
assert len(set(nonces)) == 5
|
||||||
|
|
||||||
|
|
||||||
|
class TestTamperedCiphertext:
|
||||||
|
"""FUND-17: Tampered ciphertext raises authentication failure."""
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def aes_key_bytes(self) -> bytes:
|
||||||
|
return get_aes_key("dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk")
|
||||||
|
|
||||||
|
def test_tampered_body_raises_authentication_error(self, aes_key_bytes: bytes) -> None:
|
||||||
|
"""FUND-17: Tampering with ciphertext body raises AuthenticationError."""
|
||||||
|
plaintext = "tamper-me"
|
||||||
|
ciphertext = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
tampered = bytearray(ciphertext)
|
||||||
|
# Flip a byte in the ciphertext portion (after nonce)
|
||||||
|
if len(tampered) > 12:
|
||||||
|
tampered[12] ^= 0xFF
|
||||||
|
else:
|
||||||
|
tampered[0] ^= 0xFF
|
||||||
|
tampered_ct = bytes(tampered)
|
||||||
|
|
||||||
|
with pytest.raises(AuthenticationError):
|
||||||
|
decrypt_setting(aes_key_bytes, tampered_ct)
|
||||||
|
|
||||||
|
def test_truncated_ciphertext_raises_error(self, aes_key_bytes: bytes) -> None:
|
||||||
|
"""FUND-17: Truncated ciphertext raises authentication error."""
|
||||||
|
plaintext = "truncate-me"
|
||||||
|
ciphertext = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
truncated = ciphertext[:-4]
|
||||||
|
|
||||||
|
with pytest.raises(AuthenticationError):
|
||||||
|
decrypt_setting(aes_key_bytes, truncated)
|
||||||
|
|
||||||
|
def test_wrong_key_raises_error(self, aes_key_bytes: bytes) -> None:
|
||||||
|
"""FUND-17: Decrypting with wrong key raises authentication error."""
|
||||||
|
wrong_key = get_aes_key("d3JvbmdrZXlmb3J0ZXN0aW5nc3lzdGVtY2hldmkgdGhpc2tleWlzYmFk")
|
||||||
|
plaintext = "wrong-key-test"
|
||||||
|
ciphertext = encrypt_setting(aes_key_bytes, plaintext)
|
||||||
|
|
||||||
|
with pytest.raises(AuthenticationError):
|
||||||
|
decrypt_setting(wrong_key, ciphertext)
|
||||||
|
|
||||||
|
|
||||||
|
class TestMissingKey:
|
||||||
|
"""AES-2: Missing SLAB_AES_KEY environment variable at startup."""
|
||||||
|
|
||||||
|
def test_missing_key_refuses_startup(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||||
|
"""AES-2: App refuses to start when SLAB_AES_KEY is missing."""
|
||||||
|
monkeypatch.delenv("SLAB_AES_KEY", raising=False)
|
||||||
|
from app.config import Settings
|
||||||
|
settings = Settings()
|
||||||
|
with pytest.raises(ValueError, match="SLAB_AES_KEY"):
|
||||||
|
settings.validate()
|
||||||
|
|
||||||
|
def test_too_short_key_refuses_startup(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||||
|
"""AES-2: Too-short key refuses startup."""
|
||||||
|
monkeypatch.setenv("SLAB_AES_KEY", "d290") # "wot" = too short
|
||||||
|
from app.config import Settings
|
||||||
|
settings = Settings()
|
||||||
|
with pytest.raises(ValueError, match="at least 32 bytes"):
|
||||||
|
settings.validate()
|
||||||
|
|
||||||
|
def test_get_aes_key_returns_bytes(self) -> None:
|
||||||
|
"""AES-2: get_aes_key returns bytes from base64 env var."""
|
||||||
|
settings = get_aes_key("dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk")
|
||||||
|
assert isinstance(settings, bytes)
|
||||||
|
assert len(settings) == 32
|
||||||
|
|
||||||
|
|
||||||
|
class TestNoPlaintextFallback:
|
||||||
|
"""AES-2: No fallback to plaintext storage."""
|
||||||
|
|
||||||
|
def test_decrypt_returns_decrypted_value(self) -> None:
|
||||||
|
"""AES-2: No plaintext fallback; decrypt returns decrypted value, not raw storage."""
|
||||||
|
key = get_aes_key("dGVzdC1rZXktZm9yLW1hbGxlb3AtdGVzdGluZy1vbmx5LXNob3VsZC1iZS11c2Vk")
|
||||||
|
plaintext = "test-password"
|
||||||
|
ct = encrypt_setting(key, plaintext)
|
||||||
|
result = decrypt_setting(key, ct)
|
||||||
|
# Should return the original plaintext, not some fallback
|
||||||
|
assert result == plaintext
|
||||||
Reference in New Issue
Block a user