ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature/correcion-hash

Browse Source
This commit is contained in:
hreyes
2026-04-13 15:57:40 -06:00
parent 233da55f40
commit 8f4fd7ee90
8 changed files with 208 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.env.example
View File

@@ -61,8 +61,8 @@ S3_USE_SSL=false
S3_FILE_STORAGE=true
S3_PRESIGNED_EXPIRES_SECONDS=3600
COVE_FIEL_PASSWORD=
COVE_FIEL_HASH_KEY=
COVE_FIEL_HASH_IV=
# ----- Sitar API -----
SITAR_API_URL=http://api.sitar.aduanasoft.com

2
backend/.env.example
View File

@@ -28,8 +28,8 @@ CORS_ORIGINS=http://localhost:5173,http://localhost:3000
# License Service
LICENSE_CHECK_ENABLED=True
COVE_FIEL_PASSWORD=
COVE_FIEL_HASH_KEY=
COVE_FIEL_HASH_IV=
# Synchronization (Hub & Spoke)
SYNC_SECRET_TOKEN=change-this-sync-token-in-production

14
backend/api/v1/modules/a76/factura_cove/external_service.py
View File

@@ -2,6 +2,7 @@ from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Dict
import logging
import httpx
@@ -9,6 +10,9 @@ from core.config import settings
from .schemas import FacturaCoveRequest
logger = logging.getLogger(__name__)
@dataclass
class CoveExternalResult:
"""
@@ -60,6 +64,16 @@ class CoveExternalService:
"""
url = f"{self.base_url.rstrip('/')}/api/v1/factura-cove/generar-factura-cove"
json_payload = payload.model_dump(mode="json")
configuracion_vu = json_payload.get("configuracion_vu") or {}
logger.info(
"Sending COVE payload: invoice=%s rfc_vu=%s clave_fiel_len=%s clave_ws_len=%s cer_len=%s key_len=%s",
json_payload.get("numero_factura"),
configuracion_vu.get("rfc_usuario_vu"),
len(configuracion_vu.get("clave_fiel") or ""),
len(configuracion_vu.get("clave_webservice") or ""),
len(configuracion_vu.get("archivo_cer_base64") or ""),
len(configuracion_vu.get("archivo_key_base64") or ""),
)
# NOTA: verify=False desactiva la validación de certificado SSL.
# Esto es útil en entornos de desarrollo o cuando el entorno no confía

142
backend/api/v1/modules/a76/factura_cove/service.py
View File

