feat(carbon-black-cloud): new VMware Carbon Black Cloud EDR integration

Platform API, 10 commands: device search/get, quarantine/unquarantine
(network-isolate), background scan, update device policy, alert search,
ban/unban file hash (reputation override). API-token auth, stdlib-only.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

integrations/carbon-black-cloud/manifest.yaml

@@ -0,0 +1,121 @@
+id: carbon_black_cloud
+name: VMware Carbon Black Cloud
+version: 1.0.0
+description: "VMware Carbon Black Cloud (Platform API) — endpoint containment: search and read devices, quarantine/unquarantine an endpoint, trigger a background scan, update a device's policy, search alerts, and ban/unban a file hash (reputation override). API-token authentication; stdlib-only, no extra Python dependencies."
+changelog: "1.0.0 — Initial release: device search/get, quarantine/unquarantine, background scan, update policy, alert search, ban hash, delete reputation override."
+category: endpoint
+
+# Per-instance configuration. The API token is sent as
+# 'X-Auth-Token: <api_secret_key>/<api_id>'. org_key identifies the org.
+config_schema:
+  properties:
+    base_url:
+      type: string
+      description: "Carbon Black Cloud URL (e.g. https://defense.conferdeploy.net)"
+    api_id:
+      type: string
+      description: "API key ID"
+    api_secret_key:
+      type: string
+      description: "API secret key"
+      x-soar-sensitive: true
+    org_key:
+      type: string
+      description: "Organization key"
+  required:
+    - base_url
+    - api_id
+    - api_secret_key
+    - org_key
+
+commands:
+  - id: list_devices
+    name: cbc-list-devices
+    description: "Search devices (optionally by hostname or IP)."
+    risk: read
+    inputs_schema:
+      properties:
+        query: { type: string, description: "Free-text query (hostname, user, IP)" }
+        rows: { type: number, description: "Max devices (default 50)" }
+      required: []
+    outputs_schema: { properties: {} }
+  - id: get_device
+    name: cbc-get-device
+    description: "Get a single device by ID."
+    risk: read
+    inputs_schema:
+      properties:
+        device_id: { type: string, description: "Device ID" }
+      required: [device_id]
+    outputs_schema: { properties: {} }
+  - id: quarantine_device
+    name: cbc-quarantine-device
+    description: "Quarantine (network-isolate) a device."
+    inputs_schema:
+      properties:
+        device_id: { type: string, description: "Device ID" }
+      required: [device_id]
+    outputs_schema: { properties: {} }
+  - id: unquarantine_device
+    name: cbc-unquarantine-device
+    description: "Remove a device from quarantine."
+    inputs_schema:
+      properties:
+        device_id: { type: string, description: "Device ID" }
+      required: [device_id]
+    outputs_schema: { properties: {} }
+  - id: background_scan
+    name: cbc-background-scan
+    description: "Start or stop a background scan on a device."
+    inputs_schema:
+      properties:
+        device_id: { type: string, description: "Device ID" }
+        toggle: { type: string, description: "ON or OFF (default ON)" }
+      required: [device_id]
+    outputs_schema: { properties: {} }
+  - id: update_device_policy
+    name: cbc-update-device-policy
+    description: "Assign a device to a different policy."
+    inputs_schema:
+      properties:
+        device_id: { type: string, description: "Device ID" }
+        policy_id: { type: string, description: "Target policy ID" }
+      required: [device_id, policy_id]
+    outputs_schema: { properties: {} }
+  - id: search_alerts
+    name: cbc-search-alerts
+    description: "Search alerts."
+    risk: read
+    inputs_schema:
+      properties:
+        query: { type: string, description: "Free-text alert query" }
+        rows: { type: number, description: "Max alerts (default 50)" }
+      required: []
+    outputs_schema: { properties: {} }
+  - id: ban_hash
+    name: cbc-ban-hash
+    description: "Ban a file by SHA-256 hash (reputation deny-list override)."
+    inputs_schema:
+      properties:
+        sha256: { type: string, description: "SHA-256 hash to ban" }
+        filename: { type: string, description: "Optional associated file name" }
+        description: { type: string, description: "Optional reason/description" }
+      required: [sha256]
+    outputs_schema: { properties: {} }
+  - id: delete_reputation_override
+    name: cbc-delete-reputation-override
+    description: "Delete a reputation override (unban) by its ID."
+    inputs_schema:
+      properties:
+        override_id: { type: string, description: "Reputation override ID" }
+      required: [override_id]
+    outputs_schema: { properties: {} }
+
+  - id: test_connection
+    name: cbc-test-connection
+    description: "Verify connectivity and the API token (used by the Test button)."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }

