From 7bd6ad6a69b8ae826651dce8c13a99043b9ad372 Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sat, 11 Jul 2026 23:37:53 +0200 Subject: [PATCH] feat(duo): new Cisco Duo MFA-containment integration Duo Admin API, 9 commands: get users/user, modify user status (disable/enable/bypass), delete user, bypass codes, devices, user devices, authentication logs. HMAC-SHA1 signed auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) --- integrations/duo/manifest.yaml | 111 ++++++++++++++++++ integrations/duo/scripts/delete_user.py | 75 ++++++++++++ .../duo/scripts/get_authentication_logs.py | 77 ++++++++++++ integrations/duo/scripts/get_bypass_codes.py | 74 ++++++++++++ integrations/duo/scripts/get_devices.py | 70 +++++++++++ integrations/duo/scripts/get_user.py | 74 ++++++++++++ integrations/duo/scripts/get_user_devices.py | 74 ++++++++++++ integrations/duo/scripts/get_users.py | 75 ++++++++++++ integrations/duo/scripts/modify_user.py | 79 +++++++++++++ integrations/duo/scripts/test_connection.py | 70 +++++++++++ 10 files changed, 779 insertions(+) create mode 100644 integrations/duo/manifest.yaml create mode 100644 integrations/duo/scripts/delete_user.py create mode 100644 integrations/duo/scripts/get_authentication_logs.py create mode 100644 integrations/duo/scripts/get_bypass_codes.py create mode 100644 integrations/duo/scripts/get_devices.py create mode 100644 integrations/duo/scripts/get_user.py create mode 100644 integrations/duo/scripts/get_user_devices.py create mode 100644 integrations/duo/scripts/get_users.py create mode 100644 integrations/duo/scripts/modify_user.py create mode 100644 integrations/duo/scripts/test_connection.py diff --git a/integrations/duo/manifest.yaml b/integrations/duo/manifest.yaml new file mode 100644 index 0000000..a9250c3 --- /dev/null +++ b/integrations/duo/manifest.yaml @@ -0,0 +1,111 @@ +id: duo +name: Cisco Duo +version: 1.0.0 +description: "Cisco Duo (Admin API) — MFA/identity containment and investigation: list and read users, disable/enable/bypass a user (containment), delete users, read bypass codes and devices, and pull authentication logs. HMAC-SHA1 signed authentication; stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: get users/user, modify user status (disable/enable/bypass), delete user, get bypass codes, get devices, get user devices, authentication logs." +category: identity + +# Per-instance configuration. Requests are signed (HMAC-SHA1) with the +# integration key + secret key against the Admin API hostname. +config_schema: + properties: + api_hostname: + type: string + description: "Duo Admin API hostname (e.g. api-xxxxxxxx.duosecurity.com)" + integration_key: + type: string + description: "Admin API integration key (ikey)" + x-soar-sensitive: true + secret_key: + type: string + description: "Admin API secret key (skey)" + x-soar-sensitive: true + required: + - api_hostname + - integration_key + - secret_key + +commands: + - id: get_users + name: duo-get-users + description: "List users (optionally filter by exact username)." + risk: read + inputs_schema: + properties: + username: { type: string, description: "Optional exact username to look up" } + limit: { type: number, description: "Max users (default 100)" } + required: [] + outputs_schema: { properties: {} } + - id: get_user + name: duo-get-user + description: "Get a single user by user ID." + risk: read + inputs_schema: + properties: + user_id: { type: string, description: "Duo user ID" } + required: [user_id] + outputs_schema: { properties: {} } + - id: modify_user + name: duo-modify-user + description: "Change a user's status — disable (containment), enable (active), or bypass." + inputs_schema: + properties: + user_id: { type: string, description: "Duo user ID" } + status: { type: string, description: "active | disabled | bypass" } + required: [user_id, status] + outputs_schema: { properties: {} } + - id: delete_user + name: duo-delete-user + description: "Delete a user." + inputs_schema: + properties: + user_id: { type: string, description: "Duo user ID" } + required: [user_id] + outputs_schema: { properties: {} } + - id: get_bypass_codes + name: duo-get-bypass-codes + description: "List a user's bypass codes." + risk: read + inputs_schema: + properties: + user_id: { type: string, description: "Duo user ID" } + required: [user_id] + outputs_schema: { properties: {} } + - id: get_devices + name: duo-get-devices + description: "List all phones/devices in the account." + risk: read + inputs_schema: + properties: + limit: { type: number, description: "Max devices (default 100)" } + required: [] + outputs_schema: { properties: {} } + - id: get_user_devices + name: duo-get-user-devices + description: "List the phones/devices associated with a user." + risk: read + inputs_schema: + properties: + user_id: { type: string, description: "Duo user ID" } + required: [user_id] + outputs_schema: { properties: {} } + - id: get_authentication_logs + name: duo-get-authentication-logs + description: "Pull authentication logs for a time window." + risk: read + inputs_schema: + properties: + mintime: { type: number, description: "Start time as Unix epoch milliseconds (required by Duo v2 logs)" } + maxtime: { type: number, description: "End time as Unix epoch milliseconds" } + limit: { type: number, description: "Max events (default 100, max 1000)" } + required: [mintime, maxtime] + outputs_schema: { properties: {} } + + - id: test_connection + name: duo-test-connection + description: "Verify connectivity and the signed credentials (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/duo/scripts/delete_user.py b/integrations/duo/scripts/delete_user.py new file mode 100644 index 0000000..96da182 --- /dev/null +++ b/integrations/duo/scripts/delete_user.py @@ -0,0 +1,75 @@ +import json, os, sys, hmac, hashlib, base64, email.utils +import 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 _canon_params(params): + # RFC-3986 encode each key/value, sort by key, join k=v with & + items = [] + for k in sorted(params.keys()): + ek = urllib.parse.quote(str(k), "~") + ev = urllib.parse.quote(str(params[k]), "~") + items.append(ek + "=" + ev) + return "&".join(items) + + +def _sign(method, host, path, params, cfg, now): + canon = "\n".join([now, method.upper(), host.lower(), path, _canon_params(params)]) + skey = str(cfg.get("secret_key", "")).encode("utf-8") + sig = hmac.new(skey, canon.encode("utf-8"), hashlib.sha1).hexdigest() + ikey = str(cfg.get("integration_key", "")) + auth = base64.b64encode((ikey + ":" + sig).encode("utf-8")).decode("utf-8") + return "Basic " + auth + + +def call(method, path, cfg, params=None): + params = params or {} + host = str(cfg.get("api_hostname", "")).strip() + now = email.utils.formatdate() # RFC 2822, e.g. 'Wed, 01 Jan 2020 00:00:00 -0000' + authz = _sign(method, host, path, params, cfg, now) + headers = {"Authorization": authz, "Date": now, "Accept": "application/json"} + method = method.upper() + url = "https://" + host + path + data = None + if method in ("GET", "DELETE"): + if params: + url += "?" + _canon_params(params) + else: + # POST: params go in the body, form-encoded with the SAME canonicalization + headers["Content-Type"] = "application/x-www-form-urlencoded" + data = _canon_params(params).encode("utf-8") + 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: + print(json.dumps(fn(_cfg(), _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, inputs): + user_id = str(inputs.get("user_id", "") or "").strip() + if not user_id: + raise Exception("user_id is required") + q = lambda v: urllib.parse.quote(str(v), safe="") + path = "/admin/v1/users/" + q(user_id) + call("DELETE", path, cfg) + return {"ok": True} + + +_run(main) diff --git a/integrations/duo/scripts/get_authentication_logs.py b/integrations/duo/scripts/get_authentication_logs.py new file mode 100644 index 0000000..30b4d3c --- /dev/null +++ b/integrations/duo/scripts/get_authentication_logs.py @@ -0,0 +1,77 @@ +import json, os, sys, hmac, hashlib, base64, email.utils +import 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 _canon_params(params): + # RFC-3986 encode each key/value, sort by key, join k=v with & + items = [] + for k in sorted(params.keys()): + ek = urllib.parse.quote(str(k), "~") + ev = urllib.parse.quote(str(params[k]), "~") + items.append(ek + "=" + ev) + return "&".join(items) + + +def _sign(method, host, path, params, cfg, now): + canon = "\n".join([now, method.upper(), host.lower(), path, _canon_params(params)]) + skey = str(cfg.get("secret_key", "")).encode("utf-8") + sig = hmac.new(skey, canon.encode("utf-8"), hashlib.sha1).hexdigest() + ikey = str(cfg.get("integration_key", "")) + auth = base64.b64encode((ikey + ":" + sig).encode("utf-8")).decode("utf-8") + return "Basic " + auth + + +def call(method, path, cfg, params=None): + params = params or {} + host = str(cfg.get("api_hostname", "")).strip() + now = email.utils.formatdate() # RFC 2822, e.g. 'Wed, 01 Jan 2020 00:00:00 -0000' + authz = _sign(method, host, path, params, cfg, now) + headers = {"Authorization": authz, "Date": now, "Accept": "application/json"} + method = method.upper() + url = "https://" + host + path + data = None + if method in ("GET", "DELETE"): + if params: + url += "?" + _canon_params(params) + else: + # POST: params go in the body, form-encoded with the SAME canonicalization + headers["Content-Type"] = "application/x-www-form-urlencoded" + data = _canon_params(params).encode("utf-8") + 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: + print(json.dumps(fn(_cfg(), _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, inputs): + mintime = inputs.get("mintime", "") + maxtime = inputs.get("maxtime", "") + if mintime == "" or mintime is None: + raise Exception("mintime is required") + if maxtime == "" or maxtime is None: + raise Exception("maxtime is required") + limit = inputs.get("limit", 100) + params = {"mintime": int(mintime), "maxtime": int(maxtime), "limit": int(limit)} + return call("GET", "/admin/v2/logs/authentication", cfg, params) + + +_run(main) diff --git a/integrations/duo/scripts/get_bypass_codes.py b/integrations/duo/scripts/get_bypass_codes.py new file mode 100644 index 0000000..58e6389 --- /dev/null +++ b/integrations/duo/scripts/get_bypass_codes.py @@ -0,0 +1,74 @@ +import json, os, sys, hmac, hashlib, base64, email.utils +import 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 _canon_params(params): + # RFC-3986 encode each key/value, sort by key, join k=v with & + items = [] + for k in sorted(params.keys()): + ek = urllib.parse.quote(str(k), "~") + ev = urllib.parse.quote(str(params[k]), "~") + items.append(ek + "=" + ev) + return "&".join(items) + + +def _sign(method, host, path, params, cfg, now): + canon = "\n".join([now, method.upper(), host.lower(), path, _canon_params(params)]) + skey = str(cfg.get("secret_key", "")).encode("utf-8") + sig = hmac.new(skey, canon.encode("utf-8"), hashlib.sha1).hexdigest() + ikey = str(cfg.get("integration_key", "")) + auth = base64.b64encode((ikey + ":" + sig).encode("utf-8")).decode("utf-8") + return "Basic " + auth + + +def call(method, path, cfg, params=None): + params = params or {} + host = str(cfg.get("api_hostname", "")).strip() + now = email.utils.formatdate() # RFC 2822, e.g. 'Wed, 01 Jan 2020 00:00:00 -0000' + authz = _sign(method, host, path, params, cfg, now) + headers = {"Authorization": authz, "Date": now, "Accept": "application/json"} + method = method.upper() + url = "https://" + host + path + data = None + if method in ("GET", "DELETE"): + if params: + url += "?" + _canon_params(params) + else: + # POST: params go in the body, form-encoded with the SAME canonicalization + headers["Content-Type"] = "application/x-www-form-urlencoded" + data = _canon_params(params).encode("utf-8") + 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: + print(json.dumps(fn(_cfg(), _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, inputs): + user_id = str(inputs.get("user_id", "") or "").strip() + if not user_id: + raise Exception("user_id is required") + q = lambda v: urllib.parse.quote(str(v), safe="") + path = "/admin/v1/users/" + q(user_id) + "/bypass_codes" + return call("GET", path, cfg) + + +_run(main) diff --git a/integrations/duo/scripts/get_devices.py b/integrations/duo/scripts/get_devices.py new file mode 100644 index 0000000..ebe90be --- /dev/null +++ b/integrations/duo/scripts/get_devices.py @@ -0,0 +1,70 @@ +import json, os, sys, hmac, hashlib, base64, email.utils +import 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 _canon_params(params): + # RFC-3986 encode each key/value, sort by key, join k=v with & + items = [] + for k in sorted(params.keys()): + ek = urllib.parse.quote(str(k), "~") + ev = urllib.parse.quote(str(params[k]), "~") + items.append(ek + "=" + ev) + return "&".join(items) + + +def _sign(method, host, path, params, cfg, now): + canon = "\n".join([now, method.upper(), host.lower(), path, _canon_params(params)]) + skey = str(cfg.get("secret_key", "")).encode("utf-8") + sig = hmac.new(skey, canon.encode("utf-8"), hashlib.sha1).hexdigest() + ikey = str(cfg.get("integration_key", "")) + auth = base64.b64encode((ikey + ":" + sig).encode("utf-8")).decode("utf-8") + return "Basic " + auth + + +def call(method, path, cfg, params=None): + params = params or {} + host = str(cfg.get("api_hostname", "")).strip() + now = email.utils.formatdate() # RFC 2822, e.g. 'Wed, 01 Jan 2020 00:00:00 -0000' + authz = _sign(method, host, path, params, cfg, now) + headers = {"Authorization": authz, "Date": now, "Accept": "application/json"} + method = method.upper() + url = "https://" + host + path + data = None + if method in ("GET", "DELETE"): + if params: + url += "?" + _canon_params(params) + else: + # POST: params go in the body, form-encoded with the SAME canonicalization + headers["Content-Type"] = "application/x-www-form-urlencoded" + data = _canon_params(params).encode("utf-8") + 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: + print(json.dumps(fn(_cfg(), _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, inputs): + limit = inputs.get("limit", 100) + return call("GET", "/admin/v1/phones", cfg, {"limit": int(limit)}) + + +_run(main) diff --git a/integrations/duo/scripts/get_user.py b/integrations/duo/scripts/get_user.py new file mode 100644 index 0000000..d178af0 --- /dev/null +++ b/integrations/duo/scripts/get_user.py @@ -0,0 +1,74 @@ +import json, os, sys, hmac, hashlib, base64, email.utils +import 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 _canon_params(params): + # RFC-3986 encode each key/value, sort by key, join k=v with & + items = [] + for k in sorted(params.keys()): + ek = urllib.parse.quote(str(k), "~") + ev = urllib.parse.quote(str(params[k]), "~") + items.append(ek + "=" + ev) + return "&".join(items) + + +def _sign(method, host, path, params, cfg, now): + canon = "\n".join([now, method.upper(), host.lower(), path, _canon_params(params)]) + skey = str(cfg.get("secret_key", "")).encode("utf-8") + sig = hmac.new(skey, canon.encode("utf-8"), hashlib.sha1).hexdigest() + ikey = str(cfg.get("integration_key", "")) + auth = base64.b64encode((ikey + ":" + sig).encode("utf-8")).decode("utf-8") + return "Basic " + auth + + +def call(method, path, cfg, params=None): + params = params or {} + host = str(cfg.get("api_hostname", "")).strip() + now = email.utils.formatdate() # RFC 2822, e.g. 'Wed, 01 Jan 2020 00:00:00 -0000' + authz = _sign(method, host, path, params, cfg, now) + headers = {"Authorization": authz, "Date": now, "Accept": "application/json"} + method = method.upper() + url = "https://" + host + path + data = None + if method in ("GET", "DELETE"): + if params: + url += "?" + _canon_params(params) + else: + # POST: params go in the body, form-encoded with the SAME canonicalization + headers["Content-Type"] = "application/x-www-form-urlencoded" + data = _canon_params(params).encode("utf-8") + 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: + print(json.dumps(fn(_cfg(), _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, inputs): + user_id = str(inputs.get("user_id", "") or "").strip() + if not user_id: + raise Exception("user_id is required") + q = lambda v: urllib.parse.quote(str(v), safe="") + path = "/admin/v1/users/" + q(user_id) + return call("GET", path, cfg) + + +_run(main) diff --git a/integrations/duo/scripts/get_user_devices.py b/integrations/duo/scripts/get_user_devices.py new file mode 100644 index 0000000..cfcac5e --- /dev/null +++ b/integrations/duo/scripts/get_user_devices.py @@ -0,0 +1,74 @@ +import json, os, sys, hmac, hashlib, base64, email.utils +import 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 _canon_params(params): + # RFC-3986 encode each key/value, sort by key, join k=v with & + items = [] + for k in sorted(params.keys()): + ek = urllib.parse.quote(str(k), "~") + ev = urllib.parse.quote(str(params[k]), "~") + items.append(ek + "=" + ev) + return "&".join(items) + + +def _sign(method, host, path, params, cfg, now): + canon = "\n".join([now, method.upper(), host.lower(), path, _canon_params(params)]) + skey = str(cfg.get("secret_key", "")).encode("utf-8") + sig = hmac.new(skey, canon.encode("utf-8"), hashlib.sha1).hexdigest() + ikey = str(cfg.get("integration_key", "")) + auth = base64.b64encode((ikey + ":" + sig).encode("utf-8")).decode("utf-8") + return "Basic " + auth + + +def call(method, path, cfg, params=None): + params = params or {} + host = str(cfg.get("api_hostname", "")).strip() + now = email.utils.formatdate() # RFC 2822, e.g. 'Wed, 01 Jan 2020 00:00:00 -0000' + authz = _sign(method, host, path, params, cfg, now) + headers = {"Authorization": authz, "Date": now, "Accept": "application/json"} + method = method.upper() + url = "https://" + host + path + data = None + if method in ("GET", "DELETE"): + if params: + url += "?" + _canon_params(params) + else: + # POST: params go in the body, form-encoded with the SAME canonicalization + headers["Content-Type"] = "application/x-www-form-urlencoded" + data = _canon_params(params).encode("utf-8") + 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: + print(json.dumps(fn(_cfg(), _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, inputs): + user_id = str(inputs.get("user_id", "") or "").strip() + if not user_id: + raise Exception("user_id is required") + q = lambda v: urllib.parse.quote(str(v), safe="") + path = "/admin/v1/users/" + q(user_id) + "/phones" + return call("GET", path, cfg) + + +_run(main) diff --git a/integrations/duo/scripts/get_users.py b/integrations/duo/scripts/get_users.py new file mode 100644 index 0000000..27aca7d --- /dev/null +++ b/integrations/duo/scripts/get_users.py @@ -0,0 +1,75 @@ +import json, os, sys, hmac, hashlib, base64, email.utils +import 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 _canon_params(params): + # RFC-3986 encode each key/value, sort by key, join k=v with & + items = [] + for k in sorted(params.keys()): + ek = urllib.parse.quote(str(k), "~") + ev = urllib.parse.quote(str(params[k]), "~") + items.append(ek + "=" + ev) + return "&".join(items) + + +def _sign(method, host, path, params, cfg, now): + canon = "\n".join([now, method.upper(), host.lower(), path, _canon_params(params)]) + skey = str(cfg.get("secret_key", "")).encode("utf-8") + sig = hmac.new(skey, canon.encode("utf-8"), hashlib.sha1).hexdigest() + ikey = str(cfg.get("integration_key", "")) + auth = base64.b64encode((ikey + ":" + sig).encode("utf-8")).decode("utf-8") + return "Basic " + auth + + +def call(method, path, cfg, params=None): + params = params or {} + host = str(cfg.get("api_hostname", "")).strip() + now = email.utils.formatdate() # RFC 2822, e.g. 'Wed, 01 Jan 2020 00:00:00 -0000' + authz = _sign(method, host, path, params, cfg, now) + headers = {"Authorization": authz, "Date": now, "Accept": "application/json"} + method = method.upper() + url = "https://" + host + path + data = None + if method in ("GET", "DELETE"): + if params: + url += "?" + _canon_params(params) + else: + # POST: params go in the body, form-encoded with the SAME canonicalization + headers["Content-Type"] = "application/x-www-form-urlencoded" + data = _canon_params(params).encode("utf-8") + 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: + print(json.dumps(fn(_cfg(), _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, inputs): + username = str(inputs.get("username", "") or "").strip() + if username: + params = {"username": username} + else: + limit = inputs.get("limit", 100) + params = {"limit": int(limit)} + return call("GET", "/admin/v1/users", cfg, params) + + +_run(main) diff --git a/integrations/duo/scripts/modify_user.py b/integrations/duo/scripts/modify_user.py new file mode 100644 index 0000000..7b95728 --- /dev/null +++ b/integrations/duo/scripts/modify_user.py @@ -0,0 +1,79 @@ +import json, os, sys, hmac, hashlib, base64, email.utils +import 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 _canon_params(params): + # RFC-3986 encode each key/value, sort by key, join k=v with & + items = [] + for k in sorted(params.keys()): + ek = urllib.parse.quote(str(k), "~") + ev = urllib.parse.quote(str(params[k]), "~") + items.append(ek + "=" + ev) + return "&".join(items) + + +def _sign(method, host, path, params, cfg, now): + canon = "\n".join([now, method.upper(), host.lower(), path, _canon_params(params)]) + skey = str(cfg.get("secret_key", "")).encode("utf-8") + sig = hmac.new(skey, canon.encode("utf-8"), hashlib.sha1).hexdigest() + ikey = str(cfg.get("integration_key", "")) + auth = base64.b64encode((ikey + ":" + sig).encode("utf-8")).decode("utf-8") + return "Basic " + auth + + +def call(method, path, cfg, params=None): + params = params or {} + host = str(cfg.get("api_hostname", "")).strip() + now = email.utils.formatdate() # RFC 2822, e.g. 'Wed, 01 Jan 2020 00:00:00 -0000' + authz = _sign(method, host, path, params, cfg, now) + headers = {"Authorization": authz, "Date": now, "Accept": "application/json"} + method = method.upper() + url = "https://" + host + path + data = None + if method in ("GET", "DELETE"): + if params: + url += "?" + _canon_params(params) + else: + # POST: params go in the body, form-encoded with the SAME canonicalization + headers["Content-Type"] = "application/x-www-form-urlencoded" + data = _canon_params(params).encode("utf-8") + 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: + print(json.dumps(fn(_cfg(), _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, inputs): + user_id = str(inputs.get("user_id", "") or "").strip() + if not user_id: + raise Exception("user_id is required") + status = str(inputs.get("status", "") or "").strip() + if not status: + raise Exception("status is required") + if status not in ("active", "disabled", "bypass"): + raise Exception("status must be active, disabled, or bypass") + q = lambda v: urllib.parse.quote(str(v), safe="") + path = "/admin/v1/users/" + q(user_id) + return call("POST", path, cfg, {"status": status}) + + +_run(main) diff --git a/integrations/duo/scripts/test_connection.py b/integrations/duo/scripts/test_connection.py new file mode 100644 index 0000000..7ed1cbb --- /dev/null +++ b/integrations/duo/scripts/test_connection.py @@ -0,0 +1,70 @@ +import json, os, sys, hmac, hashlib, base64, email.utils +import 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 _canon_params(params): + # RFC-3986 encode each key/value, sort by key, join k=v with & + items = [] + for k in sorted(params.keys()): + ek = urllib.parse.quote(str(k), "~") + ev = urllib.parse.quote(str(params[k]), "~") + items.append(ek + "=" + ev) + return "&".join(items) + + +def _sign(method, host, path, params, cfg, now): + canon = "\n".join([now, method.upper(), host.lower(), path, _canon_params(params)]) + skey = str(cfg.get("secret_key", "")).encode("utf-8") + sig = hmac.new(skey, canon.encode("utf-8"), hashlib.sha1).hexdigest() + ikey = str(cfg.get("integration_key", "")) + auth = base64.b64encode((ikey + ":" + sig).encode("utf-8")).decode("utf-8") + return "Basic " + auth + + +def call(method, path, cfg, params=None): + params = params or {} + host = str(cfg.get("api_hostname", "")).strip() + now = email.utils.formatdate() # RFC 2822, e.g. 'Wed, 01 Jan 2020 00:00:00 -0000' + authz = _sign(method, host, path, params, cfg, now) + headers = {"Authorization": authz, "Date": now, "Accept": "application/json"} + method = method.upper() + url = "https://" + host + path + data = None + if method in ("GET", "DELETE"): + if params: + url += "?" + _canon_params(params) + else: + # POST: params go in the body, form-encoded with the SAME canonicalization + headers["Content-Type"] = "application/x-www-form-urlencoded" + data = _canon_params(params).encode("utf-8") + 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: + print(json.dumps(fn(_cfg(), _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, inputs): + call("GET", "/admin/v1/users", cfg, {"limit": 1}) + return {"ok": True} + + +_run(main)