身份治理与生命周期流程构建

构建身份治理生命周期流程

何时使用

• 组织缺少自动化的入职-调岗-离职（JML）身份管理流程
• 访问权限配置是手动的，需要数天时间，造成生产力损失和安全缺口
• 前员工离职后仍保留系统访问权限（僵尸账户）
• 角色爆炸导致产生数千个角色，所有权不明确且权限重叠
• 合规要求强制要求记录在案的身份生命周期流程（SOX、HIPAA、GDPR）
• 全企业范围内缺乏对谁拥有哪些访问权限的集中可见性

不要用于单应用用户管理；身份治理解决的是跨系统生命周期管理，需要将权威 HR 数据源与下游应用置备进行关联。

前提条件

• 权威 HR 系统（Workday、SAP SuccessFactors、BambooHR）作为身份数据源
• IGA 平台（SailPoint、Saviynt、One Identity）或 Microsoft Entra ID Governance
• Active Directory 和/或 Azure AD 作为主要目录服务
• 用于需要自动化置备的目标系统的应用连接器
• 已定义的组织角色结构和汇报层级
• 来自 HR、IT、安全和业务部门经理的利益相关方支持

工作流程

第 1 步：定义身份生命周期状态和转换

映射从入职到离职的身份生命周期：

"""
身份生命周期状态机
定义所有身份状态及有效的转换，并包含自动化操作。
"""

IDENTITY_LIFECYCLE = {
    "states": {
        "PRE_HIRE": {
            "description": "员工在入职前，从 HR 数据源创建的身份记录",
            "automated_actions": [
                "在 IGA 平台创建身份记录",
                "生成唯一员工 ID",
                "预留邮箱",
                "根据岗位代码分配继承角色",
                "启动背景调查工作流"
            ],
            "valid_transitions": ["ACTIVE", "CANCELLED"]
        },
        "ACTIVE": {
            "description": "员工已入职，所有权限已配置完成",
            "automated_actions": [
                "创建 Active Directory 账户",
                "创建电子邮件邮箱",
                "配置继承的应用权限",
                "分配部门特定角色",
                "添加到通讯组",
                "发放 MFA 令牌/安全密钥",
                "如为远程员工，创建 VPN 账户"
            ],
            "valid_transitions": ["ROLE_CHANGE", "LEAVE_OF_ABSENCE", "TERMINATED"]
        },
        "ROLE_CHANGE": {
            "description": "员工发生调岗、晋升或部门变更",
            "automated_actions": [
                "根据新岗位代码重新计算角色分配",
                "移除前部门应用的访问权限",
                "配置新部门应用的访问权限",
                "更新组成员身份",
                "在目录中更新经理信息",
                "为保留的权限触发访问审查",
                "通知新经理有关继承的访问权限"
            ],
            "valid_transitions": ["ACTIVE", "LEAVE_OF_ABSENCE", "TERMINATED"]
        },
        "LEAVE_OF_ABSENCE": {
            "description": "员工长期休假（病假、产假、学术假）",
            "automated_actions": [
                "禁用交互式登录（保留账户）",
                "暂停 VPN 访问",
                "设置自动回复",
                "将邮箱委托给经理",
                "保留所有角色分配，待返回时恢复",
                "根据 HR 数据源设置恢复日期"
            ],
            "valid_transitions": ["ACTIVE", "TERMINATED"]
        },
        "TERMINATED": {
            "description": "员工已离职",
            "automated_actions": [
                "立即禁用 AD 账户",
                "撤销所有应用访问权限",
                "撤销 VPN 和远程访问权限",
                "将邮箱转换为共享（经理可访问90天）",
                "将 OneDrive 文件转移给经理",
                "从所有安全组和通讯组中移除",
                "撤销 OAuth 令牌和 API 密钥",
                "清除移动设备上的公司数据",
                "归档身份记录",
                "在保留期后安排账户删除"
            ],
            "valid_transitions": ["REHIRE", "DELETED"]
        },
        "REHIRE": {
            "description": "前员工重新入职",
            "automated_actions": [
                "重新激活现有身份记录",
                "重置凭据并要求重新注册 MFA",
                "根据新岗位代码配置权限（而非恢复先前权限）",
                "在前30天内标记为加强访问审查"
            ],
            "valid_transitions": ["ACTIVE"]
        },
        "DELETED": {
            "description": "保留期后永久删除账户",
            "automated_actions": [
                "删除 AD 账户",
                "删除电子邮件邮箱归档",
                "从 IGA 移除身份记录",
                "生成删除审计日志"
            ],
            "valid_transitions": []
        }
    },
    "retention_periods": {
        "terminated_to_deleted": "90 天（默认）",
        "mailbox_retention": "作为共享邮箱保留90天",
        "onedrive_retention": "经理访问权限30天，然后归档",
        "audit_log_retention": "保留7年以符合合规要求"
    }
}