integrations/carbon-black-cloud/scripts/background_scan.py

@@ -0,0 +1,60 @@
+import json, os, sys, 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 _org(cfg):
+    return str(cfg.get("org_key", ""))
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("base_url", "")).rstrip("/") + 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 = {
+        "X-Auth-Token": str(cfg.get("api_secret_key", "")) + "/" + str(cfg.get("api_id", "")),
+        "Accept": "application/json",
+    }
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) 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):
+    device_id = inputs.get("device_id")
+    if not device_id:
+        raise Exception("device_id is required")
+    toggle = inputs.get("toggle")
+    body = {
+        "action_type": "BACKGROUND_SCAN",
+        "device_id": [int(device_id)],
+        "options": {"toggle": (toggle or "ON")},
+    }
+    request("POST", "/appservices/v6/orgs/" + _org(cfg) + "/device_actions", cfg, body=body)
+    return {"ok": True}
+
+
+_run(main)

integrations/carbon-black-cloud/scripts/ban_hash.py

@@ -0,0 +1,62 @@
+import json, os, sys, 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 _org(cfg):
+    return str(cfg.get("org_key", ""))
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("base_url", "")).rstrip("/") + 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 = {
+        "X-Auth-Token": str(cfg.get("api_secret_key", "")) + "/" + str(cfg.get("api_id", "")),
+        "Accept": "application/json",
+    }
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) 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):
+    sha256 = inputs.get("sha256")
+    if not sha256:
+        raise Exception("sha256 is required")
+    filename = inputs.get("filename")
+    description = inputs.get("description")
+    body = {
+        "override_list": "BLACK_LIST",
+        "override_type": "SHA256_HASH",
+        "sha256_hash": sha256,
+        "filename": (filename or "unknown"),
+        "description": (description or "Blocked via Riposte"),
+    }
+    return request("POST", "/appservices/v6/orgs/" + _org(cfg) + "/reputations/overrides", cfg, body=body)
+
+
+_run(main)

integrations/carbon-black-cloud/scripts/delete_reputation_override.py

@@ -0,0 +1,57 @@
+import json, os, sys, 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 _org(cfg):
+    return str(cfg.get("org_key", ""))
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("base_url", "")).rstrip("/") + 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 = {
+        "X-Auth-Token": str(cfg.get("api_secret_key", "")) + "/" + str(cfg.get("api_id", "")),
+        "Accept": "application/json",
+    }
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) 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):
+    override_id = inputs.get("override_id")
+    if not override_id:
+        raise Exception("override_id is required")
+    q = lambda v: urllib.parse.quote(str(v), safe="")
+    resp = request("DELETE", "/appservices/v6/orgs/" + _org(cfg) + "/reputations/overrides/" + q(override_id), cfg)
+    if not resp:
+        return {"ok": True, "deleted": override_id}
+    return resp
+
+
+_run(main)

integrations/carbon-black-cloud/scripts/get_device.py

@@ -0,0 +1,54 @@
+import json, os, sys, 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 _org(cfg):
+    return str(cfg.get("org_key", ""))
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("base_url", "")).rstrip("/") + 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 = {
+        "X-Auth-Token": str(cfg.get("api_secret_key", "")) + "/" + str(cfg.get("api_id", "")),
+        "Accept": "application/json",
+    }
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) 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):
+    device_id = inputs.get("device_id")
+    if not device_id:
+        raise Exception("device_id is required")
+    q = lambda v: urllib.parse.quote(str(v), safe="")
+    return request("GET", "/appservices/v6/orgs/" + _org(cfg) + "/devices/" + q(device_id), cfg)
+
+
+_run(main)

integrations/carbon-black-cloud/scripts/list_devices.py

@@ -0,0 +1,55 @@
+import json, os, sys, 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 _org(cfg):
+    return str(cfg.get("org_key", ""))
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("base_url", "")).rstrip("/") + 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 = {
+        "X-Auth-Token": str(cfg.get("api_secret_key", "")) + "/" + str(cfg.get("api_id", "")),
+        "Accept": "application/json",
+    }
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) 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):
+    query = inputs.get("query")
+    rows = inputs.get("rows")
+    body = {"rows": int(rows or 50)}
+    if query:
+        body["query"] = query
+    return request("POST", "/appservices/v6/orgs/" + _org(cfg) + "/devices/_search", cfg, body=body)
+
+
+_run(main)

