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

feat(auth): add hub_admin global access support across modules

Browse Source 
- Add is_hub_admin() and resolve_tenant_id_required() to core/security
- hub_admin resolves tenant from company or uses None as global sentinel
- Update routes and services to skip tenant filter when tenant_id is None
- UserService accepts is_hub_admin flag for cross-tenant user management
- get_my_companies returns all companies for hub_admin without tenant restriction
This commit is contained in:
Galindo97
2026-05-15 09:22:35 -05:00
parent 0c876b11bf
commit 125405458c
11 changed files with 156 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
backend/api/v1/modules/a76/customs_brokers/routes.py
View File

@@ -17,7 +17,7 @@ from core.s3_keys import (
customs_broker_vu_doda_private_key_key,
customs_broker_vu_private_key_key,
)
from core.security import get_current_user, get_tenant_from_token, validate_access_to_resource
from core.security import get_current_user, get_tenant_from_token, resolve_tenant_id_required, validate_access_to_resource
from core.storage_s3 import delete_object_if_exists, put_object_bytes
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
@@ -32,25 +32,13 @@ MAX_VU_CER_KEY_BYTES = 5 * 1024 * 1024 # 5 MB
MAX_COVE_BYTES = 15 * 1024 * 1024 # 15 MB (xml/zip)
def _resolve_tenant_id_int(current_user: dict) -> int:
tid = get_tenant_from_token(current_user)
if tid is not None:
return int(tid)
raw = current_user.get("tenant_id")
if isinstance(raw, list) and raw:
raw = raw[0]
if raw is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
try:
return int(raw)
except (TypeError, ValueError):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid tenant ID in token",
)
def _resolve_tenant_id_int(
current_user: dict,
db=None,
company_id: int = None,
):
"""Delega a resolve_tenant_id_required. Hub admin resuelve tenant desde la empresa."""
return resolve_tenant_id_required(current_user, db=db, company_id=company_id)
def _remove_stored_vu_path(ref: Optional[str]) -> None:
@@ -198,7 +186,7 @@ async def upload_customs_broker_vu_file(
- DODA: doda_certificate_path, doda_key_path, doda_xml_files_path
"""
validate_access_to_resource(db, company_id, current_user, ["customs_brokers.create"])
tenant_id = _resolve_tenant_id_int(current_user)
tenant_id = _resolve_tenant_id_int(current_user, db=db, company_id=company_id)
broker = services.CustomsBrokerService.get_by_id(db, broker_key, tenant_id, company_id)
if not broker:

10
backend/api/v1/modules/a76/factura_cove/routes.py
View File

@@ -14,7 +14,7 @@ from core.config import settings
from core.database import get_core_db
from core.exceptions import ValidationException
from core.s3_keys import cove_acuse_pdf_key
from core.security import get_current_user, get_tenant_from_token, validate_access_to_resource
from core.security import get_current_user, get_tenant_from_token, resolve_tenant_id_required, validate_access_to_resource
from api.v1.modules.core.tasks_tracking import track_and_dispatch
@@ -147,14 +147,10 @@ def check_cove_eligibility(
Evalúa si la factura tiene todos los datos necesarios (VU, factura, partidas)
para poder generar un COVE. No dispara la tarea Celery.
"""
tenant_id = get_tenant_from_token(current_user)
if not tenant_id:
raise HTTPException(status_code=400, detail="Tenant ID not found in token")
tenant_id_int = int(tenant_id)
tenant_id = resolve_tenant_id_required(current_user, db=db, company_id=company_id)
service = FacturaCoveDomainService(db)
eligibility = service.check_eligibility(invoice_id=invoice_id, tenant_id=tenant_id_int, company_id=company_id)
eligibility = service.check_eligibility(invoice_id=invoice_id, tenant_id=tenant_id, company_id=company_id)
return eligibility

87
backend/api/v1/modules/a76/general_catalogs/company/routes.py
View File

@@ -23,7 +23,9 @@ from core.security import (
collect_user_role_names,
get_current_user,
get_tenant_from_token,
is_hub_admin,
resolve_effective_tenant_id_from_user,
resolve_tenant_id_required,
validate_access_to_resource,
)
from .....common.tenant_crud_routes import TenantCRUDRoutes
@@ -40,7 +42,7 @@ logger = logging.getLogger(__name__)
def _user_is_admin(current_user: dict) -> bool:
return "admin" in collect_user_role_names(current_user)
return "admin" in collect_user_role_names(current_user) or is_hub_admin(current_user)
def _assert_permission_any_company(
@@ -81,26 +83,13 @@ def _assert_permission_for_company(
return validate_access_to_resource(db, company_id, current_user, [permission_code])
def _resolve_tenant_id_int(current_user: dict) -> int:
"""Misma lógica que validate_access_to_resource: entero estable para BD y claves S3."""
tid = get_tenant_from_token(current_user)
if tid is not None:
return int(tid)
raw = current_user.get("tenant_id")
if isinstance(raw, list) and raw:
raw = raw[0]
if raw is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
try:
return int(raw)
except (TypeError, ValueError):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid tenant ID in token",
)
def _resolve_tenant_id_int(
current_user: dict,
db: Session = None,
company_id: int = None,
) -> Optional[int]:
"""Delega a resolve_tenant_id_required. Hub admin resuelve tenant desde la empresa."""
return resolve_tenant_id_required(current_user, db=db, company_id=company_id)
def _is_s3_object_key(ref: Optional[str]) -> bool:
@@ -133,11 +122,11 @@ async def create_company(
):
_assert_permission_any_company(db, current_user, "cat_company.create")
tenant_id = current_user.get("tenant_id")
if not tenant_id:
tenant_id = resolve_tenant_id_required(current_user, db=db)
if tenant_id is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
detail="Hub admin: debe especificar tenant_id para crear una empresa",
)
service = CompanyService(db)
@@ -161,12 +150,7 @@ async def list_companies(
"""Get paginated list of companies for current tenant with optional filters"""
_assert_permission_any_company(db, current_user, "cat_company.view")
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
tenant_id = resolve_tenant_id_required(current_user, db=db)
skip = (page - 1) * page_size
filters = {}
@@ -214,6 +198,24 @@ async def get_my_companies(
Un usuario solo con roles de app y sin ``tenant_id`` en /auth/me sigue pudiendo
listar sus compañías asignadas.
"""
from core.security import collect_user_role_names
user_roles = collect_user_role_names(current_user)
# Hub admin: visibilidad global sobre todas las compañías sin restricciones
# de tenant ni licencia. El Hub ya validó el rol en /auth/me.
if "hub_admin" in user_roles:
service = CompanyService(db)
all_companies = (
db.query(Company)
.filter(Company.deleted_at.is_(None))
.order_by(Company.name)
.all()
)
return [
CompanyResponseDTO.model_validate(service.flatten_company_dto(c))
for c in all_companies
]
_assert_permission_any_company(db, current_user, "cat_company.view")
user_id = current_user.get("sub") or current_user.get("id")
@@ -245,12 +247,7 @@ async def get_company(
"""Get a specific company by ID"""
_assert_permission_for_company(db, company_id, current_user, "cat_company.view")
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
tenant_id = resolve_tenant_id_required(current_user, db=db, company_id=company_id)
service = CompanyService(db)
company = CompanyService.get_by_id(db, company_id, tenant_id, 0)
@@ -277,12 +274,7 @@ async def update_company(
"""Update a company"""
_assert_permission_for_company(db, company_id, current_user, "cat_company.edit")
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
tenant_id = resolve_tenant_id_required(current_user, db=db, company_id=company_id)
service = CompanyService(db)
updated_company = service.update(db, company_id, tenant_id, 0, data)
@@ -354,12 +346,7 @@ async def delete_company(
"""Delete a company"""
_assert_permission_for_company(db, company_id, current_user, "cat_company.delete")
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
tenant_id = resolve_tenant_id_required(current_user, db=db, company_id=company_id)
success = CompanyService.delete(db, company_id, tenant_id, 0)
if not success:
@@ -385,7 +372,7 @@ async def upload_company_logo(
"""Upload a logo for a company"""
_assert_permission_for_company(db, company_id, current_user, "cat_company.edit")
tenant_id = _resolve_tenant_id_int(current_user)
tenant_id = _resolve_tenant_id_int(current_user, db=db, company_id=company_id)
# Validar que la empresa existe
company = CompanyService.get_by_id(db, company_id, tenant_id, 0)
@@ -468,7 +455,7 @@ async def upload_company_certificate(
"""
_assert_permission_for_company(db, company_id, current_user, "cat_company.edit")
tenant_id = _resolve_tenant_id_int(current_user)
tenant_id = _resolve_tenant_id_int(current_user, db=db, company_id=company_id)
# Validar que la empresa existe
service = CompanyService(db)

17
backend/api/v1/modules/a76/general_catalogs/company/service.py
View File

@@ -44,7 +44,9 @@ class CompanyService:
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[Company], int]:
"""Get all companies for a tenant with pagination"""
query = db.query(Company).filter(Company.tenant_id == tenant_id, Company.deleted_at.is_(None))
query = db.query(Company).filter(Company.deleted_at.is_(None))
if tenant_id is not None:
query = query.filter(Company.tenant_id == tenant_id)
# Apply filters if provided
if filters:
@@ -67,15 +69,10 @@ class CompanyService:
db: Session, company_id: int, tenant_id: int, company_id_unused: int
) -> Optional[Company]:
"""Get company by ID"""
return (
db.query(Company)
.filter(
Company.id == company_id,
Company.tenant_id == tenant_id,
Company.deleted_at.is_(None)
)
.first()
)
query = db.query(Company).filter(Company.id == company_id, Company.deleted_at.is_(None))
if tenant_id is not None:
query = query.filter(Company.tenant_id == tenant_id)
return query.first()
# ESTE ES EL MÉTODO VIEJO QUE CAUSABA PROBLEMAS (Lo dejamos por si acaso)
@staticmethod

14
backend/api/v1/modules/a76/reports/movements/invoices/routes.py
View File

@@ -764,20 +764,12 @@ def generate_invoice_report_async(
# Serialize filters to dict for Celery
filter_data = filters.model_dump()
user_email = current_user.get('email')
# Trigger task
tenant_id = get_tenant_from_token(current_user)
if not tenant_id:
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in token",
)
# tenant_id ya validado por validate_access_to_resource
task = track_and_dispatch(
db=db,
task=generate_invoice_movements_async,
tenant_id=int(tenant_id),
tenant_id=tenant_id,
company_id=company_id,
requested_by_user=current_user.get("preferred_username") or current_user.get("email") or current_user.get("sub"),
task_name="generate_invoice_movements_async",