Rakseeibra/arrr-erp
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enforce partner client ownership and partner-scoped imports

Browse Source
This commit is contained in:
A R R R Associates
2026-07-29 06:05:10 +05:30
parent 83f4cd2d4f
commit a58ec7b807
9 changed files with 205 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/clients/api.py
+16 -13
View File
@@ -7,7 +7,7 @@ from sqlalchemy.orm import Session
from app.core.db.deps import get_common_db
from app.core.security.session_auth import require_login
from app.modules.clients.access import ClientAccessScope
from app.modules.clients.access import ClientAccessScope, enforce_partner_scope, is_partner_role
from app.modules.clients.schemas import ClientAuditLogOut, ClientFilterOptions, ClientListResponse, ClientOut, ClientUpdate, ClientCreate
from app.modules.clients.service import (
activate_client_service,
@@ -23,6 +23,7 @@ from app.modules.clients.service import (
update_client_service,
)
from app.modules.core.rbac.permission_guard import require_permission
from app.modules.core.rbac.deps import get_user_roles
router = APIRouter(prefix="/api/v1/clients", tags=["clients-api"])
@@ -34,17 +35,19 @@ def _api_scope_from_user(db, user):
except Exception:
return False
own_only = has("clients.view.own_only") or not has("clients.assign_partner")
return ClientAccessScope(
scope = ClientAccessScope(
tenant_id=user.tenant_id,
branch_id=user.branch_id,
allow_cross_branch=has("clients.cross_branch"),
allow_cross_tenant=has("clients.cross_tenant"),
allow_all_clients=has("clients.view.all") and not own_only,
own_only=own_only,
locked_partner_id=user.id if own_only else None,
can_assign_partner=has("clients.assign_partner"),
can_change_branch=has("clients.cross_branch"),
can_change_tenant=has("clients.cross_tenant"),
)
return enforce_partner_scope(scope, user=user, role_names=get_user_roles(db, user.id))
@router.get("/filters", response_model=ClientFilterOptions)
def api_client_filters():
@@ -73,7 +76,8 @@ def api_list_clients(
tenant_id=scope.tenant_id,
branch_id=scope.branch_id,
allow_cross_branch=scope.allow_cross_branch,
partner_id=partner_id,
partner_id=None if scope.own_only else partner_id,
viewer_partner_id=user.id if scope.own_only else None,
q=q,
status=status,
client_type=client_type,
@@ -105,7 +109,8 @@ def api_export_clients(
tenant_id=scope.tenant_id,
branch_id=scope.branch_id,
allow_cross_branch=scope.allow_cross_branch,
partner_id=partner_id,
partner_id=None if scope.own_only else partner_id,
viewer_partner_id=user.id if scope.own_only else None,
q=q,
status=status,
client_type=client_type,
@@ -122,9 +127,7 @@ def api_export_clients(
def api_get_client(client_id: int, db: Session = Depends(get_common_db), user=Depends(require_login)):
require_permission(db, user, "clients.view")
scope = _api_scope_from_user(db, user)
row = get_client_or_404(db, client_id=client_id, tenant_id=scope.tenant_id, branch_id=scope.branch_id, allow_cross_branch=scope.allow_cross_branch)
if scope.own_only and row.partner_id != scope.locked_partner_id:
raise HTTPException(status_code=404, detail="Client not found.")
row = get_client_or_404(db, client_id=client_id, tenant_id=scope.tenant_id, branch_id=scope.branch_id, allow_cross_branch=scope.allow_cross_branch, allow_all_clients=scope.allow_all_clients, viewer_partner_id=user.id if scope.own_only else None, allow_review_access=True)
return row
@router.post("", response_model=ClientOut, status_code=201)
@@ -137,7 +140,7 @@ def api_create_client(data: ClientCreate, db: Session = Depends(get_common_db),
def api_update_client(client_id: int, data: ClientUpdate, db: Session = Depends(get_common_db), user=Depends(require_login)):
require_permission(db, user, "clients.edit")
scope = _api_scope_from_user(db, user)
row = get_client_or_404(db, client_id=client_id, tenant_id=scope.tenant_id, branch_id=scope.branch_id, allow_cross_branch=scope.allow_cross_branch)
row = get_client_or_404(db, client_id=client_id, tenant_id=scope.tenant_id, branch_id=scope.branch_id, allow_cross_branch=scope.allow_cross_branch, allow_all_clients=scope.allow_all_clients, viewer_partner_id=user.id if scope.own_only else None)
if scope.own_only and row.partner_id != scope.locked_partner_id:
raise HTTPException(status_code=404, detail="Client not found.")
return update_client_service(db, row=row, data=data, actor_user_id=user.id, scope=scope)
@@ -146,33 +149,33 @@ def api_update_client(client_id: int, data: ClientUpdate, db: Session = Depends(
def api_deactivate_client(client_id: int, db: Session = Depends(get_common_db), user=Depends(require_login)):
require_permission(db, user, "clients.deactivate")
scope = _api_scope_from_user(db, user)
row = get_client_or_404(db, client_id=client_id, tenant_id=scope.tenant_id, branch_id=scope.branch_id, allow_cross_branch=scope.allow_cross_branch)
row = get_client_or_404(db, client_id=client_id, tenant_id=scope.tenant_id, branch_id=scope.branch_id, allow_cross_branch=scope.allow_cross_branch, allow_all_clients=scope.allow_all_clients, viewer_partner_id=user.id if scope.own_only else None)
return deactivate_client_service(db, row=row, actor_user_id=user.id)
@router.post("/{client_id}/activate", response_model=ClientOut)
def api_activate_client(client_id: int, db: Session = Depends(get_common_db), user=Depends(require_login)):
require_permission(db, user, "clients.activate")
scope = _api_scope_from_user(db, user)
row = get_client_or_404(db, client_id=client_id, tenant_id=scope.tenant_id, branch_id=scope.branch_id, allow_cross_branch=scope.allow_cross_branch)
row = get_client_or_404(db, client_id=client_id, tenant_id=scope.tenant_id, branch_id=scope.branch_id, allow_cross_branch=scope.allow_cross_branch, allow_all_clients=scope.allow_all_clients, viewer_partner_id=user.id if scope.own_only else None)
return activate_client_service(db, row=row, actor_user_id=user.id)
@router.post("/{client_id}/archive", response_model=ClientOut)
def api_archive_client(client_id: int, db: Session = Depends(get_common_db), user=Depends(require_login)):
require_permission(db, user, "clients.archive")
scope = _api_scope_from_user(db, user)
row = get_client_or_404(db, client_id=client_id, tenant_id=scope.tenant_id, branch_id=scope.branch_id, allow_cross_branch=scope.allow_cross_branch)
row = get_client_or_404(db, client_id=client_id, tenant_id=scope.tenant_id, branch_id=scope.branch_id, allow_cross_branch=scope.allow_cross_branch, allow_all_clients=scope.allow_all_clients, viewer_partner_id=user.id if scope.own_only else None)
return archive_client_service(db, row=row, actor_user_id=user.id)
@router.post("/{client_id}/restore", response_model=ClientOut)
def api_restore_client(client_id: int, db: Session = Depends(get_common_db), user=Depends(require_login)):
require_permission(db, user, "clients.restore")
scope = _api_scope_from_user(db, user)
row = get_client_or_404(db, client_id=client_id, tenant_id=scope.tenant_id, branch_id=scope.branch_id, allow_cross_branch=scope.allow_cross_branch)
row = get_client_or_404(db, client_id=client_id, tenant_id=scope.tenant_id, branch_id=scope.branch_id, allow_cross_branch=scope.allow_cross_branch, allow_all_clients=scope.allow_all_clients, viewer_partner_id=user.id if scope.own_only else None)
return restore_client_service(db, row=row, actor_user_id=user.id)
@router.get("/{client_id}/audit-logs", response_model=list[ClientAuditLogOut])
def api_client_audit_logs(client_id: int, limit: int = Query(50, ge=1, le=200), db: Session = Depends(get_common_db), user=Depends(require_login)):
require_permission(db, user, "clients.audit_log.view")
scope = _api_scope_from_user(db, user)
row = get_client_or_404(db, client_id=client_id, tenant_id=scope.tenant_id, branch_id=scope.branch_id, allow_cross_branch=scope.allow_cross_branch)
row = get_client_or_404(db, client_id=client_id, tenant_id=scope.tenant_id, branch_id=scope.branch_id, allow_cross_branch=scope.allow_cross_branch, allow_all_clients=scope.allow_all_clients, viewer_partner_id=user.id if scope.own_only else None)
return list_client_audit_logs(db, row=row, limit=limit)