cve_2026_58073_check.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Produced by Bishop Fox Team X and released for defensive use. Use only against
# systems you are authorized to test.

"""Safe, unauthenticated patch-state check for the Veeam Service Provider Console
KB4893 vulnerabilities (CVE-2026-58073 / 58072 / 58071 / 58067).

The ConnectionHub management-agent router reads a client handshake before any
authentication or TLS, and the fix widened the protocol versions Request.Read
accepts. So one Connector handshake advertising version 7 is a binary oracle: a
patched build parses it and returns an XML error for the unknown receiver, an
unpatched build rejects the version and returns nothing. A version-6 probe runs
first and must be answered, which is what stops a quiet TCP service from being
read as unpatched.

See README.md for methodology, verdicts, safety, and limitations.
"""

from __future__ import annotations

import argparse
import concurrent.futures
import io
import json
import os
import socket
import struct
import sys
import uuid
import xml.etree.ElementTree as ET
from dataclasses import asdict, dataclass, field

DEFAULT_HUB_PORT = 9999
DEFAULT_GATEWAY_PORT = 6180
DEFAULT_TIMEOUT = 8.0
DEFAULT_WORKERS = 16

# Hardcoded in Veeam.SPP.Utilities.Agent.CloudGate; identical on every deployment.
GATEWAY_SERVICE_GUID = "c6da2169-975a-4865-9e10-f4b453fb981b"

HOST_TYPE_CONNECTOR = 2  # Veeam.SPP.Communication.ChannelHostType

TRANSPORT_AUTO = "auto"
TRANSPORT_DIRECT = "direct"
TRANSPORT_GATEWAY = "gateway"

# 6 is accepted by every build in scope, so it fingerprints; 7 only by the fixed
# build, so it discriminates. Both are fixed by the code, not tunable. See README.
PROTOCOL_FINGERPRINT = 6
PROTOCOL_DISCRIMINATOR = 7

# Makes this tool's traffic attributable in ConnectionHub.log. See README.
RECEIVER_PREFIX = "bf-probe-"

VERDICT_VULNERABLE = "VULNERABLE"
VERDICT_PATCHED = "PATCHED"
VERDICT_UNAFFECTED = "UNAFFECTED"
VERDICT_INCONCLUSIVE = "INCONCLUSIVE"
VERDICT_ERROR = "ERROR"

KB4893_CVES = ["CVE-2026-58073", "CVE-2026-58072", "CVE-2026-58071", "CVE-2026-58067"]


# --- output ------------------------------------------------------------------
class Ansi:
    RESET = "\033[0m"
    BOLD = "\033[1m"
    DIM = "\033[0;90m"
    RED = "\033[1;31m"
    GREEN = "\033[1;32m"
    YELLOW = "\033[1;33m"
    MAGENTA = "\033[1;35m"


VERDICT_STYLE = {
    VERDICT_VULNERABLE: ("[!]", Ansi.RED),
    VERDICT_PATCHED: ("[+]", Ansi.GREEN),
    VERDICT_UNAFFECTED: ("[-]", Ansi.DIM),
    VERDICT_INCONCLUSIVE: ("[?]", Ansi.YELLOW),
    VERDICT_ERROR: ("[x]", Ansi.MAGENTA),
}


def want_color(flag_no_color: bool) -> bool:
    """Honour --no-color, the NO_COLOR convention, and non-TTY output."""
    if flag_no_color or os.environ.get("NO_COLOR"):
        return False
    return sys.stdout.isatty()


def paint(text: str, style: str, enabled: bool) -> str:
    return f"{style}{text}{Ansi.RESET}" if enabled else text


# --- .NET BinaryWriter / BinaryReader codecs ---------------------------------
def write_7bit_encoded_int(value: int) -> bytes:
    """.NET BinaryWriter.Write7BitEncodedInt — LEB128 over a non-negative int."""
    if value < 0:
        raise ValueError("length must be non-negative")
    out = bytearray()
    v = value
    while v >= 0x80:
        out.append((v & 0x7F) | 0x80)
        v >>= 7
    out.append(v)
    return bytes(out)


