Identity Governance & Lifecycle

What Are We Building?

Automated Identity Governance & Administration (IGA) Platform

🔄

Identity Lifecycle Automation

Build an enterprise-grade IGA system with automated provisioning, access reviews, and compliance reporting - the skills that define Security Architects.

👤+

Joiner

New hire onboarding
Auto-provision access

→

↔️

Mover

Role change/transfer
Adjust entitlements

→

👤−

Leaver

Offboarding
Revoke all access

🏢 Enterprise IGA Systems Cost $500K+

SailPoint, Saviynt, One Identity - these platforms cost hundreds of thousands annually. You'll build the core concepts using open-source tools:

Enterprise Feature	What You'll Build	Skills Transfer
User Provisioning (SCIM)	Authentik SCIM Provider → Apps	SailPoint IdentityIQ (95%)
Access Reviews/Certifications	Custom Review Dashboard	Saviynt Access Reviews (90%)
Joiner-Mover-Leaver	Automated Workflows	One Identity Manager (90%)
Segregation of Duties (SoD)	Role Conflict Detection	SAP GRC (85%)
Compliance Reporting	Audit Trail Dashboard	Any IGA Platform (95%)

IGA Architecture Flow

📋
HR System
Source of Truth

→

🔄
IGA Engine
Authentik

→

📡
SCIM 2.0
Provisioning

→

📱
Target Apps
SaaS/On-Prem

✅
Access Reviews
Periodic Certification

←

Manager Approval Loop

⏱️

Time Investment

Week	Focus Area	Time
Week 1	Core Concepts + SCIM Setup	8 hours
Week 2	Joiner-Mover-Leaver Workflows	10 hours
Week 3	Access Reviews Dashboard	12 hours
Week 4	SoD Detection + Testing	10 hours

Total: 3-4 weeks (40 hours)

Core Concepts

Understanding Identity Governance & Administration

🧠

IGA = IAM + Governance

IGA extends IAM with compliance, certification, and lifecycle automation

📡

SCIM 2.0

System for Cross-domain Identity Management. Standard protocol for automated user provisioning.

🔄

JML Workflows

Joiner-Mover-Leaver automation. Birthright access on hire, adjust on transfer, revoke on exit.

✅

Access Reviews

Periodic certification that users still need their access. Required for SOX, PCI, HIPAA.

⚠️

Segregation of Duties

Prevent toxic role combinations. No one should both create and approve payments.

👥

Role Mining

Analyze existing access patterns to discover and define roles automatically.

📊

Entitlement Management

Fine-grained permissions beyond roles. Application-specific access rights.

📡

SCIM 2.0 Protocol Explained

SCIM is REST-based and uses JSON. Here's what happens when a user is created:

📝 SCIM User Creation Request

// POST /scim/v2/Users
// Creates a new user in the target application
{
  "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"],
  "userName": "jsmith@company.com",
  "name": {
    "givenName": "John",
    "familyName": "Smith"
  },
  "emails": [{
    "value": "jsmith@company.com",
    "type": "work",
    "primary": true
  }],
  "active": true,
  "groups": [
    {"value": "engineering-team"}
  ]
}

SCIM Operation	HTTP Method	IGA Action
Create User	POST /Users	Joiner workflow
Update User	PATCH /Users/{id}	Mover workflow
Disable User	PATCH /Users/{id}	Leaver workflow
Delete User	DELETE /Users/{id}	Full deprovisioning
List Users	GET /Users	Reconciliation

🏢

Why IGA is an Architect-Level Skill

IGA sits at the intersection of security, compliance, and business operations. Understanding it demonstrates you can:

Design enterprise identity architectures
Translate compliance requirements into technical controls
Automate manual processes that cost companies millions
Reduce audit findings and security incidents

Prerequisites

What you need before starting

📋

Build on Project A Foundation

This project extends your Authentik SSO Gateway with governance capabilities

✅

Required: Complete Project A First

You need a working Authentik instance from Project A (Zero-Trust SSO Gateway):

