Communitygithub.com

Identity-Governance-Lifecycle-Prozess

Entwerfen und automatisieren Sie Identity-Governance- und Lifecycle-Programme auf Plattformen wie SailPoint, Saviynt oder Entra ID Governance, einschließlich Joiner-Mover-Leaver (JML)-Automatisierung, Rollen-Mining, Zugriffsanfragen, periodischer Rezertifizierung und der Bereinigung von verwaisten Konten auf Basis von HR-Daten.

Was ist Identity-Governance-Lifecycle-Prozess?

Diese Fähigkeit beschreibt umfassend die Planung und Implementierung eines Identity-Governance- und Lifecycle-Management-Prozesses, der typischerweise auf Plattformen wie SailPoint, Saviynt oder Microsoft Entra ID Governance aufbaut. Der Fokus liegt auf der Automatisierung des Joiner-Mover-Leaver (JML)-Lebenszyklus von Identitäten, der durch ein autoritatives HR-System (z.B. Workday, SAP SuccessFactors) gesteuert wird. Die Skills-Anleitung definiert einen Zustandsautomaten mit sieben Identitätszuständen (z. B. PRE_HIRE, ACTIVE, TERMINATED) und spezifischen automatisierten Aktionen für jeden Zustandsübergang. Darüber hinaus enthält sie Codebeispiele für die HR-Systemintegration (einen Workday-Connector) und eine Rollen-Mining-Engine, die automatisch Berechtigungen analysiert, um Geburtsrechtsrollen zu definieren, Rollenexplosionen zu erkennen und verwaiste Zugriffe zu identifizieren. Die Fähigkeit adressiert auch die Einhaltung von Compliance-Anforderungen wie SOX, HIPAA und GDPR, indem sie dokumentierte Prozesse und Prüfprotokolle gewährleistet. Sie ist ideal für Organisationen, die ihre Identity-Governance-Programme aufbauen oder optimieren möchten, insbesondere wenn es um die systemübergreifende Automatisierung von Bereitstellungen, die Bereinigung von Zugriffen ehemaliger Mitarbeiter und die Einhaltung gesetzlicher Vorschriften geht.

Funktioniert mit~Claude Code~Codex CLI~Cursor
npx skills add https://github.com/mukul975/Anthropic-Cybersecurity-Skills/tree/main/skills/building-identity-governance-lifecycle-process

In Ihrer bevorzugten KI fragen

Öffnet einen neuen Chat, in dem dieser Agent-Skill bereits geladen ist.

Dokumentation

Erstellen eines Identity-Governance-Lifecycle-Prozesses

Wann verwenden

  • Der Organisation fehlen automatisierte Joiner-Mover-Leaver (JML)-Prozesse für das Identitätsmanagement.
  • Die Zugriffsbereitstellung erfolgt manuell und dauert Tage, was zu Produktivitätseinbußen und Sicherheitslücken führt.
  • Ehemalige Mitarbeiter behalten nach der Kündigung den Zugriff auf Systeme (verwaiste Konten).
  • Rollenexplosion hat tausende von Rollen mit unklaren Besitzverhältnissen und überlappenden Berechtigungen geschaffen.
  • Compliance-Anforderungen erzwingen dokumentierte Identitätslebenszyklus-Prozesse (SOX, HIPAA, GDPR).
  • Es gibt keine zentralisierte Transparenz darüber, wer im gesamten Unternehmen Zugriff auf welche Ressourcen hat.

Nicht verwenden für das Benutzermanagement einzelner Anwendungen; Identity Governance adressiert das systemübergreifende Lifecycle-Management, das die Korrelation autoritativer HR-Quellen mit der nachgelagerten Anwendungsbereitstellung erfordert.

Voraussetzungen

  • Autoritatives HR-System (Workday, SAP SuccessFactors, BambooHR) als Quelle der Wahrheit für Identitäten.
  • IGA-Plattform (SailPoint, Saviynt, One Identity) oder Microsoft Entra ID Governance.
  • Active Directory und/oder Azure AD als primäre Verzeichnisdienste.
  • Anwendungs-Connectoren für Zielsysteme, die eine automatisierte Bereitstellung erfordern.
  • Definierte Organisationsrollenstruktur und Berichtshierarchie.
  • Zustimmung der Stakeholder aus HR, IT, Sicherheit und Fachbereichsleitern.