def read_7bit_encoded_int(buf: io.BytesIO) -> int:
    """.NET BinaryReader.Read7BitEncodedInt. Raises on the 5-byte overflow .NET rejects."""
    count = 0
    shift = 0
    while True:
        if shift == 5 * 7:
            raise ValueError("malformed 7-bit encoded int")
        byte = buf.read(1)
        if not byte:
            raise EOFError("truncated 7-bit encoded int")
        b = byte[0]
        count |= (b & 0x7F) << shift
        shift += 7
        if not (b & 0x80):
            return count


def write_dotnet_string(s: str) -> bytes:
    """.NET BinaryWriter.Write(string) with a UTF-8 encoder."""
    raw = s.encode("utf-8")
    return write_7bit_encoded_int(len(raw)) + raw


def read_dotnet_string(buf: io.BytesIO) -> str:
    """.NET BinaryReader.ReadString() with a UTF-8 encoder."""
    length = read_7bit_encoded_int(buf)
    raw = buf.read(length)
    if len(raw) != length:
        raise EOFError(f"truncated string: wanted {length} bytes, got {len(raw)}")
    return raw.decode("utf-8")


# --- ConnectionHub handshake -------------------------------------------------
def build_connector_request(version_byte: int, receiver_name: str) -> bytes:
    """A Connector handshake for Request.Read: int16 meta, then a .NET string of XML.

    Supplying only <connectTo/> leaves ConnectorConfiguration at ConnectionType
    .Normal — deliberately not Multiplexed, whose branch would build a channel.
    """
    meta = (HOST_TYPE_CONNECTOR & 0xFF) | ((version_byte & 0xFF) << 8)
    xml = f'<Connector><connectTo receiver="{receiver_name}" /></Connector>'
    return struct.pack("<h", meta) + write_dotnet_string(xml)


@dataclass
class HubResponse:
    status: str
    attributes: dict = field(default_factory=dict)
    raw_xml: str = ""

    @property
    def message(self) -> str:
        return self.attributes.get("message", "")

    @property
    def protocol_version(self):
        v = self.attributes.get("protocolVersion")
        return int(v) if v is not None and v.isdigit() else None


def parse_hub_response(data: bytes) -> HubResponse:
    """Response.WriteTo: int32 meta, then a .NET string of <Response status=…> XML."""
    buf = io.BytesIO(data)
    head = buf.read(4)
    if len(head) != 4:
        raise EOFError("truncated 4-byte response meta")
    xml = read_dotnet_string(buf)
    root = ET.fromstring(xml)
    attrs = {}
    for attr in root.iterfind("./attributes/attr"):
        name = attr.get("name")
        if name is not None:
            attrs[name] = attr.get("value", "")
    return HubResponse(status=root.get("status", ""), attributes=attrs, raw_xml=xml)


def gateway_prologue() -> bytes:
    """The relay-selection frame from VacConnector.Behaviour.GatewaySetup.

    The gateway relays every later byte untouched, so the handshake that follows
    is byte-identical to the direct path.
    """
    guid_ascii = str(uuid.UUID(GATEWAY_SERVICE_GUID)).encode("ascii")
    payload = struct.pack("<i", 0) + struct.pack("<i", len(guid_ascii)) + guid_ascii
    return struct.pack("<i", len(payload)) + payload


def recv_exactly(sock: socket.socket, count: int):
    """Read exactly `count` bytes, or None if the peer stops short.

    A reset counts as stopping short, not as an error to propagate: a non-gateway
    rejects the relay prologue with RST, and letting that escape would abort the
    whole scan on one bad target.
    """
    out = bytearray()
    while len(out) < count:
        try:
            chunk = sock.recv(count - len(out))
        except (socket.timeout, TimeoutError):
            return None
        except OSError:
            return None
        if not chunk:
            return None
        out.extend(chunk)
    return bytes(out)


def read_gateway_reply(sock: socket.socket):
    """(ok, error). 32-byte reply; bit 0 of the leading uint32 flags an error."""
    reply = recv_exactly(sock, 32)
    if reply is None:
        return False, "gateway closed the connection without replying"
    flags = struct.unpack("<I", reply[:4])[0]
    if flags & 1:
        length_bytes = recv_exactly(sock, 4)
        if length_bytes is None:
            return False, f"gateway error, flags=0x{flags:08x} (no message)"
        msg = recv_exactly(sock, struct.unpack("<i", length_bytes)[0]) or b""
        return False, f"gateway error: {msg.decode('utf-8', 'replace')}"
    return True, ""


