feat(microsoft-defender-endpoint): new Defender for Endpoint integration
22 commands (Security Center API): alert ingestion + triage, machine isolate/unisolate, restrict/unrestrict app execution, AV scan, stop & quarantine file, collect investigation package, offboard, tag, list machine actions, advanced hunting (KQL), and custom indicators. Azure AD OAuth 2.0 client-credentials, stdlib-only. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
@@ -0,0 +1,3 @@
|
||||
name: "Microsoft Defender Alert"
|
||||
color: "#0078d4"
|
||||
icon: "alert"
|
||||
@@ -0,0 +1,262 @@
|
||||
id: microsoft_defender_endpoint
|
||||
name: Microsoft Defender for Endpoint
|
||||
version: 1.0.0
|
||||
description: "Microsoft Defender for Endpoint (Security Center API) — machine isolation, app-execution restriction, antivirus scan, file quarantine, investigation package collection, offboarding, tagging, alert triage, advanced hunting (KQL), custom indicators, and alert ingestion (get_incidents) with an OCSF mapper. Azure AD OAuth 2.0 client-credentials authentication. Stdlib-only, no extra Python dependencies."
|
||||
changelog: "1.0.0 — Initial release: alert ingestion + triage, machine actions (isolate/unisolate, restrict/unrestrict app execution, AV scan, quarantine file, collect package, offboard, tag), advanced hunting, custom indicators, and machine/action listing."
|
||||
category: endpoint
|
||||
|
||||
# Per-instance configuration. Register an Azure AD application, grant it the
|
||||
# WindowsDefenderATP application permissions (Machine.Isolate, Machine.Scan,
|
||||
# Alert.Read.All, Ti.ReadWrite, AdvancedQuery.Read.All, …) and admin-consent
|
||||
# them. The scripts request a token from Azure AD (client-credentials) for the
|
||||
# Security Center resource, then call https://api.securitycenter.microsoft.com.
|
||||
config_schema:
|
||||
properties:
|
||||
tenant_id:
|
||||
type: string
|
||||
description: "Azure AD tenant (directory) ID"
|
||||
client_id:
|
||||
type: string
|
||||
description: "Application (client) ID"
|
||||
client_secret:
|
||||
type: string
|
||||
description: "Client secret"
|
||||
x-soar-sensitive: true
|
||||
required:
|
||||
- tenant_id
|
||||
- client_id
|
||||
- client_secret
|
||||
|
||||
commands:
|
||||
# ── Ingestion / alerts ────────────────────────────────────────────────────
|
||||
- id: get_incidents
|
||||
name: microsoft-defender-endpoint-get-incidents
|
||||
description: "Fetch Defender alerts for ingestion. Returns {result:[...]}; use result as the alert rule results path."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
status: { type: string, description: "Filter by status (New, InProgress, Resolved)" }
|
||||
severity: { type: string, description: "Filter by severity (Informational, Low, Medium, High)" }
|
||||
created_after: { type: string, description: "Lower bound on alertCreationTime, ISO8601 (incremental fetch watermark)" }
|
||||
limit: { type: number, description: "Maximum alerts to fetch (default 50)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
ingest:
|
||||
results_path: result
|
||||
dedup_key: id
|
||||
incremental_field: created_after
|
||||
- id: list_alerts
|
||||
name: microsoft-defender-endpoint-list-alerts
|
||||
description: "List alerts with an optional OData filter."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
filter: { type: string, description: "OData $filter expression" }
|
||||
limit: { type: number, description: "Maximum alerts (default 50)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_alert
|
||||
name: microsoft-defender-endpoint-get-alert
|
||||
description: "Get a single alert by ID."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
alert_id: { type: string, description: "Alert ID" }
|
||||
required: [alert_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: update_alert
|
||||
name: microsoft-defender-endpoint-update-alert
|
||||
description: "Update an alert's status, classification, determination or assignment."
|
||||
inputs_schema:
|
||||
properties:
|
||||
alert_id: { type: string, description: "Alert ID" }
|
||||
status: { type: string, description: "New, InProgress or Resolved" }
|
||||
classification: { type: string, description: "TruePositive, FalsePositive or InformationalExpectedActivity" }
|
||||
determination: { type: string, description: "Determination, e.g. Malware, SecurityTesting" }
|
||||
assigned_to: { type: string, description: "Assignee" }
|
||||
comment: { type: string, description: "Comment" }
|
||||
required: [alert_id]
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
# ── Machines ──────────────────────────────────────────────────────────────
|
||||
- id: list_machines
|
||||
name: microsoft-defender-endpoint-list-machines
|
||||
description: "List machines with an optional OData filter."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
filter: { type: string, description: "OData $filter expression" }
|
||||
limit: { type: number, description: "Maximum machines (default 50)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_machine
|
||||
name: microsoft-defender-endpoint-get-machine
|
||||
description: "Get a machine's details by machine ID."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
machine_id: { type: string, description: "Machine (device) ID" }
|
||||
required: [machine_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_machine_by_ip
|
||||
name: microsoft-defender-endpoint-get-machine-by-ip
|
||||
description: "Find machines that last reported the given IP at a point in time."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
ip: { type: string, description: "IP address" }
|
||||
timestamp: { type: string, description: "Point in time, ISO8601 (default now)" }
|
||||
required: [ip]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: isolate_machine
|
||||
name: microsoft-defender-endpoint-isolate-machine
|
||||
description: "Network-isolate a machine."
|
||||
inputs_schema:
|
||||
properties:
|
||||
machine_id: { type: string, description: "Machine ID" }
|
||||
comment: { type: string, description: "Reason for the action" }
|
||||
isolation_type: { type: string, description: "Full or Selective (default Full)" }
|
||||
required: [machine_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: unisolate_machine
|
||||
name: microsoft-defender-endpoint-unisolate-machine
|
||||
description: "Release a machine from network isolation."
|
||||
inputs_schema:
|
||||
properties:
|
||||
machine_id: { type: string, description: "Machine ID" }
|
||||
comment: { type: string, description: "Reason for the action" }
|
||||
required: [machine_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: restrict_app_execution
|
||||
name: microsoft-defender-endpoint-restrict-app-execution
|
||||
description: "Restrict application execution on a machine (allow only Microsoft-signed binaries)."
|
||||
inputs_schema:
|
||||
properties:
|
||||
machine_id: { type: string, description: "Machine ID" }
|
||||
comment: { type: string, description: "Reason for the action" }
|
||||
required: [machine_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: remove_app_restriction
|
||||
name: microsoft-defender-endpoint-remove-app-restriction
|
||||
description: "Remove the application-execution restriction from a machine."
|
||||
inputs_schema:
|
||||
properties:
|
||||
machine_id: { type: string, description: "Machine ID" }
|
||||
comment: { type: string, description: "Reason for the action" }
|
||||
required: [machine_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: run_antivirus_scan
|
||||
name: microsoft-defender-endpoint-run-antivirus-scan
|
||||
description: "Trigger a Defender antivirus scan on a machine."
|
||||
inputs_schema:
|
||||
properties:
|
||||
machine_id: { type: string, description: "Machine ID" }
|
||||
scan_type: { type: string, description: "Quick or Full (default Quick)" }
|
||||
comment: { type: string, description: "Reason for the action" }
|
||||
required: [machine_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: stop_and_quarantine_file
|
||||
name: microsoft-defender-endpoint-stop-and-quarantine-file
|
||||
description: "Stop execution of a file (by SHA1) and quarantine it on a machine."
|
||||
inputs_schema:
|
||||
properties:
|
||||
machine_id: { type: string, description: "Machine ID" }
|
||||
sha1: { type: string, description: "SHA1 of the file" }
|
||||
comment: { type: string, description: "Reason for the action" }
|
||||
required: [machine_id, sha1]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: collect_investigation_package
|
||||
name: microsoft-defender-endpoint-collect-investigation-package
|
||||
description: "Collect a forensic investigation package from a machine (returns a machine action; poll it for the download URI)."
|
||||
inputs_schema:
|
||||
properties:
|
||||
machine_id: { type: string, description: "Machine ID" }
|
||||
comment: { type: string, description: "Reason for the action" }
|
||||
required: [machine_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: offboard_machine
|
||||
name: microsoft-defender-endpoint-offboard-machine
|
||||
description: "Offboard a machine from Defender for Endpoint."
|
||||
inputs_schema:
|
||||
properties:
|
||||
machine_id: { type: string, description: "Machine ID" }
|
||||
comment: { type: string, description: "Reason for the action" }
|
||||
required: [machine_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: tag_machine
|
||||
name: microsoft-defender-endpoint-tag-machine
|
||||
description: "Add or remove a tag on a machine."
|
||||
inputs_schema:
|
||||
properties:
|
||||
machine_id: { type: string, description: "Machine ID" }
|
||||
tag: { type: string, description: "Tag value" }
|
||||
action: { type: string, description: "Add or Remove (default Add)" }
|
||||
required: [machine_id, tag]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: list_machine_actions
|
||||
name: microsoft-defender-endpoint-list-machine-actions
|
||||
description: "List machine actions (their status), optionally filtered."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
filter: { type: string, description: "OData $filter expression (e.g. machineId eq '...')" }
|
||||
limit: { type: number, description: "Maximum actions (default 50)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
# ── Hunting / indicators ──────────────────────────────────────────────────
|
||||
- id: advanced_hunting
|
||||
name: microsoft-defender-endpoint-advanced-hunting
|
||||
description: "Run an advanced hunting KQL query."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
query: { type: string, description: "KQL query" }
|
||||
required: [query]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: list_indicators
|
||||
name: microsoft-defender-endpoint-list-indicators
|
||||
description: "List custom threat indicators."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
filter: { type: string, description: "OData $filter expression" }
|
||||
limit: { type: number, description: "Maximum indicators (default 50)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: create_indicator
|
||||
name: microsoft-defender-endpoint-create-indicator
|
||||
description: "Create a custom indicator (file hash, IP, URL or domain) with an action."
|
||||
inputs_schema:
|
||||
properties:
|
||||
indicator_value: { type: string, description: "Indicator value (hash, IP, URL or domain)" }
|
||||
indicator_type: { type: string, description: "FileSha256, FileSha1, FileMd5, IpAddress, DomainName or Url" }
|
||||
action: { type: string, description: "Alert, AlertAndBlock, Block, Allowed or Warn (default Alert)" }
|
||||
title: { type: string, description: "Indicator title" }
|
||||
description: { type: string, description: "Indicator description" }
|
||||
severity: { type: string, description: "Informational, Low, Medium or High" }
|
||||
expiration_time: { type: string, description: "Expiration, ISO8601" }
|
||||
required: [indicator_value, indicator_type]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: delete_indicator
|
||||
name: microsoft-defender-endpoint-delete-indicator
|
||||
description: "Delete a custom indicator by ID."
|
||||
inputs_schema:
|
||||
properties:
|
||||
indicator_id: { type: string, description: "Indicator ID" }
|
||||
required: [indicator_id]
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
- id: test_connection
|
||||
name: microsoft-defender-endpoint-test-connection
|
||||
description: "Verify connectivity and credentials (used by the Test button)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
ingestion:
|
||||
command: get_incidents
|
||||
mapper: get_incidents
|
||||
default_incident_type: "Microsoft Defender Alert"
|
||||
@@ -0,0 +1,21 @@
|
||||
name: "Microsoft Defender Alerts → OCSF"
|
||||
description: "Maps a Defender for Endpoint alert (get_incidents, results_path = result) to OCSF finding fields. Uses the first evidence entry for host/user/file where present."
|
||||
field_mappings:
|
||||
title: "title"
|
||||
severity: "severity = 'High' ? 4 : (severity = 'Medium' ? 3 : (severity = 'Low' ? 2 : 1))"
|
||||
description: "description"
|
||||
ocsf:
|
||||
- { source_path: "id", ocsf_field: "finding_info.uid" }
|
||||
- { source_path: "title", ocsf_field: "finding_info.title" }
|
||||
- { source_path: "description", ocsf_field: "finding_info.desc" }
|
||||
- { source_path: "category", ocsf_field: "finding_info.analytic.category" }
|
||||
- { source_path: "alertCreationTime", ocsf_field: "finding_info.created_time" }
|
||||
- { source_path: "lastUpdateTime", ocsf_field: "finding_info.modified_time" }
|
||||
- { source_path: "status", ocsf_field: "status" }
|
||||
- { source_path: "determination", ocsf_field: "disposition" }
|
||||
- { source_path: "detectionSource", ocsf_field: "finding_info.analytic.name" }
|
||||
- { source_path: "threatFamilyName", ocsf_field: "malware.name" }
|
||||
- { source_path: "computerDnsName", ocsf_field: "device.hostname" }
|
||||
- { source_path: "machineId", ocsf_field: "device.uid" }
|
||||
- { source_path: "assignedTo", ocsf_field: "assignee.name" }
|
||||
- { source_path: "mitreTechniques[0]", ocsf_field: "attacks.technique.uid" }
|
||||
@@ -0,0 +1,62 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
query = inputs.get("query")
|
||||
if not query:
|
||||
raise Exception("query is required")
|
||||
body = {"Query": query}
|
||||
result = request("POST", "/advancedqueries/run", body=body)
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
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, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
machine_id = inputs.get("machine_id")
|
||||
if not machine_id:
|
||||
raise Exception("machine_id is required")
|
||||
comment = inputs.get("comment")
|
||||
body = {"Comment": comment or "Collect package via Riposte"}
|
||||
path = "/machines/" + urllib.parse.quote(str(machine_id), safe="") + "/collectInvestigationPackage"
|
||||
result = request("POST", path, body=body)
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
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,81 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
indicator_value = inputs.get("indicator_value")
|
||||
if not indicator_value:
|
||||
raise Exception("indicator_value is required")
|
||||
indicator_type = inputs.get("indicator_type")
|
||||
if not indicator_type:
|
||||
raise Exception("indicator_type is required")
|
||||
action = inputs.get("action")
|
||||
title = inputs.get("title")
|
||||
description = inputs.get("description")
|
||||
severity = inputs.get("severity")
|
||||
expiration_time = inputs.get("expiration_time")
|
||||
|
||||
body = {}
|
||||
body["indicatorValue"] = indicator_value
|
||||
body["indicatorType"] = indicator_type
|
||||
body["action"] = action or "Alert"
|
||||
body["title"] = title or "Riposte indicator"
|
||||
body["description"] = description or "Created via Riposte"
|
||||
if severity not in (None, ""):
|
||||
body["severity"] = severity
|
||||
if expiration_time not in (None, ""):
|
||||
body["expirationTime"] = expiration_time
|
||||
|
||||
result = request("POST", "/indicators", body=body)
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
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, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
indicator_id = inputs.get("indicator_id")
|
||||
if not indicator_id:
|
||||
raise Exception("indicator_id is required")
|
||||
path = "/indicators/" + urllib.parse.quote(str(indicator_id), safe="")
|
||||
result = request("DELETE", path)
|
||||
if not result:
|
||||
result = {"ok": True, "indicator_id": indicator_id}
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
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, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 {}
|
||||
|
||||
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
quote = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
alert_id = inputs.get("alert_id")
|
||||
if not alert_id:
|
||||
raise Exception("alert_id is required")
|
||||
res = request("GET", "/alerts/" + quote(alert_id))
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
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,89 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
from datetime import datetime, timezone
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 {}
|
||||
|
||||
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
quote = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def _iso(s):
|
||||
s = str(s)
|
||||
if s.isdigit():
|
||||
ts = int(s)
|
||||
if ts > 10 ** 12:
|
||||
ts = ts / 1000.0
|
||||
return datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
try:
|
||||
return datetime.fromisoformat(s.replace("Z", "+00:00")).astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
except Exception:
|
||||
return s
|
||||
|
||||
|
||||
def main():
|
||||
status = inputs.get("status")
|
||||
severity = inputs.get("severity")
|
||||
created_after = inputs.get("created_after")
|
||||
limit = inputs.get("limit") or 50
|
||||
|
||||
parts = []
|
||||
if status:
|
||||
parts.append("status eq '" + str(status) + "'")
|
||||
if severity:
|
||||
parts.append("severity eq '" + str(severity) + "'")
|
||||
if created_after:
|
||||
parts.append("alertCreationTime ge " + _iso(created_after))
|
||||
filt = " and ".join(parts)
|
||||
|
||||
res = request("GET", "/alerts", params={"$filter": filt or None, "$top": limit, "$orderby": "alertCreationTime asc"})
|
||||
print(json.dumps({"result": res.get("value", [])}))
|
||||
|
||||
|
||||
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, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 {}
|
||||
|
||||
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
quote = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
machine_id = inputs.get("machine_id")
|
||||
if not machine_id:
|
||||
raise Exception("machine_id is required")
|
||||
res = request("GET", "/machines/" + quote(machine_id))
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
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,67 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
from datetime import datetime, timezone
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 {}
|
||||
|
||||
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
quote = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
ip = inputs.get("ip")
|
||||
if not ip:
|
||||
raise Exception("ip is required")
|
||||
ts = inputs.get("timestamp") or datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
url = API + "/machines/findbyip(ip='" + str(ip) + "',timestamp=" + str(ts) + ")"
|
||||
res = request("GET", "", full_url=url)
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
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,67 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 {}
|
||||
|
||||
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
quote = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
machine_id = inputs.get("machine_id")
|
||||
if not machine_id:
|
||||
raise Exception("machine_id is required")
|
||||
comment = inputs.get("comment")
|
||||
isolation_type = inputs.get("isolation_type")
|
||||
body = {"Comment": comment or "Isolated via Riposte", "IsolationType": isolation_type or "Full"}
|
||||
res = request("POST", "/machines/" + quote(machine_id) + "/isolate", body=body)
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
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, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 {}
|
||||
|
||||
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
quote = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
filt = inputs.get("filter")
|
||||
limit = inputs.get("limit") or 50
|
||||
res = request("GET", "/alerts", params={"$filter": filt or None, "$top": limit})
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
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,61 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
filter_ = inputs.get("filter")
|
||||
limit = inputs.get("limit") or 50
|
||||
params = {"$filter": filter_ or None, "$top": limit}
|
||||
result = request("GET", "/indicators", params=params)
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
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,61 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
filter_ = inputs.get("filter")
|
||||
limit = inputs.get("limit") or 50
|
||||
params = {"$filter": filter_ or None, "$top": limit}
|
||||
result = request("GET", "/machineactions", params=params)
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
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, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 {}
|
||||
|
||||
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
quote = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
filt = inputs.get("filter")
|
||||
limit = inputs.get("limit") or 50
|
||||
res = request("GET", "/machines", params={"$filter": filt or None, "$top": limit})
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
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, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
machine_id = inputs.get("machine_id")
|
||||
if not machine_id:
|
||||
raise Exception("machine_id is required")
|
||||
comment = inputs.get("comment")
|
||||
body = {"Comment": comment or "Offboard via Riposte"}
|
||||
path = "/machines/" + urllib.parse.quote(str(machine_id), safe="") + "/offboard"
|
||||
result = request("POST", path, body=body)
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
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, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 {}
|
||||
|
||||
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
quote = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
machine_id = inputs.get("machine_id")
|
||||
if not machine_id:
|
||||
raise Exception("machine_id is required")
|
||||
comment = inputs.get("comment")
|
||||
body = {"Comment": comment or "Unrestricted via Riposte"}
|
||||
res = request("POST", "/machines/" + quote(machine_id) + "/unrestrictCodeExecution", body=body)
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
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, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 {}
|
||||
|
||||
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
quote = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
machine_id = inputs.get("machine_id")
|
||||
if not machine_id:
|
||||
raise Exception("machine_id is required")
|
||||
comment = inputs.get("comment")
|
||||
body = {"Comment": comment or "Restricted via Riposte"}
|
||||
res = request("POST", "/machines/" + quote(machine_id) + "/restrictCodeExecution", body=body)
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
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,67 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 {}
|
||||
|
||||
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
quote = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
machine_id = inputs.get("machine_id")
|
||||
if not machine_id:
|
||||
raise Exception("machine_id is required")
|
||||
scan_type = inputs.get("scan_type")
|
||||
comment = inputs.get("comment")
|
||||
body = {"Comment": comment or "AV scan via Riposte", "ScanType": scan_type or "Quick"}
|
||||
res = request("POST", "/machines/" + quote(machine_id) + "/runAntiVirusScan", body=body)
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
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,69 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 {}
|
||||
|
||||
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
quote = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
machine_id = inputs.get("machine_id")
|
||||
sha1 = inputs.get("sha1")
|
||||
if not machine_id:
|
||||
raise Exception("machine_id is required")
|
||||
if not sha1:
|
||||
raise Exception("sha1 is required")
|
||||
comment = inputs.get("comment")
|
||||
body = {"Comment": comment or "Quarantine via Riposte", "Sha1": sha1}
|
||||
res = request("POST", "/machines/" + quote(machine_id) + "/StopAndQuarantineFile", body=body)
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
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,67 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
machine_id = inputs.get("machine_id")
|
||||
if not machine_id:
|
||||
raise Exception("machine_id is required")
|
||||
tag = inputs.get("tag")
|
||||
if not tag:
|
||||
raise Exception("tag is required")
|
||||
action = inputs.get("action")
|
||||
body = {"Value": tag, "Action": action or "Add"}
|
||||
path = "/machines/" + urllib.parse.quote(str(machine_id), safe="") + "/tags"
|
||||
result = request("POST", path, body=body)
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
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, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 {}
|
||||
|
||||
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
quote = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
res = request("GET", "/alerts", params={"$top": 1})
|
||||
if "value" not in res:
|
||||
raise Exception("unexpected response")
|
||||
print(json.dumps({"ok": True}))
|
||||
|
||||
|
||||
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, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 {}
|
||||
|
||||
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
quote = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
machine_id = inputs.get("machine_id")
|
||||
if not machine_id:
|
||||
raise Exception("machine_id is required")
|
||||
comment = inputs.get("comment")
|
||||
body = {"Comment": comment or "Released via Riposte"}
|
||||
res = request("POST", "/machines/" + quote(machine_id) + "/unisolate", body=body)
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
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,79 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
SCOPE = "https://api.securitycenter.microsoft.com/.default"
|
||||
API = "https://api.securitycenter.microsoft.com/api"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _token():
|
||||
cfg = _cfg()
|
||||
data = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id") or ""),
|
||||
"client_secret": str(cfg.get("client_secret") or ""),
|
||||
"scope": SCOPE,
|
||||
}).encode("utf-8")
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id") or "") + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
method="POST")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
tok = json.loads(r.read())
|
||||
if not tok.get("access_token"):
|
||||
raise Exception("Token request failed: " + json.dumps(tok))
|
||||
return tok["access_token"]
|
||||
|
||||
|
||||
def request(method, path, params=None, body=None, full_url=None, token=None):
|
||||
url = full_url or (API + path)
|
||||
q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
|
||||
if q:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + (token or _token())}
|
||||
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 {}
|
||||
|
||||
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
quote = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
alert_id = inputs.get("alert_id")
|
||||
if not alert_id:
|
||||
raise Exception("alert_id is required")
|
||||
|
||||
body = {}
|
||||
if inputs.get("status"):
|
||||
body["status"] = inputs.get("status")
|
||||
if inputs.get("classification"):
|
||||
body["classification"] = inputs.get("classification")
|
||||
if inputs.get("determination"):
|
||||
body["determination"] = inputs.get("determination")
|
||||
if inputs.get("assigned_to"):
|
||||
body["assignedTo"] = inputs.get("assigned_to")
|
||||
if inputs.get("comment"):
|
||||
body["comment"] = inputs.get("comment")
|
||||
if not body:
|
||||
raise Exception("nothing to update")
|
||||
|
||||
res = request("PATCH", "/alerts/" + quote(alert_id), body=body)
|
||||
print(json.dumps(res))
|
||||
|
||||
|
||||
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