IAM Application Risk Governance Pipeline

Introduction

30 Second Summary

Large enterprises run hundreds of applications, each needing secure identity connections to systems like single sign-on, privileged access, and lifecycle management. When those connections are scattered across multiple spreadsheets and registries with no single source of truth, security gaps hide in plain sight.

In this project, you will build a Python data pipeline that ingests application data from three enterprise sources, maps them to real IAM products, classifies risk, detects governance gaps, and produces audit-ready outputs including a professional HTML dashboard.

What You'll Build

A complete IAM Application Risk Governance Pipeline that consolidates scattered enterprise data into a unified governance map with risk classification, gap detection, and management-ready reporting.

By the end of this project, you'll have:

• A Python pipeline with modular architecture that ingests data from three sources and produces a unified, risk-classified IAM governance map modelling real enterprise tools like Microsoft Entra ID, SailPoint, and CyberArk.
• A YAML-based feeding rules engine defining authoritative sources, conflict resolution, remediation SLAs, and communication channels.
• Audit-ready outputs including a master map CSV, gap reports, department scorecards, a remediation priority queue, and a professional HTML dashboard.
• Secret Mission: Add a SCIM provisioning simulation modelling SailPoint-to-CyberArk identity lifecycle flows with drift detection.

Are there any prerequisites?

This project assumes you have Python experience and a conceptual understanding of IAM topics like SSO, MFA, PAM, IGA, and RBAC. You do not need hands-on experience with any specific IAM product.

Before We Start

Before We Start...

Set Up the Project Environment

Before you can build the governance pipeline, you need a clean workspace with the right tools installed.

In this step, you will create a dedicated project folder, set up an isolated Python environment, and install the three libraries that power the pipeline: pandas for data processing, Jinja2 for HTML templating, and PyYAML for parsing the feeding rules.

In this step, get ready to:

• Create a project directory and Python virtual environment.
• Install required packages (pandas, Jinja2, PyYAML).
• Create the project folder structure with all directories and starter files.

Create a project directory and virtual environment

A virtual environment keeps your pipeline's dependencies isolated from the rest of your system. This prevents version conflicts with other Python projects.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Visual Studio Code and press Enter to open it.
• Open the integrated terminal by pressing Ctrl+` (macOS and Windows).

First, confirm your Python version meets the requirement. This project uses pandas==3.0.3 which requires Python 3.11 or higher.

• Check your Python version by running:

python3 --version

✔️ I see version 3.11 or higher

Your Python version is compatible. Continue to the next instruction below.

ⓧ I see a version lower than 3.11

You need Python 3.11+ because pandas==3.0.3 does not support older versions. On macOS, the system Python is often outdated even after installing a newer version.

• Download and install Python 3.11 or higher from the official site: https://www.python.org/downloads/.
• After installation, close and reopen your terminal, then run python3 --version again to confirm the new version appears.
• On macOS, if the old version persists, try running python3.11 --version or python3.12 --version to use the specific binary you installed.

ⓧ Command not found

Python is not installed or not in your system PATH.

• Download and install Python 3.11 or higher from the official site: https://www.python.org/downloads/.
• On Windows, make sure to check the Add Python to PATH box during installation.
• Close and reopen VS Code after installation, then try python3 --version (macOS/Linux) or python --version (Windows) again.

• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir iam-governance-pipeline
cd iam-governance-pipeline

• Create a virtual environment by running:

python3 -m venv venv

• Activate the virtual environment by running the command for your operating system:

macOS

source venv/bin/activate

Windows

venv\Scripts\Activate.ps1

You should see (venv) appear at the beginning of your terminal prompt. This confirms you are working inside the virtual environment.

Not seeing (venv) in your prompt?

On Windows PowerShell, you may see an error about execution policies. Run Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser first, then try the activation command again.

On macOS, if python3 -m venv venv failed silently, check that Python 3.11+ is installed correctly. The venv folder must exist before you can activate it.

help me activate my Python virtual environment

Install the required packages

The pipeline relies on three libraries. Each handles a distinct layer of the system.

• Install all three packages by running:

pip install pandas==3.0.3 Jinja2==3.1.6 PyYAML==6.0.3

What do these packages do?

• pandas is the data processing engine. It loads CSVs, merges datasets, and produces the governance map.
• Jinja2 is a templating library. It renders the HTML governance dashboard from data.
• PyYAML parses the feeding_rules.yaml file that defines data ownership, conflict resolution, and SLAs.

You should see output confirming all three packages were downloaded and installed successfully.

Seeing a resolution error or build failure?

If pip cannot find a matching version of pandas==3.0.3, your Python version is below 3.11. Go back to the version check above and install Python 3.11 or higher.

Make sure (venv) is visible in your prompt. If not, the packages will install globally instead of in your virtual environment.

help me fix pip install errors for pandas, Jinja2, or PyYAML

Create the project folder structure

The pipeline uses a modular layout. Data sources live in data/, processing modules in src/, and all generated reports go to output/.

• Create the three directories by running:

mkdir data src output

• Create the root-level starter files and the Python package marker by running:

macOS

touch pipeline.py feeding_rules.yaml requirements.txt src/__init__.py

Windows

New-Item pipeline.py, feeding_rules.yaml, requirements.txt, src/__init__.py

Why create src/__init__.py?

The empty __init__.py file tells Python that the src/ folder is a package. Without it, pipeline.py would not be able to import modules like src.data_processor or src.risk_classifier.

• Now open requirements.txt in VS Code by clicking on it in the Explorer sidebar (left panel).
• Paste the following package pins into the file:

pandas==3.0.3
Jinja2==3.1.6
PyYAML==6.0.3

What does requirements.txt do?

This file records the exact package versions your project depends on. Anyone who clones your project later can run pip install -r requirements.txt to install the same versions you used.

• Save requirements.txt by pressing Cmd+S (macOS) or Ctrl+S (Windows).
• Verify the full project structure by running:

ls

You should see data, feeding_rules.yaml, output, pipeline.py, requirements.txt, src, and venv in the output.

Missing files or folders?

Double-check that you ran the mkdir and file creation commands from inside the iam-governance-pipeline folder. Run pwd (macOS) or cd (Windows) to see your current location.

If you see src/__init__.py is missing, make sure you included the path prefix src/ in the command.

help me fix my project folder structure

✔️ Awesome, I've got everything!

Great. Double check you have saved requirements.txt with the three package pins.

ⓧ I'd like to double check the full code

Here is what requirements.txt should contain:

pandas==3.0.3
Jinja2==3.1.6
PyYAML==6.0.3

The file src/__init__.py should be completely empty (zero bytes). Its presence alone is what matters.

Your workspace is ready. Next up, you will create the synthetic enterprise data sources that simulate a real company's application inventory, IAM product registry, and connection logs.

Create the Enterprise Data Sources

Your project structure is ready. Now you need the raw data that your pipeline will process.

In a real enterprise, application data lives in scattered systems. HR maintains one spreadsheet, the IAM team manages a product registry, and connection logs accumulate in a separate database. Your pipeline needs to consolidate all of these into a single governance map. In this step, you will create three synthetic data sources that mirror this real-world mess.

In this step, get ready to:

• Create an application inventory CSV with 75 enterprise applications across 14 departments.
• Build an IAM product registry JSON defining 5 real IAM products and their coverage requirements.
• Generate a connection log CSV with 255 records that includes realistic data quality issues.

Create the application inventory

The application inventory is the foundation of your governance map. It represents every business application across the enterprise that needs IAM coverage.

You will create a Python script that generates realistic synthetic data. This gives you control over the data quality issues the pipeline will need to handle later.

• Create a new file called generate_data.py in the root of your iam-governance-pipeline folder by running:

touch generate_data.py

• Open generate_data.py in VS Code and paste the following code to generate the application inventory:

import csv
import random
import json
from datetime import datetime, timedelta

# Set seed for reproducible data generation
random.seed(42)

# 14 departments representing a large enterprise
DEPARTMENTS = [
    'Finance', 'HR', 'IT Operations', 'Cybersecurity', 'Supply Chain',
    'Legal', 'Marketing', 'Customer Service', 'R&D', 'Engineering',
    'Procurement', 'Executive', 'Compliance', 'Data Analytics'
]

# Realistic enterprise application names grouped by department
APP_NAMES = {
    'Finance': ['SAP S/4HANA Finance', 'Oracle HFM', 'Bloomberg Terminal', 'Concur Expense', 'Kyriba Treasury', 'BlackLine'],
    'HR': ['Workday HCM', 'SAP SuccessFactors', 'ADP Workforce Now', 'Cornerstone OnDemand', 'ServiceNow HRSD'],
    'IT Operations': ['ServiceNow ITSM', 'Jira Service Management', 'PagerDuty', 'Datadog', 'Splunk Enterprise', 'Ansible Tower'],
    'Cybersecurity': ['CrowdStrike Falcon', 'Palo Alto Cortex', 'Qualys VMDR', 'Tenable.io', 'Rapid7 InsightVM'],
    'Supply Chain': ['SAP TMS', 'Oracle SCM Cloud', 'Blue Yonder WMS', 'Kinaxis RapidResponse', 'FourKites', 'project44'],
    'Legal': ['DocuSign CLM', 'Relativity', 'ContractPodAi', 'Thomson Reuters Legal', 'iManage Work'],
    'Marketing': ['Salesforce Marketing Cloud', 'Adobe Experience Platform', 'HubSpot', 'Marketo Engage', 'Google Analytics 360'],
    'Customer Service': ['Salesforce Service Cloud', 'Zendesk Enterprise', 'Genesys Cloud CX', 'NICE CXone', 'Five9'],
    'R&D': ['Jira Software', 'Confluence', 'GitLab Ultimate', 'SonarQube', 'Artifactory'],
    'Engineering': ['GitHub Enterprise', 'Jenkins', 'Terraform Cloud', 'AWS Console', 'Azure Portal', 'Docker Enterprise'],
    'Procurement': ['SAP Ariba', 'Coupa', 'Jaggaer', 'GEP SMART', 'Oracle Procurement Cloud'],
    'Executive': ['Tableau Server', 'Power BI Premium', 'Board M.I.', 'Anaplan', 'Domo'],
    'Compliance': ['Archer GRC', 'OneTrust', 'LogicGate', 'Diligent', 'MetricStream'],
    'Data Analytics': ['Snowflake', 'Databricks', 'Palantir Foundry', 'Alteryx', 'ThoughtSpot']
}

SENSITIVITY_LEVELS = ['HIGH', 'MEDIUM', 'LOW']
ENVIRONMENTS = ['Production', 'Staging', 'Development']
REGIONS = ['EMEA', 'APAC', 'Americas', 'Global']
AUTH_PROTOCOLS = ['SAML', 'OIDC', 'LDAP', 'Local']
DATA_CLASSIFICATIONS = ['Confidential', 'Internal', 'Public', 'Restricted']


def generate_application_inventory():
    """Generate 75 synthetic applications across 14 departments."""
    apps = []
    app_id = 1

    for dept in DEPARTMENTS:
        dept_apps = APP_NAMES[dept]
        for app_name in dept_apps:
            # Assign realistic sensitivity based on department
            if dept in ('Finance', 'Cybersecurity', 'Executive', 'Compliance'):
                sensitivity = random.choices(SENSITIVITY_LEVELS, weights=[0.6, 0.3, 0.1])[0]
            elif dept in ('HR', 'Legal', 'Engineering'):
                sensitivity = random.choices(SENSITIVITY_LEVELS, weights=[0.3, 0.5, 0.2])[0]
            else:
                sensitivity = random.choices(SENSITIVITY_LEVELS, weights=[0.2, 0.4, 0.4])[0]

            # Business critical apps are more likely in Production
            environment = random.choices(ENVIRONMENTS, weights=[0.6, 0.2, 0.2])[0]
            business_critical = (environment == 'Production' and sensitivity == 'HIGH') or random.random() < 0.3

            # Some apps intentionally have missing owners
            if random.random() < 0.08:
                owner = ''
            else:
                first_names = ['Sarah', 'James', 'Maria', 'David', 'Priya', 'Thomas', 'Aisha', 'Robert', 'Lin', 'Carlos']
                last_names = ['Chen', 'Smith', 'Dubois', 'Nakamura', 'Patel', 'Wilson', 'Hassan', 'Garcia', 'Kim', 'Andersen']
                owner = f"{random.choice(first_names)} {random.choice(last_names)}"

            apps.append({
                'app_id': f"APP-{app_id:03d}",
                'app_name': app_name,
                'department': dept,
                'owner': owner,
                'sensitivity': sensitivity,
                'business_critical': business_critical,
                'environment': environment,
                'region': random.choice(REGIONS),
                'authentication_protocol': random.choice(AUTH_PROTOCOLS),
                'data_classification': random.choice(DATA_CLASSIFICATIONS)
            })
            app_id += 1

    return apps


def write_application_inventory(apps, filepath='data/application_inventory.csv'):
    """Write the application inventory to CSV."""
    fieldnames = ['app_id', 'app_name', 'department', 'owner', 'sensitivity',
                  'business_critical', 'environment', 'region',
                  'authentication_protocol', 'data_classification']

    with open(filepath, 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(apps)

    print(f"Generated {len(apps)} applications in {filepath}")


if __name__ == '__main__':
    apps = generate_application_inventory()
    write_application_inventory(apps)

What does this code do?

This script generates realistic enterprise application data. Each of the 75 applications gets assigned a department, owner, sensitivity level, and environment.

The sensitivity weights are tuned by department. Finance and Cybersecurity apps skew HIGH because they handle sensitive data. Marketing and Customer Service apps skew MEDIUM or LOW. Some applications intentionally have blank owners to create governance gaps the pipeline will detect later.

• Save generate_data.py.
• Run the script to generate the application inventory:

python generate_data.py

• You should see output confirming 75 applications were generated in data/application_inventory.csv.

Seeing a FileNotFoundError?

Make sure your data/ directory exists. You created it in Step 1. If it is missing, run mkdir data in your terminal and try again.

help me fix a FileNotFoundError when running my generate_data.py script

Create the IAM product registry

The IAM product registry defines which identity products your enterprise uses and what coverage they require. Each product specifies which sensitivity levels it must protect.

• Open data/iam_product_registry.json in VS Code and paste the following:

{
    "products": [
        {
            "product_id": "entra_id",
            "product_name": "Microsoft Entra ID",
            "category": "SSO/IdP",
            "required_for_sensitivity": ["HIGH", "MEDIUM", "LOW"],
            "protocols": ["SAML", "OIDC"],
            "review_cycle_days": 180,
            "description": "Primary identity provider for SSO across all enterprise applications"
        },
        {
            "product_id": "sailpoint_isc",
            "product_name": "SailPoint Identity Security Cloud",
            "category": "IGA",
            "required_for_sensitivity": ["HIGH", "MEDIUM"],
            "protocols": ["SCIM", "SAML"],
            "review_cycle_days": 90,
            "description": "Identity Governance and Administration for lifecycle management and access reviews"
        },
        {
            "product_id": "cyberark_pc",
            "product_name": "CyberArk Privilege Cloud",
            "category": "PAM",
            "required_for_sensitivity": ["HIGH"],
            "protocols": ["SCIM", "SAML", "OIDC"],
            "review_cycle_days": 90,
            "description": "Privileged Access Management for securing high-sensitivity administrative access"
        },
        {
            "product_id": "enterprise_mfa",
            "product_name": "Enterprise MFA",
            "category": "MFA",
            "required_for_sensitivity": ["HIGH", "MEDIUM"],
            "protocols": ["SAML", "OIDC"],
            "review_cycle_days": 180,
            "description": "Multi-factor authentication enforcement for elevated-risk applications"
        },
        {
            "product_id": "rbac_engine",
            "product_name": "RBAC Engine",
            "category": "Access Control",
            "required_for_sensitivity": ["HIGH", "MEDIUM", "LOW"],
            "protocols": ["LDAP", "SCIM"],
            "review_cycle_days": 365,
            "description": "Role-Based Access Control engine for consistent permission assignment"
        }
    ]
}

What do these products cover?

• Microsoft Entra ID is the SSO identity provider. All apps must connect regardless of sensitivity.
• SailPoint Identity Security Cloud handles identity governance. HIGH and MEDIUM apps need lifecycle management.
• CyberArk Privilege Cloud secures privileged access. Only HIGH sensitivity apps require it.
• Enterprise MFA adds step-up authentication for HIGH and MEDIUM apps.
• RBAC Engine enforces role-based permissions across all sensitivity levels.

The required_for_sensitivity field is what drives gap detection. If a HIGH-sensitivity app is not connected to CyberArk Privilege Cloud, the pipeline flags it as a MISSING_CONNECTION gap.

• Save data/iam_product_registry.json.

JSON syntax error later?

If you see a JSON decode error when running the pipeline later, check for trailing commas after the last item in the products array. JSON does not allow trailing commas like Python does.

help me fix a JSON parsing error in my iam_product_registry.json file

Create the connection log

The connection log records which applications are actually connected to which IAM products. In a real enterprise, this data accumulates over time and develops quality issues. Stale reviews, missing dates, and unknown reviewers are all common.

• Return to generate_data.py and replace the entire file with the following updated version that adds connection log generation:

import csv
import random
import json
from datetime import datetime, timedelta

# Set seed for reproducible data generation
random.seed(42)

# 14 departments representing a large enterprise
DEPARTMENTS = [
    'Finance', 'HR', 'IT Operations', 'Cybersecurity', 'Supply Chain',
    'Legal', 'Marketing', 'Customer Service', 'R&D', 'Engineering',
    'Procurement', 'Executive', 'Compliance', 'Data Analytics'
]

# Realistic enterprise application names grouped by department
APP_NAMES = {
    'Finance': ['SAP S/4HANA Finance', 'Oracle HFM', 'Bloomberg Terminal', 'Concur Expense', 'Kyriba Treasury', 'BlackLine'],
    'HR': ['Workday HCM', 'SAP SuccessFactors', 'ADP Workforce Now', 'Cornerstone OnDemand', 'ServiceNow HRSD'],
    'IT Operations': ['ServiceNow ITSM', 'Jira Service Management', 'PagerDuty', 'Datadog', 'Splunk Enterprise', 'Ansible Tower'],
    'Cybersecurity': ['CrowdStrike Falcon', 'Palo Alto Cortex', 'Qualys VMDR', 'Tenable.io', 'Rapid7 InsightVM'],
    'Supply Chain': ['SAP TMS', 'Oracle SCM Cloud', 'Blue Yonder WMS', 'Kinaxis RapidResponse', 'FourKites', 'project44'],
    'Legal': ['DocuSign CLM', 'Relativity', 'ContractPodAi', 'Thomson Reuters Legal', 'iManage Work'],
    'Marketing': ['Salesforce Marketing Cloud', 'Adobe Experience Platform', 'HubSpot', 'Marketo Engage', 'Google Analytics 360'],
    'Customer Service': ['Salesforce Service Cloud', 'Zendesk Enterprise', 'Genesys Cloud CX', 'NICE CXone', 'Five9'],
    'R&D': ['Jira Software', 'Confluence', 'GitLab Ultimate', 'SonarQube', 'Artifactory'],
    'Engineering': ['GitHub Enterprise', 'Jenkins', 'Terraform Cloud', 'AWS Console', 'Azure Portal', 'Docker Enterprise'],
    'Procurement': ['SAP Ariba', 'Coupa', 'Jaggaer', 'GEP SMART', 'Oracle Procurement Cloud'],
    'Executive': ['Tableau Server', 'Power BI Premium', 'Board M.I.', 'Anaplan', 'Domo'],
    'Compliance': ['Archer GRC', 'OneTrust', 'LogicGate', 'Diligent', 'MetricStream'],
    'Data Analytics': ['Snowflake', 'Databricks', 'Palantir Foundry', 'Alteryx', 'ThoughtSpot']
}

SENSITIVITY_LEVELS = ['HIGH', 'MEDIUM', 'LOW']
ENVIRONMENTS = ['Production', 'Staging', 'Development']
REGIONS = ['EMEA', 'APAC', 'Americas', 'Global']
AUTH_PROTOCOLS = ['SAML', 'OIDC', 'LDAP', 'Local']
DATA_CLASSIFICATIONS = ['Confidential', 'Internal', 'Public', 'Restricted']

IAM_PRODUCTS = ['entra_id', 'sailpoint_isc', 'cyberark_pc', 'enterprise_mfa', 'rbac_engine']
CONNECTION_STATUSES = ['active', 'pending', 'inactive', 'undocumented', '']
REVIEWERS = [
    'Alice Martin', 'Bob Zhang', 'Clara Singh', 'Derek Johnson',
    'Eva Kowalski', 'Frank Yamamoto', 'Grace Okafor', 'UNKNOWN', ''
]


def generate_application_inventory():
    """Generate 75 synthetic applications across 14 departments."""
    apps = []
    app_id = 1

    for dept in DEPARTMENTS:
        dept_apps = APP_NAMES[dept]
        for app_name in dept_apps:
            # Assign realistic sensitivity based on department
            if dept in ('Finance', 'Cybersecurity', 'Executive', 'Compliance'):
                sensitivity = random.choices(SENSITIVITY_LEVELS, weights=[0.6, 0.3, 0.1])[0]
            elif dept in ('HR', 'Legal', 'Engineering'):
                sensitivity = random.choices(SENSITIVITY_LEVELS, weights=[0.3, 0.5, 0.2])[0]
            else:
                sensitivity = random.choices(SENSITIVITY_LEVELS, weights=[0.2, 0.4, 0.4])[0]

            # Business critical apps are more likely in Production
            environment = random.choices(ENVIRONMENTS, weights=[0.6, 0.2, 0.2])[0]
            business_critical = (environment == 'Production' and sensitivity == 'HIGH') or random.random() < 0.3

            # Some apps intentionally have missing owners
            if random.random() < 0.08:
                owner = ''
            else:
                first_names = ['Sarah', 'James', 'Maria', 'David', 'Priya', 'Thomas', 'Aisha', 'Robert', 'Lin', 'Carlos']
                last_names = ['Chen', 'Smith', 'Dubois', 'Nakamura', 'Patel', 'Wilson', 'Hassan', 'Garcia', 'Kim', 'Andersen']
                owner = f"{random.choice(first_names)} {random.choice(last_names)}"

            apps.append({
                'app_id': f"APP-{app_id:03d}",
                'app_name': app_name,
                'department': dept,
                'owner': owner,
                'sensitivity': sensitivity,
                'business_critical': business_critical,
                'environment': environment,
                'region': random.choice(REGIONS),
                'authentication_protocol': random.choice(AUTH_PROTOCOLS),
                'data_classification': random.choice(DATA_CLASSIFICATIONS)
            })
            app_id += 1

    return apps


def generate_connection_log(apps):
    """Generate 255 connection records with realistic data quality issues."""
    connections = []
    today = datetime.now()

    for app in apps:
        app_id = app['app_id']
        sensitivity = app['sensitivity']

        # Determine how many IAM products this app connects to (1-5)
        if sensitivity == 'HIGH':
            num_connections = random.randint(3, 5)
        elif sensitivity == 'MEDIUM':
            num_connections = random.randint(2, 4)
        else:
            num_connections = random.randint(1, 3)

        # Select IAM products for this app
        selected_products = random.sample(IAM_PRODUCTS, min(num_connections, len(IAM_PRODUCTS)))

        for product in selected_products:
            # Generate connection date (1-3 years ago)
            days_ago = random.randint(30, 1095)
            conn_date = today - timedelta(days=days_ago)

            # Introduce data quality issues in review dates
            review_date = None
            if random.random() < 0.15:
                # 15% chance: missing review date entirely
                review_date = ''
            elif random.random() < 0.25:
                # 25% chance: stale review (older than 180 days)
                stale_days = random.randint(181, 730)
                review_date = (today - timedelta(days=stale_days)).strftime('%Y-%m-%d')
            elif random.random() < 0.1:
                # 10% chance: inconsistent date format
                recent_days = random.randint(1, 150)
                review_dt = today - timedelta(days=recent_days)
                # Use different date formats to simulate messy data
                formats = ['%d/%m/%Y', '%m-%d-%Y', '%Y/%m/%d']
                review_date = review_dt.strftime(random.choice(formats))
            else:
                # Normal: recent review date in standard format
                recent_days = random.randint(1, 150)
                review_date = (today - timedelta(days=recent_days)).strftime('%Y-%m-%d')

            # Assign connection status with some blanks
            status = random.choice(CONNECTION_STATUSES)

            # Assign reviewer with some unknowns
            reviewer = random.choice(REVIEWERS)

            connections.append({
                'app_id': app_id,
                'iam_product': product,
                'connection_date': conn_date.strftime('%Y-%m-%d'),
                'last_review_date': review_date,
                'connection_status': status,
                'reviewer': reviewer
            })

    # Trim or pad to exactly 255 records
    if len(connections) > 255:
        connections = connections[:255]
    elif len(connections) < 255:
        # Duplicate some entries with variations to reach 255
        while len(connections) < 255:
            base = random.choice(connections[:75])
            new_conn = base.copy()
            new_conn['iam_product'] = random.choice(IAM_PRODUCTS)
            new_conn['connection_status'] = random.choice(CONNECTION_STATUSES)
            connections.append(new_conn)

    return connections


def write_application_inventory(apps, filepath='data/application_inventory.csv'):
    """Write the application inventory to CSV."""
    fieldnames = ['app_id', 'app_name', 'department', 'owner', 'sensitivity',
                  'business_critical', 'environment', 'region',
                  'authentication_protocol', 'data_classification']

    with open(filepath, 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(apps)

    print(f"Generated {len(apps)} applications in {filepath}")


def write_connection_log(connections, filepath='data/connection_log.csv'):
    """Write the connection log to CSV."""
    fieldnames = ['app_id', 'iam_product', 'connection_date',
                  'last_review_date', 'connection_status', 'reviewer']

    with open(filepath, 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(connections)

    print(f"Generated {len(connections)} connection records in {filepath}")


if __name__ == '__main__':
    # Generate application inventory
    apps = generate_application_inventory()
    write_application_inventory(apps)

    # Generate connection log linked to applications
    connections = generate_connection_log(apps)
    write_connection_log(connections)

    print("\nData generation complete!")
    print(f"  - {len(apps)} applications across {len(DEPARTMENTS)} departments")
    print(f"  - {len(connections)} connection records with intentional data quality gaps")

What does the connection log generation do?

The generate_connection_log function creates records that link each application to one or more IAM products. HIGH-sensitivity apps get more connections because they need more protection.

The function introduces five types of data quality issues on purpose:

• Missing review dates (15% of records have blank last_review_date).
• Stale reviews older than 180 days (25% chance).
• Inconsistent date formats mixing DD/MM/YYYY, MM-DD-YYYY, and YYYY/MM/DD.
• Empty connection statuses simulating undocumented connections.
• UNKNOWN reviewers representing ownership gaps.

Your pipeline will detect all of these issues later as governance gaps.

• Save generate_data.py.
• Run the updated script to generate both data files:

python generate_data.py

• You should see output similar to:

Generated 75 applications in data/application_inventory.csv
Generated 255 connection records in data/connection_log.csv

Data generation complete!
  - 75 applications across 14 departments
  - 255 connection records with intentional data quality gaps

• Confirm the files exist by running:

ls data/

• You should see application_inventory.csv, connection_log.csv, and iam_product_registry.json listed in the output.

Only seeing 2 files?

If iam_product_registry.json is missing, go back to the previous substep and make sure you saved the JSON file. If the CSV files are missing, check that the script ran without errors.

help me troubleshoot missing data files in my iam-governance-pipeline project

✔️ Awesome, I've got everything!

Great. Double check that you have saved all three files in your data/ directory.

ⓧ I'd like to double check the full code

Here is the complete generate_data.py and data/iam_product_registry.json for comparison:

import csv
import random
import json
from datetime import datetime, timedelta

# Set seed for reproducible data generation
random.seed(42)

# 14 departments representing a large enterprise
DEPARTMENTS = [
    'Finance', 'HR', 'IT Operations', 'Cybersecurity', 'Supply Chain',
    'Legal', 'Marketing', 'Customer Service', 'R&D', 'Engineering',
    'Procurement', 'Executive', 'Compliance', 'Data Analytics'
]

# Realistic enterprise application names grouped by department
APP_NAMES = {
    'Finance': ['SAP S/4HANA Finance', 'Oracle HFM', 'Bloomberg Terminal', 'Concur Expense', 'Kyriba Treasury', 'BlackLine'],
    'HR': ['Workday HCM', 'SAP SuccessFactors', 'ADP Workforce Now', 'Cornerstone OnDemand', 'ServiceNow HRSD'],
    'IT Operations': ['ServiceNow ITSM', 'Jira Service Management', 'PagerDuty', 'Datadog', 'Splunk Enterprise', 'Ansible Tower'],
    'Cybersecurity': ['CrowdStrike Falcon', 'Palo Alto Cortex', 'Qualys VMDR', 'Tenable.io', 'Rapid7 InsightVM'],
    'Supply Chain': ['SAP TMS', 'Oracle SCM Cloud', 'Blue Yonder WMS', 'Kinaxis RapidResponse', 'FourKites', 'project44'],
    'Legal': ['DocuSign CLM', 'Relativity', 'ContractPodAi', 'Thomson Reuters Legal', 'iManage Work'],
    'Marketing': ['Salesforce Marketing Cloud', 'Adobe Experience Platform', 'HubSpot', 'Marketo Engage', 'Google Analytics 360'],
    'Customer Service': ['Salesforce Service Cloud', 'Zendesk Enterprise', 'Genesys Cloud CX', 'NICE CXone', 'Five9'],
    'R&D': ['Jira Software', 'Confluence', 'GitLab Ultimate', 'SonarQube', 'Artifactory'],
    'Engineering': ['GitHub Enterprise', 'Jenkins', 'Terraform Cloud', 'AWS Console', 'Azure Portal', 'Docker Enterprise'],
    'Procurement': ['SAP Ariba', 'Coupa', 'Jaggaer', 'GEP SMART', 'Oracle Procurement Cloud'],
    'Executive': ['Tableau Server', 'Power BI Premium', 'Board M.I.', 'Anaplan', 'Domo'],
    'Compliance': ['Archer GRC', 'OneTrust', 'LogicGate', 'Diligent', 'MetricStream'],
    'Data Analytics': ['Snowflake', 'Databricks', 'Palantir Foundry', 'Alteryx', 'ThoughtSpot']
}

SENSITIVITY_LEVELS = ['HIGH', 'MEDIUM', 'LOW']
ENVIRONMENTS = ['Production', 'Staging', 'Development']
REGIONS = ['EMEA', 'APAC', 'Americas', 'Global']
AUTH_PROTOCOLS = ['SAML', 'OIDC', 'LDAP', 'Local']
DATA_CLASSIFICATIONS = ['Confidential', 'Internal', 'Public', 'Restricted']

IAM_PRODUCTS = ['entra_id', 'sailpoint_isc', 'cyberark_pc', 'enterprise_mfa', 'rbac_engine']
CONNECTION_STATUSES = ['active', 'pending', 'inactive', 'undocumented', '']
REVIEWERS = [
    'Alice Martin', 'Bob Zhang', 'Clara Singh', 'Derek Johnson',
    'Eva Kowalski', 'Frank Yamamoto', 'Grace Okafor', 'UNKNOWN', ''
]


def generate_application_inventory():
    """Generate 75 synthetic applications across 14 departments."""
    apps = []
    app_id = 1

    for dept in DEPARTMENTS:
        dept_apps = APP_NAMES[dept]
        for app_name in dept_apps:
            # Assign realistic sensitivity based on department
            if dept in ('Finance', 'Cybersecurity', 'Executive', 'Compliance'):
                sensitivity = random.choices(SENSITIVITY_LEVELS, weights=[0.6, 0.3, 0.1])[0]
            elif dept in ('HR', 'Legal', 'Engineering'):
                sensitivity = random.choices(SENSITIVITY_LEVELS, weights=[0.3, 0.5, 0.2])[0]
            else:
                sensitivity = random.choices(SENSITIVITY_LEVELS, weights=[0.2, 0.4, 0.4])[0]

            # Business critical apps are more likely in Production
            environment = random.choices(ENVIRONMENTS, weights=[0.6, 0.2, 0.2])[0]
            business_critical = (environment == 'Production' and sensitivity == 'HIGH') or random.random() < 0.3

            # Some apps intentionally have missing owners
            if random.random() < 0.08:
                owner = ''
            else:
                first_names = ['Sarah', 'James', 'Maria', 'David', 'Priya', 'Thomas', 'Aisha', 'Robert', 'Lin', 'Carlos']
                last_names = ['Chen', 'Smith', 'Dubois', 'Nakamura', 'Patel', 'Wilson', 'Hassan', 'Garcia', 'Kim', 'Andersen']
                owner = f"{random.choice(first_names)} {random.choice(last_names)}"

            apps.append({
                'app_id': f"APP-{app_id:03d}",
                'app_name': app_name,
                'department': dept,
                'owner': owner,
                'sensitivity': sensitivity,
                'business_critical': business_critical,
                'environment': environment,
                'region': random.choice(REGIONS),
                'authentication_protocol': random.choice(AUTH_PROTOCOLS),
                'data_classification': random.choice(DATA_CLASSIFICATIONS)
            })
            app_id += 1

    return apps


def generate_connection_log(apps):
    """Generate 255 connection records with realistic data quality issues."""
    connections = []
    today = datetime.now()

    for app in apps:
        app_id = app['app_id']
        sensitivity = app['sensitivity']

        # Determine how many IAM products this app connects to (1-5)
        if sensitivity == 'HIGH':
            num_connections = random.randint(3, 5)
        elif sensitivity == 'MEDIUM':
            num_connections = random.randint(2, 4)
        else:
            num_connections = random.randint(1, 3)

        # Select IAM products for this app
        selected_products = random.sample(IAM_PRODUCTS, min(num_connections, len(IAM_PRODUCTS)))

        for product in selected_products:
            # Generate connection date (1-3 years ago)
            days_ago = random.randint(30, 1095)
            conn_date = today - timedelta(days=days_ago)

            # Introduce data quality issues in review dates
            review_date = None
            if random.random() < 0.15:
                # 15% chance: missing review date entirely
                review_date = ''
            elif random.random() < 0.25:
                # 25% chance: stale review (older than 180 days)
                stale_days = random.randint(181, 730)
                review_date = (today - timedelta(days=stale_days)).strftime('%Y-%m-%d')
            elif random.random() < 0.1:
                # 10% chance: inconsistent date format
                recent_days = random.randint(1, 150)
                review_dt = today - timedelta(days=recent_days)
                # Use different date formats to simulate messy data
                formats = ['%d/%m/%Y', '%m-%d-%Y', '%Y/%m/%d']
                review_date = review_dt.strftime(random.choice(formats))
            else:
                # Normal: recent review date in standard format
                recent_days = random.randint(1, 150)
                review_date = (today - timedelta(days=recent_days)).strftime('%Y-%m-%d')

            # Assign connection status with some blanks
            status = random.choice(CONNECTION_STATUSES)

            # Assign reviewer with some unknowns
            reviewer = random.choice(REVIEWERS)

            connections.append({
                'app_id': app_id,
                'iam_product': product,
                'connection_date': conn_date.strftime('%Y-%m-%d'),
                'last_review_date': review_date,
                'connection_status': status,
                'reviewer': reviewer
            })

    # Trim or pad to exactly 255 records
    if len(connections) > 255:
        connections = connections[:255]
    elif len(connections) < 255:
        # Duplicate some entries with variations to reach 255
        while len(connections) < 255:
            base = random.choice(connections[:75])
            new_conn = base.copy()
            new_conn['iam_product'] = random.choice(IAM_PRODUCTS)
            new_conn['connection_status'] = random.choice(CONNECTION_STATUSES)
            connections.append(new_conn)

    return connections


def write_application_inventory(apps, filepath='data/application_inventory.csv'):
    """Write the application inventory to CSV."""
    fieldnames = ['app_id', 'app_name', 'department', 'owner', 'sensitivity',
                  'business_critical', 'environment', 'region',
                  'authentication_protocol', 'data_classification']

    with open(filepath, 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(apps)

    print(f"Generated {len(apps)} applications in {filepath}")


def write_connection_log(connections, filepath='data/connection_log.csv'):
    """Write the connection log to CSV."""
    fieldnames = ['app_id', 'iam_product', 'connection_date',
                  'last_review_date', 'connection_status', 'reviewer']

    with open(filepath, 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(connections)

    print(f"Generated {len(connections)} connection records in {filepath}")


if __name__ == '__main__':
    # Generate application inventory
    apps = generate_application_inventory()
    write_application_inventory(apps)

    # Generate connection log linked to applications
    connections = generate_connection_log(apps)
    write_connection_log(connections)

    print("\nData generation complete!")
    print(f"  - {len(apps)} applications across {len(DEPARTMENTS)} departments")
    print(f"  - {len(connections)} connection records with intentional data quality gaps")

{
    "products": [
        {
            "product_id": "entra_id",
            "product_name": "Microsoft Entra ID",
            "category": "SSO/IdP",
            "required_for_sensitivity": ["HIGH", "MEDIUM", "LOW"],
            "protocols": ["SAML", "OIDC"],
            "review_cycle_days": 180,
            "description": "Primary identity provider for SSO across all enterprise applications"
        },
        {
            "product_id": "sailpoint_isc",
            "product_name": "SailPoint Identity Security Cloud",
            "category": "IGA",
            "required_for_sensitivity": ["HIGH", "MEDIUM"],
            "protocols": ["SCIM", "SAML"],
            "review_cycle_days": 90,
            "description": "Identity Governance and Administration for lifecycle management and access reviews"
        },
        {
            "product_id": "cyberark_pc",
            "product_name": "CyberArk Privilege Cloud",
            "category": "PAM",
            "required_for_sensitivity": ["HIGH"],
            "protocols": ["SCIM", "SAML", "OIDC"],
            "review_cycle_days": 90,
            "description": "Privileged Access Management for securing high-sensitivity administrative access"
        },
        {
            "product_id": "enterprise_mfa",
            "product_name": "Enterprise MFA",
            "category": "MFA",
            "required_for_sensitivity": ["HIGH", "MEDIUM"],
            "protocols": ["SAML", "OIDC"],
            "review_cycle_days": 180,
            "description": "Multi-factor authentication enforcement for elevated-risk applications"
        },
        {
            "product_id": "rbac_engine",
            "product_name": "RBAC Engine",
            "category": "Access Control",
            "required_for_sensitivity": ["HIGH", "MEDIUM", "LOW"],
            "protocols": ["LDAP", "SCIM"],
            "review_cycle_days": 365,
            "description": "Role-Based Access Control engine for consistent permission assignment"
        }
    ]
}

You now have three realistic enterprise data sources sitting in your data/ folder. Next up, you will build the feeding rules engine that defines who owns each data field and how conflicts get resolved.

Build the Feeding Rules Engine

Your three data sources are in place. Right now they exist as disconnected files with no rules governing which source is authoritative, how conflicts get resolved, or what happens when a gap is detected.

In a real Control Tower, a feeding rules engine solves this. It defines which system "owns" each data field, how to handle conflicting information, and how quickly teams must remediate governance gaps based on risk severity.

In this step, get ready to:

• Define authoritative sources, conflict resolution, and SLA timelines in a YAML configuration file.
• Build a Python module that loads and parses the feeding rules.
• Implement remediation SLA deadlines and communication channel mappings.

Create the feeding rules configuration

The feeding rules configuration is the backbone of your governance pipeline. It answers three critical questions: who owns each piece of data, what happens when sources disagree, and how fast must issues be fixed.

• Create a new file called feeding_rules.yaml in your project root (the iam-governance-pipeline folder, alongside requirements.txt).
• Paste the following YAML into the file:

authoritative_sources:
  app_metadata:
    source: application_inventory
    update_mode: manual
    owner: Application Portfolio Team
  iam_product_details:
    source: iam_product_registry
    update_mode: scheduled
    schedule_days: 180
    owner: IAM Engineering Team
  connection_status:
    source: connection_log
    update_mode: manual
    owner: Application Owners
  risk_classification:
    source: pipeline_calculated
    update_mode: automatic
    owner: Control Tower

conflict_resolution:
  sensitivity_mismatch:
    action: escalate
    escalation_target: IAM Manager
  owner_conflict:
    action: flag
    resolution_method: most_recent_update
  duplicate_connection:
    action: reject
    notification: Application Owner

remediation_slas:
  HIGH:
    days: 7
    escalation_after_days: 3
    notification_channel: servicenow_ticket
  MEDIUM:
    days: 30
    escalation_after_days: 14
    notification_channel: email
  LOW:
    days: 90
    escalation_after_days: 45
    notification_channel: email

communication_channels:
  master_map:
    format: csv
    audience: GRC / Audit systems
    channel: automated_integration
  governance_gaps:
    format: csv
    audience: Application owners
    channel: weekly_email
  department_scorecard:
    format: csv
    audience: IAM Manager
    channel: internal_reporting
  remediation_queue:
    format: csv
    audience: Application owners + IAM team
    channel: servicenow_tickets
  governance_report:
    format: html
    audience: Control Tower / Management
    channel: intranet_sharepoint

stale_review_threshold_days: 180

What does this configuration do?

This YAML file has five sections that control how your pipeline governs data:

• authoritative_sources declares which system is the single source of truth for each data field. If two systems disagree about an application's sensitivity level, the pipeline trusts the source listed here.
• conflict_resolution defines what happens when sources conflict. Actions are escalate (send to a manager), flag (mark for review), or reject (block the record).
• remediation_slas sets deadlines by risk level. HIGH-risk gaps must be fixed within 7 days. MEDIUM gets 30 days. LOW gets 90 days.
• communication_channels maps each output to its audience and delivery method. GRC teams get CSV via automated integration. Management gets an HTML dashboard on SharePoint.
• stale_review_threshold_days sets 180 days as the cutoff for flagging a connection review as stale.

• Save feeding_rules.yaml.
• Verify the file is valid YAML by running this command in your terminal:

python -c "import yaml; yaml.safe_load(open('feeding_rules.yaml')); print('YAML loaded successfully')"

You should see YAML loaded successfully printed in your terminal.

Seeing a YAML parsing error?

• YAML is sensitive to indentation. Make sure you are using spaces (not tabs) and that each nested level is indented by exactly 2 or 4 spaces consistently.
• Check that colons have a space after them (key: value not key:value).
• Make sure your terminal is in the iam-governance-pipeline directory where feeding_rules.yaml lives.

Help me debug my YAML parsing error in feeding_rules.yaml

Create the rules engine module

The YAML file defines your rules. Now you need Python code to load those rules and make them available to the rest of the pipeline.

• Create a new file called src/rules_engine.py inside the src/ folder.
• Add the following code to define the rules loader:

import yaml
from datetime import datetime, timedelta


def load_feeding_rules(filepath):
    # Open and parse the YAML feeding rules file
    with open(filepath, 'r') as f:
        rules = yaml.safe_load(f)
    return rules

What does this code do?

• yaml.safe_load() parses the YAML file into a Python dictionary. The safe_load variant prevents execution of arbitrary code embedded in YAML.
• The returned dictionary gives the pipeline access to all the rules (authoritative sources, conflict resolution, SLAs) as nested Python objects.
• The datetime and timedelta imports are used by the SLA functions you will add next.

• Save src/rules_engine.py.

You won't see visible output yet. The next chunk adds the SLA functions that use these imports.

Implement remediation SLA logic

When the pipeline detects a governance gap, it needs to assign a remediation deadline based on the gap's severity. HIGH-risk gaps get 7 days. MEDIUM gets 30. LOW gets 90.

• In src/rules_engine.py, add the following three functions below load_feeding_rules:

def get_remediation_deadline(risk_level, rules):
    # Calculate the remediation deadline based on risk severity
    sla = rules.get('remediation_slas', {}).get(risk_level, {})
    days = sla.get('days', 90)
    return datetime.now() + timedelta(days=days)


def get_escalation_deadline(risk_level, rules):
    # Calculate when the gap should be escalated if unresolved
    sla = rules.get('remediation_slas', {}).get(risk_level, {})
    days = sla.get('escalation_after_days', 30)
    return datetime.now() + timedelta(days=days)


def apply_sla_to_gaps(gaps_df, rules):
    # Enrich each gap with its deadline, escalation date, and notification channel
    if gaps_df.empty:
        return gaps_df
    gaps_df['remediation_deadline'] = gaps_df['severity'].apply(
        lambda x: get_remediation_deadline(x, rules)
    )
    gaps_df['escalation_deadline'] = gaps_df['severity'].apply(
        lambda x: get_escalation_deadline(x, rules)
    )
    gaps_df['notification_channel'] = gaps_df['severity'].apply(
        lambda x: rules.get('remediation_slas', {}).get(x, {}).get('notification_channel', 'email')
    )
    return gaps_df

What does this code do?

• get_remediation_deadline() looks up the SLA days for a given risk level (7, 30, or 90) and returns a future date that many days from now.
• get_escalation_deadline() works the same way but uses the shorter escalation_after_days value. If a gap is not resolved within this window, it gets escalated to management.
• apply_sla_to_gaps() is the main function that enriches a DataFrame of gaps with three new columns: remediation_deadline, escalation_deadline, and notification_channel. This connects the YAML rules to actual data.
• The notification_channel maps to the communication channels in your YAML. HIGH-risk gaps trigger ServiceNow tickets. MEDIUM and LOW use email.

• Save src/rules_engine.py.
• Verify the module loads correctly by running this command:

python -c "from src.rules_engine import load_feeding_rules; rules = load_feeding_rules('feeding_rules.yaml'); print(f'Loaded {len(rules)} rule sections'); print(f'HIGH SLA: {rules[\"remediation_slas\"][\"HIGH\"][\"days\"]} days')"

You should see Loaded 5 rule sections and HIGH SLA: 7 days printed in your terminal.

Seeing a ModuleNotFoundError?

• Make sure the src/__init__.py file exists. Python needs it to recognize src/ as a package.
• Confirm your terminal is in the iam-governance-pipeline root folder, not inside src/.
• Check that your virtual environment is activated. You should see (venv) at the start of your terminal prompt.

Help me fix a ModuleNotFoundError when importing rules_engine

✔️ Awesome, I've got everything!

Great. Double check you have saved both feeding_rules.yaml and src/rules_engine.py.

ⓧ I'd like to double check the full code

Here is the complete src/rules_engine.py file:

import yaml
from datetime import datetime, timedelta


def load_feeding_rules(filepath):
    # Open and parse the YAML feeding rules file
    with open(filepath, 'r') as f:
        rules = yaml.safe_load(f)
    return rules


def get_remediation_deadline(risk_level, rules):
    # Calculate the remediation deadline based on risk severity
    sla = rules.get('remediation_slas', {}).get(risk_level, {})
    days = sla.get('days', 90)
    return datetime.now() + timedelta(days=days)


def get_escalation_deadline(risk_level, rules):
    # Calculate when the gap should be escalated if unresolved
    sla = rules.get('remediation_slas', {}).get(risk_level, {})
    days = sla.get('escalation_after_days', 30)
    return datetime.now() + timedelta(days=days)


def apply_sla_to_gaps(gaps_df, rules):
    # Enrich each gap with its deadline, escalation date, and notification channel
    if gaps_df.empty:
        return gaps_df
    gaps_df['remediation_deadline'] = gaps_df['severity'].apply(
        lambda x: get_remediation_deadline(x, rules)
    )
    gaps_df['escalation_deadline'] = gaps_df['severity'].apply(
        lambda x: get_escalation_deadline(x, rules)
    )
    gaps_df['notification_channel'] = gaps_df['severity'].apply(
        lambda x: rules.get('remediation_slas', {}).get(x, {}).get('notification_channel', 'email')
    )
    return gaps_df

Your feeding rules engine is ready. It defines who owns the data, how conflicts resolve, and how fast gaps must be fixed. Next, you will build the risk classification and gap detection logic that uses these rules to evaluate every application in your portfolio.

Build the Risk Classifier and Gap Detector

Your feeding rules engine defines how the pipeline should behave. Now you need the analytical core that actually classifies risk and identifies governance failures across your 75 applications.

In this step, you will build three Python modules that form the decision-making backbone of your pipeline. A risk classifier assigns each application-connection record to a tier (HIGH, MEDIUM, or LOW). A gap detector scans the consolidated data for five types of governance failures. A data processor loads, cleans, and merges your three data sources into the unified structure these engines consume.

In this step, get ready to:

• Create a three-tier risk classification engine based on sensitivity, criticality, and environment.
• Implement gap detection logic that identifies 5 governance failure types across your application portfolio.
• Build a data processor that loads, cleans, and consolidates your three enterprise data sources into a unified DataFrame.

Create the risk classification engine

Every application-to-IAM-product connection needs a risk level so the Control Tower team knows where to focus attention. The classifier uses a simple decision tree: if an application handles HIGH sensitivity data, is business critical, AND runs in Production, it gets a HIGH risk rating. Applications with MEDIUM sensitivity, stale reviews, or undocumented statuses get MEDIUM. Everything else is LOW.

• Create a new file called risk_classifier.py inside the src/ folder.
• Add the following risk classification logic:

import pandas as pd


def classify_risk(row):
    sensitivity = str(row.get('sensitivity', '')).upper()
    business_critical = bool(row.get('business_critical', False))
    environment = str(row.get('environment', '')).strip()
    is_stale = bool(row.get('is_stale', False))
    status = str(row.get('connection_status', '')).strip().lower()

    if sensitivity == 'HIGH' and business_critical and environment == 'Production':
        return 'HIGH'
    elif sensitivity == 'MEDIUM' or is_stale or status in ('pending', 'undocumented', ''):
        return 'MEDIUM'
    else:
        return 'LOW'


def apply_risk_classification(df):
    df['risk_level'] = df.apply(classify_risk, axis=1)
    return df

What does this code do?

The classify_risk function examines each row of merged data and assigns a risk tier using three decision branches.

• HIGH requires ALL three conditions: HIGH sensitivity, business critical, and Production environment.
• MEDIUM triggers on ANY of: MEDIUM sensitivity, a stale review flag, or a connection status that is pending, undocumented, or empty.
• LOW is the fallback for everything that does not match the above conditions.
• apply_risk_classification is a wrapper that applies classify_risk row-by-row across an entire DataFrame, adding a new risk_level column.

Why this three-tier model?

Enterprise IAM governance typically uses a tiered risk model to prioritise remediation. HIGH risk items get 7-day SLAs and ServiceNow tickets. MEDIUM items get 30 days. LOW items get 90 days. Without tiers, the Control Tower team would treat every gap with equal urgency, which means nothing gets fixed quickly.

• Save src/risk_classifier.py.

You will not see output from this file yet. The pipeline orchestrator in a later step will call these functions on your consolidated data.

Seeing a syntax error?

• Make sure the elif line has the tuple ('pending', 'undocumented', '') with matching parentheses and quotes.
• Verify that your file is saved inside src/, not at the project root.

Help me debug my risk_classifier.py file

Implement the gap detection engine

Risk classification tells you how dangerous a gap is. The gap detector finds the gaps themselves. It scans merged application data for five distinct failure types that a Control Tower team needs to track and remediate.

• Create a new file called gap_detector.py inside the src/ folder.
• Add the first two detection functions by pasting this code:

import pandas as pd


def detect_missing_connections(merged_df, registry):
    gaps = []
    for _, row in merged_df.drop_duplicates(subset='app_id').iterrows():
        app_id = row['app_id']
        sensitivity = str(row.get('sensitivity', '')).upper()
        app_connections = merged_df[merged_df['app_id'] == app_id]['iam_product'].dropna().unique()

        for product in registry.get('products', []):
            required_for = product.get('required_for_sensitivity', [])
            if sensitivity in required_for and product['product_id'] not in app_connections:
                gaps.append({
                    'app_id': app_id,
                    'app_name': row.get('app_name', ''),
                    'gap_type': 'MISSING_CONNECTION',
                    'detail': f"Required product '{product['product_name']}' not connected",
                    'severity': row.get('risk_level', 'MEDIUM')
                })
    return gaps


def detect_stale_reviews(merged_df):
    gaps = []
    stale = merged_df[merged_df['is_stale'] == True]
    for _, row in stale.iterrows():
        if pd.notna(row.get('iam_product')):
            gaps.append({
                'app_id': row['app_id'],
                'app_name': row.get('app_name', ''),
                'gap_type': 'STALE_REVIEW',
                'detail': f"Connection to '{row['iam_product']}' not reviewed within cycle",
                'severity': row.get('risk_level', 'MEDIUM')
            })
    return gaps

What do these detection functions do?

• detect_missing_connections checks each application's sensitivity level against the IAM product registry. If the registry says a product is required for that sensitivity level but the app has no connection to it, that is a MISSING_CONNECTION gap.
• detect_stale_reviews finds connections flagged as stale (not reviewed within the 180-day threshold). These represent governance decay where connections exist but nobody is verifying they are still appropriate.

• Add the remaining three detection functions below the code you just pasted:

def detect_undocumented_connections(merged_df):
    gaps = []
    undoc = merged_df[merged_df['connection_status'].isna() | (merged_df['connection_status'] == '')]
    for _, row in undoc.iterrows():
        if pd.notna(row.get('iam_product')):
            gaps.append({
                'app_id': row['app_id'],
                'app_name': row.get('app_name', ''),
                'gap_type': 'UNDOCUMENTED_CONNECTION',
                'detail': f"Connection to '{row['iam_product']}' has no documented status",
                'severity': row.get('risk_level', 'MEDIUM')
            })
    return gaps


def detect_unknown_owners(merged_df):
    gaps = []
    unknown = merged_df[merged_df['owner'] == 'UNKNOWN'].drop_duplicates(subset='app_id')
    for _, row in unknown.iterrows():
        gaps.append({
            'app_id': row['app_id'],
            'app_name': row.get('app_name', ''),
            'gap_type': 'UNKNOWN_OWNER',
            'detail': 'Application has no registered owner',
            'severity': row.get('risk_level', 'MEDIUM')
        })
    return gaps


def detect_missing_review_dates(merged_df):
    gaps = []
    missing = merged_df[merged_df['last_review_date'].isna() & merged_df['iam_product'].notna()]
    for _, row in missing.iterrows():
        status = str(row.get('connection_status', '')).lower()
        if status == 'active':
            gaps.append({
                'app_id': row['app_id'],
                'app_name': row.get('app_name', ''),
                'gap_type': 'MISSING_REVIEW_DATE',
                'detail': f"Active connection to '{row['iam_product']}' has no review date",
                'severity': row.get('risk_level', 'MEDIUM')
            })
    return gaps

What do these three functions detect?

• UNDOCUMENTED_CONNECTION flags connections where a link to an IAM product exists but has no documented status (empty or null).
• UNKNOWN_OWNER flags applications where no registered owner could be found. Without an owner, nobody is accountable for remediation.
• MISSING_REVIEW_DATE finds active connections that have never been reviewed. The connection is live but there is no evidence of governance oversight.

• Add the orchestrator function that runs all five detectors at the bottom of the file:

def run_all_gap_detection(merged_df, registry):
    all_gaps = []
    all_gaps.extend(detect_missing_connections(merged_df, registry))
    all_gaps.extend(detect_stale_reviews(merged_df))
    all_gaps.extend(detect_undocumented_connections(merged_df))
    all_gaps.extend(detect_unknown_owners(merged_df))
    all_gaps.extend(detect_missing_review_dates(merged_df))
    return pd.DataFrame(all_gaps)

Why a single orchestrator function?

run_all_gap_detection collects results from every detector into one list, then converts it to a pandas DataFrame. This gives the pipeline a single function to call rather than wiring five separate detectors manually. It also makes it easy to add new gap types later without changing the pipeline's main entry point.

• Save src/gap_detector.py.

You will not see output from this file yet. The pipeline will call run_all_gap_detection after the data processor merges all three sources.

Seeing indentation errors?

• Each function must start at the top-level indentation (no leading spaces). The dictionary keys inside each gaps.append({}) call must align consistently.
• Make sure you pasted the second and third code blocks BELOW the first, not inside an existing function.

Help me fix indentation in my gap_detector.py

✔️ Awesome, I've got everything!

Great. Double check you have saved src/gap_detector.py with all six functions.

ⓧ I'd like to double check the full code

Here is the complete src/gap_detector.py file for comparison:

import pandas as pd


def detect_missing_connections(merged_df, registry):
    gaps = []
    for _, row in merged_df.drop_duplicates(subset='app_id').iterrows():
        app_id = row['app_id']
        sensitivity = str(row.get('sensitivity', '')).upper()
        app_connections = merged_df[merged_df['app_id'] == app_id]['iam_product'].dropna().unique()

        for product in registry.get('products', []):
            required_for = product.get('required_for_sensitivity', [])
            if sensitivity in required_for and product['product_id'] not in app_connections:
                gaps.append({
                    'app_id': app_id,
                    'app_name': row.get('app_name', ''),
                    'gap_type': 'MISSING_CONNECTION',
                    'detail': f"Required product '{product['product_name']}' not connected",
                    'severity': row.get('risk_level', 'MEDIUM')
                })
    return gaps


def detect_stale_reviews(merged_df):
    gaps = []
    stale = merged_df[merged_df['is_stale'] == True]
    for _, row in stale.iterrows():
        if pd.notna(row.get('iam_product')):
            gaps.append({
                'app_id': row['app_id'],
                'app_name': row.get('app_name', ''),
                'gap_type': 'STALE_REVIEW',
                'detail': f"Connection to '{row['iam_product']}' not reviewed within cycle",
                'severity': row.get('risk_level', 'MEDIUM')
            })
    return gaps


def detect_undocumented_connections(merged_df):
    gaps = []
    undoc = merged_df[merged_df['connection_status'].isna() | (merged_df['connection_status'] == '')]
    for _, row in undoc.iterrows():
        if pd.notna(row.get('iam_product')):
            gaps.append({
                'app_id': row['app_id'],
                'app_name': row.get('app_name', ''),
                'gap_type': 'UNDOCUMENTED_CONNECTION',
                'detail': f"Connection to '{row['iam_product']}' has no documented status",
                'severity': row.get('risk_level', 'MEDIUM')
            })
    return gaps


def detect_unknown_owners(merged_df):
    gaps = []
    unknown = merged_df[merged_df['owner'] == 'UNKNOWN'].drop_duplicates(subset='app_id')
    for _, row in unknown.iterrows():
        gaps.append({
            'app_id': row['app_id'],
            'app_name': row.get('app_name', ''),
            'gap_type': 'UNKNOWN_OWNER',
            'detail': 'Application has no registered owner',
            'severity': row.get('risk_level', 'MEDIUM')
        })
    return gaps


def detect_missing_review_dates(merged_df):
    gaps = []
    missing = merged_df[merged_df['last_review_date'].isna() & merged_df['iam_product'].notna()]
    for _, row in missing.iterrows():
        status = str(row.get('connection_status', '')).lower()
        if status == 'active':
            gaps.append({
                'app_id': row['app_id'],
                'app_name': row.get('app_name', ''),
                'gap_type': 'MISSING_REVIEW_DATE',
                'detail': f"Active connection to '{row['iam_product']}' has no review date",
                'severity': row.get('risk_level', 'MEDIUM')
            })
    return gaps


def run_all_gap_detection(merged_df, registry):
    all_gaps = []
    all_gaps.extend(detect_missing_connections(merged_df, registry))
    all_gaps.extend(detect_stale_reviews(merged_df))
    all_gaps.extend(detect_undocumented_connections(merged_df))
    all_gaps.extend(detect_unknown_owners(merged_df))
    all_gaps.extend(detect_missing_review_dates(merged_df))
    return pd.DataFrame(all_gaps)

Build the data processor that consolidates your three sources

The risk classifier and gap detector both expect a single, cleaned DataFrame as input. Your raw data lives in three separate files with inconsistent formats. The data processor bridges that gap by loading each source, normalising fields, filling missing values, flagging stale reviews, and merging everything into one unified structure.

• Create a new file called data_processor.py inside the src/ folder.
• Add the imports and data loading functions by pasting this code:

import pandas as pd
import json
from datetime import datetime, timedelta


def load_application_inventory(filepath):
    df = pd.read_csv(filepath)
    df['app_id'] = df['app_id'].astype(str).str.strip()
    df['sensitivity'] = df['sensitivity'].str.upper().str.strip()
    df['environment'] = df['environment'].str.strip()
    df['business_critical'] = df['business_critical'].astype(bool)
    return df


def load_iam_product_registry(filepath):
    with open(filepath, 'r') as f:
        registry = json.load(f)
    return registry


def load_connection_log(filepath):
    df = pd.read_csv(filepath)
    df['app_id'] = df['app_id'].astype(str).str.strip()
    df['last_review_date'] = pd.to_datetime(df['last_review_date'], format='mixed', errors='coerce')
    df['connection_date'] = pd.to_datetime(df['connection_date'], format='mixed', errors='coerce')
    return df

What does each loader do?

• load_application_inventory reads the CSV and normalises fields: strips whitespace from IDs, uppercases sensitivity levels, and casts business_critical to boolean.
• load_iam_product_registry reads the JSON product definitions directly into a Python dictionary.
• load_connection_log reads the CSV and parses date columns using format='mixed' to handle inconsistent date formats in the raw data. Invalid dates become NaT (Not a Time) instead of crashing.

• Add the cleaning and consolidation functions below the loaders:

def fill_missing_owners(app_df):
    unknown_mask = app_df['owner'].isna() | (app_df['owner'].str.strip() == '')
    app_df.loc[unknown_mask, 'owner'] = 'UNKNOWN'
    return app_df


def flag_stale_reviews(conn_df, threshold_days=180):
    cutoff = datetime.now() - timedelta(days=threshold_days)
    conn_df['is_stale'] = conn_df['last_review_date'] < cutoff
    conn_df.loc[conn_df['last_review_date'].isna(), 'is_stale'] = True
    return conn_df


def consolidate_data(app_df, conn_df, registry):
    app_df = fill_missing_owners(app_df)
    conn_df = flag_stale_reviews(conn_df)

    merged = conn_df.merge(
        app_df[['app_id', 'app_name', 'department', 'owner', 'sensitivity',
                'business_critical', 'environment', 'region',
                'authentication_protocol', 'data_classification']],
        on='app_id',
        how='right'
    )
    return merged, registry

What does the consolidation logic do?

• fill_missing_owners replaces empty or null owner fields with the string UNKNOWN. This enables the gap detector to flag these apps as UNKNOWN_OWNER gaps.
• flag_stale_reviews adds an is_stale boolean column. Any connection not reviewed within 180 days (or with no review date at all) is marked as stale.
• consolidate_data merges the connection log with the application inventory using a right join on app_id. The right join ensures every application appears in the output, even if it has no connections.

• Save src/data_processor.py.
• Confirm your src/ folder now contains four files by running:

ls src/

You should see __init__.py, data_processor.py, gap_detector.py, risk_classifier.py, and rules_engine.py listed in the output.

Missing a file?

• If data_processor.py is missing, make sure you created the file inside src/ and not at the project root.
• Check that your terminal is in the iam-governance-pipeline directory. If you are inside src/, run cd .. first.

Help me find my missing Python files

✔️ Awesome, I've got everything!

Great. Make sure you have saved src/data_processor.py with all six functions (three loaders, two cleaners, and one consolidator).

ⓧ I'd like to double check the full code

Here is the complete src/data_processor.py file for comparison:

import pandas as pd
import json
from datetime import datetime, timedelta


def load_application_inventory(filepath):
    df = pd.read_csv(filepath)
    df['app_id'] = df['app_id'].astype(str).str.strip()
    df['sensitivity'] = df['sensitivity'].str.upper().str.strip()
    df['environment'] = df['environment'].str.strip()
    df['business_critical'] = df['business_critical'].astype(bool)
    return df


def load_iam_product_registry(filepath):
    with open(filepath, 'r') as f:
        registry = json.load(f)
    return registry


def load_connection_log(filepath):
    df = pd.read_csv(filepath)
    df['app_id'] = df['app_id'].astype(str).str.strip()
    df['last_review_date'] = pd.to_datetime(df['last_review_date'], format='mixed', errors='coerce')
    df['connection_date'] = pd.to_datetime(df['connection_date'], format='mixed', errors='coerce')
    return df


def fill_missing_owners(app_df):
    unknown_mask = app_df['owner'].isna() | (app_df['owner'].str.strip() == '')
    app_df.loc[unknown_mask, 'owner'] = 'UNKNOWN'
    return app_df


def flag_stale_reviews(conn_df, threshold_days=180):
    cutoff = datetime.now() - timedelta(days=threshold_days)
    conn_df['is_stale'] = conn_df['last_review_date'] < cutoff
    conn_df.loc[conn_df['last_review_date'].isna(), 'is_stale'] = True
    return conn_df


def consolidate_data(app_df, conn_df, registry):
    app_df = fill_missing_owners(app_df)
    conn_df = flag_stale_reviews(conn_df)

    merged = conn_df.merge(
        app_df[['app_id', 'app_name', 'department', 'owner', 'sensitivity',
                'business_critical', 'environment', 'region',
                'authentication_protocol', 'data_classification']],
        on='app_id',
        how='right'
    )
    return merged, registry

Your analytical engine is complete. You have a risk classifier that assigns three tiers, a gap detector that identifies five governance failure types, and a data processor that merges your scattered sources into one clean DataFrame. Next up, you will build the reporting layer that transforms these results into management-ready outputs.

Build the Report Generator and HTML Dashboard

Your pipeline can now classify risk and detect governance gaps. The missing piece is translating that analysis into outputs that different stakeholders can actually use.

A Control Tower team produces multiple output formats for different audiences: CSV files for GRC audit systems, department scorecards for IAM managers, and visual dashboards for executive stakeholders. In this step, you will build the reporting layer that generates all of these outputs.

In this step, get ready to:

• Create the report generator module that produces CSV outputs and an audit trail log.
• Create the scorecard module that aggregates risk by department and builds a remediation priority queue.
• Add a Jinja2 HTML dashboard template with executive summary cards, risk distribution, and scorecard tables.

Create the report generator for CSV outputs

The report generator is responsible for writing the pipeline's results into structured files. Each output format is defined in your feeding_rules.yaml under communication_channels.

• Create a new file called report_generator.py inside your src/ folder.
• Add the imports and the two CSV output functions by pasting this code:

import pandas as pd
from datetime import datetime
from jinja2 import Template
import os


def generate_master_map(merged_df, output_dir):
    # Select the columns that form the unified governance map
    cols = ['app_id', 'app_name', 'department', 'owner', 'sensitivity',
            'business_critical', 'environment', 'iam_product',
            'connection_status', 'last_review_date', 'risk_level']
    available_cols = [c for c in cols if c in merged_df.columns]
    merged_df[available_cols].to_csv(
        os.path.join(output_dir, 'iam_connection_master_map.csv'), index=False
    )


def generate_gap_report(gaps_df, output_dir):
    # Write all detected governance gaps to a CSV for application owners
    gaps_df.to_csv(os.path.join(output_dir, 'governance_gaps.csv'), index=False)

What does this code do?

These two functions handle the CSV outputs that feed into downstream systems.

• generate_master_map writes the unified governance map. It selects only the columns that GRC and audit systems need, filtering out any internal processing columns.
• generate_gap_report exports all detected gaps as a CSV. Application owners receive this as a weekly email attachment per the feeding rules.

• Add the write_run_log function below generate_gap_report to create a timestamped audit trail:

def write_run_log(output_dir, stats):
    # Append a timestamped entry to the pipeline run log
    log_path = os.path.join(output_dir, 'pipeline_run_log.txt')
    with open(log_path, 'a', encoding='utf-8') as f:
        f.write(f"\n{'='*60}\n")
        f.write(f"Pipeline Run: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
        f.write(f"{'='*60}\n")
        for key, value in stats.items():
            f.write(f"  {key}: {value}\n")
        f.write(f"{'='*60}\n")

What does this code do?

write_run_log appends a new entry every time the pipeline executes. Each entry records the timestamp plus key statistics (applications processed, gaps detected, critical items). Using append mode ('a') means you build a cumulative history across runs.

This audit trail is essential for governance. Auditors need to prove the pipeline runs regularly and that gap counts are trending in the right direction.

• Save src/report_generator.py.

Seeing a syntax error?

• Check that the f-string using {'='*60} has matching curly braces. The outer braces belong to the f-string and the inner expression multiplies the = character 60 times.
• Make sure from jinja2 import Template uses a lowercase jinja2 for the package name and uppercase Template for the class.

Help me fix a syntax error in my report_generator.py file

Create the scorecard module

Department scorecards give IAM managers a per-team view of risk posture. The scorecard aggregates applications, risk levels, and gaps for each of the 14 departments in your data.

• Create a new file called scorecard.py inside your src/ folder.
• Add the import and the department scorecard function by pasting this code:

import pandas as pd


def generate_department_scorecard(merged_df, gaps_df):
    # Group applications by department to calculate per-department metrics
    departments = merged_df.drop_duplicates(subset='app_id').groupby('department')

    scorecard_rows = []
    for dept, group in departments:
        total_apps = len(group)
        high_risk = len(group[group['risk_level'] == 'HIGH'])
        medium_risk = len(group[group['risk_level'] == 'MEDIUM'])
        low_risk = len(group[group['risk_level'] == 'LOW'])

        # Count gaps belonging to this department's applications
        dept_gaps = gaps_df[gaps_df['app_id'].isin(group['app_id'])] if not gaps_df.empty else pd.DataFrame()
        total_gaps = len(dept_gaps)
        critical_gaps = len(dept_gaps[dept_gaps['severity'] == 'HIGH']) if not dept_gaps.empty else 0

        # Coverage percentage: ratio of gap-free connections to total possible
        coverage = round((1 - total_gaps / max(total_apps * 5, 1)) * 100, 1)
        coverage = max(0, min(100, coverage))

        scorecard_rows.append({
            'department': dept,
            'total_apps': total_apps,
            'high_risk': high_risk,
            'medium_risk': medium_risk,
            'low_risk': low_risk,
            'total_gaps': total_gaps,
            'critical_gaps': critical_gaps,
            'coverage_pct': coverage
        })

    return pd.DataFrame(scorecard_rows).sort_values('critical_gaps', ascending=False)

What does this code do?

generate_department_scorecard loops through each department and calculates six metrics: total applications, counts per risk tier (HIGH/MEDIUM/LOW), total governance gaps, and critical gaps.

The coverage_pct formula divides total gaps by the maximum possible gaps (apps multiplied by 5 IAM products). A department with zero gaps scores 100% coverage. The result is sorted so the departments with the most critical gaps appear first.

• Add the remediation queue function below generate_department_scorecard:

def generate_remediation_queue(gaps_df):
    # Sort gaps by severity priority (HIGH first) then by deadline
    if gaps_df.empty:
        return gaps_df
    priority_order = {'HIGH': 0, 'MEDIUM': 1, 'LOW': 2}
    gaps_df['priority_sort'] = gaps_df['severity'].map(priority_order)
    queue = gaps_df.sort_values(['priority_sort', 'remediation_deadline'])
    queue = queue.drop(columns=['priority_sort'])
    return queue

What does this code do?

generate_remediation_queue transforms the flat gaps list into a prioritised queue. It assigns numeric sort values (HIGH=0, MEDIUM=1, LOW=2) so that high-severity gaps with the nearest deadlines appear at the top.

This is the output that maps to ServiceNow tickets in your feeding rules. Each item in the queue represents a remediation action assigned to an application owner.

• Save src/scorecard.py.
• Confirm the file exists by running:

ls src/

You should see scorecard.py listed alongside report_generator.py, data_processor.py, risk_classifier.py, gap_detector.py, and rules_engine.py.

Missing scorecard.py in the listing?

• Make sure you created the file inside the src/ directory, not in the project root.
• Check the filename is exactly scorecard.py with no typos or extra extensions.

Help me find where my scorecard.py file ended up

✔️ Awesome, I've got everything!

Great. Double check you have saved src/scorecard.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete src/scorecard.py file:

import pandas as pd


def generate_department_scorecard(merged_df, gaps_df):
    # Group applications by department to calculate per-department metrics
    departments = merged_df.drop_duplicates(subset='app_id').groupby('department')

    scorecard_rows = []
    for dept, group in departments:
        total_apps = len(group)
        high_risk = len(group[group['risk_level'] == 'HIGH'])
        medium_risk = len(group[group['risk_level'] == 'MEDIUM'])
        low_risk = len(group[group['risk_level'] == 'LOW'])

        # Count gaps belonging to this department's applications
        dept_gaps = gaps_df[gaps_df['app_id'].isin(group['app_id'])] if not gaps_df.empty else pd.DataFrame()
        total_gaps = len(dept_gaps)
        critical_gaps = len(dept_gaps[dept_gaps['severity'] == 'HIGH']) if not dept_gaps.empty else 0

        # Coverage percentage: ratio of gap-free connections to total possible
        coverage = round((1 - total_gaps / max(total_apps * 5, 1)) * 100, 1)
        coverage = max(0, min(100, coverage))

        scorecard_rows.append({
            'department': dept,
            'total_apps': total_apps,
            'high_risk': high_risk,
            'medium_risk': medium_risk,
            'low_risk': low_risk,
            'total_gaps': total_gaps,
            'critical_gaps': critical_gaps,
            'coverage_pct': coverage
        })

    return pd.DataFrame(scorecard_rows).sort_values('critical_gaps', ascending=False)


def generate_remediation_queue(gaps_df):
    # Sort gaps by severity priority (HIGH first) then by deadline
    if gaps_df.empty:
        return gaps_df
    priority_order = {'HIGH': 0, 'MEDIUM': 1, 'LOW': 2}
    gaps_df['priority_sort'] = gaps_df['severity'].map(priority_order)
    queue = gaps_df.sort_values(['priority_sort', 'remediation_deadline'])
    queue = queue.drop(columns=['priority_sort'])
    return queue

Add the HTML dashboard template

The HTML dashboard is the management-facing output. It uses Jinja2 template rendering to produce a self-contained HTML file with inline CSS bar charts, colour-coded risk indicators, and sortable tables.

• Switch back to src/report_generator.py.
• Find the line that says import os at the top of the file.
• Add the DASHBOARD_TEMPLATE constant below it, leaving two blank lines between the imports and the template. Paste this code between import os and def generate_master_map:

DASHBOARD_TEMPLATE = """<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>IAM Governance Report</title>
    <style>
        body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif; margin: 0; padding: 20px; background: #f5f5f5; }
        .container { max-width: 1200px; margin: 0 auto; }
        h1 { color: #1a237e; border-bottom: 3px solid #1a237e; padding-bottom: 10px; }
        h2 { color: #283593; margin-top: 30px; }
        .summary-cards { display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 15px; margin: 20px 0; }
        .card { background: white; border-radius: 8px; padding: 20px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); text-align: center; }
        .card .number { font-size: 2.5em; font-weight: bold; }
        .card .label { color: #666; margin-top: 5px; }
        .card.critical { border-left: 4px solid #d32f2f; }
        .card.warning { border-left: 4px solid #f57c00; }
        .card.success { border-left: 4px solid #388e3c; }
        .card.info { border-left: 4px solid #1976d2; }
        table { width: 100%; border-collapse: collapse; background: white; border-radius: 8px; overflow: hidden; box-shadow: 0 2px 4px rgba(0,0,0,0.1); margin: 15px 0; }
        th { background: #1a237e; color: white; padding: 12px; text-align: left; }
        td { padding: 10px 12px; border-bottom: 1px solid #eee; }
        tr:hover { background: #f5f5f5; }
        .risk-high { color: #d32f2f; font-weight: bold; }
        .risk-medium { color: #f57c00; font-weight: bold; }
        .risk-low { color: #388e3c; font-weight: bold; }
        .bar-container { background: #e0e0e0; border-radius: 4px; height: 20px; width: 100%; }
        .bar-fill { height: 100%; border-radius: 4px; }
        .timestamp { color: #999; font-size: 0.9em; margin-top: 30px; }
    </style>
</head>
<body>
<div class="container">
    <h1>IAM Governance Report</h1>
    <p class="timestamp">Generated: {{ generated_at }}</p>

    <h2>Executive Summary</h2>
    <div class="summary-cards">
        <div class="card info"><div class="number">{{ total_apps }}</div><div class="label">Applications</div></div>
        <div class="card info"><div class="number">{{ total_connections }}</div><div class="label">Connections</div></div>
        <div class="card critical"><div class="number">{{ total_gaps }}</div><div class="label">Governance Gaps</div></div>
        <div class="card critical"><div class="number">{{ critical_items }}</div><div class="label">Critical Items</div></div>
    </div>

    <h2>Risk Distribution</h2>
    <div class="summary-cards">
        <div class="card critical"><div class="number">{{ high_risk }}</div><div class="label">HIGH Risk Apps</div></div>
        <div class="card warning"><div class="number">{{ medium_risk }}</div><div class="label">MEDIUM Risk Apps</div></div>
        <div class="card success"><div class="number">{{ low_risk }}</div><div class="label">LOW Risk Apps</div></div>
    </div>

    <h2>Department Scorecard</h2>
    <table>
        <tr><th>Department</th><th>Apps</th><th>High Risk</th><th>Gaps</th><th>Critical</th><th>Coverage</th></tr>
        {% for row in scorecard %}
        <tr>
            <td>{{ row.department }}</td>
            <td>{{ row.total_apps }}</td>
            <td class="risk-high">{{ row.high_risk }}</td>
            <td>{{ row.total_gaps }}</td>
            <td class="risk-high">{{ row.critical_gaps }}</td>
            <td>
                <div class="bar-container">
                    <div class="bar-fill" style="width: {{ row.coverage_pct }}%; background: {% if row.coverage_pct >= 80 %}#388e3c{% elif row.coverage_pct >= 50 %}#f57c00{% else %}#d32f2f{% endif %};"></div>
                </div>
                {{ row.coverage_pct }}%
            </td>
        </tr>
        {% endfor %}
    </table>

    <h2>Top Remediation Items</h2>
    <table>
        <tr><th>App</th><th>Gap Type</th><th>Severity</th><th>Detail</th><th>Deadline</th></tr>
        {% for row in top_remediation %}
        <tr>
            <td>{{ row.app_name }}</td>
            <td>{{ row.gap_type }}</td>
            <td class="risk-{{ row.severity | lower }}">{{ row.severity }}</td>
            <td>{{ row.detail }}</td>
            <td>{{ row.remediation_deadline }}</td>
        </tr>
        {% endfor %}
    </table>
</div>
</body>
</html>"""

What does this template do?

This is a Jinja2 template stored as a Python string. The {{ }} syntax marks places where Python data gets injected at render time.

The template has four sections: an executive summary with card-style KPIs, a risk distribution breakdown, a department scorecard table with inline CSS bar charts for coverage, and a remediation items table showing the top 15 gaps by priority.

The bar chart uses a simple technique. A div with width set to the coverage percentage creates a visual bar, with the background colour changing from green to orange to red based on thresholds.

• Now add the generate_html_dashboard function that renders this template. Place it between generate_gap_report and write_run_log:

def generate_html_dashboard(merged_df, gaps_df, scorecard_df, output_dir):
    # Calculate summary statistics for the executive dashboard
    apps_unique = merged_df.drop_duplicates(subset='app_id')
    total_apps = len(apps_unique)
    total_connections = len(merged_df[merged_df['iam_product'].notna()])
    total_gaps = len(gaps_df)
    critical_items = len(gaps_df[gaps_df['severity'] == 'HIGH']) if not gaps_df.empty else 0
    high_risk = len(apps_unique[apps_unique['risk_level'] == 'HIGH'])
    medium_risk = len(apps_unique[apps_unique['risk_level'] == 'MEDIUM'])
    low_risk = len(apps_unique[apps_unique['risk_level'] == 'LOW'])

    # Format the top remediation items for display
    top_remediation = gaps_df.head(15).to_dict('records') if not gaps_df.empty else []
    for item in top_remediation:
        if 'remediation_deadline' in item and pd.notna(item['remediation_deadline']):
            item['remediation_deadline'] = pd.Timestamp(item['remediation_deadline']).strftime('%Y-%m-%d')

    # Render the Jinja2 template with all computed values
    template = Template(DASHBOARD_TEMPLATE)
    html = template.render(
        generated_at=datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        total_apps=total_apps,
        total_connections=total_connections,
        total_gaps=total_gaps,
        critical_items=critical_items,
        high_risk=high_risk,
        medium_risk=medium_risk,
        low_risk=low_risk,
        scorecard=scorecard_df.to_dict('records'),
        top_remediation=top_remediation
    )

    # Write the rendered HTML to the output directory
    with open(os.path.join(output_dir, 'iam_governance_report.html'), 'w', encoding='utf-8') as f:
        f.write(html)

What does this code do?

generate_html_dashboard aggregates the pipeline's results into a single HTML file. It deduplicates applications to count unique entries, calculates risk tier distributions, and formats deadline timestamps for human readability.

The Template(DASHBOARD_TEMPLATE).render() call injects all computed values into the HTML template. The result is a self-contained file that stakeholders can open in any browser without needing Python installed.

• Save src/report_generator.py.
• Verify the file has no syntax errors by running:

python -c "import src.report_generator"

If there is no output and no error, the module imported successfully.

Seeing an ImportError or SyntaxError?

• A SyntaxError pointing at the triple-quoted string usually means a missing or extra quotation mark in DASHBOARD_TEMPLATE. Check that it starts and ends with """.
• An IndentationError means a function body is misaligned. Ensure generate_html_dashboard is at the module level (no indentation on the def line).
• If you see ModuleNotFoundError: No module named 'jinja2', make sure your virtual environment is activated.

Help me debug an import error in my report_generator module

✔️ Awesome, I've got everything!

Great. Double check you have saved src/report_generator.py with all five elements: DASHBOARD_TEMPLATE, generate_master_map, generate_gap_report, generate_html_dashboard, and write_run_log.

ⓧ I'd like to double check the full code

Here is the complete src/report_generator.py file:

import pandas as pd
from datetime import datetime
from jinja2 import Template
import os


DASHBOARD_TEMPLATE = """<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>IAM Governance Report</title>
    <style>
        body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif; margin: 0; padding: 20px; background: #f5f5f5; }
        .container { max-width: 1200px; margin: 0 auto; }
        h1 { color: #1a237e; border-bottom: 3px solid #1a237e; padding-bottom: 10px; }
        h2 { color: #283593; margin-top: 30px; }
        .summary-cards { display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 15px; margin: 20px 0; }
        .card { background: white; border-radius: 8px; padding: 20px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); text-align: center; }
        .card .number { font-size: 2.5em; font-weight: bold; }
        .card .label { color: #666; margin-top: 5px; }
        .card.critical { border-left: 4px solid #d32f2f; }
        .card.warning { border-left: 4px solid #f57c00; }
        .card.success { border-left: 4px solid #388e3c; }
        .card.info { border-left: 4px solid #1976d2; }
        table { width: 100%; border-collapse: collapse; background: white; border-radius: 8px; overflow: hidden; box-shadow: 0 2px 4px rgba(0,0,0,0.1); margin: 15px 0; }
        th { background: #1a237e; color: white; padding: 12px; text-align: left; }
        td { padding: 10px 12px; border-bottom: 1px solid #eee; }
        tr:hover { background: #f5f5f5; }
        .risk-high { color: #d32f2f; font-weight: bold; }
        .risk-medium { color: #f57c00; font-weight: bold; }
        .risk-low { color: #388e3c; font-weight: bold; }
        .bar-container { background: #e0e0e0; border-radius: 4px; height: 20px; width: 100%; }
        .bar-fill { height: 100%; border-radius: 4px; }
        .timestamp { color: #999; font-size: 0.9em; margin-top: 30px; }
    </style>
</head>
<body>
<div class="container">
    <h1>IAM Governance Report</h1>
    <p class="timestamp">Generated: {{ generated_at }}</p>

    <h2>Executive Summary</h2>
    <div class="summary-cards">
        <div class="card info"><div class="number">{{ total_apps }}</div><div class="label">Applications</div></div>
        <div class="card info"><div class="number">{{ total_connections }}</div><div class="label">Connections</div></div>
        <div class="card critical"><div class="number">{{ total_gaps }}</div><div class="label">Governance Gaps</div></div>
        <div class="card critical"><div class="number">{{ critical_items }}</div><div class="label">Critical Items</div></div>
    </div>

    <h2>Risk Distribution</h2>
    <div class="summary-cards">
        <div class="card critical"><div class="number">{{ high_risk }}</div><div class="label">HIGH Risk Apps</div></div>
        <div class="card warning"><div class="number">{{ medium_risk }}</div><div class="label">MEDIUM Risk Apps</div></div>
        <div class="card success"><div class="number">{{ low_risk }}</div><div class="label">LOW Risk Apps</div></div>
    </div>

    <h2>Department Scorecard</h2>
    <table>
        <tr><th>Department</th><th>Apps</th><th>High Risk</th><th>Gaps</th><th>Critical</th><th>Coverage</th></tr>
        {% for row in scorecard %}
        <tr>
            <td>{{ row.department }}</td>
            <td>{{ row.total_apps }}</td>
            <td class="risk-high">{{ row.high_risk }}</td>
            <td>{{ row.total_gaps }}</td>
            <td class="risk-high">{{ row.critical_gaps }}</td>
            <td>
                <div class="bar-container">
                    <div class="bar-fill" style="width: {{ row.coverage_pct }}%; background: {% if row.coverage_pct >= 80 %}#388e3c{% elif row.coverage_pct >= 50 %}#f57c00{% else %}#d32f2f{% endif %};"></div>
                </div>
                {{ row.coverage_pct }}%
            </td>
        </tr>
        {% endfor %}
    </table>

    <h2>Top Remediation Items</h2>
    <table>
        <tr><th>App</th><th>Gap Type</th><th>Severity</th><th>Detail</th><th>Deadline</th></tr>
        {% for row in top_remediation %}
        <tr>
            <td>{{ row.app_name }}</td>
            <td>{{ row.gap_type }}</td>
            <td class="risk-{{ row.severity | lower }}">{{ row.severity }}</td>
            <td>{{ row.detail }}</td>
            <td>{{ row.remediation_deadline }}</td>
        </tr>
        {% endfor %}
    </table>
</div>
</body>
</html>"""


def generate_master_map(merged_df, output_dir):
    # Select the columns that form the unified governance map
    cols = ['app_id', 'app_name', 'department', 'owner', 'sensitivity',
            'business_critical', 'environment', 'iam_product',
            'connection_status', 'last_review_date', 'risk_level']
    available_cols = [c for c in cols if c in merged_df.columns]
    merged_df[available_cols].to_csv(
        os.path.join(output_dir, 'iam_connection_master_map.csv'), index=False
    )


def generate_gap_report(gaps_df, output_dir):
    # Write all detected governance gaps to a CSV for application owners
    gaps_df.to_csv(os.path.join(output_dir, 'governance_gaps.csv'), index=False)


def generate_html_dashboard(merged_df, gaps_df, scorecard_df, output_dir):
    # Calculate summary statistics for the executive dashboard
    apps_unique = merged_df.drop_duplicates(subset='app_id')
    total_apps = len(apps_unique)
    total_connections = len(merged_df[merged_df['iam_product'].notna()])
    total_gaps = len(gaps_df)
    critical_items = len(gaps_df[gaps_df['severity'] == 'HIGH']) if not gaps_df.empty else 0
    high_risk = len(apps_unique[apps_unique['risk_level'] == 'HIGH'])
    medium_risk = len(apps_unique[apps_unique['risk_level'] == 'MEDIUM'])
    low_risk = len(apps_unique[apps_unique['risk_level'] == 'LOW'])

    # Format the top remediation items for display
    top_remediation = gaps_df.head(15).to_dict('records') if not gaps_df.empty else []
    for item in top_remediation:
        if 'remediation_deadline' in item and pd.notna(item['remediation_deadline']):
            item['remediation_deadline'] = pd.Timestamp(item['remediation_deadline']).strftime('%Y-%m-%d')

    # Render the Jinja2 template with all computed values
    template = Template(DASHBOARD_TEMPLATE)
    html = template.render(
        generated_at=datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        total_apps=total_apps,
        total_connections=total_connections,
        total_gaps=total_gaps,
        critical_items=critical_items,
        high_risk=high_risk,
        medium_risk=medium_risk,
        low_risk=low_risk,
        scorecard=scorecard_df.to_dict('records'),
        top_remediation=top_remediation
    )

    # Write the rendered HTML to the output directory
    with open(os.path.join(output_dir, 'iam_governance_report.html'), 'w', encoding='utf-8') as f:
        f.write(html)


def write_run_log(output_dir, stats):
    # Append a timestamped entry to the pipeline run log
    log_path = os.path.join(output_dir, 'pipeline_run_log.txt')
    with open(log_path, 'a', encoding='utf-8') as f:
        f.write(f"\n{'='*60}\n")
        f.write(f"Pipeline Run: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
        f.write(f"{'='*60}\n")
        for key, value in stats.items():
            f.write(f"  {key}: {value}\n")
        f.write(f"{'='*60}\n")

Your reporting layer is complete. Next up, you will assemble all the modules into the main pipeline script, run it end to end, and open the HTML dashboard to see your governance data visualised.

Assemble and Run the Pipeline

You have built every module your governance pipeline needs. The data processor, risk classifier, gap detector, rules engine, scorecard generator, and report generator are all complete and tested individually.

Now you need a single entry point that calls each module in the right order, passes data between them, and writes all outputs to disk. That entry point is pipeline.py.

In this step, get ready to:

• Write the pipeline orchestrator that chains all modules together.
• Run the full pipeline and verify all 6 output files are generated.
• Review the HTML governance dashboard and CSV exports in your browser.

Write the pipeline orchestrator

The orchestrator imports every function you built across the src/ modules, then calls them in a strict sequence. Each phase feeds its output into the next.

• In VS Code, open pipeline.py at the project root (it should be empty).
• Add the imports and path setup by pasting the following code:

import os
import sys

# Ensure the project root is on the Python path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

# Import all pipeline modules
from src.data_processor import (
    load_application_inventory,
    load_iam_product_registry,
    load_connection_log,
    consolidate_data
)
from src.risk_classifier import apply_risk_classification
from src.gap_detector import run_all_gap_detection
from src.rules_engine import load_feeding_rules, apply_sla_to_gaps
from src.scorecard import generate_department_scorecard, generate_remediation_queue
from src.report_generator import (
    generate_master_map,
    generate_gap_report,
    generate_html_dashboard,
    write_run_log
)

What does this code do?

The sys.path.insert line ensures Python can find your src/ package regardless of which directory you run the script from.

Each from src...import line pulls in the specific functions you built in previous steps. This keeps pipeline.py clean. It only orchestrates, never implements.

You won't see a change yet. The next chunk defines the main() function that uses all these imports.

• Below the imports, add the main() function and entry point by pasting this code:

def main():
    # Phase 1: Load all three data sources
    print("[1/5] Loading data sources...")
    app_df = load_application_inventory('data/application_inventory.csv')
    registry = load_iam_product_registry('data/iam_product_registry.json')
    conn_df = load_connection_log('data/connection_log.csv')
    print(f"  Loaded: {len(app_df)} applications, {len(conn_df)} connections")

    # Phase 2: Apply feeding rules and merge data
    print("[2/5] Applying feeding rules and consolidating data...")
    rules = load_feeding_rules('feeding_rules.yaml')
    merged_df, registry = consolidate_data(app_df, conn_df, registry)
    print(f"  Consolidated: {len(merged_df)} records")

    # Phase 3: Classify risk for every record
    print("[3/5] Classifying risk...")
    merged_df = apply_risk_classification(merged_df)
    high = len(merged_df[merged_df['risk_level'] == 'HIGH'])
    medium = len(merged_df[merged_df['risk_level'] == 'MEDIUM'])
    low = len(merged_df[merged_df['risk_level'] == 'LOW'])
    print(f"  Risk distribution: HIGH={high}, MEDIUM={medium}, LOW={low}")

    # Phase 4: Detect governance gaps and apply SLAs
    print("[4/5] Detecting governance gaps...")
    gaps_df = run_all_gap_detection(merged_df, registry)
    gaps_df = apply_sla_to_gaps(gaps_df, rules)
    critical = len(gaps_df[gaps_df['severity'] == 'HIGH']) if not gaps_df.empty else 0
    print(f"  Detected: {len(gaps_df)} gaps ({critical} CRITICAL)")

    # Phase 5: Generate all output files
    print("[5/5] Generating outputs...")
    os.makedirs('output', exist_ok=True)

    generate_master_map(merged_df, 'output')
    generate_gap_report(gaps_df, 'output')

    scorecard_df = generate_department_scorecard(merged_df, gaps_df)
    scorecard_df.to_csv('output/department_risk_scorecard.csv', index=False)

    remediation_df = generate_remediation_queue(gaps_df)
    remediation_df.to_csv('output/remediation_priority_queue.csv', index=False)

    generate_html_dashboard(merged_df, gaps_df, scorecard_df, 'output')

    # Write audit trail
    stats = {
        'Applications processed': len(app_df),
        'Connections analysed': len(conn_df),
        'Governance gaps detected': len(gaps_df),
        'Critical items': critical,
        'Departments covered': merged_df['department'].nunique()
    }
    write_run_log('output', stats)

    print("  Master map, gap report, scorecard, remediation queue, HTML dashboard")
    print(f"  {critical} CRITICAL items requiring immediate action")
    print("\nPipeline complete. Open output/iam_governance_report.html to view dashboard.")


if __name__ == '__main__':
    main()

What does this code do?

The main() function runs 5 phases in strict order, printing progress at each stage:

• Phase 1 loads all three data sources (CSV inventory, JSON registry, CSV connection log).
• Phase 2 loads the YAML feeding rules and consolidates the three sources into a single merged DataFrame.
• Phase 3 applies risk classification (HIGH/MEDIUM/LOW) to every record.
• Phase 4 runs all 5 gap detection algorithms and attaches SLA deadlines from the feeding rules.
• Phase 5 writes 6 output files: the master map, gap report, department scorecard, remediation queue, HTML dashboard, and pipeline run log.

The if __name__ == '__main__' guard ensures main() only runs when you execute the file directly, not when it is imported by another script.

• Save pipeline.py.

Seeing a syntax error on save?

Make sure the main() function starts below a blank line after the last import block. If you accidentally pasted inside the import parentheses, Python will raise a SyntaxError.

Also check that the f-strings use curly braces {len(app_df)} and not square brackets.

help me fix a SyntaxError in my pipeline.py orchestrator file

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py before moving on.

ⓧ I'd like to double check the full code

import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from src.data_processor import (
    load_application_inventory,
    load_iam_product_registry,
    load_connection_log,
    consolidate_data
)
from src.risk_classifier import apply_risk_classification
from src.gap_detector import run_all_gap_detection
from src.rules_engine import load_feeding_rules, apply_sla_to_gaps
from src.scorecard import generate_department_scorecard, generate_remediation_queue
from src.report_generator import (
    generate_master_map,
    generate_gap_report,
    generate_html_dashboard,
    write_run_log
)


def main():
    print("[1/5] Loading data sources...")
    app_df = load_application_inventory('data/application_inventory.csv')
    registry = load_iam_product_registry('data/iam_product_registry.json')
    conn_df = load_connection_log('data/connection_log.csv')
    print(f"  Loaded: {len(app_df)} applications, {len(conn_df)} connections")

    print("[2/5] Applying feeding rules and consolidating data...")
    rules = load_feeding_rules('feeding_rules.yaml')
    merged_df, registry = consolidate_data(app_df, conn_df, registry)
    print(f"  Consolidated: {len(merged_df)} records")

    print("[3/5] Classifying risk...")
    merged_df = apply_risk_classification(merged_df)
    high = len(merged_df[merged_df['risk_level'] == 'HIGH'])
    medium = len(merged_df[merged_df['risk_level'] == 'MEDIUM'])
    low = len(merged_df[merged_df['risk_level'] == 'LOW'])
    print(f"  Risk distribution: HIGH={high}, MEDIUM={medium}, LOW={low}")

    print("[4/5] Detecting governance gaps...")
    gaps_df = run_all_gap_detection(merged_df, registry)
    gaps_df = apply_sla_to_gaps(gaps_df, rules)
    critical = len(gaps_df[gaps_df['severity'] == 'HIGH']) if not gaps_df.empty else 0
    print(f"  Detected: {len(gaps_df)} gaps ({critical} CRITICAL)")

    print("[5/5] Generating outputs...")
    os.makedirs('output', exist_ok=True)

    generate_master_map(merged_df, 'output')
    generate_gap_report(gaps_df, 'output')

    scorecard_df = generate_department_scorecard(merged_df, gaps_df)
    scorecard_df.to_csv('output/department_risk_scorecard.csv', index=False)

    remediation_df = generate_remediation_queue(gaps_df)
    remediation_df.to_csv('output/remediation_priority_queue.csv', index=False)

    generate_html_dashboard(merged_df, gaps_df, scorecard_df, 'output')

    stats = {
        'Applications processed': len(app_df),
        'Connections analysed': len(conn_df),
        'Governance gaps detected': len(gaps_df),
        'Critical items': critical,
        'Departments covered': merged_df['department'].nunique()
    }
    write_run_log('output', stats)

    print("  Master map, gap report, scorecard, remediation queue, HTML dashboard")
    print(f"  {critical} CRITICAL items requiring immediate action")
    print("\nPipeline complete. Open output/iam_governance_report.html to view dashboard.")


if __name__ == '__main__':
    main()

Run the pipeline

With pipeline.py complete, you can now execute the full governance pipeline from end to end.

• Run the pipeline by executing this command in your terminal:

python pipeline.py

You should see progress output for all 5 phases, ending with a summary of detected gaps and the message Pipeline complete.

• Confirm all 6 output files were generated by running:

ls output/

You should see these 6 files listed: iam_connection_master_map.csv, governance_gaps.csv, department_risk_scorecard.csv, remediation_priority_queue.csv, iam_governance_report.html, and pipeline_run_log.txt.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is still activated. You should see (venv) at the start of your terminal prompt.

Also verify you are running the command from the iam-governance-pipeline project root, not from inside the src/ folder.

help me fix a ModuleNotFoundError when running my IAM governance pipeline

Review the dashboard and CSV outputs

The pipeline produced a professional HTML dashboard for management stakeholders and structured CSVs for audit systems. Time to verify them.

• Open output/iam_governance_report.html in your browser.
• You can do this by right-clicking the file in VS Code's Explorer sidebar and selecting Reveal in Finder (macOS) or Reveal in File Explorer (Windows), then double-clicking the HTML file.

You should see a dashboard with four sections: an Executive Summary showing total applications, connections, gaps, and critical items. A Risk Distribution section with HIGH, MEDIUM, and LOW counts. A Department Scorecard table with coverage percentages. A Top Remediation Items table listing the most urgent gaps.

What do the coverage percentages mean?

Each department's coverage percentage represents how many of its potential governance slots are gap-free. A coverage of 100% means every required IAM product connection exists, is documented, has a known owner, and has been reviewed on schedule.

Departments with lower coverage have more gaps relative to their application count. These are the ones a Control Tower team would prioritise for remediation.

• Open output/governance_gaps.csv in VS Code to inspect the gap records.
• Verify you can see columns for app_id, gap_type, severity, remediation_deadline, and notification_channel.
• Open output/remediation_priority_queue.csv and confirm that HIGH severity gaps appear at the top, sorted by deadline.

Why are there multiple output formats?

The feeding rules define different communication channels for different audiences. CSV files feed into GRC platforms and ServiceNow tickets automatically. The HTML dashboard goes on the intranet for management visibility.

This mirrors real enterprise IAM governance where a single pipeline run produces outputs consumed by auditors, application owners, and executives through different channels.

Dashboard looks empty or shows zero gaps?

Check that your data/connection_log.csv contains intentional gaps (missing review dates, empty statuses). The gap detector relies on these data quality issues to find governance failures.

Also confirm your data/iam_product_registry.json has required_for_sensitivity arrays populated. Without these, the MISSING_CONNECTION detector has no basis for flagging gaps.

help me debug why my IAM governance dashboard shows zero gaps

Secret mission

Add SCIM Provisioning Simulation

Real enterprise IAM doesn't stop at mapping applications to products. Identity lifecycle automation is where governance meets the real world. In this secret mission, you'll simulate how SailPoint provisions users to CyberArk via SCIM, detect drift between provisioning events and your governance map, and surface provisioning health on your dashboard.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing costs.

Resources you used:

• The iam-governance-pipeline project folder on your Desktop (includes all data files, source code, output files, and the virtual environment).

Keep everything running

No action needed. Choose this if you're still actively building or want to revisit the pipeline later.

• Your project folder stays on your Desktop with all source code, data files, and generated outputs intact.
• The virtual environment (venv/) retains all installed packages so you can reactivate it and run the pipeline again anytime.
• To pick up where you left off, open the iam-governance-pipeline folder in VS Code and reactivate the virtual environment.

Pause - I'll come back to this later

Since this project is entirely local and uses no running processes, there is nothing to pause. Your files sit idle on disk and consume no memory or CPU.

• Deactivate the virtual environment (if still active) by running:
• Close VS Code to free up memory.

deactivate

When you're ready to return, reopen the folder in VS Code and reactivate the environment.

Delete - I don't want to use this again

Remove all project files from your machine. This deletes your source code, synthetic data, generated outputs, and the virtual environment.

• Deactivate the virtual environment first (if still active):

deactivate

• Delete the entire project folder by running one of the following commands from your Desktop:

macOS

cd ~/Desktop
rm -rf iam-governance-pipeline

Windows

cd ~/Desktop
rmdir /s /q iam-governance-pipeline

• Confirm the folder is gone by running:

ls

You should no longer see iam-governance-pipeline in the output.

Nice Work!

Nice Work!

You just built a complete IAM Application Risk Governance Pipeline that maps 75 enterprise applications to 5 real IAM products, classifies risk across three tiers, and detects governance gaps. This is the exact type of system a Control Tower team at a company like CMA CGM would build and maintain.

You've learned how to:

• Build a Python data pipeline that ingests three enterprise data sources, consolidates them into a unified governance map, and applies three-tier risk classification across 75 applications mapped to 5 real IAM products (Microsoft Entra ID, SailPoint, CyberArk).
• Create a YAML-based feeding rules engine defining authoritative data sources, conflict resolution logic, and remediation SLAs with escalation timelines and communication channel mappings.
• Generate audit-ready outputs including a master governance map, gap reports, department scorecards, a remediation priority queue, and a professional HTML dashboard for management stakeholders.
• Add a SCIM provisioning simulator modelling SailPoint-to-CyberArk identity lifecycle flows with drift detection that flags discrepancies between expected provisioning state and actual connection state.

Ready to quiz yourself?