821675ab4a
Specs stay at repo root (cross-VM). Move deploy and code into logical projects with README per domain, updated manifest.yaml, and symlinks at legacy paths for VM122 backward compatibility.
54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
"""Encrypted credential vault — Spec 019."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
|
|
_KEY_ENV = "MIGRATION_CREDENTIALS_KEY"
|
|
|
|
|
|
def _fernet() -> Fernet:
|
|
raw = os.getenv(_KEY_ENV, "").strip()
|
|
if not raw:
|
|
raw = Fernet.generate_key().decode()
|
|
if len(raw) != 44:
|
|
raw = Fernet.generate_key().decode()
|
|
return Fernet(raw.encode() if isinstance(raw, str) else raw)
|
|
|
|
|
|
def store_secret(conn, mailbox_id: int, secret: dict[str, Any]) -> str:
|
|
cred_id = str(uuid.uuid4())
|
|
blob = _fernet().encrypt(json.dumps(secret).encode())
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO migration_credentials (id, mailbox_id, secret_blob, created_at)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
(cred_id, mailbox_id, blob, now),
|
|
)
|
|
conn.execute(
|
|
"UPDATE migration_mailboxes SET credentials_ref = ? WHERE id = ?",
|
|
(cred_id, mailbox_id),
|
|
)
|
|
conn.commit()
|
|
return cred_id
|
|
|
|
|
|
def load_secret(conn, cred_id: str) -> dict[str, Any] | None:
|
|
row = conn.execute(
|
|
"SELECT secret_blob FROM migration_credentials WHERE id = ?",
|
|
(cred_id,),
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
try:
|
|
return json.loads(_fernet().decrypt(bytes(row["secret_blob"])).decode())
|
|
except (InvalidToken, json.JSONDecodeError):
|
|
return None
|