building-identity-governance-lifecycle-process

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

适用场景

• 组织缺乏用于身份管理的自动化入职-调动-离职（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 State Machine
Defines all identity states and valid transitions with automated actions.
"""

IDENTITY_LIFECYCLE = {
    "states": {
        "PRE_HIRE": {
            "description": "Identity created from HR feed before start date",
            "automated_actions": [
                "Create identity record in IGA platform",
                "Generate unique employee ID",
                "Create mailbox reservation",
                "Assign birthright roles based on job code",
                "Initiate background check workflow"
            ],
            "valid_transitions": ["ACTIVE", "CANCELLED"]
        },
        "ACTIVE": {
            "description": "Employee has started, full access provisioned",
            "automated_actions": [
                "Create Active Directory account",
                "Create email mailbox",
                "Provision birthright application access",
                "Assign department-specific roles",
                "Add to distribution groups",
                "Issue MFA token/security key",
                "Create VPN account if remote worker"
            ],
            "valid_transitions": ["ROLE_CHANGE", "LEAVE_OF_ABSENCE", "TERMINATED"]
        },
        "ROLE_CHANGE": {
            "description": "Employee transferred, promoted, or changed departments",
            "automated_actions": [
                "Recalculate role assignments based on new job code",
                "Remove access from previous department applications",
                "Provision access for new department applications",
                "Update group memberships",
                "Transfer manager in directory",
                "Trigger access review for retained entitlements",
                "Notify new manager of inherited access"
            ],
            "valid_transitions": ["ACTIVE", "LEAVE_OF_ABSENCE", "TERMINATED"]
        },
        "LEAVE_OF_ABSENCE": {
            "description": "Employee on extended leave (medical, parental, sabbatical)",
            "automated_actions": [
                "Disable interactive login (preserve account)",
                "Suspend VPN access",
                "Set out-of-office auto-reply",
                "Delegate mailbox to manager",
                "Preserve all role assignments for return",
                "Set reactivation date from HR feed"
            ],
            "valid_transitions": ["ACTIVE", "TERMINATED"]
        },
        "TERMINATED": {
            "description": "Employee has left the organization",
            "automated_actions": [
                "Disable AD account immediately",
                "Revoke all application access",
                "Revoke VPN and remote access",
                "Convert mailbox to shared (manager access for 90 days)",
                "Transfer OneDrive files to manager",
                "Remove from all security and distribution groups",
                "Revoke OAuth tokens and API keys",
                "Wipe corporate data from mobile devices",
                "Archive identity record",
                "Schedule account deletion after retention period"
            ],
            "valid_transitions": ["REHIRE", "DELETED"]
        },
        "REHIRE": {
            "description": "Previously terminated employee returning",
            "automated_actions": [
                "Reactivate existing identity record",
                "Reset credentials and require MFA re-enrollment",
                "Provision based on new job code (not previous access)",
                "Flag for enhanced access review in first 30 days"
            ],
            "valid_transitions": ["ACTIVE"]
        },
        "DELETED": {
            "description": "Account permanently removed after retention period",
            "automated_actions": [
                "Delete AD account",
                "Delete email mailbox archive",
                "Remove identity record from IGA",
                "Generate deletion audit log"
            ],
            "valid_transitions": []
        }
    },
    "retention_periods": {
        "terminated_to_deleted": "90 days (default)",
        "mailbox_retention": "90 days as shared mailbox",
        "onedrive_retention": "30 days manager access, then archived",
        "audit_log_retention": "7 years for compliance"
    }
}

第 2 步：实现权威来源集成

将 HR 系统作为身份数据的唯一真实来源进行连接：

"""
HR Source Integration - Workday to IGA Platform Connector
Polls Workday for employee lifecycle events and triggers provisioning.
"""
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):
        """Authenticate to 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):
        """Fetch all worker lifecycle events since the last sync."""
        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):
        """Map Workday worker data to identity lifecycle event."""
        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"),
        }

        # Determine lifecycle transition
        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"  # Department or role change
        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):
        """Main processing loop for identity lifecycle events."""
        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 步：实现角色挖掘与默认权限

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

