diff --git a/backend/api/v1/modules/crm/uploads/routes.py b/backend/api/v1/modules/crm/uploads/routes.py index 7968a27..e67a2a9 100644 --- a/backend/api/v1/modules/crm/uploads/routes.py +++ b/backend/api/v1/modules/crm/uploads/routes.py @@ -17,6 +17,26 @@ router = APIRouter() MAX_UPLOAD_BYTES = 25 * 1024 * 1024 # 25 MB _SAFE_NAME = re.compile(r"[^A-Za-z0-9._-]+") +# Extensiones que el CRM acepta subir. Es una ALLOWLIST y no una lista de vetados: lo segundo +# deja pasar todo lo que nadie pensó en prohibir. +EXTENSIONES_PERMITIDAS = frozenset({ + # Documentos + "pdf", "xml", "csv", "txt", "doc", "docx", "xls", "xlsx", "ppt", "pptx", "odt", "ods", + # Imágenes (fotos de maniobras, sellos, evidencias) + "jpg", "jpeg", "png", "gif", "webp", "bmp", "tif", "tiff", + # Paquetes (juegos de documentos de un embarque) + "zip", "rar", "7z", + # Correo, que en comercio exterior se archiva como evidencia + "msg", "eml", +}) + +# Prefijos —ya dentro de ``tenants/{tid}/companies/{cid}/``— que son documentos del CRM. El +# alcance de estos endpoints es «los archivos que el CRM subió», NO todo el almacén de la +# company: ahí conviven los certificados de la FIEL y del CSD, los CFDI, los CSV de importación +# y el branding, que nada tienen que ver con el permiso de módulo ``crm.access``. +_PREFIJOS_DOCUMENTOS = ("crm-docs/",) +_PATRON_DOCUMENTOS = re.compile(r"^expedientes/\d+/documents/") + def _safe_filename(name: str | None) -> str: base = (name or "archivo").strip().replace(" ", "_") @@ -24,6 +44,41 @@ def _safe_filename(name: str | None) -> str: return base[:120] +def validar_extension(name: str | None) -> None: + """Rechaza lo que no esté en la allowlist. Un archivo sin extensión tampoco pasa. + + Se valida el NOMBRE y no el ``content_type``: el segundo lo pone el navegador y quien sube + el archivo lo controla, así que no es una comprobación. + """ + _, punto, extension = (name or "").rpartition(".") + if not punto or extension.lower() not in EXTENSIONES_PERMITIDAS: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="Ese tipo de archivo no está permitido.", + ) + + +def _validar_alcance(key: str, tenant_id: int, company_id: int) -> None: + """Comprueba que la key sea un documento del CRM de ESTE tenant y company. + + Son dos guardas y la segunda no reemplaza a la primera. El aislamiento por + tenant/company evita leer el almacén de otro cliente; el alcance por prefijo evita que el + permiso de módulo del CRM sirva para firmar un objeto que pertenece a otro módulo. + """ + prefijo = f"tenants/{tenant_id}/companies/{company_id}/" + if not key.startswith(prefijo): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, detail="Archivo fuera de tu alcance" + ) + + relativa = key[len(prefijo):] + if relativa.startswith(_PREFIJOS_DOCUMENTOS) or _PATRON_DOCUMENTOS.match(relativa): + return + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, detail="Archivo fuera de tu alcance" + ) + + @router.post("/uploads") async def upload_file( file: UploadFile = File(...), @@ -31,6 +86,7 @@ async def upload_file( current_user: dict = Depends(get_current_user), ): tenant_id = current_user["tenant_id"] + validar_extension(file.filename) content = await file.read() if len(content) > MAX_UPLOAD_BYTES: raise HTTPException( @@ -55,11 +111,7 @@ def get_upload_url( company_id: int = Query(..., description="Company ID"), current_user: dict = Depends(get_current_user), ): - tenant_id = current_user["tenant_id"] - # Un archivo solo puede consultarse dentro de su propio tenant/company (aislamiento). - prefix = f"tenants/{tenant_id}/companies/{company_id}/" - if not key.startswith(prefix): - raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Archivo fuera de tu alcance") + _validar_alcance(key, current_user["tenant_id"], company_id) return {"url": presigned_get_url(key)} @@ -72,11 +124,12 @@ def download_file( """Transmite el archivo por el backend (sin exponer MinIO al navegador). Evita el bug de la URL prefirmada que apunta al host interno ``minio:9000``. + + Mismo alcance que ``/uploads/url``, y por la misma razón: este endpoint entrega los BYTES, + así que dejarlo más abierto que el que solo firma una URL sería la puerta grande al lado de + la que se acaba de cerrar. """ - tenant_id = current_user["tenant_id"] - prefix = f"tenants/{tenant_id}/companies/{company_id}/" - if not key.startswith(prefix): - raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Archivo fuera de tu alcance") + _validar_alcance(key, current_user["tenant_id"], company_id) try: data = get_object_bytes(key) except Exception: diff --git a/backend/tests/test_uploads_alcance.py b/backend/tests/test_uploads_alcance.py index 915d0fe..f6a544d 100644 --- a/backend/tests/test_uploads_alcance.py +++ b/backend/tests/test_uploads_alcance.py @@ -10,7 +10,7 @@ almacén de la company». import pytest from fastapi import HTTPException -from api.v1.modules.crm.uploads.routes import get_upload_url, validar_extension +from api.v1.modules.crm.uploads.routes import download_file, get_upload_url, validar_extension from tests.conftest import COMPANY_ID, TENANT_ID USUARIO = {"tenant_id": TENANT_ID, "sub": "user-1"} @@ -83,3 +83,40 @@ def test_la_allowlist_de_extensiones_rechaza_lo_ejecutable(): def test_la_allowlist_acepta_los_formatos_de_documento(): for nombre in ("guia.pdf", "factura.XML", "foto.JPG", "hoja.xlsx", "carta.docx", "paquete.zip"): validar_extension(nombre) # no lanza + + +# ── /uploads/download: el mismo alcance ────────────────────────────────────── +# +# El endpoint que entrega los BYTES no puede ser más permisivo que el que solo firma una URL: +# sería la puerta grande justo al lado de la que se cerró. + +@pytest.fixture() +def _sin_descarga(monkeypatch): + import api.v1.modules.crm.uploads.routes as uploads + + monkeypatch.setattr(uploads, "get_object_bytes", lambda key: b"contenido") + + +@pytest.mark.parametrize( + "sufijo", + [ + "certificates/fiel_20260101.key", + "invoices/9/cove/cove.xml", + "branding/logo.png", + ], +) +def test_la_descarga_tampoco_alcanza_fuera_de_los_documentos(sufijo, _sin_descarga): + with pytest.raises(HTTPException) as exc: + download_file(PREFIJO + sufijo, COMPANY_ID, USUARIO) + assert exc.value.status_code == 403 + + +def test_la_descarga_si_entrega_un_documento_del_crm(_sin_descarga): + resp = download_file(PREFIJO + "crm-docs/abc123/contrato.pdf", COMPANY_ID, USUARIO) + assert resp.body == b"contenido" + + +def test_la_descarga_no_alcanza_otro_tenant(_sin_descarga): + with pytest.raises(HTTPException) as exc: + download_file("tenants/999/companies/1/crm-docs/a/b.pdf", COMPANY_ID, USUARIO) + assert exc.value.status_code == 403