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) <noreply@anthropic.com>
2026-07-11 23:37:53 +02:00
parent 26694b512a
commit 7bd6ad6a69
10 changed files with 779 additions and 0 deletions
@@ -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)