sowai-fiscal-svc/src/fiscal_svc/certificates/crypto.py

"""A1 certificate (.pfx) lifecycle: parse + validate metadata, encrypt/
decrypt the binary for storage (Fernet), and decrypt-in-memory for signing
(consumed by `emission.service`, Task 5). Ported VERBATIM (mechanism +
docstrings) from the auto's `app/modules/fiscal/certificate.py` -- no
tenancy-shaped adaptation needed, this module never touches
product_id/tenant_ref/branch_ref, only the PFX bytes themselves.

`FISCAL_CERT_ENCRYPTION_KEY` is this SERVICE's OWN env var (Global
Constraints/porte table: "secrets never live in Settings", same convention
as the auto) -- deliberately a DIFFERENT key/secret than the auto's own
`FISCAL_CERT_ENCRYPTION_KEY` (they are two different Kubernetes Secrets in
two different namespaces/deployments; the NAME collides on purpose, mirroring
the auto's env var name 1:1, but the VALUE never does -- ties the blast
radius of a compromised key to exactly one service's certificates)."""
import os
from dataclasses import dataclass
from datetime import datetime
from functools import lru_cache

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
from cryptography.hazmat.primitives.serialization import pkcs12
from cryptography.x509 import Certificate
from cryptography.x509.oid import NameOID

CURRENT_ENCRYPTION_KEY_ID = "fiscal-svc-cert-fernet-v1-env"


class FiscalCertEncryptionKeyNotConfiguredError(RuntimeError):
    pass


class InvalidPfxError(Exception):
    """O arquivo não é um PKCS#12 (.pfx/.p12) válido -- corrompido ou de
    outro formato inteiro. Distinto de `WrongPasswordError` (ver a
    docstring de `parse_pfx` para como -- e o quão bem -- essa distinção é
    feita)."""


class WrongPasswordError(Exception):
    """A senha informada não abre o PFX. `cryptography` não distingue com
    perfeição "senha errada" de "PFX corrompido de um jeito que só se
    manifesta na fase de decriptação" -- ver a docstring de `parse_pfx`."""


class CertificateExpiredError(Exception):
    """O certificado já passou do `not_valid_after` -- fail-closed: emitir
    com um A1 vencido é rejeitado pela própria SEFAZ, então isto é
    detectado no UPLOAD, não só na emissão."""

    def __init__(self, not_valid_after: datetime):
        self.not_valid_after = not_valid_after
        super().__init__(f"Certificado vencido em {not_valid_after.isoformat()}")


@dataclass(frozen=True)
class CertInfo:
    subject_cn: str
    cnpj: str
    not_valid_before: datetime
    not_valid_after: datetime


@lru_cache(maxsize=1)
def _get_fernet() -> Fernet:
    """A ausência de `FISCAL_CERT_ENCRYPTION_KEY` é um erro de DEPLOY (o
    Secret do k8s não foi provisionado/montado), detectado aqui no PRIMEIRO
    uso -- o primeiro upload ou emissão que precisar cifrar/decifrar um
    certificado, não antes (mesma convenção do módulo que este porta)."""
    key = os.environ.get("FISCAL_CERT_ENCRYPTION_KEY")
    if not key:
        raise FiscalCertEncryptionKeyNotConfiguredError(
            "FISCAL_CERT_ENCRYPTION_KEY não configurada. Em produção vem de um "
            "secrets manager (Vault) injetada como variável de ambiente -- nunca "
            "de um arquivo .env versionado."
        )
    return Fernet(key.encode("utf-8"))


def encrypt_bytes(raw: bytes) -> bytes:
    return _get_fernet().encrypt(raw)


def decrypt_bytes(token: bytes) -> bytes:
    return _get_fernet().decrypt(token)


def _extract_cnpj(certificate: Certificate) -> str:
    """CNPJ do certificado A1 e-CNPJ (ICP-Brasil, DOC-ICP-04): o padrão de
    mercado é `CN=RAZAO SOCIAL:CNPJ` e um atributo `SERIALNUMBER` (OID
    2.5.4.5) carregando o CNPJ puro. Tenta o `SERIALNUMBER` primeiro; cai
    para o sufixo do `CN` depois de `:` quando o `SERIALNUMBER` não vem ou
    não tem 14 dígitos (algumas ACs variam)."""
    serial_attrs = certificate.subject.get_attributes_for_oid(NameOID.SERIAL_NUMBER)
    if serial_attrs:
        digits = "".join(c for c in serial_attrs[0].value if c.isdigit())
        if len(digits) == 14:
            return digits

    cn_attrs = certificate.subject.get_attributes_for_oid(NameOID.COMMON_NAME)
    if cn_attrs and ":" in cn_attrs[0].value:
        tail = cn_attrs[0].value.rsplit(":", 1)[-1]
        digits = "".join(c for c in tail if c.isdigit())
        if len(digits) == 14:
            return digits

    raise InvalidPfxError(
        "Não foi possível extrair um CNPJ (14 dígitos) do subject do certificado"
    )


def parse_pfx(pfx_bytes: bytes, password: str) -> CertInfo:
    """Abre o .pfx com a senha informada e extrai os metadados.

    `cryptography.hazmat.primitives.serialization.pkcs12.
    load_key_and_certificates` levanta `ValueError` tanto para "não é um
    PKCS#12 válido" quanto para "senha errada" -- mensagens diferentes,
    confirmadas empiricamente (heurística, não contrato): dado malformado
    -> "Could not deserialize PKCS12 data"; senha errada sobre um PKCS#12
    estruturalmente válido -> "Invalid password or PKCS12 data". Ambas
    viram 422 no router de qualquer forma -- a distinção importa só para a
    MENSAGEM ao usuário."""
    try:
        private_key, certificate, _ca_certs = pkcs12.load_key_and_certificates(
            pfx_bytes, password.encode("utf-8")
        )
    except ValueError as exc:
        if "deserialize" in str(exc).lower():
            raise InvalidPfxError(f"Arquivo não é um PKCS#12 (.pfx) válido: {exc}") from exc
        raise WrongPasswordError("Senha do certificado incorreta") from exc

    if certificate is None or private_key is None:
        raise InvalidPfxError("PFX não contém certificado e/ou chave privada")

    cn_attrs = certificate.subject.get_attributes_for_oid(NameOID.COMMON_NAME)
    subject_cn = cn_attrs[0].value if cn_attrs else certificate.subject.rfc4514_string()
    cnpj = _extract_cnpj(certificate)

    return CertInfo(
        subject_cn=subject_cn,
        cnpj=cnpj,
        not_valid_before=certificate.not_valid_before_utc,
        not_valid_after=certificate.not_valid_after_utc,
    )


def load_private_key_and_cert(certificate) -> tuple[RSAPrivateKey, Certificate]:
    """Decifra o PFX/senha de um `FiscalCertificate` row EM MEMÓRIA e
    devolve `(private_key, certificate)` prontos para assinatura
    (`erpbrasil.assinatura`, consumido por `emission.service`, Task 5).
    Nunca grava nada em disco/tmp; o retorno vive só na pilha do
    chamador."""
    pfx_bytes = decrypt_bytes(certificate.pfx_encrypted)
    password = decrypt_bytes(certificate.password_encrypted).decode("utf-8")
    private_key, cert, _ca_certs = pkcs12.load_key_and_certificates(
        pfx_bytes, password.encode("utf-8")
    )
    return private_key, cert