security: Fernet encryption, export sanitization, sensitive settings protection
- Replace base64 _enc/_dec with Fernet (AES-128-CBC + HMAC-SHA256) derived from APP_SECRET - Auto-migrate legacy base64 data to Fernet on startup (backward compatible) - Remove all credentials from config export (private keys, passwords, OCIDs, passphrases) - Block sensitive settings (oidc_*, secret_*) from non-admin read - Block sensitive settings import - Import OCI configs as metadata-only (credentials reconfigured manually)
This commit is contained in:
nogueiraguh
101
backend/app.py
101
backend/app.py
@@ -491,6 +491,25 @@ def init_db():
|
||||
GROUP BY cm.session_id""")
|
||||
except Exception as e:
|
||||
log.warning(f"DB migration: chat_sessions backfill failed: {e}")
|
||||
# Migrate legacy base64 fields to Fernet encryption
|
||||
_enc_migrations = [
|
||||
("oci_configs", ["tenancy_ocid", "user_ocid", "fingerprint", "compartment_id"]),
|
||||
("adb_vector_configs", ["password_enc", "wallet_password_enc"]),
|
||||
]
|
||||
for table, cols in _enc_migrations:
|
||||
for col in cols:
|
||||
try:
|
||||
rows = c.execute(f"SELECT id, {col} FROM {table} WHERE {col} IS NOT NULL AND {col} != ''").fetchall()
|
||||
for row in rows:
|
||||
val = row[col]
|
||||
if val and not val.startswith(_FERNET_PREFIX):
|
||||
try:
|
||||
plaintext = base64.b64decode(val.encode()).decode()
|
||||
c.execute(f"UPDATE {table} SET {col}=? WHERE id=?", (_enc(plaintext), row["id"]))
|
||||
except Exception:
|
||||
pass
|
||||
except Exception as e:
|
||||
log.warning(f"DB migration: encrypt {table}.{col} failed: {e}")
|
||||
# Seed default system prompt if none exists for chat agent
|
||||
if not c.execute("SELECT 1 FROM system_prompts WHERE agent='chat'").fetchone():
|
||||
c.execute("INSERT INTO system_prompts (id,name,agent,content,is_active) VALUES (?,?,?,?,?)",
|
||||
@@ -534,8 +553,30 @@ def _totp_verify(secret,code,window=1):
|
||||
def _totp_uri(secret,user): return f"otpauth://totp/AI-Agent:{user}?secret={secret}&issuer=AI-Agent"
|
||||
def _make_token(uid,role,sid):
|
||||
return pyjwt.encode({"sub":uid,"role":role,"sid":sid,"exp":datetime.utcnow()+timedelta(hours=JWT_EXP_H),"iat":datetime.utcnow()},APP_SECRET,algorithm=JWT_ALG)
|
||||
def _enc(v): return base64.b64encode(v.encode()).decode()
|
||||
def _dec(v): return base64.b64decode(v.encode()).decode()
|
||||
|
||||
# ── Fernet encryption (AES-128-CBC + HMAC-SHA256) ────────────────────────────
|
||||
from cryptography.fernet import Fernet as _Fernet
|
||||
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC as _PBKDF2HMAC
|
||||
from cryptography.hazmat.primitives import hashes as _hashes
|
||||
_FERNET_PREFIX = "fernet:"
|
||||
def _derive_fernet_key(secret: str) -> bytes:
|
||||
kdf = _PBKDF2HMAC(algorithm=_hashes.SHA256(), length=32, salt=b"oci-cis-agent-fernet-v1", iterations=100_000)
|
||||
return base64.urlsafe_b64encode(kdf.derive(secret.encode()))
|
||||
_fernet = _Fernet(_derive_fernet_key(APP_SECRET))
|
||||
|
||||
def _enc(v):
|
||||
"""Encrypt a string using Fernet (AES + HMAC). Returns prefixed token."""
|
||||
if not v: return v
|
||||
return _FERNET_PREFIX + _fernet.encrypt(v.encode()).decode()
|
||||
|
||||
def _dec(v):
|
||||
"""Decrypt a value. Supports Fernet (prefixed) and legacy base64 (unprefixed)."""
|
||||
if not v: return v
|
||||
if v.startswith(_FERNET_PREFIX):
|
||||
return _fernet.decrypt(v[len(_FERNET_PREFIX):].encode()).decode()
|
||||
# Legacy base64 fallback
|
||||
return base64.b64decode(v.encode()).decode()
|
||||
|
||||
def _mask(v, show=6):
|
||||
"""Mask a sensitive value, showing only the last `show` characters."""
|
||||
if not v or len(v) <= show:
|
||||
@@ -9668,8 +9709,12 @@ async def get_chat_logs(
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
# ── App Settings ──────────────────────────────────────────────────────────────
|
||||
_SENSITIVE_SETTING_PREFIXES = ("oidc_", "secret_", "app_secret", "encryption_")
|
||||
|
||||
@app.get("/api/settings/{key}")
|
||||
async def get_setting(key: str, u=Depends(current_user)):
|
||||
if any(key.startswith(p) or key == p for p in _SENSITIVE_SETTING_PREFIXES) and u["role"] != "admin":
|
||||
raise HTTPException(403, "Acesso negado a configuração sensível")
|
||||
with db() as c:
|
||||
row = c.execute("SELECT value FROM app_settings WHERE key=?", (key,)).fetchone()
|
||||
return {"key": key, "value": row["value"] if row else ""}
|
||||
@@ -9764,37 +9809,24 @@ async def delete_prompt(pid: str, u=Depends(require("admin"))):
|
||||
# ── Export / Import Configuration ─────────────────────────────────────────────
|
||||
@app.get("/api/config/export")
|
||||
async def export_config(u=Depends(require("admin"))):
|
||||
"""Export all configurations as JSON (OCI configs with keys, GenAI, MCP, prompts, settings)."""
|
||||
"""Export configurations as JSON. Excludes sensitive credentials (private keys, passwords, secrets)."""
|
||||
_OCI_SAFE_COLS = "id,user_id,tenancy_name,region,compartment_id,auth_type,ssh_public_key,created_at"
|
||||
_ADB_SAFE_COLS = "id,user_id,config_name,dsn,username,table_name,use_mtls,is_active,is_global,genai_config_id,embedding_model_id,created_at"
|
||||
data = {"version": VERSION, "exported_at": datetime.now().isoformat()}
|
||||
with db() as c:
|
||||
# OCI configs
|
||||
oci_rows = c.execute("SELECT * FROM oci_configs").fetchall()
|
||||
oci_cfgs = []
|
||||
for r in oci_rows:
|
||||
cfg = dict(r)
|
||||
# Include private key content (base64)
|
||||
kp = Path(cfg.get("key_file_path", ""))
|
||||
if kp.exists():
|
||||
cfg["_private_key_b64"] = base64.b64encode(kp.read_bytes()).decode()
|
||||
# Include passphrase from config file if exists
|
||||
cfg_file = kp.parent / "config" if kp.parent.exists() else None
|
||||
if cfg_file and cfg_file.exists():
|
||||
for line in cfg_file.read_text().splitlines():
|
||||
if line.startswith("pass_phrase="):
|
||||
cfg["_key_passphrase"] = line.split("=", 1)[1]
|
||||
oci_cfgs.append(cfg)
|
||||
data["oci_configs"] = oci_cfgs
|
||||
# GenAI configs
|
||||
# OCI configs (no private keys, no encrypted OCIDs, no passphrases)
|
||||
data["oci_configs"] = [dict(r) for r in c.execute(f"SELECT {_OCI_SAFE_COLS} FROM oci_configs").fetchall()]
|
||||
# GenAI configs (no secrets — model info only)
|
||||
data["genai_configs"] = [dict(r) for r in c.execute("SELECT * FROM genai_configs").fetchall()]
|
||||
# MCP servers (exclude system-auto-registered if tools are empty)
|
||||
# MCP servers
|
||||
data["mcp_servers"] = [dict(r) for r in c.execute("SELECT * FROM mcp_servers").fetchall()]
|
||||
# System prompts
|
||||
data["system_prompts"] = [dict(r) for r in c.execute("SELECT * FROM system_prompts").fetchall()]
|
||||
# App settings
|
||||
data["app_settings"] = [dict(r) for r in c.execute("SELECT * FROM app_settings").fetchall()]
|
||||
# ADB vector configs
|
||||
adb_rows = c.execute("SELECT * FROM adb_vector_configs").fetchall()
|
||||
data["adb_vector_configs"] = [dict(r) for r in adb_rows]
|
||||
# App settings (exclude sensitive keys)
|
||||
data["app_settings"] = [dict(r) for r in c.execute("SELECT * FROM app_settings").fetchall()
|
||||
if not any(r["key"].startswith(p) for p in _SENSITIVE_SETTING_PREFIXES)]
|
||||
# ADB vector configs (no passwords, no wallet passwords)
|
||||
data["adb_vector_configs"] = [dict(r) for r in c.execute(f"SELECT {_ADB_SAFE_COLS} FROM adb_vector_configs").fetchall()]
|
||||
# ADB vector tables
|
||||
data["adb_vector_tables"] = [dict(r) for r in c.execute("SELECT * FROM adb_vector_tables").fetchall()]
|
||||
return StreamingResponse(
|
||||
@@ -9820,12 +9852,12 @@ async def import_config(file: UploadFile = File(...), u=Depends(require("admin")
|
||||
for cfg in data.get("oci_configs", []):
|
||||
if c.execute("SELECT 1 FROM oci_configs WHERE id=?", (cfg["id"],)).fetchone():
|
||||
counts["skipped"] += 1; continue
|
||||
# Restore private key file
|
||||
# Import metadata only — credentials must be reconfigured manually
|
||||
cdir = OCI_DIR / cfg["id"]; cdir.mkdir(parents=True, exist_ok=True)
|
||||
kp = cdir / "oci_api_key.pem"
|
||||
# Restore private key only if present in legacy export format
|
||||
if cfg.get("_private_key_b64"):
|
||||
kp.write_bytes(base64.b64decode(cfg["_private_key_b64"])); kp.chmod(0o600)
|
||||
# Write config file
|
||||
cfg_file = cdir / "config"
|
||||
cfg_content = (f"[DEFAULT]\nuser={_dec(cfg['user_ocid']) if cfg.get('user_ocid') else ''}\n"
|
||||
f"fingerprint={_dec(cfg['fingerprint']) if cfg.get('fingerprint') else ''}\n"
|
||||
@@ -9835,8 +9867,13 @@ async def import_config(file: UploadFile = File(...), u=Depends(require("admin")
|
||||
cfg_content += f"pass_phrase={cfg['_key_passphrase']}\n"
|
||||
cfg_file.write_text(cfg_content); cfg_file.chmod(0o600)
|
||||
c.execute("INSERT INTO oci_configs (id,user_id,tenancy_name,tenancy_ocid,user_ocid,fingerprint,region,key_file_path,compartment_id,created_at) VALUES (?,?,?,?,?,?,?,?,?,?)",
|
||||
(cfg["id"], cfg.get("user_id", u["id"]), cfg.get("tenancy_name",""), cfg.get("tenancy_ocid",""), cfg.get("user_ocid",""),
|
||||
cfg.get("fingerprint",""), cfg.get("region",""), str(kp), cfg.get("compartment_id"), cfg.get("created_at")))
|
||||
(cfg["id"], cfg.get("user_id", u["id"]), cfg.get("tenancy_name",""),
|
||||
_enc(cfg["tenancy_ocid"]) if cfg.get("tenancy_ocid") else "",
|
||||
_enc(cfg["user_ocid"]) if cfg.get("user_ocid") else "",
|
||||
_enc(cfg["fingerprint"]) if cfg.get("fingerprint") else "",
|
||||
cfg.get("region",""), str(kp),
|
||||
_enc(cfg["compartment_id"]) if cfg.get("compartment_id") else None,
|
||||
cfg.get("created_at")))
|
||||
counts["oci_configs"] += 1
|
||||
# GenAI configs
|
||||
for cfg in data.get("genai_configs", []):
|
||||
@@ -9867,6 +9904,8 @@ async def import_config(file: UploadFile = File(...), u=Depends(require("admin")
|
||||
counts["system_prompts"] += 1
|
||||
# App settings
|
||||
for s in data.get("app_settings", []):
|
||||
if any(s["key"].startswith(p) for p in _SENSITIVE_SETTING_PREFIXES):
|
||||
counts["skipped"] += 1; continue
|
||||
c.execute("INSERT INTO app_settings (key,value,updated_at) VALUES (?,?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at",
|
||||
(s["key"], s.get("value",""), s.get("updated_at")))
|
||||
counts["app_settings"] += 1
|
||||
|
||||
Reference in New Issue
Block a user