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

アイデンティティガバナンスライフサイクルプロセスの構築 は何をしますか？

使用する場合

• 組織にアイデンティティ管理のための自動化されたジョイナー・ムーバー・リーバー（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 to IGA Platform Connector
従業員ライフサイクルイベントについて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"ライフサイクルイベント: {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"])
            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"],
                    "department": row["department"]
                })

ステップ 4: アクセス要求と承認ワークフローを実装する

ユーザーが追加のアクセスを要求し、承認されるメカニズムを作成します:

"""
アクセス要求API
ユーザーが追加のアクセスを要求し、承認者は承認/拒否できます。
"""
from datetime import datetime
from enum import Enum
import uuid

class RequestStatus(Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"
    CANCELLED = "cancelled"
    ESCALATED = "escalated"

class AccessRequest:
    def __init__(self, requester_id, application, entitlement, justification):
        self.request_id = str(uuid.uuid4())
        self.requester_id = requester_id
        self.application = application
        self.entitlement = entitlement
        self.justification = justification
        self.status = RequestStatus.PENDING
        self.created_at = datetime.utcnow()
        self.approval_chain = []
        self.current_approver = None
        self.escalation_level = 0

class AccessRequestWorkflow:
    def __init__(self, approval_policies):
        """
        approval_policies: 承認ロジックを定義するポリシーの辞書
        形式: { "application:entitlement": {
                    "approvers": ["manager", "department_head", "app_owner"],
                    "escalation_minutes": [0, 60, 120]
                }
        }
        """
        self.approval_policies = approval_policies
        self.requests = {}

    def submit_request(self, requester_id, application, entitlement, justification):
        """新しいアクセス要求を送信します。"""
        request = AccessRequest(
            requester_id, application, entitlement, justification
        )
        self.requests[request.request_id] = request

        # 最初の承認者を決定
        policy = self.approval_policies.get(
            f"{application}:{entitlement}",
            self.approval_policies.get("default", {
                "approvers": ["manager"],
                "escalation_minutes": [0]
            })
        )

        request.approval_chain = policy["approvers"]
        request.current_approver = policy["approvers"][0]
        request.escalation_level = 0

        return request.request_id

    def approve_request(self, request_id, approver_id, comments=""):
        """保留中の要求を承認します。"""
        request = self.requests.get(request_id)
        if not request or request.status != RequestStatus.PENDING:
            return False

        # 承認者が現在の承認者であることを確認
        if request.current_approver != approver_id:
            return False

        # 次の承認者に進むか、完了する
        current_index = request.approval_chain.index(approver_id)
        if current_index + 1 < len(request.approval_chain):
            request.current_approver = request.approval_chain[current_index + 1]
            request.status = RequestStatus.PENDING
        else:
            request.status = RequestStatus.APPROVED
            request.approved_at = datetime.utcnow()
            # IGAプラットフォームでプロビジョニングをトリガー
            self._provision_access(request)

        return True

    def reject_request(self, request_id, approver_id, reason=""):
        """保留中の要求を拒否します。"""
        request = self.requests.get(request_id)
        if not request or request.status != RequestStatus.PENDING:
            return False

        request.status = RequestStatus.REJECTED
        request.rejection_reason = reason
        request.rejected_by = approver_id
        request.rejected_at = datetime.utcnow()
        return True

    def check_escalations(self):
        """タイムアウトした要求をエスカレーションします。"""
        now = datetime.utcnow()
        for request_id, request in self.requests.items():
            if request.status != RequestStatus.PENDING:
                continue

            elapsed = (now - request.created_at).total_seconds() / 60
            escalation_minutes = self.approval_policies.get(
                f"{request.application}:{request.entitlement}",
                self.approval_policies.get("default", {"escalation_minutes": [60]})
            )["escalation_minutes"]

            if request.escalation_level < len(escalation_minutes):
                if elapsed > escalation_minutes[request.escalation_level]:
                    request.escalation_level += 1
                    if request.escalation_level < len(request.approval_chain):
                        request.current_approver = request.approval_chain[request.escalation_level]
                        request.status = RequestStatus.ESCALATED

    def _provision_access(self, request):
        """承認されたアクセスプロビジョニングをトリガーします。"""
        # 実際の実装では、ここでIGAプラットフォームAPIを呼び出します
        self.logger.info(f"プロビジョニングアクセス: {request.requester_id} -> {request.application}:{request.entitlement}")

ステップ 5: 定期的な再認定（アクセスレビュー）を実装する

定期的なアクセスレビューキャンペーンを設定します:

"""
アクセスレビューキャンペーンエンジン
定期的なアクセスレビューをスケジュールおよび管理します。
"""
from datetime import datetime, timedelta
from enum import Enum

class ReviewStatus(Enum):
    SCHEDULED = "scheduled"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    OVERDUE = "overdue"

class CertificationCampaign:
    def __init__(self, name, campaign_type, start_date, end_date, reviewers):
        self.name = name
        self.campaign_type = campaign_type  # "SOX", "HIPAA", "GDPR", "GENERAL"
        self.start_date = start_date
        self.end_date = end_date
        self.reviewers = reviewers  # { reviewer_id: [entitlements_to_review] }
        self.status = ReviewStatus.SCHEDULED
        self.review_results = {}  # { (user_id, entitlement): "approve"|"revoke" }
        self.reminder_schedule = [7, 3, 1]  # 残り日数でのリマインダー送信

class CertEngine:
    def __init__(self, access_data, compliance_rules):
        """
        access_data: 現在のアクセス権限
        compliance_rules: コンプライアンス要件の定義
        """
        self.access_data = access_data
        self.compliance_rules = compliance_rules
        self.campaigns = []

    def create_campaign(self, name, campaign_type, due_date):
        """特定のコンプライアンス要件のための新しい再認定キャンペーンを作成します。"""
        # レビュー対象のアクセス決定
        if campaign_type == "SOX":
            # SOXアプリケーションにアクセスできるすべてのユーザーを含める
            relevant_access = self.access_data[
                self.access_data["application"].isin(
                    self.compliance_rules.get("SOX", {}).get("applications", [])
                )
            ]
        elif campaign_type == "HIPAA":
            relevant_access = self.access_data[
                self.access_data["application"].isin(
                    self.compliance_rules.get("HIPAA", {}).get("applications", [])
                )
            ]
        elif campaign_type == "GDPR":
            # EU在住のユーザー、またはEUユーザーデータを処理するすべての人
            relevant_access = self.access_data[
                (self.access_data["region"] == "EU") |
                (self.access_data["application"].isin(
                    self.compliance_rules.get("GDPR", {}).get("data_processing_apps", [])
                ))
            ]
        else:
            relevant_access = self.access_data

        # アクセス所有権に基づいてレビュー担当者に割り当て
        reviews = {}
        for _, row in relevant_access.iterrows():
            # アプリケーションの所有者またはアプリケーションの管理者をレビュー担当者として割り当て
            reviewer = self._determine_reviewer(row["application"])
            if reviewer not in reviews:
                reviews[reviewer] = []
            reviews[reviewer].append(
                (row["employee_id"], row["application"], row["entitlement"])
            )

        campaign = CertificationCampaign(
            name=name,
            campaign_type=campaign_type,
            start_date=datetime.utcnow(),
            end_date=due_date,
            reviewers=reviews
        )

        self.campaigns.append(campaign)
        return campaign

    def submit_review_decision(self, campaign, reviewer_id, decisions):
        """レビュー担当者がアクセスレビューの決定を提出します。"""
        """
        decisions: [{ "user_id": "...", "entitlement": "...", "decision": "approve"|"revoke" }]
        """
        if campaign.status not in [ReviewStatus.IN_PROGRESS, ReviewStatus.OVERDUE]:
            return False

        for decision in decisions:
            key = (decision["user_id"], decision["entitlement"])
            campaign.review_results[key] = decision["decision"]

            if decision["decision"] == "revoke":
                # IGA APIを介してアクセス削除をトリガー
                self._revoke_access(
                    decision["user_id"],
                    decision["application"],
                    decision["entitlement"]
                )

        # すべてのレビューが完了したかどうかを確認
        total_items = sum(len(items) for items in campaign.reviewers.values())
        reviewed_items = len(campaign.review_results)

        if reviewed_items >= total_items:
            campaign.status = ReviewStatus.COMPLETED
            self._generate_certification_report(campaign)

        return True

    def _generate_certification_report(self, campaign):
        """コンプライアンス監査用の再認定レポートを生成します。"""
        report = {
            "campaign_name": campaign.name,
            "campaign_type": campaign.campaign_type,
            "start_date": campaign.start_date.isoformat(),
            "end_date": campaign.end_date.isoformat(),
            "status": campaign.status.value,
            "total_entitlements_reviewed": len(campaign.review_results),
            "approved_count": sum(
                1 for d in campaign.review_results.values() if d == "approve"
            ),
            "revoked_count": sum(
                1 for d in campaign.review_results.values() if d == "revoke"
            ),
            "reviewers": list(campaign.reviewers.keys()),
            "completion_rate": 100.0
        }
        return report

ステップ 6: コンプライアンスレポートと監査証跡を実装する

規制要件のための監査証跡とレポートを維持します:

"""
監査レポートエンジン
コンプライアンス監査のためのアイデンティティアクティビティと権限の包括的なレポートを生成します。
"""
from datetime import datetime
import json

class AuditReportGenerator:
    def __init__(self, identity_store, access_history):
        """
        identity_store: 現在のアイデンティティとアクセスのデータベース
        access_history: アクセス変更の時系列ログ
        """
        self.identity_store = identity_store
        self.access_history = access_history

    def generate_sox_report(self, fiscal_year, fiscal_quarter):
        """SOXコンプライアンス監査のためのレポートを生成します。
        これにはアクセスレビューの完了、ユーザーアクセス権限、
        特権アカウントの変更が含まれます。"""
        report = {
            "report_type": "SOX",
            "fiscal_year": fiscal_year,
            "fiscal_quarter": fiscal_quarter,
            "generated_at": datetime.utcnow().isoformat(),
            "sox_applications": self._get_sox_applications(),
            "user_access_summary": self._get_user_access_summary(),
            "privileged_access_report": self._get_privileged_access(),
            "access_review_completion": self._get_review_completion_rate(),
            "segregation_of_duties_conflicts": self._get_sod_conflicts(),
            "significant_changes": self._get_significant_changes(fiscal_year, fiscal_quarter)
        }
        return report

    def generate_hipaa_report(self, date_range):
        """HIPAAコンプライアンス監査のためのレポートを生成します。
        これにはPHIデータへのアクセス、不正アクセス試行、
        アクセス権限の変更が含まれます。"""
        report = {
            "report_type": "HIPAA",
            "date_range": date_range,
            "generated_at": datetime.utcnow().isoformat(),
            "phi_application_access": self._get_phi_access(),
            "unauthorized_access_attempts": self._get_unauthorized_attempts(),
            "access_modifications": self._get_access_modifications(date_range),
            "terminated_employee_access": self._get_terminated_access(),
            "breach_assessment": self._assess_breach_risk()
        }
        return report

    def generate_gdpr_report(self, data_subject_id=None):
        """GDPRコンプライアンスのためのレポートを生成します。
        データ主体のアクセス要求（DSAR）とデータ処理アクティビティをサポートします。"""
        if data_subject_id:
            # 特定のデータ主体のレポート
            report = {
                "report_type": "GDPR_DSAR",
                "data_subject_id": data_subject_id,
                "generated_at": datetime.utcnow().isoformat(),
                "personal_data": self._get_personal_data(data_subject_id),
                "processing_activities": self._get_processing_activities(data_subject_id),
                "data_retention": self._get_retention_details(data_subject_id),
                "third_party_sharing": self._get_third_party_sharing(data_subject_id)
            }
        else:
            # 組織全体のレポート
            report = {
                "report_type": "GDPR_ORGANIZATIONAL",
                "generated_at": datetime.utcnow().isoformat(),
                "data_processing_register": self._get_processing_register(),
                "data_retention_policies": self._get_retention_policies(),
                "consent_management": self._get_consent_records(),
                "data_breach_log": self._get_breach_log(),
                "data_protection_impact_assessments": self._get_dpias()
            }
        return report

    def export_to_csv(self, report, filename):
        """監査レポートをCSVファイルにエクスポートします。"""
        import csv
        with open(filename, 'w', newline='') as csvfile:
            writer = csv.writer(csvfile)
            # フラット化されたレポートデータを書き込む
            self._flatten_and_write(report, writer)
        return filename

設定リファレンス

以下は、本番環境用にIGAソリューションを設定するためのサンプル設定です:

# IGAプラットフォーム設定
iga:
  platform: "sailpoint"  # または "saviynt", "entra_id"
  api_version: "v3"
  base_url: "https://identity.company.com/api"

# プロビジョニング設定
provisioning:
  batch_size: 100  # バッチあたりのユーザー数
  retry_attempts: 3
  retry_delay_seconds: 60
  timeouts:
    ad_account_creation: 30  # 秒
    mailbox_provisioning: 120
    application_provisioning: 300

# 再認定スケジュール
certification:
  schedules:
    - name: "SOX Quarterly Review"
      frequency: "QUARTERLY"
      months: [3, 6, 9, 12]
      due_day: 15
      reviewer_assignment: "app_owner"
    - name: "HIPAA Monthly Review"
      frequency: "MONTHLY"
      days: [1]
      due_day: 7
      reviewer_assignment: "department_head"
    - name: "GDPR Semi-Annual Review"
      frequency: "SEMI_ANNUAL"
      months: [6, 12]
      due_day: 31
      reviewer_assignment: "data_protection_officer"

# HRソース設定
hr_source:
  type: "workday"  # または "successfactors", "bamboo"
  sync_interval_minutes: 15
  retry_on_failure: true
  fields_to_map:
    employee_id: "Employee_ID"
    first_name: "FirstName"
    last_name: "LastName"
    email: "Email"
    job_code: "Job_Code"
    department: "Department"
    manager_id: "Manager_ID"
    hire_date: "Hire_Date"
    termination_date: "Termination_Date"
    status: "Employment_Status"

トラブルシューティング

一般的な問題と解決策

問題	考えられる原因	解決策
HR同期がイベントを検出しない	API認証エラー	APIトークンを更新し、アクセス許可を確認
ユーザーに生得権アクセスがプロビジョニングされない	ロールマッピングがない	roleMiningEngineを使用してジョブコードロールを確認
アクセス要求が承認されない	承認ワークフローが不完全	approval_policiesの構成を確認
再認定キャンペーンが期限を過ぎている	レビュー担当者が割り当てられていない	キャンペーン作成を検証し、レビュー担当者を手動で割り当て
退職した従業員がまだアクティブ	AD無効化が失敗した	ADコネクタのログを確認し、アカウントを手動で無効化
ロールの乱立が多すぎる	役割定義の重複	detect_role_explosion()を実行し、推奨に従って統合

ログ取得

# IGAコネクタのログを表示
tail -f /var/log/iga/connector.log

# デバッグモードを有効化
export IGA_LOG_LEVEL=DEBUG

# 監査ログを確認
cat /var/log/iga/audit.log | grep "TERMINATED"

# プロビジョニングエラーを監視
journalctl -u iga-provisioner -f --since "5 minutes ago"

ベストプラクティス

設計上の考慮事項

1. HRを信頼できる情報源として維持する: アイデンティティデータをHRシステム以外で直接変更しないでください。IGAはHRから変更を読み取り、反映する必要があります。

2. 最小権限の原則から始める: 新しいロールには生得権アクセスを最小限に抑え、ユーザーが要求を通じて適切なアクセスをリクエストできるようにします。

3. 変更管理を実装する: ロールの定義、アクセスポリシー、プロビジョニングルールへのすべての変更は、展開前にレビューおよび承認される必要があります。

4. コンプライアンスのための監査: 規制監査のための明確な証跡を提供するために、すべてのアイデンティティ変更、アクセス要求、承認、レビュー決定をログに記録します。

5. パフォーマンスを計画する: 大量のHR同期（例：合併、大量採用）中にシステムパフォーマンスを確保するために、バッチ処理と非同期プロビジョニングを使用します。

6. 例外ケースをテストする: アカウントがHRにないがIGAには存在する場合（シャドウIT）、または重複レコードがある場合など、エッジケースを処理するために堅牢なエラーハンドリングを実装します。

セキュリティ考慮事項

• IGAプラットフォームAPIへのすべてのアクセスには、サービスアカウントと最小権限のアクセストークンを使用します
• HRとIGAの間のAPI通信にはTLS 1.2+を強制します
• 機密性の高い権限（特権アクセス、管理者ロール）には、分離された義務（SoD）チェックを実装します
• 個人を特定できる情報（PII）を含む監査ログは、アクセス制御された安全な場所に保存します
• IGAプラットフォーム自体へのアクセスには、多要素認証（MFA）を要求します