fix(crm): restringe /uploads/url y /uploads/download a documentos del CRM
tests/test_uploads_alcance.py llegó en el rebase del carril EFC (5c4df59) sin el código
de producción que lo acompañaba: venía de la rama del expediente paralelo que se
descartó. El módulo no importaba —`validar_extension` no existe— así que la suite
nunca corrió y el hueco quedó invisible.
El hueco: ambos endpoints solo comprobaban que la key empezara con
tenants/{tid}/companies/{cid}/. Con el permiso de módulo crm.access eso alcanzaba para
firmar o descargar CUALQUIER objeto de la company, incluidos certificates/*.key — la
llave privada del CSD con la que se sellan los CFDI.
- _validar_alcance: además del aislamiento por tenant/company, la key tiene que ser un
documento del CRM (crm-docs/ o expedientes/{n}/documents/). Se aplica a los dos
endpoints: el que entrega bytes no puede ser más laxo que el que firma una URL.
- validar_extension: allowlist de extensiones en la subida, no lista de vetados.
Verificado que el frontend solo pasa file_key de documentos a estos endpoints
(uploads.ts, sus dos únicos llamadores). Se agregan 5 casos para /uploads/download,
que la prueba original no cubría.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user