fix(settings): encrypt the AI provider token at rest

ai_config was missing from PROTECTED_SETTING_KEYS, so the AI provider's
api_token sat in plaintext in the settings table, and GET returned it in
full to the browser. It is now Fernet-encrypted like every other
integration config, and GET shows the token as ***set***. The AI clients
and the model-list fallback read the decrypted value. The fallback also
treats the ***set*** placeholder as "use the stored token".

Rows saved in plaintext before their key was protected used to stay that
way until the next edit. encrypt_plaintext_rows now encrypts them at
startup, for ai_config and any key added to the list later.
This commit is contained in:
2026-11-04 09:03:36 +01:00
parent d50c5144f4
commit 4bc3633551
7 changed files with 118 additions and 16 deletions
+5
View File
@@ -2887,6 +2887,11 @@ run). The synchronous `POST /eol-check` remains for API clients.
| Key | Purpose |
|---|---|
| `github_pat` | Optional GitHub token. Lifts GHSA **and** the MFSA/repo-advisory fetches from 60 to 5000 req/h. Fine-grained, no scopes needed (public data). |
| `ai_config` | AI provider, model, base URL and `api_token`. GET shows the token as `***set***`; the model list falls back to the stored one. |
Every key in `PROTECTED_SETTING_KEYS` (`app/auth/setting_crypto.py`) is
Fernet-encrypted on write. A row still in plaintext (saved before its key
joined the list) is encrypted at startup by `encrypt_plaintext_rows`.
## UI
+19 -1
View File
@@ -5,7 +5,8 @@ Setting rows whose `key` is in PROTECTED_SETTING_KEYS get their `value`
column encrypted with the AUTH_PROVIDER_CRYPTO_KEY Fernet key (same key
that protects TOTP secrets). Encrypted values are prefixed with
`ENC_PREFIX` so reads can detect ciphertext vs. legacy plaintext and
auto-migrate on the next write.
auto-migrate on the next write; `encrypt_plaintext_rows` (run at startup)
encrypts any still-plaintext row, e.g. after a key joins the list.
"""
from __future__ import annotations
@@ -37,6 +38,7 @@ PROTECTED_SETTING_KEYS: frozenset[str] = frozenset({
"dockhand_config",
"xenserver_config",
"github_pat",
"ai_config",
})
@@ -72,6 +74,22 @@ def read_setting_value(db: Session, key: str) -> Optional[str]:
return row.value
def encrypt_plaintext_rows(db: Session) -> int:
"""Encrypt protected rows still stored in plaintext. Returns how many."""
rows = (db.query(Setting)
.filter(Setting.key.in_(PROTECTED_SETTING_KEYS))
.filter(Setting.value.isnot(None), Setting.value != "")
.all())
pending = [r for r in rows if not r.value.startswith(ENC_PREFIX)]
for row in pending:
row.value = encrypt_value(row.value)
if pending:
db.commit()
logger.info("Encrypted %d plaintext setting(s): %s",
len(pending), ", ".join(r.key for r in pending))
return len(pending)
def write_setting_value(
db: Session,
key: str,
+4 -4
View File
@@ -11,7 +11,7 @@ import httpx
from fastapi import Depends
from sqlalchemy.orm import Session
from app.database import get_db
from app.models.setting import Setting
from app.auth.setting_crypto import read_setting_value
logger = logging.getLogger(__name__)
@@ -246,10 +246,10 @@ Structure your response as JSON ONLY:
self.client.close()
def get_ai_client(db: Session = Depends(get_db)) -> UniversalAIClient:
config_setting = db.query(Setting).filter(Setting.key == "ai_config").first()
if config_setting and config_setting.value:
stored = read_setting_value(db, "ai_config") # encrypted at rest
if stored:
try:
config = json.loads(config_setting.value)
config = json.loads(stored)
return UniversalAIClient.from_config(config)
except Exception:
logger.error("Invalid AI config in DB")
+4 -4
View File
@@ -8,7 +8,7 @@ from typing import Optional, Dict, Any, List
from sqlalchemy.orm import Session
from fastapi import Depends
from app.database import get_db
from app.models.setting import Setting
from app.auth.setting_crypto import read_setting_value
from app.integrations.ai_client import UniversalAIClient, AIError
logger = logging.getLogger(__name__)
@@ -21,10 +21,10 @@ def get_infomaniak_ai_client(db: Session = Depends(get_db)) -> UniversalAIClient
"""
Factory function for dependency injection.
"""
config_setting = db.query(Setting).filter(Setting.key == "ai_config").first()
if config_setting and config_setting.value:
stored = read_setting_value(db, "ai_config") # encrypted at rest
if stored:
try:
config = json.loads(config_setting.value)
config = json.loads(stored)
return UniversalAIClient.from_config(config)
except Exception:
logger.error("Invalid AI config in DB")
+10
View File
@@ -102,6 +102,16 @@ async def lifespan(app: FastAPI):
except Exception as e:
logger.error(f"Error creating initial data: {e}")
# Secrets saved before their key was protected sit in plaintext until the
# next edit — encrypt them now.
try:
from app.auth.setting_crypto import encrypt_plaintext_rows
db = SessionLocal()
encrypt_plaintext_rows(db)
db.close()
except Exception as e:
logger.error(f"Plaintext secret settings could not be encrypted: {e}")
# Custom CAs before the scheduler: the first sync must already trust them.
try:
from app.services.ca_trust_service import load_trust_bundle
+9 -7
View File
@@ -98,8 +98,8 @@ def _changed_fields(old: Optional[str], new: Optional[str]) -> Optional[List[str
def _audit_value(key: str, plain: Optional[str]) -> Optional[str]:
"""What an audit row may show of a setting value. Secret subfields are
masked for every key, not only protected ones: ai_config is unprotected
and carries api_token, and editors can read a resource-scoped trail."""
masked for every key, not only protected ones: an unprotected config may
still carry a token, and editors can read a resource-scoped trail."""
try:
data = json.loads(plain) if plain else None
except (json.JSONDecodeError, TypeError):
@@ -252,12 +252,14 @@ async def list_ai_models(
provider = (body.provider or "").strip().lower()
token = (body.api_token or "").strip()
if not token:
# Fall back to what is stored, so the dropdown works without retyping.
row = db.query(Setting).filter(Setting.key == "ai_config").first()
if row and row.value:
if not token or token == "***set***":
# Fall back to what is stored, so the dropdown works without retyping
# (the UI only ever sees the redacted placeholder).
stored = read_setting_value(db, "ai_config")
token = ""
if stored:
try:
token = (_json.loads(row.value) or {}).get("api_token") or ""
token = (_json.loads(stored) or {}).get("api_token") or ""
except (ValueError, TypeError):
token = ""
base = (body.base_url or "").strip().rstrip("/")
+67
View File
@@ -0,0 +1,67 @@
"""
ai_config carries the AI provider's api_token, so it is encrypted at rest like
every other integration config — and a row saved in plaintext before that is
encrypted on the next start, not left until somebody edits it.
"""
import json
import os
os.environ.setdefault("JWT_SECRET_KEY", "0" * 64)
import pytest # noqa: E402
from cryptography.fernet import Fernet # noqa: E402
from sqlalchemy import create_engine # noqa: E402
from sqlalchemy.orm import sessionmaker # noqa: E402
from sqlalchemy.pool import StaticPool # noqa: E402
import app.models # noqa: F401,E402
from app.auth import setting_crypto, totp # noqa: E402
from app.integrations.ai_client import get_ai_client # noqa: E402
from app.models.base import Base # noqa: E402
from app.models.setting import Setting # noqa: E402
from app.models.user import User, UserRole # noqa: E402
from app.routers.settings import SettingUpdate, get_setting, update_setting # noqa: E402
TOKEN = "sk-test-not-a-real-token"
CFG = {"provider": "openai", "api_token": TOKEN, "model": "gpt-x", "base_url": ""}
@pytest.fixture
def db(monkeypatch):
monkeypatch.setenv("AUTH_PROVIDER_CRYPTO_KEY", Fernet.generate_key().decode())
monkeypatch.setattr(totp, "_cached_key", None)
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
Base.metadata.create_all(engine)
session = sessionmaker(bind=engine)()
session.add(User(username="admin", email="a@example.com", role=UserRole.ADMIN, is_active=True))
session.commit()
return session
def _row(db):
return db.query(Setting).filter_by(key="ai_config").one()
def test_saved_token_is_ciphertext_and_redacted_on_read(db):
admin = db.query(User).one()
update_setting("ai_config", SettingUpdate(value=json.dumps(CFG)), db=db, current_user=admin)
assert TOKEN not in _row(db).value
assert _row(db).value.startswith(setting_crypto.ENC_PREFIX)
shown = json.loads(get_setting("ai_config", db=db, current_user=admin)["value"])
assert shown["api_token"] == "***set***" and shown["model"] == "gpt-x"
def test_the_ai_client_reads_the_encrypted_config(db):
setting_crypto.write_setting_value(db, "ai_config", json.dumps(CFG))
assert get_ai_client(db).api_key == TOKEN
def test_plaintext_row_from_before_is_encrypted_at_startup(db):
db.add(Setting(key="ai_config", value=json.dumps(CFG)))
db.add(Setting(key="notification_mode", value="digest")) # not a secret: untouched
db.commit()
assert setting_crypto.encrypt_plaintext_rows(db) == 1
assert _row(db).value.startswith(setting_crypto.ENC_PREFIX)
assert json.loads(setting_crypto.read_setting_value(db, "ai_config"))["api_token"] == TOKEN
assert db.query(Setting).filter_by(key="notification_mode").one().value == "digest"
assert setting_crypto.encrypt_plaintext_rows(db) == 0 # idempotent