Method handle_batch

bbot/modules/httpx.py:105–224

(self, *events)

        return url_hash

    async def handle_batch(self, *events):
        stdin = {}

        for event in events:
            url, url_hash = self.make_url_metadata(event)
            stdin[url] = event

        if not stdin:
            return

        command = [
            "httpx",
            "-silent",
            "-json",
            "-include-response",
            "-threads",
            self.threads,
            "-timeout",
            self.scan.httpx_timeout,
            "-retries",
            self.scan.httpx_retries,
            "-header",
            f"User-Agent: {self.scan.useragent}",
            "-response-size-to-read",
            f"{self.max_response_size}",
        ]

        if self.store_responses:
            response_dir = self.scan.home / "httpx"
            self.helpers.mkdir(response_dir)
            command += ["-srd", str(response_dir)]

        dns_resolvers = ",".join(self.helpers.system_resolvers)
        if dns_resolvers:
            command += ["-r", dns_resolvers]

        if self.probe_all_ips:
            command += ["-probe-all-ips"]

        # Add custom HTTP headers
        for hk, hv in self.scan.custom_http_headers.items():
            command += ["-header", f"{hk}: {hv}"]

        # Add custom HTTP cookies as a single header
        if self.scan.custom_http_cookies:
            cookie = SimpleCookie()
            for ck, cv in self.scan.custom_http_cookies.items():
                cookie[ck] = cv

            # Build the cookie header
            cookie_header = f"Cookie: {cookie.output(header='', sep='; ').strip()}"
            command += ["-header", cookie_header]

        proxy = self.scan.http_proxy
        if proxy:
            command += ["-http-proxy", proxy]
        async for line in self.run_process_live(command, text=False, input=list(stdin), stderr=subprocess.DEVNULL):
            try: