Construcción de Procesos de Ciclo de Vida de Gobierno de Identidad
Cuándo Usar
- La organización carece de procesos automatizados de incorporación-traslado-baja (JML) para la gestión de identidades
- El aprovisionamiento de acceso es manual y toma días, creando pérdida de productividad y brechas de seguridad
- Exempleados retienen acceso a sistemas después de la baja (cuentas huérfanas)
- La explosión de roles ha creado miles de roles con propietarios poco claros y permisos superpuestos
- Los requisitos de cumplimiento normativo exigen procesos documentados del ciclo de vida de la identidad (SOX, HIPAA, GDPR)
- No hay visibilidad centralizada de quién tiene acceso a qué en toda la empresa
No usar para la gestión de usuarios de una sola aplicación; el gobierno de identidades aborda la gestión del ciclo de vida entre sistemas, requiriendo la correlación de fuentes autoritativas de RRHH con el aprovisionamiento de aplicaciones posteriores.
Prerrequisitos
- Sistema de RRHH autoritativo (Workday, SAP SuccessFactors, BambooHR) como fuente única de verdad de identidad
- Plataforma IGA (SailPoint, Saviynt, One Identity) o Microsoft Entra ID Governance
- Active Directory y/o Azure AD como servicios de directorio primarios
- Conectores de aplicaciones para los sistemas de destino que requieren aprovisionamiento automatizado
- Estructura de roles organizacional definida y jerarquía de reportes
- Aceptación de las partes interesadas de RRHH, TI, seguridad y gerentes de unidad de negocio
Flujo de Trabajo
Paso 1: Definir Estados y Transiciones del Ciclo de Vida de la Identidad
Mapear el ciclo de vida de la identidad desde la contratación hasta la baja:
"""
Máquina de Estados del Ciclo de Vida de la Identidad
Define todos los estados de identidad y las transiciones válidas con acciones automatizadas.
"""
IDENTITY_LIFECYCLE = {
"states": {
"PRE_HIRE": {
"description": "Identidad creada desde el feed de RRHH antes de la fecha de inicio",
"automated_actions": [
"Crear registro de identidad en la plataforma IGA",
"Generar ID de empleado único",
"Crear reserva de buzón de correo",
"Asignar roles de nacimiento según el código de puesto",
"Iniciar flujo de trabajo de verificación de antecedentes"
],
"valid_transitions": ["ACTIVE", "CANCELLED"]
},
"ACTIVE": {
"description": "El empleado ha comenzado, acceso completo aprovisionado",
"automated_actions": [
"Crear cuenta de Active Directory",
"Crear buzón de correo electrónico",
"Aprovisionar acceso de aplicaciones de nacimiento",
"Asignar roles específicos del departamento",
"Agregar a grupos de distribución",
"Emitir token MFA/llave de seguridad",
"Crear cuenta VPN si es trabajador remoto"
],
"valid_transitions": ["ROLE_CHANGE", "LEAVE_OF_ABSENCE", "TERMINATED"]
},
"ROLE_CHANGE": {
"description": "Empleado transferido, promovido o cambiado de departamento",
"automated_actions": [
"Recalcular asignaciones de roles según el nuevo código de puesto",
"Eliminar acceso a aplicaciones del departamento anterior",
"Aprovisionar acceso para aplicaciones del nuevo departamento",
"Actualizar membresías de grupos",
"Transferir gerente en el directorio",
"Activar revisión de acceso para permisos retenidos",
"Notificar al nuevo gerente sobre el acceso heredado"
],
"valid_transitions": ["ACTIVE", "LEAVE_OF_ABSENCE", "TERMINATED"]
},
"LEAVE_OF_ABSENCE": {
"description": "Empleado con licencia extendida (médica, parental, sabática)",
"automated_actions": [
"Deshabilitar inicio de sesión interactivo (preservar cuenta)",
"Suspender acceso VPN",
"Configurar respuesta automática de fuera de oficina",
"Delegar buzón al gerente",
"Preservar todas las asignaciones de roles para el regreso",
"Establecer fecha de reactivación desde el feed de RRHH"
],
"valid_transitions": ["ACTIVE", "TERMINATED"]
},
"TERMINATED": {
"description": "El empleado ha dejado la organización",
"automated_actions": [
"Deshabilitar cuenta de AD inmediatamente",
"Revocar todo el acceso a aplicaciones",
"Revocar acceso VPN y remoto",
"Convertir buzón a compartido (acceso del gerente por 90 días)",
"Transferir archivos de OneDrive al gerente",
"Eliminar de todos los grupos de seguridad y distribución",
"Revocar tokens OAuth y claves API",
"Borrar datos corporativos de dispositivos móviles",
"Archivar registro de identidad",
"Programar eliminación de cuenta después del período de retención"
],
"valid_transitions": ["REHIRE", "DELETED"]
},
"REHIRE": {
"description": "Empleado previamente dado de baja que regresa",
"automated_actions": [
"Reactivar registro de identidad existente",
"Restablecer credenciales y requerir reinscripción MFA",
"Aprovisionar según el nuevo código de puesto (no el acceso anterior)",
"Marcar para revisión de acceso mejorada en los primeros 30 días"
],
"valid_transitions": ["ACTIVE"]
},
"DELETED": {
"description": "Cuenta eliminada permanentemente después del período de retención",
"automated_actions": [
"Eliminar cuenta de AD",
"Eliminar archivo de buzón de correo electrónico",
"Eliminar registro de identidad del IGA",
"Generar registro de auditoría de eliminación"
],
"valid_transitions": []
}
},
"retention_periods": {
"terminated_to_deleted": "90 días (predeterminado)",
"mailbox_retention": "90 días como buzón compartido",
"onedrive_retention": "30 días de acceso del gerente, luego archivado",
"audit_log_retention": "7 años para cumplimiento"
}
}
Paso 2: Implementar Integración de Fuente Autoritativa
Conectar el sistema de RRHH como la única fuente de verdad para los datos de identidad:
"""
Integración de Fuente de RRHH - Conector de Workday a Plataforma IGA
Sondea Workday en busca de eventos del ciclo de vida del empleado y activa el aprovisionamiento.
"""
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):
"""Autenticar en la API REST de Workday."""
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):
"""Obtener todos los eventos del ciclo de vida del trabajador desde la última sincronización."""
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):
"""Mapear los datos del trabajador de Workday a un evento del ciclo de vida de la identidad."""
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"),
}
# Determinar la transición del ciclo de vida
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" # Cambio de departamento o rol
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):
"""Bucle de procesamiento principal para eventos del ciclo de vida de la identidad."""
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"Evento de ciclo de vida: {event['lifecycle_event']} para "
f"{event['first_name']} {event['last_name']} "
f"(EmpID: {event['employee_id']})"
)
return events
Paso 3: Implementar Minería de Roles y Acceso de Nacimiento
Definir roles basados en funciones laborales para el aprovisionamiento automatizado:
"""
Motor de Minería de Roles
Analiza patrones de acceso existentes para derivar definiciones de roles
para el aprovisionamiento de nacimiento (automático).
"""
import pandas as pd
from collections import Counter
from itertools import combinations
class RoleMiningEngine:
def __init__(self, access_data):
"""
access_data: DataFrame con columnas
[employee_id, job_code, department, application, entitlement]
"""
self.access_data = access_data
def mine_birthright_roles(self, min_assignment_pct=0.8):
"""
Identificar permisos que deben asignarse automáticamente
según el código de puesto. Si el 80%+ de los usuarios con el mismo
código de puesto tienen un permiso, se convierte en acceso de nacimiento.
"""
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):
"""Identificar roles con superposición excesiva que indiquen necesidad de consolidación."""
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):
"""
Encontrar permisos que ya no se alinean con ninguna definición de rol.
Estas son excepciones que se acumularon con el tiempo.
"""
# Obtener definiciones de nacimiento
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"],
"application": row["application"],
"entitlement": row["entitlement"],
"job_code": row["job_code"]
})
return orphaned
Paso 4: Implementar Flujo de Solicitud de Acceso y Aprobación
Proporcionar un modelo de solicitud de acceso autoservicio con escalado de aprobación:
"""
Motor de Solicitudes de Acceso
Maneja el ciclo de vida de las solicitudes de acceso de socios comerciales,
incluyendo enrutamiento de aprobación, cumplimiento de políticas y aprovisionamiento.
"""
from enum import Enum
from datetime import datetime, timedelta
import uuid
class RequestStatus(Enum):
DRAFT = "draft"
PENDING_MANAGER = "pending_manager"
PENDING_IT = "pending_it"
PENDING_COMPLIANCE = "pending_compliance"
APPROVED = "approved"
REJECTED = "rejected"
PROVISIONED = "provisioned"
FAILED = "failed"
CANCELLED = "cancelled"
class AccessRequestHandler:
def __init__(self, policy_store, provisioning_engine):
self.policy_store = policy_store
self.provisioning_engine = provisioning_engine
self.requests = {}
def create_access_request(self, requester_id, target_employee_id, entitlements, justification):
"""Crear una nueva solicitud de acceso con los aprobadores adecuados."""
request_id = str(uuid.uuid4())
request = {
"request_id": request_id,
"requester": requester_id,
"target_employee": target_employee_id,
"entitlements": entitlements,
"justification": justification,
"status": RequestStatus.DRAFT.value,
"created_at": datetime.utcnow().isoformat(),
"updated_at": datetime.utcnow().isoformat(),
"approval_chain": self._build_approval_chain(requester_id, entitlements),
"approvals": [],
"provisioning_results": []
}
# Iniciar flujo de aprobación
request["status"] = RequestStatus.PENDING_MANAGER.value
self._notify_approver(request["approval_chain"][0], request)
self.requests[request_id] = request
return request
def _build_approval_chain(self, requester_id, entitlements):
"""Determinar la cadena de aprobación según el nivel de riesgo."""
# Lógica simple: gerente -> TI para derechos estándar, agregar cumplimiento para privilegiado
chain = [
{"level": "manager", "approvers": [], "policy": "manager_approval"},
{"level": "it_security", "approvers": [], "policy": "it_approval"}
]
# Verificar si algún permiso requiere aprobación de cumplimiento
for ent in entitlements:
if self.policy_store.is_privileged(ent):
chain.append({
"level": "compliance",
"approvers": [],
"policy": "compliance_approval"
})
break
return chain
def approve_request(self, request_id, approver_id, comments=""):
"""Aprobar una solicitud en un nivel de aprobación."""
request = self.requests.get(request_id)
if not request:
raise ValueError(f"Solicitud {request_id} no encontrada")
current_level = request["approval_chain"][len(request["approvals"])]
request["approvals"].append({
"level": current_level["level"],
"approver": approver_id,
"decision": "approved",
"comments": comments,
"timestamp": datetime.utcnow().isoformat()
})
# Verificar si todas las aprobaciones están completas
if len(request["approvals"]) == len(request["approval_chain"]):
request["status"] = RequestStatus.APPROVED.value
self._provision_access(request)
else:
next_approvers = request["approval_chain"][len(request["approvals"])]
self._notify_approver(next_approvers, request)
request["updated_at"] = datetime.utcnow().isoformat()
return request
def _provision_access(self, request):
"""Ejecutar aprovisionamiento real de acceso a través del motor de aprovisionamiento."""
try:
results = self.provisioning_engine.provision(
request["target_employee"],
request["entitlements"]
)
request["provisioning_results"] = results
request["status"] = RequestStatus.PROVISIONED.value
except Exception as e:
request["status"] = RequestStatus.FAILED.value
request["error"] = str(e)
Paso 5: Configurar Recertificaciones y Campañas de Acceso
Establecer revisiones periódicas de acceso para cumplimiento:
"""
Motor de Campañas de Recertificación
Gestiona revisiones periódicas de acceso con recordatorios, escalaciones
y revocaciones automáticas de acceso no certificado.
"""
from datetime import datetime, timedelta
import uuid
class RecertificationCampaign:
def __init__(self, identity_store, notification_service):
self.identity_store = identity_store
self.notification_service = notification_service
self.campaigns = {}
def create_campaign(self, config):
"""
Crear una campaña de recertificación.
Config: {
"name": str,
"scope": ["all" | "department" | "role"],
"scope_value": str | None,
"reviewer_type": "manager" | "application_owner",
"due_in_days": 30,
"escalation_days": 7,
"auto_revoke": True
}
"""
campaign_id = str(uuid.uuid4())
campaign = {
"campaign_id": campaign_id,
"config": config,
"status": "draft",
"created_at": datetime.utcnow().isoformat(),
"due_date": (datetime.utcnow() + timedelta(days=config["due_in_days"])).isoformat(),
"escalation_date": (datetime.utcnow() + timedelta(days=config["due_in_days"] - config["escalation_days"])).isoformat(),
"reviews": [],
"statistics": {
"total_items": 0,
"reviewed": 0,
"certified": 0,
"revoked": 0,
"escalated": 0
}
}
# Identificar las asignaciones de acceso para su revisión
review_items = self._get_review_items(campaign)
campaign["statistics"]["total_items"] = len(review_items)
# Asignar revisores
for item in review_items:
reviewer = self._assign_reviewer(item, config["reviewer_type"])
item["reviewer"] = reviewer
item["review_id"] = str(uuid.uuid4())
item["decision"] = None
item["reviewed_at"] = None
campaign["reviews"] = review_items
campaign["status"] = "active"
self.campaigns[campaign_id] = campaign
# Enviar notificaciones iniciales
self._send_campaign_notifications(campaign)
return campaign
def _get_review_items(self, campaign):
"""Obtener los elementos de revisión según el alcance de la campaña."""
scope = campaign["config"]["scope"]
scope_value = campaign["config"].get("scope_value")
if scope == "all":
assignments = self.identity_store.get_all_assignments()
elif scope == "department":
assignments = self.identity_store.get_assignments_by_department(scope_value)
elif scope == "role":
assignments = self.identity_store.get_assignments_by_role(scope_value)
else:
raise ValueError(f"Alcance de campaña no soportado: {scope}")
return assignments
def submit_review_decision(self, review_id, decision, comments=""):
"""Registrar la decisión de un revisor sobre un elemento de revisión."""
for campaign_id, campaign in self.campaigns.items():
for item in campaign["reviews"]:
if item["review_id"] == review_id:
item["decision"] = decision
item["comments"] = comments
item["reviewed_at"] = datetime.utcnow().isoformat()
if decision == "certify":
campaign["statistics"]["certified"] += 1
elif decision == "revoke":
campaign["statistics"]["revoked"] += 1
self._queue_access_revocation(item)
campaign["statistics"]["reviewed"] += 1
# Verificar si la campaña está completa
if campaign["statistics"]["reviewed"] >= campaign["statistics"]["total_items"]:
campaign["status"] = "completed"
self._generate_campaign_report(campaign)
return True
return False
def process_escalations(self):
"""Manejar elementos de revisión vencidos: notificar y escalar."""
now = datetime.utcnow()
for campaign_id, campaign in self.campaigns.items():
if campaign["status"] != "active":
continue
escalation_date = datetime.fromisoformat(campaign["escalation_date"])
if now >= escalation_date:
for item in campaign["reviews"]:
if item["decision"] is None:
# Escalar al gerente del revisor
manager = self.identity_store.get_manager(item["reviewer"])
self.notification_service.notify(
manager,
f"Revisión de acceso atrasada: {item}"
)
campaign["statistics"]["escalated"] += 1
# Auto-revocar si está vencido y configurado
if campaign["config"].get("auto_revoke"):
due_date = datetime.fromisoformat(campaign["due_date"])
if now >= due_date:
for item in campaign["reviews"]:
if item["decision"] is None:
self._revoke_access(item)
item["decision"] = "auto_revoked"
campaign["statistics"]["revoked"] += 1
campaign["statistics"]["reviewed"] += 1
campaign["status"] = "completed"
Mejores Prácticas
Gestión de Datos de Identidad
- Mantener RRHH como fuente única de verdad: Todos los eventos del ciclo de vida deben originarse en el sistema de RRHH; no permitir cambios de identidad desde fuentes secundarias
- Implementar marco de confianza cero: Para sensibilidades altas, requerir re-autenticación incluso para cambios de rol aprobados
- Establecer acuerdos de nivel de servicio (SLA) claros: Definir plazos de aprovisionamiento aceptables para cada tipo de evento (por ejemplo, desactivación de cuenta en baja en <5 minutos)
Integración con Sistemas de Destino
- Usar conectores estándar de la industria (SCIM, LDAP, PowerShell remoto) para aplicaciones compatibles
- Desarrollar conectores personalizados para aplicaciones heredadas que no soportan protocolos modernos
- Implementar registro de aprovisionamiento con seguimiento de éxito/fallo por sistema de destino
Seguridad y Auditoría
- Registrar todas las acciones de aprovisionamiento con marca de tiempo, actor y resultado
- Implementar reversión automática si el aprovisionamiento falla después de cambios parciales
- Cifrar todos los datos de identidad en reposo y en tránsito
- Realizar revisiones trimestrales de acceso para aplicaciones críticas para el negocio
Consideraciones de Cumplimiento
| Estándar | Requisito Clave | Implementación |
|---|---|---|
| SOX | Segregación de funciones (SoD) | Definir políticas de SoD en el motor de solicitudes de acceso |
| HIPAA | Control de acceso mínimo necesario | Asignar roles con permisos mínimos según el rol laboral |
| GDPR | Derecho de supresión (olvido) | Implementar flujo de trabajo de eliminación completa de identidad con retención de registros de auditoría |
| PCI DSS | Revocación de acceso en baja inmediata | Priorizar la desactivación de cuenta como acción de alta urgencia |
Consejos para el Éxito
- Comenzar con un POC con un único sistema de destino y una población acotada de usuarios
- Involucrar a los dueños de aplicaciones temprano en el proceso de minería de roles
- Automatizar las revisiones de acceso para usuarios privilegiados y directivos primero
- Establecer un proceso de excepción para desviaciones justificadas de las asignaciones de roles estándar
- Monitorear el rendimiento del aprovisionamiento y configurar alertas para cuellos de botella
- Realizar pruebas de conciliación periódicas entre el IGA y los sistemas de destino
Resolución de Problemas
Los eventos del ciclo de vida no se procesan
- Verificar la conectividad con la API de RRHH (prueba de ping, renovar tokens OAuth)
- Validar que el formato de datos coincida entre el sistema de RRHH y el mapeo del IGA
- Revisar los registros del conector en busca de errores de mapeo (especialmente formatos de fecha)
El aprovisionamiento falla para sistemas de destino
- Confirmar que las credenciales de la cuenta de servicio tengan los permisos adecuados en el sistema de destino
- Verificar que los nombres de los sistemas de destino existan y sean accesibles en la red
- Probar el aprovisionamiento manualmente usando las mismas API para aislar el problema
Los roles minados no reflejan la realidad
- Aumentar el tamaño de la muestra de datos de acceso (mínimo 100 usuarios por código de puesto)
- Ajustar el porcentaje de asignación requerido (comenzar con 0,7 para minimizar falsos positivos)
- Revisar y excluir roles de servicio o cuentas de sistema de los datos de minería