integrations/carbon-black-cloud/scripts/quarantine_device.py

@@ -0,0 +1,61 @@
+import json, os, sys, 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 _org(cfg):
+    return str(cfg.get("org_key", ""))
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("base_url", "")).rstrip("/") + 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 = {
+        "X-Auth-Token": str(cfg.get("api_secret_key", "")) + "/" + str(cfg.get("api_id", "")),
+        "Accept": "application/json",
+    }
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) 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):
+    device_id = inputs.get("device_id")
+    if not device_id:
+        raise Exception("device_id is required")
+    body = {
+        "action_type": "QUARANTINE",
+        "device_id": [int(device_id)],
+        "options": {"toggle": "ON"},
+    }
+    resp = request("POST", "/appservices/v6/orgs/" + _org(cfg) + "/device_actions", cfg, body=body)
+    if not resp:
+        return {"ok": True, "device_id": device_id, "action": "quarantine"}
+    return resp
+
+
+_run(main)

integrations/carbon-black-cloud/scripts/search_alerts.py

@@ -0,0 +1,55 @@
+import json, os, sys, 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 _org(cfg):
+    return str(cfg.get("org_key", ""))
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("base_url", "")).rstrip("/") + 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 = {
+        "X-Auth-Token": str(cfg.get("api_secret_key", "")) + "/" + str(cfg.get("api_id", "")),
+        "Accept": "application/json",
+    }
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) 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):
+    query = inputs.get("query")
+    rows = inputs.get("rows")
+    body = {"rows": int(rows or 50)}
+    if query:
+        body["query"] = query
+    return request("POST", "/api/alerts/v7/orgs/" + _org(cfg) + "/alerts/_search", cfg, body=body)
+
+
+_run(main)

integrations/carbon-black-cloud/scripts/test_connection.py

@@ -0,0 +1,51 @@
+import json, os, sys, 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 _org(cfg):
+    return str(cfg.get("org_key", ""))
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("base_url", "")).rstrip("/") + 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 = {
+        "X-Auth-Token": str(cfg.get("api_secret_key", "")) + "/" + str(cfg.get("api_id", "")),
+        "Accept": "application/json",
+    }
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) 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):
+    request("POST", "/appservices/v6/orgs/" + _org(cfg) + "/devices/_search", cfg, body={"rows": 1})
+    return {"ok": True}
+
+
+_run(main)

integrations/carbon-black-cloud/scripts/unquarantine_device.py

@@ -0,0 +1,61 @@
+import json, os, sys, 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 _org(cfg):
+    return str(cfg.get("org_key", ""))
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("base_url", "")).rstrip("/") + 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 = {
+        "X-Auth-Token": str(cfg.get("api_secret_key", "")) + "/" + str(cfg.get("api_id", "")),
+        "Accept": "application/json",
+    }
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) 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):
+    device_id = inputs.get("device_id")
+    if not device_id:
+        raise Exception("device_id is required")
+    body = {
+        "action_type": "QUARANTINE",
+        "device_id": [int(device_id)],
+        "options": {"toggle": "OFF"},
+    }
+    resp = request("POST", "/appservices/v6/orgs/" + _org(cfg) + "/device_actions", cfg, body=body)
+    if not resp:
+        return {"ok": True, "device_id": device_id, "action": "unquarantine"}
+    return resp
+
+
+_run(main)

integrations/carbon-black-cloud/scripts/update_device_policy.py

@@ -0,0 +1,62 @@
+import json, os, sys, 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 _org(cfg):
+    return str(cfg.get("org_key", ""))
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("base_url", "")).rstrip("/") + 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 = {
+        "X-Auth-Token": str(cfg.get("api_secret_key", "")) + "/" + str(cfg.get("api_id", "")),
+        "Accept": "application/json",
+    }
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) 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):
+    device_id = inputs.get("device_id")
+    if not device_id:
+        raise Exception("device_id is required")
+    policy_id = inputs.get("policy_id")
+    if not policy_id:
+        raise Exception("policy_id is required")
+    body = {
+        "action_type": "UPDATE_POLICY",
+        "device_id": [int(device_id)],
+        "options": {"policy_id": int(policy_id)},
+    }
+    request("POST", "/appservices/v6/orgs/" + _org(cfg) + "/device_actions", cfg, body=body)
+    return {"ok": True}
+
+
+_run(main)