feat(qualys): new Qualys VMDR vulnerability integration

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>

integrations/qualys/manifest.yaml

@@ -0,0 +1,88 @@
+id: qualys
+name: Qualys VMDR
+version: 1.0.0
+description: "Qualys Vulnerability Management (API v2) — asset and vulnerability context: list hosts, list host vulnerability detections, look up vulnerability (QID) knowledge-base details, and list/launch VM scans. HTTP Basic authentication; stdlib-only, no extra Python dependencies."
+changelog: "1.0.0 — Initial release: host list, host detections, knowledge-base vulnerability details, scan list, launch scan."
+category: vulnerability
+
+# Per-instance configuration. HTTP Basic auth against the Qualys API server
+# (regional, e.g. https://qualysapi.qg2.apps.qualys.com). All requests send the
+# required 'X-Requested-With' header. Responses are XML.
+config_schema:
+  properties:
+    base_url:
+      type: string
+      description: "Qualys API server URL (e.g. https://qualysapi.qualys.com)"
+    username:
+      type: string
+      description: "Qualys username"
+    password:
+      type: string
+      description: "Qualys password"
+      x-soar-sensitive: true
+  required:
+    - base_url
+    - username
+    - password
+
+commands:
+  - id: host_list
+    name: qualys-host-list
+    description: "List scanned hosts (optionally filtered by IP range)."
+    risk: read
+    inputs_schema:
+      properties:
+        ips: { type: string, description: "Optional IP/range filter (e.g. 10.0.0.1-10.0.0.50)" }
+        truncation_limit: { type: number, description: "Max hosts (default 100)" }
+      required: []
+    outputs_schema: { properties: {} }
+  - id: host_detections
+    name: qualys-host-detections
+    description: "List vulnerability detections on hosts (optionally filtered by IP and severity)."
+    risk: read
+    inputs_schema:
+      properties:
+        ips: { type: string, description: "Optional IP/range filter" }
+        severities: { type: string, description: "Optional severity filter (e.g. 4,5)" }
+        truncation_limit: { type: number, description: "Max hosts (default 100)" }
+      required: []
+    outputs_schema: { properties: {} }
+  - id: vuln_details
+    name: qualys-vuln-details
+    description: "Get knowledge-base details for one or more vulnerabilities by QID."
+    risk: read
+    inputs_schema:
+      properties:
+        qids: { type: string, description: "Comma-separated QIDs" }
+      required: [qids]
+    outputs_schema: { properties: {} }
+  - id: list_scans
+    name: qualys-list-scans
+    description: "List VM scans."
+    risk: read
+    inputs_schema:
+      properties:
+        state: { type: string, description: "Optional state filter (Running, Finished, Canceled, ...)" }
+      required: []
+    outputs_schema: { properties: {} }
+  - id: launch_scan
+    name: qualys-launch-scan
+    description: "Launch a VM scan against a set of IPs."
+    inputs_schema:
+      properties:
+        scan_title: { type: string, description: "Title for the scan" }
+        ip: { type: string, description: "Comma-separated target IPs/ranges" }
+        option_title: { type: string, description: "Option profile title (provide this or option_id)" }
+        option_id: { type: string, description: "Option profile ID (provide this or option_title)" }
+        iscanner_name: { type: string, description: "Scanner appliance name (optional)" }
+      required: [ip]
+    outputs_schema: { properties: {} }
+
+  - id: test_connection
+    name: qualys-test-connection
+    description: "Verify connectivity and credentials (used by the Test button)."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }

integrations/qualys/scripts/host_detections.py

@@ -0,0 +1,89 @@
+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)

integrations/qualys/scripts/host_list.py

@@ -0,0 +1,86 @@
+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")
+    truncation_limit = inputs.get("truncation_limit")
+    params = {
+        "action": "list",
+        "truncation_limit": int(truncation_limit or 100),
+    }
+    if ips:
+        params["ips"] = ips
+    return call("GET", "/api/2.0/fo/asset/host/", cfg, params)
+
+
+_run(main)

integrations/qualys/scripts/launch_scan.py

@@ -0,0 +1,102 @@
+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):
+    ip = inputs.get("ip")
+    if not ip:
+        raise Exception("ip is required")
+    scan_title = inputs.get("scan_title")
+    option_title = inputs.get("option_title")
+    option_id = inputs.get("option_id")
+    iscanner_name = inputs.get("iscanner_name")
+
+    if not option_title and not option_id:
+        raise Exception("option_title or option_id is required")
+
+    params = {
+        "action": "launch",
+        "ip": ip,
+    }
+    if scan_title:
+        params["scan_title"] = scan_title
+    if option_id:
+        params["option_id"] = option_id
+    elif option_title:
+        params["option_title"] = option_title
+    if iscanner_name:
+        params["iscanner_name"] = iscanner_name
+
+    return call("POST", "/api/2.0/fo/scan/", cfg, params)
+
+
+_run(main)

integrations/qualys/scripts/list_scans.py

@@ -0,0 +1,82 @@
+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):
+    state = inputs.get("state")
+    params = {"action": "list"}
+    if state:
+        params["state"] = state
+    return call("GET", "/api/2.0/fo/scan/", cfg, params)
+
+
+_run(main)

integrations/qualys/scripts/test_connection.py

@@ -0,0 +1,79 @@
+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):
+    call("GET", "/api/2.0/fo/scan/", cfg, {"action": "list"})
+    return {"ok": True}
+
+
+_run(main)

integrations/qualys/scripts/vuln_details.py

@@ -0,0 +1,85 @@
+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):
+    qids = inputs.get("qids")
+    if not qids:
+        raise Exception("qids is required")
+    params = {
+        "action": "list",
+        "ids": qids,
+    }
+    return call("GET", "/api/2.0/fo/knowledge_base/vuln/", cfg, params)
+
+
+_run(main)