from collections.abc import Iterable GLOBAL_ADMIN_ROLE = "global_admin" EMPLOYEE_ROLE = "employee" AUDITOR_ROLE = "auditor" ASSET_ADMIN_ROLE = "asset_admin" # 2026-06-11: ``asset_global_admin`` (全局资产管理员) — manually granted # (RBAC page), unlike ``asset_admin`` which is auto-derived from # ``asset_location_admins``. Semantics: asset-domain authority over # EVERY location, like global_admin but scoped to the asset domain. # # Two invariants keep the blast radius small: # # 1. Holders ALWAYS co-hold ``asset_admin`` — the derivation in # ``location_admin_sync._person_should_hold_role`` (and the daily # reconciler) treats "holds asset_global_admin" as a location-less # grant source. Every existing entry gate (``require_asset_admin``, # frontend route guards, menu checks) therefore works unchanged. # 2. The ONLY new authority is the region bypass: every location-scope # gate short-circuits via :func:`has_global_asset_scope` instead of # checking ``global_admin`` alone. # # It does NOT grant anything outside the asset domain (no HR / cloud / # integrations / user governance). ASSET_GLOBAL_ADMIN_ROLE = "asset_global_admin" HR_ROLE = "hr" FINANCE_ROLE = "finance" # 2026-04-29: ``cloud_admin`` REVIVED with a NEW semantic. # # The role was deleted in migration 040 (collapsed into global_admin # when integrations got unified). User direction 2026-04-29 brought # it back with a different meaning: not a "cloud-only IT admin" but # a marker for SREs/DevOps whose cloud grants must NEVER expire — # real organizations have a small population of permanent cloud # operators and the 90-day expiry sweep is operationally hostile to # them (they get paged, they renew, they get paged again). A # cloud_admin's grants are issued with ``expires_at = NULL`` and # the expiry sweeper skips them. Since the 2026-08-02 policy update, # global_admin grants share that permanence through the lifecycle # service's broader ``load_perpetual_person_ids`` predicate; this # constant/helper remains the identity of the specialist role only. # # It is NOT a substitute for global_admin: cloud_admin alone does # NOT grant permission to read all assets / hr / finance / # integrations. To get full IT admin authority a person still needs # global_admin. Migration 080 re-creates the role row. CLOUD_ADMIN_ROLE = "cloud_admin" # 2026-05-02: ``management`` (管理层) — read-only viewer for company # leadership / executives. Holders see the same admin surface as # ``global_admin`` (assets, HR, cloud, audit, finance, security, # insights) but cannot mutate anything: every ``can_manage_*`` / # write helper deliberately excludes this role. They also do NOT # appear in any admin-broadcast notification fan-out (the 7 sites # that hard-code ``("global_admin", "it_admin")`` in scope-request, # lifecycle-service, provisioning, offboarding, hr-digest, and # leaver-blocked SLA), so management holders aren't paged by routine # operational alerts. Personal notifications targeted at them # (their own offboarding, their own grants) still arrive. MANAGEMENT_ROLE = "management" # 2026-08-25: ``security_officer`` (安全员) — the people who actually run # the security operations desk: they triage externally reported # vulnerabilities, set severity, decide bounty amounts and reply to # researchers. That is WRITE work, so this is not a read-only viewer role. # # Scope: the security domain, and the WHOLE of it. Holders pass # ``can_view_security`` and ``can_manage_security``, which is what gates # the 安全合规 menu group — the bounty console, the findings worklist, the # compliance controls, the evidence surfaces and the access-certification # item decisions (``compliance_router._require_manage`` is # ``can_manage_security``). Inside this domain a security officer has the # same authority as an auditor; that is what "读写安全合规" means and there # is no narrower carve-out pretending otherwise. # # What it deliberately does NOT grant: # # * ``global_admin`` — no users / roles / integrations / HR / finance / # scheduled jobs / asset or cloud administration, # * membership of any admin-broadcast notification fan-out that # hard-codes ``global_admin``: a new security seat should not silently # join every operational page-out. # # Distinct from ``auditor`` as an identity rather than as a permission # set: an auditor audits, a security officer operates, and the two appear # separately in the role switcher, the menu and the audit trail. Anyone # who needs the difference to be an authority boundary too has to add a # narrower gate — widening ``can_manage_security`` was the explicit # decision (2026-08-25), not an accident. SECURITY_OFFICER_ROLE = "security_officer" # 2026-08-29: ``github_admin`` (GitHub 组织管理员) —— 那 5 个 GitHub 组织 # 的 owner 本人。他们不是 IT,是研发侧的人:决定谁该进哪个组织、哪些 # 部门天然就该在里面、以及一个离职的人是不是仍然是社区里合法的 # maintainer。以前这些判断只存在于飞书审批流的「流程管理员」字段和几个 # 人的记忆里,Helix 接管执行之后,配置权必须跟着落到同一批人手上 —— # 否则每加一条自动加入规则都要走 IT,自动化就退化成了换一个人工窗口。 # # 作用域:**只有 GitHub 组织域**,而且是这个域的全部(组织登记、自动 # 加入规则、成员台账、手工增删、对账与漂移处置)。域内它与 global_admin # 等权,域外一无所有 —— 不给 HR / 财务 / 资产 / 云 / 用户与角色治理, # 也不加入任何硬编码 ``global_admin`` 的管理员广播扇出:一个研发侧的 # 组织 owner 不该因为拿了这个角色就开始收所有运维告警。 # # 与 ``security_officer`` 同款设计,理由也同款:域内操作权,不是只读位。 GITHUB_ADMIN_ROLE = "github_admin" LEGACY_ROLE_ALIASES = { "it_admin": GLOBAL_ADMIN_ROLE, "security_auditor": AUDITOR_ROLE, } CANONICAL_ROLE_NAMES = ( GLOBAL_ADMIN_ROLE, EMPLOYEE_ROLE, AUDITOR_ROLE, ASSET_ADMIN_ROLE, ASSET_GLOBAL_ADMIN_ROLE, HR_ROLE, FINANCE_ROLE, CLOUD_ADMIN_ROLE, MANAGEMENT_ROLE, SECURITY_OFFICER_ROLE, GITHUB_ADMIN_ROLE, ) # Privileged platform roles a leaver MUST NOT keep. When a person is # offboarded (``employment_status='left'``) these manually-granted # elevated grants are revoked so a departed account can never retain # admin/auditor/finance/cloud authority. Two canonical roles are # deliberately EXCLUDED: # * ``employee`` — the base self-scope role everyone holds; it confers # no authority over others and is harmless on a disabled account. # * ``asset_admin`` — NOT a manual grant; it is reconciler-derived from # ``asset_location_admins`` (location_admin_sync) and from # ``asset_global_admin`` co-derivation. Deleting it here would just # be re-added by the daily reconciler while the location-admin row # stands, so cleanup of that role belongs to the location-admin side. # Note ``asset_global_admin`` (a manual grant) IS in this set, and # revoking it co-derives-drops ``asset_admin`` via # ``AuthService._sync_asset_admin_co_derivation``. LEAVER_REVOCABLE_ROLES = frozenset(CANONICAL_ROLE_NAMES) - { EMPLOYEE_ROLE, ASSET_ADMIN_ROLE, } ROLE_EQUIVALENTS = { GLOBAL_ADMIN_ROLE: {GLOBAL_ADMIN_ROLE, "it_admin"}, EMPLOYEE_ROLE: {EMPLOYEE_ROLE}, AUDITOR_ROLE: {AUDITOR_ROLE, "security_auditor"}, ASSET_ADMIN_ROLE: {ASSET_ADMIN_ROLE}, ASSET_GLOBAL_ADMIN_ROLE: {ASSET_GLOBAL_ADMIN_ROLE}, HR_ROLE: {HR_ROLE}, FINANCE_ROLE: {FINANCE_ROLE}, CLOUD_ADMIN_ROLE: {CLOUD_ADMIN_ROLE}, MANAGEMENT_ROLE: {MANAGEMENT_ROLE}, SECURITY_OFFICER_ROLE: {SECURITY_OFFICER_ROLE}, GITHUB_ADMIN_ROLE: {GITHUB_ADMIN_ROLE}, } def canonicalize_role_name(role_name: str) -> str: return LEGACY_ROLE_ALIASES.get(role_name, role_name) def normalize_role_names(role_names: Iterable[str]) -> list[str]: seen: set[str] = set() normalized: list[str] = [] for role_name in role_names: canonical = canonicalize_role_name(role_name) if canonical in seen: continue seen.add(canonical) normalized.append(canonical) return normalized def db_role_names_for(role_name: str) -> set[str]: return ROLE_EQUIVALENTS.get(canonicalize_role_name(role_name), {canonicalize_role_name(role_name)}) def has_role(role_names: Iterable[str], *required_roles: str) -> bool: normalized = set(normalize_role_names(role_names)) return bool(normalized.intersection(canonicalize_role_name(role) for role in required_roles)) # View-role narrowing — the request-time self-downgrade mechanism that # backs the header role-switcher chip in the web UI. # # Frontend sends ``X-View-Role: `` to declare "treat this request # as if I only hold this role". The header is honoured only when it # describes a STRICT NARROWING of the user's real role_names: # # * ``leader`` is the UI synonym for "employee + manager-scope". The # manager-scope (subordinate-id expansion) is org-tree-driven and # already kicks in for any non-admin caller with subordinates, so # ``leader`` narrows to ``[employee]`` and lets the existing # subordinate-expansion logic do the rest. # * Any other value must already appear in the user's canonical # ``role_names``. A user with only ``employee`` cannot escalate by # sending ``X-View-Role: global_admin`` — the header is silently # ignored when invalid. # # Every active person carries the ``employee`` role (see # ``HrService._grant_default_employee_role``), so a global_admin viewing # the platform "as employee" is a real narrowing of an existing role, # not a forged downgrade. This matters: the narrowed view should behave # identically to a real employee — the same scope filtering, the same # 403s on admin endpoints — so admins testing the employee experience # actually exercise the production code paths the employees hit. # # The override is request-scoped and lives on a private attribute of the # request's ``IdentityPerson`` instance. ``role_names`` reads it before # falling back to the relationship walk; everything downstream # (``can_*`` / ``has_role`` / ``require_*``) sees the narrowed list # transparently. No call-site changes were required at 325 backend # RBAC call sites — the seam is the property. _VIEW_ROLE_UI_SYNONYMS = {"leader"} def apply_view_role_narrow(user, requested: str | None) -> None: """Narrow ``user.role_names`` to ``requested`` when the request is a valid self-downgrade. Silently no-op otherwise. The function mutates the in-memory ORM instance (``user.__dict__``) — it does NOT touch the DB, never widens the role set, and never raises on invalid input. Frontend may set the header speculatively on every request; defense-in-depth lives here. """ if not requested: return requested = requested.strip() if not requested: return canonical = canonicalize_role_name(requested) if canonical not in CANONICAL_ROLE_NAMES and requested not in _VIEW_ROLE_UI_SYNONYMS: return # We walk the eager-loaded relationship directly (not ``user.role_names``) # because the property would respect any override already on the # instance — circular logic at first narrowing on a request that # somehow re-enters this helper. actual = normalize_role_names( ur.role.name for ur in user.user_roles ) # Two carve-outs collapse to the same narrowed shape ``[EMPLOYEE_ROLE]``: # # * ``leader`` is a UI synonym for "employee with manager-scope". The # org-tree subordinate-id expansion in routers kicks in for any # non-admin caller, so we don't need a distinct backend role. # * ``employee`` is the universal self-downgrade. Every active person # synced from HRIS is granted ``employee`` via # ``HrService._grant_default_employee_role``; second-pass security # scan flagged the edge case where a manual / contractor person # authenticates without the grant (e.g. it was revoked). For those # people we refuse to grant ``employee`` via narrowing — narrowing # can only ever NARROW the real role set, never WIDEN. if requested in _VIEW_ROLE_UI_SYNONYMS or canonical == EMPLOYEE_ROLE: if EMPLOYEE_ROLE not in actual: return narrowed = [EMPLOYEE_ROLE] else: # The user must actually hold this canonical role. if canonical not in actual: return narrowed = [canonical] # ``asset_global_admin`` is a scope-widening modifier of the # asset-admin view, not a separate view: holders co-hold # ``asset_admin`` (auto-derived) and the frontend surfaces # exactly one "资产管理员" chip for them. Narrowing to that # chip must NOT strip the global scope — otherwise the # office-region gates (which see no office rows for a global # holder) would lock them out of the very view that is their # daily driver. Both directions collapse to the same pair so # role_names-based surfaces stay coherent either way. Still a # strict narrowing: nothing outside the asset domain survives. if ( canonical in (ASSET_ADMIN_ROLE, ASSET_GLOBAL_ADMIN_ROLE) and ASSET_GLOBAL_ADMIN_ROLE in actual ): narrowed = [ r for r in (ASSET_ADMIN_ROLE, ASSET_GLOBAL_ADMIN_ROLE) if r in actual ] user.__dict__["_role_names_override"] = narrowed def can_manage_assets(role_names: Iterable[str]) -> bool: return has_role(role_names, GLOBAL_ADMIN_ROLE, ASSET_ADMIN_ROLE) def has_global_asset_scope(role_names: Iterable[str]) -> bool: """Location-region bypass for asset write gates. True for ``global_admin`` (full platform authority) and ``asset_global_admin`` (asset-domain-wide authority). Every location-scope gate (``person_can_manage_asset``, the ``_enforce_*_region`` helpers in the asset router and their MCP twins) short-circuits through this instead of hand-rolling the ``"global_admin" in role_names or "it_admin" in role_names`` pattern — ``has_role`` canonicalizes the legacy alias. NOT a substitute for the entry gate: callers still pass ``require_asset_admin`` first (asset_global_admin co-holds ``asset_admin`` via the location-admin derivation, so that gate admits them unchanged). """ return has_role(role_names, GLOBAL_ADMIN_ROLE, ASSET_GLOBAL_ADMIN_ROLE) def can_read_all_assets(role_names: Iterable[str]) -> bool: # ``finance`` reads the full hardware fleet (2026-06-13): it already # sees software-seat / cloud / AI / procurement cost in depth, and # the hardware financial lifecycle (book value, depreciation, # disposal P&L) closes the cost-of-ownership gap. Read-only — asset # mutations stay on ``can_manage_assets`` (global_admin / asset_admin). return has_role( role_names, GLOBAL_ADMIN_ROLE, ASSET_ADMIN_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE, FINANCE_ROLE, ) def can_access_audit(role_names: Iterable[str]) -> bool: return has_role( role_names, GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE, ) def can_manage_cloud(role_names: Iterable[str]) -> bool: """Mutate cloud integrations / scopes / IAM bindings. 2026-05-20: ``cloud_admin`` was promoted from "perpetual grant marker" to a real write authority over the cloud domain. User direction: "云管理员能看到云相关的东西" (cloud admins should actually be able to operate cloud resources, not just be marked as long-lived grantees). Concretely this allows cloud_admin to: * mutate cloud servers / load-balancers / object storage / IAM users / domains / DNS / CDN annotations * trigger sync jobs and capability runs * manage cloud scopes (members, ownership, alerts) — short of irreversible decommission, which still goes through ``can_manage_cloud_scope_lifecycle`` (global_admin only) * write to Bailian admin endpoints (AliCloud LLM gateway) What cloud_admin still does NOT get: * close / decommission a cloud sub-account / scope (``can_manage_cloud_scope_lifecycle``) * edit raw integration credentials / connector secrets (``/admin/integrations`` is route-gated to global_admin) * mutate HR / asset / finance / approval data outside the cloud domain The perpetual-grant-marker semantic is preserved via ``is_cloud_admin`` — that helper still drives the ``expires_at = NULL`` carve-out in the lifecycle service and the expiry sweeper. The two roles (cloud_admin grant marker vs. cloud_admin authority) collapse into one because in practice the same population — SREs / DevOps — needs both long-lived access AND the ability to manage the cloud resources they hold grants on. """ return has_role(role_names, GLOBAL_ADMIN_ROLE, CLOUD_ADMIN_ROLE) def can_manage_cloud_scope_lifecycle(role_names: Iterable[str]) -> bool: """Decommission a cloud scope (irreversible) / scan idle candidates. Restricted to ``global_admin`` only. ``cloud_admin`` is documented in CLAUDE.md and migration 080 as a "perpetual grant marker for SREs/DevOps whose access must never expire" — explicitly NOT a substitute for global_admin. Account-close API calls are irreversible (the cloud-side close cannot be reactivated); the four-eyes posture is "only the IT admin team can fire this". Idle-candidate scans are kept admin-only too — the candidate list is a sensitive forecast of "scopes I'd close next week" that should not be visible to perpetual grant-holders by default. """ return has_role(role_names, GLOBAL_ADMIN_ROLE) def can_read_cloud(role_names: Iterable[str]) -> bool: """Read access to the admin-tier cloud surface (fleet aggregates, LB / OSS / IAM / server lists, billing exports). Matches /help permission matrix row 4 (云资产) + row 6 (AI 花费 / 云账单). Finance is admitted because the same finance team that sees billing also needs the underlying resource catalogue to do cost attribution — they cannot reconcile a $X charge against rows they cannot list. Pure network metadata (DNS / domain / CDN — matrix row 5) is NOT routed through this helper; those endpoints stay on ``can_view_cloud_resources`` (open to authenticated employees, finance therefore admitted incidentally as an employee). """ return has_role( role_names, GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, CLOUD_ADMIN_ROLE, FINANCE_ROLE, MANAGEMENT_ROLE, ) def is_cloud_admin(role_names: Iterable[str]) -> bool: """Person holds the ``cloud_admin`` system role. This is a role-identity predicate and therefore does not fold in ``global_admin``. Cloud-grant permanence is broader as of 2026-08-02: the lifecycle service's ``load_perpetual_person_ids`` admits both cloud and global admins. Callers deciding expiry semantics must use that lifecycle predicate rather than this role-specific helper. """ return has_role(role_names, CLOUD_ADMIN_ROLE) def can_view_cloud_resources(role_names: Iterable[str]) -> bool: """Baseline read gate for the cloud surface (servers / domains / scopes / DNS / CDN / LBs / per-id GETs). ``cloud_admin`` is admitted here because the view-role chip strips every other role when narrowing to ``[cloud_admin]`` — a cloud admin browsing the platform "as cloud admin" must not lose read access to the very resources they manage. Without this, 30+ /cmdb/* endpoints 403 in cloud-admin view and the ``/admin/cloud/scopes`` page surfaces "loadFailed". ``finance`` is admitted for the SAME narrowed-view reason (2026-06-03): finance already reads the admin-tier cloud surface via ``can_read_cloud`` (cost attribution needs the resource catalogue), and day-to-day passes this gate incidentally through the ``employee`` bit. But when a finance user clicks the「财务」view chip, ``role_names`` narrows to ``[finance]`` — dropping ``employee`` — and the scope list/search (`GET /cmdb/scopes`) 403'd, so finance could not look up which cloud sub-account/subscription a given owner holds. Admitting ``finance`` here closes that gap and makes the existing finance scope-detail bypass (`get_scope`, 2026-05-16) reachable in finance view. The membership filter in ``list_scopes`` is keyed off ``can_read_cloud`` (which already includes finance), so finance sees the full org-wide list, not a self-scoped subset. Account management and sync remain restricted via ``can_manage_cloud`` / ``can_manage_cloud_scope_lifecycle`` — this helper governs READ-ONLY visibility only. """ return has_role( role_names, GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, EMPLOYEE_ROLE, MANAGEMENT_ROLE, CLOUD_ADMIN_ROLE, FINANCE_ROLE, ) def can_view_finance(role_names: Iterable[str]) -> bool: """Finance role and global admin can view billing/cost-related data. Finance is read-only; monetary mutations remain with global admin. """ return has_role( role_names, GLOBAL_ADMIN_ROLE, FINANCE_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE, ) # 实付（invoice / post-discount）口径的保密圈。Lives in core, not in a # domain, because it is a PLATFORM boundary: the cmdb cost surfaces, the # finance ledger reads and ``app.core.masking`` all have to agree on # exactly one definition of "who may see a post-discount cloud figure". # ``app.domains.cmdb.cost_view`` re-exports it under the same name for # its existing callers. # # Deliberately NOT ``can_view_finance``: that predicate admits auditor + # management, who sit outside the circle. Handing them an invoice total # alongside the list total the same gate already gives them # (``/cost/breakdown``, ``/cmdb/billing/monthly``) discloses the company # discount rate by division — DECISIONS 2026-07-29 #1. _INVOICE_MEASURE_ROLES = (FINANCE_ROLE, CLOUD_ADMIN_ROLE, GLOBAL_ADMIN_ROLE) def can_view_invoice_measure(role_names: Iterable[str]) -> bool: """实付口径（及财务瀑布 / 折扣桶金额）的可见性闸。""" return has_role(role_names, *_INVOICE_MEASURE_ROLES) def can_manage_finance(role_names: Iterable[str]) -> bool: """Only global admin can mutate finance-related rows (suppliers, procurement orders, business payments, legal entities). """ return has_role(role_names, GLOBAL_ADMIN_ROLE) def can_operate_finance(role_names: Iterable[str]) -> bool: """Finance-controller operations: period close/reopen and governance finding triage (acknowledge / snooze / resolve). These are the finance team's OWN workflow actions — closing a month is a controller decision, and triaging cost-governance findings is the finance analyst's day job — so the ``finance`` role gets them in addition to global admin (2026-06-18, DECISIONS.md). This is distinct from ``can_manage_finance`` (global-admin-only), which still gates the master-data / irreversible mutations (FX pins, rollup refresh, budget CRUD, supplier/PO/payment edits). """ return has_role(role_names, GLOBAL_ADMIN_ROLE, FINANCE_ROLE) def can_view_ai_spending(role_names: Iterable[str]) -> bool: """Who can read the AI Spending dashboard. Per user direction 2026-04-28: IT Admin + Finance + Auditor. Management added 2026-05-02 as the read-only leadership view. Cloud admin added 2026-05-20 to align with /help matrix row 6 (AI 花费 / 云账单 / 部门成本): cloud_admin operates the underlying cloud accounts whose Bailian / Bedrock / Vertex / Foundry meters feed this dashboard — they need spend visibility to size / rebalance / decommission, same reason finance gets the page. Mutations stay on ``can_manage_ai_spending`` (global_admin only). """ return has_role( role_names, GLOBAL_ADMIN_ROLE, FINANCE_ROLE, AUDITOR_ROLE, CLOUD_ADMIN_ROLE, MANAGEMENT_ROLE, ) def can_manage_ai_spending(role_names: Iterable[str]) -> bool: """Mutate AI Spending config / regenerate summary / assign owners. Only IT Admin. Finance is read-only per the same convention as ``can_manage_finance``. """ return has_role(role_names, GLOBAL_ADMIN_ROLE) def can_view_hr_data(role_names: Iterable[str]) -> bool: """HR role can read org tree, HRIS sync batches, directory, lifecycle events. No HR-driven writes in this phase. """ return has_role( role_names, GLOBAL_ADMIN_ROLE, HR_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE, ) def can_view_directory(role_names: Iterable[str]) -> bool: """Read the lightweight people / department directory. Wider than ``can_view_hr_data`` — every admin-tier role legitimately needs to look up *who* a person is to do their job: * ``asset_admin`` assigns laptops / triggers transfers * ``finance`` resolves payment requesters / approvers * ``cloud_admin`` adds cloud-resource owners and alert ccs * ``hr`` / ``global_admin`` / ``auditor`` / ``management`` are the traditional governance readers The directory endpoints (:func:`/api/v1/directory/people`, :func:`/api/v1/auth/departments`) return MINIMAL fields: ``{id, display_name, work_email, employee_no, department, employment_status}``. Sensitive HR-only fields (manager hierarchy, last_login_at, hris_record_state, roles[]) stay on the narrower ``/api/v1/users`` (gated by :func:`can_view_hr_data`). Background: PR 2026-06-01 — 各地资产管理员（区域 asset_admin）反馈 资产管理页的「持有人 / 部门」筛选下拉永远是空的。后台日志显示 /users + /auth/departments 一直 403,前端 try/catch 静默吞掉。 这个 helper 区分了「人物简介」(所有 admin 视图都需要)和 「HR 治理画像」(只有 HR 系视图需要),让前者可被所有 admin 角色 访问而不放宽后者。 ``employee`` is intentionally EXCLUDED. The directory endpoint is an internal admin lookup; an employee searching the org directory has their own (more constrained) surface via the directory tab on /me. Letting an unfiltered employee call the picker endpoint would let them scrape the full roster. """ return has_role( role_names, GLOBAL_ADMIN_ROLE, HR_ROLE, ASSET_ADMIN_ROLE, FINANCE_ROLE, CLOUD_ADMIN_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE, ) # ── Active-employee gate (Leaver Blocked runtime) ─────────────────── def is_active_employee(user) -> bool: """True iff the user is currently allowed to perform mutating actions in Helix. A user is considered "active" when: * ``employment_status == 'active'``, AND * ``deleted_at`` is unset. Leaver_pending users are EXCLUDED — once HR has confirmed an offboarding event, the person loses mutate authority immediately even before their termination_date passes. This is the runtime half of the Leaver Blocked invariant; the input half (rejection of flips to 'left' while blockers remain) lives in ``app.domains.leaver_blocked.guard``. Read-only access remains. The matching FastAPI dependency is ``app.core.dependencies.require_active_employee``. """ status = getattr(user, "employment_status", None) if status != "active": return False if getattr(user, "deleted_at", None) is not None: return False return True # ── Security Auditor gates ────────────────────────────────────────── def can_view_security(role_names: Iterable[str]) -> bool: """Read access to the Security Auditor tooling (MFA / privileged role review / public exposure / secret rotation / signin anomaly / cert expiry / risk score / evidence package). Open to global_admin, auditor, security_auditor (legacy alias), security_officer (the operations desk) and management (read-only leadership view). Finance is intentionally NOT here — security evidence is its own governance lane. """ return has_role( role_names, GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE, SECURITY_OFFICER_ROLE, ) def can_manage_security(role_names: Iterable[str]) -> bool: """Mutate access — open / close privileged-role reviews, decide review items, acknowledge findings, regenerate evidence packages, update component weights for the risk score. Auditor explicitly mutates: the auditor's job IS to make decisions. global_admin retains the same ability, and ``security_officer`` is added 2026-08-25 — the security desk triages vulnerabilities, sets severity, decides bounty amounts and replies to researchers, all of which are writes. ``management`` is deliberately absent: it is the read-only leadership view. """ return has_role( role_names, GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, SECURITY_OFFICER_ROLE, ) # ── GitHub 组织域闸门 ──────────────────────────────────────────────── def can_view_github_org(role_names: Iterable[str]) -> bool: """读 GitHub 组织的管理面:组织登记、成员台账、自动加入规则、 与 GitHub 的对账差异。 比写面宽:除了 GitHub 管理员和 global_admin,还开给 ``auditor`` / ``security_officer``(「谁能往我们的代码仓库里写东西」是标准的访问 治理问题,访问复核要读它)和 ``management``(只读领导视图)。 员工看**自己**在哪些组织里不走这个闸 —— 那是 ``/api/v1/me/...`` 的自助面,按「是不是本人」判,不按角色判。 """ return has_role( role_names, GLOBAL_ADMIN_ROLE, GITHUB_ADMIN_ROLE, AUDITOR_ROLE, SECURITY_OFFICER_ROLE, MANAGEMENT_ROLE, ) def can_manage_github_org(role_names: Iterable[str]) -> bool: """写 GitHub 组织域:登记/停用组织、配自动加入规则、手工增删成员、 处置对账漂移、重发或取消邀请。 ``github_admin`` 就是为这件事存在的(见角色常量处的说明)。 ``auditor`` / ``security_officer`` 在这里**只读不写** —— 审计的职责 是看见和判定,不是替 org owner 决定谁进哪个组织;``management`` 同样缺席,它是只读领导视图。 """ return has_role(role_names, GLOBAL_ADMIN_ROLE, GITHUB_ADMIN_ROLE) # ── Internal system registry gates ────────────────────────────────── def can_view_system_registry(role_names: Iterable[str]) -> bool: """Read the internal system registry (系统台账). Wider than the security lane on purpose. Three groups need the same answer to "who owns this system": * the security lane, triaging an inbound vulnerability report, * HR / global_admin, running an offboarding handover for a departing owner, * management, reading the ownership-coverage number. An employee's own systems are NOT read through here — that is the self-service ``/api/v1/me/systems`` surface, keyed on being the owner rather than on a role. """ return has_role( role_names, GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE, SECURITY_OFFICER_ROLE, HR_ROLE, ) def can_manage_system_registry(role_names: Iterable[str]) -> bool: """Create / edit / retire a registry entry, and set its owner. ``global_admin`` only, per the platform owner's call 2026-08-25: IT maintains the register. Deliberately narrower than :func:`can_manage_security` — the security desk consumes ownership to route a finding, it does not decide who owns a system, and a registry where two desks both edit ownership is a registry nobody trusts. The bounty console links a report to an existing entry (which is a bounty write, gated by ``can_manage_security``) and surfaces "not registered yet?" pointing here. """ return has_role(role_names, GLOBAL_ADMIN_ROLE) # ── Approval-flow registry gates ──────────────────────────────────── def can_view_approval_registry(role_names: Iterable[str]) -> bool: """Read the approval-flow register's ADMIN surface (审批流程台账). Same shape as the system registry's read gate, and for the same reasons: the governance lane reads coverage, HR runs the handover when an owner leaves, management reads the number. This gate does NOT guard the published list. That one is ``GET /api/v1/approval-flows/published``, open to any signed-in employee — the whole point of the register is that anyone can find out which approval to use and who to ask. What this gate protects is the operational view: sync bookkeeping, inference evidence, unpublished rows and the ownership trail. """ return has_role( role_names, GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE, SECURITY_OFFICER_ROLE, HR_ROLE, ) def can_manage_approval_registry(role_names: Iterable[str]) -> bool: """Register / edit / retire an approval flow, set its owner, run the Feishu sweep. ``global_admin`` only — IT maintains the register, matching :func:`can_manage_system_registry`. HR owns many of the *flows*, but ownership records are the register's own accountability data: a table where every department edits its own row is a table whose coverage number means nothing. HR's route to a correction is the dispute button or a word with IT. """ return has_role(role_names, GLOBAL_ADMIN_ROLE) # ── Leaver Blocked admin gate ─────────────────────────────────────── def can_manage_leaver_blocked(role_names: Iterable[str]) -> bool: """Initiate dual-sign overrides, force-advance leavers via the dashboard, view the full leaver-block audit log. HR is included because they own the offboarding pipeline end-to-end. """ return has_role(role_names, GLOBAL_ADMIN_ROLE, HR_ROLE) def can_decide_github_id_change(role_names: Iterable[str]) -> bool: """Backstop authority over a self-service GitHub-ID change request. The NORMAL decider is not a role at all: it is the requester's own HR Ops (``hris_hrops_person_id``), falling back to their HRBP — a data-driven duty, the same shape as a mail-group owner. This gate only covers the two cases that duty cannot: the ~4 employees who carry neither relationship (their requests would otherwise rot), and standing in when the named approver is away. HR is here because the field is HR master data (SF ``cust_Other``) and the manual process this replaces was literally "contact your HRBP". """ return has_role(role_names, GLOBAL_ADMIN_ROLE, HR_ROLE) def can_secondary_sign_leaver_override(role_names: Iterable[str]) -> bool: """Approve / reject a dual-sign override. ANY person can be a secondary signer if their identity matches the leaver's department director chain — that check is enforced at the service layer against the actual person's manager hierarchy. This gate covers the BASELINE: only managers (anyone with subordinates) and admin roles can act as secondary; the chain check narrows further. """ return has_role(role_names, GLOBAL_ADMIN_ROLE, HR_ROLE) def can_force_bypass_leaver_grace(role_names: Iterable[str]) -> bool: """Force the T+0 24h IT grace window to close NOW. Stricter than ``can_manage_leaver_blocked``: HR is excluded. The grace window exists to give IT a manual-cleanup buffer; only IT can legitimately decide IT is ready to skip it. HR initiating this bypass would defeat its purpose. global_admin (alias it_admin per the role normalization table) only. """ return has_role(role_names, GLOBAL_ADMIN_ROLE) # ── Relationship graph gate ───────────────────────────────────────── def can_view_relationship_graph(role_names: Iterable[str]) -> bool: """View the resource relationship graph at /admin/relationship-graph. Open to admins, auditors, and finance (finance reads cost-related edges). Regular employees see only their own first-degree edges via the ``/me`` embed; they don't reach the admin page. """ return has_role( role_names, GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, FINANCE_ROLE, ASSET_ADMIN_ROLE, CLOUD_ADMIN_ROLE, HR_ROLE, MANAGEMENT_ROLE, ) def can_alias_for_others(role_names: Iterable[str]) -> bool: """HR (and global admin) can create email aliases targeting any user or mail group. Employees can only alias their own mailbox / mail groups they own. """ return has_role(role_names, GLOBAL_ADMIN_ROLE, HR_ROLE) def can_delete_any_mail_group(role_names: Iterable[str]) -> bool: """HR + global admin can delete any mail group. Regular owners can delete their own groups via the owner check in the service layer. """ return has_role(role_names, GLOBAL_ADMIN_ROLE, HR_ROLE) def can_manage_email_groups(role_names: Iterable[str]) -> bool: """HR + global admin hold the mail-group management desk. Covers BOTH flavours of group under the 邮件组管理 entry: * **dynamic** groups — membership computed from a rule over synced HR attributes. No per-group owner exists (membership is rule-driven, and the rules key off department / office / country data HR owns end-to-end), so management is a role gate rather than the owner-OR-admin check static groups use. Per the 2026-07-22 decision HR edits apply immediately (no four-eyes), fully audited. * **static** groups — the admin-tier view over the whole estate (governance filters, operation log, cleanup workbench). Regular owners keep their own narrower path through the public directory and the owner-OR-admin checks in :class:`~app.domains.email_group.service.EmailGroupService`; this gate is what lets HR act on a group they do not own. Same tier as :func:`can_delete_any_mail_group` — a role that can delete any group can also manage the estate around it. """ return has_role(role_names, GLOBAL_ADMIN_ROLE, HR_ROLE) # Pre-2026-09 name, kept as an alias so existing dynamic-group call # sites (service gate, MCP tool docs) keep working. The gate widened # from "dynamic groups" to "the whole mail-group desk" when the two # management surfaces merged into one menu entry; the role-set is # identical, so this is a rename rather than a behaviour change. can_manage_dynamic_email_groups = can_manage_email_groups # ── Cloud-scope owner RBAC (PR 2026-04-23) ────────────────────── # # ``scope_owner`` is NOT a DB role — it's derived from # ``cmdb_cloud_scopes.owner_person_id`` at request time. A person is a # scope owner iff at least one active scope names them; the router # layer resolves this via :func:`is_cloud_scope_owner_of` for the # specific scope the request targets. async def is_cloud_scope_owner_of( session: "object", person_id: str, scope_id: str, ) -> bool: """Return True when person owns the given scope. Router-level gate.""" from sqlalchemy import select as _select # noqa: PLC0415 from app.domains.cmdb.models import ( # noqa: PLC0415 CmdbCloudScope, ) row = await session.scalar( # type: ignore[attr-defined] _select(CmdbCloudScope).where( CmdbCloudScope.id == scope_id, CmdbCloudScope.owner_person_id == str(person_id), CmdbCloudScope.deleted_at.is_(None), ) ) return row is not None async def get_subordinate_person_ids( session: "object", manager_person_id: str, *, max_depth: int | None = None, ) -> list[str]: """Return the person_ids of all subordinates reporting (transitively) to ``manager_person_id``. Bounded BFS, dedup'd, cycle-safe. ``max_depth`` defaults to the canonical :data:`app.core.person_scope.DEFAULT_MAX_DEPTH` (10) rather than a hard-coded 5. PingCAP reporting chains run 6–7 levels deep, so the old default silently dropped the deepest reports of a senior manager / VP from every "my team's resources" view (cloud / permission / MCP) that calls this wrapper — none of the ~18 callers pass ``max_depth``. Keeping the two entry points in sync avoids that under-inclusion. Used by ``/me/*`` endpoints across cloud / asset / hr to expand the visibility filter from ``person_id == self`` to ``person_id IN (self ∪ subordinates)``. Returns the list EXCLUDING the caller — caller adds themselves to the IN clause. Thin compatibility wrapper over the canonical :func:`app.core.person_scope.subordinate_person_ids` with ``scope=ALL`` — i.e. it returns every non-deleted transitive report regardless of ``employment_status`` (the cloud-resource visibility views that call this carry their own ``include_left`` toggle and deliberately need to see what a departed report still holds). Callers that want the leaver-lockout invariant should call ``subordinate_person_ids`` directly with ``MANAGED`` / ``ACTIVE_ONLY``. """ from app.core.person_scope import ( # noqa: PLC0415 DEFAULT_MAX_DEPTH, SubordinateScope, subordinate_person_ids, ) rows = await subordinate_person_ids( session, # type: ignore[arg-type] manager_person_id, scope=SubordinateScope.ALL, max_depth=DEFAULT_MAX_DEPTH if max_depth is None else max_depth, ) # Historical contract: this helper returns ``str`` ids (its cmdb / mcp # / permission callers build ``set(...)`` / ``.in_(...)`` over strings). return [str(pid) for pid in rows] async def person_owns_any_cloud_scope(session: "object", person_id: str) -> bool: """Global "is a cloud scope owner" check — used by the frontend to decide whether to show the /my/cloud/owner-console entry point. """ from sqlalchemy import func as _func # noqa: PLC0415 from sqlalchemy import select as _select # noqa: PLC0415 from app.domains.cmdb.models import CmdbCloudScope # noqa: PLC0415 count = await session.scalar( # type: ignore[attr-defined] _select(_func.count(CmdbCloudScope.id)).where( CmdbCloudScope.owner_person_id == str(person_id), CmdbCloudScope.deleted_at.is_(None), ) ) return bool(count and count > 0)