Spark
735d7cdfd0
- Tables forge_connections/users/workspaces (+ membres, projets, états, audit) ; migration des anciennes bases avec portées (workspace_id, connection_id) par défaut (1, 1). - Jetons chiffrés Fernet (KANBANTE_MASTER_KEY ou clé générée en base), jamais réaffichés. - ForgeRegistry : connexion héritée créée depuis l'environnement, espace Principal, instantanés préfixés par connexion. - GitHubClient complet (Bearer, pagination Link, étiquettes par nom, PATCH archived, branches via git/refs). - Paramètres surchargeables en base (résolveur config, P3).
71 lines
2.0 KiB
Python
71 lines
2.0 KiB
Python
"""Encryption at rest for forge tokens (0.24.0, P0/P2).
|
|
|
|
Forge access tokens are the keys to the kingdom: they are
|
|
stored Fernet-encrypted (AES-128-CBC + HMAC) in
|
|
forge_connections.token_encrypted and never shown again after
|
|
entry (the admin UI only ever displays a mask).
|
|
|
|
Master key resolution (decision D5):
|
|
1. KANBANTE_MASTER_KEY (urlsafe base64, 32 bytes) — recommended;
|
|
2. otherwise a random key generated ONCE and kept in kv_store
|
|
("master_key_generated"): as safe as the database file
|
|
itself, zero configuration, but it travels with DB backups —
|
|
set the env var for anything stricter.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import secrets as _secrets
|
|
|
|
from cryptography.fernet import Fernet
|
|
|
|
from app import models
|
|
|
|
logger = logging.getLogger("kanbante.crypto")
|
|
|
|
_fernet: Fernet | None = None
|
|
|
|
|
|
def _master_key() -> bytes:
|
|
env_key = os.getenv("KANBANTE_MASTER_KEY", "").strip()
|
|
if env_key:
|
|
return env_key.encode("utf-8")
|
|
stored = models.get_kv("master_key_generated")
|
|
if stored:
|
|
return stored.encode("utf-8")
|
|
generated = Fernet.generate_key()
|
|
models.set_kv("master_key_generated", generated.decode("utf-8"))
|
|
logger.warning(
|
|
"KANBANTE_MASTER_KEY not set: generated a master key and "
|
|
"stored it in the database (kv_store). Set the env var to "
|
|
"control it explicitly."
|
|
)
|
|
return generated
|
|
|
|
|
|
def _get_fernet() -> Fernet:
|
|
global _fernet
|
|
if _fernet is None:
|
|
_fernet = Fernet(_master_key())
|
|
return _fernet
|
|
|
|
|
|
def encrypt_token(plain: str) -> str:
|
|
if not plain:
|
|
return ""
|
|
return _get_fernet().encrypt(plain.encode("utf-8")).decode("utf-8")
|
|
|
|
|
|
def decrypt_token(stored: str) -> str:
|
|
if not stored:
|
|
return ""
|
|
try:
|
|
return _get_fernet().decrypt(stored.encode("utf-8")).decode("utf-8")
|
|
except Exception as exc:
|
|
logger.error("forge token decryption failed: %s", exc)
|
|
return ""
|
|
|
|
|
|
def mask_token(stored: str) -> str:
|
|
return "••••••••" if stored else "—"
|