diff --git a/integrations/qualys/manifest.yaml b/integrations/qualys/manifest.yaml new file mode 100644 index 0000000..4db5193 --- /dev/null +++ b/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: {} } diff --git a/integrations/qualys/scripts/host_detections.py b/integrations/qualys/scripts/host_detections.py new file mode 100644 index 0000000..71e6c2e --- /dev/null +++ b/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) diff --git a/integrations/qualys/scripts/host_list.py b/integrations/qualys/scripts/host_list.py new file mode 100644 index 0000000..80a06c7 --- /dev/null +++ b/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) diff --git a/integrations/qualys/scripts/launch_scan.py b/integrations/qualys/scripts/launch_scan.py new file mode 100644 index 0000000..745df0a --- /dev/null +++ b/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) diff --git a/integrations/qualys/scripts/list_scans.py b/integrations/qualys/scripts/list_scans.py new file mode 100644 index 0000000..f14b758 --- /dev/null +++ b/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) diff --git a/integrations/qualys/scripts/test_connection.py b/integrations/qualys/scripts/test_connection.py new file mode 100644 index 0000000..72cff90 --- /dev/null +++ b/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) diff --git a/integrations/qualys/scripts/vuln_details.py b/integrations/qualys/scripts/vuln_details.py new file mode 100644 index 0000000..5bec639 --- /dev/null +++ b/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)