#!/usr/bin/env python3 # Copyright 2025 Ping Identity Corporation. All Rights Reserved # # This code is to be used exclusively in connection with Ping Identity # Corporation software or services. Ping Identity Corporation only offers # such software or services to legal entities who have entered into a # binding license agreement with Ping Identity Corporation. # -*- coding: utf-8 -*- """ This program will create IGA Applications based on the arguments provided. It is intended to run standalone and can be used inside a conf.yaml using the ShellTask. It requires at the minimum three arguments: --tenant-url, --tenant-admin, --tenant-password The program will create an Application Owner if --create-owner is provided and similarly create an Authrotative Source if --with-as is provided. These are needed if you are creating Applicaitons for the very first time on a fresh tenant or after a data reset. If you want to just create an authoratiave source then use --as-only. For creating applications you have to provide --start-with and --end-with. For example just to create Applicaiton0 use --start-with 0 --end-with 0. Or if you want to create applications 5 onward then use --start-with 5 --end-with XX. Use --end-with 101 to create 100 applications. Finally use --dry-run to see what the program will do without actually creating anything. TODO: Add support for running recon TODO: Add support for recon status check TODO: Add support for creating files insead of making HTTP calls TODO: Make the program more modular and thus readable """ # Python imports import time import argparse import json import requests ############ Enable for Debugging only ####### # import logging # import http.client # logging.basicConfig(level=logging.DEBUG) # http.client.HTTPConnection.debuglevel = 1 ############################################## parser = argparse.ArgumentParser() parser.add_argument("--start-with", default=0, help="Application number to start with", type=int) parser.add_argument("--end-with", default=101, help="Application number to end with", type=int) parser.add_argument("--create-owner", action="store_true", help="Create Application Owner") parser.add_argument("--as-only", action="store_true", help="Create Authoritative Source Only") parser.add_argument("--with-as", action="store_true", help="Create Authoritative Source Also") parser.add_argument("--dry-run", action="store_true", help="Dry run") parser.add_argument("--files-only", action="store_true", help="Generate files only") # Not implemented yet parser.add_argument("--role-file", default=None, help="Create roles from a file") parser.add_argument("--appid-file", default=None, help="Deletes the applications in the file") parser.add_argument("--tenant-url", default=None, help="Tenant Base URL. For example https://openam-perf-iga-deve1.forgeblocks.com") parser.add_argument("--tenant-admin", default=None, help="Tenant Admin ID. For example foo.bar@pingidentity.com") parser.add_argument("--tenant-password", default=None, help="Tenant Admin Password") args = parser.parse_args() START_WITH = args.start_with END_WITH = args.end_with CREATE_OWNER = True if args.create_owner else False AS_ONLY = True if args.as_only else False WITH_AS = True if args.with_as else False FILES_ONLY = args.files_only DRY_RUN = True if FILES_ONLY else args.dry_run BASE_URL = args.tenant_url TENANT_ADMIN = args.tenant_admin TENANT_ADMIN_PASSWORD = args.tenant_password ROLE_FILE = args.role_file APP_ID_FILE = args.appid_file if not DRY_RUN: if BASE_URL is None or TENANT_ADMIN is None or TENANT_ADMIN_PASSWORD is None: parser.error("--tenant-url and --tenant-admin and --tenant-password are required") REALM_PREFIX = "alpha_" TEMPLATE_DIR = "templates" IS_TENANT = True IDM_URL = f"{BASE_URL}/openidm" MANAGED_USER_URL = f"{IDM_URL}/managed/{REALM_PREFIX}user" MANAGED_APPLICATION_URL = f"{IDM_URL}/managed/{REALM_PREFIX}application" MANAGED_ROLE_URL = f"{IDM_URL}/managed/{REALM_PREFIX}role" CONFIG_URL = f"{IDM_URL}/config" OPENICF_CONFIG_URL = f"{CONFIG_URL}/provisioner.openicf" IGA_API_URL = f"{BASE_URL}/iga" def main(): """ Creates an Application for IGA feature. This will not work on CDM. This has been reversed engineered from the Platform UI. Unfortunately there is no single API to create an application in IGA. """ access_token = "dummy-access-token" # required for dry-run if not DRY_RUN: cookie_name, am_token = get_am_token(TENANT_ADMIN, TENANT_ADMIN_PASSWORD) access_token = get_access_token(am_token, cookie_name) if not access_token: print("=> Failed to get access_token. Exiting!!!") exit(1) headers = { "User-Agent": "gen-app-standalone", "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", } if ROLE_FILE: create_role(headers, ROLE_FILE) exit(0) if APP_ID_FILE: delete_application(headers, APP_ID_FILE) exit(0) owner = "20d93847-dbda-4b02-80da-ae8114ad7f96" # Application Owner UUID (_id) app_id_array = [] # Array to store application names # Build an array of applications names depending on the arguments apps_array = [f"Application{n}" for n in range(START_WITH, END_WITH + 1)] if AS_ONLY: apps_array = ["AuthoritativeSource"] if WITH_AS: apps_array = ["AuthoritativeSource"] + apps_array # Read all the templates into memory with open(f"{TEMPLATE_DIR}/template_sync.json", "r") as file: sync_json_blob = file.read() with open(f"{TEMPLATE_DIR}/template_mapping_application_outbound_user.json", "r") as file: outbound_app_user = file.read() with open(f"{TEMPLATE_DIR}/template_provisioner_application.json", "r") as file: provisioner_app = file.read() with open(f"{TEMPLATE_DIR}/template_mapping_application_inbound_user.json", "r") as file: inbound_app_user = file.read() with open(f"{TEMPLATE_DIR}/template_mapping_application_inbound_assignment.json", "r") as file: inbound_app_assign = file.read() with open(f"{TEMPLATE_DIR}/template_patch_payload.json", "r") as file: patch_payload_blob = file.read() # Step 0 (optional): First create the application owner user if CREATE_OWNER: url = f"{MANAGED_USER_URL}/{owner}" data = { "userName": "saint", "givenName": "Simon", "sn": "Templer", "mail": "saint@example.com", "password": "Pa_ssw0rd", } get_http_response(url=url, method="PUT", headers=headers, json=data) print("=> Application Owner Created") # Now loop through the applications and create provisoners and mappings count = 0 for app_name in apps_array: print(f"=================> Creating {app_name} <=================") # Step 1: Create the application (container) url = f"{MANAGED_APPLICATION_URL}/?_action=create" authoritative = False if app_name == "AuthoritativeSource": authoritative = True data = { "name": f"{app_name}", "owners": [{"_ref": f"managed/{REALM_PREFIX}user/{owner}", "_refProperties": {}}], "icon": "", "templateName": "ds.ldap", "templateVersion": "2.5", "authoritative": authoritative, } response = get_http_response(url=url, method="POST", headers=headers, json=data) if DRY_RUN: response_json = {"ok": True, "_id": "dummy_id", "_rev": "dummy_rev"} else: response_json = response.json() app_id = response_json["_id"] app_rev = response_json["_rev"] app_id_array.append(app_id) print("=> Step 1: Success") # Step 2: Create core configuration. Make sure bundleVersion is correct for the connector url = f"{IDM_URL}/system?_action=createCoreConfig" data = { "connectorRef": { "connectorHostRef": "connectorserver", "displayName": f"{app_name} LDAP Connector", "bundleVersion": "[1.5.0.0, 1.6.0.0]", "systemType": "provisioner.openicf", "bundleName": "org.forgerock.openicf.connectors.ldap-connector", "connectorName": "org.identityconnectors.ldap.LdapConnector", } } get_http_response(url=url, method="POST", headers=headers, json=data) print("=> Step 2: Success") # Step 2.1: Wait for completion - This step might note be needed and probably can be achieved by a sleep url = f"{CONFIG_URL}/provisioner.openicf.connectorinfoprovider?waitForCompletion=true" timeout = 120 start_time = time.time() while True: if time.time() - start_time > timeout: print("Timeout reached. Exiting") break try: response = get_http_response(url=url, headers=headers) if DRY_RUN: break if response.status_code in [200, 201]: print("=> Step 2.1: Success") break else: print( "=> Step 2.1: Retrying as expected response not received", response.status_code, response.text, ) except Exception as e: print(f"An http error occurred: {e}. Checking again in 5 seconds") time.sleep(5) # Step 3: Create full configuration using provisioner as payload url = f"{IDM_URL}/system?_action=createFullConfig" # Do not add frist and last double quotes to the base_context if authoritative: base_context = 'ou=groups,ou=identities", "ou=people,ou=identities", "ou=accounts,ou=identities' blob = provisioner_app % (base_context) elif app_name == "Application0": base_context = f'ou=app{count},ou=applications,ou=identities", "ou=people,ou=identities' blob = provisioner_app % (base_context) count += 1 else: base_context = f"ou=app{count},ou=applications,ou=identities" blob = provisioner_app % (base_context) count += 1 data = json.loads(blob) # make the blob truly json get_http_response(url=url, method="POST", headers=headers, json=data) print("=> Step 3: Success") # Step 4: Create provisioner file for applicaiton and wait for completion url = f"{OPENICF_CONFIG_URL}/{app_name}?waitForCompletion=true" get_http_response(url=url, method="PUT", headers=headers, json=data) # use data from step3 again print("=> Step 4: Success") # Step 5: Create inbound mapping file for application if authoritative: url = f"{CONFIG_URL}/sync?waitForCompletion=true" data = json.loads(sync_json_blob % (REALM_PREFIX, REALM_PREFIX, REALM_PREFIX, REALM_PREFIX)) else: url = f"{CONFIG_URL}/mapping/system{app_name}User_managedAlpha_user?waitForCompletion=true" data = json.loads(inbound_app_user % (app_name, app_name, app_name, REALM_PREFIX, app_name)) get_http_response(url=url, method="PUT", headers=headers, json=data) print("=> Step 5: Success") if authoritative: continue # skip the rest of the steps for authoritative only # Step 6: Create outbound mapping file for application url = f"{CONFIG_URL}/mapping/managedAlpha_user_system{app_name}User?waitForCompletion=true" data = json.loads( outbound_app_user % (app_name, app_name, REALM_PREFIX, app_name, app_id, app_name, app_id, app_name, app_name) ) get_http_response(url=url, method="PUT", headers=headers, json=data) print("=> Step 6: Success") # Step 7: Create inbound mapping file for application assignment url = f"{CONFIG_URL}/mapping/system{app_name}Group_managedAlpha_assignment?waitForCompletion=true" data = json.loads( inbound_app_assign % (app_name, app_name, REALM_PREFIX, app_name, app_name, app_name, app_name) ) get_http_response(url=url, method="PUT", headers=headers, json=data) print("=> Step 7: Success") # Step 8: Check if application is ready url = f"{IDM_URL}/system/{app_name}?_action=test" data = {} timeout = 120 start_time = time.time() while True: if DRY_RUN: print("DRY_RUN is set. Skipping...") break if time.time() - start_time > timeout: print("Timeout reached. Exiting") break try: response = get_http_response(url=url, method="POST", headers=headers, json=data) if response.status_code in [200, 201]: print("=> Step 8: Success") response_json = response.json() if response_json["ok"] == True: print("Application is ready") break else: print( "=> Step 8: Retrying as expected json response not received...", response.status_code, response.text, ) except Exception as e: print(f"An http error occurred: {e}. Checking again in 5 seconds...") time.sleep(5) # Step 9: Patch the application. This is needed for provisioner tab in the UI to work properly headers2 = headers.copy() # shallow copy headers2["if-match"] = f"{app_rev}" # add "if-match" to headers which is required for this patch url = f"{IDM_URL}/managed/{REALM_PREFIX}application/{app_id}" data = json.loads(patch_payload_blob % (app_name, app_name, app_name, app_name)) get_http_response(url=url, method="PATCH", headers=headers2, json=data) print("=> Step 9: Success") # Step 10: Create Glossary (on IGA enabled tenant) if IS_TENANT: if not DRY_RUN: # Check if IGA health endpoint is responding indicating that IGA feature is enabled health = f"{IGA_API_URL}/health" response = get_http_response(url=health, headers=headers) if response.status_code in [200, 201]: url = f"{IGA_API_URL}/governance/application/{app_id}/glossary" data = {} get_http_response(url=url, method="POST", headers=headers, json=data) print("=> Step 10: Success") # The app_id are written to a file for future use with open("app_id.txt", "w") as file: for id in app_id_array: file.write(f"{id}\n") ######### end main() ######### def delete_application(headers, app_id_file): """ Deletes all IGA application in the file The provisiioner and mapping files are not deleted Delete them manually if needed via customer-config """ with open(app_id_file, "r") as file: app_id = file.read().strip() for id in app_id.split("\n"): url = f"{IDM_URL}/managed/alpha_application/{id}" response = get_http_response(url=url, method='DELETE', headers=headers) if response.status_code in [200, 201]: print(f"Application {app_id} deleted") else: print(f"Error deleting application {app_id}: {response.status_code}, {response.text}") def create_role(headers, role_file): """ Creates roles from json file into IDM """ with open(role_file, "r") as file: role = file.read() role_json = json.loads(role) url = f"{IDM_URL}/managed/alpha_role?_action=create" for item in role_json: data = { "name": f"{item['roleName']}", "condition": f"{item['condition']}" } start_time = time.time() response = get_http_response(url=url, method='POST', headers=headers, json=data, timeout=1200) if response.status_code in [200, 201]: print(f"Role {item['roleName']} created in {time.time() - start_time} seconds") else: print(f"Error creating role {item['roleName']}: {response.status_code}, {response.text}") def get_am_token(username, password): headers = {"Accept-API-Version": "resource=2.1, protocol=1.0", "Content-Type": "application/json"} authn_url = f"{BASE_URL}/am/json/realms/root/authenticate" response = get_http_response(url=authn_url, method="POST", json={}, headers=headers) response_json = response.json() response_json["callbacks"][0]["input"][0]["value"] = username response_json["callbacks"][1]["input"][0]["value"] = password response = get_http_response(url=authn_url, method="POST", json=response_json, headers=headers) response_json = response.json() response_json["callbacks"][2]["input"][0]["value"] = "Skip" response = get_http_response(url=authn_url, method="POST", json=response_json, headers=headers) cookie = response.cookies.get_dict() for key, value in cookie.items(): if len(key) == 15: cookie_name = key cookie_value = value # this is the tokenId break #token_id = response.json.get('tokenId') # You can get the tokenId from the response too return cookie_name, cookie_value def get_access_token(am_token, cookie_name): authz_url = f"{BASE_URL}/am/oauth2/access_token" client_id = "idmAdminClient" redirect_uri = f"{BASE_URL}/platform/appAuthHelperRedirect.html" scope = "fr:idm:*" code_challenge = "H8VHP73pYVOJ0f7Y9lG3J5DU3gjTBCElj1L3LM6FgO" code_verifier = code_challenge # becaue it is plain url = f"{BASE_URL}/am/oauth2/realms/root/authorize" headers = {"Content-Type": "application/x-www-form-urlencoded", "Cookie": f"{cookie_name}={am_token}"} pkce_data = { "client_id": f"{client_id}", "redirect_uri": f"{redirect_uri}", "response_type": "code", "prompt": "none", "nonce": "12345", "decision": "allow", "csrf": f"{am_token}", "scope": f"{scope}", "code_challenge": f"{code_challenge}", "code_challenge_method": "plain", } response = get_http_response(url=url, method="POST", headers=headers, data=pkce_data) code = response.url.split("code=")[1].split("&")[0] token_data = { "grant_type": "authorization_code", "code": f"{code}", "redirect_uri": f"{redirect_uri}", "client_id": f"{client_id}", "code_verifier": f"{code_verifier}", } response = get_http_response(url=authz_url, method="POST", headers=headers, data=token_data) if response.status_code == 200: token_info = response.json() access_token = token_info["access_token"] return access_token else: print(f"Error in getting access_token: {response.status_code}, {response.text}") return None def get_http_response(url, method="GET", params=None, data=None, json=None, headers=None, timeout=60): """ Sends an HTTP request and returns the response. Args: url (str): The URL to send the request to. method (str): The HTTP method to use (GET, POST, etc.). Defaults to 'GET'. params (dict): URL parameters for GET requests. Defaults to None. data (dict): Form data for POST requests. Defaults to None. json (dict): JSON payload for POST requests. Defaults to None. headers (dict): Custom headers to send with the request. Defaults to None. timeout (int): The maximum time to wait for a response (in seconds). Defaults to 10. Returns: response: The response object from the `requests` library. """ if DRY_RUN: print("============= Fake HTTP request ================ \n") print(f"URL : {url}") print(f"Method : {method}") print(f"Params : {params}") print(f"Data : {data}") print(f"JSON : {json}") print(f"Headers: {headers}") else: try: # Choose the correct HTTP method if method.upper() == "GET": response = requests.get(url, params=params, headers=headers, timeout=timeout) elif method.upper() == "POST": response = requests.post(url, data=data, json=json, headers=headers, timeout=timeout) elif method.upper() == "PUT": response = requests.put(url, data=data, json=json, headers=headers, timeout=timeout) elif method.upper() == "PATCH": response = requests.patch(url, data=data, json=json, headers=headers, timeout=timeout) elif method.upper() == "DELETE": response = requests.delete(url, headers=headers, timeout=timeout) else: raise ValueError("Unsupported HTTP method") # Check if the response was successful response.raise_for_status() # Return the response object return response except requests.exceptions.Timeout: print("The request timed out.") except requests.exceptions.RequestException as e: print(f"An error occurred: {e}") except ValueError as ve: print(ve) return None if __name__ == "__main__": main()