FabledSteward/steward/core/crypto.py

# steward/core/crypto.py
"""Encryption-at-rest for sensitive settings (smtp/oidc/ldap/ansible secrets).

Values are encrypted with Fernet (AES-128-CBC + HMAC) using a key derived from
the app secret key (the same /data/secret.key used for sessions). Encrypted
values are stored with an ``enc:v1:`` prefix so reads can transparently tell
ciphertext from legacy plaintext during/after migration.

Key-loss caveat: if the app secret key is lost or changed, encrypted secrets
become unrecoverable — they must be re-entered. This ties secret recovery to
the same key the rest of the app already depends on; back it up with the data.
"""
from __future__ import annotations

import base64
import hashlib
import logging
import os
from pathlib import Path

from cryptography.fernet import Fernet, InvalidToken
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

logger = logging.getLogger(__name__)

ENC_PREFIX = "enc:v1:"
_SECRET_KEY_FILE = Path("/data/secret.key")
_fernet: Fernet | None = None


def _make_fernet(secret_key: str) -> Fernet:
    # Derive a stable 32-byte Fernet key from the app secret (any length string).
    digest = hashlib.sha256(secret_key.encode("utf-8")).digest()
    return Fernet(base64.urlsafe_b64encode(digest))


def init_crypto(secret_key: str) -> None:
    """Bind the encryptor to the app's secret key (called once at startup)."""
    global _fernet
    if secret_key:
        _fernet = _make_fernet(secret_key)


def _resolve_secret_key() -> str | None:
    """Fallback key resolution mirroring config (env → /data/secret.key)."""
    k = os.environ.get("STEWARD_SECRET_KEY")
    if k:
        return k
    try:
        if _SECRET_KEY_FILE.exists():
            v = _SECRET_KEY_FILE.read_text().strip()
            if v:
                return v
    except OSError:
        pass
    return None


def _get() -> Fernet | None:
    global _fernet
    if _fernet is None:
        k = _resolve_secret_key()
        if k:
            _fernet = _make_fernet(k)
    return _fernet


def is_encrypted(value) -> bool:
    return isinstance(value, str) and value.startswith(ENC_PREFIX)


def encrypt_secret(plaintext: str) -> str:
    """Return an ``enc:v1:`` token, or the input unchanged if empty / no key."""
    if not plaintext:
        return plaintext
    f = _get()
    if f is None:
        logger.warning("No secret key available — storing a secret in PLAINTEXT")
        return plaintext
    return ENC_PREFIX + f.encrypt(plaintext.encode("utf-8")).decode("ascii")


def decrypt_secret(value: str) -> str:
    """Decrypt an ``enc:v1:`` token; pass through plaintext / undecryptable values."""
    if not is_encrypted(value):
        return value
    f = _get()
    if f is None:
        return value
    try:
        return f.decrypt(value[len(ENC_PREFIX):].encode("ascii")).decode("utf-8")
    except InvalidToken:
        logger.error("Could not decrypt a stored secret (wrong/rotated key?)")
        return value


def generate_ssh_keypair(comment: str = "steward-managed") -> tuple[str, str]:
    """Mint a fresh ed25519 keypair for Steward's managed Ansible identity.

    Returns (private_openssh_pem, public_openssh_line). ed25519 is small, fast,
    and universally supported by modern OpenSSH. The private key is unencrypted
    OpenSSH PEM (Steward stores it encrypted-at-rest itself); the public key is
    the single-line ``ssh-ed25519 AAAA… comment`` form for authorized_keys.
    """
    key = Ed25519PrivateKey.generate()
    private_pem = key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.OpenSSH,
        encryption_algorithm=serialization.NoEncryption(),
    ).decode("ascii")
    public_line = key.public_key().public_bytes(
        encoding=serialization.Encoding.OpenSSH,
        format=serialization.PublicFormat.OpenSSH,
    ).decode("ascii")
    if comment:
        public_line = f"{public_line} {comment}"
    return private_pem, public_line