feature/fracciones-permisos

backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/routes.py

@@ -9,7 +9,7 @@ from uuid import uuid4
 
 from fastapi import APIRouter, File, HTTPException, Query, UploadFile, Depends
 from sqlalchemy.orm import Session
-from typing import Dict, Any
+from typing import Dict, Any, Optional
 
 from core.celery_app import celery_app
 from core.database import get_core_db
@@ -39,6 +39,50 @@ def _get_redis():
     return redis.Redis.from_url(url, decode_responses=False)
 
 
+def _assert_us_tariff_csv_job_access(
+    db: Session,
+    job_id: str,
+    current_user: Dict[str, Any],
+) -> None:
+    from api.v1.modules.core.tasks_tracking.models import TaskRun
+
+    company_id: Optional[int] = None
+    row = db.query(TaskRun).filter(TaskRun.task_id == job_id).first()
+    if row is not None and row.company_id is not None:
+        company_id = int(row.company_id)
+
+    r = _get_redis()
+    if company_id is None:
+        raw = r.get(f"{FA_IMPORT_META_PREFIX}{job_id}")
+        if raw:
+            meta = json.loads(raw.decode("utf-8"))
+            cid = meta.get("company_id")
+            if cid is not None:
+                company_id = int(cid)
+
+    if company_id is None and row is not None and row.meta_payload:
+        layout_jid = row.meta_payload.get("layout_import_job_id")
+        if layout_jid:
+            raw2 = r.get(f"{FA_IMPORT_META_PREFIX}{layout_jid}")
+            if raw2:
+                meta2 = json.loads(raw2.decode("utf-8"))
+                cid2 = meta2.get("company_id")
+                if cid2 is not None:
+                    company_id = int(cid2)
+
+    if company_id is None:
+        raise HTTPException(status_code=403, detail="Sin acceso a este job")
+    try:
+        validate_access_to_resource(
+            db, company_id, current_user, ["csv_upload.process"]
+        )
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("US tariff fractions import: job access validation failed: %s", e)
+        raise HTTPException(status_code=403, detail="Sin acceso a este job") from None
+
+
 @router.post("/upload", response_model=ImportJobResponse)
 async def upload_import_file(
     file: UploadFile = File(...),
@@ -112,10 +156,15 @@ async def upload_import_file(
 
 
 @router.get("/{job_id}/status")
-async def get_import_status(job_id: str):
+async def get_import_status(
+    job_id: str,
+    db: Session = Depends(get_core_db),
+    current_user: Dict[str, Any] = Depends(get_current_user),
+):
     """
     Polling: estado del escaneo o del commit.
     """
+    _assert_us_tariff_csv_job_access(db, job_id, current_user)
     task_result = celery_app.AsyncResult(job_id)
 
     if task_result.state == "PENDING":
@@ -179,6 +228,7 @@ async def commit_import_job(
         task_name="us_tariff_fractions_insert_valid_rows",
         task_origin="a76/layouts_csv/us_tariff_fractions/commit",
         args=[job_id],
+        required_permissions=["csv_upload.process"],
     )
     return {
         "status": "committing",
@@ -188,5 +238,10 @@ async def commit_import_job(
 
 
 @router.get("/{job_id}/errors/scan-csv")
-async def download_scan_errors_csv(job_id: str):
+async def download_scan_errors_csv(
+    job_id: str,
+    db: Session = Depends(get_core_db),
+    current_user: Dict[str, Any] = Depends(get_current_user),
+):
+    _assert_us_tariff_csv_job_access(db, job_id, current_user)
     return download_scan_errors_csv_stream("fa", job_id)