LAB 64 ‐ AI ‐ Extract all user, group,dynamic group and policy in every compartment for IAM family in CSV & also to object storage - DevWithNehaCloud/Oracle-Cloud-Infrastructure-wiki GitHub Wiki

python script

import oci
import csv


# ============================================================
# OCI IAM INVENTORY
# ============================================================
#
# This script extracts:
#
# 1. All Users
# 2. All Groups
# 3. All Dynamic Groups
# 4. All Policies from every compartment
#
# Output:
#
# CSV file:
# IAM Inventory.csv
#
# Object Storage bucket:
# compute-capacity-inventory
#
# Authentication:
# OCI config file (~/.oci/config)
#
# ============================================================


# ============================================================
# OCI AUTHENTICATION
# ============================================================

# Uses OCI config file (~/.oci/config)
config = oci.config.from_file()


# ============================================================
# OCI REGION
# ============================================================

region = "us-ashburn-1"

# Add region into config
config["region"] = region


# ============================================================
# OCI IDENTITY CLIENT
# ============================================================

identity = oci.identity.IdentityClient(config)


# ============================================================
# OBJECT STORAGE CLIENT
# ============================================================

object_storage = oci.object_storage.ObjectStorageClient(config)


# ============================================================
# TENANCY OCID
# ============================================================

tenancy_id = config["tenancy"]


# ============================================================
# OBJECT STORAGE CONFIGURATION
# ============================================================

bucket_name = "compute-capacity-inventory"

object_name = "IAM Inventory.csv"


# Get Object Storage namespace automatically
namespace = object_storage.get_namespace().data


# ============================================================
# START MESSAGE
# ============================================================

print("\n============================================================")
print("                 OCI IAM INVENTORY")
print("============================================================")

print(f"Region       : {region}")
print(f"Namespace    : {namespace}")
print(f"Bucket       : {bucket_name}")
print(f"Object Name  : {object_name}")


# ============================================================
# GET ALL COMPARTMENTS
# ============================================================

print("\n------------------------------------------------------------")
print("STEP 1 - Getting Compartments")
print("------------------------------------------------------------")


compartments = [
    {
        "id": tenancy_id,
        "name": "Tenancy Root"
    }
]


try:

    compartment_response = oci.pagination.list_call_get_all_results(
        identity.list_compartments,
        tenancy_id,
        compartment_id_in_subtree=True,
        access_level="ANY"
    )


    for compartment in compartment_response.data:

        if compartment.lifecycle_state == "ACTIVE":

            compartments.append({
                "id": compartment.id,
                "name": compartment.name
            })


    print(
        f"Total compartments scanned: "
        f"{len(compartments)}"
    )


except Exception as e:

    print("\nERROR while getting compartments.")

    print(e)

    raise


# ============================================================
# COUNTERS
# ============================================================

total_users = 0

total_groups = 0

total_dynamic_groups = 0

total_policies = 0


# ============================================================
# CSV ROWS
# ============================================================

rows = []


# ============================================================
# STEP 2 - GET ALL USERS
# ============================================================

print("\n------------------------------------------------------------")
print("STEP 2 - Getting Users")
print("------------------------------------------------------------")


try:

    users = oci.pagination.list_call_get_all_results(
        identity.list_users,
        tenancy_id
    ).data


    for user in users:

        total_users += 1


        rows.append([

            region,

            "User",

            "Tenancy Root",

            user.name,

            user.id,

            user.description or "",

            user.lifecycle_state,

            user.email or "",

            "",

            "",

            "",

            user.time_created

        ])


    print(
        f"Total users found: "
        f"{total_users}"
    )


except Exception as e:

    print("\nERROR while getting users.")

    print(e)


# ============================================================
# STEP 3 - GET ALL GROUPS
# ============================================================

print("\n------------------------------------------------------------")
print("STEP 3 - Getting Groups")
print("------------------------------------------------------------")


try:

    groups = oci.pagination.list_call_get_all_results(
        identity.list_groups,
        tenancy_id
    ).data


    for group in groups:

        total_groups += 1


        rows.append([

            region,

            "Group",

            "Tenancy Root",

            group.name,

            group.id,

            group.description or "",

            group.lifecycle_state,

            "",

            "",

            "",

            "",

            group.time_created

        ])


    print(
        f"Total groups found: "
        f"{total_groups}"
    )


except Exception as e:

    print("\nERROR while getting groups.")

    print(e)


# ============================================================
# STEP 4 - GET ALL DYNAMIC GROUPS
# ============================================================

print("\n------------------------------------------------------------")
print("STEP 4 - Getting Dynamic Groups")
print("------------------------------------------------------------")


try:

    dynamic_groups = oci.pagination.list_call_get_all_results(
        identity.list_dynamic_groups,
        tenancy_id
    ).data


    for dynamic_group in dynamic_groups:

        total_dynamic_groups += 1


        rows.append([

            region,

            "Dynamic Group",

            "Tenancy Root",

            dynamic_group.name,

            dynamic_group.id,

            dynamic_group.description or "",

            dynamic_group.lifecycle_state,

            "",

            "",

            "",

            "",

            dynamic_group.time_created

        ])


    print(
        f"Total dynamic groups found: "
        f"{total_dynamic_groups}"
    )


