feat(azure-security): new Microsoft Azure cloud-containment integration
Azure Resource Manager (Defender for Cloud + Network), 9 commands: list/get security alerts, update alert state, secure score, list/get NSGs, create/delete NSG security rules (deny inbound to isolate). OAuth2 client-credentials, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
@@ -0,0 +1,120 @@
|
||||
id: azure_security
|
||||
name: Microsoft Azure
|
||||
version: 1.0.0
|
||||
description: "Microsoft Azure (Resource Manager: Defender for Cloud + Network) — cloud containment: list and read Defender for Cloud security alerts and update their state, read the secure score, list/read network security groups (NSGs), and add or delete NSG security rules (deny inbound to isolate). Azure AD OAuth2 client-credentials authentication; stdlib-only, no extra Python dependencies."
|
||||
changelog: "1.0.0 — Initial release: list/get security alerts, update alert state, secure score, list/get NSGs, create/delete NSG security rules."
|
||||
category: cloud
|
||||
|
||||
# Per-instance configuration. Uses application (client-credentials) OAuth2 with
|
||||
# a service principal that has Reader + Security Admin + Network Contributor on
|
||||
# the subscription.
|
||||
config_schema:
|
||||
properties:
|
||||
tenant_id:
|
||||
type: string
|
||||
description: "Azure AD tenant ID"
|
||||
client_id:
|
||||
type: string
|
||||
description: "Service principal (client) ID"
|
||||
client_secret:
|
||||
type: string
|
||||
description: "Service principal client secret"
|
||||
x-soar-sensitive: true
|
||||
subscription_id:
|
||||
type: string
|
||||
description: "Azure subscription ID"
|
||||
required:
|
||||
- tenant_id
|
||||
- client_id
|
||||
- client_secret
|
||||
- subscription_id
|
||||
|
||||
commands:
|
||||
- id: list_alerts
|
||||
name: azure-list-alerts
|
||||
description: "List Microsoft Defender for Cloud security alerts in the subscription."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_alert
|
||||
name: azure-get-alert
|
||||
description: "Get a single security alert by its full ARM resource ID."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
alert_id: { type: string, description: "Full ARM resource ID of the alert (from azure-list-alerts)" }
|
||||
required: [alert_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: update_alert_state
|
||||
name: azure-update-alert-state
|
||||
description: "Change a security alert's state (dismiss, resolve, activate, or inProgress)."
|
||||
inputs_schema:
|
||||
properties:
|
||||
alert_id: { type: string, description: "Full ARM resource ID of the alert" }
|
||||
state: { type: string, description: "dismiss | resolve | activate | inProgress" }
|
||||
required: [alert_id, state]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_secure_score
|
||||
name: azure-get-secure-score
|
||||
description: "Get the subscription's Defender for Cloud secure score."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: list_nsgs
|
||||
name: azure-list-nsgs
|
||||
description: "List network security groups in the subscription."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_nsg
|
||||
name: azure-get-nsg
|
||||
description: "Get a single network security group."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
resource_group: { type: string, description: "Resource group name" }
|
||||
nsg_name: { type: string, description: "NSG name" }
|
||||
required: [resource_group, nsg_name]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: create_nsg_rule
|
||||
name: azure-create-nsg-rule
|
||||
description: "Create or update an NSG security rule (e.g. a Deny inbound rule to isolate a resource)."
|
||||
inputs_schema:
|
||||
properties:
|
||||
resource_group: { type: string, description: "Resource group name" }
|
||||
nsg_name: { type: string, description: "NSG name" }
|
||||
rule_name: { type: string, description: "Security rule name" }
|
||||
priority: { type: number, description: "Rule priority (100-4096)" }
|
||||
direction: { type: string, description: "Inbound or Outbound (default Inbound)" }
|
||||
access: { type: string, description: "Allow or Deny (default Deny)" }
|
||||
protocol: { type: string, description: "Tcp, Udp, or * (default *)" }
|
||||
source: { type: string, description: "Source address prefix (CIDR or *, default *)" }
|
||||
destination: { type: string, description: "Destination address prefix (default *)" }
|
||||
destination_port: { type: string, description: "Destination port range (default *)" }
|
||||
required: [resource_group, nsg_name, rule_name, priority]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: delete_nsg_rule
|
||||
name: azure-delete-nsg-rule
|
||||
description: "Delete an NSG security rule."
|
||||
inputs_schema:
|
||||
properties:
|
||||
resource_group: { type: string, description: "Resource group name" }
|
||||
nsg_name: { type: string, description: "NSG name" }
|
||||
rule_name: { type: string, description: "Security rule name" }
|
||||
required: [resource_group, nsg_name, rule_name]
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
- id: test_connection
|
||||
name: azure-test-connection
|
||||
description: "Verify connectivity and the service-principal credentials (used by the Test button)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
@@ -0,0 +1,103 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
ARM = "https://management.azure.com"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _token(cfg):
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id", "")) + "/oauth2/v2.0/token"
|
||||
form = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id", "")),
|
||||
"client_secret": str(cfg.get("client_secret", "")),
|
||||
"scope": "https://management.azure.com/.default",
|
||||
}).encode("utf-8")
|
||||
req = urllib.request.Request(url, data=form, headers={"Content-Type": "application/x-www-form-urlencoded"}, 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 arm(method, path, token, api_version, body=None, params=None, full_url=None):
|
||||
url = (full_url if full_url else ARM + path)
|
||||
qp = {"api-version": api_version}
|
||||
if params:
|
||||
qp.update({k: v for k, v in params.items() if v not in (None, "")})
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(qp)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + token, "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
cfg = _cfg()
|
||||
inputs = _inputs()
|
||||
token = _token(cfg)
|
||||
print(json.dumps(fn(cfg, token, inputs)))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
sub = lambda cfg: str(cfg.get("subscription_id", ""))
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main(cfg, token, inputs):
|
||||
resource_group = inputs.get("resource_group")
|
||||
if not resource_group:
|
||||
raise Exception("resource_group is required")
|
||||
nsg_name = inputs.get("nsg_name")
|
||||
if not nsg_name:
|
||||
raise Exception("nsg_name is required")
|
||||
rule_name = inputs.get("rule_name")
|
||||
if not rule_name:
|
||||
raise Exception("rule_name is required")
|
||||
priority = inputs.get("priority")
|
||||
if priority in (None, ""):
|
||||
raise Exception("priority is required")
|
||||
|
||||
direction = inputs.get("direction")
|
||||
access = inputs.get("access")
|
||||
protocol = inputs.get("protocol")
|
||||
source = inputs.get("source")
|
||||
destination = inputs.get("destination")
|
||||
destination_port = inputs.get("destination_port")
|
||||
|
||||
path = ("/subscriptions/" + sub(cfg) + "/resourceGroups/" + q(resource_group) +
|
||||
"/providers/Microsoft.Network/networkSecurityGroups/" + q(nsg_name) +
|
||||
"/securityRules/" + q(rule_name))
|
||||
body = {
|
||||
"properties": {
|
||||
"priority": int(priority),
|
||||
"direction": direction or "Inbound",
|
||||
"access": access or "Deny",
|
||||
"protocol": protocol or "*",
|
||||
"sourceAddressPrefix": source or "*",
|
||||
"destinationAddressPrefix": destination or "*",
|
||||
"sourcePortRange": "*",
|
||||
"destinationPortRange": destination_port or "*",
|
||||
}
|
||||
}
|
||||
return arm("PUT", path, token, "2023-09-01", body=body)
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,84 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
ARM = "https://management.azure.com"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _token(cfg):
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id", "")) + "/oauth2/v2.0/token"
|
||||
form = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id", "")),
|
||||
"client_secret": str(cfg.get("client_secret", "")),
|
||||
"scope": "https://management.azure.com/.default",
|
||||
}).encode("utf-8")
|
||||
req = urllib.request.Request(url, data=form, headers={"Content-Type": "application/x-www-form-urlencoded"}, 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 arm(method, path, token, api_version, body=None, params=None, full_url=None):
|
||||
url = (full_url if full_url else ARM + path)
|
||||
qp = {"api-version": api_version}
|
||||
if params:
|
||||
qp.update({k: v for k, v in params.items() if v not in (None, "")})
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(qp)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + token, "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
cfg = _cfg()
|
||||
inputs = _inputs()
|
||||
token = _token(cfg)
|
||||
print(json.dumps(fn(cfg, token, inputs)))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
sub = lambda cfg: str(cfg.get("subscription_id", ""))
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main(cfg, token, inputs):
|
||||
resource_group = inputs.get("resource_group")
|
||||
if not resource_group:
|
||||
raise Exception("resource_group is required")
|
||||
nsg_name = inputs.get("nsg_name")
|
||||
if not nsg_name:
|
||||
raise Exception("nsg_name is required")
|
||||
rule_name = inputs.get("rule_name")
|
||||
if not rule_name:
|
||||
raise Exception("rule_name is required")
|
||||
|
||||
path = ("/subscriptions/" + sub(cfg) + "/resourceGroups/" + q(resource_group) +
|
||||
"/providers/Microsoft.Network/networkSecurityGroups/" + q(nsg_name) +
|
||||
"/securityRules/" + q(rule_name))
|
||||
result = arm("DELETE", path, token, "2023-09-01")
|
||||
if not result:
|
||||
return {"ok": True, "deleted": rule_name}
|
||||
return result
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,67 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
ARM = "https://management.azure.com"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _token(cfg):
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id", "")) + "/oauth2/v2.0/token"
|
||||
form = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id", "")),
|
||||
"client_secret": str(cfg.get("client_secret", "")),
|
||||
"scope": "https://management.azure.com/.default",
|
||||
}).encode("utf-8")
|
||||
req = urllib.request.Request(url, data=form, headers={"Content-Type": "application/x-www-form-urlencoded"}, 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 arm(method, path, token, api_version, body=None, params=None, full_url=None):
|
||||
url = (full_url if full_url else ARM + path)
|
||||
qp = {"api-version": api_version}
|
||||
if params:
|
||||
qp.update({k: v for k, v in params.items() if v not in (None, "")})
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(qp)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + token, "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
cfg = _cfg()
|
||||
inputs = _inputs()
|
||||
token = _token(cfg)
|
||||
print(json.dumps(fn(cfg, token, inputs)))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def main(cfg, token, inputs):
|
||||
alert_id = inputs.get("alert_id")
|
||||
if not alert_id:
|
||||
raise Exception("alert_id is required")
|
||||
return arm("GET", None, token, "2022-01-01", full_url=ARM + str(alert_id))
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,76 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
ARM = "https://management.azure.com"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _token(cfg):
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id", "")) + "/oauth2/v2.0/token"
|
||||
form = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id", "")),
|
||||
"client_secret": str(cfg.get("client_secret", "")),
|
||||
"scope": "https://management.azure.com/.default",
|
||||
}).encode("utf-8")
|
||||
req = urllib.request.Request(url, data=form, headers={"Content-Type": "application/x-www-form-urlencoded"}, 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 arm(method, path, token, api_version, body=None, params=None, full_url=None):
|
||||
url = (full_url if full_url else ARM + path)
|
||||
qp = {"api-version": api_version}
|
||||
if params:
|
||||
qp.update({k: v for k, v in params.items() if v not in (None, "")})
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(qp)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + token, "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
cfg = _cfg()
|
||||
inputs = _inputs()
|
||||
token = _token(cfg)
|
||||
print(json.dumps(fn(cfg, token, inputs)))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
sub = lambda cfg: str(cfg.get("subscription_id", ""))
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main(cfg, token, inputs):
|
||||
resource_group = inputs.get("resource_group")
|
||||
if not resource_group:
|
||||
raise Exception("resource_group is required")
|
||||
nsg_name = inputs.get("nsg_name")
|
||||
if not nsg_name:
|
||||
raise Exception("nsg_name is required")
|
||||
path = ("/subscriptions/" + sub(cfg) + "/resourceGroups/" + q(resource_group) +
|
||||
"/providers/Microsoft.Network/networkSecurityGroups/" + q(nsg_name))
|
||||
return arm("GET", path, token, "2023-09-01")
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,68 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
ARM = "https://management.azure.com"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _token(cfg):
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id", "")) + "/oauth2/v2.0/token"
|
||||
form = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id", "")),
|
||||
"client_secret": str(cfg.get("client_secret", "")),
|
||||
"scope": "https://management.azure.com/.default",
|
||||
}).encode("utf-8")
|
||||
req = urllib.request.Request(url, data=form, headers={"Content-Type": "application/x-www-form-urlencoded"}, 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 arm(method, path, token, api_version, body=None, params=None, full_url=None):
|
||||
url = (full_url if full_url else ARM + path)
|
||||
qp = {"api-version": api_version}
|
||||
if params:
|
||||
qp.update({k: v for k, v in params.items() if v not in (None, "")})
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(qp)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + token, "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
cfg = _cfg()
|
||||
inputs = _inputs()
|
||||
token = _token(cfg)
|
||||
print(json.dumps(fn(cfg, token, inputs)))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
sub = lambda cfg: str(cfg.get("subscription_id", ""))
|
||||
|
||||
|
||||
def main(cfg, token, inputs):
|
||||
path = "/subscriptions/" + sub(cfg) + "/providers/Microsoft.Security/secureScores/ascScore"
|
||||
return arm("GET", path, token, "2020-01-01")
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,68 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
ARM = "https://management.azure.com"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _token(cfg):
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id", "")) + "/oauth2/v2.0/token"
|
||||
form = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id", "")),
|
||||
"client_secret": str(cfg.get("client_secret", "")),
|
||||
"scope": "https://management.azure.com/.default",
|
||||
}).encode("utf-8")
|
||||
req = urllib.request.Request(url, data=form, headers={"Content-Type": "application/x-www-form-urlencoded"}, 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 arm(method, path, token, api_version, body=None, params=None, full_url=None):
|
||||
url = (full_url if full_url else ARM + path)
|
||||
qp = {"api-version": api_version}
|
||||
if params:
|
||||
qp.update({k: v for k, v in params.items() if v not in (None, "")})
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(qp)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + token, "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
cfg = _cfg()
|
||||
inputs = _inputs()
|
||||
token = _token(cfg)
|
||||
print(json.dumps(fn(cfg, token, inputs)))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
sub = lambda cfg: str(cfg.get("subscription_id", ""))
|
||||
|
||||
|
||||
def main(cfg, token, inputs):
|
||||
path = "/subscriptions/" + sub(cfg) + "/providers/Microsoft.Security/alerts"
|
||||
return arm("GET", path, token, "2022-01-01")
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,68 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
ARM = "https://management.azure.com"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _token(cfg):
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id", "")) + "/oauth2/v2.0/token"
|
||||
form = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id", "")),
|
||||
"client_secret": str(cfg.get("client_secret", "")),
|
||||
"scope": "https://management.azure.com/.default",
|
||||
}).encode("utf-8")
|
||||
req = urllib.request.Request(url, data=form, headers={"Content-Type": "application/x-www-form-urlencoded"}, 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 arm(method, path, token, api_version, body=None, params=None, full_url=None):
|
||||
url = (full_url if full_url else ARM + path)
|
||||
qp = {"api-version": api_version}
|
||||
if params:
|
||||
qp.update({k: v for k, v in params.items() if v not in (None, "")})
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(qp)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + token, "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
cfg = _cfg()
|
||||
inputs = _inputs()
|
||||
token = _token(cfg)
|
||||
print(json.dumps(fn(cfg, token, inputs)))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
sub = lambda cfg: str(cfg.get("subscription_id", ""))
|
||||
|
||||
|
||||
def main(cfg, token, inputs):
|
||||
path = "/subscriptions/" + sub(cfg) + "/providers/Microsoft.Network/networkSecurityGroups"
|
||||
return arm("GET", path, token, "2023-09-01")
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,69 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
ARM = "https://management.azure.com"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _token(cfg):
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id", "")) + "/oauth2/v2.0/token"
|
||||
form = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id", "")),
|
||||
"client_secret": str(cfg.get("client_secret", "")),
|
||||
"scope": "https://management.azure.com/.default",
|
||||
}).encode("utf-8")
|
||||
req = urllib.request.Request(url, data=form, headers={"Content-Type": "application/x-www-form-urlencoded"}, 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 arm(method, path, token, api_version, body=None, params=None, full_url=None):
|
||||
url = (full_url if full_url else ARM + path)
|
||||
qp = {"api-version": api_version}
|
||||
if params:
|
||||
qp.update({k: v for k, v in params.items() if v not in (None, "")})
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(qp)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + token, "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
cfg = _cfg()
|
||||
inputs = _inputs()
|
||||
token = _token(cfg)
|
||||
print(json.dumps(fn(cfg, token, inputs)))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
sub = lambda cfg: str(cfg.get("subscription_id", ""))
|
||||
|
||||
|
||||
def main(cfg, token, inputs):
|
||||
path = "/subscriptions/" + sub(cfg)
|
||||
arm("GET", path, token, "2022-12-01")
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,78 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
ARM = "https://management.azure.com"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _token(cfg):
|
||||
url = "https://login.microsoftonline.com/" + str(cfg.get("tenant_id", "")) + "/oauth2/v2.0/token"
|
||||
form = urllib.parse.urlencode({
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": str(cfg.get("client_id", "")),
|
||||
"client_secret": str(cfg.get("client_secret", "")),
|
||||
"scope": "https://management.azure.com/.default",
|
||||
}).encode("utf-8")
|
||||
req = urllib.request.Request(url, data=form, headers={"Content-Type": "application/x-www-form-urlencoded"}, 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 arm(method, path, token, api_version, body=None, params=None, full_url=None):
|
||||
url = (full_url if full_url else ARM + path)
|
||||
qp = {"api-version": api_version}
|
||||
if params:
|
||||
qp.update({k: v for k, v in params.items() if v not in (None, "")})
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(qp)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + token, "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
cfg = _cfg()
|
||||
inputs = _inputs()
|
||||
token = _token(cfg)
|
||||
print(json.dumps(fn(cfg, token, inputs)))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
VALID_STATES = ("dismiss", "resolve", "activate", "inProgress")
|
||||
|
||||
|
||||
def main(cfg, token, inputs):
|
||||
alert_id = inputs.get("alert_id")
|
||||
if not alert_id:
|
||||
raise Exception("alert_id is required")
|
||||
state = inputs.get("state")
|
||||
if not state:
|
||||
raise Exception("state is required")
|
||||
if state not in VALID_STATES:
|
||||
raise Exception("state must be one of: " + ", ".join(VALID_STATES))
|
||||
result = arm("POST", None, token, "2022-01-01", body=None, full_url=ARM + str(alert_id) + "/" + state)
|
||||
if not result:
|
||||
return {"ok": True, "state": state}
|
||||
return result
|
||||
|
||||
|
||||
_run(main)
|
||||
Reference in New Issue
Block a user