{"uuid": "0743b31b-dbdf-4c9c-b7c4-6d173e2d83fe", "vulnerability_lookup_origin": "1a89b78e-f703-45f3-bb86-59eb712668bd", "author": "9f56dd64-161d-43a6-b9c3-555944290a09", "vulnerability": "CVE-2026-87902", "type": "seen", "source": "https://gist.github.com/giaoimgiao/1d0ed1adba5b9ca37dc4f48ba7fb5276", "content": "#!/usr/bin/env python3\n# -*- coding: utf-8 -*-\n\"\"\"Rolling CVE-2026-87902 hunter over a masscan host list.\n\nDesign constraints learned the hard way:\n  * 8.8 million hosts cannot be probed naively. WordPress is a small fraction\n    of open 80/443, so stage 1 is a cheap fingerprint and stage 2 only runs\n    against hosts that actually look like WordPress.\n  * Resume is mandatory: a multi-hour job will be interrupted. Tested hosts\n    and hits are appended, never rewritten.\n  * Output is line-delimited JSON so a later report can be rebuilt from disk\n    without re-probing.\n\"\"\"\n\nimport argparse\nimport json\nimport os\nimport sys\nimport time\n\nimport wp87902\n\n\ndef load_done(path):\n    done = set()\n    if not os.path.exists(path):\n        return done\n    with open(path, \"r\", errors=\"replace\") as fh:\n        for line in fh:\n            line = line.strip()\n            if line:\n                done.add(line.split()[0])\n    return done\n\n\ndef interesting(result):\n    \"\"\"Keep only verdicts worth a human look.\"\"\"\n    return result.get(\"verdict\") in (\n        \"VULN_HIGH\", \"VULN_MAYBE\", \"VER_ONLY\", \"LFI_EVIDENCE\", \"WAF_BLOCKED\",\n    )\n\n\ndef main():\n    ap = argparse.ArgumentParser()\n    ap.add_argument(\"--hosts\", default=\"/root/wp_hosts.txt\")\n    ap.add_argument(\"--tested\", default=\"/root/wp_tested.txt\")\n    ap.add_argument(\"--hits\", default=\"/root/wp_hits.jsonl\")\n    ap.add_argument(\"-w\", \"--workers\", type=int, default=80)\n    ap.add_argument(\"-t\", \"--timeout\", type=float, default=6.0)\n    ap.add_argument(\"--batch\", type=int, default=2000)\n    ap.add_argument(\"--verify\", action=\"store_true\",\n                    help=\"active LFI probe on version-range hosts only\")\n    ap.add_argument(\"--max\", type=int, default=0, help=\"stop after N new hosts (0=all)\")\n    args = ap.parse_args()\n\n    done = load_done(args.tested)\n    pending = []\n    with open(args.hosts, \"r\", errors=\"replace\") as fh:\n        for line in fh:\n            host = line.strip()\n            if host and host not in done:\n                pending.append(host)\n            if args.max and len(pending) >= args.max:\n                break\n    print(\"[hunter] hosts=%s already_tested=%d pending=%d verify=%s\"\n          % (args.hosts, len(done), len(pending), args.verify), flush=True)\n    if not pending:\n        print(\"[hunter] nothing pending\", flush=True)\n        return 0\n\n    hits_fh = open(args.hits, \"a\")\n    tested_fh = open(args.tested, \"a\")\n    t0 = time.time()\n    n = 0\n    hit_n = 0\n\n    def consume(result):\n        nonlocal n, hit_n\n        n += 1\n        tested_fh.write(result[\"target\"] + \"\\n\")\n        if interesting(result):\n            hit_n += 1\n            hits_fh.write(json.dumps(result, ensure_ascii=False) + \"\\n\")\n            hits_fh.flush()\n        if n % 200 == 0:\n            tested_fh.flush()\n            rate = n / max(1.0, time.time() - t0)\n            print(\"[hunter] %d/%d hits=%d %.1f/s last=%s %s\"\n                  % (n, len(pending), hit_n, rate, result[\"target\"], result[\"verdict\"]),\n                  flush=True)\n\n    import concurrent.futures as futures\n    with futures.ThreadPoolExecutor(max_workers=args.workers) as ex:\n        # submit in batches so a crash loses at most one batch of in-flight work\n        for i in range(0, len(pending), args.batch):\n            chunk = pending[i:i + args.batch]\n            futs = [ex.submit(wp87902.check, h, verify=args.verify, timeout=args.timeout)\n                   for h in chunk]\n            for fut in futures.as_completed(futs):\n                try:\n                    consume(fut.result())\n                except Exception as exc:\n                    print(\"[hunter] worker error: %s\" % exc, flush=True)\n    tested_fh.flush()\n    hits_fh.close()\n    tested_fh.close()\n    print(\"[hunter] done tested=%d hits=%d elapsed=%.0fs\" % (n, hit_n, time.time() - t0),\n          flush=True)\n    return 0\n\n\nif __name__ == \"__main__\":\n    sys.exit(main())", "creation_timestamp": "2026-09-28T07:59:52.000000Z"}