Guillaume BOURGEOIS
fcf30da5fc
Qualys API v2 (XML), 6 commands: host list, host detections, knowledge-base vulnerability details, scan list, launch scan. HTTP Basic auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
90 lines
2.5 KiB
Python
90 lines
2.5 KiB
Python
import json, os, sys, base64, urllib.parse, urllib.request, urllib.error
|
|
import xml.etree.ElementTree as ET
|
|
|
|
|
|
def _cfg():
|
|
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
|
|
|
|
|
def _inputs():
|
|
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
|
|
|
|
|
def _strip_ns(tag):
|
|
return tag.split("}", 1)[1] if "}" in tag else tag
|
|
|
|
|
|
def _xml_to_dict(elem):
|
|
children = list(elem)
|
|
if not children:
|
|
return (elem.text or "").strip()
|
|
d = {}
|
|
for c in children:
|
|
tag = _strip_ns(c.tag)
|
|
val = _xml_to_dict(c)
|
|
if tag in d:
|
|
if not isinstance(d[tag], list):
|
|
d[tag] = [d[tag]]
|
|
d[tag].append(val)
|
|
else:
|
|
d[tag] = val
|
|
return d
|
|
|
|
|
|
def _auth(cfg):
|
|
raw = str(cfg.get("username", "")) + ":" + str(cfg.get("password", ""))
|
|
return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8")
|
|
|
|
|
|
def call(method, path, cfg, params=None):
|
|
base = str(cfg.get("base_url", "")).rstrip("/")
|
|
url = base + path
|
|
params = {k: v for k, v in (params or {}).items() if v not in (None, "")}
|
|
data = None
|
|
if method == "GET":
|
|
if params:
|
|
url += "?" + urllib.parse.urlencode(params)
|
|
else:
|
|
data = urllib.parse.urlencode(params).encode("utf-8")
|
|
headers = {
|
|
"Authorization": _auth(cfg),
|
|
"X-Requested-With": "Riposte",
|
|
"Accept": "application/xml",
|
|
}
|
|
if data is not None:
|
|
headers["Content-Type"] = "application/x-www-form-urlencoded"
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
with urllib.request.urlopen(req, timeout=120) as r:
|
|
raw = r.read()
|
|
root = ET.fromstring(raw)
|
|
return {_strip_ns(root.tag): _xml_to_dict(root)}
|
|
|
|
|
|
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):
|
|
ips = inputs.get("ips")
|
|
severities = inputs.get("severities")
|
|
truncation_limit = inputs.get("truncation_limit")
|
|
params = {
|
|
"action": "list",
|
|
"truncation_limit": int(truncation_limit or 100),
|
|
}
|
|
if ips:
|
|
params["ips"] = ips
|
|
if severities:
|
|
params["severities"] = severities
|
|
return call("GET", "/api/2.0/fo/asset/host/vm/detection/", cfg, params)
|
|
|
|
|
|
_run(main)
|