feat(tenable-io): new Tenable Vulnerability Management integration
Tenable.io API, 9 commands: list/get assets, list vulnerabilities, per-asset vulnerabilities, vulnerability (plugin) details, list scans, scan status, launch scan. API-key (access/secret) auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,107 @@
|
|||||||
|
id: tenable_io
|
||||||
|
name: Tenable Vulnerability Management
|
||||||
|
version: 1.0.0
|
||||||
|
description: "Tenable Vulnerability Management (Tenable.io API) — asset and vulnerability context for incidents: list and read assets, list vulnerabilities and per-asset vulnerabilities, read vulnerability (plugin) details, and list/launch/status scans. API-key (access/secret) authentication; stdlib-only, no extra Python dependencies."
|
||||||
|
changelog: "1.0.0 — Initial release: list/get assets, list vulnerabilities, per-asset vulnerabilities, vulnerability details, list scans, scan status, launch scan."
|
||||||
|
category: vulnerability
|
||||||
|
|
||||||
|
# Per-instance configuration. Keys are sent as
|
||||||
|
# 'X-ApiKeys: accessKey=<access_key>; secretKey=<secret_key>'.
|
||||||
|
config_schema:
|
||||||
|
properties:
|
||||||
|
url:
|
||||||
|
type: string
|
||||||
|
description: "Tenable.io API base URL"
|
||||||
|
default: "https://cloud.tenable.com"
|
||||||
|
access_key:
|
||||||
|
type: string
|
||||||
|
description: "Tenable API access key"
|
||||||
|
x-soar-sensitive: true
|
||||||
|
secret_key:
|
||||||
|
type: string
|
||||||
|
description: "Tenable API secret key"
|
||||||
|
x-soar-sensitive: true
|
||||||
|
required:
|
||||||
|
- access_key
|
||||||
|
- secret_key
|
||||||
|
|
||||||
|
commands:
|
||||||
|
- id: list_assets
|
||||||
|
name: tenable-list-assets
|
||||||
|
description: "List assets."
|
||||||
|
risk: read
|
||||||
|
inputs_schema:
|
||||||
|
properties: {}
|
||||||
|
required: []
|
||||||
|
outputs_schema: { properties: {} }
|
||||||
|
- id: get_asset
|
||||||
|
name: tenable-get-asset
|
||||||
|
description: "Get a single asset's details by UUID."
|
||||||
|
risk: read
|
||||||
|
inputs_schema:
|
||||||
|
properties:
|
||||||
|
asset_uuid: { type: string, description: "Asset UUID" }
|
||||||
|
required: [asset_uuid]
|
||||||
|
outputs_schema: { properties: {} }
|
||||||
|
- id: asset_vulnerabilities
|
||||||
|
name: tenable-asset-vulnerabilities
|
||||||
|
description: "List the vulnerabilities found on a single asset."
|
||||||
|
risk: read
|
||||||
|
inputs_schema:
|
||||||
|
properties:
|
||||||
|
asset_uuid: { type: string, description: "Asset UUID" }
|
||||||
|
required: [asset_uuid]
|
||||||
|
outputs_schema: { properties: {} }
|
||||||
|
- id: list_vulnerabilities
|
||||||
|
name: tenable-list-vulnerabilities
|
||||||
|
description: "List vulnerabilities across the workbench (optionally filtered by severity age)."
|
||||||
|
risk: read
|
||||||
|
inputs_schema:
|
||||||
|
properties:
|
||||||
|
age: { type: number, description: "Only vulnerabilities seen in the last N days (default 30)" }
|
||||||
|
required: []
|
||||||
|
outputs_schema: { properties: {} }
|
||||||
|
- id: get_vulnerability
|
||||||
|
name: tenable-get-vulnerability
|
||||||
|
description: "Get details for a vulnerability (Nessus plugin) by plugin ID."
|
||||||
|
risk: read
|
||||||
|
inputs_schema:
|
||||||
|
properties:
|
||||||
|
plugin_id: { type: string, description: "Nessus plugin ID" }
|
||||||
|
required: [plugin_id]
|
||||||
|
outputs_schema: { properties: {} }
|
||||||
|
- id: list_scans
|
||||||
|
name: tenable-list-scans
|
||||||
|
description: "List scans."
|
||||||
|
risk: read
|
||||||
|
inputs_schema:
|
||||||
|
properties: {}
|
||||||
|
required: []
|
||||||
|
outputs_schema: { properties: {} }
|
||||||
|
- id: get_scan_status
|
||||||
|
name: tenable-get-scan-status
|
||||||
|
description: "Get the status of a scan."
|
||||||
|
risk: read
|
||||||
|
inputs_schema:
|
||||||
|
properties:
|
||||||
|
scan_id: { type: string, description: "Scan ID" }
|
||||||
|
required: [scan_id]
|
||||||
|
outputs_schema: { properties: {} }
|
||||||
|
- id: launch_scan
|
||||||
|
name: tenable-launch-scan
|
||||||
|
description: "Launch a scan."
|
||||||
|
inputs_schema:
|
||||||
|
properties:
|
||||||
|
scan_id: { type: string, description: "Scan ID to launch" }
|
||||||
|
targets: { type: string, description: "Optional comma-separated alternate targets" }
|
||||||
|
required: [scan_id]
|
||||||
|
outputs_schema: { properties: {} }
|
||||||
|
|
||||||
|
- id: test_connection
|
||||||
|
name: tenable-test-connection
|
||||||
|
description: "Verify connectivity and the API keys (used by the Test button)."
|
||||||
|
risk: read
|
||||||
|
inputs_schema:
|
||||||
|
properties: {}
|
||||||
|
required: []
|
||||||
|
outputs_schema: { properties: {} }
|
||||||
@@ -0,0 +1,54 @@
|
|||||||
|
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||||
|
|
||||||
|
|
||||||
|
def _cfg():
|
||||||
|
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||||
|
|
||||||
|
|
||||||
|
def _inputs():
|
||||||
|
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||||
|
|
||||||
|
|
||||||
|
def _base(cfg):
|
||||||
|
return (str(cfg.get("url") or "https://cloud.tenable.com")).rstrip("/")
|
||||||
|
|
||||||
|
|
||||||
|
def request(method, path, cfg, body=None, params=None):
|
||||||
|
url = _base(cfg) + path
|
||||||
|
if params:
|
||||||
|
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||||
|
if clean:
|
||||||
|
url += "?" + urllib.parse.urlencode(clean)
|
||||||
|
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||||
|
headers = {
|
||||||
|
"X-ApiKeys": "accessKey=" + str(cfg.get("access_key", "")) + "; secretKey=" + str(cfg.get("secret_key", "")),
|
||||||
|
"Accept": "application/json",
|
||||||
|
}
|
||||||
|
if data is not None:
|
||||||
|
headers["Content-Type"] = "application/json"
|
||||||
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||||
|
with urllib.request.urlopen(req, timeout=90) as r:
|
||||||
|
raw = r.read()
|
||||||
|
return json.loads(raw) if raw else {}
|
||||||
|
|
||||||
|
|
||||||
|
def _run(fn):
|
||||||
|
try:
|
||||||
|
print(json.dumps(fn(_cfg(), _inputs())))
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||||
|
sys.exit(1)
|
||||||
|
except Exception as e:
|
||||||
|
print(json.dumps({"error": str(e)}))
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
def main(cfg, inputs):
|
||||||
|
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||||
|
asset_uuid = inputs.get("asset_uuid")
|
||||||
|
if not asset_uuid:
|
||||||
|
raise Exception("asset_uuid is required")
|
||||||
|
return request("GET", "/workbenches/assets/" + q(asset_uuid) + "/vulnerabilities", cfg)
|
||||||
|
|
||||||
|
|
||||||
|
_run(main)
|
||||||
@@ -0,0 +1,54 @@
|
|||||||
|
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||||
|
|
||||||
|
|
||||||
|
def _cfg():
|
||||||
|
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||||
|
|
||||||
|
|
||||||
|
def _inputs():
|
||||||
|
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||||
|
|
||||||
|
|
||||||
|
def _base(cfg):
|
||||||
|
return (str(cfg.get("url") or "https://cloud.tenable.com")).rstrip("/")
|
||||||
|
|
||||||
|
|
||||||
|
def request(method, path, cfg, body=None, params=None):
|
||||||
|
url = _base(cfg) + path
|
||||||
|
if params:
|
||||||
|
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||||
|
if clean:
|
||||||
|
url += "?" + urllib.parse.urlencode(clean)
|
||||||
|
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||||
|
headers = {
|
||||||
|
"X-ApiKeys": "accessKey=" + str(cfg.get("access_key", "")) + "; secretKey=" + str(cfg.get("secret_key", "")),
|
||||||
|
"Accept": "application/json",
|
||||||
|
}
|
||||||
|
if data is not None:
|
||||||
|
headers["Content-Type"] = "application/json"
|
||||||
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||||
|
with urllib.request.urlopen(req, timeout=90) as r:
|
||||||
|
raw = r.read()
|
||||||
|
return json.loads(raw) if raw else {}
|
||||||
|
|
||||||
|
|
||||||
|
def _run(fn):
|
||||||
|
try:
|
||||||
|
print(json.dumps(fn(_cfg(), _inputs())))
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||||
|
sys.exit(1)
|
||||||
|
except Exception as e:
|
||||||
|
print(json.dumps({"error": str(e)}))
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
def main(cfg, inputs):
|
||||||
|
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||||
|
asset_uuid = inputs.get("asset_uuid")
|
||||||
|
if not asset_uuid:
|
||||||
|
raise Exception("asset_uuid is required")
|
||||||
|
return request("GET", "/assets/" + q(asset_uuid), cfg)
|
||||||
|
|
||||||
|
|
||||||
|
_run(main)
|
||||||
@@ -0,0 +1,54 @@
|
|||||||
|
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||||
|
|
||||||
|
|
||||||
|
def _cfg():
|
||||||
|
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||||
|
|
||||||
|
|
||||||
|
def _inputs():
|
||||||
|
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||||
|
|
||||||
|
|
||||||
|
def _base(cfg):
|
||||||
|
return (str(cfg.get("url") or "https://cloud.tenable.com")).rstrip("/")
|
||||||
|
|
||||||
|
|
||||||
|
def request(method, path, cfg, body=None, params=None):
|
||||||
|
url = _base(cfg) + path
|
||||||
|
if params:
|
||||||
|
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||||
|
if clean:
|
||||||
|
url += "?" + urllib.parse.urlencode(clean)
|
||||||
|
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||||
|
headers = {
|
||||||
|
"X-ApiKeys": "accessKey=" + str(cfg.get("access_key", "")) + "; secretKey=" + str(cfg.get("secret_key", "")),
|
||||||
|
"Accept": "application/json",
|
||||||
|
}
|
||||||
|
if data is not None:
|
||||||
|
headers["Content-Type"] = "application/json"
|
||||||
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||||
|
with urllib.request.urlopen(req, timeout=90) as r:
|
||||||
|
raw = r.read()
|
||||||
|
return json.loads(raw) if raw else {}
|
||||||
|
|
||||||
|
|
||||||
|
def _run(fn):
|
||||||
|
try:
|
||||||
|
print(json.dumps(fn(_cfg(), _inputs())))
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||||
|
sys.exit(1)
|
||||||
|
except Exception as e:
|
||||||
|
print(json.dumps({"error": str(e)}))
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
def main(cfg, inputs):
|
||||||
|
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||||
|
scan_id = inputs.get("scan_id")
|
||||||
|
if not scan_id:
|
||||||
|
raise Exception("scan_id is required")
|
||||||
|
return request("GET", "/scans/" + q(scan_id) + "/latest-status", cfg)
|
||||||
|
|
||||||
|
|
||||||
|
_run(main)
|
||||||
@@ -0,0 +1,54 @@
|
|||||||
|
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||||
|
|
||||||
|
|
||||||
|
def _cfg():
|
||||||
|
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||||
|
|
||||||
|
|
||||||
|
def _inputs():
|
||||||
|
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||||
|
|
||||||
|
|
||||||
|
def _base(cfg):
|
||||||
|
return (str(cfg.get("url") or "https://cloud.tenable.com")).rstrip("/")
|
||||||
|
|
||||||
|
|
||||||
|
def request(method, path, cfg, body=None, params=None):
|
||||||
|
url = _base(cfg) + path
|
||||||
|
if params:
|
||||||
|
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||||
|
if clean:
|
||||||
|
url += "?" + urllib.parse.urlencode(clean)
|
||||||
|
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||||
|
headers = {
|
||||||
|
"X-ApiKeys": "accessKey=" + str(cfg.get("access_key", "")) + "; secretKey=" + str(cfg.get("secret_key", "")),
|
||||||
|
"Accept": "application/json",
|
||||||
|
}
|
||||||
|
if data is not None:
|
||||||
|
headers["Content-Type"] = "application/json"
|
||||||
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||||
|
with urllib.request.urlopen(req, timeout=90) as r:
|
||||||
|
raw = r.read()
|
||||||
|
return json.loads(raw) if raw else {}
|
||||||
|
|
||||||
|
|
||||||
|
def _run(fn):
|
||||||
|
try:
|
||||||
|
print(json.dumps(fn(_cfg(), _inputs())))
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||||
|
sys.exit(1)
|
||||||
|
except Exception as e:
|
||||||
|
print(json.dumps({"error": str(e)}))
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
def main(cfg, inputs):
|
||||||
|
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||||
|
plugin_id = inputs.get("plugin_id")
|
||||||
|
if not plugin_id:
|
||||||
|
raise Exception("plugin_id is required")
|
||||||
|
return request("GET", "/workbenches/vulnerabilities/" + q(plugin_id) + "/info", cfg)
|
||||||
|
|
||||||
|
|
||||||
|
_run(main)
|
||||||
@@ -0,0 +1,57 @@
|
|||||||
|
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||||
|
|
||||||
|
|
||||||
|
def _cfg():
|
||||||
|
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||||
|
|
||||||
|
|
||||||
|
def _inputs():
|
||||||
|
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||||
|
|
||||||
|
|
||||||
|
def _base(cfg):
|
||||||
|
return (str(cfg.get("url") or "https://cloud.tenable.com")).rstrip("/")
|
||||||
|
|
||||||
|
|
||||||
|
def request(method, path, cfg, body=None, params=None):
|
||||||
|
url = _base(cfg) + path
|
||||||
|
if params:
|
||||||
|
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||||
|
if clean:
|
||||||
|
url += "?" + urllib.parse.urlencode(clean)
|
||||||
|
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||||
|
headers = {
|
||||||
|
"X-ApiKeys": "accessKey=" + str(cfg.get("access_key", "")) + "; secretKey=" + str(cfg.get("secret_key", "")),
|
||||||
|
"Accept": "application/json",
|
||||||
|
}
|
||||||
|
if data is not None:
|
||||||
|
headers["Content-Type"] = "application/json"
|
||||||
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||||
|
with urllib.request.urlopen(req, timeout=90) as r:
|
||||||
|
raw = r.read()
|
||||||
|
return json.loads(raw) if raw else {}
|
||||||
|
|
||||||
|
|
||||||
|
def _run(fn):
|
||||||
|
try:
|
||||||
|
print(json.dumps(fn(_cfg(), _inputs())))
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||||
|
sys.exit(1)
|
||||||
|
except Exception as e:
|
||||||
|
print(json.dumps({"error": str(e)}))
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
def main(cfg, inputs):
|
||||||
|
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||||
|
scan_id = inputs.get("scan_id")
|
||||||
|
if not scan_id:
|
||||||
|
raise Exception("scan_id is required")
|
||||||
|
targets_raw = inputs.get("targets")
|
||||||
|
targets = [s.strip() for s in str(targets_raw).split(",") if s.strip()] if targets_raw else []
|
||||||
|
body = {"alt_targets": targets} if targets else {}
|
||||||
|
return request("POST", "/scans/" + q(scan_id) + "/launch", cfg, body=body)
|
||||||
|
|
||||||
|
|
||||||
|
_run(main)
|
||||||
@@ -0,0 +1,50 @@
|
|||||||
|
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 _base(cfg):
|
||||||
|
return (str(cfg.get("url") or "https://cloud.tenable.com")).rstrip("/")
|
||||||
|
|
||||||
|
|
||||||
|
def request(method, path, cfg, body=None, params=None):
|
||||||
|
url = _base(cfg) + path
|
||||||
|
if params:
|
||||||
|
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||||
|
if clean:
|
||||||
|
url += "?" + urllib.parse.urlencode(clean)
|
||||||
|
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||||
|
headers = {
|
||||||
|
"X-ApiKeys": "accessKey=" + str(cfg.get("access_key", "")) + "; secretKey=" + str(cfg.get("secret_key", "")),
|
||||||
|
"Accept": "application/json",
|
||||||
|
}
|
||||||
|
if data is not None:
|
||||||
|
headers["Content-Type"] = "application/json"
|
||||||
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||||
|
with urllib.request.urlopen(req, timeout=90) as r:
|
||||||
|
raw = r.read()
|
||||||
|
return json.loads(raw) if raw else {}
|
||||||
|
|
||||||
|
|
||||||
|
def _run(fn):
|
||||||
|
try:
|
||||||
|
print(json.dumps(fn(_cfg(), _inputs())))
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||||
|
sys.exit(1)
|
||||||
|
except Exception as e:
|
||||||
|
print(json.dumps({"error": str(e)}))
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
def main(cfg, inputs):
|
||||||
|
return request("GET", "/assets", cfg)
|
||||||
|
|
||||||
|
|
||||||
|
_run(main)
|
||||||
@@ -0,0 +1,50 @@
|
|||||||
|
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 _base(cfg):
|
||||||
|
return (str(cfg.get("url") or "https://cloud.tenable.com")).rstrip("/")
|
||||||
|
|
||||||
|
|
||||||
|
def request(method, path, cfg, body=None, params=None):
|
||||||
|
url = _base(cfg) + path
|
||||||
|
if params:
|
||||||
|
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||||
|
if clean:
|
||||||
|
url += "?" + urllib.parse.urlencode(clean)
|
||||||
|
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||||
|
headers = {
|
||||||
|
"X-ApiKeys": "accessKey=" + str(cfg.get("access_key", "")) + "; secretKey=" + str(cfg.get("secret_key", "")),
|
||||||
|
"Accept": "application/json",
|
||||||
|
}
|
||||||
|
if data is not None:
|
||||||
|
headers["Content-Type"] = "application/json"
|
||||||
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||||
|
with urllib.request.urlopen(req, timeout=90) as r:
|
||||||
|
raw = r.read()
|
||||||
|
return json.loads(raw) if raw else {}
|
||||||
|
|
||||||
|
|
||||||
|
def _run(fn):
|
||||||
|
try:
|
||||||
|
print(json.dumps(fn(_cfg(), _inputs())))
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||||
|
sys.exit(1)
|
||||||
|
except Exception as e:
|
||||||
|
print(json.dumps({"error": str(e)}))
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
def main(cfg, inputs):
|
||||||
|
return request("GET", "/scans", cfg)
|
||||||
|
|
||||||
|
|
||||||
|
_run(main)
|
||||||
@@ -0,0 +1,52 @@
|
|||||||
|
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 _base(cfg):
|
||||||
|
return (str(cfg.get("url") or "https://cloud.tenable.com")).rstrip("/")
|
||||||
|
|
||||||
|
|
||||||
|
def request(method, path, cfg, body=None, params=None):
|
||||||
|
url = _base(cfg) + path
|
||||||
|
if params:
|
||||||
|
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||||
|
if clean:
|
||||||
|
url += "?" + urllib.parse.urlencode(clean)
|
||||||
|
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||||
|
headers = {
|
||||||
|
"X-ApiKeys": "accessKey=" + str(cfg.get("access_key", "")) + "; secretKey=" + str(cfg.get("secret_key", "")),
|
||||||
|
"Accept": "application/json",
|
||||||
|
}
|
||||||
|
if data is not None:
|
||||||
|
headers["Content-Type"] = "application/json"
|
||||||
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||||
|
with urllib.request.urlopen(req, timeout=90) as r:
|
||||||
|
raw = r.read()
|
||||||
|
return json.loads(raw) if raw else {}
|
||||||
|
|
||||||
|
|
||||||
|
def _run(fn):
|
||||||
|
try:
|
||||||
|
print(json.dumps(fn(_cfg(), _inputs())))
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||||
|
sys.exit(1)
|
||||||
|
except Exception as e:
|
||||||
|
print(json.dumps({"error": str(e)}))
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
def main(cfg, inputs):
|
||||||
|
age = inputs.get("age")
|
||||||
|
age = int(age) if age not in (None, "") else 30
|
||||||
|
return request("GET", "/workbenches/vulnerabilities", cfg, params={"date_range": age})
|
||||||
|
|
||||||
|
|
||||||
|
_run(main)
|
||||||
@@ -0,0 +1,51 @@
|
|||||||
|
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||||
|
|
||||||
|
|
||||||
|
def _cfg():
|
||||||
|
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||||
|
|
||||||
|
|
||||||
|
def _inputs():
|
||||||
|
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||||
|
|
||||||
|
|
||||||
|
def _base(cfg):
|
||||||
|
return (str(cfg.get("url") or "https://cloud.tenable.com")).rstrip("/")
|
||||||
|
|
||||||
|
|
||||||
|
def request(method, path, cfg, body=None, params=None):
|
||||||
|
url = _base(cfg) + path
|
||||||
|
if params:
|
||||||
|
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||||
|
if clean:
|
||||||
|
url += "?" + urllib.parse.urlencode(clean)
|
||||||
|
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||||
|
headers = {
|
||||||
|
"X-ApiKeys": "accessKey=" + str(cfg.get("access_key", "")) + "; secretKey=" + str(cfg.get("secret_key", "")),
|
||||||
|
"Accept": "application/json",
|
||||||
|
}
|
||||||
|
if data is not None:
|
||||||
|
headers["Content-Type"] = "application/json"
|
||||||
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||||
|
with urllib.request.urlopen(req, timeout=90) as r:
|
||||||
|
raw = r.read()
|
||||||
|
return json.loads(raw) if raw else {}
|
||||||
|
|
||||||
|
|
||||||
|
def _run(fn):
|
||||||
|
try:
|
||||||
|
print(json.dumps(fn(_cfg(), _inputs())))
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||||
|
sys.exit(1)
|
||||||
|
except Exception as e:
|
||||||
|
print(json.dumps({"error": str(e)}))
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
|
def main(cfg, inputs):
|
||||||
|
request("GET", "/server/properties", cfg)
|
||||||
|
return {"ok": True}
|
||||||
|
|
||||||
|
|
||||||
|
_run(main)
|
||||||
Reference in New Issue
Block a user