def recv_available(sock: socket.socket, limit: int = 65536) -> bytes:
    """Read until the peer closes or stops. Zero bytes is the signal we care about.

    A reset is a normal termination here: a rejecting server disposes the socket
    with our handshake still unread, which emits RST. Treating it as an error
    would make every vulnerable host report INCONCLUSIVE.
    """
    out = bytearray()
    while len(out) < limit:
        try:
            chunk = sock.recv(4096)
        except (socket.timeout, TimeoutError):
            break
        except (ConnectionResetError, ConnectionAbortedError):
            break
        except OSError:
            break
        if not chunk:
            break
        out.extend(chunk)
        # A complete response is small; stop as soon as it parses.
        try:
            parse_hub_response(bytes(out))
            break
        except Exception:
            continue
    return bytes(out)


# --- probing -----------------------------------------------------------------
@dataclass
class Probe:
    version_byte: int
    transport: str
    # Apart from `error` so auto-detection can tell "nothing is listening" (no
    # transport helps) from "answered, but not as expected".
    connected: bool
    responded: bool
    status: str = ""
    message: str = ""
    error: str = ""


@dataclass
class Result:
    target: str
    host: str
    port: int
    transport: str
    verdict: str
    reason: str
    detail: str
    protocol_version: object = None
    affected_cves: list = field(default_factory=list)
    probes: list = field(default_factory=list)


def probe_once(host: str, port: int, version_byte: int, transport: str,
               timeout: float) -> Probe:
    """Send one handshake over one transport and classify the reply.

    Failing to *reach* the target is an error, never a verdict; connecting and
    answering nothing is a protocol outcome, and for version 7 it is the
    vulnerable signal. The two are kept apart.
    """
    receiver = f"{RECEIVER_PREFIX}{uuid.uuid4()}"

    def probe(connected, responded, **kw):
        return Probe(version_byte, transport, connected, responded, **kw)

    try:
        sock = socket.create_connection((host, port), timeout=timeout)
    except OSError as exc:
        return probe(False, False, error=f"{type(exc).__name__}: {exc}")

    try:
        sock.settimeout(timeout)
        if transport == TRANSPORT_GATEWAY:
            try:
                sock.sendall(gateway_prologue())
            except OSError as exc:
                return probe(True, False,
                             error=f"gateway prologue send failed: {exc}")
            ok, err = read_gateway_reply(sock)
            if not ok:
                return probe(True, False, error=err)
        try:
            sock.sendall(build_connector_request(version_byte, receiver))
        except (ConnectionResetError, BrokenPipeError, ConnectionAbortedError):
            # Server tore the connection down mid-handshake. Same meaning as
            # answering nothing.
            return probe(True, False)
        except OSError as exc:
            return probe(True, False, error=f"send failed: {exc}")
        data = recv_available(sock)
    finally:
        try:
            sock.close()
        except OSError:
            pass

    if not data:
        # Server disposed the socket without writing. For version 7 this is the
        # vulnerable signal.
        return probe(True, False)
    try:
        resp = parse_hub_response(data)
    except Exception as exc:
        return probe(True, True,
                     error=f"unparseable response ({exc}): {data[:64]!r}")
    return probe(True, True, status=resp.status, message=resp.message)


def transport_order(port: int, pinned: str) -> list:
    """Transports to try for the fingerprint probe, in order.

    Leading with the service that owns the port is deliberate: a handshake sent to
    a gateway is read as a ~1 GB frame length and burns the whole timeout, while
    the reverse mismatch is rejected in one round trip. See README.
    """
    if pinned != TRANSPORT_AUTO:
        return [pinned]
    if port == DEFAULT_GATEWAY_PORT:
        return [TRANSPORT_GATEWAY, TRANSPORT_DIRECT]
    return [TRANSPORT_DIRECT, TRANSPORT_GATEWAY]


def passes_gate(p: Probe) -> bool:
    """True when a fingerprint probe proves a VSPC ConnectionHub is on the far end."""
    return (p.responded and p.status == "Error"
            and "receiver not found" in p.message.lower())