Workflow

Schritt 1: Identitätslebenszyklus-Zustände und -Übergänge definieren

Kartieren Sie den Identitätslebenszyklus von der Einstellung bis zur Kündigung:

"""
Identity Lifecycle State Machine
Definiert alle Identitätszustände und gültigen Übergänge mit automatisierten Aktionen.
"""

IDENTITY_LIFECYCLE = {
    "states": {
        "PRE_HIRE": {
            "description": "Identität aus HR-Feed vor dem Startdatum erstellt",
            "automated_actions": [
                "Identitätsdatensatz in der IGA-Plattform erstellen",
                "Eindeutige Mitarbeiter-ID generieren",
                "Postfachreservierung erstellen",
                "Geburtsrechtsrollen basierend auf der Stellenkennung zuweisen",
                "Hintergrundüberprüfungs-Workflow initiieren"
            ],
            "valid_transitions": ["ACTIVE", "CANCELLED"]
        },
        "ACTIVE": {
            "description": "Mitarbeiter hat begonnen, vollständiger Zugriff bereitgestellt",
            "automated_actions": [
                "Active Directory-Konto erstellen",
                "E-Mail-Postfach erstellen",
                "Birthright-Anwendungszugriff bereitstellen",
                "Abteilungsspezifische Rollen zuweisen",
                "Zu Verteilerlisten hinzufügen",
                "MFA-Token/Sicherheitsschlüssel ausstellen",
                "VPN-Konto erstellen, wenn Remote-Mitarbeiter"
            ],
            "valid_transitions": ["ROLE_CHANGE", "LEAVE_OF_ABSENCE", "TERMINATED"]
        },
        "ROLE_CHANGE": {
            "description": "Mitarbeiter versetzt, befördert oder Abteilungswechsel",
            "automated_actions": [
                "Rollen zuweisungen basierend auf neuer Stellenkennung neu berechnen",
                "Zugriff auf vorherige Abteilungsanwendungen entfernen",
                "Zugriff für neue Abteilungsanwendungen bereitstellen",
                "Gruppenmitgliedschaften aktualisieren",
                "Vorgesetzten im Verzeichnis übertragen",
                "Zugriffsüberprüfung für beibehaltene Berechtigungen auslösen",
                "Neuen Vorgesetzten über geerbten Zugriff benachrichtigen"
            ],
            "valid_transitions": ["ACTIVE", "LEAVE_OF_ABSENCE", "TERMINATED"]
        },
        "LEAVE_OF_ABSENCE": {
            "description": "Mitarbeiter in längerer Abwesenheit (medizinisch, Elternzeit, Sabbatical)",
            "automated_actions": [
                "Interaktiven Login deaktivieren (Konto erhalten)",
                "VPN-Zugriff sperren",
                "Abwesenheitsnotiz einrichten",
                "Postfach an Vorgesetzten delegieren",
                "Alle Rollenzuweisungen für die Rückkehr erhalten",
                "Reaktivierungsdatum aus HR-Feed setzen"
            ],
            "valid_transitions": ["ACTIVE", "TERMINATED"]
        },
        "TERMINATED": {
            "description": "Mitarbeiter hat das Unternehmen verlassen",
            "automated_actions": [
                "AD-Konto sofort deaktivieren",
                "Allen Anwendungszugriff widerrufen",
                "VPN- und Remotezugriff widerrufen",
                "Postfach in freigegebenes Postfach umwandeln (Vorgesetztenzugriff für 90 Tage)",
                "OneDrive-Dateien an Vorgesetzten übertragen",
                "Aus allen Sicherheits- und Verteilerlisten entfernen",
                "OAuth-Tokens und API-Schlüssel widerrufen",
                "Unternehmensdaten von mobilen Geräten löschen",
                "Identitätsdatensatz archivieren",
                "Kontolöschung nach Aufbewahrungsfrist planen"
            ],
            "valid_transitions": ["REHIRE", "DELETED"]
        },
        "REHIRE": {
            "description": "Zuvor gekündigter Mitarbeiter kehrt zurück",
            "automated_actions": [
                "Vorhandenen Identitätsdatensatz reaktivieren",
                "Anmeldeinformationen zurücksetzen und erneute MFA-Registrierung anfordern",
                "Bereitstellung basierend auf neuer Stellenkennung (nicht vorherigem Zugriff)",
                "Für erweiterte Zugriffsüberprüfung in den ersten 30 Tagen markieren"
            ],
            "valid_transitions": ["ACTIVE"]
        },
        "DELETED": {
            "description": "Konto nach Aufbewahrungsfrist endgültig entfernt",
            "automated_actions": [
                "AD-Konto löschen",
                "E-Mail-Postfacharchiv löschen",
                "Identitätsdatensatz aus IGA entfernen",
                "Löschungs-Prüfprotokoll erstellen"
            ],
            "valid_transitions": []
        }
    },
    "retention_periods": {
        "terminated_to_deleted": "90 Tage (Standard)",
        "mailbox_retention": "90 Tage als freigegebenes Postfach",
        "onedrive_retention": "30 Tage Vorgesetztenzugriff, dann archiviert",
        "audit_log_retention": "7 Jahre für Compliance"
    }
}

