auto-stock/backend/app/core/crypto.py

"""사용자 KIS 자격증명 암복호화 (멀티테넌트 1단계).

- 키 파생: PBKDF2-HMAC-SHA256(200_000 회) → Fernet 키.
  소스는 settings.CREDENTIALS_ENCRYPTION_KEY, 비어있으면 settings.SECRET_KEY 폴백.
- 로테이션: CREDENTIALS_ENCRYPTION_KEY_PREVIOUS 가 설정되면 MultiFernet 으로
  구 키 복호화(읽기)만 허용, 암호화(쓰기)는 항상 현재 키(version 1)를 사용.
  버전 2단계 확장 시 key_version 값을 늘리고 _derive_key(version=N) 로 파생하면 됨.

사용:
    from app.core.crypto import encrypt, decrypt, current_key_version, is_encryption_configured
    token = encrypt("plain-app-key")
    plain = decrypt(token, key_version=1)
"""
import base64
import functools
import hashlib
import logging

from cryptography.fernet import Fernet, InvalidToken, MultiFernet

from app.core.config import settings

logger = logging.getLogger(__name__)

_KDF_ITERATIONS = 200_000
_DEFAULT_SECRET = "change-this-secret-key"


class CredentialDecryptError(ValueError):
    """자격증명 복호화 실패 — 키 로테이션/설정 변경 등 서버 측 원인 진단용 (L2).

    기존 호출부가 `ValueError`를 기대할 수 있으므로 하위 호환을 위해 `ValueError`를 상속한다.
    """


@functools.lru_cache(maxsize=8)
def _derive_key(secret: str, version: int) -> bytes:
    """PBKDF2 로 Fernet 대칭키(urlsafe base64, 32바이트) 파생.

    PBKDF2 200_000회 반복은 호출당 ~0.1~0.2초 CPU 연산이라(M4) (secret, version) 조합별로
    캐시한다 — secret 후보는 CREDENTIALS_ENCRYPTION_KEY/SECRET_KEY(+PREVIOUS) 뿐이라
    프로세스 수명 동안 조합 수가 매우 작다.
    """
    salt = f"autostock-kis-cred-v{version}".encode()
    raw = hashlib.pbkdf2_hmac("sha256", secret.encode(), salt, _KDF_ITERATIONS)
    return base64.urlsafe_b64encode(raw)


def _key_source() -> str:
    return settings.CREDENTIALS_ENCRYPTION_KEY or settings.SECRET_KEY


def is_encryption_configured() -> bool:
    """SECRET_KEY 가 기본값이거나 너무 짧으면(길이<32) 암호화 저장을 거부."""
    secret = _key_source()
    if secret == _DEFAULT_SECRET:
        return False
    if len(secret) < 32:
        return False
    return True


def current_key_version() -> int:
    """현재 암호화에 사용하는 키 버전. 1단계는 항상 1(로테이션은 §9 2단계에서 증가)."""
    return 1


def _fernet_for_write() -> Fernet:
    return Fernet(_derive_key(_key_source(), current_key_version()))


def _fernet_for_read(key_version: int) -> MultiFernet:
    """복호화용 — 현재 키 + (있으면) 이전 키 순으로 시도."""
    keys = [Fernet(_derive_key(_key_source(), key_version))]
    if settings.CREDENTIALS_ENCRYPTION_KEY_PREVIOUS:
        keys.append(Fernet(_derive_key(settings.CREDENTIALS_ENCRYPTION_KEY_PREVIOUS, key_version)))
    return MultiFernet(keys)


def encrypt(plain: str) -> str:
    """평문을 암호화해 저장용 문자열(토큰)로 반환. 항상 current_key_version() 기준."""
    if not is_encryption_configured():
        raise RuntimeError(
            "CREDENTIALS_ENCRYPTION_KEY/SECRET_KEY 미설정 또는 기본값 — 자격증명 암호화 저장 불가"
        )
    return _fernet_for_write().encrypt(plain.encode()).decode()


def decrypt(token: str, key_version: int = 1) -> str:
    """저장된 토큰을 평문으로 복호화. 실패 시 `CredentialDecryptError`(L2, `ValueError` 서브클래스)."""
    try:
        return _fernet_for_read(key_version).decrypt(token.encode()).decode()
    except InvalidToken as e:
        logger.error("[Crypto] 자격증명 복호화 실패 (key_version=%s)", key_version)
        raise CredentialDecryptError("자격증명 복호화 실패 — 암호화 키 불일치") from e