def assess(host: str, port: int, pinned_transport: str, timeout: float) -> Result:
    def result(verdict, reason, detail, transport, **kw):
        label = f"{host}:{port}"
        if transport == TRANSPORT_GATEWAY:
            label += " (gateway)"
        return Result(label, host, port, transport, verdict, reason, detail, **kw)

    # Step 1 — fingerprint, and under `auto` settle the transport too. Passing the
    # gate is what makes silence in step 2 mean something. See README.
    probes = []
    used = None
    for transport in transport_order(port, pinned_transport):
        fp = probe_once(host, port, PROTOCOL_FINGERPRINT, transport, timeout)
        probes.append(fp)
        if passes_gate(fp):
            used = transport
            break
        if not fp.connected:
            break  # TCP never opened; no transport helps, and a sweep saves a timeout

    if used is None:
        # The primary attempt wins unless something replied: a fallback's refused
        # prologue would turn every quiet TCP service into ERROR, not UNAFFECTED.
        witness = next((p for p in probes if p.responded), probes[0])
        if witness.responded:
            detail = witness.error or (f"status={witness.status!r}, "
                                       f"message={witness.message!r}")
            return result(VERDICT_INCONCLUSIVE, "unexpected-reply",
                          f"unexpected handshake reply over the "
                          f"{witness.transport} transport ({detail})",
                          witness.transport, probes=probes)
        if witness.error:
            return result(VERDICT_ERROR, "unreachable", witness.error,
                          witness.transport, probes=probes)
        tried = " or ".join(p.transport for p in probes)
        return result(VERDICT_UNAFFECTED, "not-vspc",
                      f"no response to a valid ConnectionHub handshake over the "
                      f"{tried} transport, so this is not a VSPC ConnectionHub",
                      witness.transport, probes=probes)

    # Step 2 — discriminate over the *same* transport, never falling back, or
    # silence stops being attributable to the version check. See README.
    disc = probe_once(host, port, PROTOCOL_DISCRIMINATOR, used, timeout)
    probes.append(disc)

    if disc.responded and disc.status == "Error":
        return result(VERDICT_PATCHED, "protocol-7-accepted",
                      "ConnectionHub accepts protocol 7, so the KB4893 fixes are "
                      "present (>= 9.3.0.35057)",
                      used, protocol_version=7, probes=probes)
    if not disc.responded and not disc.error:
        return result(VERDICT_VULNERABLE, "protocol-7-rejected",
                      "ConnectionHub rejects protocol 7 but accepts 6, so the "
                      "KB4893 fixes are absent (<= 9.2.1.33875)",
                      used, protocol_version=6, affected_cves=list(KB4893_CVES),
                      probes=probes)
    return result(VERDICT_INCONCLUSIVE, "inconclusive-discriminator",
                  disc.error or f"inconclusive discriminator reply "
                                f"(status={disc.status!r})",
                  used, probes=probes)


