feat(wiz): new Wiz CNAPP integration

Wiz GraphQL API, 6 commands: list/get issues, update issue status, list
vulnerabilities, list cloud resources. OAuth2 client-credentials auth,
stdlib-only.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
2026-07-12 00:53:02 +02:00
parent 52356523c9
commit fe7dae8bd3
7 changed files with 564 additions and 0 deletions
+90
View File
@@ -0,0 +1,90 @@
id: wiz
name: Wiz
version: 1.0.0
description: "Wiz (CNAPP GraphQL API) — cloud security posture: list and read issues, update an issue's status, list vulnerability findings, and search cloud resources. OAuth2 client-credentials authentication; stdlib-only, no extra Python dependencies."
changelog: "1.0.0 — Initial release: list/get issues, update issue status, list vulnerabilities, list cloud resources."
category: cloud
# Per-instance configuration. Client credentials are exchanged at the auth URL
# for a bearer token used against the tenant GraphQL endpoint.
config_schema:
properties:
api_url:
type: string
description: "Wiz GraphQL API endpoint (e.g. https://api.us1.app.wiz.io/graphql)"
auth_url:
type: string
description: "Wiz OAuth token URL"
default: "https://auth.app.wiz.io/oauth/token"
client_id:
type: string
description: "Wiz service account client ID"
client_secret:
type: string
description: "Wiz service account client secret"
x-soar-sensitive: true
required:
- api_url
- client_id
- client_secret
commands:
- id: list_issues
name: wiz-list-issues
description: "List issues (optionally filtered by status and severity)."
risk: read
inputs_schema:
properties:
status: { type: string, description: "Status filter (OPEN, IN_PROGRESS, RESOLVED, REJECTED)" }
severity: { type: string, description: "Severity filter (CRITICAL, HIGH, MEDIUM, LOW, INFORMATIONAL)" }
first: { type: number, description: "Max issues (default 20)" }
required: []
outputs_schema: { properties: {} }
- id: get_issue
name: wiz-get-issue
description: "Get a single issue by ID."
risk: read
inputs_schema:
properties:
issue_id: { type: string, description: "Issue ID" }
required: [issue_id]
outputs_schema: { properties: {} }
- id: update_issue_status
name: wiz-update-issue-status
description: "Update an issue's status (e.g. resolve or reject)."
inputs_schema:
properties:
issue_id: { type: string, description: "Issue ID" }
status: { type: string, description: "OPEN, IN_PROGRESS, RESOLVED, or REJECTED" }
note: { type: string, description: "Optional note" }
required: [issue_id, status]
outputs_schema: { properties: {} }
- id: list_vulnerabilities
name: wiz-list-vulnerabilities
description: "List vulnerability findings."
risk: read
inputs_schema:
properties:
severity: { type: string, description: "Severity filter (CRITICAL, HIGH, MEDIUM, LOW)" }
first: { type: number, description: "Max findings (default 20)" }
required: []
outputs_schema: { properties: {} }
- id: list_cloud_resources
name: wiz-list-cloud-resources
description: "Search cloud resources by a text query."
risk: read
inputs_schema:
properties:
search: { type: string, description: "Free-text search over resource names" }
first: { type: number, description: "Max resources (default 20)" }
required: []
outputs_schema: { properties: {} }
- id: test_connection
name: wiz-test-connection
description: "Verify the token exchange and GraphQL access (used by the Test button)."
risk: read
inputs_schema:
properties: {}
required: []
outputs_schema: { properties: {} }
+75
View File
@@ -0,0 +1,75 @@
import json, os, sys, urllib.parse, urllib.request, urllib.error
def _cfg():
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
def _inputs():
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
def _token(cfg):
auth_url = str(cfg.get("auth_url") or "https://auth.app.wiz.io/oauth/token")
form = urllib.parse.urlencode({
"grant_type": "client_credentials",
"client_id": str(cfg.get("client_id", "")),
"client_secret": str(cfg.get("client_secret", "")),
"audience": "wiz-api",
}).encode("utf-8")
req = urllib.request.Request(auth_url, data=form,
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 graphql(cfg, token, query, variables):
body = json.dumps({"query": query, "variables": variables}).encode("utf-8")
req = urllib.request.Request(str(cfg.get("api_url", "")), data=body, headers={
"Authorization": "Bearer " + token,
"Content-Type": "application/json",
"Accept": "application/json",
}, method="POST")
with urllib.request.urlopen(req, timeout=90) as r:
raw = r.read()
resp = json.loads(raw) if raw else {}
if resp.get("errors"):
raise Exception("GraphQL error: " + json.dumps(resp["errors"]))
return resp.get("data", resp)
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)
QUERY = """
query Issue($id: ID!) {
issue(id: $id) { id status severity type createdAt notes { text } entitySnapshot { name type } }
}
"""
def main(cfg, token, inputs):
issue_id = inputs.get("issue_id")
if not issue_id:
raise Exception("issue_id is required")
variables = {"id": issue_id}
return graphql(cfg, token, QUERY, variables)
_run(main)
@@ -0,0 +1,81 @@
import json, os, sys, urllib.parse, urllib.request, urllib.error
def _cfg():
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
def _inputs():
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
def _token(cfg):
auth_url = str(cfg.get("auth_url") or "https://auth.app.wiz.io/oauth/token")
form = urllib.parse.urlencode({
"grant_type": "client_credentials",
"client_id": str(cfg.get("client_id", "")),
"client_secret": str(cfg.get("client_secret", "")),
"audience": "wiz-api",
}).encode("utf-8")
req = urllib.request.Request(auth_url, data=form,
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 graphql(cfg, token, query, variables):
body = json.dumps({"query": query, "variables": variables}).encode("utf-8")
req = urllib.request.Request(str(cfg.get("api_url", "")), data=body, headers={
"Authorization": "Bearer " + token,
"Content-Type": "application/json",
"Accept": "application/json",
}, method="POST")
with urllib.request.urlopen(req, timeout=90) as r:
raw = r.read()
resp = json.loads(raw) if raw else {}
if resp.get("errors"):
raise Exception("GraphQL error: " + json.dumps(resp["errors"]))
return resp.get("data", resp)
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)
QUERY = """
query CloudResources($first: Int, $filterBy: CloudResourceFilters) {
cloudResources(first: $first, filterBy: $filterBy) {
nodes { id name type subscriptionId region }
pageInfo { hasNextPage endCursor }
}
}
"""
def main(cfg, token, inputs):
search = inputs.get("search")
first = inputs.get("first")
filter_by = {}
if search:
filter_by["search"] = search
variables = {"first": int(first or 20), "filterBy": filter_by}
return graphql(cfg, token, QUERY, variables)
_run(main)
+84
View File
@@ -0,0 +1,84 @@
import json, os, sys, urllib.parse, urllib.request, urllib.error
def _cfg():
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
def _inputs():
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
def _token(cfg):
auth_url = str(cfg.get("auth_url") or "https://auth.app.wiz.io/oauth/token")
form = urllib.parse.urlencode({
"grant_type": "client_credentials",
"client_id": str(cfg.get("client_id", "")),
"client_secret": str(cfg.get("client_secret", "")),
"audience": "wiz-api",
}).encode("utf-8")
req = urllib.request.Request(auth_url, data=form,
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 graphql(cfg, token, query, variables):
body = json.dumps({"query": query, "variables": variables}).encode("utf-8")
req = urllib.request.Request(str(cfg.get("api_url", "")), data=body, headers={
"Authorization": "Bearer " + token,
"Content-Type": "application/json",
"Accept": "application/json",
}, method="POST")
with urllib.request.urlopen(req, timeout=90) as r:
raw = r.read()
resp = json.loads(raw) if raw else {}
if resp.get("errors"):
raise Exception("GraphQL error: " + json.dumps(resp["errors"]))
return resp.get("data", resp)
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)
QUERY = """
query Issues($first: Int, $filterBy: IssueFilters) {
issues(first: $first, filterBy: $filterBy) {
nodes { id status severity createdAt type entitySnapshot { name type } }
pageInfo { hasNextPage endCursor }
}
}
"""
def main(cfg, token, inputs):
status = inputs.get("status")
severity = inputs.get("severity")
first = inputs.get("first")
filter_by = {}
if status:
filter_by["status"] = [status]
if severity:
filter_by["severity"] = [severity]
variables = {"first": int(first or 20), "filterBy": filter_by}
return graphql(cfg, token, QUERY, variables)
_run(main)
@@ -0,0 +1,81 @@
import json, os, sys, urllib.parse, urllib.request, urllib.error
def _cfg():
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
def _inputs():
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
def _token(cfg):
auth_url = str(cfg.get("auth_url") or "https://auth.app.wiz.io/oauth/token")
form = urllib.parse.urlencode({
"grant_type": "client_credentials",
"client_id": str(cfg.get("client_id", "")),
"client_secret": str(cfg.get("client_secret", "")),
"audience": "wiz-api",
}).encode("utf-8")
req = urllib.request.Request(auth_url, data=form,
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 graphql(cfg, token, query, variables):
body = json.dumps({"query": query, "variables": variables}).encode("utf-8")
req = urllib.request.Request(str(cfg.get("api_url", "")), data=body, headers={
"Authorization": "Bearer " + token,
"Content-Type": "application/json",
"Accept": "application/json",
}, method="POST")
with urllib.request.urlopen(req, timeout=90) as r:
raw = r.read()
resp = json.loads(raw) if raw else {}
if resp.get("errors"):
raise Exception("GraphQL error: " + json.dumps(resp["errors"]))
return resp.get("data", resp)
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)
QUERY = """
query Vulns($first: Int, $filterBy: VulnerabilityFindingFilters) {
vulnerabilityFindings(first: $first, filterBy: $filterBy) {
nodes { id name severity cvssSeverity vendorSeverity firstDetectedAt }
pageInfo { hasNextPage endCursor }
}
}
"""
def main(cfg, token, inputs):
severity = inputs.get("severity")
first = inputs.get("first")
filter_by = {}
if severity:
filter_by["severity"] = [severity]
variables = {"first": int(first or 20), "filterBy": filter_by}
return graphql(cfg, token, QUERY, variables)
_run(main)
@@ -0,0 +1,69 @@
import json, os, sys, urllib.parse, urllib.request, urllib.error
def _cfg():
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
def _inputs():
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
def _token(cfg):
auth_url = str(cfg.get("auth_url") or "https://auth.app.wiz.io/oauth/token")
form = urllib.parse.urlencode({
"grant_type": "client_credentials",
"client_id": str(cfg.get("client_id", "")),
"client_secret": str(cfg.get("client_secret", "")),
"audience": "wiz-api",
}).encode("utf-8")
req = urllib.request.Request(auth_url, data=form,
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 graphql(cfg, token, query, variables):
body = json.dumps({"query": query, "variables": variables}).encode("utf-8")
req = urllib.request.Request(str(cfg.get("api_url", "")), data=body, headers={
"Authorization": "Bearer " + token,
"Content-Type": "application/json",
"Accept": "application/json",
}, method="POST")
with urllib.request.urlopen(req, timeout=90) as r:
raw = r.read()
resp = json.loads(raw) if raw else {}
if resp.get("errors"):
raise Exception("GraphQL error: " + json.dumps(resp["errors"]))
return resp.get("data", resp)
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)
QUERY = """
query { issues(first: 1) { nodes { id } } }
"""
def main(cfg, token, inputs):
graphql(cfg, token, QUERY, {})
return {"ok": True}
_run(main)
@@ -0,0 +1,84 @@
import json, os, sys, urllib.parse, urllib.request, urllib.error
def _cfg():
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
def _inputs():
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
def _token(cfg):
auth_url = str(cfg.get("auth_url") or "https://auth.app.wiz.io/oauth/token")
form = urllib.parse.urlencode({
"grant_type": "client_credentials",
"client_id": str(cfg.get("client_id", "")),
"client_secret": str(cfg.get("client_secret", "")),
"audience": "wiz-api",
}).encode("utf-8")
req = urllib.request.Request(auth_url, data=form,
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 graphql(cfg, token, query, variables):
body = json.dumps({"query": query, "variables": variables}).encode("utf-8")
req = urllib.request.Request(str(cfg.get("api_url", "")), data=body, headers={
"Authorization": "Bearer " + token,
"Content-Type": "application/json",
"Accept": "application/json",
}, method="POST")
with urllib.request.urlopen(req, timeout=90) as r:
raw = r.read()
resp = json.loads(raw) if raw else {}
if resp.get("errors"):
raise Exception("GraphQL error: " + json.dumps(resp["errors"]))
return resp.get("data", resp)
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)
QUERY = """
mutation UpdateIssue($input: UpdateIssueInput!) {
updateIssue(input: $input) { issue { id status } }
}
"""
def main(cfg, token, inputs):
issue_id = inputs.get("issue_id")
status = inputs.get("status")
note = inputs.get("note")
if not issue_id:
raise Exception("issue_id is required")
if not status:
raise Exception("status is required")
patch = {"status": status}
if note:
patch["note"] = note
input_obj = {"id": issue_id, "patch": patch}
variables = {"input": input_obj}
return graphql(cfg, token, QUERY, variables)
_run(main)