"""
Role Mining Engine
Analyzes existing access patterns to derive role definitions
for birthright (automatic) provisioning.
"""
import pandas as pd
from collections import Counter
from itertools import combinations

class RoleMiningEngine:
    def __init__(self, access_data):
        """
        access_data: DataFrame with columns
        [employee_id, job_code, department, application, entitlement]
        """
        self.access_data = access_data

    def mine_birthright_roles(self, min_assignment_pct=0.8):
        """
        Identify entitlements that should be automatically assigned
        based on job code. If 80%+ of users with same job code
        have an entitlement, it becomes birthright access.
        """
        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):
        """Identify roles with excessive overlap indicating need for consolidation."""
        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):
        """
        Find entitlements that no longer align with any role definition.
        These are exceptions that accumulated over time.
        """
        # Get birthright definitions
        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"])
            if current_ent not in expected_ents:
                orphaned.append({
                    "employee_id": row["employee_id"],
                    "job_code": row["job_code"],
                    "application": row["application"],
                    "entitlement": row["entitlement"],
                    "recommendation": "Review for revocation"
                })

        return pd.DataFrame(orphaned)

第 4 步：构建访问请求与审批工作流

实现基于风险审批的自助式访问请求：

"""
Access Request Workflow Engine
Handles self-service access requests with multi-level approvals
based on risk classification of requested entitlements.
"""

ACCESS_REQUEST_WORKFLOW = {
    "risk_levels": {
        "LOW": {
            "description": "Standard business applications",
            "examples": ["Email distribution groups", "SharePoint team sites", "Standard SaaS apps"],
            "approval_chain": ["manager"],
            "sla_hours": 4,
            "auto_approve_if_birthright": True
        },
        "MEDIUM": {
            "description": "Sensitive data access or elevated permissions",
            "examples": ["CRM admin", "Financial reporting", "HR systems"],
            "approval_chain": ["manager", "application_owner"],
            "sla_hours": 24,
            "auto_approve_if_birthright": False
        },
        "HIGH": {
            "description": "Privileged access or regulated data",
            "examples": ["Database admin", "Cloud admin", "PAM vault access"],
            "approval_chain": ["manager", "application_owner", "security_team"],
            "sla_hours": 48,
            "auto_approve_if_birthright": False,
            "require_justification": True,
            "require_time_limit": True
        },
        "CRITICAL": {
            "description": "Domain admin, root access, or production data modification",
            "examples": ["Domain Admin", "AWS root", "Production DB write"],
            "approval_chain": ["manager", "application_owner", "security_team", "ciso"],
            "sla_hours": 72,
            "auto_approve_if_birthright": False,
            "require_justification": True,
            "require_time_limit": True,
            "require_sod_check": True,
            "max_duration_days": 90
        }
    }
}

class AccessRequestEngine:
    def __init__(self, iga_client, risk_catalog):
        self.iga = iga_client
        self.risk_catalog = risk_catalog

    def submit_request(self, requester_id, entitlement_id, justification, duration_days=None):
        """Submit an access request with automatic risk classification."""
        # Classify risk level of requested entitlement
        risk_level = self.risk_catalog.get_risk_level(entitlement_id)
        workflow = ACCESS_REQUEST_WORKFLOW["risk_levels"][risk_level]

        # Check if entitlement is birthright for requester's role
        requester = self.iga.get_identity(requester_id)
        is_birthright = self.iga.is_birthright_for_role(
            entitlement_id, requester["job_code"]
        )

        if is_birthright and workflow.get("auto_approve_if_birthright"):
            return self._auto_approve(requester_id, entitlement_id, "Birthright access")

        # Run SOD check if required
        if workflow.get("require_sod_check"):
            sod_violations = self.iga.check_sod(requester_id, entitlement_id)
            if sod_violations:
                return {
                    "status": "SOD_VIOLATION",
                    "violations": sod_violations,
                    "action": "Request requires compensating control approval"
                }

        # Create approval chain
        request = {
            "requester": requester_id,
            "entitlement": entitlement_id,
            "risk_level": risk_level,
            "justification": justification,
            "duration_days": duration_days or workflow.get("max_duration_days"),
            "approval_chain": self._build_approval_chain(
                requester, workflow["approval_chain"]
            ),
            "sla_deadline": workflow["sla_hours"],
            "status": "PENDING_APPROVAL"
        }

        return self.iga.create_request(request)

    def _build_approval_chain(self, requester, approver_types):
        """Resolve approval chain to actual approver identities."""
        chain = []
        for approver_type in approver_types:
            if approver_type == "manager":
                chain.append({
                    "type": "manager",
                    "identity": requester["manager_id"],
                    "fallback": requester.get("skip_manager_id")
                })
            elif approver_type == "application_owner":
                chain.append({
                    "type": "application_owner",
                    "identity": "resolved_at_runtime",
                    "fallback": "it-governance-team"
                })
            elif approver_type == "security_team":
                chain.append({
                    "type": "group",
                    "identity": "security-governance-team",
                    "required_approvals": 1
                })
            elif approver_type == "ciso":
                chain.append({
                    "type": "role",
                    "identity": "CISO",
                    "fallback": "deputy-ciso"
                })
        return chain

