아이덴티티 거버넌스 라이프사이클 프로세스 구축

아이덴티티 거버넌스 라이프사이클 프로세스 구축은(는) 무엇을 하나요?

사용 시기

• 조직에 아이덴티티 관리를 위한 자동화된 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']} - "
                f"{event['first_name']} {event['last_name']} "
                f"(직원 ID: {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"])