第 2 步：实现权威数据源集成

将 HR 系统作为身份数据的单一权威来源进行连接：

"""
HR 数据源集成 - Workday 到 IGA 平台连接器
轮询 Workday 的员工生命周期事件，并触发置备操作。
"""
import requests
from datetime import datetime, timedelta
import logging

class WorkdayIdentityConnector:
    def __init__(self, config):
        self.base_url = config["workday_api_url"]
        self.tenant = config["tenant"]
        self.client_id = config["client_id"]
        self.client_secret = config["client_secret"]
        self.session = requests.Session()
        self.logger = logging.getLogger("workday_connector")

    def get_access_token(self):
        """向 Workday REST API 进行身份验证。"""
        token_url = f"{self.base_url}/ccx/oauth2/{self.tenant}/token"
        response = self.session.post(token_url, data={
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        })
        response.raise_for_status()
        return response.json()["access_token"]

    def fetch_worker_changes(self, since_datetime):
        """获取自上次同步以来的所有员工生命周期事件。"""
        headers = {"Authorization": f"Bearer {self.get_access_token()}"}
        params = {
            "Updated_From": since_datetime.isoformat(),
            "Updated_Through": datetime.utcnow().isoformat(),
            "Count": 100
        }

        workers = []
        url = f"{self.base_url}/ccx/api/v1/{self.tenant}/workers"

        while url:
            response = self.session.get(url, headers=headers, params=params)
            response.raise_for_status()
            data = response.json()
            workers.extend(data.get("data", []))
            url = data.get("next", None)
            params = {}

        return workers

    def map_lifecycle_event(self, worker):
        """将 Workday 员工数据映射到身份生命周期事件。"""
        worker_data = worker.get("workerData", {})
        employment = worker_data.get("employmentData", {})
        personal = worker_data.get("personalData", {})

        event = {
            "employee_id": worker.get("id"),
            "first_name": personal.get("legalName", {}).get("firstName"),
            "last_name": personal.get("legalName", {}).get("lastName"),
            "email": worker_data.get("emailAddress"),
            "job_code": employment.get("jobProfile", {}).get("id"),
            "job_title": employment.get("jobProfile", {}).get("name"),
            "department": employment.get("organization", {}).get("name"),
            "department_code": employment.get("organization", {}).get("id"),
            "manager_id": employment.get("managerId"),
            "location": employment.get("location", {}).get("name"),
            "cost_center": employment.get("costCenter", {}).get("id"),
            "hire_date": employment.get("hireDate"),
            "termination_date": employment.get("terminationDate"),
            "status": employment.get("status"),
            "worker_type": employment.get("workerType"),
        }

        # 确定生命周期转换
        if event["status"] == "Active" and event["hire_date"]:
            hire_date = datetime.fromisoformat(event["hire_date"])
            if hire_date > datetime.utcnow():
                event["lifecycle_event"] = "PRE_HIRE"
            else:
                event["lifecycle_event"] = "JOINER"
        elif event["status"] == "Active":
            event["lifecycle_event"] = "MOVER"  # 部门或角色变更
        elif event["status"] == "Terminated":
            event["lifecycle_event"] = "LEAVER"
        elif event["status"] == "On Leave":
            event["lifecycle_event"] = "LEAVE_OF_ABSENCE"

        return event

    def process_lifecycle_events(self, since_datetime):
        """身份生命周期事件的主处理循环。"""
        workers = self.fetch_worker_changes(since_datetime)
        events = []

        for worker in workers:
            event = self.map_lifecycle_event(worker)
            events.append(event)
            self.logger.info(
                f"Lifecycle event: {event['lifecycle_event']} for "
                f"{event['first_name']} {event['last_name']} "
                f"(EmpID: {event['employee_id']})"
            )

        return events

