building-identity-governance-lifecycle-process

アイデンティティガバナンスのライフサイクルプロセスを構築する

使用するタイミング

• アイデンティティ管理における入社・異動・退職（JML）プロセスが自動化されていない
• アクセス権の付与が手作業で、完了まで数日かかり、生産性の低下やセキュリティ上の不備が生じている
• 退職後も元従業員がシステムへのアクセス権を保持している（孤立アカウント）
• ロールが増えすぎて、所有者が不明確で権限が重複するロールが何千も存在している
• コンプライアンス要件により、文書化されたアイデンティティライフサイクルプロセスが求められている（SOX、HIPAA、GDPR）
• 企業全体で誰が何にアクセスできるかを一元的に把握できていない

次の用途には使用しないでください：単一アプリケーションのユーザー管理。アイデンティティガバナンスは、信頼できる人事情報源と下流のアプリケーションへのアクセス権付与を関連付ける必要がある、システム横断的なライフサイクル管理に対応します。

前提条件

• アイデンティティの信頼できる唯一の情報源としての人事システム（Workday、SAP SuccessFactors、BambooHR）
• IGA プラットフォーム（SailPoint、Saviynt、One Identity）または Microsoft Entra ID Governance
• 主要なディレクトリサービスとしての Active Directory および／または Azure AD
• 自動的なアクセス権付与が必要な対象システム向けのアプリケーションコネクター
• 定義済みの組織ロール構造とレポート階層
• 人事、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 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）	従業員の入社（入社者）、役割や部署の変更（異動者）、退職時の手続き（退職者）を対象とする、アイデンティティライフサイクルの主要な遷移
標準アクセス権	アクセス申請を必要とせず、職務コード、部署、所在地に基づいて自動的に付与される基本的な権限
ロールマイニング	類似の職務に共通する権限のグループを特定し、既存のアクセスパターンを分析してロール定義を導き出すこと
孤立アカウント	信頼できる人事情報源に対応する有効なアイデンティティがなく、セキュリティリスクとなるアプリケーションアカウント
信頼できる情報源	アイデンティティ属性と雇用状況の唯一の信頼できる情報源となる記録システム（通常は人事システム）
アクセス申請ワークフロー	ユーザーが追加の権限を申請できるセルフサービスプロセス。リスクに基づいて承認先を割り当てます

ツールとシステム

• SailPoint IdentityIQ/IdentityNow：ライフサイクル管理、アクセス認定、自動的なアクセス権付与に対応するエンタープライズ向け IGA プラットフォーム
• Saviynt Enterprise Identity Cloud：アイデンティティウェアハウス、アクセスガバナンス、アプリケーションアクセス管理を備えたクラウドネイティブ IGA
• Microsoft Entra ID Governance：ライフサイクルワークフロー、アクセスレビュー、権限管理などのアイデンティティガバナンス機能
• One Identity Manager：ビジネスロール管理、証明、アクセス申請用の IT ショップを備えた IGA ソリューション

一般的なシナリオ

シナリオ：従業員10,000人の組織向けJMLプロセスの構築

コンテキスト：急成長中の企業では、IDライフサイクルが自動化されていません。IT部門がアカウントを手動で作成しているため、新入社員のアカウント作成に3～5日かかります。退職した従業員は数週間にわたってアクセス権を保持しています。監査では、45個のアプリケーション全体で2,300件の孤立アカウントが見つかりました。

アプローチ：

1. Workdayを信頼できる情報源として統合し、IGAプラットフォームと毎日の差分同期を行います
2. 既存のアクセスパターンを分析して、上位20件の職務コード（従業員の80％をカバー）に対する初期付与ロールを定義します
3. 開始日の7日前にトリガーされる入社前のプロビジョニングを実装し、AD、メール、初期付与アプリに適用します
4. 人事ステータスの変更から1時間以内にすべてのアクセスを無効化する退職ワークフローを構築します
5. 職務コードまたは部門が変更されたときにロールを再計算する異動ワークフローを構築します
6. リスクベースの承認フローを備えたセルフサービスのアクセス申請ポータルを導入します
7. 孤立アカウントの検出を実行し、既存の2,300件の孤立アカウントを特定して是正します
8. アクセス権の蓄積を防ぐため、四半期ごとのアクセス認定をスケジュールします

注意点：

• 信頼できる情報源を一元化して定義しないと、複数の人事システムから競合するIDデータが発生します
• ビジネス部門による検証なしにロールを分析すると、組織構造に合わない技術的なロールが作成されます
• 引き継ぎのための猶予期間を設けずに退職を自動化すると、業務マネージャーの不満を招きます
• 人事システムに登録されていない契約社員やベンダーのIDを処理できません

出力形式

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