From 86fdd8483a1414602a6501c2049a3c4f408919dd Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sun, 12 Jul 2026 15:14:01 +0200 Subject: [PATCH] feat(pingone): new PingOne identity-containment integration PingOne Platform API, 6 commands: list/get users, enable/disable user, list populations. OAuth2 client-credentials auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) --- integrations/pingone/manifest.yaml | 87 +++++++++++++++++++ integrations/pingone/scripts/disable_user.py | 76 ++++++++++++++++ integrations/pingone/scripts/enable_user.py | 76 ++++++++++++++++ integrations/pingone/scripts/get_user.py | 76 ++++++++++++++++ .../pingone/scripts/list_populations.py | 70 +++++++++++++++ integrations/pingone/scripts/list_users.py | 72 +++++++++++++++ .../pingone/scripts/test_connection.py | 71 +++++++++++++++ 7 files changed, 528 insertions(+) create mode 100644 integrations/pingone/manifest.yaml create mode 100644 integrations/pingone/scripts/disable_user.py create mode 100644 integrations/pingone/scripts/enable_user.py create mode 100644 integrations/pingone/scripts/get_user.py create mode 100644 integrations/pingone/scripts/list_populations.py create mode 100644 integrations/pingone/scripts/list_users.py create mode 100644 integrations/pingone/scripts/test_connection.py diff --git a/integrations/pingone/manifest.yaml b/integrations/pingone/manifest.yaml new file mode 100644 index 0000000..40aabe1 --- /dev/null +++ b/integrations/pingone/manifest.yaml @@ -0,0 +1,87 @@ +id: pingone +name: PingOne +version: 1.0.0 +description: "Ping Identity PingOne (Platform API) — identity containment: list and read users, enable/disable a user, and list populations. OAuth2 client-credentials authentication; stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: list/get users, enable/disable user, list populations." +category: identity + +# Per-instance configuration. Client credentials are exchanged for a token at the +# environment's token endpoint; environment_id scopes all API calls. +config_schema: + properties: + auth_url: + type: string + description: "PingOne auth base URL (region-specific)" + default: "https://auth.pingone.com" + api_url: + type: string + description: "PingOne API base URL (region-specific)" + default: "https://api.pingone.com" + environment_id: + type: string + description: "PingOne environment ID" + client_id: + type: string + description: "Worker application client ID" + client_secret: + type: string + description: "Worker application client secret" + x-soar-sensitive: true + required: + - environment_id + - client_id + - client_secret + +commands: + - id: list_users + name: pingone-list-users + description: "List users (optionally filter by username or email)." + risk: read + inputs_schema: + properties: + filter: { type: string, description: "SCIM filter (e.g. username eq \"jdoe\")" } + limit: { type: number, description: "Max users (default 50)" } + required: [] + outputs_schema: { properties: {} } + - id: get_user + name: pingone-get-user + description: "Get a single user by ID." + risk: read + inputs_schema: + properties: + user_id: { type: string, description: "User ID" } + required: [user_id] + outputs_schema: { properties: {} } + - id: disable_user + name: pingone-disable-user + description: "Disable a user account (containment)." + inputs_schema: + properties: + user_id: { type: string, description: "User ID" } + required: [user_id] + outputs_schema: { properties: {} } + - id: enable_user + name: pingone-enable-user + description: "Enable a user account." + inputs_schema: + properties: + user_id: { type: string, description: "User ID" } + required: [user_id] + outputs_schema: { properties: {} } + - id: list_populations + name: pingone-list-populations + description: "List populations." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } + + - id: test_connection + name: pingone-test-connection + description: "Verify the credentials via the token exchange (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/pingone/scripts/disable_user.py b/integrations/pingone/scripts/disable_user.py new file mode 100644 index 0000000..5cf4232 --- /dev/null +++ b/integrations/pingone/scripts/disable_user.py @@ -0,0 +1,76 @@ +import json, os, sys, base64, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _auth_url(cfg): + return (str(cfg.get("auth_url") or "https://auth.pingone.com")).rstrip("/") + + +def _api_base(cfg): + return (str(cfg.get("api_url") or "https://api.pingone.com")).rstrip("/") + "/v1/environments/" + str(cfg.get("environment_id", "")) + + +def _token(cfg): + basic = base64.b64encode((str(cfg.get("client_id", "")) + ":" + str(cfg.get("client_secret", ""))).encode("utf-8")).decode("utf-8") + form = urllib.parse.urlencode({"grant_type": "client_credentials"}).encode("utf-8") + url = _auth_url(cfg) + "/" + str(cfg.get("environment_id", "")) + "/as/token" + req = urllib.request.Request(url, data=form, headers={ + "Authorization": "Basic " + basic, + "Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json", + }, method="POST") + with urllib.request.urlopen(req, timeout=60) as r: + tok = json.loads(r.read()) + if not tok.get("access_token"): + raise Exception("Token request failed: " + json.dumps(tok)) + return tok["access_token"] + + +def request(method, path, cfg, token, body=None, params=None, content_type="application/json"): + url = _api_base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + token, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = content_type + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + token = _token(cfg) + print(json.dumps(fn(cfg, token, inputs))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(cfg, token, inputs): + user_id = inputs.get("user_id") + if not user_id: + raise Exception("user_id is required") + return request("PUT", "/users/" + q(user_id) + "/enabled", cfg, token, body={"enabled": False}, content_type="application/vnd.pingidentity.user.enabled+json") + + +_run(main) diff --git a/integrations/pingone/scripts/enable_user.py b/integrations/pingone/scripts/enable_user.py new file mode 100644 index 0000000..2ed3431 --- /dev/null +++ b/integrations/pingone/scripts/enable_user.py @@ -0,0 +1,76 @@ +import json, os, sys, base64, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _auth_url(cfg): + return (str(cfg.get("auth_url") or "https://auth.pingone.com")).rstrip("/") + + +def _api_base(cfg): + return (str(cfg.get("api_url") or "https://api.pingone.com")).rstrip("/") + "/v1/environments/" + str(cfg.get("environment_id", "")) + + +def _token(cfg): + basic = base64.b64encode((str(cfg.get("client_id", "")) + ":" + str(cfg.get("client_secret", ""))).encode("utf-8")).decode("utf-8") + form = urllib.parse.urlencode({"grant_type": "client_credentials"}).encode("utf-8") + url = _auth_url(cfg) + "/" + str(cfg.get("environment_id", "")) + "/as/token" + req = urllib.request.Request(url, data=form, headers={ + "Authorization": "Basic " + basic, + "Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json", + }, method="POST") + with urllib.request.urlopen(req, timeout=60) as r: + tok = json.loads(r.read()) + if not tok.get("access_token"): + raise Exception("Token request failed: " + json.dumps(tok)) + return tok["access_token"] + + +def request(method, path, cfg, token, body=None, params=None, content_type="application/json"): + url = _api_base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + token, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = content_type + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + token = _token(cfg) + print(json.dumps(fn(cfg, token, inputs))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(cfg, token, inputs): + user_id = inputs.get("user_id") + if not user_id: + raise Exception("user_id is required") + return request("PUT", "/users/" + q(user_id) + "/enabled", cfg, token, body={"enabled": True}, content_type="application/vnd.pingidentity.user.enabled+json") + + +_run(main) diff --git a/integrations/pingone/scripts/get_user.py b/integrations/pingone/scripts/get_user.py new file mode 100644 index 0000000..c8038d4 --- /dev/null +++ b/integrations/pingone/scripts/get_user.py @@ -0,0 +1,76 @@ +import json, os, sys, base64, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _auth_url(cfg): + return (str(cfg.get("auth_url") or "https://auth.pingone.com")).rstrip("/") + + +def _api_base(cfg): + return (str(cfg.get("api_url") or "https://api.pingone.com")).rstrip("/") + "/v1/environments/" + str(cfg.get("environment_id", "")) + + +def _token(cfg): + basic = base64.b64encode((str(cfg.get("client_id", "")) + ":" + str(cfg.get("client_secret", ""))).encode("utf-8")).decode("utf-8") + form = urllib.parse.urlencode({"grant_type": "client_credentials"}).encode("utf-8") + url = _auth_url(cfg) + "/" + str(cfg.get("environment_id", "")) + "/as/token" + req = urllib.request.Request(url, data=form, headers={ + "Authorization": "Basic " + basic, + "Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json", + }, method="POST") + with urllib.request.urlopen(req, timeout=60) as r: + tok = json.loads(r.read()) + if not tok.get("access_token"): + raise Exception("Token request failed: " + json.dumps(tok)) + return tok["access_token"] + + +def request(method, path, cfg, token, body=None, params=None, content_type="application/json"): + url = _api_base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + token, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = content_type + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + token = _token(cfg) + print(json.dumps(fn(cfg, token, inputs))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(cfg, token, inputs): + user_id = inputs.get("user_id") + if not user_id: + raise Exception("user_id is required") + return request("GET", "/users/" + q(user_id), cfg, token) + + +_run(main) diff --git a/integrations/pingone/scripts/list_populations.py b/integrations/pingone/scripts/list_populations.py new file mode 100644 index 0000000..33e4800 --- /dev/null +++ b/integrations/pingone/scripts/list_populations.py @@ -0,0 +1,70 @@ +import json, os, sys, base64, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _auth_url(cfg): + return (str(cfg.get("auth_url") or "https://auth.pingone.com")).rstrip("/") + + +def _api_base(cfg): + return (str(cfg.get("api_url") or "https://api.pingone.com")).rstrip("/") + "/v1/environments/" + str(cfg.get("environment_id", "")) + + +def _token(cfg): + basic = base64.b64encode((str(cfg.get("client_id", "")) + ":" + str(cfg.get("client_secret", ""))).encode("utf-8")).decode("utf-8") + form = urllib.parse.urlencode({"grant_type": "client_credentials"}).encode("utf-8") + url = _auth_url(cfg) + "/" + str(cfg.get("environment_id", "")) + "/as/token" + req = urllib.request.Request(url, data=form, headers={ + "Authorization": "Basic " + basic, + "Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json", + }, method="POST") + with urllib.request.urlopen(req, timeout=60) as r: + tok = json.loads(r.read()) + if not tok.get("access_token"): + raise Exception("Token request failed: " + json.dumps(tok)) + return tok["access_token"] + + +def request(method, path, cfg, token, body=None, params=None, content_type="application/json"): + url = _api_base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + token, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = content_type + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + token = _token(cfg) + print(json.dumps(fn(cfg, token, inputs))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, token, inputs): + return request("GET", "/populations", cfg, token) + + +_run(main) diff --git a/integrations/pingone/scripts/list_users.py b/integrations/pingone/scripts/list_users.py new file mode 100644 index 0000000..10c05ad --- /dev/null +++ b/integrations/pingone/scripts/list_users.py @@ -0,0 +1,72 @@ +import json, os, sys, base64, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _auth_url(cfg): + return (str(cfg.get("auth_url") or "https://auth.pingone.com")).rstrip("/") + + +def _api_base(cfg): + return (str(cfg.get("api_url") or "https://api.pingone.com")).rstrip("/") + "/v1/environments/" + str(cfg.get("environment_id", "")) + + +def _token(cfg): + basic = base64.b64encode((str(cfg.get("client_id", "")) + ":" + str(cfg.get("client_secret", ""))).encode("utf-8")).decode("utf-8") + form = urllib.parse.urlencode({"grant_type": "client_credentials"}).encode("utf-8") + url = _auth_url(cfg) + "/" + str(cfg.get("environment_id", "")) + "/as/token" + req = urllib.request.Request(url, data=form, headers={ + "Authorization": "Basic " + basic, + "Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json", + }, method="POST") + with urllib.request.urlopen(req, timeout=60) as r: + tok = json.loads(r.read()) + if not tok.get("access_token"): + raise Exception("Token request failed: " + json.dumps(tok)) + return tok["access_token"] + + +def request(method, path, cfg, token, body=None, params=None, content_type="application/json"): + url = _api_base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + token, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = content_type + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + token = _token(cfg) + print(json.dumps(fn(cfg, token, inputs))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, token, inputs): + filter_ = inputs.get("filter") + limit = inputs.get("limit") + return request("GET", "/users", cfg, token, params={"filter": filter_, "limit": int(limit or 50)}) + + +_run(main) diff --git a/integrations/pingone/scripts/test_connection.py b/integrations/pingone/scripts/test_connection.py new file mode 100644 index 0000000..5b92f27 --- /dev/null +++ b/integrations/pingone/scripts/test_connection.py @@ -0,0 +1,71 @@ +import json, os, sys, base64, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _auth_url(cfg): + return (str(cfg.get("auth_url") or "https://auth.pingone.com")).rstrip("/") + + +def _api_base(cfg): + return (str(cfg.get("api_url") or "https://api.pingone.com")).rstrip("/") + "/v1/environments/" + str(cfg.get("environment_id", "")) + + +def _token(cfg): + basic = base64.b64encode((str(cfg.get("client_id", "")) + ":" + str(cfg.get("client_secret", ""))).encode("utf-8")).decode("utf-8") + form = urllib.parse.urlencode({"grant_type": "client_credentials"}).encode("utf-8") + url = _auth_url(cfg) + "/" + str(cfg.get("environment_id", "")) + "/as/token" + req = urllib.request.Request(url, data=form, headers={ + "Authorization": "Basic " + basic, + "Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json", + }, method="POST") + with urllib.request.urlopen(req, timeout=60) as r: + tok = json.loads(r.read()) + if not tok.get("access_token"): + raise Exception("Token request failed: " + json.dumps(tok)) + return tok["access_token"] + + +def request(method, path, cfg, token, body=None, params=None, content_type="application/json"): + url = _api_base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + token, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = content_type + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + token = _token(cfg) + print(json.dumps(fn(cfg, token, inputs))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, token, inputs): + request("GET", "/users", cfg, token, params={"limit": 1}) + return {"ok": True} + + +_run(main)