第 3 步：实现角色挖掘和继承权限

根据岗位职能定义角色，以实现自动化配置：

"""
角色挖掘引擎
分析现有访问模式，推导出用于自动配置的角色定义。
"""
import pandas as pd
from collections import Counter
from itertools import combinations

class RoleMiningEngine:
    def __init__(self, access_data):
        """
        access_data: DataFrame，包含列
        [employee_id, job_code, department, application, entitlement]
        """
        self.access_data = access_data

    def mine_birthright_roles(self, min_assignment_pct=0.8):
        """
        识别应根据岗位代码自动分配的权限。
        如果具有相同岗位代码的用户中80%以上拥有某个权限，
        则该权限成为继承访问权限。
        """
        birthright_roles = {}

        for job_code, group in self.access_data.groupby("job_code"):
            total_users = group["employee_id"].nunique()
            entitlement_counts = group.groupby(
                ["application", "entitlement"]
            )["employee_id"].nunique()

            birthright_entitlements = []
            for (app, ent), count in entitlement_counts.items():
                pct = count / total_users
                if pct >= min_assignment_pct:
                    birthright_entitlements.append({
                        "application": app,
                        "entitlement": ent,
                        "assignment_percentage": round(pct * 100, 1),
                        "user_count": count
                    })

            if birthright_entitlements:
                birthright_roles[job_code] = {
                    "job_code": job_code,
                    "total_users": total_users,
                    "birthright_entitlements": birthright_entitlements
                }

        return birthright_roles

    def detect_role_explosion(self):
        """识别权限重叠过多的角色，表明需要合并。"""
        roles = self.access_data.groupby("job_code").apply(
            lambda x: set(zip(x["application"], x["entitlement"]))
        )

        overlap_report = []
        for (role1, ents1), (role2, ents2) in combinations(roles.items(), 2):
            if len(ents1) == 0 or len(ents2) == 0:
                continue
            overlap = len(ents1 & ents2)
            max_size = max(len(ents1), len(ents2))
            overlap_pct = overlap / max_size * 100

            if overlap_pct > 70:
                overlap_report.append({
                    "role_1": role1,
                    "role_2": role2,
                    "role_1_entitlements": len(ents1),
                    "role_2_entitlements": len(ents2),
                    "overlapping_entitlements": overlap,
                    "overlap_percentage": round(overlap_pct, 1),
                    "recommendation": "CONSOLIDATE" if overlap_pct > 90 else "REVIEW"
                })

        return sorted(overlap_report, key=lambda x: x["overlap_percentage"], reverse=True)

    def find_orphaned_access(self):
        """
        查找不再属于任何角色定义的权限。
        这些是随着时间的推移累积的异常权限。
        """
        # 获取继承权限定义
        birthright = self.mine_birthright_roles(min_assignment_pct=0.5)

        orphaned = []
        for _, row in self.access_data.iterrows():
            job_birthright = birthright.get(row["job_code"], {})
            expected_ents = set()
            for ent in job_birthright.get("birthright_entitlements", []):
                expected_ents.add((ent["application"], ent["entitlement"]))

            current_ent = (row["application"], row["entitlement"])