第 5 步：实现孤儿账户检测与修复

识别并修复没有有效身份关联的账户：

"""
Orphaned Account Detection
Identifies accounts in target systems that have no corresponding
active identity in the authoritative HR source.
"""

class OrphanedAccountDetector:
    def __init__(self, hr_connector, app_connectors):
        self.hr = hr_connector
        self.apps = app_connectors

    def detect_orphaned_accounts(self):
        """Compare application accounts against HR active employees."""
        active_employees = set(self.hr.get_active_employee_ids())
        orphaned_accounts = []

        for app_name, connector in self.apps.items():
            app_accounts = connector.get_all_accounts()

            for account in app_accounts:
                correlated_id = account.get("employee_id") or account.get("correlation_id")

                if correlated_id and correlated_id not in active_employees:
                    # Check if recently terminated (within grace period)
                    termination_info = self.hr.get_termination_info(correlated_id)

                    orphaned_accounts.append({
                        "application": app_name,
                        "account_name": account["username"],
                        "correlated_employee_id": correlated_id,
                        "account_status": account.get("status", "unknown"),
                        "last_login": account.get("last_login"),
                        "termination_date": termination_info.get("date") if termination_info else None,
                        "days_since_termination": (
                            (datetime.utcnow() - termination_info["date"]).days
                            if termination_info and termination_info.get("date") else None
                        ),
                        "risk_level": self._assess_orphan_risk(account, termination_info)
                    })

                elif not correlated_id:
                    # Uncorrelated account - no link to any employee
                    orphaned_accounts.append({
                        "application": app_name,
                        "account_name": account["username"],
                        "correlated_employee_id": None,
                        "account_status": account.get("status", "unknown"),
                        "last_login": account.get("last_login"),
                        "risk_level": "HIGH",
                        "reason": "Uncorrelated - no employee association"
                    })

        return orphaned_accounts

    def _assess_orphan_risk(self, account, termination_info):
        """Assess risk level of orphaned account."""
        if account.get("is_privileged"):
            return "CRITICAL"
        if termination_info and termination_info.get("involuntary"):
            return "HIGH"
        if account.get("status") == "active":
            return "HIGH"
        return "MEDIUM"

    def generate_remediation_plan(self, orphaned_accounts):
        """Create remediation actions for orphaned accounts."""
        plan = []
        for account in orphaned_accounts:
            if account["risk_level"] == "CRITICAL":
                action = "DISABLE_IMMEDIATELY"
                sla = "4 hours"
            elif account["risk_level"] == "HIGH":
                action = "DISABLE_WITHIN_24H"
                sla = "24 hours"
            else:
                action = "REVIEW_AND_DISABLE"
                sla = "7 days"

            plan.append({
                **account,
                "remediation_action": action,
                "sla": sla,
                "assigned_to": "identity-governance-team"
            })

        return sorted(plan, key=lambda x: ["CRITICAL", "HIGH", "MEDIUM", "LOW"].index(x["risk_level"]))

