feat(cortex-xdr): full command coverage (50 commands) — endpoints, scripts, exclusions, tags, audits, RBAC, distributions
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
@@ -1,8 +1,8 @@
|
||||
id: cortex_xdr
|
||||
name: Cortex XDR
|
||||
version: 1.0.0
|
||||
description: "Palo Alto Cortex XDR (public API v1) — incident ingestion, endpoint isolation, RTR scripts, hash block/allow lists, file quarantine/restore, scans and alert retrieval."
|
||||
changelog: "1.0.0 — Initial release: incident ingestion (get_incidents) with OCSF mapper, endpoints, isolate/unisolate, scan, hash blocklist/allowlist, quarantine/restore, run script + results, alerts retrieval, distributions and action status. Standard or Advanced API authentication."
|
||||
version: 1.1.0
|
||||
description: "Palo Alto Cortex XDR (public API v1) — incident ingestion, endpoint isolation/scan/delete/tagging, RTR scripts, hash block/allow lists, file quarantine/restore/retrieval, alert exclusions, device-control violations, audits, distributions and RBAC/risk."
|
||||
changelog: "1.1.0 — Full command coverage: added delete/alias/tag endpoints, abort scan, original alerts, script metadata/code/snippet/exec-status, file retrieval (+details), alert exclusions, device-control violations, audits, distribution url/status/create and RBAC (users, roles, groups, risk score, risky users/hosts). 1.0.0 — Initial release: incident ingestion (get_incidents) with OCSF mapper, endpoints, isolate/unisolate, scan, hash blocklist/allowlist, quarantine/restore, run script + results, alerts retrieval, distributions and action status. Standard or Advanced API authentication."
|
||||
category: endpoint
|
||||
|
||||
# Per-instance configuration. The base URL is the tenant API root, e.g.
|
||||
@@ -260,6 +260,279 @@ commands:
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
- id: delete_endpoints
|
||||
name: cortex-xdr-delete-endpoints
|
||||
description: "Delete endpoints (agents) by ID."
|
||||
inputs_schema:
|
||||
properties:
|
||||
endpoint_ids: { type: string, description: "Comma-separated endpoint IDs to delete" }
|
||||
required: [endpoint_ids]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: set_endpoint_alias
|
||||
name: cortex-xdr-set-endpoint-alias
|
||||
description: "Set the alias name on endpoints."
|
||||
inputs_schema:
|
||||
properties:
|
||||
endpoint_ids: { type: string, description: "Comma-separated endpoint IDs" }
|
||||
alias: { type: string, description: "New alias name (empty to clear)" }
|
||||
required: [endpoint_ids]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: abort_scan
|
||||
name: cortex-xdr-abort-scan
|
||||
description: "Abort a running malware scan on endpoints matching the filters (or all)."
|
||||
inputs_schema:
|
||||
properties:
|
||||
endpoint_id_list: { type: string, description: "Comma-separated endpoint IDs (omit to abort all)" }
|
||||
incident_id: { type: string, description: "Optional incident ID" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_original_alerts
|
||||
name: cortex-xdr-get-original-alerts
|
||||
description: "Get the original raw alert events by alert ID."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
alert_id_list: { type: string, description: "Comma-separated alert IDs" }
|
||||
required: [alert_id_list]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_script_metadata
|
||||
name: cortex-xdr-get-script-metadata
|
||||
description: "Get metadata for a script by UID."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
script_uid: { type: string, description: "Script UID" }
|
||||
required: [script_uid]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_script_code
|
||||
name: cortex-xdr-get-script-code
|
||||
description: "Get the source code of a script by UID."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
script_uid: { type: string, description: "Script UID" }
|
||||
required: [script_uid]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: run_snippet_code_script
|
||||
name: cortex-xdr-run-snippet-code-script
|
||||
description: "Run an arbitrary code snippet on endpoints."
|
||||
inputs_schema:
|
||||
properties:
|
||||
snippet_code: { type: string, description: "Code snippet to run" }
|
||||
endpoint_ids: { type: string, description: "Comma-separated endpoint IDs" }
|
||||
incident_id: { type: string, description: "Optional incident ID" }
|
||||
required: [snippet_code, endpoint_ids]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_script_execution_status
|
||||
name: cortex-xdr-get-script-execution-status
|
||||
description: "Get the status of a script execution by action ID."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
action_id: { type: string, description: "Action ID" }
|
||||
required: [action_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: retrieve_file
|
||||
name: cortex-xdr-retrieve-file
|
||||
description: "Retrieve files from endpoints by OS-specific path lists."
|
||||
inputs_schema:
|
||||
properties:
|
||||
endpoint_ids: { type: string, description: "Comma-separated endpoint IDs" }
|
||||
windows: { type: string, description: "Comma-separated Windows file paths" }
|
||||
linux: { type: string, description: "Comma-separated Linux file paths" }
|
||||
macos: { type: string, description: "Comma-separated macOS file paths" }
|
||||
incident_id: { type: string, description: "Optional incident ID" }
|
||||
required: [endpoint_ids]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: retrieve_file_details
|
||||
name: cortex-xdr-retrieve-file-details
|
||||
description: "Get the download details of a file-retrieval action."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
action_id: { type: string, description: "Group action ID from retrieve_file" }
|
||||
required: [action_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: add_exclusion
|
||||
name: cortex-xdr-add-exclusion
|
||||
description: "Add an alert exclusion."
|
||||
inputs_schema:
|
||||
properties:
|
||||
name: { type: string, description: "Exclusion name" }
|
||||
indicator: { type: string, description: "JSON indicator object for the exclusion filter" }
|
||||
status: { type: string, description: "ENABLED or DISABLED (default ENABLED)" }
|
||||
required: [name, indicator]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: delete_exclusion
|
||||
name: cortex-xdr-delete-exclusion
|
||||
description: "Delete an alert exclusion by ID."
|
||||
inputs_schema:
|
||||
properties:
|
||||
alert_exclusion_id: { type: number, description: "Exclusion ID" }
|
||||
required: [alert_exclusion_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_exclusion
|
||||
name: cortex-xdr-get-exclusion
|
||||
description: "List alert exclusions."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
tenant_id: { type: string, description: "Optional tenant ID" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: add_tag_endpoint
|
||||
name: cortex-xdr-add-tag-endpoint
|
||||
description: "Add a tag to endpoints."
|
||||
inputs_schema:
|
||||
properties:
|
||||
endpoint_ids: { type: string, description: "Comma-separated endpoint IDs" }
|
||||
tag: { type: string, description: "Tag to add" }
|
||||
required: [endpoint_ids, tag]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: remove_tag_endpoint
|
||||
name: cortex-xdr-remove-tag-endpoint
|
||||
description: "Remove a tag from endpoints."
|
||||
inputs_schema:
|
||||
properties:
|
||||
endpoint_ids: { type: string, description: "Comma-separated endpoint IDs" }
|
||||
tag: { type: string, description: "Tag to remove" }
|
||||
required: [endpoint_ids, tag]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_device_control_violations
|
||||
name: cortex-xdr-get-device-control-violations
|
||||
description: "List device control (USB) violations."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
endpoint_ids: { type: string, description: "Comma-separated endpoint IDs" }
|
||||
type: { type: string, description: "Comma-separated violation types" }
|
||||
timestamp_gte: { type: string, description: "Lower bound timestamp (ISO8601 or epoch ms)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: list_users
|
||||
name: cortex-xdr-list-users
|
||||
description: "List Cortex XDR users."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_risk_score
|
||||
name: cortex-xdr-get-risk-score
|
||||
description: "Get the risk score of a user or host by ID."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
id: { type: string, description: "User or host identifier" }
|
||||
required: [id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: list_risky_users
|
||||
name: cortex-xdr-list-risky-users
|
||||
description: "List risky users."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: list_risky_hosts
|
||||
name: cortex-xdr-list-risky-hosts
|
||||
description: "List risky hosts."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: list_user_groups
|
||||
name: cortex-xdr-list-user-groups
|
||||
description: "Get user groups by name."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
group_names: { type: string, description: "Comma-separated group names" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: list_roles
|
||||
name: cortex-xdr-list-roles
|
||||
description: "Get roles by name."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
role_names: { type: string, description: "Comma-separated role names" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: set_user_role
|
||||
name: cortex-xdr-set-user-role
|
||||
description: "Assign a role to users."
|
||||
inputs_schema:
|
||||
properties:
|
||||
user_emails: { type: string, description: "Comma-separated user emails" }
|
||||
role_name: { type: string, description: "Role name to assign" }
|
||||
required: [user_emails, role_name]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: remove_user_role
|
||||
name: cortex-xdr-remove-user-role
|
||||
description: "Remove the role assignment from users."
|
||||
inputs_schema:
|
||||
properties:
|
||||
user_emails: { type: string, description: "Comma-separated user emails" }
|
||||
required: [user_emails]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: audit_management_logs
|
||||
name: cortex-xdr-audit-management-logs
|
||||
description: "Retrieve management audit logs."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
email: { type: string, description: "Comma-separated emails" }
|
||||
type: { type: string, description: "Comma-separated log types" }
|
||||
timestamp_gte: { type: string, description: "Lower bound timestamp (ISO8601 or epoch ms)" }
|
||||
limit: { type: number, description: "Max records (default 100)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_audit_agent_reports
|
||||
name: cortex-xdr-get-audit-agent-reports
|
||||
description: "Retrieve agent audit reports."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
endpoint_ids: { type: string, description: "Comma-separated endpoint IDs" }
|
||||
type: { type: string, description: "Comma-separated report types" }
|
||||
timestamp_gte: { type: string, description: "Lower bound timestamp (ISO8601 or epoch ms)" }
|
||||
limit: { type: number, description: "Max records (default 100)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_distribution_url
|
||||
name: cortex-xdr-get-distribution-url
|
||||
description: "Get the download URL of an agent installer distribution."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
distribution_id: { type: string, description: "Distribution ID" }
|
||||
package_type: { type: string, description: "Package type (sh, rpm, deb, pkg, x86, x64, ...)" }
|
||||
required: [distribution_id, package_type]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_distribution_status
|
||||
name: cortex-xdr-get-distribution-status
|
||||
description: "Get the build status of an agent installer distribution."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
distribution_id: { type: string, description: "Distribution ID" }
|
||||
required: [distribution_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: create_distribution
|
||||
name: cortex-xdr-create-distribution
|
||||
description: "Create an agent installer distribution."
|
||||
inputs_schema:
|
||||
properties:
|
||||
name: { type: string, description: "Distribution name" }
|
||||
platform: { type: string, description: "Platform (windows, linux, macos, android)" }
|
||||
package_type: { type: string, description: "standalone or upgrade (default standalone)" }
|
||||
agent_version: { type: string, description: "Agent version" }
|
||||
description: { type: string, description: "Optional description" }
|
||||
required: [name, platform]
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
- id: test_connection
|
||||
name: cortex-xdr-test-connection
|
||||
description: "Verify connectivity and credentials (used by the Test button)."
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
filters = []
|
||||
if inputs.get("endpoint_id_list"):
|
||||
filters.append({"field": "endpoint_id_list", "operator": "in", "value": csv(inputs["endpoint_id_list"])})
|
||||
rd = {"filters": filters if filters else "all"}
|
||||
if inputs.get("incident_id"):
|
||||
rd["incident_id"] = inputs["incident_id"]
|
||||
print(json.dumps(post("/endpoints/abort_scan/", rd)))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,64 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
rd = {"indicator": json.loads(inputs.get("indicator") or "{}"), "name": inputs.get("name"), "status": inputs.get("status") or "ENABLED"}
|
||||
print(json.dumps(post("/alerts_exclusion/add/", rd)))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,65 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
eids = csv(inputs.get("endpoint_ids"))
|
||||
body = {"context": {"lcaas_id": eids}, "request_data": {"filters": [{"field": "endpoint_id_list", "operator": "in", "value": eids}], "tag": inputs.get("tag")}}
|
||||
print(json.dumps(post_raw("/tags/agents/assign/", body)))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,74 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
filters = []
|
||||
if inputs.get("email"):
|
||||
filters.append({"field": "email", "operator": "in", "value": csv(inputs["email"])})
|
||||
if inputs.get("type"):
|
||||
filters.append({"field": "type", "operator": "in", "value": csv(inputs["type"])})
|
||||
g = to_ms(inputs.get("timestamp_gte"))
|
||||
if g is not None:
|
||||
filters.append({"field": "timestamp", "operator": "gte", "value": g})
|
||||
rd = {"search_from": 0, "search_to": int(inputs.get("limit") or 100)}
|
||||
if filters:
|
||||
rd["filters"] = filters
|
||||
print(json.dumps(post("/audits/management_logs/", rd)))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,64 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
rd = {"name": inputs.get("name"), "platform": inputs.get("platform"), "package_type": inputs.get("package_type") or "standalone", "agent_version": inputs.get("agent_version"), "description": inputs.get("description") or ""}
|
||||
print(json.dumps(post("/distributions/create/", rd)))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,64 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
rd = {"filters": [{"field": "endpoint_id_list", "operator": "in", "value": csv(inputs.get("endpoint_ids"))}]}
|
||||
print(json.dumps(post("/endpoints/delete/", rd)))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,63 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
print(json.dumps(post("/alerts_exclusion/delete/", {"alert_exclusion_id": int(inputs.get("alert_exclusion_id"))})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,74 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
filters = []
|
||||
if inputs.get("endpoint_ids"):
|
||||
filters.append({"field": "endpoint_id", "operator": "in", "value": csv(inputs["endpoint_ids"])})
|
||||
if inputs.get("type"):
|
||||
filters.append({"field": "type", "operator": "in", "value": csv(inputs["type"])})
|
||||
g = to_ms(inputs.get("timestamp_gte"))
|
||||
if g is not None:
|
||||
filters.append({"field": "timestamp", "operator": "gte", "value": g})
|
||||
rd = {"search_from": 0, "search_to": int(inputs.get("limit") or 100)}
|
||||
if filters:
|
||||
rd["filters"] = filters
|
||||
print(json.dumps(post("/audits/agents_reports/", rd)))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,71 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
filters = []
|
||||
if inputs.get("endpoint_ids"):
|
||||
filters.append({"field": "endpoint_id_list", "operator": "in", "value": csv(inputs["endpoint_ids"])})
|
||||
if inputs.get("type"):
|
||||
filters.append({"field": "type", "operator": "in", "value": csv(inputs["type"])})
|
||||
g = to_ms(inputs.get("timestamp_gte"))
|
||||
if g is not None:
|
||||
filters.append({"field": "timestamp", "operator": "gte", "value": g})
|
||||
print(json.dumps(post("/device_control/get_violations/", {"filters": filters})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,63 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
print(json.dumps(post("/distributions/get_status/", {"distribution_id": inputs.get("distribution_id")})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,63 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
print(json.dumps(post("/distributions/get_dist_url/", {"distribution_id": inputs.get("distribution_id"), "package_type": inputs.get("package_type")})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,66 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
rd = {}
|
||||
if inputs.get("tenant_id"):
|
||||
rd["tenant_id"] = inputs["tenant_id"]
|
||||
print(json.dumps(post("/alerts_exclusion/", rd)))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,63 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
print(json.dumps(post("/alerts/get_original_alerts/", {"alert_id_list": csv(inputs.get("alert_id_list"))})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,63 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
print(json.dumps(post("/get_risk_score/", {"id": inputs.get("id")})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,63 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
print(json.dumps(post("/scripts/get_script_code/", {"script_uid": inputs.get("script_uid")})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,63 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
print(json.dumps(post("/scripts/get_script_execution_status/", {"action_id": inputs.get("action_id")})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,63 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
print(json.dumps(post("/scripts/get_script_metadata/", {"script_uid": inputs.get("script_uid")})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,62 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
print(json.dumps(post("/get_risky_hosts/")))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,62 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
print(json.dumps(post("/get_risky_users/")))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,63 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
print(json.dumps(post("/rbac/get_roles/", {"role_names": csv(inputs.get("role_names"))})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,63 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
print(json.dumps(post("/rbac/get_user_group/", {"group_names": csv(inputs.get("group_names"))})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,62 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
print(json.dumps(post("/rbac/get_users/", {})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,65 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
eids = csv(inputs.get("endpoint_ids"))
|
||||
body = {"context": {"lcaas_id": eids}, "request_data": {"filters": [{"field": "endpoint_id_list", "operator": "in", "value": eids}], "tag": inputs.get("tag")}}
|
||||
print(json.dumps(post_raw("/tags/agents/remove/", body)))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,63 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
print(json.dumps(post("/rbac/set_user_role/", {"user_emails": csv(inputs.get("user_emails"))})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,74 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
eids = csv(inputs.get("endpoint_ids"))
|
||||
files = {}
|
||||
if inputs.get("windows"):
|
||||
files["windows"] = csv(inputs["windows"])
|
||||
if inputs.get("linux"):
|
||||
files["linux"] = csv(inputs["linux"])
|
||||
if inputs.get("macos"):
|
||||
files["macos"] = csv(inputs["macos"])
|
||||
rd = {"filters": [{"field": "endpoint_id_list", "operator": "in", "value": eids}], "files": files}
|
||||
if inputs.get("incident_id"):
|
||||
rd["incident_id"] = inputs["incident_id"]
|
||||
print(json.dumps(post("/endpoints/file_retrieval/", rd)))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,63 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
print(json.dumps(post("/actions/file_retrieval_details/", {"group_action_id": int(inputs.get("action_id"))})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,66 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
rd = {"filters": [{"field": "endpoint_id_list", "operator": "in", "value": csv(inputs.get("endpoint_ids"))}], "snippet_code": inputs.get("snippet_code")}
|
||||
if inputs.get("incident_id"):
|
||||
rd["incident_id"] = inputs["incident_id"]
|
||||
print(json.dumps(post("/scripts/run_snippet_code_script", rd)))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,64 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
rd = {"filters": [{"field": "endpoint_id_list", "operator": "in", "value": csv(inputs.get("endpoint_ids"))}], "alias": inputs.get("alias")}
|
||||
print(json.dumps(post("/endpoints/update_agent_name/", rd)))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
@@ -0,0 +1,63 @@
|
||||
import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _client():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = s.get("url", "").rstrip("/") + "/public_api/v1"
|
||||
key = s.get("api_key", "")
|
||||
kid = str(s.get("api_key_id", ""))
|
||||
headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
|
||||
if str(s.get("auth_type") or "standard").lower() == "advanced":
|
||||
nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
|
||||
ts = str(int(time.time()) * 1000)
|
||||
headers["x-xdr-nonce"] = nonce
|
||||
headers["x-xdr-timestamp"] = ts
|
||||
headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
|
||||
else:
|
||||
headers["Authorization"] = key
|
||||
return base, headers
|
||||
|
||||
|
||||
def post_raw(path, body):
|
||||
base, headers = _client()
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def post(path, request_data=None):
|
||||
return post_raw(path, {"request_data": request_data} if request_data is not None else None)
|
||||
|
||||
|
||||
def csv(v):
|
||||
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
||||
|
||||
|
||||
def to_ms(v):
|
||||
if v in (None, ""):
|
||||
return None
|
||||
s = str(v)
|
||||
if s.isdigit():
|
||||
return int(s)
|
||||
try:
|
||||
from datetime import datetime
|
||||
return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
print(json.dumps(post("/rbac/set_user_role/", {"user_emails": csv(inputs.get("user_emails")), "role_name": inputs.get("role_name")})))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
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)
|
||||
Reference in New Issue
Block a user