Schritt 2: Implementierung der autoritativen Quellintegration

Verbinden Sie das HR-System als einzige Quelle der Wahrheit für Identitätsdaten:

"""
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

Schritt 3: Rollen-Mining und Birthright-Zugriff implementieren

Definieren Sie Rollen basierend auf Aufgabenbereichen für die automatisierte Bereitstellung:

"""
Role Mining Engine
Analysiert bestehende Zugriffsmuster, um Rollendefinitionen
für die Birthright-Bereitstellung abzuleiten.
"""
import pandas as pd
from collections import Counter
from itertools import combinations

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

    def mine_birthright_roles(self, min_assignment_pct=0.8):
        """
        Identifiziert Berechtigungen, die automatisch zugewiesen werden sollen,
        basierend auf der Stellenkennung. Wenn 80%+ der Benutzer mit derselben
        Stellenkennung eine Berechtigung haben, wird sie zum Birthright-Zugriff.
        """
        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):
        """Identifiziert Rollen mit übermäßigen Überschneidungen, die auf Konsolidierungsbedarf hinweisen."""
        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):
        """
        Findet Berechtigungen, die nicht mehr mit einer Rollendefinition übereinstimmen.
        Dies sind Ausnahmen, die sich im Laufe der Zeit angesammelt haben.
        """
        # Birthright-Definitionen abrufen
        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"],
                    "reason": "Not in birthright definition"
                })

        return orphaned

    def generate_role_catalog(self):
        """Erstellt einen vollständigen Rollenkatalog mit Metadaten."""
        birthright = self.mine_birthright_roles()
        catalog = []

        for job_code, details in birthright.items():
            catalog.append({
                "role_name": f"Birthright_{job_code}",
                "job_code": job_code,
                "user_count": details["total_users"],
                "entitlement_count": len(details["birthright_entitlements"]),
                "entitlements": details["birthright_entitlements"],
                "role_type": "Birthright",
                "owner": "TBD - Assign based on department"
            })

        return catalog

Schritt 4: Zugriffsanforderungs-Workflow entwerfen

Erstellen Sie einen Self-Service-Zugriffsanforderungsprozess mit Genehmigung:

"""
Access Request Workflow Engine
Verwaltet Self-Service-Zugriffsanforderungen mit mehrstufigen Genehmigungen.
"""
from enum import Enum
from datetime import datetime

class RequestStatus(Enum):
    DRAFT = "Draft"
    PENDING_APPROVAL = "Pending Approval"
    APPROVED = "Approved"
    DENIED = "Denied"
    PROVISIONING = "Provisioning"
    COMPLETED = "Completed"
    ERROR = "Error"

class AccessRequest:
    def __init__(self, requester_id, justification):
        self.request_id = None
        self.requester_id = requester_id
        self.justification = justification
        self.items = []
        self.approvals = []
        self.status = RequestStatus.DRAFT
        self.created_at = datetime.utcnow()

    def add_request_item(self, application, entitlement, owner_id):
        """Ein einzelnes Zugriffsanforderungselement hinzufügen."""
        self.items.append({
            "application": application,
            "entitlement": entitlement,
            "owner_id": owner_id,
            "status": "Pending"
        })

    def submit(self):
        """Anforderung zur Genehmigung einreichen."""
        if not self.items:
            raise ValueError("Anforderung muss mindestens ein Element enthalten")
        self.status = RequestStatus.PENDING_APPROVAL
        self.submitted_at = datetime.utcnow()
        return self._generate_approval_chain()

    def _generate_approval_chain(self):
        """Generiert Genehmigungskette basierend auf Anwendungseigentümern."""
        approval_chain = []
        for item in self.items:
            approval_chain.append({
                "request_item": item,
                "approver": item["owner_id"],
                "status": "Pending",
                "level": 1
            })
        return approval_chain

    def approve(self, approver_id, item_index=None):
        """Ein bestimmtes Element oder alle ausstehenden Elemente genehmigen."""
        if item_index is not None:
            self.items[item_index]["status"] = "Approved"
        else:
            for item in self.items:
                if item["status"] == "Pending":
                    item["status"] = "Approved"

        self.approvals.append({
            "approver_id": approver_id,
            "action": "Approved",
            "timestamp": datetime.utcnow()
        })

        # Prüfen, ob alle genehmigt sind
        if all(item["status"] == "Approved" for item in self.items):
            self.status = RequestStatus.APPROVED
        elif all(item["status"] in ["Approved", "Denied"] for item in self.items):
            self.status = RequestStatus.COMPLETED if any(item["status"] == "Approved" for item in self.items) else RequestStatus.DENIED

    def deny(self, approver_id, reason, item_index=None):
        """Ein bestimmtes Element oder alle ausstehenden Elemente ablehnen."""
        if item_index is not None:
            self.items[item_index]["status"] = "Denied"
            self.items[item_index]["denial_reason"] = reason
        else:
            for item in self.items:
                if item["status"] == "Pending":
                    item["status"] = "Denied"
                    item["denial_reason"] = reason

        self.approvals.append({
            "approver_id": approver_id,
            "action": "Denied",
            "reason": reason,
            "timestamp": datetime.utcnow()
        })

        if all(item["status"] == "Denied" for item in self.items):
            self.status = RequestStatus.DENIED

class AccessRequestManager:
    def __init__(self):
        self.requests = {}

    def create_request(self, requester_id, justification):
        """Eine neue Zugriffsanforderung erstellen."""
        request = AccessRequest(requester_id, justification)
        request.request_id = f"REQ-{len(self.requests) + 1:06d}"
        self.requests[request.request_id] = request
        return request

    def get_pending_approvals(self, approver_id):
        """Alle ausstehenden Genehmigungen für einen bestimmten Genehmiger abrufen."""
        pending = []
        for req_id, request in self.requests.items():
            if request.status == RequestStatus.PENDING_APPROVAL:
                for item in request.items:
                    if item["status"] == "Pending" and item["owner_id"] == approver_id:
                        pending.append({
                            "request_id": req_id,
                            "requester": request.requester_id,
                            "justification": request.justification,
                            "item": item,
                            "submitted_at": request.submitted_at
                        })
        return pending

Schritt 5: Periodische Zugriffsüberprüfung (Rezertifizierung) implementieren

Richten Sie regelmäßige Überprüfungen ein, um Compliance sicherzustellen:

"""
Access Certification Campaign Engine
Verwaltet periodische Zugriffsüberprüfungskampagnen für Compliance.
"""
from datetime import datetime, timedelta
from enum import Enum

class CampaignStatus(Enum):
    DRAFT = "Draft"
    ACTIVE = "Active"
    COMPLETED = "Completed"
    CANCELLED = "Cancelled"

class CertificationCampaign:
    def __init__(self, name, campaign_type, due_date):
        self.name = name
        self.campaign_type = campaign_type  # SOX, HIPAA, GDPR, Internal
        self.due_date = due_date
        self.status = CampaignStatus.DRAFT
        self.reviewers = []
        self.reviews = []
        self.created_at = datetime.utcnow()

    def add_reviewer(self, reviewer_id, scope):
        """Einen Prüfer mit einem definierten Bereich hinzufügen."""
        self.reviewers.append({
            "reviewer_id": reviewer_id,
            "scope": scope,  # z.B. "Department: Engineering", "Application: SAP"
            "progress": {
                "total": 0,
                "reviewed": 0,
                "certified": 0,
                "revoked": 0
            }
        })

    def start_campaign(self):
        """Kampagne aktivieren und Prüfungsaufgaben zuweisen."""
        self.status = CampaignStatus.ACTIVE
        self.started_at = datetime.utcnow()

        for reviewer in self.reviewers:
            # Zugriffsdatensätze basierend auf dem Bereich abrufen
            access_records = self._get_access_records(reviewer["scope"])
            reviewer["progress"]["total"] = len(access_records)

            # Prüfungsaufgaben erstellen
            for record in access_records:
                self.reviews.append({
                    "reviewer_id": reviewer["reviewer_id"],
                    "access_record": record,
                    "decision": None,
                    "comments": None,
                    "reviewed_at": None
                })

    def _get_access_records(self, scope):
        """Zugriffsdatensätze für einen bestimmten Bereich abrufen."""
        # Dummy-Implementierung – in der Praxis würde dies die IGA-Plattform abfragen
        return [
            {
                "employee_id": "EMP001",
                "application": "SAP",
                "entitlement": "Finance_Admin",
                "last_used": "2024-01-15",
                "assigned_date": "2023-06-01"
            },
            {
                "employee_id": "EMP002",
                "application": "Salesforce",
                "entitlement": "Sales_Manager",
                "last_used": "2024-02-01",
                "assigned_date": "2023-01-15"
            }
        ]

    def review_access(self, reviewer_id, record_index, decision, comments=""):
        """Eine Zugriffsüberprüfungsentscheidung aufzeichnen."""
        review = self.reviews[record_index]
        if review["reviewer_id"] != reviewer_id:
            raise ValueError("Prüfer ist nicht für diesen Datensatz verantwortlich")

        review["decision"] = decision  # "Certify" oder "Revoke"
        review["comments"] = comments
        review["reviewed_at"] = datetime.utcnow()

        # Fortschritt aktualisieren
        for reviewer in self.reviewers:
            if reviewer["reviewer_id"] == reviewer_id:
                reviewer["progress"]["reviewed"] += 1
                if decision == "Certify":
                    reviewer["progress"]["certified"] += 1
                elif decision == "Revoke":
                    reviewer["progress"]["revoked"] += 1

        # Prüfen, ob Kampagne abgeschlossen ist
        if self._is_complete():
            self.status = CampaignStatus.COMPLETED
            self.completed_at = datetime.utcnow()

    def _is_complete(self):
        """Prüfen, ob alle Überprüfungen abgeschlossen sind."""
        return all(review["decision"] is not None for review in self.reviews)

    def get_reminder_list(self):
        """Prüfer abrufen, die ihre Überprüfungen noch nicht abgeschlossen haben."""
        incomplete = []
        for reviewer in self.reviewers:
            if reviewer["progress"]["reviewed"] < reviewer["progress"]["total"]:
                incomplete.append(reviewer["reviewer_id"])
        return incomplete

Schritt 6: Verwaiste Konten und Zugriffsbereinigung umsetzen

Identifizieren und beheben Sie Konten von ehemaligen Mitarbeitern:

"""
Orphaned Account Remediation Engine
Identifiziert Konten ohne aktiven Mitarbeiter und leitet Bereinigung ein.
"""
from datetime import datetime, timedelta

class OrphanedAccountRemediation:
    def __init__(self, hr_system, application_connectors):
        """
        hr_system: Schnittstelle zum autoritativen HR-System
        application_connectors: Liste der Anwendungs-Connectoren
        """
        self.hr_system = hr_system
        self.application_connectors = application_connectors

    def detect_orphaned_accounts(self):
        """Konten mit HR zuordnen und verwaiste identifizieren."""
        active_employees = set(self.hr_system.get_active_employee_ids())
        orphaned = []

        for connector in self.application_connectors:
            app_accounts = connector.get_all_accounts()

            for account in app_accounts:
                # Prüfen, ob das Konto mit einem aktiven Mitarbeiter verknüpft ist
                linked_employee = self._find_linked_employee(account)

                if linked_employee is None:
                    # Konto ohne HR-Verknüpfung – als verwaist markieren
                    orphaned.append({
                        "account_id": account["id"],
                        "application": connector.name,
                        "username": account.get("username"),
                        "last_login": account.get("last_login"),
                        "created_date": account.get("created_date"),
                        "risk_score": self._calculate_risk_score(account),
                        "reason": "No linked employee record"
                    })
                elif account["employee_id"] not in active_employees:
                    # Konto mit ehemaligem Mitarbeiter verknüpft
                    orphaned.append({
                        "account_id": account["id"],
                        "application": connector.name,
                        "username": account.get("username"),
                        "last_login": account.get("last_login"),
                        "created_date": account.get("created_date"),
                        "risk_score": self._calculate_risk_score(account),
                        "reason": "Employee terminated",
                        "terminated_date": linked_employee.get("termination_date")
                    })

        return sorted(orphaned, key=lambda x: x["risk_score"], reverse=True)

    def _find_linked_employee(self, account):
        """Versuch, ein Konto mit einem HR-Datensatz zu verknüpfen."""
        # Implementieren Sie die Logik zur Verknüpfung von Konten basierend auf:
        # - E-Mail-Adresse
        # - employee_id im Attribut
        # - Benutzername
        # - Benutzerdefinierte Felder
        if account.get("employee_id"):
            return self.hr_system.get_employee(account["employee_id"])
        elif account.get("email"):
            return self.hr_system.find_by_email(account["email"])
        return None

    def _calculate_risk_score(self, account):
        """Risikobewertung für verwaistes Konto berechnen."""
        score = 0

        # Risiko erhöhen, wenn Konto zuletzt kurz nach Beendigung verwendet wurde
        if account.get("last_login"):
            last_login = datetime.fromisoformat(account["last_login"])
            days_since_login = (datetime.utcnow() - last_login).days
            if days_since_login < 30:  # Kürzlich verwendet
                score += 30
            elif days_since_login < 90:
                score += 10

        # Risiko für privilegierte Konten (z.B. Admin-Rollen)
        if account.get("is_privileged", False):
            score += 50

        # Risiko für Dienstkonten
        if account.get("account_type") == "Service":
            score += 20

        return min(score, 100)

    def remediate_account(self, account_id, action="disable"):
        """Bereinigungsmaßnahme für ein verwaistes Konto durchführen."""
        remediation_actions = {
            "disable": self._disable_account,
            "remove_access": self._remove_sensitive_access,
            "convert_to_shared": self._convert_to_shared,
            "delete": self._delete_account
        }

        if action not in remediation_actions:
            raise ValueError(f"Unbekannte Aktion: {action}")

        result = remediation_actions[action](account_id)
        self._log_remediation_action(account_id, action, result)
        return result

    def _disable_account(self, account_id):
        """Konto deaktivieren, dabei Daten für Reaktivierung erhalten."""
        # Connectoren aufrufen, um Konto zu deaktivieren
        pass

    def _remove_sensitive_access(self, account_id):
        """Nur sensible Berechtigungen entfernen, Konto intakt lassen."""
        pass

    def _convert_to_shared(self, account_id):
        """Persönliches Konto in freigegebenes Konto umwandeln."""
        pass

    def _delete_account(self, account_id):
        """Konto dauerhaft löschen (nach Aufbewahrungsfrist)."""
        pass

    def _log_remediation_action(self, account_id, action, result):
        """Bereinigungsmaßnahme für Prüfzwecke protokollieren."""
        audit_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "account_id": account_id,
            "action": action,
            "result": result,
            "initiated_by": "Automated - Orphaned Account Remediation"
        }
        # In Prüfprotokolldatenbank schreiben
        pass

Best Practices

Prozess-Governance

  1. Autoritative Quelle zuerst: Integrieren Sie immer das HR-System, bevor Sie JML-Regeln definieren. Ohne eine einzige Quelle der Wahrheit führt das Lifecycle-Management zu Inkonsistenzen.
  2. Birthright-Richtlinien: Beginn Sie mit 80 % Schwellenwert für Birthright-Zugriff; senken Sie ihn schrittweise, wenn die Zugriffsbereitstellung reift und Ausnahmen reduziert werden.
  3. Genehmigungsstufen: Verwenden Sie standardmäßig eine einstufige Genehmigung für Birthright-Zugriff und zweistufig für privilegierten Zugriff.
  4. Kampagnenhäufigkeit: Führen Sie Zugriffsüberprüfungskampagnen vierteljährlich für sensible Systeme und jährlich für Standardsysteme durch.
  5. Aufbewahrungsfristen: Richten Sie Aufbewahrungsfristen für Konten entsprechend der gesetzlichen Anforderungen ein (z. B. 90 Tage für die allgemeine Löschung, 7 Jahre für Prüfprotokolle).

Automatisierungsstrategie

  • Just-in-Time (JIT): Verwenden Sie JIT-Bereitstellung für privilegierten Zugriff, immer wenn möglich, um Dauerzugriffsrechte zu reduzieren.
  • Event-gesteuert: Implementieren Sie ereignisgesteuerte Verarbeitung (Webhooks, Event Hubs) gegenüber Polling, wann immer die Plattform dies unterstützt.
  • Ausnahmeverwaltung: Erstellen Sie ein separates Workflow für Zugriffsausnahmen mit strengeren Genehmigungsanforderungen und Ablaufdaten.
  • Rollback-Mechanismen: Implementieren Sie Rollback-Funktionen für fehlgeschlagene Bereitstellungen und automatisierte Tests.

Sicherheitsüberlegungen

  • Privileged Access Management (PAM): Integrieren Sie PAM-Lösungen für die Bereitstellung von Admin-Zugriff mit Zeitsperre.
  • Trennungsbefugnis (SoD): Überprüfen Sie Berechtigungen automatisch auf SoD-Konflikte, bevor Sie Rollen zuweisen.
  • Prüfpfad: Führen Sie unveränderliche Prüfprotokolle aller Lebenszyklus-Ereignisse, Änderungen und Überprüfungsentscheidungen.
  • Verschlüsselung: Verschlüsseln Sie Identitätsdaten sowohl im Ruhezustand als auch während der Übertragung.

Verwandte Fähigkeiten

  • workload-identity-federation: Implementieren Sie die Workload-Identitätsföderation für nicht-menschliche Identitäten (Servicekonten, Anwendungen).
  • pet-model-infrastructure-deployment: Entwerfen Sie eine „Haustier-Modell“-Infrastrukturbereitstellung, die Identitätsbereitstellung als Teil der Infrastruktur betrachtet.
  • cattle-model-infrastructure-deployment: Erkunden Sie die „Herden-Modell“-Bereitstellung für kurzlebige Umgebungen ohne dauerhafte Identitäten.

Zuletzt aktualisiert: 2024-05-23

Individual skills in this repo

This repo contains 20 individual skills — each has its own dedicated page.

abusing-dpapi-for-credential-access

Extract and decrypt Windows DPAPI-protected secrets (Credential Manager, browser logins/cookies, Wi-Fi credentials, KeePass keys) online or offline using SharpDPAPI, SharpChrome, Mimikatz, or Impacket

abusing-shadow-credentials-for-privesc

Take over Active Directory accounts by writing attacker-controlled public keys to msDS-KeyCredentialLink (Shadow Credentials) with pyWhisker, Whisker, or Certipy, then authenticate via PKINIT to recover the target

achieving-cmmc-level-2-compliance

>-

acquiring-disk-image-with-dd-and-dcfldd

Create forensically sound bit-for-bit disk images with dd or dcfldd on a Linux forensic workstation, preserving evidence integrity through hash verification (MD5/SHA) during acquisition. Use when imaging a suspect drive, USB device, or memory card for investigation, preserving volatile disk evidence during incident response, or producing a verified copy for legal or law-enforcement proceedings before any destructive analysis.

analyzing-active-directory-acl-abuse

Detect dangerous ACL misconfigurations in Active Directory using ldap3

analyzing-android-malware-with-apktool

Perform static analysis of Android APK malware using apktool for resource decompilation, jadx for Java source recovery, and androguard for manifest inspection, dangerous permission-combination detection, and identification of obfuscated code, dynamic code loading, and reflection-based API calls. Use to statically triage a suspicious APK without executing it or to build mobile malware detection rules.

analyzing-api-gateway-access-logs

Parses API Gateway access logs (AWS API Gateway, Kong, Nginx) to detect

analyzing-apt-group-with-mitre-navigator

Query ATT&CK data with attackcti, mitreattack-python, and stix2, then build MITRE ATT&CK Navigator layers and multi-layer heatmap overlays mapping one or more APT groups

analyzing-azure-activity-logs-for-threats

Queries Azure Monitor activity logs and sign-in logs via azure-monitor-query

analyzing-bootkit-and-rootkit-samples

Analyzes bootkit and advanced rootkit malware infecting the Master

analyzing-browser-forensics-with-hindsight

Parse Chromium-based browser databases with Hindsight to extract and correlate browsing history, downloads, cookies, cached content, autofill data, saved passwords, and extensions from Chrome, Edge, Brave, Opera, and Vivaldi into a unified timeline (XLSX, JSON, or SQLite output). Use during incident response, insider-threat investigations, or criminal cases when you need to reconstruct a user

analyzing-campaign-attribution-evidence

Systematically evaluate cyber-campaign evidence to attribute an operation to a threat actor, using the Diamond Model and Analysis of Competing Hypotheses (ACH) to weigh infrastructure overlaps, TTP consistency, malware code similarity, and timing/language artifacts into confidence-weighted attribution assessments. Use when an incident investigation needs a defensible attribution confidence level.

analyzing-certificate-transparency-for-phishing

Monitor Certificate Transparency logs using crt.sh and Certstream to

analyzing-cloud-storage-access-patterns

Detect abnormal access in AWS S3, GCS, and Azure Blob Storage by analyzing CloudTrail Data Events, GCS audit logs, and Azure Storage Analytics for after-hours bulk downloads, new-IP access, and API-call spikes (e.g. GetObject) via statistical baselines and time-series anomaly detection. Use when investigating suspected cloud data exfiltration or building related detection rules.

analyzing-cobalt-strike-beacon-configuration

Extract and analyze Cobalt Strike beacon configuration from PE files

analyzing-cobaltstrike-malleable-c2-profiles

Parse and analyze Cobalt Strike Malleable C2 profiles with dissect.cobaltstrike (profiles and beacon-payload configs) and pyMalleableC2 (AST parsing) to extract HTTP/DNS transforms, URIs, headers, sleep/jitter, and injection behavior, then generate network detection signatures. Use when reverse-engineering a captured malleable profile or building detections against Cobalt Strike Beacon traffic.

analyzing-command-and-control-communication

Analyzes malware C2 communication over HTTP, HTTPS, DNS, and custom

analyzing-cyber-kill-chain

Analyzes intrusion activity against the Lockheed Martin Cyber Kill Chain

analyzing-disk-image-with-autopsy

Perform comprehensive forensic analysis of raw (dd), E01, or AFF disk images with Autopsy and The Sleuth Kit, recovering deleted files, examining metadata and embedded artifacts, keyword searching, and building investigation timelines with visual reports. Use for structured analysis of a forensic disk image or when stakeholders need visual reports from evidence.

analyzing-dns-logs-for-exfiltration

Analyzes DNS query logs to detect data exfiltration via DNS tunneling,

Verwandte Skills