@@ -1,11 +1,12 @@
from __future__ import annotations
import base64
import hashlib
from dataclasses import dataclass
from decimal import Decimal
from typing import List, Tuple
from cryptography.hazmat.primitives import padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from sqlalchemy.orm import Session
from core.config import settings
@@ -79,23 +80,46 @@ class FacturaCoveDomainService:
return InvoiceContext(invoice=invoice, broker=broker, vu=vu)
def _hash_fiel(self, raw_fiel: str) -> str:
def _get_company(self, ctx: InvoiceContext) -> Company | None:
company_id = getattr(ctx.invoice, "company_id", None)
if not company_id:
return None
return self.db.get(Company, company_id)
def _get_company_fiel_certificate(self, company: Company | None):
if not company:
return None
for certificate in company.digital_certificates or []:
if (certificate.certificate_type or "").strip().lower() == "fiel":
return certificate
return None
def _encrypt_fiel(self, raw_fiel: str) -> str:
"""
Calcula un hash determinista (base64) de la clave FIEL usando una
clave de encriptado fija obtenida desde configuración. Este valor es
el que se enviará como `clave_fiel` al API externo cuando exista una
clave FIEL capturada en VU.
Cifra la clave FIEL con el mismo esquema del sistema legado PHP:
AES-256-CBC + PKCS7 + base64.
"""
if not raw_fiel:
normalized_fiel = (raw_fiel or "").strip()
if not normalized_fiel:
return ""
hash_key = (settings.COVE_FIEL_HASH_KEY or "").strip()
if not hash_key:
encryption_key = (settings.COVE_FIEL_HASH_KEY or "").encode("utf-8")
encryption_iv = (settings.COVE_FIEL_HASH_IV or "").encode("utf-8")
if not encryption_key or not encryption_iv:
return ""
data = f"{hash_key}:{raw_fiel}".encode("utf-8")
digest = hashlib.sha256(data).digest()
return base64.b64encode(digest).decode("ascii")
key_bytes = encryption_key[:32].ljust(32, b"\0")
iv_bytes = encryption_iv[:16].ljust(16, b"\0")
padder = padding.PKCS7(algorithms.AES.block_size).padder()
padded_data = padder.update(normalized_fiel.encode("utf-8")) + padder.finalize()
cipher = Cipher(algorithms.AES(key_bytes), modes.CBC(iv_bytes))
encryptor = cipher.encryptor()
encrypted = encryptor.update(padded_data) + encryptor.finalize()
return base64.b64encode(encrypted).decode("ascii")
def _build_configuracion_vu(self, ctx: InvoiceContext, errors: ErrorCollector) -> ConfiguracionVU | None:
"""
@@ -106,13 +130,16 @@ class FacturaCoveDomainService:
al API externo de COVE.
"""
vu = ctx.vu
company = self._get_company(ctx)
company_vu = company.ventanilla_unica if company else None
company_fiel_certificate = self._get_company_fiel_certificate(company)
if not vu:
if not vu and not company_vu and not company_fiel_certificate:
errors.add_error(
field="vu",
message="La factura no tiene configuración VU asociada en el agente aduanal",
message="La factura no tiene configuración VU asociada ni configuración VU/certificado FIEL en la empresa",
solution=[
"Configura los datos VU del agente aduanal y sube certificado (.cer) y llave (.key) antes de generar COVE."
"Configura los datos VU del agente aduanal o de la empresa y sube certificado (.cer) y llave (.key) antes de generar COVE."
],
code="MISSING_VU_CONFIGURATION",
)
@@ -121,23 +148,32 @@ class FacturaCoveDomainService:
# Determinar usuario efectivo de WebService:
# - Preferimos el usuario configurado en VU (web_service_user)
# - Si no existe, usamos el de DODA-PITA (doda_web_service_user)
# - Si no existe, usamos la configuración VU de la empresa
effective_ws_user = (
(vu.web_service_user or vu.doda_web_service_user or "").strip() if vu else ""
(
vu.web_service_user
or vu.doda_web_service_user
or getattr(company_vu, "webservice_user", None)
or ""
).strip()
if (vu or company_vu)
else ""
)
# Determinar clave FIEL efectiva.
# Orden de prioridad:
# 1) Si existe vu.fiel_access_key, se hashea.
# 2) En caso contrario, se usa la variable de entorno COVE_FIEL_PASSWORD.
# Determinar clave FIEL efectiva desde la configuración persistida.
# Se envía cifrada con el mismo esquema AES-256-CBC del sistema legado.
clave_fiel_value = ""
if vu and getattr(vu, "fiel_access_key", None):
# Usar la clave capturada en VU, hasheada con hashlib + COVE_FIEL_HASH_KEY
clave_fiel_value = self._hash_fiel(vu.fiel_access_key or "")
if not clave_fiel_value:
# Fallback: usar la contraseña FIEL configurada vía entorno
clave_fiel_value = (settings.COVE_FIEL_PASSWORD or "").strip()
clave_fiel_value = self._encrypt_fiel(vu.fiel_access_key or "")
elif company_fiel_certificate:
# Fallback en base de datos: certificado FIEL de la empresa
company_fiel_secret = (
getattr(company_fiel_certificate, "access_key", None)
or getattr(company_fiel_certificate, "password", None)
or ""
)
clave_fiel_value = self._encrypt_fiel(str(company_fiel_secret))
# Validación básica de credenciales VU: para COVE necesitamos al menos
# un usuario de web service (VU o DODA) y una clave FIEL no vacía.
@@ -146,8 +182,8 @@ class FacturaCoveDomainService:
field="vu",
message="Faltan credenciales de web service o clave FIEL en VU",
solution=[
"Captura usuario y clave de web service en la pestaña VU o DODA del agente "
"y la clave FIEL en la configuración VU del agente aduanal."
"Captura usuario y clave de web service en la pestaña VU o DODA del agente, "
"o completa la configuración VU de la empresa y su certificado FIEL."
],
code="MISSING_VU_CREDENTIALS",
)
@@ -155,20 +191,34 @@ class FacturaCoveDomainService:
if not clave_fiel_value:
errors.add_error(
field="vu.clave_fiel",
message="La clave FIEL para COVE no está configurada. Define COVE_FIEL_PASSWORD o captura la FIEL en VU.",
message="La clave FIEL para COVE no está configurada ni en VU ni en la empresa.",
solution=[
"Configura la variable de entorno COVE_FIEL_PASSWORD con la contraseña FIEL de VUCEM, "
"o captura la clave FIEL en la configuración VU del agente aduanal."
"Captura la clave FIEL en la configuración VU del agente aduanal o en el certificado FIEL de la empresa."
],
code="MISSING_FIEL_PASSWORD",
)
if not (vu.certificate_path and vu.key_path):
certificate_path = (
(getattr(vu, "certificate_path", None) or "").strip() if vu else ""
) or (
(getattr(company_fiel_certificate, "cer_file_path", None) or "").strip()
if company_fiel_certificate
else ""
)
key_path = (
(getattr(vu, "key_path", None) or "").strip() if vu else ""
) or (
(getattr(company_fiel_certificate, "key_file_path", None) or "").strip()
if company_fiel_certificate
else ""
)
if not (certificate_path and key_path):
errors.add_error(
field="vu",
message="No hay rutas de certificado o llave en VU",
solution=[
"Sube el certificado (.cer) y la llave (.key) del VU desde el módulo de agentes aduanales."
"Sube el certificado (.cer) y la llave (.key) en la configuración VU del agente aduanal o en los certificados digitales de la empresa."
],
code="MISSING_VU_CERT_KEY",
)
@@ -179,39 +229,43 @@ class FacturaCoveDomainService:
key_b64 = None
try:
if not object_exists(vu.certificate_path):
if not object_exists(certificate_path):
errors.add_error(
field="vu.certificate_path",
message="El certificado VU no existe en el almacenamiento de objetos",
solution=["Vuelve a subir el certificado VU en el agente aduanal."],
solution=["Vuelve a subir el certificado en la configuración VU del agente aduanal o en los certificados digitales de la empresa."],
code="VU_CERT_NOT_FOUND",
)
else:
cer_bytes = get_object_bytes(vu.certificate_path)
cer_bytes = get_object_bytes(certificate_path)
cer_b64 = base64.b64encode(cer_bytes).decode("ascii")
if not object_exists(vu.key_path):
if not object_exists(key_path):
errors.add_error(
field="vu.key_path",
message="La llave VU no existe en el almacenamiento de objetos",
solution=["Vuelve a subir la llave VU en el agente aduanal."],
solution=["Vuelve a subir la llave en la configuración VU del agente aduanal o en los certificados digitales de la empresa."],
code="VU_KEY_NOT_FOUND",
)
else:
key_bytes = get_object_bytes(vu.key_path)
key_bytes = get_object_bytes(key_path)
key_b64 = base64.b64encode(key_bytes).decode("ascii")
except Exception as exc: # pragma: no cover - errores de IO externos
errors.add_error(
field="vu",
message="Error leyendo certificados VU desde almacenamiento de objetos.",
solution=["Verifica la configuración de MinIO/S3 y las rutas de certificados/llaves en la configuración VU."],
solution=["Verifica la configuración de MinIO/S3 y las rutas de certificados/llaves en VU o en los certificados digitales de la empresa."],
code="VU_STORAGE_ERROR",
)
if errors.has_errors():
return None
rfc_usuario_vu = vu.query_tax_id or ""
rfc_usuario_vu = (
(getattr(vu, "query_tax_id", None) or "").strip() if vu else ""
) or (
(getattr(company_vu, "query_rfc", None) or "").strip() if company_vu else ""
)
# Clave/token del webservice: usar el valor de VU si existe, o una
# clave fija de pruebas mientras se termina la configuración real.
@@ -221,7 +275,11 @@ class FacturaCoveDomainService:
return ConfiguracionVU(
rfc_usuario_vu=rfc_usuario_vu,
clave_webservice=(vu.web_service_access_key or hardcoded_ws_key),
clave_webservice=(
(getattr(vu, "web_service_access_key", None) or "").strip()
or (getattr(company_vu, "webservice_password", None) or "").strip()
or hardcoded_ws_key
),
archivo_cer_base64=cer_b64 or "",
archivo_key_base64=key_b64 or "",
clave_fiel=clave_fiel_value,

2
backend/core/config.py
View File

@@ -53,8 +53,8 @@ class Settings(BaseSettings):
# External APIs
SITAR_API_URL: str = "api.sitar.aduanasoft.com:880"
COVE_API_URL: str = ""
COVE_FIEL_PASSWORD: str = ""
COVE_FIEL_HASH_KEY: str = ""
COVE_FIEL_HASH_IV: str = ""
SITAR_API_USER: str = ""
SITAR_API_PASSWORD: str = ""

78
backend/tests/unit/factura_cove/test_service.py
View File

@@ -5,7 +5,7 @@ from core.exceptions import ErrorCollector
def test_build_configuracion_vu_returns_validation_error_when_vu_is_missing() -> None:
service = FacturaCoveDomainService(db=SimpleNamespace())
service = FacturaCoveDomainService(db=SimpleNamespace(get=lambda *args, **kwargs: None))
ctx = InvoiceContext(
invoice=SimpleNamespace(),
broker=None,
@@ -20,10 +20,82 @@ def test_build_configuracion_vu_returns_validation_error_when_vu_is_missing() ->
assert errors.get_errors() == [
{
"field": "vu",
"message": "La factura no tiene configuración VU asociada en el agente aduanal",
"message": "La factura no tiene configuración VU asociada ni configuración VU/certificado FIEL en la empresa",
"solution": [
"Configura los datos VU del agente aduanal y sube certificado (.cer) y llave (.key) antes de generar COVE."
"Configura los datos VU del agente aduanal o de la empresa y sube certificado (.cer) y llave (.key) antes de generar COVE."
],
"code": "MISSING_VU_CONFIGURATION",
}
]
def test_build_configuracion_vu_uses_company_fiel_when_vu_has_no_fiel(monkeypatch) -> None:
company = SimpleNamespace(
ventanilla_unica=SimpleNamespace(
webservice_user="company-ws-user",
webservice_password="company-ws-pass",
query_rfc="RFCEMPRESA123",
),
digital_certificates=[
SimpleNamespace(
certificate_type="fiel",
access_key="company-fiel-access",
cer_file_path="company/fiel.cer",
key_file_path="company/fiel.key",
)
],
)
service = FacturaCoveDomainService(db=SimpleNamespace(get=lambda *args, **kwargs: company))
monkeypatch.setattr("api.v1.modules.a76.factura_cove.service.object_exists", lambda path: True)
monkeypatch.setattr(
"api.v1.modules.a76.factura_cove.service.get_object_bytes",
lambda path: f"content:{path}".encode("utf-8"),
)
ctx = InvoiceContext(invoice=SimpleNamespace(company_id=10), broker=None, vu=None)
errors = ErrorCollector()
configuracion = service._build_configuracion_vu(ctx, errors)
assert configuracion is not None
assert not errors.has_errors()
assert configuracion.rfc_usuario_vu == "RFCEMPRESA123"
assert configuracion.clave_webservice == "company-ws-pass"
assert configuracion.clave_fiel == "company-fiel-access"
def test_build_configuracion_vu_reports_missing_fiel_when_absent_in_vu_and_company(monkeypatch) -> None:
company = SimpleNamespace(
ventanilla_unica=SimpleNamespace(
webservice_user="company-ws-user",
webservice_password="company-ws-pass",
query_rfc="RFCEMPRESA123",
),
digital_certificates=[],
)
service = FacturaCoveDomainService(db=SimpleNamespace(get=lambda *args, **kwargs: company))
vu = SimpleNamespace(
web_service_user="",
doda_web_service_user=None,
fiel_access_key=None,
certificate_path="cert.cer",
key_path="key.key",
query_tax_id="",
web_service_access_key="",
)
ctx = InvoiceContext(invoice=SimpleNamespace(company_id=10), broker=None, vu=vu)
errors = ErrorCollector()
configuracion = service._build_configuracion_vu(ctx, errors)
assert configuracion is None
assert errors.has_errors()
assert errors.get_errors() == [
{
"field": "vu.clave_fiel",
"message": "La clave FIEL para COVE no está configurada ni en VU ni en la empresa.",
"solution": [
"Captura la clave FIEL en la configuración VU del agente aduanal o en el certificado FIEL de la empresa."
],
"code": "MISSING_FIEL_PASSWORD",
}
]

8
docker-compose.prod.yml
View File

@@ -173,7 +173,9 @@ services:
- CORS_ORIGINS=${CORS_ORIGINS:-https://anexo76-dev.aduanasoft.com,http://localhost:3000}
- SITAR_API_URL=${SITAR_API_URL}
- SITAR_API_USER=${SITAR_API_USER}
- SITAR_API_PASSWORD=${SITAR_API_PASSWORD}
- SITAR_API_PASSWORD=${SITAR_API_PASSWORD}
- COVE_FIEL_HASH_KEY=${COVE_FIEL_HASH_KEY}
- COVE_FIEL_HASH_IV=${COVE_FIEL_HASH_IV}
- VALKEY_URL=${VALKEY_URL:-redis://valkey:6379/0}
- CENTRAL_SERVER_URL=${CENTRAL_SERVER_URL:-""}
- SYNC_SECRET_TOKEN=${SYNC_SECRET_TOKEN:-change-this-sync-token-in-production}
@@ -242,6 +244,8 @@ services:
- SITAR_API_URL=${SITAR_API_URL}
- SITAR_API_USER=${SITAR_API_USER}
- SITAR_API_PASSWORD=${SITAR_API_PASSWORD}
- COVE_FIEL_HASH_KEY=${COVE_FIEL_HASH_KEY}
- COVE_FIEL_HASH_IV=${COVE_FIEL_HASH_IV}
- CSV_IMPORT_STORAGE=${CSV_IMPORT_STORAGE:-minio}
- S3_ENDPOINT_URL=${S3_ENDPOINT_URL:-http://minio:9000}
- S3_ACCESS_KEY=${S3_ACCESS_KEY:-${MINIO_ROOT_USER:-minioadmin}}
@@ -275,6 +279,8 @@ services:
- SITAR_API_URL=${SITAR_API_URL}
- SITAR_API_USER=${SITAR_API_USER}
- SITAR_API_PASSWORD=${SITAR_API_PASSWORD}
- COVE_FIEL_HASH_KEY=${COVE_FIEL_HASH_KEY}
- COVE_FIEL_HASH_IV=${COVE_FIEL_HASH_IV}
- CSV_IMPORT_STORAGE=${CSV_IMPORT_STORAGE:-minio}
- S3_ENDPOINT_URL=${S3_ENDPOINT_URL:-http://minio:9000}
- S3_ACCESS_KEY=${S3_ACCESS_KEY:-${MINIO_ROOT_USER:-minioadmin}}

9
docker-compose.yml
View File

@@ -178,6 +178,8 @@ services:
- SITAR_API_URL=${SITAR_API_URL}
- SITAR_API_USER=${SITAR_API_USER}
- SITAR_API_PASSWORD=${SITAR_API_PASSWORD}
- COVE_FIEL_HASH_KEY=${COVE_FIEL_HASH_KEY}
- COVE_FIEL_HASH_IV=${COVE_FIEL_HASH_IV}
- VALKEY_URL=${VALKEY_URL:-redis://valkey:6379/0}
- CENTRAL_SERVER_URL=${CENTRAL_SERVER_URL:-""}
- SYNC_SECRET_TOKEN=${SYNC_SECRET_TOKEN:-change-this-sync-token-in-production}
@@ -304,6 +306,8 @@ services:
- SITAR_API_URL=${SITAR_API_URL}
- SITAR_API_USER=${SITAR_API_USER}
- SITAR_API_PASSWORD=${SITAR_API_PASSWORD}
- COVE_FIEL_HASH_KEY=${COVE_FIEL_HASH_KEY}
- COVE_FIEL_HASH_IV=${COVE_FIEL_HASH_IV}
- CSV_IMPORT_STORAGE=${CSV_IMPORT_STORAGE:-minio}
- S3_ENDPOINT_URL=${S3_ENDPOINT_URL:-http://minio:9000}
- S3_ACCESS_KEY=${S3_ACCESS_KEY:-${MINIO_ROOT_USER:-minioadmin}}
@@ -336,6 +340,11 @@ services:
- CORE_DB_NAME=${CORE_DB_NAME:-anexo76_core}
- CORE_DB_USER=${CORE_DB_USER:-postgres}
- CORE_DB_PASSWORD=${POSTGRES_APP_PASSWORD:-postgres}
- SITAR_API_URL=${SITAR_API_URL}
- SITAR_API_USER=${SITAR_API_USER}
- SITAR_API_PASSWORD=${SITAR_API_PASSWORD}
- COVE_FIEL_HASH_KEY=${COVE_FIEL_HASH_KEY}
- COVE_FIEL_HASH_IV=${COVE_FIEL_HASH_IV}
- CSV_IMPORT_STORAGE=${CSV_IMPORT_STORAGE:-minio}
- S3_ENDPOINT_URL=${S3_ENDPOINT_URL:-http://minio:9000}
- S3_ACCESS_KEY=${S3_ACCESS_KEY:-${MINIO_ROOT_USER:-minioadmin}}