# --- self-test — validates the codecs without touching the network -----------
def self_test() -> int:
    failures = []

    def check(name, got, want):
        if got != want:
            failures.append(f"{name}: got {got!r}, want {want!r}")

    # 7-bit encoded int, against known .NET output
    check("7bit(0)", write_7bit_encoded_int(0), b"\x00")
    check("7bit(127)", write_7bit_encoded_int(127), b"\x7f")
    check("7bit(128)", write_7bit_encoded_int(128), b"\x80\x01")
    check("7bit(300)", write_7bit_encoded_int(300), b"\xac\x02")
    check("7bit(16384)", write_7bit_encoded_int(16384), b"\x80\x80\x01")
    for n in (0, 1, 127, 128, 300, 16383, 16384, 2097151, 2097152):
        check(f"7bit roundtrip({n})",
              read_7bit_encoded_int(io.BytesIO(write_7bit_encoded_int(n))), n)

    # string codec
    for s in ("", "a", "<Connector/>", "ünïcødé " * 40):
        check(f"string roundtrip({s[:12]!r})",
              read_dotnet_string(io.BytesIO(write_dotnet_string(s))), s)

    # request framing
    req = build_connector_request(7, "bf-probe-x")
    check("request meta", struct.unpack("<h", req[:2])[0], 2 | (7 << 8))
    check("request type byte", req[0], HOST_TYPE_CONNECTOR)
    check("request version byte", req[1], 7)
    check("request xml", read_dotnet_string(io.BytesIO(req[2:])),
          '<Connector><connectTo receiver="bf-probe-x" /></Connector>')

    # response parser, against the exact XML shape HostResponse serializes to
    xml = ('<Response status="Error"><attributes>'
           '<attr name="status" value="1" />'
           '<attr name="message" value="Cannot connect transmitter. '
           'Requested receiver not found (receiver name:bf-probe-x)" />'
           '</attributes></Response>')
    wire = struct.pack("<i", 0) + write_dotnet_string(xml)
    parsed = parse_hub_response(wire)
    check("response status", parsed.status, "Error")
    check("response message contains", "receiver not found" in parsed.message.lower(), True)

    ok_xml = ('<Response status="Ok"><attributes>'
              '<attr name="channelId" value="00000000-0000-0000-0000-000000000000" />'
              '<attr name="protocolVersion" value="7" />'
              '<attr name="pierName" value="" />'
              '</attributes></Response>')
    parsed_ok = parse_hub_response(struct.pack("<i", 0) + write_dotnet_string(ok_xml))
    check("ok status", parsed_ok.status, "Ok")
    check("ok protocolVersion", parsed_ok.protocol_version, 7)

    # transport ordering
    check("order auto on hub port", transport_order(DEFAULT_HUB_PORT, TRANSPORT_AUTO),
          [TRANSPORT_DIRECT, TRANSPORT_GATEWAY])
    check("order auto on gateway port",
          transport_order(DEFAULT_GATEWAY_PORT, TRANSPORT_AUTO),
          [TRANSPORT_GATEWAY, TRANSPORT_DIRECT])
    check("order auto on odd port", transport_order(1234, TRANSPORT_AUTO),
          [TRANSPORT_DIRECT, TRANSPORT_GATEWAY])
    check("order pinned direct ignores port",
          transport_order(DEFAULT_GATEWAY_PORT, TRANSPORT_DIRECT), [TRANSPORT_DIRECT])
    check("order pinned gateway ignores port",
          transport_order(DEFAULT_HUB_PORT, TRANSPORT_GATEWAY), [TRANSPORT_GATEWAY])

    # fingerprint gate — the sole guard against calling a quiet service VULNERABLE
    def fp_probe(**kw):
        return Probe(PROTOCOL_FINGERPRINT, TRANSPORT_DIRECT, True, **kw)

    check("gate passes on receiver-not-found",
          passes_gate(fp_probe(responded=True, status="Error",
                               message="Cannot connect transmitter. Requested "
                                       "receiver not found (receiver name:x)")), True)
    check("gate rejects silence", passes_gate(fp_probe(responded=False)), False)
    check("gate rejects ok status",
          passes_gate(fp_probe(responded=True, status="Ok")), False)
    check("gate rejects other error",
          passes_gate(fp_probe(responded=True, status="Error",
                               message="Unsupported client version")), False)

    # gateway prologue
    pro = gateway_prologue()
    check("prologue total length", len(pro), 4 + 4 + 4 + 36)
    check("prologue declared length", struct.unpack("<i", pro[:4])[0], 44)
    check("prologue reserved", struct.unpack("<i", pro[4:8])[0], 0)
    check("prologue guid length", struct.unpack("<i", pro[8:12])[0], 36)
    check("prologue guid", pro[12:].decode("ascii"), GATEWAY_SERVICE_GUID)

    if failures:
        print("SELF-TEST FAILED", file=sys.stderr)
        for f in failures:
            print(f"  - {f}", file=sys.stderr)
        return 1
    print("self-test OK — codecs match the .NET wire format")
    return 0


def print_human(r: Result, color: bool) -> None:
    marker, style = VERDICT_STYLE.get(r.verdict, ("[?]", Ansi.YELLOW))
    print(f"{paint(marker, style, color)} {paint(r.target, Ansi.BOLD, color)}: "
          f"{paint(r.verdict, style, color)}  [{r.reason}]")
    print(f"      {paint(r.detail, Ansi.DIM, color)}")


def print_brief(r: Result, color: bool) -> None:
    """One aligned line per target — convenient for scanning many hosts."""
    _, style = VERDICT_STYLE.get(r.verdict, ("[?]", Ansi.YELLOW))
    status = paint(f"{r.verdict:<13}", style, color)
    print(f"{status} {r.target:<32} {r.reason}")


