diff --git a/.env.example b/.env.example index 171db7e..e491622 100644 --- a/.env.example +++ b/.env.example @@ -1,2 +1,8 @@ DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/fiscal_svc_dev TEST_DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/fiscal_svc_test + +# Fernet key for A1 certificate ciphertext at rest (certificates.crypto, +# Task 4) -- read straight from os.environ, never through Settings (same +# "secrets don't live in Settings" convention as the auto). Generate one +# with: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" +FISCAL_CERT_ENCRYPTION_KEY= diff --git a/pyproject.toml b/pyproject.toml index 06820ee..a27ce37 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,6 +13,7 @@ dependencies = [ "passlib[bcrypt]>=1.7.4", "pydantic-settings>=2.0", "pydantic>=2.13.4", + "python-multipart>=0.0.20", "sowai-fiscal", "sqlalchemy[asyncio]>=2.0", "uvicorn[standard]>=0.49.0", diff --git a/src/fiscal_svc/certificates/__init__.py b/src/fiscal_svc/certificates/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/fiscal_svc/certificates/crypto.py b/src/fiscal_svc/certificates/crypto.py new file mode 100644 index 0000000..bf39a17 --- /dev/null +++ b/src/fiscal_svc/certificates/crypto.py @@ -0,0 +1,158 @@ +"""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 diff --git a/src/fiscal_svc/certificates/router.py b/src/fiscal_svc/certificates/router.py new file mode 100644 index 0000000..d5b4f9e --- /dev/null +++ b/src/fiscal_svc/certificates/router.py @@ -0,0 +1,111 @@ +"""`POST/GET/DELETE /v1/certificados` -- Task 4. `tenant_ref`/`branch_ref` +travel as QUERY params on all three verbs (consistent shape across POST/GET/ +DELETE, since POST's body is multipart -- file + form fields -- and cannot +also carry a JSON body); `cnpj`/`password` are `Form(...)` fields alongside +the file, same multipart shape as the auto's own certificate upload.""" +from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, UploadFile, status +from sqlalchemy.ext.asyncio import AsyncSession + +from fiscal_svc.certificates import crypto as certificate_lib +from fiscal_svc.certificates import service +from fiscal_svc.certificates.schemas import FiscalCertificateRead +from fiscal_svc.core.db import get_session +from fiscal_svc.shared.errors import conflict +from fiscal_svc.tenancy.deps import require_product +from fiscal_svc.tenancy.models import Product + +router = APIRouter(prefix="/v1/certificados", tags=["certificados"]) + +# Teto de tamanho do upload do .pfx -- mesmo valor/racional do auto +# (`fiscal.router._MAX_PFX_UPLOAD_BYTES`): um A1 típico fica na casa de +# poucos KB; 256 KiB é generoso o bastante sem deixar o endpoint aceitar um +# upload arbitrariamente grande. +_MAX_PFX_UPLOAD_BYTES = 256 * 1024 + + +async def _read_upload_capped(file: UploadFile, max_bytes: int) -> bytes: + chunk_size = 64 * 1024 + chunks: list[bytes] = [] + total = 0 + while True: + chunk = await file.read(chunk_size) + if not chunk: + break + total += len(chunk) + if total > max_bytes: + raise HTTPException( + status_code=status.HTTP_413_CONTENT_TOO_LARGE, + detail=f"Certificado excede o tamanho máximo permitido ({max_bytes} bytes)", + ) + chunks.append(chunk) + return b"".join(chunks) + + +@router.post("", response_model=FiscalCertificateRead, status_code=status.HTTP_201_CREATED) +async def upload_certificate_endpoint( + tenant_ref: str = Query(..., max_length=64), + branch_ref: str = Query(..., max_length=64), + cnpj: str = Form(..., min_length=14, max_length=14), + file: UploadFile = File(...), + password: str = Form(...), + product: Product = Depends(require_product), + session: AsyncSession = Depends(get_session), +) -> FiscalCertificateRead: + pfx_bytes = await _read_upload_capped(file, _MAX_PFX_UPLOAD_BYTES) + try: + certificate = await service.upload_certificate( + session, product.id, tenant_ref, branch_ref, cnpj, pfx_bytes, password + ) + except (certificate_lib.InvalidPfxError, certificate_lib.WrongPasswordError) as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc) + ) from exc + except service.CertificateCnpjMismatchError as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc) + ) from exc + except certificate_lib.CertificateExpiredError as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc) + ) from exc + except service.CertificateNotYetValidError as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc) + ) from exc + except service.CertificateUploadConflictError as exc: + raise conflict("certificate_upload_conflict", str(exc)) from exc + return FiscalCertificateRead.model_validate(certificate) + + +@router.get("", response_model=FiscalCertificateRead) +async def read_certificate_endpoint( + tenant_ref: str = Query(..., max_length=64), + branch_ref: str = Query(..., max_length=64), + product: Product = Depends(require_product), + session: AsyncSession = Depends(get_session), +) -> FiscalCertificateRead: + """Metadados do certificado VIVO -- NUNCA o binário (o .pfx/senha + cifrados nunca saem do banco por esta rota; só `certificates.crypto. + load_private_key_and_cert`, uso interno da emissão, Task 5, os + descriptografa, e só em memória).""" + certificate = await service.get_certificate(session, product.id, tenant_ref, branch_ref) + if certificate is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Certificado não encontrado" + ) + return FiscalCertificateRead.model_validate(certificate) + + +@router.delete("", status_code=status.HTTP_204_NO_CONTENT) +async def deactivate_certificate_endpoint( + tenant_ref: str = Query(..., max_length=64), + branch_ref: str = Query(..., max_length=64), + product: Product = Depends(require_product), + session: AsyncSession = Depends(get_session), +) -> None: + try: + await service.deactivate_certificate(session, product.id, tenant_ref, branch_ref) + except service.CertificateNotFoundError as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Certificado não encontrado" + ) from exc diff --git a/src/fiscal_svc/certificates/schemas.py b/src/fiscal_svc/certificates/schemas.py new file mode 100644 index 0000000..188c99c --- /dev/null +++ b/src/fiscal_svc/certificates/schemas.py @@ -0,0 +1,23 @@ +import uuid +from datetime import datetime + +from pydantic import BaseModel, ConfigDict + + +class FiscalCertificateRead(BaseModel): + """Metadados do A1 -- NUNCA o binário (`pfx_encrypted`/ + `password_encrypted` ficam de fora de propósito; ver + `certificates.router.read_certificate_endpoint`'s docstring).""" + + model_config = ConfigDict(from_attributes=True) + + id: uuid.UUID + product_id: uuid.UUID + tenant_ref: str + branch_ref: str + cnpj: str + subject_cn: str + cnpj_certificado: str + not_valid_before: datetime + not_valid_after: datetime + created_at: datetime diff --git a/src/fiscal_svc/certificates/service.py b/src/fiscal_svc/certificates/service.py new file mode 100644 index 0000000..057e738 --- /dev/null +++ b/src/fiscal_svc/certificates/service.py @@ -0,0 +1,225 @@ +"""Task 4: `FiscalCertificate` lifecycle -- ported from the auto's +`app/modules/fiscal/service.py` (the certificate half only; TaxRule/preset/ +simulate stay in the product per design spec decision #3, "o motor de +regras FICA no produto"). Porte table applied: `organization_id`/ +`branch_id` (a real FK, validated via `tenants_service.get_branch`) become +`product_id`/`branch_ref` (an OPAQUE string this service never resolves +against a Branch row -- there is no Branch row here). Consequently every +`CertificateBranchNotFoundError` check from the auto is GONE: there is no +branch existence to 404 on, only a `product_id`/`tenant_ref`/`branch_ref` +scope the caller declares. + +The "vínculo forte" (design spec decision #4) that used to be `Branch.cnpj` +is now the `cnpj` FIELD ON THE UPLOAD REQUEST ITSELF (the router's +`FiscalCertificateUpload.cnpj`, Task 4) -- the caller (the product) declares +which CNPJ this branch_ref is FOR, and this service validates the +certificate's own CNPJ against THAT declaration, never against a row it +owns.""" +import uuid +from datetime import datetime, timezone + +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from fiscal_svc.certificates import crypto as certificate_lib +from fiscal_svc.documents.models import FiscalCertificate + + +class CertificateNotFoundError(Exception): + """No `(product_id, tenant_ref, branch_ref)` scope has a live + `FiscalCertificate` -- raised both when none was ever uploaded and when + a previous one was soft-deleted, same "revoked = gone" semantics as the + rest of this service.""" + + def __init__(self, product_id: uuid.UUID, branch_ref: str): + self.product_id = product_id + self.branch_ref = branch_ref + super().__init__( + f"Nenhum certificado vivo para product_id={product_id}, branch_ref={branch_ref!r}" + ) + + +class CertificateCnpjMismatchError(Exception): + """O CNPJ extraído do certificado não bate com o CNPJ declarado no + payload do upload -- um A1 é emitido para UM CNPJ específico (design + spec decisão #4: o CNPJ é o vínculo forte); um certificado de outro + CNPJ nunca pode ser aceito, mesmo que a senha esteja correta e o + arquivo seja um PFX íntegro.""" + + def __init__(self, payload_cnpj: str, cert_cnpj: str): + self.payload_cnpj = payload_cnpj + self.cert_cnpj = cert_cnpj + super().__init__( + f"CNPJ do certificado ({cert_cnpj}) diverge do CNPJ informado ({payload_cnpj})" + ) + + +class CertificateNotYetValidError(Exception): + """O `not_valid_before` do certificado ainda não chegou -- um A1 emitido + com validade FUTURA (ou lido sob um relógio de cliente adiantado) não + está "ainda válido" agora, e a SEFAZ rejeita a assinatura de um + certificado fora da janela de validade em QUALQUER direção, não só + vencido. Mesma camada fail-closed de `CertificateExpiredError` (checado + no UPLOAD, não só na emissão).""" + + def __init__(self, not_valid_before: datetime): + self.not_valid_before = not_valid_before + super().__init__( + f"Certificado ainda não é válido -- válido a partir de " + f"{not_valid_before.isoformat()}" + ) + + +class CertificateUploadConflictError(Exception): + """Traduz o `IntegrityError` da violação do índice parcial único + `ix_fiscal_certificates_product_branch_live` (`(product_id, branch_ref) + WHERE deleted_at IS NULL`, Task 3's `FiscalCertificate.__table_args__`) + -- disparado quando DUAS chamadas genuinamente concorrentes de + `upload_certificate` para o MESMO `(product_id, branch_ref)` ambas leem + `_get_live_certificate_by_branch() -> None` antes de qualquer uma + commitar. A PRIMEIRA a commitar vence; a segunda recebe este erro (409 + `certificate_upload_conflict` no router) em vez de silenciosamente + criar um segundo certificado vivo.""" + + def __init__(self, product_id: uuid.UUID, branch_ref: str): + self.product_id = product_id + self.branch_ref = branch_ref + super().__init__( + f"Upload de certificado concorrente para product_id={product_id}, " + f"branch_ref={branch_ref!r} -- outro upload venceu a corrida; tente novamente" + ) + + +def _is_fiscal_certificate_branch_live_violation(exc: IntegrityError) -> bool: + """True iff `exc` violates `ix_fiscal_certificates_product_branch_live` + -- substring match on the index name, which Task 3's migration names + explicitly (unlike the auto's un-named-in-create_all equivalent, this + one is identical across `Base.metadata.create_all` and Alembic, so a + plain substring check suffices, no "unique constraint" reinforcement + needed).""" + detail = str(getattr(exc.orig, "args", [""])[0]) if exc.orig else str(exc) + return "ix_fiscal_certificates_product_branch_live" in detail + + +async def _get_live_certificate_by_branch( + session: AsyncSession, product_id: uuid.UUID, branch_ref: str +) -> FiscalCertificate | None: + """Scoped to `(product_id, branch_ref)` ONLY -- matches exactly the + scope of the partial unique index this function's callers (`upload_ + certificate`) need to pre-check against. NOT the public lookup for GET/ + DELETE (see `get_certificate`/`deactivate_certificate` below, which + additionally scope by `tenant_ref` for the anti-oracle boundary, design + spec decision #4: "anti-oracle 404 por (product_id, tenant_ref)").""" + result = await session.execute( + select(FiscalCertificate).where( + FiscalCertificate.product_id == product_id, + FiscalCertificate.branch_ref == branch_ref, + FiscalCertificate.deleted_at.is_(None), + ) + ) + return result.scalar_one_or_none() + + +async def _get_live_certificate( + session: AsyncSession, product_id: uuid.UUID, tenant_ref: str, branch_ref: str +) -> FiscalCertificate | None: + """Public lookup, scoped to `(product_id, tenant_ref, branch_ref)` -- + the anti-oracle boundary (design spec decision #4). Used by GET/DELETE + AND by `emission.service` (Task 5, which duplicates this as a small + private SELECT rather than importing it -- same "not worth the cross- + module coupling for a few-line query" reasoning the auto's own + `fiscal.emissao._get_live_certificate` docstring gives for not reusing + `fiscal.service._get_live_certificate` directly).""" + result = await session.execute( + select(FiscalCertificate).where( + FiscalCertificate.product_id == product_id, + FiscalCertificate.tenant_ref == tenant_ref, + FiscalCertificate.branch_ref == branch_ref, + FiscalCertificate.deleted_at.is_(None), + ) + ) + return result.scalar_one_or_none() + + +async def upload_certificate( + session: AsyncSession, + product_id: uuid.UUID, + tenant_ref: str, + branch_ref: str, + cnpj: str, + pfx_bytes: bytes, + password: str, +) -> FiscalCertificate: + """Valida e armazena um A1 para `(product_id, tenant_ref, branch_ref)`. + Fail-closed, na ordem: PFX abre com a senha (`certificate_lib.parse_pfx` + -- `InvalidPfxError`/`WrongPasswordError`) -> CNPJ do certificado == + `cnpj` declarado no payload (`CertificateCnpjMismatchError`) -> dentro + da janela de validade, nas DUAS direções: não vencido + (`CertificateExpiredError`) e já vigente (`CertificateNotYetValidError`). + Só depois de TODAS as checagens passarem é que qualquer escrita + acontece: o certificado anterior (se houver, para o MESMO + `(product_id, branch_ref)`) é soft-deletado e o novo é inserido -- um + único VIVO por `(product_id, branch_ref)`, nunca dois, nunca um + update-in-place. + + O INSERT final é protegido pelo índice parcial único (Task 3) contra a + corrida de dois uploads genuinamente concorrentes -- o `commit()` do + PERDEDOR levanta `IntegrityError`, capturado aqui e traduzido em + `CertificateUploadConflictError` (409 no router).""" + info = certificate_lib.parse_pfx(pfx_bytes, password) + + if cnpj != info.cnpj: + raise CertificateCnpjMismatchError(cnpj, info.cnpj) + + now = datetime.now(timezone.utc) + if info.not_valid_after < now: + raise certificate_lib.CertificateExpiredError(info.not_valid_after) + if info.not_valid_before > now: + raise CertificateNotYetValidError(info.not_valid_before) + + previous = await _get_live_certificate_by_branch(session, product_id, branch_ref) + if previous is not None: + previous.deleted_at = now + + certificate = FiscalCertificate( + product_id=product_id, + tenant_ref=tenant_ref, + branch_ref=branch_ref, + cnpj=cnpj, + pfx_encrypted=certificate_lib.encrypt_bytes(pfx_bytes), + password_encrypted=certificate_lib.encrypt_bytes(password.encode("utf-8")), + subject_cn=info.subject_cn, + cnpj_certificado=info.cnpj, + not_valid_before=info.not_valid_before, + not_valid_after=info.not_valid_after, + ) + session.add(certificate) + try: + await session.commit() + except IntegrityError as exc: + await session.rollback() + if _is_fiscal_certificate_branch_live_violation(exc): + raise CertificateUploadConflictError(product_id, branch_ref) from exc + raise + await session.refresh(certificate) + return certificate + + +async def get_certificate( + session: AsyncSession, product_id: uuid.UUID, tenant_ref: str, branch_ref: str +) -> FiscalCertificate | None: + """Metadados do certificado VIVO -- NUNCA descriptografa o binário aqui + (ver `certificates.crypto.load_private_key_and_cert`, uso exclusivo da + emissão, Task 5).""" + return await _get_live_certificate(session, product_id, tenant_ref, branch_ref) + + +async def deactivate_certificate( + session: AsyncSession, product_id: uuid.UUID, tenant_ref: str, branch_ref: str +) -> None: + certificate = await _get_live_certificate(session, product_id, tenant_ref, branch_ref) + if certificate is None: + raise CertificateNotFoundError(product_id, branch_ref) + certificate.deleted_at = datetime.now(timezone.utc) + await session.commit() diff --git a/src/fiscal_svc/main.py b/src/fiscal_svc/main.py index 6af5233..0660bee 100644 --- a/src/fiscal_svc/main.py +++ b/src/fiscal_svc/main.py @@ -1,9 +1,14 @@ from fastapi import FastAPI +from fiscal_svc.certificates.router import router as certificates_router from fiscal_svc.core.config import settings +from fiscal_svc.series.router import router as series_router app = FastAPI(title=settings.app_name) +app.include_router(certificates_router) +app.include_router(series_router) + @app.get("/v1/health") async def health_check() -> dict[str, str]: diff --git a/src/fiscal_svc/series/__init__.py b/src/fiscal_svc/series/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/fiscal_svc/series/router.py b/src/fiscal_svc/series/router.py new file mode 100644 index 0000000..6482cf3 --- /dev/null +++ b/src/fiscal_svc/series/router.py @@ -0,0 +1,55 @@ +import uuid + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from sqlalchemy.ext.asyncio import AsyncSession + +from fiscal_svc.core.db import get_session +from fiscal_svc.series import service +from fiscal_svc.series.schemas import FiscalSeriesCreate, FiscalSeriesPatch, FiscalSeriesRead +from fiscal_svc.shared.errors import conflict +from fiscal_svc.tenancy.deps import require_product +from fiscal_svc.tenancy.models import Product + +router = APIRouter(prefix="/v1/series", tags=["series"]) + + +@router.post("", response_model=FiscalSeriesRead, status_code=status.HTTP_201_CREATED) +async def create_fiscal_series_endpoint( + payload: FiscalSeriesCreate, + product: Product = Depends(require_product), + session: AsyncSession = Depends(get_session), +) -> FiscalSeriesRead: + try: + series = await service.create_fiscal_series(session, product.id, payload) + except service.DuplicateFiscalSeriesError as exc: + raise conflict("duplicate_fiscal_series", str(exc)) from exc + return FiscalSeriesRead.model_validate(series) + + +@router.get("", response_model=list[FiscalSeriesRead]) +async def list_fiscal_series_endpoint( + tenant_ref: str = Query(..., max_length=64), + branch_ref: str = Query(..., max_length=64), + product: Product = Depends(require_product), + session: AsyncSession = Depends(get_session), +) -> list[FiscalSeriesRead]: + series_list = await service.list_fiscal_series(session, product.id, tenant_ref, branch_ref) + return [FiscalSeriesRead.model_validate(series) for series in series_list] + + +@router.patch("/{series_id}", response_model=FiscalSeriesRead) +async def update_fiscal_series_endpoint( + series_id: uuid.UUID, + payload: FiscalSeriesPatch, + product: Product = Depends(require_product), + session: AsyncSession = Depends(get_session), +) -> FiscalSeriesRead: + try: + series = await service.update_fiscal_series(session, product.id, series_id, payload) + except service.FiscalSeriesNotFoundError as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Série fiscal não encontrada" + ) from exc + except service.FiscalSeriesNumberRegressionError as exc: + raise conflict("fiscal_series_number_regression", str(exc)) from exc + return FiscalSeriesRead.model_validate(series) diff --git a/src/fiscal_svc/series/schemas.py b/src/fiscal_svc/series/schemas.py new file mode 100644 index 0000000..a7a8eb7 --- /dev/null +++ b/src/fiscal_svc/series/schemas.py @@ -0,0 +1,55 @@ +import uuid +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, field_validator + + +class FiscalSeriesCreate(BaseModel): + """POST /v1/series body. `serie` allows `0` (SEFAZ convention for + "série única"/sem série formal) -- `ge=0`, unlike `next_number` which + must start at 1 (`ge=1`, a document numbering never starts at 0). Only + `"55"` is accepted for now (Global Constraints/spec: F2 covers NF-e 55 + emission only -- `xml_builder.build_nfe` itself hardcodes `mod="55"`; + accepting `"65"` here would create a series nothing can ever allocate + against without silently wrong output).""" + + tenant_ref: str = Field(max_length=64) + branch_ref: str = Field(max_length=64) + document_model: Literal["55"] = "55" + serie: int = Field(ge=0) + next_number: int = Field(ge=1) + + +class FiscalSeriesRead(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: uuid.UUID + product_id: uuid.UUID + tenant_ref: str + branch_ref: str + document_model: str + serie: int + next_number: int + + +class FiscalSeriesPatch(BaseModel): + """PATCH /v1/series/{series_id} body -- ONLY `next_number` is editable + (a manual correction, e.g. realigning the counter after a migration + from another emissor). `tenant_ref`/`branch_ref`/`document_model`/ + `serie` are immutable once created -- delete and recreate the series + instead if it was set up wrong (no DELETE endpoint exists yet for this + resource per the F2 API contract; ported convention documented here for + when it does). `next_number` maps to a NOT NULL column, so explicit + `null` is rejected with 422, same convention as the auto's + `FiscalDocumentSeriesPatch`.""" + + model_config = ConfigDict(extra="forbid") + + next_number: int | None = Field(default=None, ge=1) + + @field_validator("next_number", mode="before") + @classmethod + def _reject_explicit_null(cls, value: object) -> object: + if value is None: + raise ValueError("não pode ser nulo (coluna obrigatória)") + return value diff --git a/src/fiscal_svc/series/service.py b/src/fiscal_svc/series/service.py new file mode 100644 index 0000000..844bfbd --- /dev/null +++ b/src/fiscal_svc/series/service.py @@ -0,0 +1,215 @@ +"""Task 4: `FiscalSeries` CRUD (create/list/update) -- ported from the +auto's `app/modules/tenants/service.py` FiscalDocumentSeries CRUD, porte +table applied (`organization_id`/`branch_id` -> `product_id`/`tenant_ref`/ +`branch_ref`). `allocate_fiscal_number` itself (the atomic number-granting +function) already lives in `documents.service` (Task 3, ported first since +`FiscalSeries` is its home table) -- this module owns everything ELSE about +a series: creating one, listing them, and the ADMIN edit path (`PATCH`) +with the "guard retroativo" (`next_number` can never regress below what +this série has already emitted) that Task 3's `FiscalSeries` docstring +explicitly deferred to this task. + +Deliberately its OWN `FiscalSeriesNotFoundError` (distinct class from +`documents.service.FiscalSeriesNotFoundError`, same name, different +module): that one is raised by a `(product_id, tenant_ref, branch_ref, +document_model, serie)` TUPLE lookup (`allocate_fiscal_number`'s exact +lookup shape); this one is raised by a bare `series_id` lookup (`PATCH +/v1/series/{series_id}`'s shape) -- the auto's own `tenants.service. +FiscalSeriesNotFoundError` supported BOTH shapes via optional constructor +args in one class; this port keeps the shapes SEPARATE instead of carrying +that same either/or constructor across two different modules-by-porte.""" +import uuid + +from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from fiscal_svc.documents.models import FiscalDocument, FiscalSeries +from fiscal_svc.series.schemas import FiscalSeriesCreate, FiscalSeriesPatch + + +class FiscalSeriesNotFoundError(Exception): + """`series_id` does not resolve to a live row for this `product_id` -- + same anti-oracle 404 either way (does not exist vs. belongs to another + product) as the rest of this service.""" + + def __init__(self, series_id: uuid.UUID): + self.series_id = series_id + super().__init__(f"Série fiscal {series_id} não encontrada") + + +class DuplicateFiscalSeriesError(Exception): + """Raised when a `(product_id, tenant_ref, branch_ref, document_model, + serie)` tuple collides with an existing `FiscalSeries` row -- live OR + soft-deleted (the DB constraint, Task 3's + `uq_fiscal_series_product_tenant_branch_model_serie`, has no `WHERE + deleted_at IS NULL`, so a soft-deleted series still blocks recreation + with the same tuple).""" + + def __init__(self, document_model: str, serie: int): + self.document_model = document_model + self.serie = serie + super().__init__( + f"Já existe uma série fiscal para o modelo {document_model} e série {serie} " + "neste branch_ref" + ) + + +class FiscalSeriesNumberRegressionError(Exception): + """The "requisito herdado" guard from Task 3's `FiscalSeries` docstring, + finally closed now that `FiscalDocument` exists to check against: + `PATCH /v1/series/{id}` setting `next_number` to a value that is NOT + strictly greater than the highest `numero` this series has already + emitted (a LIVE, i.e. non-soft-deleted, `FiscalDocument`) would let the + NEXT allocation hand out a number that was already used -- either an + outright repeat (SEFAZ duplicate-key rejection) or, worse, a silent + re-use if the earlier document was never transmitted. Blocked + unconditionally whenever the series has emitted at least one document, + regardless of whether the new `next_number` is higher or lower than the + CURRENT `next_number` -- "regression" here means "against what SEFAZ has + already seen for this série", not "against the previous column value".""" + + def __init__(self, series_id: uuid.UUID, next_number: int, max_numero: int): + self.series_id = series_id + self.next_number = next_number + self.max_numero = max_numero + super().__init__( + f"série {series_id} já emitiu até o número {max_numero}; " + f"next_number ({next_number}) deve ser maior que {max_numero}" + ) + + +def _is_fiscal_series_constraint_violation(exc: IntegrityError) -> bool: + """True iff `exc` violates `uq_fiscal_series_product_tenant_branch_ + model_serie` -- Task 3's migration names this constraint EXPLICITLY + (unlike the auto's un-named equivalent), so a plain substring match on + the name suffices; no need for the auto's "serie" + "unique constraint" + double-marker workaround (that existed there only because the + auto-generated constraint name there collided with the table's own + name).""" + detail = str(getattr(exc.orig, "args", [""])[0]) if exc.orig else str(exc) + return "uq_fiscal_series_product_tenant_branch_model_serie" in detail + + +async def _check_duplicate_fiscal_series( + session: AsyncSession, + product_id: uuid.UUID, + tenant_ref: str, + branch_ref: str, + document_model: str, + serie: int, +) -> None: + """Pre-flight check against LIVE series only -- fast, friendlier-error + happy path. Does NOT see soft-deleted rows; the DB constraint plus + `create_fiscal_series`'s `except IntegrityError` still block those, + same two-layer pattern as the rest of this codebase's uniqueness + guards.""" + result = await session.execute( + select(FiscalSeries.id).where( + FiscalSeries.product_id == product_id, + FiscalSeries.tenant_ref == tenant_ref, + FiscalSeries.branch_ref == branch_ref, + FiscalSeries.document_model == document_model, + FiscalSeries.serie == serie, + FiscalSeries.deleted_at.is_(None), + ) + ) + if result.scalar_one_or_none() is not None: + raise DuplicateFiscalSeriesError(document_model, serie) + + +async def create_fiscal_series( + session: AsyncSession, product_id: uuid.UUID, data: FiscalSeriesCreate +) -> FiscalSeries: + await _check_duplicate_fiscal_series( + session, product_id, data.tenant_ref, data.branch_ref, data.document_model, data.serie + ) + + series = FiscalSeries( + product_id=product_id, + tenant_ref=data.tenant_ref, + branch_ref=data.branch_ref, + document_model=data.document_model, + serie=data.serie, + next_number=data.next_number, + ) + session.add(series) + try: + await session.commit() + except IntegrityError as exc: + await session.rollback() + if _is_fiscal_series_constraint_violation(exc): + raise DuplicateFiscalSeriesError(data.document_model, data.serie) from exc + raise + await session.refresh(series) + return series + + +async def list_fiscal_series( + session: AsyncSession, product_id: uuid.UUID, tenant_ref: str, branch_ref: str +) -> list[FiscalSeries]: + """`(product_id, tenant_ref, branch_ref)`-scoped, non-soft-deleted + list, ordered by `(document_model, serie)` -- small per-branch catalog, + unpaginated (mirrors the auto's `list_fiscal_series`).""" + result = await session.execute( + select(FiscalSeries) + .where( + FiscalSeries.product_id == product_id, + FiscalSeries.tenant_ref == tenant_ref, + FiscalSeries.branch_ref == branch_ref, + FiscalSeries.deleted_at.is_(None), + ) + .order_by(FiscalSeries.document_model, FiscalSeries.serie) + ) + return list(result.scalars().all()) + + +async def update_fiscal_series( + session: AsyncSession, product_id: uuid.UUID, series_id: uuid.UUID, data: FiscalSeriesPatch +) -> FiscalSeries: + """Applies the allowlisted partial edit -- ONLY `next_number`. Uses + `.with_for_update()` + `.execution_options(populate_existing=True)` -- + the SAME two-part fix as `documents.service.allocate_fiscal_number` (see + its docstring): this PATCH can race a concurrent `allocate_fiscal_ + number` call against the SAME row, and the regression guard below needs + a freshly-locked read to be meaningful.""" + result = await session.execute( + select(FiscalSeries) + .where( + FiscalSeries.id == series_id, + FiscalSeries.product_id == product_id, + FiscalSeries.deleted_at.is_(None), + ) + .with_for_update() + .execution_options(populate_existing=True) + ) + series = result.scalar_one_or_none() + if series is None: + raise FiscalSeriesNotFoundError(series_id=series_id) + + changes = data.model_dump(exclude_unset=True) + + # Guard retroativo (Task 3's `FiscalSeries` docstring, closed here): + # roda DEPOIS do `.with_for_update()` acima (a mesma linha travada + # serializa esta checagem contra `allocate_fiscal_number`) e ANTES de + # aplicar qualquer mudança -- um `next_number` que regride é rejeitado + # inteiro, nenhum campo do patch é aplicado. + if "next_number" in changes: + max_result = await session.execute( + select(func.max(FiscalDocument.numero)).where( + FiscalDocument.series_id == series.id, + FiscalDocument.deleted_at.is_(None), + ) + ) + max_numero = max_result.scalar_one_or_none() + if max_numero is not None and changes["next_number"] <= max_numero: + raise FiscalSeriesNumberRegressionError( + series_id=series.id, next_number=changes["next_number"], max_numero=max_numero + ) + + for field, value in changes.items(): + setattr(series, field, value) + + await session.commit() + await session.refresh(series) + return series diff --git a/tests/certificates/__init__.py b/tests/certificates/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/certificates/test_certificates.py b/tests/certificates/test_certificates.py new file mode 100644 index 0000000..0287414 --- /dev/null +++ b/tests/certificates/test_certificates.py @@ -0,0 +1,387 @@ +"""Task 4: `POST/GET/DELETE /v1/certificados` -- ported from the auto's +`tests/modules/fiscal/test_certificate.py`, porte table applied (JWT bearer ++ `branch_id` path segment -> `X-Api-Key` + `tenant_ref`/`branch_ref` query +params; anti-oracle now by `(product_id, tenant_ref)` instead of +`organization_id`). + +Fixture de PFX: gerado em memória via `cryptography` (chave RSA 2048 + +certificado self-signed com o CNPJ no subject, formato ICP-Brasil e-CNPJ +real -- `CN=RAZAO SOCIAL:CNPJ` + atributo `SERIALNUMBER`) -- NUNCA um +certificado real no repo, mesma convenção do auto.""" +import asyncio +import uuid +from datetime import datetime, timedelta, timezone + +import pytest +from cryptography import x509 +from cryptography.fernet import Fernet +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.hazmat.primitives.serialization import pkcs12 +from cryptography.x509.oid import NameOID +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker + +from fiscal_svc.certificates import crypto as certificate_lib +from fiscal_svc.certificates import service as certificate_service +from fiscal_svc.core.db import get_session +from fiscal_svc.documents.models import FiscalCertificate +from fiscal_svc.main import app +from fiscal_svc.tenancy.service import create_product + + +@pytest.fixture(autouse=True) +def _override_db(db_session): + async def _get_session_override(): + yield db_session + + app.dependency_overrides[get_session] = _get_session_override + yield + app.dependency_overrides.clear() + + +@pytest.fixture(autouse=True) +def _fiscal_cert_encryption_key(monkeypatch): + monkeypatch.setenv("FISCAL_CERT_ENCRYPTION_KEY", Fernet.generate_key().decode("ascii")) + certificate_lib._get_fernet.cache_clear() + yield + certificate_lib._get_fernet.cache_clear() + + +def _build_test_pfx( + *, + cnpj: str = "14200166000187", + password: str = "correct-horse-battery", + not_valid_before: datetime | None = None, + not_valid_after: datetime | None = None, + cn: str | None = None, +) -> bytes: + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + subject = issuer = x509.Name( + [ + x509.NameAttribute(NameOID.COMMON_NAME, cn or f"EMPRESA TESTE LTDA:{cnpj}"), + x509.NameAttribute(NameOID.SERIAL_NUMBER, cnpj), + ] + ) + now = datetime.now(timezone.utc) + nvb = not_valid_before if not_valid_before is not None else now - timedelta(days=1) + nva = not_valid_after if not_valid_after is not None else now + timedelta(days=365) + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(nvb) + .not_valid_after(nva) + .sign(key, hashes.SHA256()) + ) + return pkcs12.serialize_key_and_certificates( + name=b"test", + key=key, + cert=cert, + cas=None, + encryption_algorithm=serialization.BestAvailableEncryption(password.encode("utf-8")), + ) + + +def _headers(api_key: str) -> dict[str, str]: + return {"X-Api-Key": api_key} + + +async def _upload(client, api_key, tenant_ref, branch_ref, cnpj, pfx_bytes, password): + return await client.post( + "/v1/certificados", + params={"tenant_ref": tenant_ref, "branch_ref": branch_ref}, + files={"file": ("cert.pfx", pfx_bytes, "application/x-pkcs12")}, + data={"password": password, "cnpj": cnpj}, + headers=_headers(api_key), + ) + + +@pytest.mark.asyncio +async def test_upload_valid_pfx_returns_201_with_correct_metadata_and_encrypted_binary(db_session): + key = f"k-{uuid.uuid4().hex}" + await create_product(db_session, name="auto", api_key=key) + pfx_bytes = _build_test_pfx(cnpj="14200166000187", password="senha123") + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await _upload(client, key, "tenant-1", "branch-1", "14200166000187", pfx_bytes, "senha123") + + assert response.status_code == 201, response.text + body = response.json() + assert body["branch_ref"] == "branch-1" + assert body["tenant_ref"] == "tenant-1" + assert body["cnpj_certificado"] == "14200166000187" + assert "EMPRESA TESTE LTDA" in body["subject_cn"] + assert "pfx_encrypted" not in body + assert "password_encrypted" not in body + + result = await db_session.execute( + select(FiscalCertificate).where(FiscalCertificate.id == uuid.UUID(body["id"])) + ) + row = result.scalar_one() + assert row.pfx_encrypted != pfx_bytes + assert pfx_bytes not in row.pfx_encrypted + assert certificate_lib.decrypt_bytes(row.pfx_encrypted) == pfx_bytes + assert certificate_lib.decrypt_bytes(row.password_encrypted) == b"senha123" + + +@pytest.mark.asyncio +async def test_upload_wrong_password_is_422(db_session): + key = f"k-{uuid.uuid4().hex}" + await create_product(db_session, name="auto", api_key=key) + pfx_bytes = _build_test_pfx(cnpj="14200166000187", password="correct-pw") + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await _upload(client, key, "t1", "b1", "14200166000187", pfx_bytes, "wrong-pw") + + assert response.status_code == 422, response.text + + +@pytest.mark.asyncio +async def test_upload_non_pfx_file_is_422(db_session): + key = f"k-{uuid.uuid4().hex}" + await create_product(db_session, name="auto", api_key=key) + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await _upload(client, key, "t1", "b1", "14200166000187", b"isso nao e um pfx", "qualquer") + + assert response.status_code == 422, response.text + + +@pytest.mark.asyncio +async def test_upload_expired_certificate_is_422(db_session): + key = f"k-{uuid.uuid4().hex}" + await create_product(db_session, name="auto", api_key=key) + now = datetime.now(timezone.utc) + pfx_bytes = _build_test_pfx( + cnpj="14200166000187", password="senha123", + not_valid_before=now - timedelta(days=400), not_valid_after=now - timedelta(days=10), + ) + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await _upload(client, key, "t1", "b1", "14200166000187", pfx_bytes, "senha123") + + assert response.status_code == 422, response.text + + +@pytest.mark.asyncio +async def test_upload_not_valid_before_in_future_is_422(db_session): + key = f"k-{uuid.uuid4().hex}" + await create_product(db_session, name="auto", api_key=key) + now = datetime.now(timezone.utc) + pfx_bytes = _build_test_pfx( + cnpj="14200166000187", password="senha123", + not_valid_before=now + timedelta(days=5), not_valid_after=now + timedelta(days=400), + ) + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await _upload(client, key, "t1", "b1", "14200166000187", pfx_bytes, "senha123") + + assert response.status_code == 422, response.text + + +@pytest.mark.asyncio +async def test_upload_cnpj_mismatch_between_payload_and_certificate_is_422_naming_both(db_session): + key = f"k-{uuid.uuid4().hex}" + await create_product(db_session, name="auto", api_key=key) + pfx_bytes = _build_test_pfx(cnpj="99887766000155", password="senha123") + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await _upload(client, key, "t1", "b1", "14200166000187", pfx_bytes, "senha123") + + assert response.status_code == 422, response.text + assert "14200166000187" in response.text + assert "99887766000155" in response.text + + +@pytest.mark.asyncio +async def test_get_certificate_returns_metadata_without_binary(db_session): + key = f"k-{uuid.uuid4().hex}" + await create_product(db_session, name="auto", api_key=key) + pfx_bytes = _build_test_pfx(cnpj="14200166000187", password="senha123") + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + await _upload(client, key, "t1", "b1", "14200166000187", pfx_bytes, "senha123") + response = await client.get( + "/v1/certificados", params={"tenant_ref": "t1", "branch_ref": "b1"}, headers=_headers(key) + ) + + assert response.status_code == 200, response.text + body = response.json() + assert body["cnpj_certificado"] == "14200166000187" + assert "pfx_encrypted" not in body + assert "password_encrypted" not in body + + +@pytest.mark.asyncio +async def test_cross_product_branch_is_404_not_leaked(db_session): + key_a = f"k-{uuid.uuid4().hex}" + key_b = f"k-{uuid.uuid4().hex}" + await create_product(db_session, name="auto", api_key=key_a) + await create_product(db_session, name="crm", api_key=key_b) + pfx_bytes = _build_test_pfx(cnpj="14200166000187", password="senha123") + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + await _upload(client, key_a, "t1", "b1", "14200166000187", pfx_bytes, "senha123") + get_response = await client.get( + "/v1/certificados", params={"tenant_ref": "t1", "branch_ref": "b1"}, headers=_headers(key_b) + ) + + assert get_response.status_code == 404, get_response.text + + +@pytest.mark.asyncio +async def test_cross_tenant_ref_same_product_is_404(db_session): + """Anti-oracle boundary is `(product_id, tenant_ref)` (design spec + decision #4) -- even under the SAME product, a wrong `tenant_ref` for a + real `branch_ref` must 404, not leak the certificate.""" + key = f"k-{uuid.uuid4().hex}" + await create_product(db_session, name="auto", api_key=key) + pfx_bytes = _build_test_pfx(cnpj="14200166000187", password="senha123") + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + await _upload(client, key, "tenant-a", "b1", "14200166000187", pfx_bytes, "senha123") + get_response = await client.get( + "/v1/certificados", params={"tenant_ref": "tenant-b", "branch_ref": "b1"}, headers=_headers(key) + ) + + assert get_response.status_code == 404, get_response.text + + +@pytest.mark.asyncio +async def test_second_upload_replaces_the_first_soft_deleting_it(db_session): + key = f"k-{uuid.uuid4().hex}" + await create_product(db_session, name="auto", api_key=key) + first_pfx = _build_test_pfx(cnpj="14200166000187", password="senha123", cn="PRIMEIRO:14200166000187") + second_pfx = _build_test_pfx(cnpj="14200166000187", password="senha456", cn="SEGUNDO:14200166000187") + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + first_response = await _upload(client, key, "t1", "b1", "14200166000187", first_pfx, "senha123") + second_response = await _upload(client, key, "t1", "b1", "14200166000187", second_pfx, "senha456") + get_response = await client.get( + "/v1/certificados", params={"tenant_ref": "t1", "branch_ref": "b1"}, headers=_headers(key) + ) + + assert first_response.status_code == 201, first_response.text + assert second_response.status_code == 201, second_response.text + assert get_response.json()["id"] == second_response.json()["id"] + assert "SEGUNDO" in get_response.json()["subject_cn"] + + result = await db_session.execute( + select(FiscalCertificate).where(FiscalCertificate.branch_ref == "b1") + ) + rows = result.scalars().all() + assert len(rows) == 2 + live = [r for r in rows if r.deleted_at is None] + dead = [r for r in rows if r.deleted_at is not None] + assert len(live) == 1 + assert len(dead) == 1 + assert live[0].id == uuid.UUID(second_response.json()["id"]) + assert dead[0].id == uuid.UUID(first_response.json()["id"]) + + +@pytest.mark.asyncio +async def test_delete_soft_deletes_and_get_then_404s(db_session): + key = f"k-{uuid.uuid4().hex}" + await create_product(db_session, name="auto", api_key=key) + pfx_bytes = _build_test_pfx(cnpj="14200166000187", password="senha123") + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + await _upload(client, key, "t1", "b1", "14200166000187", pfx_bytes, "senha123") + delete_response = await client.delete( + "/v1/certificados", params={"tenant_ref": "t1", "branch_ref": "b1"}, headers=_headers(key) + ) + get_response = await client.get( + "/v1/certificados", params={"tenant_ref": "t1", "branch_ref": "b1"}, headers=_headers(key) + ) + + assert delete_response.status_code == 204, delete_response.text + assert get_response.status_code == 404, get_response.text + + +@pytest.mark.asyncio +async def test_delete_without_certificate_is_404(db_session): + key = f"k-{uuid.uuid4().hex}" + await create_product(db_session, name="auto", api_key=key) + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.delete( + "/v1/certificados", params={"tenant_ref": "t1", "branch_ref": "b1"}, headers=_headers(key) + ) + + assert response.status_code == 404, response.text + + +@pytest.mark.asyncio +async def test_missing_api_key_is_401(db_session): + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.get("/v1/certificados", params={"tenant_ref": "t1", "branch_ref": "b1"}) + + assert response.status_code == 401, response.text + + +@pytest.mark.asyncio +async def test_concurrent_uploads_for_same_product_branch_only_one_wins_the_other_gets_409( + test_engine, db_session +): + """Mesma prova do auto (`test_concurrent_uploads_for_same_branch_only_ + one_wins_the_other_gets_409`): duas `AsyncSession` distintas contra o + MESMO `test_engine`, disparadas via `asyncio.gather` -- concorrência + REAL, não simulada.""" + key = f"k-{uuid.uuid4().hex}" + product = await create_product(db_session, name="auto", api_key=key) + pfx_a = _build_test_pfx(cnpj="14200166000187", password="senha123", cn="A:14200166000187") + pfx_b = _build_test_pfx(cnpj="14200166000187", password="senha456", cn="B:14200166000187") + + session_maker = async_sessionmaker(test_engine, expire_on_commit=False) + session_a = session_maker() + session_b = session_maker() + try: + results = await asyncio.gather( + certificate_service.upload_certificate( + session_a, product.id, "t1", "b1", "14200166000187", pfx_a, "senha123" + ), + certificate_service.upload_certificate( + session_b, product.id, "t1", "b1", "14200166000187", pfx_b, "senha456" + ), + return_exceptions=True, + ) + finally: + await session_a.close() + await session_b.close() + + successes = [r for r in results if not isinstance(r, BaseException)] + errors = [r for r in results if isinstance(r, BaseException)] + + assert len(successes) == 1, f"expected exactly 1 winner, got {len(successes)}: {results!r}" + assert len(errors) == 1, f"expected exactly 1 conflict error, got {len(errors)}: {results!r}" + assert isinstance(errors[0], certificate_service.CertificateUploadConflictError), errors[0] + + live_result = await db_session.execute( + select(FiscalCertificate).where( + FiscalCertificate.product_id == product.id, + FiscalCertificate.branch_ref == "b1", + FiscalCertificate.deleted_at.is_(None), + ) + ) + live_rows = live_result.scalars().all() + assert len(live_rows) == 1, ( + f"expected exactly 1 live certificate after the race, found {len(live_rows)} -- " + "the DB-level partial unique index should have blocked the second insert" + ) diff --git a/tests/series/__init__.py b/tests/series/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/series/test_series.py b/tests/series/test_series.py new file mode 100644 index 0000000..789f262 --- /dev/null +++ b/tests/series/test_series.py @@ -0,0 +1,263 @@ +"""Task 4: `POST/GET /v1/series`, `PATCH /v1/series/{id}` -- ported from the +auto's `tests/modules/tenants/test_fiscal_series.py` CRUD half (allocation +itself, and its concurrency proofs, are Task 3's +`tests/documents/test_allocate_fiscal_number.py`). Covers: CRUD happy path, +duplicate tuple -> 409, anti-oracle 404 (cross-product), missing API key -> +401, and the "guard retroativo" this Task closes: `PATCH .../next_number` +regressing below the highest `numero` a series has already emitted -> 409.""" +import uuid +from datetime import datetime, timezone + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select + +from fiscal_svc.core.db import get_session +from fiscal_svc.documents.models import FiscalDocument, FiscalSeries +from fiscal_svc.main import app +from fiscal_svc.tenancy.service import create_product + + +@pytest.fixture(autouse=True) +def _override_db(db_session): + async def _get_session_override(): + yield db_session + + app.dependency_overrides[get_session] = _get_session_override + yield + app.dependency_overrides.clear() + + +def _headers(api_key: str) -> dict[str, str]: + return {"X-Api-Key": api_key} + + +async def _product_and_key(db_session, name="auto"): + key = f"k-{uuid.uuid4().hex}" + product = await create_product(db_session, name=name, api_key=key) + return product, key + + +@pytest.mark.asyncio +async def test_create_fiscal_series_returns_201(db_session): + _, key = await _product_and_key(db_session) + payload = {"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1014} + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.post("/v1/series", json=payload, headers=_headers(key)) + + assert response.status_code == 201, response.text + body = response.json() + assert body["tenant_ref"] == "t1" + assert body["branch_ref"] == "b1" + assert body["document_model"] == "55" + assert body["serie"] == 1 + assert body["next_number"] == 1014 + + +@pytest.mark.asyncio +async def test_list_fiscal_series_returns_created_series(db_session): + _, key = await _product_and_key(db_session) + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + await client.post( + "/v1/series", + json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1}, + headers=_headers(key), + ) + await client.post( + "/v1/series", + json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 2, "next_number": 1}, + headers=_headers(key), + ) + response = await client.get( + "/v1/series", params={"tenant_ref": "t1", "branch_ref": "b1"}, headers=_headers(key) + ) + + assert response.status_code == 200, response.text + series_list = response.json() + assert len(series_list) == 2 + assert {s["serie"] for s in series_list} == {1, 2} + + +@pytest.mark.asyncio +async def test_patch_fiscal_series_updates_next_number(db_session): + _, key = await _product_and_key(db_session) + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + create_response = await client.post( + "/v1/series", + json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1}, + headers=_headers(key), + ) + series_id = create_response.json()["id"] + + response = await client.patch( + f"/v1/series/{series_id}", json={"next_number": 500}, headers=_headers(key) + ) + + assert response.status_code == 200, response.text + assert response.json()["next_number"] == 500 + + +async def _series_with_documents(db_session, product, *, max_numero: int, next_number: int) -> FiscalSeries: + """Fabrica uma série + um `FiscalDocument` VIVO com `numero=max_numero` + -- exatamente como o guard vai ler (`numero`/`series_id`/`deleted_at IS + NULL`), sem passar pelo fluxo real de emissão (Task 5).""" + series = FiscalSeries( + product_id=product.id, tenant_ref="t1", branch_ref="b1", + document_model="55", serie=1, next_number=next_number, + ) + db_session.add(series) + await db_session.flush() + + document = FiscalDocument( + product_id=product.id, tenant_ref="t1", branch_ref="b1", series_id=series.id, + document_model="55", serie=1, numero=max_numero, + chave_acesso=str(uuid.uuid4().int)[:44].zfill(44), codigo_numerico="12345678", + status="ASSINADO", ambiente="homologacao", xml_assinado="", + ) + db_session.add(document) + await db_session.commit() + await db_session.refresh(series) + return series + + +@pytest.mark.asyncio +async def test_patch_fiscal_series_next_number_regression_is_409(db_session): + product, key = await _product_and_key(db_session) + series = await _series_with_documents(db_session, product, max_numero=1014, next_number=1015) + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + regressed = await client.patch( + f"/v1/series/{series.id}", json={"next_number": 1000}, headers=_headers(key) + ) + equal_to_max = await client.patch( + f"/v1/series/{series.id}", json={"next_number": 1014}, headers=_headers(key) + ) + allowed = await client.patch( + f"/v1/series/{series.id}", json={"next_number": 1015}, headers=_headers(key) + ) + + assert regressed.status_code == 409, regressed.text + assert regressed.json()["detail"]["code"] == "fiscal_series_number_regression" + assert equal_to_max.status_code == 409, equal_to_max.text + assert allowed.status_code == 200, allowed.text + assert allowed.json()["next_number"] == 1015 + + +@pytest.mark.asyncio +async def test_patch_fiscal_series_next_number_ignores_soft_deleted_documents(db_session): + """O guard só olha documentos VIVOS -- um `FiscalDocument` + soft-deletado não deve travar o `next_number` para sempre.""" + product, key = await _product_and_key(db_session) + series = await _series_with_documents(db_session, product, max_numero=1014, next_number=1015) + + result = await db_session.execute(select(FiscalDocument).where(FiscalDocument.series_id == series.id)) + document = result.scalar_one() + document.deleted_at = datetime.now(timezone.utc) + await db_session.commit() + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.patch( + f"/v1/series/{series.id}", json={"next_number": 1}, headers=_headers(key) + ) + + assert response.status_code == 200, response.text + assert response.json()["next_number"] == 1 + + +@pytest.mark.asyncio +async def test_patch_fiscal_series_rejects_explicit_null(db_session): + _, key = await _product_and_key(db_session) + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + create_response = await client.post( + "/v1/series", + json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1}, + headers=_headers(key), + ) + series_id = create_response.json()["id"] + + response = await client.patch( + f"/v1/series/{series_id}", json={"next_number": None}, headers=_headers(key) + ) + + assert response.status_code == 422, response.text + + +@pytest.mark.asyncio +async def test_create_duplicate_fiscal_series_is_409(db_session): + _, key = await _product_and_key(db_session) + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + await client.post( + "/v1/series", + json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1}, + headers=_headers(key), + ) + response = await client.post( + "/v1/series", + json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 999}, + headers=_headers(key), + ) + + assert response.status_code == 409, response.text + assert response.json()["detail"]["code"] == "duplicate_fiscal_series" + + +@pytest.mark.asyncio +async def test_different_serie_same_model_is_ok(db_session): + _, key = await _product_and_key(db_session) + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + await client.post( + "/v1/series", + json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1}, + headers=_headers(key), + ) + response = await client.post( + "/v1/series", + json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 2, "next_number": 1}, + headers=_headers(key), + ) + + assert response.status_code == 201, response.text + + +@pytest.mark.asyncio +async def test_patch_fiscal_series_other_product_series_is_404(db_session): + product_a, key_a = await _product_and_key(db_session, name="auto") + _, key_b = await _product_and_key(db_session, name="crm") + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + create_response = await client.post( + "/v1/series", + json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1}, + headers=_headers(key_a), + ) + series_id = create_response.json()["id"] + + response = await client.patch( + f"/v1/series/{series_id}", json={"next_number": 42}, headers=_headers(key_b) + ) + + assert response.status_code == 404, response.text + + +@pytest.mark.asyncio +async def test_list_fiscal_series_missing_api_key_is_401(db_session): + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.get("/v1/series", params={"tenant_ref": "t1", "branch_ref": "b1"}) + + assert response.status_code == 401, response.text diff --git a/uv.lock b/uv.lock index a7d8d27..e65a16a 100644 --- a/uv.lock +++ b/uv.lock @@ -961,6 +961,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0b/d7/1959b9648791274998a9c3526f6d0ec8fd2233e4d4acce81bbae76b44b2a/python_dotenv-1.2.2-py3-none-any.whl", hash = "sha256:1d8214789a24de455a8b8bd8ae6fe3c6b69a5e3d64aa8a8e5d68e694bbcb285a", size = 22101, upload-time = "2026-03-01T16:00:25.09Z" }, ] +[[package]] +name = "python-multipart" +version = "0.0.32" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/5b/42/55c32bb9b12693c092ad250a0e82edb5b31ddeda6eb772de5f308b3804ad/python_multipart-0.0.32.tar.gz", hash = "sha256:be54b7f3fa167bb83e4fcd936b887b708f4e57fe75911c02aebf53efaf8d938e", size = 46881, upload-time = "2026-06-04T16:18:58.647Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e1/04/e8135ebd1ad02c56ec633277529b2602ff99ff634be76cdba5744cf554fd/python_multipart-0.0.32-py3-none-any.whl", hash = "sha256:ff6d3f776f16878c894e52e107296ffc890e913c611b1a4ec6c44e2821fe2e23", size = 30042, upload-time = "2026-06-04T16:18:57.319Z" }, +] + [[package]] name = "pytz" version = "2026.2" @@ -1064,6 +1073,7 @@ dependencies = [ { name = "passlib", extra = ["bcrypt"] }, { name = "pydantic" }, { name = "pydantic-settings" }, + { name = "python-multipart" }, { name = "sowai-fiscal" }, { name = "sqlalchemy", extra = ["asyncio"] }, { name = "uvicorn", extra = ["standard"] }, @@ -1087,6 +1097,7 @@ requires-dist = [ { name = "passlib", extras = ["bcrypt"], specifier = ">=1.7.4" }, { name = "pydantic", specifier = ">=2.13.4" }, { name = "pydantic-settings", specifier = ">=2.0" }, + { name = "python-multipart", specifier = ">=0.0.20" }, { name = "sowai-fiscal", git = "https://git.sowai.com.br/jonatan/sowai-fiscal.git?rev=v0.1.0" }, { name = "sqlalchemy", extras = ["asyncio"], specifier = ">=2.0" }, { name = "uvicorn", extras = ["standard"], specifier = ">=0.49.0" },