except Exception as e:

    print("\nERROR while getting dynamic groups.")

    print(e)


# ============================================================
# STEP 5 - GET POLICIES FROM EVERY COMPARTMENT
# ============================================================

print("\n------------------------------------------------------------")
print("STEP 5 - Getting Policies")
print("------------------------------------------------------------")


for compartment in compartments:

    compartment_id = compartment["id"]

    compartment_name = compartment["name"]


    print(
        f"Scanning policies in compartment: "
        f"{compartment_name}"
    )


    try:

        policies = oci.pagination.list_call_get_all_results(
            identity.list_policies,
            compartment_id
        ).data


        for policy in policies:

            total_policies += 1


            # ------------------------------------------------
            # POLICY STATEMENTS
            # ------------------------------------------------

            policy_statements = ""


            if policy.statements:

                policy_statements = " | ".join(
                    policy.statements
                )


            # ------------------------------------------------
            # POLICY VERSION
            # ------------------------------------------------

            policy_version = (
                policy.version_date
                if policy.version_date
                else ""
            )


            # ------------------------------------------------
            # POLICY TIME CREATED
            # ------------------------------------------------

            time_created = (
                policy.time_created
                if policy.time_created
                else ""
            )


            rows.append([

                region,

                "Policy",

                compartment_name,

                policy.name,

                policy.id,

                policy.description or "",

                policy.lifecycle_state,

                "",

                policy_version,

                policy_statements,

                policy.compartment_id,

                time_created

            ])


    except Exception as e:

        print(
            f"Unable to get policies from: "
            f"{compartment_name}"
        )

        print(
            f"Error: {e}"
        )


print(
    f"Total policies found: "
    f"{total_policies}"
)


# ============================================================
# CREATE CSV
# ============================================================

print("\n------------------------------------------------------------")
print("STEP 6 - Creating CSV")
print("------------------------------------------------------------")


csv_file = "IAM Inventory.csv"


with open(
    csv_file,
    "w",
    newline="",
    encoding="utf-8-sig"
) as file:


    writer = csv.writer(
        file,
        delimiter=",",
        quoting=csv.QUOTE_MINIMAL
    )


    # ========================================================
    # CSV HEADERS
    # ========================================================

    writer.writerow([

        "Region",

        "Resource Type",

        "Compartment",

        "Name",

        "OCID",

        "Description",

        "Lifecycle State",

        "Email",

        "Policy Version",

        "Policy Statements",

        "Policy Compartment OCID",

        "Time Created"

    ])


    # ========================================================
    # WRITE DATA
    # ========================================================

    writer.writerows(
        rows
    )


print(
    f"CSV generated successfully: "
    f"{csv_file}"
)


# ============================================================
# DISPLAY SUMMARY
# ============================================================

print("\n============================================================")
print("                 OCI IAM INVENTORY")
print("============================================================")


print(
    f"Total Users           : "
    f"{total_users}"
)


print(
    f"Total Groups          : "
    f"{total_groups}"
)


print(
    f"Total Dynamic Groups  : "
    f"{total_dynamic_groups}"
)


print(
    f"Total Policies        : "
    f"{total_policies}"
)


print(
    f"Total CSV Rows        : "
    f"{len(rows)}"
)


print(
    f"\nCSV generated successfully: "
    f"{csv_file}"
)


# ============================================================
# OBJECT STORAGE UPLOAD
# ============================================================

print("\n------------------------------------------------------------")
print("STEP 7 - Uploading CSV to Object Storage")
print("------------------------------------------------------------")


try:

    with open(
        csv_file,
        "rb"
    ) as file:


        object_storage.put_object(

            namespace_name=namespace,

            bucket_name=bucket_name,

            object_name=object_name,

            put_object_body=file

        )


    print(
        "\nCSV uploaded successfully!"
    )


    print(
        f"Bucket : {bucket_name}"
    )


    print(
        f"Object : {object_name}"
    )


except Exception as e:

    print(
        "\nObject Storage upload failed."
    )


    print(
        "Please check:"
    )


    print(
        "1. Bucket name"
    )


    print(
        "2. OCI authentication"
    )


    print(
        "3. Object Storage permissions"
    )


    print(
        "4. Object Storage namespace"
    )


    print(
        f"\nError: {e}"
    )


# ============================================================
# SCRIPT COMPLETED
# ============================================================

print("\n============================================================")
print("              SCRIPT COMPLETED")
print("============================================================")


print(
    "\nOCI IAM Inventory completed successfully."
)


print(
    f"Local CSV    : {csv_file}"
)


print(
    f"Object Store : {bucket_name}"
)


print(
    f"Object Name  : {object_name}"
)


print("============================================================")

to run the script

python iam_capacity_inventory.py

verify the csv in object storage bucket