def parse_target(raw: str, default_port: int):
    raw = raw.strip()
    if raw.startswith("["):  # bracketed IPv6
        host, _, rest = raw[1:].partition("]")
        rest = rest.lstrip(":")
        return host, int(rest) if rest else default_port
    if raw.count(":") == 1:
        host, _, port = raw.partition(":")
        return host, int(port)
    return raw, default_port


def build_parser():
    p = argparse.ArgumentParser(
        prog="cve_2026_58073_check.py",
        description="Safe detector for Veeam Service Provider Console KB4893 "
                    "(CVE-2026-58073/58072/58071/58067). Reports whether the "
                    "KB4893 fixes are present, without authenticating.",
        epilog="examples:\n"
               "  cve_2026_58073_check.py vspc.example.com\n"
               "  cve_2026_58073_check.py vspc.example.com:9999 10.0.0.5\n"
               "  cve_2026_58073_check.py -f targets.txt --brief\n"
               "  cve_2026_58073_check.py -f targets.txt --json > results.json\n"
               "  cve_2026_58073_check.py cc-gw.example.com:6180\n"
               "  cve_2026_58073_check.py --transport gateway cc-gw.example.com\n\n"
               "Only scan systems you are authorized to test.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    p.add_argument("targets", nargs="*", help="host or host:port")
    p.add_argument("-f", "--targets-file", metavar="FILE",
                   help="file with one target per line ('#' comments allowed)")
    p.add_argument("--transport", default=TRANSPORT_AUTO,
                   choices=[TRANSPORT_AUTO, TRANSPORT_DIRECT, TRANSPORT_GATEWAY],
                   help="how to reach the ConnectionHub: 'direct' speaks to it "
                        "straight, 'gateway' prepends the Veeam Cloud Connect "
                        "relay prologue, 'auto' (default) detects which one works "
                        "per target")
    p.add_argument("-p", "--port", type=int, metavar="PORT",
                   help=f"override the default port ({DEFAULT_HUB_PORT}, or "
                        f"{DEFAULT_GATEWAY_PORT} with --transport gateway)")
    p.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT, metavar="SECS",
                   help=f"per-probe timeout in seconds (default: {DEFAULT_TIMEOUT:g})")
    p.add_argument("--workers", type=int, default=DEFAULT_WORKERS, metavar="N",
                   help=f"concurrent targets (default: {DEFAULT_WORKERS})")
    p.add_argument("-b", "--brief", action="store_true",
                   help="single aligned line per target (for scanning many hosts)")
    p.add_argument("--json", action="store_true",
                   help="emit JSON results, including every probe sent per target")
    p.add_argument("--no-color", action="store_true",
                   help="disable coloured output")
    p.add_argument("--self-test", action="store_true",
                   help="validate the wire codecs and exit; no network access")
    return p


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()

    if args.self_test:
        return self_test()

    raw_targets = list(args.targets)
    if args.targets_file:
        try:
            with open(args.targets_file) as fh:
                raw_targets += [ln.split("#")[0].strip() for ln in fh
                                if ln.split("#")[0].strip()]
        except OSError as exc:
            print(f"error: cannot read targets file: {exc}", file=sys.stderr)
            return 2
    if not raw_targets:
        parser.error("no targets given (positional or --targets-file)")

    default_port = args.port or (DEFAULT_GATEWAY_PORT
                                 if args.transport == TRANSPORT_GATEWAY
                                 else DEFAULT_HUB_PORT)
    try:
        targets = [parse_target(t, default_port) for t in raw_targets]
    except ValueError as exc:
        print(f"error: bad target ({exc})", file=sys.stderr)
        return 2

    color = want_color(args.no_color)
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max(1, args.workers)) as pool:
        # map() yields in input order, so output is deterministic while probes
        # still run concurrently.
        stream = pool.map(
            lambda hp: assess(hp[0], hp[1], args.transport, args.timeout), targets)
        for r in stream:
            results.append(r)
            if args.json:
                continue
            print_brief(r, color) if args.brief else print_human(r, color)

    if args.json:
        print(json.dumps([asdict(r) for r in results], indent=2))

    return 1 if any(r.verdict == VERDICT_VULNERABLE for r in results) else 0


if __name__ == "__main__":
    sys.exit(main())