"""Region gates for asset writes — the single implementation. Every asset mutation is gated twice: the route-level ``require_asset_admin`` dependency confirms the caller holds an admin role, then one of these narrows it to the caller's region — the set of managed locations they administer (``asset_location_admins``). These live here rather than in the router because the MCP tool surface has to apply exactly the same gates. It used to carry its own copies, and they had drifted: the MCP twin of the target-person gate had lost the explicit-location fallback and had nested the pre-hire bypass under a truthiness check the HTTP version did not have. Two implementations of an authorization rule is one too many — MCP is a first-class surface (``feedback_mcp_sync_rule``), so it imports the same functions the HTTP routes call. ``global_admin`` and ``asset_global_admin`` short-circuit every gate through :func:`app.core.rbac.has_global_asset_scope`. """ from __future__ import annotations from uuid import UUID from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.auth.models import AuthUser from app.core.dependencies import assert_can_manage_asset from app.core.exceptions import ForbiddenError, NotFoundError from app.core.rbac import has_global_asset_scope from app.domains.asset.models import AssetItem async def enforce_asset_region( db: AsyncSession, current_user: AuthUser, asset_id: UUID, ) -> AssetItem: """Load the asset by id and enforce region-scoped write authority. Used at the head of every ``/{asset_id}/*`` mutating endpoint so asset_admin holders can only mutate assets at the locations they administer. global_admin always passes; auditor / management never reach this code because the route-level ``require_asset_admin`` already rejects them. Returns the asset for caller reuse (the service layer also re-fetches with ``FOR UPDATE`` — the extra read here is cheap because the row is still warm in the SQLA identity map). """ asset = await db.get(AssetItem, asset_id) if asset is None or asset.deleted_at is not None: raise NotFoundError("Asset", str(asset_id)) await assert_can_manage_asset(db, current_user, asset) return asset async def assert_target_person_in_region( db: AsyncSession, current_user: AuthUser, target_person_id: UUID, *, target_location: str | None = None, ) -> None: """Reject when the asset's NEW holder sits at a location outside the caller's managed set. Applied to every endpoint that changes ``assigned_to`` (claim, change_assignee, transfer, assign). Without this gate a region-scoped asset_admin could pass the source-region check (asset is at my location) and then re-assign the asset to a Shanghai employee, "laundering" it out of their region without Shanghai-admin involvement. global_admin bypasses. Two further bypasses for the asset_admin path that fall out of operational reality: * **Self-claim**: an asset_admin claiming an asset to themselves is always allowed when they administer at least one location. The admin's personal ``work_location_name`` may differ from any location they administer (traveling / remote IT admins), so a target-resolution check on their own row could 403 the very action they own (claim a Beijing laptop for themselves to bring to the site). * **Onboarding cliff**: a new hire whose HRIS sync hasn't landed yet has ``employment_status='pre_hire'`` (the actual enum value — ``'pending'`` is NOT a valid status and would be rejected by the ``@validates`` hook on ``IdentityPerson.employment_status``). Their ``work_location_name`` / ``city`` are typically empty until the first full HRIS sweep, so strict target-region matching would block IT from claiming their first laptop. Allow when target is ``'pre_hire'`` AND the caller administers at least one location — the audit log still captures the actor, and global_admin is the fallback if pre-hire status is missing. Otherwise the target person's resolved location MUST be one the caller administers. If HR has not populated any location for the person, a location explicitly selected on the dispatch form may establish it. That is deliberately not a bypass for a person with a known, foreign HR location. """ role_names = {r.role.name for r in (current_user.user_roles or [])} if has_global_asset_scope(role_names): return from app.domains.asset.location_admin_sync import ( get_location_ids_managed_by, resolve_person_location_ids_bulk, ) from app.domains.asset.location_resolver import LocationIndex from app.identity.person_models import IdentityPerson managed = await get_location_ids_managed_by(db, current_user.id) if not managed: raise ForbiddenError( "asset_admin without any location cannot assign assets", ) # Self-claim bypass. if target_person_id == current_user.id: return # New-hire onboarding bypass. # # **Invariant** (please preserve): only the HRIS sync writes # ``employment_status='pre_hire'``. If a future admin/self-service # tool gains the ability to write this field, this bypass becomes # a privilege-escalation primitive (an asset_admin could flip an # arbitrary person to pre_hire and then claim any asset to them). # If that lands, gate this bypass on # ``person.created_at > now - INTERVAL 30 DAY`` as well. target = await db.get(IdentityPerson, target_person_id) if target is not None and (target.employment_status or "") == "pre_hire": return index = await LocationIndex.load(db) resolved = await resolve_person_location_ids_bulk( db, [target_person_id], index=index, ) target_location_id = resolved.get(target_person_id) if target_location_id is not None: if target_location_id in managed: return raise ForbiddenError("target user is outside your managed-location region") if target_location: explicit = index.resolve_id(target_location) if explicit is not None and explicit in managed: return raise ForbiddenError("target user is outside your managed-location region") async def assert_target_office_in_region( db: AsyncSession, current_user: AuthUser, target_office_id: UUID, ) -> None: """Reject when the destination office sits at a location outside the caller's managed set. Applied to ``recycle-to-office`` / ``bulk-recycle-to-office`` — the destination ``office_id`` parameter is otherwise unconstrained and an in-region admin could push an asset to a foreign office. The office is an address, not an authority: it is mapped to the managed location it sits in and the check is made there. An office whose city resolves to no managed location is out of region for every regional admin — recycling into it needs the global tier, which is the honest answer when we cannot say whose region it is. """ role_names = {r.role.name for r in (current_user.user_roles or [])} if has_global_asset_scope(role_names): return from app.domains.asset.location_admin_sync import ( get_location_ids_managed_by, ) from app.domains.asset.office_local_admins import get_office_location_id managed = await get_location_ids_managed_by(db, current_user.id) office_location_id = await get_office_location_id(db, target_office_id) if office_location_id is None or office_location_id not in managed: raise ForbiddenError( "target office is outside your managed-location region", ) async def enforce_intake_region( db: AsyncSession, current_user: AuthUser, *, holder_office_id: UUID | None, location: str | None, ) -> None: """Region gate for asset intake (new asset creation). global_admin bypasses. A regional asset_admin must stamp the new asset with a region they administer, and every region signal on the request has to agree with that: * ``location`` resolves through the managed-location master; * ``holder_office_id``, when present, resolves to the location its office sits in. Both are checked when both are given. Checking only the first one that happens to resolve would let an admin pass with their own ``location`` while parking the physical asset at an office in somebody else's region. At least one signal must resolve — an intake with no determinable region is rejected rather than being quietly filed as global. """ role_names = {r.role.name for r in (current_user.user_roles or [])} if has_global_asset_scope(role_names): return from app.domains.asset.location_admin_sync import ( get_location_ids_managed_by, ) from app.domains.asset.location_resolver import LocationIndex from app.domains.asset.office_local_admins import get_office_location_id managed = await get_location_ids_managed_by(db, current_user.id) if not managed: raise ForbiddenError( "asset_admin without any location cannot intake assets", ) index = await LocationIndex.load(db) resolved_any = False if location: location_id = index.resolve_id(location) if location_id is not None: if location_id not in managed: raise ForbiddenError( "intake location is outside your managed-location region", ) resolved_any = True if holder_office_id is not None: office_location_id = await get_office_location_id( db, holder_office_id, index=index, ) if office_location_id is None or office_location_id not in managed: raise ForbiddenError( "intake holder_office_id is outside your managed-location region", ) resolved_any = True if resolved_any: return raise ForbiddenError( "asset_admin must specify a holder_office_id or location within their region", ) async def enforce_asset_region_bulk( db: AsyncSession, current_user: AuthUser, asset_ids: list[UUID], ) -> None: """Bulk variant of :func:`enforce_asset_region`. Loads every asset id and rejects the call if: * ANY id is missing / soft-deleted (mirrors the single-asset gate which raises NotFoundError — no silent pass-through); * ANY asset is outside the caller's managed-location set. For global_admin this is an O(1) check (the helper short-circuits before fetching the managed set). For asset_admin the managed set and every asset's location are resolved in one bulk pass — the per-asset resolver in a loop was an N+1 that grew with the size of the batch it was meant to make cheap. """ role_names = {r.role.name for r in (current_user.user_roles or [])} if has_global_asset_scope(role_names): return from app.domains.asset.location_admin_sync import ( get_location_ids_managed_by, resolve_asset_location_ids_bulk, ) managed = await get_location_ids_managed_by(db, current_user.id) rows = await db.execute( select(AssetItem).where( AssetItem.id.in_(asset_ids), AssetItem.deleted_at.is_(None), ) ) found_assets = list(rows.scalars().all()) found_ids = {a.id for a in found_assets} missing = set(asset_ids) - found_ids if missing: # Match the single-asset gate's NotFoundError surface — also # closes the existence-probe oracle (region error vs not-found # error leaked information about which ids exist). raise NotFoundError( "Asset", ",".join(str(m) for m in sorted(missing)), ) location_ids = await resolve_asset_location_ids_bulk(db, found_assets) for asset in found_assets: location_id = location_ids.get(asset.id) if location_id is None or location_id not in managed: raise ForbiddenError( f"asset {asset.id} is outside your managed-location region", )