Merge pull request 'feature/optimizacion-de-permisos' (#460) from feature/optimizacion-de-permisos into development
Reviewed-on: ADUANASOFT/anexo76#460
This commit is contained in:
@@ -20,7 +20,6 @@ from core.s3_keys import company_certificate_key, company_logo_key
|
||||
from core.storage_s3 import delete_object_if_exists, get_object_bytes, put_object_bytes
|
||||
from core.security import (
|
||||
collect_company_ids_from_app_membership,
|
||||
collect_user_role_names,
|
||||
get_current_user,
|
||||
get_tenant_from_token,
|
||||
is_hub_admin,
|
||||
@@ -42,7 +41,13 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _user_is_admin(current_user: dict) -> bool:
|
||||
return "admin" in collect_user_role_names(current_user) or is_hub_admin(current_user)
|
||||
"""
|
||||
Bypass cross-tenant para endpoints sin company_id en la firma. La única
|
||||
fuente válida es hub_admin atestado por el Hub; el antiguo bypass por rol
|
||||
"admin" del realm Keycloak se eliminó para no depender de claims del JWT.
|
||||
Los super_admin locales operan a nivel de compañía (validate_access_to_resource).
|
||||
"""
|
||||
return is_hub_admin(current_user)
|
||||
|
||||
|
||||
def _assert_permission_any_company(
|
||||
@@ -213,12 +218,10 @@ async def get_my_companies(
|
||||
(``user_company_roles``, ``user_company_permissions``, ``user_tenants``).
|
||||
Hub admin: si hay X-Tenant-Override muestra solo ese tenant; si no, muestra todos.
|
||||
"""
|
||||
from core.security import collect_user_role_names
|
||||
user_roles = collect_user_role_names(current_user)
|
||||
|
||||
# Hub admin: respeta X-Tenant-Override para filtrar por tenant seleccionado;
|
||||
# sin override devuelve todo (visibilidad global).
|
||||
if "hub_admin" in user_roles:
|
||||
# sin override devuelve todo (visibilidad global). Se consulta el booleano
|
||||
# normalizado por el Hub (no se lee del JWT) para no acoplar a claims.
|
||||
if is_hub_admin(current_user):
|
||||
service = CompanyService(db)
|
||||
override = request.headers.get("X-Tenant-Override")
|
||||
query = db.query(Company).filter(Company.deleted_at.is_(None))
|
||||
|
||||
@@ -2,7 +2,12 @@ from typing import List, Dict, Any
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||
from sqlalchemy.orm import Session
|
||||
from core.database import get_core_db
|
||||
from core.security import get_current_user, validate_access_to_resource
|
||||
from core.security import (
|
||||
get_current_user,
|
||||
validate_access_to_resource,
|
||||
is_hub_admin,
|
||||
_has_local_super_admin_role,
|
||||
)
|
||||
from api.v1.modules.a76.invoice_settings import services
|
||||
from api.v1.modules.a76.invoice_settings.dto import InvoiceSettingsRequest, InvoiceSettingsResponse, OperationType
|
||||
from api.v1.modules.core.permissions.service import PermissionService
|
||||
@@ -30,10 +35,11 @@ def _can_read_invoice_settings_row(
|
||||
Ver configuración por tipo/op: settings_general.view O ver facturas de ese mismo contexto
|
||||
(para cargar defaults en alta/edición sin abrir la pantalla de parámetros).
|
||||
"""
|
||||
user_roles = current_user.get("realm_access", {}).get("roles", [])
|
||||
if "admin" in user_roles:
|
||||
return True
|
||||
# Bypass de checks: hub_admin (atestado por el Hub) o super_admin local en
|
||||
# la compañía. La autorización NO se decide por claims del JWT.
|
||||
user_id = current_user.get("sub") or current_user.get("id")
|
||||
if is_hub_admin(current_user) or _has_local_super_admin_role(db, user_id, company_id):
|
||||
return True
|
||||
if not user_id:
|
||||
return False
|
||||
ps = PermissionService(db)
|
||||
|
||||
@@ -4,7 +4,13 @@ from typing import Dict, Any, Literal, Optional
|
||||
from core.config import settings
|
||||
from core.database import get_core_db
|
||||
from core.exceptions import BaseAPIException
|
||||
from core.security import collect_user_role_names, get_current_user, validate_access_to_resource, get_active_system
|
||||
from core.security import (
|
||||
_has_local_super_admin_role,
|
||||
get_active_system,
|
||||
get_current_user,
|
||||
is_hub_admin,
|
||||
validate_access_to_resource,
|
||||
)
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, Path, Request
|
||||
from sqlalchemy import func, or_, and_
|
||||
from sqlalchemy.orm import Session
|
||||
@@ -534,12 +540,15 @@ def list_invoices(
|
||||
List invoices with optional filters and granular permission enforcement.
|
||||
"""
|
||||
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
||||
# Roles vienen del Hub (/auth/me), no de realm_access del JWT crudo.
|
||||
is_hub_admin = "admin" in collect_user_role_names(current_user)
|
||||
|
||||
from api.v1.modules.core.permissions.service import PermissionService
|
||||
|
||||
user_id = current_user.get("sub") or current_user.get("id")
|
||||
# Bypass para "ver todo": hub_admin atestado por el Hub o super_admin local
|
||||
# en la compañía. Se evita depender de claims del JWT crudo.
|
||||
is_global_admin = is_hub_admin(current_user) or _has_local_super_admin_role(
|
||||
db, user_id, company_id
|
||||
)
|
||||
perm_service = PermissionService(db)
|
||||
perm_codes = perm_service.get_user_permissions(user_id, company_id)
|
||||
|
||||
@@ -558,9 +567,9 @@ def list_invoices(
|
||||
allowed_filters = build_allowed_types_from_view_permissions(perm_codes)
|
||||
|
||||
if not allowed_filters:
|
||||
if is_hub_admin:
|
||||
if is_global_admin:
|
||||
logger.debug(
|
||||
"[invoices.list] hub admin without invoice view perms — unfiltered types fallback"
|
||||
"[invoices.list] global admin without invoice view perms — unfiltered types fallback"
|
||||
)
|
||||
allowed_filters = None
|
||||
else:
|
||||
|
||||
Reference in New Issue
Block a user