Authentik deployed and running
At least one application configured
Test users and groups created
Basic understanding of Authentik admin interface

🛠️

Additional Tools for This Project

Tool	Purpose	Install Command
Python 3.10+	Workflow scripts & dashboard	apt install python3
Flask	Web dashboard framework	pip install flask
SQLite	Access review database	Built into Python
Postman/curl	SCIM API testing	Already installed

💻 Install Python Dependencies

# ============================================
# INSTALL PYTHON DEPENDENCIES
# For IGA dashboard and workflows
# ============================================

# Create project directory
mkdir -p ~/iga-lab
cd ~/iga-lab

# Create virtual environment
python3 -m venv venv
source venv/bin/activate

# Install required packages
pip install flask requests python-dotenv

# Create requirements.txt
cat > requirements.txt << 'EOF'
flask==3.0.0
requests==2.31.0
python-dotenv==1.0.0
EOF

echo "Dependencies installed!"

SCIM Provider Setup

Configure Authentik as a SCIM identity source

📡

Automated User Provisioning

SCIM automatically creates, updates, and disables users in target applications

Create SCIM Provider in Authentik

⏱️ 15 min

Open Authentik Admin: https://authentik.yourdomain.com/if/admin/
Navigate to: Applications → Providers
Click Create → Select SCIM Provider

Configure:

Name	IGA-SCIM-Provider
URL	Target app's SCIM endpoint (we'll create one)
Token	Generate secure token
Filter Group	Select users to provision

💡

SCIM Provider vs SCIM Source

Provider: Authentik pushes users TO other apps (outbound)
Source: Authentik receives users FROM other systems (inbound)

Create Mock SCIM Server (for Testing)

⏱️ 30 min

Build a simple Flask app that receives SCIM requests - simulates a SaaS application:

🐍 scim_server.py - Mock SCIM Endpoint

#!/usr/bin/env python3
"""
============================================
MOCK SCIM 2.0 SERVER
Simulates a SaaS application receiving
provisioning requests from Authentik
============================================
"""

from flask import Flask, request, jsonify
import sqlite3
import uuid
from datetime import datetime

app = Flask(__name__)

# ============================================
# DATABASE SETUP
# SQLite for storing provisioned users
# ============================================