关键概念

术语	定义
入职-调动-离职（JML）	核心身份生命周期转换，涵盖员工入职（入职）、角色/部门变更（调动）和离职（离职）
默认权限	根据岗位代码、部门或地点自动配置的基线权限，无需发起访问请求
角色挖掘	通过识别相似岗位职能中的常见权限分组，分析现有访问模式以推导角色定义
孤儿账户	在权威 HR 来源中不再有对应有效身份的应用程序账户，构成安全风险
权威来源	作为身份属性和雇佣状态唯一真实来源的记录系统（通常为 HR）
访问请求工作流	使用户能够请求额外权限，并按基于风险的审批路由进行的自助式流程

工具与系统

• SailPoint IdentityIQ/IdentityNow：用于生命周期管理、访问认证和自动化配置的企业级 IGA 平台
• Saviynt Enterprise Identity Cloud：云原生 IGA，具备身份仓库、访问治理和应用程序访问管理功能
• Microsoft Entra ID Governance：身份治理能力，包括生命周期工作流、访问审查和权限管理
• One Identity Manager：具备业务角色管理、认证和用于访问请求的 IT 商店的 IGA 解决方案

常见场景

场景：为 10,000 名员工的组织构建 JML 流程

背景：一家快速成长的公司没有自动化的身份生命周期管理。IT 手动创建账户，新员工入职需要 3-5 天。被解雇的员工在数周内仍保留访问权限。审计发现 45 个应用中存在 2,300 个孤儿账户。

方法：

1. 将 Workday 集成为权威源，每天向 IGA 平台增量同步
2. 挖掘现有访问模式，为排名前 20 的职位代码（覆盖 80% 的员工）定义默认角色
3. 为 AD、电子邮件和默认应用实施入职前预置，在入职日期前 7 天触发
4. 构建离职工作流，在 HR 状态变更后 1 小时内禁用所有访问权限
5. 创建调岗工作流，在职位代码或部门变更时重新计算角色
6. 部署自助访问请求门户，配备基于风险的审批链
7. 运行孤儿账户检测，识别并修复现有的 2,300 个孤儿账户
8. 安排季度访问认证，防止权限累积

陷阱：

• 未定义单一权威源会导致来自多个 HR 系统的身份数据相互冲突
• 未经业务验证就挖掘角色，会创建与组织结构不一致的技术角色
• 在自动化离职时未留出知识转移的宽限期，会引起业务经理的不满
• 未处理存在于 HR 系统之外的承包商和供应商身份

输出格式

IDENTITY GOVERNANCE LIFECYCLE REPORT
=======================================
Authoritative Source:   Workday
IGA Platform:          SailPoint IdentityIQ
Total Identities:      10,247
Active Employees:      9,834
Contractors:           413

LIFECYCLE AUTOMATION
Joiner (Pre-Hire) SLA:     Target: 0 days | Actual: 0.2 days avg
Mover Processing SLA:      Target: 1 day  | Actual: 0.8 days avg
Leaver Disablement SLA:    Target: 1 hour | Actual: 0.5 hours avg

PROVISIONING METRICS (Last 30 Days)
New Hires Provisioned:     187
  Auto-Provisioned:        174 (93.0%)
  Manual Intervention:     13 (7.0%)
Role Changes Processed:    89
Terminations Processed:    43
  Within 1-Hour SLA:       41 (95.3%)

ROLE GOVERNANCE
Defined Roles:             127
Birthright Roles:          48
Average Entitlements/Role: 12.3
Role Overlap > 70%:        8 pairs (consolidation recommended)

ORPHANED ACCOUNTS
Detected:                  23
  Critical:                2 (privileged accounts)
  High:                    8
  Medium:                  13
Remediated (30 days):      19
Outstanding:               4

ACCESS REQUESTS
Submitted:                 342
Auto-Approved (Birthright):87 (25.4%)
Approved:                  231 (67.5%)
Denied:                    24 (7.0%)
Average Approval Time:     6.2 hours
SOD Violations Flagged:    12