def init_db():
    """Initialize the database with users table"""
    conn = sqlite3.connect('scim_users.db')
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id TEXT PRIMARY KEY,
            userName TEXT UNIQUE NOT NULL,
            givenName TEXT,
            familyName TEXT,
            email TEXT,
            active INTEGER DEFAULT 1,
            created_at TEXT,
            updated_at TEXT
        )
    ''')
    conn.commit()
    conn.close()

# ============================================
# SCIM ENDPOINTS
# ============================================

@app.route('/scim/v2/Users', methods=['POST'])
def create_user():
    """
    SCIM Create User (POST /Users)
    Called when Authentik provisions a new user
    """
    data = request.get_json()
    
    # Generate unique SCIM ID
    user_id = str(uuid.uuid4())
    
    # Extract user attributes from SCIM request
    user_name = data.get('userName')
    name = data.get('name', {})
    emails = data.get('emails', [])
    
    # Insert into database
    conn = sqlite3.connect('scim_users.db')
    cursor = conn.cursor()
    
    try:
        cursor.execute('''
            INSERT INTO users (id, userName, givenName, familyName, email, active, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, 1, ?, ?)
        ''', (
            user_id,
            user_name,
            name.get('givenName', ''),
            name.get('familyName', ''),
            emails[0].get('value', '') if emails else '',
            datetime.utcnow().isoformat(),
            datetime.utcnow().isoformat()
        ))
        conn.commit()
        
        # Log the provisioning event
        print(f"✅ JOINER: Created user {user_name} (ID: {user_id})")
        
        # Return SCIM-compliant response
        return jsonify({
            "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"],
            "id": user_id,
            "userName": user_name,
            "name": name,
            "emails": emails,
            "active": True,
            "meta": {
                "resourceType": "User",
                "created": datetime.utcnow().isoformat(),
                "location": f"/scim/v2/Users/{user_id}"
            }
        }), 201
        
    except sqlite3.IntegrityError:
        return jsonify({"error": "User already exists"}), 409
    finally:
        conn.close()


@app.route('/scim/v2/Users/<user_id>', methods=['PATCH'])
def update_user(user_id):
    """
    SCIM Update User (PATCH /Users/{id})
    Called for Mover (role change) or Leaver (disable) events
    """
    data = request.get_json()
    operations = data.get('Operations', [])
    
    conn = sqlite3.connect('scim_users.db')
    cursor = conn.cursor()
    
    for op in operations:
        if op.get('op') == 'replace':
            path = op.get('path')
            value = op.get('value')
            
            # Handle user disable (Leaver workflow)
            if path == 'active':
                cursor.execute(
                    'UPDATE users SET active = ?, updated_at = ? WHERE id = ?',
                    (1 if value else 0, datetime.utcnow().isoformat(), user_id)
                )
                action = "LEAVER" if not value else "REACTIVATED"
                print(f"⚠️ {action}: User {user_id} active={value}")
    
    conn.commit()
    conn.close()
    
    return jsonify({"status": "updated"}), 200


@app.route('/scim/v2/Users/<user_id>', methods=['DELETE'])
def delete_user(user_id):
    """
    SCIM Delete User (DELETE /Users/{id})
    Full deprovisioning - removes user completely
    """
    conn = sqlite3.connect('scim_users.db')
    cursor = conn.cursor()
    cursor.execute('DELETE FROM users WHERE id = ?', (user_id,))
    conn.commit()
    conn.close()
    
    print(f"🗑️ DELETED: User {user_id} fully deprovisioned")
    return '', 204


@app.route('/scim/v2/Users', methods=['GET'])
def list_users():
    """
    SCIM List Users (GET /Users)
    Used for reconciliation - verify sync status
    """
    conn = sqlite3.connect('scim_users.db')
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM users')
    users = cursor.fetchall()
    conn.close()
    
    resources = []
    for user in users:
        resources.append({
            "id": user[0],
            "userName": user[1],
            "name": {"givenName": user[2], "familyName": user[3]},
            "emails": [{"value": user[4]}] if user[4] else [],
            "active": bool(user[5])
        })
    
    return jsonify({
        "schemas": ["urn:ietf:params:scim:api:messages:2.0:ListResponse"],
        "totalResults": len(resources),
        "Resources": resources
    })


if __name__ == '__main__':
    init_db()
    print("🚀 Mock SCIM Server running on http://localhost:5001")
    app.run(host='0.0.0.0', port=5001, debug=True)

Test SCIM Provisioning

⏱️ 10 min

🧪 Test SCIM with curl

# Start the mock SCIM server
python3 scim_server.py &

# Test creating a user (Joiner)
curl -X POST http://localhost:5001/scim/v2/Users \
  -H "Content-Type: application/json" \
  -d '{
    "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"],
    "userName": "jsmith@company.com",
    "name": {"givenName": "John", "familyName": "Smith"},
    "emails": [{"value": "jsmith@company.com", "primary": true}],
    "active": true
  }'

# List all users
curl http://localhost:5001/scim/v2/Users | python3 -m json.tool

# Disable user (Leaver)
curl -X PATCH http://localhost:5001/scim/v2/Users/{USER_ID} \
  -H "Content-Type: application/json" \
  -d '{
    "schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
    "Operations": [{"op": "replace", "path": "active", "value": false}]
  }'

Joiner-Mover-Leaver Workflows

Automating the identity lifecycle

🔄

HR-Driven Identity Automation

When HR updates an employee record, access automatically adjusts

Create JML Workflow Engine

⏱️ 45 min

This script simulates HR events and triggers appropriate provisioning actions:

🐍 jml_workflow.py - Lifecycle Automation

#!/usr/bin/env python3
"""
============================================
JOINER-MOVER-LEAVER WORKFLOW ENGINE
Automates identity lifecycle based on HR events
============================================
"""

import requests
import json
from datetime import datetime
from dataclasses import dataclass
from typing import List, Dict

# SCIM server configuration
SCIM_BASE_URL = "http://localhost:5001/scim/v2"
SCIM_TOKEN = "your-scim-token"  # From Authentik provider


@dataclass
class Employee:
    """Represents an HR employee record"""
    employee_id: str
    email: str
    first_name: str
    last_name: str
    department: str
    job_title: str
    manager_email: str
    start_date: str
    end_date: str = None


# ============================================
# BIRTHRIGHT ACCESS MAPPING
# Maps departments to automatic group membership
# ============================================

BIRTHRIGHT_ACCESS = {
    "Engineering": ["engineering-team", "github-access", "jira-users"],
    "Finance": ["finance-team", "sap-users", "expense-system"],
    "HR": ["hr-team", "workday-admin", "benefits-portal"],
    "Sales": ["sales-team", "salesforce-users", "zoom-pro"],
    "IT": ["it-team", "admin-tools", "aws-console"],
}

# All employees get these
BASELINE_ACCESS = ["all-employees", "email-access", "slack-users"]


class JMLWorkflowEngine:
    """
    Handles Joiner-Mover-Leaver lifecycle events
    """
    
    def __init__(self, scim_url: str, token: str):
        self.scim_url = scim_url
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }
        self.audit_log = []
    
    def log_event(self, event_type: str, employee: Employee, details: str):
        """Record audit trail for compliance"""
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "event_type": event_type,
            "employee_id": employee.employee_id,
            "email": employee.email,
            "details": details
        }
        self.audit_log.append(entry)
        print(f"📝 AUDIT: {event_type} - {employee.email} - {details}")
    
    # ============================================
    # JOINER WORKFLOW
    # ============================================
    
    def process_joiner(self, employee: Employee) -> bool:
        """
        Process new hire onboarding
        1. Create user account
        2. Assign birthright access based on department
        3. Notify manager
        """
        print(f"\n{'='*50}")
        print(f"👤+ JOINER WORKFLOW: {employee.first_name} {employee.last_name}")
        print(f"{'='*50}")
        
        # Determine access groups
        groups = BASELINE_ACCESS.copy()
        dept_groups = BIRTHRIGHT_ACCESS.get(employee.department, [])
        groups.extend(dept_groups)
        
        # Create SCIM user
        scim_user = {
            "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"],
            "userName": employee.email,
            "name": {
                "givenName": employee.first_name,
                "familyName": employee.last_name
            },
            "emails": [{"value": employee.email, "primary": True}],
            "active": True,
            "title": employee.job_title,
            "department": employee.department
        }
        
        try:
            response = requests.post(
                f"{self.scim_url}/Users",
                headers=self.headers,
                json=scim_user
            )
            
            if response.status_code == 201:
                self.log_event("JOINER", employee, 
                    f"Created with groups: {', '.join(groups)}")
                print(f"✅ User created successfully")
                print(f"✅ Assigned groups: {groups}")
                return True
            else:
                print(f"❌ Failed: {response.text}")
                return False
                
        except Exception as e:
            print(f"❌ Error: {e}")
            return False
    
    # ============================================
    # MOVER WORKFLOW
    # ============================================
    
    def process_mover(self, employee: Employee, old_dept: str, new_dept: str) -> bool:
        """
        Process role change / department transfer
        1. Remove old department access
        2. Add new department access
        3. Keep baseline access
        """
        print(f"\n{'='*50}")
        print(f"↔️ MOVER WORKFLOW: {employee.email}")
        print(f"   {old_dept} → {new_dept}")
        print(f"{'='*50}")
        
        # Determine access changes
        old_groups = BIRTHRIGHT_ACCESS.get(old_dept, [])
        new_groups = BIRTHRIGHT_ACCESS.get(new_dept, [])
        
        groups_to_remove = set(old_groups) - set(new_groups)
        groups_to_add = set(new_groups) - set(old_groups)
        
        print(f"🔴 Removing: {groups_to_remove}")
        print(f"🟢 Adding: {groups_to_add}")
        
        self.log_event("MOVER", employee,
            f"Transferred {old_dept} → {new_dept}. Removed: {groups_to_remove}, Added: {groups_to_add}")
        
        return True
    
    # ============================================
    # LEAVER WORKFLOW
    # ============================================
    
    def process_leaver(self, employee: Employee, user_id: str) -> bool:
        """
        Process employee termination
        1. Disable account immediately
        2. Revoke all access
        3. Archive for compliance
        """
        print(f"\n{'='*50}")
        print(f"👤− LEAVER WORKFLOW: {employee.email}")
        print(f"{'='*50}")
        
        # SCIM PATCH to disable
        patch_data = {
            "schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
            "Operations": [
                {"op": "replace", "path": "active", "value": False}
            ]
        }
        
        try:
            response = requests.patch(
                f"{self.scim_url}/Users/{user_id}",
                headers=self.headers,
                json=patch_data
            )
            
            if response.status_code == 200:
                self.log_event("LEAVER", employee,
                    "Account disabled, all access revoked")
                print(f"✅ Account disabled")
                print(f"✅ All groups removed")
                return True
            else:
                print(f"❌ Failed: {response.text}")
                return False
                
        except Exception as e:
            print(f"❌ Error: {e}")
            return False


# ============================================
# DEMO: Simulate HR Events
# ============================================

if __name__ == "__main__":
    engine = JMLWorkflowEngine(SCIM_BASE_URL, SCIM_TOKEN)
    
    # Simulate new hire
    new_employee = Employee(
        employee_id="EMP001",
        email="jsmith@company.com",
        first_name="John",
        last_name="Smith",
        department="Engineering",
        job_title="Software Engineer",
        manager_email="manager@company.com",
        start_date="2024-01-15"
    )
    
    # Process joiner
    engine.process_joiner(new_employee)
    
    # Simulate department transfer
    engine.process_mover(new_employee, "Engineering", "IT")
    
    # Print audit log
    print("\n📋 AUDIT LOG:")
    for entry in engine.audit_log:
        print(json.dumps(entry, indent=2))

🏢

Enterprise JML Best Practices

Joiner: Provision 24-48 hours before start date
Mover: Remove old access same day, add new next day
Leaver: Disable within 1 hour of termination notice
Audit: Log everything for 7 years (SOX requirement)

Access Reviews Dashboard

Periodic access certification for compliance

✅

Manager Certification Campaigns

"Does John Smith still need access to the Finance system?" - Managers answer quarterly

847

Certified

123

Pending

Revoked

94%

Completion

Create Access Review Web Dashboard

⏱️ 60 min

🐍 access_review_app.py - Flask Dashboard

#!/usr/bin/env python3
"""
============================================
ACCESS REVIEW DASHBOARD
Web interface for manager certification campaigns
============================================
"""

from flask import Flask, render_template_string, request, jsonify
import sqlite3
from datetime import datetime, timedelta
import uuid

app = Flask(__name__)

# ============================================
# DATABASE SETUP
# ============================================

def init_review_db():
    """Initialize access review database"""
    conn = sqlite3.connect('access_reviews.db')
    cursor = conn.cursor()
    
    # Access review campaigns
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS campaigns (
            id TEXT PRIMARY KEY,
            name TEXT NOT NULL,
            description TEXT,
            start_date TEXT,
            end_date TEXT,
            status TEXT DEFAULT 'active'
        )
    ''')
    
    # Individual review items
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS review_items (
            id TEXT PRIMARY KEY,
            campaign_id TEXT,
            user_email TEXT,
            user_name TEXT,
            application TEXT,
            access_level TEXT,
            reviewer_email TEXT,
            decision TEXT DEFAULT 'pending',
            decision_date TEXT,
            justification TEXT,
            FOREIGN KEY (campaign_id) REFERENCES campaigns(id)
        )
    ''')
    
    conn.commit()
    conn.close()


# ============================================
# HTML TEMPLATE (Embedded for simplicity)
# ============================================

DASHBOARD_HTML = '''
<!DOCTYPE html>
<html>
<head>
    <title>Access Review Dashboard</title>
    <style>
        body { 
            font-family: 'Segoe UI', sans-serif; 
            background: #0a0a1a; 
            color: #e8e8ff;
            margin: 0;
            padding: 20px;
        }
        .header { 
            text-align: center; 
            padding: 20px;
            border-bottom: 2px solid #14b8a6;
        }
        h1 { color: #14b8a6; }
        .stats { 
            display: flex; 
            justify-content: center; 
            gap: 20px;
            margin: 20px 0;
        }
        .stat-card {
            background: rgba(20, 184, 166, 0.1);
            border: 1px solid #14b8a6;
            border-radius: 10px;
            padding: 20px;
            text-align: center;
            min-width: 120px;
        }
        .stat-number { 
            font-size: 2rem; 
            font-weight: bold;
            color: #5eead4;
        }
        .stat-label { color: #6b6b8d; }
        table {
            width: 100%;
            border-collapse: collapse;
            margin-top: 20px;
        }
        th {
            background: rgba(20, 184, 166, 0.2);
            color: #14b8a6;
            padding: 12px;
            text-align: left;
        }
        td {
            padding: 12px;
            border-bottom: 1px solid rgba(20, 184, 166, 0.2);
        }
        .btn {
            padding: 8px 16px;
            border: none;
            border-radius: 5px;
            cursor: pointer;
            margin: 2px;
        }
        .btn-approve { background: #22c55e; color: white; }
        .btn-revoke { background: #ef4444; color: white; }
        .badge {
            padding: 4px 12px;
            border-radius: 20px;
            font-size: 0.85rem;
        }
        .badge-pending { background: rgba(234, 179, 8, 0.2); color: #eab308; }
        .badge-approved { background: rgba(34, 197, 94, 0.2); color: #22c55e; }
        .badge-revoked { background: rgba(239, 68, 68, 0.2); color: #ef4444; }
    </style>
</head>
<body>
    <div class="header">
        <h1>🔐 Access Review Dashboard</h1>
        <p>Q4 2024 Access Certification Campaign</p>
    </div>
    
    <div class="stats">
        <div class="stat-card">
            <div class="stat-number">{{ stats.total }}</div>
            <div class="stat-label">Total Reviews</div>
        </div>
        <div class="stat-card">
            <div class="stat-number">{{ stats.pending }}</div>
            <div class="stat-label">Pending</div>
        </div>
        <div class="stat-card">
            <div class="stat-number">{{ stats.approved }}</div>
            <div class="stat-label">Approved</div>
        </div>
        <div class="stat-card">
            <div class="stat-number">{{ stats.revoked }}</div>
            <div class="stat-label">Revoked</div>
        </div>
    </div>
    
    <table>
        <thead>
            <tr>
                <th>User</th>
                <th>Application</th>
                <th>Access Level</th>
                <th>Status</th>
                <th>Actions</th>
            </tr>
        </thead>
        <tbody>
            {% for item in reviews %}
            <tr>
                <td>{{ item.user_name }}<br><small>{{ item.user_email }}</small></td>
                <td>{{ item.application }}</td>
                <td>{{ item.access_level }}</td>
                <td>
                    <span class="badge badge-{{ item.decision }}">
                        {{ item.decision|upper }}
                    </span>
                </td>
                <td>
                    {% if item.decision == 'pending' %}
                    <button class="btn btn-approve" onclick="decide('{{ item.id }}', 'approved')">✓ Approve</button>
                    <button class="btn btn-revoke" onclick="decide('{{ item.id }}', 'revoked')">✗ Revoke</button>
                    {% else %}
                    <small>{{ item.decision_date }}</small>
                    {% endif %}
                </td>
            </tr>
            {% endfor %}
        </tbody>
    </table>
    
    <script>
    function decide(itemId, decision) {
        fetch('/api/review/' + itemId, {
            method: 'POST',
            headers: {'Content-Type': 'application/json'},
            body: JSON.stringify({decision: decision})
        }).then(() => location.reload());
    }
    </script>
</body>
</html>
'''


# ============================================
# ROUTES
# ============================================

@app.route('/')
def dashboard():
    """Main dashboard view"""
    conn = sqlite3.connect('access_reviews.db')
    cursor = conn.cursor()
    
    # Get all review items
    cursor.execute('SELECT * FROM review_items')
    rows = cursor.fetchall()
    
    reviews = []
    for row in rows:
        reviews.append({
            'id': row[0],
            'user_email': row[2],
            'user_name': row[3],
            'application': row[4],
            'access_level': row[5],
            'decision': row[7],
            'decision_date': row[8]
        })
    
    # Calculate stats
    stats = {
        'total': len(reviews),
        'pending': len([r for r in reviews if r['decision'] == 'pending']),
        'approved': len([r for r in reviews if r['decision'] == 'approved']),
        'revoked': len([r for r in reviews if r['decision'] == 'revoked'])
    }
    
    conn.close()
    return render_template_string(DASHBOARD_HTML, reviews=reviews, stats=stats)


@app.route('/api/review/<item_id>', methods=['POST'])
def make_decision(item_id):
    """Record manager's decision"""
    data = request.get_json()
    decision = data.get('decision')
    
    conn = sqlite3.connect('access_reviews.db')
    cursor = conn.cursor()
    cursor.execute('''
        UPDATE review_items 
        SET decision = ?, decision_date = ?
        WHERE id = ?
    ''', (decision, datetime.utcnow().isoformat(), item_id))
    conn.commit()
    conn.close()
    
    # If revoked, trigger deprovisioning
    if decision == 'revoked':
        print(f"🔴 ACCESS REVOKED: {item_id} - Triggering deprovisioning...")
    
    return jsonify({'status': 'success'})


def seed_sample_data():
    """Add sample review items for demo"""
    conn = sqlite3.connect('access_reviews.db')
    cursor = conn.cursor()
    
    # Check if already seeded
    cursor.execute('SELECT COUNT(*) FROM review_items')
    if cursor.fetchone()[0] > 0:
        conn.close()
        return
    
    # Sample data
    samples = [
        ('jsmith@company.com', 'John Smith', 'Salesforce', 'Admin'),
        ('jsmith@company.com', 'John Smith', 'GitHub', 'Write'),
        ('mjones@company.com', 'Mary Jones', 'SAP', 'Read'),
        ('mjones@company.com', 'Mary Jones', 'AWS Console', 'PowerUser'),
        ('bwilson@company.com', 'Bob Wilson', 'Jira', 'Admin'),
    ]
    
    for email, name, app, level in samples:
        cursor.execute('''
            INSERT INTO review_items (id, user_email, user_name, application, access_level, reviewer_email)
            VALUES (?, ?, ?, ?, ?, ?)
        ''', (str(uuid.uuid4()), email, name, app, level, 'manager@company.com'))
    
    conn.commit()
    conn.close()
    print("✅ Sample data seeded")


if __name__ == '__main__':
    init_review_db()
    seed_sample_data()
    print("🚀 Access Review Dashboard: http://localhost:5002")
    app.run(host='0.0.0.0', port=5002, debug=True)

💡

Run the Dashboard

python3 access_review_app.py
Then open http://localhost:5002 in your browser

Segregation of Duties (SoD)

Detecting toxic role combinations

⚠️

Prevent Fraud Through Role Separation

No single person should be able to initiate AND approve financial transactions

🔒

Common SoD Violations

Role A	Role B	Risk
Create Vendor	Approve Payments	High - Ghost vendor fraud
Create User	Assign Privileges	High - Privilege escalation
Develop Code	Deploy to Production	Medium - Unauthorized changes
Receive Inventory	Adjust Inventory	High - Theft concealment

Create SoD Detection Engine