{"owner":"blacklanternsecurity","repo":"bbot","hasSkills":true,"hasMcp":false,"mcpConfig":null,"found":["AGENTS.md","CLAUDE.md"],"skills":{"AGENTS.md":"# BBOT Developer Guide\n\n## Core Principles\n\n### Modularity Principle\nWhen writing a BBOT module, make sure all module-specific code lives in the module itself. Don't hard-code module-specific things in core or in helpers.\n\n### DRY Principle\nDon't Repeat Yourself -- and interpret this broadly. If two pieces of code aren't identical but follow a similar enough pattern that they could be generalized, they should be. Extract shared logic into a common abstraction rather than duplicating the pattern. Usually this means creating a shared helper, or a shared module template in `bbot/modules/templates`. When you notice structural similarity, unify it.\n\n### Engineering Principle\nEvery system that is implemented must be implemented properly. No hacks, no hardcoding, no shortcuts. If we implement one of something, we build a proper system for it. It's okay to take a step back from the current task, in order to do things right. This relates directly to the Modularity Principle above.\n\n### Testing Principle\nBBOT has extremely thorough tests, including **one or more individual tests for each module, with no exceptions**. This is critical to maintaining stability in a recursive tool, which by its nature flirts with race conditions and infinite loops. If you add a module, you write a test. If you change a module, you make sure its test still passes.\n\n---\n\n## Tooling\n\n- **Package manager**: [uv](https://docs.astral.sh/uv/)\n- **Linter/formatter**: [ruff](https://docs.astral.sh/ruff/) (pinned to 0.15.10)\n- **Test framework**: [pytest](https://docs.pytest.org/) with pytest-asyncio\n- **Python**: 3.10 - 3.14\n\n---\n\n## Dev Environment Setup\n\n```bash\n# 1. Fork and clone\ngit clone git@github.com:<you>/bbot.git\ncd bbot\n\n# 2. Switch to dev branch, then create a feature branch\ngit checkout dev\ngit checkout -b my-feature\n\n# 3. Install uv (if you haven't already)\ncurl -LsSf https://astral.sh/uv/install.sh | sh\n\n# 4. Install all dependencies (including dev)\nuv sync --group dev\n\n# 5. Install pre-commit hooks (ruff, file checks, etc.)\nuv run pre-commit install\n\n# 6. Activate the virtualenv\nsource .venv/bin/activate\n\n# 7. Verify\nbbot --help\n```\n\n### Running Tests\n\n```bash\n# Run the full suite\n./bbot/test/run_tests.sh\n\n# Run specific module tests\n./bbot/test/run_tests.sh robots,sslcert\n\n# Run a single test file directly\npytest bbot/test/test_step_2/module_tests/test_module_robots.py -x -vv\n```\n\n### Linting\n\n```bash\nruff check          # lint\nruff format          # auto-format\nruff format --check  # verify formatting without changes\n```\n\n### Git Workflow\n\n- `stable` - production releases\n- `dev` - active development, **almost all PRs should target this branch**\n- Feature branches should be created from `dev`\n\n---\n\n## AI Use Disclosure\n\nUse of AI is not prohibited -- and in many cases, encouraged. However, when reviewing a PR, it is helpful for the reviewer to know the extent to which AI was used, and which model.\nPlease add a small section at the bottom of the PR with the header: `### AI Use Disclosure`, followed by the following information:\n\n* Extent of the AI use. For example, was this fully autonomous by the AI, or was it a collaborative back-and-forth, or did the user just use the AI to review their work, etc.\n* Model Used\n\nThis should only apply to external contributors, not members of the blacklanternsecurity organization.\n\n---\n\n## Architecture Overview\n\n### How a Scan Works\n\nBBOT is an async, recursive OSINT tool. A scan starts with **seed events** (targets) and passes them through a pipeline of **modules**. Each module watches for specific event types, processes them, and may emit new events, which feed back into the pipeline. This continues until no module has anything left to do.\n\n```\n                     Seeds (targets)\n                          |\n                          v\n                   +--------------+\n                   | ScanIngress  |  dedup, blacklist, scope check\n                   +--------------+\n                          |\n                          v\n                   +--------------+\n                   |  Intercept   |  dns, cloud tagging\n                   |   Modules    |  (modify/tag events before distribution)\n                   +--------------+\n                          |\n                          v\n                   +--------------+\n                   |  ScanEgress  |  scope filtering, graph management\n                   +--------------+\n                          |\n                          v\n              +-----------+-----------+\n              |           |           |\n           Module A    Module B    Module C   ...\n              |           |           |\n              +-----------+-----------+\n                          |\n                          v\n                   Output Modules (json, csv, neo4j, ...)\n```\n\n### Events\n\nEvents are the currency of BBOT. Every piece of data -- a hostname, IP, URL, open port, finding -- is an event. Events have:\n\n- **type**: `DNS_NAME`, `IP_ADDRESS`, `URL`, `OPEN_TCP_PORT`, `HTTP_RESPONSE`, `FINDING`, `EMAIL_ADDRESS`, etc.\n- **data**: the actual data (a string, dict, etc.)\n- **parent**: the event that led to this one (forming a discovery chain)\n- **scope_distance**: how many hops from the original target (0 = in-scope)\n- **tags**: metadata like `in-scope`, `affiliate`, `cloud-azure`, `open-port`, etc.\n- **module**: which module discovered it\n\n### Scope Distance\n\nScope distance tracks how far an event is from the original target:\n- `0` = explicitly in-scope (matches target or discovered in-scope)\n- `1` = one hop away (e.g. a hostname found in an SSL cert of an in-scope host)\n- `2+` = further away\n\nThe scan's `scope.search_distance` (default 0) controls how far modules are allowed to look. A module's `scope_distance_modifier` adjusts this per-module.\n\n### Helpers\n\nBBOT has a helper for almost everything. **Please use them.** They're accessible via `self.helpers` inside any module.\n\nKey helpers:\n\n| Helper | What it does |\n|--------|-------------|\n| `self.helpers.request(url)` | Make an HTTP request (with retries, SSL handling, etc.) |\n| `self.helpers.blasthttp` | Shared blasthttp client (rate-limited via `web.http_rate_limit` config) |\n| `self.helpers.resolve(host)` | DNS resolution |\n| `self.helpers.is_ip(s)` | Check if string is an IP |\n| `self.helpers.is_dns_name(s)` | Check if string is a hostname |\n| `self.helpers.split_domain(host)` | Split into subdomain + root domain |\n| `self.helpers.domain_parents(domain)` | Get all parent domains |\n| `self.helpers.make_netloc(host, port)` | Format `host:port` (handles IPv6) |\n| `self.helpers.parent_domain(domain)` | Get immediate parent domain |\n| `self.helpers.beautifulsoup(html, parser)` | Parse HTML |\n| `self.helpers.validators.validate_host(h)` | Validate and normalize a hostname |\n| `self.helpers.tempfile(data, pipe=False)` | Create a temp file with content |\n| `self.helpers.run(command)` | Run a shell command |\n| `self.helpers.run_live(command)` | Run a shell command, stream output |\n| `self.helpers.as_completed(tasks)` | Async iteration of completed tasks |\n| `self.helpers.wordlist(url_or_path)` | Download/cache a wordlist |\n| `self.helpers.rand_string(n)` | Random string of length n |\n| `self.helpers.regexes.email_regex` | Pre-compiled email regex |\n| `self.helpers.add_get_params(url, params)` | Add query params to a URL |\n| `self.helpers.quote(s)` | URL-encode a string |\n| `self.helpers.make_ip_type(s)` | Convert string to `ipaddress` object |\n| `self.helpers.parse_port_string(s)` | Parse port range string (e.g. `\"80,443,8000-9000\"`) |\n| `self.helpers.top_tcp_ports(n)` | Get top N TCP ports |\n\nThere are hundreds more in `bbot/core/helpers/misc.py`. Browse them before writing utility code yourself.\n\n---\n\n## Writing a Module\n\n### Quick Start\n\n1. Create `bbot/modules/my_module.py`\n2. Define a class that inherits from `BaseModule`\n3. Set `watched_events`, `produced_events`, `flags`, and `meta`\n4. Implement `handle_event()`\n5. Create `bbot/test/test_step_2/module_tests/test_module_my_module.py`\n\nHere's a minimal module:\n\n```python\nfrom bbot.modules.base import BaseModule\n\n\nclass my_module(BaseModule):\n    watched_events = [\"DNS_NAME\"]\n    produced_events = [\"EMAIL_ADDRESS\"]\n    flags = [\"passive\", \"email-enum\"]\n    meta = {\n        \"description\": \"Query example.com for email addresses\",\n        \"created_date\": \"2025-01-01\",\n        \"author\": \"@you\",\n    }\n\n    async def handle_event(self, event):\n        url = f\"https://api.example.com/lookup?domain={event.data}\"\n        r = await self.helpers.request(url)\n        if r and r.status_code == 200:\n            for email in r.json().get(\"emails\", []):\n                await self.emit_event(\n                    email,\n                    \"EMAIL_ADDRESS\",\n                    parent=event,\n                    context=f\"{{module}} queried example.com and found {{event.type}}: {{event.data}}\",\n                )\n```\n\nAnd its test:\n\n```python\nfrom .base import ModuleTestBase\n\n\nclass TestMyModule(ModuleTestBase):\n    async def setup_after_prep(self, module_test):\n        module_test.blasthttp_mock.add_response(\n            url=\"https://api.example.com/lookup?domain=blacklanternsecurity.com\",\n            json={\"emails\": [\"info@blacklanternsecurity.com\"]},\n        )\n\n    def check(self, module_test, events):\n        assert any(\n            e.data == \"info@blacklanternsecurity.com\" and e.type == \"EMAIL_ADDRESS\"\n            for e in events\n        ), \"Failed to find email\"\n```\n\n### Module Lifecycle\n\n```\nsetup()  -->  handle_event() (called many times)  -->  finish()  -->  report()  -->  cleanup()\n```\n\n1. **`setup()`** - one-time initialization (validate config, download data, check API keys)\n2. **`handle_event(event)`** - called for each matching event\n3. **`finish()`** - called when the scan is finishing; can still emit events\n4. **`report()`** - called once after finish; for summary output\n5. **`cleanup()`** - called last; close files, delete temp data; **cannot** emit events\n\n---\n\n### Module Attributes Reference\n\n#### Event Configuration\n\n##### `watched_events` (list)\nEvent types this module wants to process. The module's `handle_event()` is only called for these types.\n\n```python\n# sslcert.py - watches for open ports to grab SSL certs from\nwatched_events = [\"OPEN_TCP_PORT\"]\n\n# newsletters.py - watches HTTP responses to scan HTML\nwatched_events = [\"HTTP_RESPONSE\"]\n\n# json.py (output module) - watches everything\nwatched_events = [\"*\"]\n```\n\n##### `produced_events` (list)\nEvent types this module may emit. Used for dependency resolution and documentation.\n\n```python\n# sslcert.py - can discover hostnames and emails from certificates\nproduced_events = [\"DNS_NAME\", \"EMAIL_ADDRESS\"]\n\n# portscan.py - finds open ports\nproduced_events = [\"OPEN_TCP_PORT\"]\n```\n\n##### `flags` (list)\nTags that describe the module's behavior. Must include at least one activity flag (`passive` or `active`). Must also include `safe`, `loud`, or `invasive` (or a combination of `loud` and `invasive`).\n\nCommon flags:\n- `passive` / `active` - whether the module touches the target directly\n- `safe` - non-intrusive and non-destructive\n- `loud` - generates a large amount of network traffic\n- `invasive` - intrusive or potentially destructive\n- `subdomain-enum` - participates in subdomain enumeration\n- `web` - basic web scanning\n- `email-enum` - email discovery\n\n```python\n# crt.py - queries a third-party API, never touches the target\nflags = [\"subdomain-enum\", \"passive\"]\n\n# sslcert.py - connects directly to target ports\nflags = [\"affiliates\", \"subdomain-enum\", \"email-enum\", \"active\", \"web\"]\n```\n\n##### `meta` (dict)\nModule metadata. Must include `description`, `created_date`, and `author`. Set `auth_required: True` if the module needs an API key.\n\n```python\nmeta = {\n    \"description\": \"Query crt.sh (certificate transparency) for subdomains\",\n    \"created_date\": \"2022-05-13\",\n    \"author\": \"@TheTechromancer\",\n}\n\n# For API-key modules:\nmeta = {\"description\": \"Query API for subdomains\", \"auth_required\": True}\n```\n\n---\n\n#### Options\n\n##### `options` / `options_desc` (dict)\nUser-configurable settings. Access them via `self.config.get(\"option_name\")`.\n\n```python\n# robots.py - configurable parsing options\noptions = {\"include_sitemap\": False, \"include_allow\": True, \"include_disallow\": True}\noptions_desc = {\n    \"include_sitemap\": \"Include 'sitemap' entries\",\n    \"include_allow\": \"Include 'Allow' Entries\",\n    \"include_disallow\": \"Include 'Disallow' Entries\",\n}\n\n# In handle_event():\nif self.config.get(\"include_sitemap\") is True:\n    ...\n```\n\n```python\n# sslcert.py - timeout and behavior options\noptions = {\"timeout\": 5.0, \"skip_non_ssl\": True}\noptions_desc = {\"timeout\": \"Socket connect timeout in seconds\", \"skip_non_ssl\": \"Don't try common non-SSL ports\"}\n```\n\n---\n\n#### Scope & Filtering\n\n##### `scope_distance_modifier` (int or None) -- default: `0`\nControls which events the module accepts based on how far they are from the target.\n\n- `0` (default) - accept events up to the scan's configured search distance\n- `1` - accept events up to search distance + 1\n- `None` - accept all events regardless of distance\n\n```python\n# sslcert.py - looks one hop beyond normal scope, because certificate names\n# found on in-scope hosts often reveal related infrastructure\nscope_distance_modifier = 1\n```\n\n##### `in_scope_only` (bool) -- default: `False`\nOnly accept events that are explicitly in-scope (distance == 0). More restrictive than `scope_distance_modifier = 0`.\n\n```python\n# robots.py - only fetch robots.txt for in-scope hosts\nin_scope_only = True\n```\n\n##### `target_only` (bool) -- default: `False`\nOnly accept the initial target/seed events. Useful for modules that should only run once against the original targets.\n\n##### `accept_seeds` (bool) -- default: `True` for passive, `False` for active\nWhether to process seed events (the initial targets provided to the scan).\n\n##### `accept_url_special` (bool) -- default: `False`\nWhether to accept \"special\" URLs (e.g. JavaScript files) that are not normally distributed to web modules.\n\n```python\n# http.py - needs to process all URLs including special ones\naccept_url_special = True\n```\n\n---\n\n#### Deduplication\n\n##### `accept_dupes` (bool) -- default: `False`\nWhether to accept the same event more than once. Most modules should leave this `False`.\n\n```python\n# Output modules set this True because they need to see every event\naccept_dupes = True\n```\n\n##### `suppress_dupes` (bool) -- default: `True`\nWhether to suppress duplicate *outgoing* events. Prevents the same event from being emitted twice.\n\n##### `per_host_only` (bool) -- default: `False`\nOnly process one event per unique host. After processing `1.2.3.4`, skip any future events for `1.2.3.4`.\n\n##### `per_hostport_only` (bool) -- default: `False`\nOnly process one event per unique host:port combination.\n\n```python\n# robots.py - only fetch robots.txt once per host:port\nper_hostport_only = True\n```\n\n##### `per_domain_only` (bool) -- default: `False`\nOnly process one event per unique root domain. After processing `www.example.com`, skip `api.example.com`.\n\n```python\n# emailformat.py - one API query per domain is enough\nper_domain_only = True\n```\n\n##### `_incoming_dedup_hash(self, event)` -- override for custom dedup\nOverride this to define custom deduplication logic. Return a hash (int) or `(hash, reason_string)`.\n\n```python\n# securitytxt.py - dedupe by parent domain so we only check security.txt once\n# per parent domain, not once per subdomain\ndef _incoming_dedup_hash(self, event):\n    parent_domain = self.helpers.parent_domain(event.data)\n    return hash(parent_domain), \"already processed parent domain\"\n```\n\n```python\n# subdomain_enum.py template - dedupe by highest or lowest parent domain\ndef _incoming_dedup_hash(self, event):\n    return hash(self.make_query(event)), f\"dedup_strategy={self.dedup_strategy}\"\n```\n\n---\n\n#### Concurrency & Batching\n\n##### `_module_threads` (int) -- default: `1`\nHow many `handle_event()` calls can run concurrently. Increase this for I/O-bound modules.\n\n```python\n# sslcert.py - connects to many hosts in parallel\n_module_threads = 25\n```\n\n##### `_batch_size` (int) -- default: `1`\nWhen > 1, events are collected into batches and passed to `handle_batch(*events)` instead of `handle_event()`. Useful for tools that work better with bulk input.\n\n```python\n# portscan.py - masscan is most efficient with all targets at once\nbatch_size = 1000000\n\nasync def handle_batch(self, *events):\n    targets, correlator = await self.make_targets(events, self.syn_scanned)\n    async for ip, port, parent_event in self.masscan(targets, correlator):\n        await self.emit_open_port(ip, port, parent_event)\n```\n\n##### `_shuffle_incoming_queue` (bool) -- default: `True`\nWhether to randomize the order of incoming events. Set to `False` when order matters.\n\n```python\n# portscan.py - processes all events together, order doesn't matter but\n# we disable shuffle because batch_size is huge\n_shuffle_incoming_queue = False\n```\n\n---\n\n#### Dependencies\n\n##### `deps_pip` (list)\nPython packages to install.\n\n```python\n# sslcert.py\ndeps_pip = [\"pyOpenSSL~=25.3.0\"]\n```\n\n##### `deps_apt` (list)\nSystem packages to install.\n\n```python\n# sslcert.py\ndeps_apt = [\"openssl\"]\n```\n\n##### `deps_modules` (list)\nOther BBOT modules that must be enabled for this module to work.\n\n##### `deps_shell` (list)\nShell commands to run for installation (uses Ansible's `shell` module).\n\n##### `deps_ansible` (list)\nAnsible tasks for complex dependency installation (downloading binaries, etc.).\n\n```python\n# fingerprintx.py - downloads a Go binary\ndeps_ansible = [\n    {\n        \"name\": \"Download fingerprintx\",\n        \"unarchive\": {\n            \"src\": \"https://github.com/.../fingerprintx_{version}_{platform}_{arch}.tar.gz\",\n            \"include\": \"fingerprintx\",\n            \"dest\": \"#{BBOT_TOOLS}\",\n            \"remote_src\": True,\n        },\n    },\n]\n```\n\n---\n\n#### Priority & Queue\n\n##### `_priority` (int) -- default: `3`\nModule priority from 1 (highest) to 5 (lowest). Lower-priority modules get events first.\n\n```python\n# sslcert.py - runs early because other modules depend on the hostnames it discovers\n_priority = 2\n```\n\n##### `_qsize` (int) -- default: `1000`\nOutgoing event queue size. A smaller queue creates backpressure that helps with rate limiting.\n\n```python\n# subdomain_enum.py template - small queue to combat API rate limiting\n_qsize = 10\n```\n\n##### `_preserve_graph` (bool) -- default: `False`\nAccept duplicate events that are needed for complete event chain construction. Only used by output modules.\n\n```python\n# json.py - needs complete event chains for accurate output\n_preserve_graph = True\n```\n\n##### `_stats_exclude` (bool) -- default: `False`\nExclude this module from scan statistics. Used by output and report modules.\n\n##### `_disable_auto_module_deps` (bool) -- default: `False`\nPrevent BBOT from automatically enabling dependency modules. For example, if your module watches `URL` events, BBOT normally auto-enables `http`. Set this to `True` to prevent that.\n\n---\n\n### Key Methods\n\n#### `handle_event(self, event)` -- the core method\n\nCalled once for each matching event. This is where your module does its work.\n\n```python\n# robots.py - fetch and parse robots.txt\nasync def handle_event(self, event):\n    host = f\"{event.parsed_url.scheme}://{event.parsed_url.netloc}/\"\n    url = f\"{host}robots.txt\"\n    result = await self.helpers.request(url)\n    if result:\n        body = result.text\n        if body:\n            for line in body.split(\"\\n\"):\n                if line.startswith(\"Disallow:\"):\n                    path = line.split(\": \", 1)[1].lstrip(\"/\")\n                    await self.emit_event(\n                        f\"{host}{path}\",\n                        \"URL_UNVERIFIED\",\n                        parent=event,\n                        tags=[\"spider-danger\"],\n                    )\n```\n\n#### `handle_batch(self, *events)` -- bulk processing\n\nUsed when `_batch_size > 1`. Receives multiple events at once.\n\n```python\n# portscan.py - bulk port scanning with masscan\nasync def handle_batch(self, *events):\n    targets, correlator = await self.make_targets(events, self.syn_scanned)\n    async for ip, port, parent_event in self.masscan(targets, correlator):\n        await self.emit_open_port(ip, port, parent_event)\n```\n\n#### `filter_event(self, event)` -- custom event filtering\n\nCalled before `handle_event()`. Return `True` to accept, `False` to reject, or `(False, \"reason\")` to reject with a logged reason.\n\n```python\n# sslcert.py - skip ports that don't typically use SSL\nasync def filter_event(self, event):\n    if self.skip_non_ssl and event.port in self.non_ssl_ports:\n        return False, f\"Port {event.port} doesn't typically use SSL\"\n    return True\n```\n\n```python\n# subdomain_enum.py template - reject wildcards and cloud resources\nasync def filter_event(self, event):\n    query = self.make_query(event)\n    is_wildcard = await self._is_wildcard(query)\n    if self.reject_wildcards and is_wildcard:\n        return False, \"Event is a wildcard domain\"\n    return True, \"\"\n```\n\n#### `setup(self)` -- one-time initialization\n\nReturn values:\n- `True` -- success\n- `(True, \"message\")` -- success with message\n- `None` or `(None, \"message\")` -- **soft fail**: module is disabled, scan continues\n- `False` or `(False, \"message\")` -- **hard fail**: scan aborts\n\n```python\n# portscan.py - validates config, checks masscan, checks IPv6 support\nasync def setup(self):\n    self.top_ports = self.config.get(\"top_ports\", 100)\n    self.rate = self.config.get(\"rate\", 300)\n    self.ports = self.config.get(\"ports\", \"\")\n    if self.ports:\n        try:\n            self.helpers.parse_port_string(self.ports)\n        except ValueError as e:\n            return False, f\"Error parsing ports '{self.ports}': {e}\"\n    # ...\n    return True\n```\n\n```python\n# subdomain_enum_apikey template - soft-fail if API key is missing\nasync def setup(self):\n    await super().setup()\n    return await self.require_api_key()\n    # Returns (None, \"No API key set\") if missing, disabling the module\n```\n\n#### `finish(self)` -- called when scan is finishing\n\nCan still emit events. May be called multiple times if new activity is detected.\n\n#### `report(self)` -- summary output\n\nCalled once after `finish()`. Use for generating tables or summary data.\n\n```python\n# asn.py - output ASN statistics\nasync def report(self):\n    self.log_table(table_data, headers=[\"ASN\", \"Subnet\", \"Count\"], table_name=\"asns\")\n```\n\n#### `cleanup(self)` -- resource cleanup\n\nCalled once at the very end. Close files, delete temp files. **Cannot emit events.**\n\n```python\n# json.py\nasync def cleanup(self):\n    if getattr(self, \"_file\", None) is not None:\n        with suppress(Exception):\n            self.file.close()\n```\n\n```python\n# portscan.py\nasync def cleanup(self):\n    with suppress(Exception):\n        self.exclude_file.unlink()\n```\n\n---\n\n### Emitting Events\n\n#### `emit_event(data, event_type, parent, **kwargs)`\n\nCreates and queues an event for processing by other modules.\n\n```python\n# Simple string event\nawait self.emit_event(\"sub.example.com\", \"DNS_NAME\", parent=event)\n\n# With context (used for discovery chain documentation)\nawait self.emit_event(\n    \"sub.example.com\",\n    \"DNS_NAME\",\n    parent=event,\n    context=f\"{{module}} queried crt.sh and found {{event.type}}: {{event.data}}\",\n)\n\n# With tags\nawait self.emit_event(url, \"URL_UNVERIFIED\", parent=event, tags=[\"spider-danger\"])\n\n# FINDING event (dict data)\nawait self.emit_event(\n    {\n        \"host\": str(event.host),\n        \"description\": \"Found something interesting\",\n        \"url\": event.data[\"url\"],\n        \"severity\": \"HIGH\",\n    },\n    \"FINDING\",\n    parent=event,\n)\n```\n\n#### `make_event(data, event_type, parent, **kwargs)`\n\nCreates an event without emitting it. Useful when you need to inspect or modify it first.\n\n```python\nssl_event = self.make_event(hostname, \"DNS_NAME\", parent=event, raise_error=True)\nif ssl_event:\n    await self.emit_event(ssl_event, tags=[\"affiliate\"])\n```\n\n---\n\n### API Helpers\n\n#### `api_request(url, **kwargs)`\n\nHTTP request with automatic retry, rate-limit handling (429), API key cycling, and failure tracking. After too many failures, the module enters error state.\n\n```python\nr = await self.api_request(\"https://api.example.com/search?q=test\")\nif r and r.status_code == 200:\n    data = r.json()\n```\n\n#### `require_api_key()`\n\nValidates that an API key is configured. Call in `setup()`.\n\n```python\nasync def setup(self):\n    return await self.require_api_key()\n```\n\n#### `api_page_iter(url, page_size=100, **kwargs)`\n\nAsync generator for paginated API results. URL can contain `{page}`, `{page_size}`, and `{offset}` placeholders.\n\n```python\nasync for page in self.api_page_iter(\n    \"https://api.example.com/search?q=test&page={page}&limit={page_size}\"\n):\n    if not page.get(\"results\"):\n        break\n    for result in page[\"results\"]:\n        await self.emit_event(result[\"hostname\"], \"DNS_NAME\", parent=event)\n```\n\n---\n\n### Running External Processes\n\n```python\n# Run a command and get the result\nresult = await self.run_process([\"nmap\", \"-p\", \"22,80\", target])\nif result.returncode == 0:\n    output = result.stdout\n\n# Stream output line-by-line (for long-running tools)\nasync for line in self.run_process_live([\"masscan\", \"-oJ\", \"-\", ...]):\n    data = json.loads(line)\n```\n\n---\n\n### Logging\n\n```python\nself.debug(\"Low-level detail\")        # only visible with -d flag\nself.verbose(\"Useful but not critical\") # visible with -v flag\nself.info(\"Standard info\")\nself.success(\"Something good happened\") # green\nself.warning(\"Something concerning\")    # orange\nself.error(\"Something failed\")          # red\n\n# \"Huge\" variants: entire line in bold color\nself.hugesuccess(\"Major discovery!\")\nself.hugewarning(\"Major concern!\")\n```\n\n---\n\n### Templates\n\nFor common patterns, inherit from a template instead of `BaseModule` directly. Templates live in `bbot/modules/templates/`:\n\n- **`subdomain_enum`** - passive subdomain enumeration via free API. Handles dedup, wildcard rejection, query building.\n- **`subdomain_enum_apikey`** - same as above but requires an API key.\n- **`shodan`** - Shodan API integration.\n- **`github`** - GitHub API integration.\n- **`censys`** - Censys API integration.\n- **`bucket`** - Cloud storage bucket enumeration.\n- **`webhook`** - Webhook output.\n\nExample: `crt.py` inherits from `subdomain_enum` and only needs to override the request/parse logic:\n\n```python\nfrom bbot.modules.templates.subdomain_enum import subdomain_enum\n\n\nclass crt(subdomain_enum):\n    flags = [\"subdomain-enum\", \"passive\"]\n    watched_events = [\"DNS_NAME\"]\n    produced_events = [\"DNS_NAME\"]\n    meta = {\n        \"description\": \"Query crt.sh (certificate transparency) for subdomains\",\n        \"created_date\": \"2022-05-13\",\n        \"author\": \"@TheTechromancer\",\n    }\n    base_url = \"https://crt.sh\"\n\n    async def request_url(self, query):\n        params = {\"q\": f\"%.{query}\", \"output\": \"json\"}\n        url = self.helpers.add_get_params(self.base_url, params).geturl()\n        return await self.api_request(url, timeout=self.http_timeout + 30)\n\n    async def parse_results(self, r, query):\n        results = set()\n        for cert_info in r.json():\n            domain = cert_info.get(\"name_value\")\n            if domain:\n                for d in domain.splitlines():\n                    results.add(d.lower())\n        return results\n```\n\n---\n\n### Module Types\n\n#### Scan Modules (default)\nNormal modules that watch events and produce new ones. This is what you'll write 95% of the time.\n\n#### Output Modules\nInherit from `BaseOutputModule`. Receive all events and write them somewhere.\n\n```python\nfrom bbot.modules.output.base import BaseOutputModule\n\nclass my_output(BaseOutputModule):\n    watched_events = [\"*\"]\n    meta = {\"description\": \"Custom output\"}\n    _preserve_graph = True  # maintain complete event chains\n\n    async def handle_event(self, event):\n        # write event to file, database, API, etc.\n        ...\n```\n\nOutput modules automatically get:\n- `accept_dupes = True`\n- `scope_distance_modifier = None` (see all events)\n- `_stats_exclude = True`\n\n#### Internal Modules\nInherit from `BaseInternalModule`. System-level modules that aren't exposed to users.\n\n#### Intercept Modules\nInherit from `BaseInterceptModule`. Special high-priority modules that can modify or reject events before they reach normal modules. Used for DNS resolution, cloud detection, etc. You probably don't need to write one.\n\n---\n\n### Writing Tests\n\nEvery module needs a test in `bbot/test/test_step_2/module_tests/`. The test file must be named `test_module_<name>.py`.\n\nTest classes inherit from `ModuleTestBase` and follow this pattern:\n\n```python\nfrom .base import ModuleTestBase\n\n\nclass TestMyModule(ModuleTestBase):\n    # Optional: override targets (default: [\"blacklanternsecurity.com\"])\n    targets = [\"http://127.0.0.1:8888\"]\n\n    # Optional: override which modules are enabled\n    modules_overrides = [\"http\", \"my_module\"]\n\n    # Optional: override config\n    config_overrides = {\"modules\": {\"my_module\": {\"some_option\": True}}}\n\n    async def setup_before_prep(self, module_test):\n        \"\"\"Called BEFORE the scan is prepared. Set up HTTP mocks here.\"\"\"\n        pass\n\n    async def setup_after_prep(self, module_test):\n        \"\"\"Called AFTER the scan is prepared. Modify modules, add mocks here.\"\"\"\n        # Mock an HTTP response\n        module_test.blasthttp_mock.add_response(\n            url=\"https://api.example.com/lookup?domain=blacklanternsecurity.com\",\n            json={\"results\": [\"sub.blacklanternsecurity.com\"]},\n        )\n\n        # Mock DNS\n        await module_test.mock_dns({\n            \"blacklanternsecurity.com\": {\"A\": [\"127.0.0.88\"]},\n        })\n\n        # Mock an HTTP server response\n        module_test.set_expect_requests(\n            expect_args={\"method\": \"GET\", \"uri\": \"/robots.txt\"},\n            respond_args={\"response_data\": \"Disallow: /secret/\"},\n        )\n\n    def check(self, module_test, events):\n        \"\"\"Verify the scan produced the expected events.\"\"\"\n        assert any(\n            e.data == \"sub.blacklanternsecurity.com\" and e.type == \"DNS_NAME\"\n            for e in events\n        ), \"Failed to find subdomain\"\n```\n\nThe test lifecycle runs:\n1. `setup_before_prep()` - set up mocks\n2. Scan `_prep()` - loads modules, config\n3. `setup_after_prep()` - modify scan state\n4. Scan runs and collects events\n5. `check()` - your assertions\n\n### Test Utilities\n\n- **`module_test.blasthttp_mock`** - mock HTTP responses \n- **`module_test.httpserver`** - real HTTP server on port 8888\n- **`module_test.httpserver_ssl`** - real HTTPS server on port 9999\n- **`module_test.mock_dns(data)`** - mock DNS responses\n- **`module_test.mock_interactsh(name)`** - mock out-of-band interactions\n- **`module_test.module`** - reference to the module instance being tested\n- **`module_test.scan`** - reference to the Scanner instance\n\nReal example -- `test_module_robots.py`:\n\n```python\nclass TestRobots(ModuleTestBase):\n    targets = [\"http://127.0.0.1:8888\"]\n    modules_overrides = [\"http\", \"robots\"]\n    config_overrides = {\"modules\": {\"robots\": {\"include_sitemap\": True}}}\n\n    async def setup_after_prep(self, module_test):\n        robots = \"Allow: /allow/\\nDisallow: /disallow/\\nSitemap: http://127.0.0.1:8888/sitemap.txt\"\n        module_test.set_expect_requests(\n            expect_args={\"method\": \"GET\", \"uri\": \"/robots.txt\"},\n            respond_args={\"response_data\": robots},\n        )\n\n    def check(self, module_test, events):\n        assert any(e.data == \"http://127.0.0.1:8888/allow/\" for e in events)\n        assert any(e.data == \"http://127.0.0.1:8888/disallow/\" for e in events)\n        assert any(e.data == \"http://127.0.0.1:8888/sitemap.txt\" for e in events)\n```\n","CLAUDE.md":"Treat @AGENTS.md the same way you'd treat CLAUDE.md."},"files":{"AGENTS.md":"# BBOT Developer Guide\n\n## Core Principles\n\n### Modularity Principle\nWhen writing a BBOT module, make sure all module-specific code lives in the module itself. Don't hard-code module-specific things in core or in helpers.\n\n### DRY Principle\nDon't Repeat Yourself -- and interpret this broadly. If two pieces of code aren't identical but follow a similar enough pattern that they could be generalized, they should be. Extract shared logic into a common abstraction rather than duplicating the pattern. Usually this means creating a shared helper, or a shared module template in `bbot/modules/templates`. When you notice structural similarity, unify it.\n\n### Engineering Principle\nEvery system that is implemented must be implemented properly. No hacks, no hardcoding, no shortcuts. If we implement one of something, we build a proper system for it. It's okay to take a step back from the current task, in order to do things right. This relates directly to the Modularity Principle above.\n\n### Testing Principle\nBBOT has extremely thorough tests, including **one or more individual tests for each module, with no exceptions**. This is critical to maintaining stability in a recursive tool, which by its nature flirts with race conditions and infinite loops. If you add a module, you write a test. If you change a module, you make sure its test still passes.\n\n---\n\n## Tooling\n\n- **Package manager**: [uv](https://docs.astral.sh/uv/)\n- **Linter/formatter**: [ruff](https://docs.astral.sh/ruff/) (pinned to 0.15.10)\n- **Test framework**: [pytest](https://docs.pytest.org/) with pytest-asyncio\n- **Python**: 3.10 - 3.14\n\n---\n\n## Dev Environment Setup\n\n```bash\n# 1. Fork and clone\ngit clone git@github.com:<you>/bbot.git\ncd bbot\n\n# 2. Switch to dev branch, then create a feature branch\ngit checkout dev\ngit checkout -b my-feature\n\n# 3. Install uv (if you haven't already)\ncurl -LsSf https://astral.sh/uv/install.sh | sh\n\n# 4. Install all dependencies (including dev)\nuv sync --group dev\n\n# 5. Install pre-commit hooks (ruff, file checks, etc.)\nuv run pre-commit install\n\n# 6. Activate the virtualenv\nsource .venv/bin/activate\n\n# 7. Verify\nbbot --help\n```\n\n### Running Tests\n\n```bash\n# Run the full suite\n./bbot/test/run_tests.sh\n\n# Run specific module tests\n./bbot/test/run_tests.sh robots,sslcert\n\n# Run a single test file directly\npytest bbot/test/test_step_2/module_tests/test_module_robots.py -x -vv\n```\n\n### Linting\n\n```bash\nruff check          # lint\nruff format          # auto-format\nruff format --check  # verify formatting without changes\n```\n\n### Git Workflow\n\n- `stable` - production releases\n- `dev` - active development, **almost all PRs should target this branch**\n- Feature branches should be created from `dev`\n\n---\n\n## AI Use Disclosure\n\nUse of AI is not prohibited -- and in many cases, encouraged. However, when reviewing a PR, it is helpful for the reviewer to know the extent to which AI was used, and which model.\nPlease add a small section at the bottom of the PR with the header: `### AI Use Disclosure`, followed by the following information:\n\n* Extent of the AI use. For example, was this fully autonomous by the AI, or was it a collaborative back-and-forth, or did the user just use the AI to review their work, etc.\n* Model Used\n\nThis should only apply to external contributors, not members of the blacklanternsecurity organization.\n\n---\n\n## Architecture Overview\n\n### How a Scan Works\n\nBBOT is an async, recursive OSINT tool. A scan starts with **seed events** (targets) and passes them through a pipeline of **modules**. Each module watches for specific event types, processes them, and may emit new events, which feed back into the pipeline. This continues until no module has anything left to do.\n\n```\n                     Seeds (targets)\n                          |\n                          v\n                   +--------------+\n                   | ScanIngress  |  dedup, blacklist, scope check\n                   +--------------+\n                          |\n                          v\n                   +--------------+\n                   |  Intercept   |  dns, cloud tagging\n                   |   Modules    |  (modify/tag events before distribution)\n                   +--------------+\n                          |\n                          v\n                   +--------------+\n                   |  ScanEgress  |  scope filtering, graph management\n                   +--------------+\n                          |\n                          v\n              +-----------+-----------+\n              |           |           |\n           Module A    Module B    Module C   ...\n              |           |           |\n              +-----------+-----------+\n                          |\n                          v\n                   Output Modules (json, csv, neo4j, ...)\n```\n\n### Events\n\nEvents are the currency of BBOT. Every piece of data -- a hostname, IP, URL, open port, finding -- is an event. Events have:\n\n- **type**: `DNS_NAME`, `IP_ADDRESS`, `URL`, `OPEN_TCP_PORT`, `HTTP_RESPONSE`, `FINDING`, `EMAIL_ADDRESS`, etc.\n- **data**: the actual data (a string, dict, etc.)\n- **parent**: the event that led to this one (forming a discovery chain)\n- **scope_distance**: how many hops from the original target (0 = in-scope)\n- **tags**: metadata like `in-scope`, `affiliate`, `cloud-azure`, `open-port`, etc.\n- **module**: which module discovered it\n\n### Scope Distance\n\nScope distance tracks how far an event is from the original target:\n- `0` = explicitly in-scope (matches target or discovered in-scope)\n- `1` = one hop away (e.g. a hostname found in an SSL cert of an in-scope host)\n- `2+` = further away\n\nThe scan's `scope.search_distance` (default 0) controls how far modules are allowed to look. A module's `scope_distance_modifier` adjusts this per-module.\n\n### Helpers\n\nBBOT has a helper for almost everything. **Please use them.** They're accessible via `self.helpers` inside any module.\n\nKey helpers:\n\n| Helper | What it does |\n|--------|-------------|\n| `self.helpers.request(url)` | Make an HTTP request (with retries, SSL handling, etc.) |\n| `self.helpers.blasthttp` | Shared blasthttp client (rate-limited via `web.http_rate_limit` config) |\n| `self.helpers.resolve(host)` | DNS resolution |\n| `self.helpers.is_ip(s)` | Check if string is an IP |\n| `self.helpers.is_dns_name(s)` | Check if string is a hostname |\n| `self.helpers.split_domain(host)` | Split into subdomain + root domain |\n| `self.helpers.domain_parents(domain)` | Get all parent domains |\n| `self.helpers.make_netloc(host, port)` | Format `host:port` (handles IPv6) |\n| `self.helpers.parent_domain(domain)` | Get immediate parent domain |\n| `self.helpers.beautifulsoup(html, parser)` | Parse HTML |\n| `self.helpers.validators.validate_host(h)` | Validate and normalize a hostname |\n| `self.helpers.tempfile(data, pipe=False)` | Create a temp file with content |\n| `self.helpers.run(command)` | Run a shell command |\n| `self.helpers.run_live(command)` | Run a shell command, stream output |\n| `self.helpers.as_completed(tasks)` | Async iteration of completed tasks |\n| `self.helpers.wordlist(url_or_path)` | Download/cache a wordlist |\n| `self.helpers.rand_string(n)` | Random string of length n |\n| `self.helpers.regexes.email_regex` | Pre-compiled email regex |\n| `self.helpers.add_get_params(url, params)` | Add query params to a URL |\n| `self.helpers.quote(s)` | URL-encode a string |\n| `self.helpers.make_ip_type(s)` | Convert string to `ipaddress` object |\n| `self.helpers.parse_port_string(s)` | Parse port range string (e.g. `\"80,443,8000-9000\"`) |\n| `self.helpers.top_tcp_ports(n)` | Get top N TCP ports |\n\nThere are hundreds more in `bbot/core/helpers/misc.py`. Browse them before writing utility code yourself.\n\n---\n\n## Writing a Module\n\n### Quick Start\n\n1. Create `bbot/modules/my_module.py`\n2. Define a class that inherits from `BaseModule`\n3. Set `watched_events`, `produced_events`, `flags`, and `meta`\n4. Implement `handle_event()`\n5. Create `bbot/test/test_step_2/module_tests/test_module_my_module.py`\n\nHere's a minimal module:\n\n```python\nfrom bbot.modules.base import BaseModule\n\n\nclass my_module(BaseModule):\n    watched_events = [\"DNS_NAME\"]\n    produced_events = [\"EMAIL_ADDRESS\"]\n    flags = [\"passive\", \"email-enum\"]\n    meta = {\n        \"description\": \"Query example.com for email addresses\",\n        \"created_date\": \"2025-01-01\",\n        \"author\": \"@you\",\n    }\n\n    async def handle_event(self, event):\n        url = f\"https://api.example.com/lookup?domain={event.data}\"\n        r = await self.helpers.request(url)\n        if r and r.status_code == 200:\n            for email in r.json().get(\"emails\", []):\n                await self.emit_event(\n                    email,\n                    \"EMAIL_ADDRESS\",\n                    parent=event,\n                    context=f\"{{module}} queried example.com and found {{event.type}}: {{event.data}}\",\n                )\n```\n\nAnd its test:\n\n```python\nfrom .base import ModuleTestBase\n\n\nclass TestMyModule(ModuleTestBase):\n    async def setup_after_prep(self, module_test):\n        module_test.blasthttp_mock.add_response(\n            url=\"https://api.example.com/lookup?domain=blacklanternsecurity.com\",\n            json={\"emails\": [\"info@blacklanternsecurity.com\"]},\n        )\n\n    def check(self, module_test, events):\n        assert any(\n            e.data == \"info@blacklanternsecurity.com\" and e.type == \"EMAIL_ADDRESS\"\n            for e in events\n        ), \"Failed to find email\"\n```\n\n### Module Lifecycle\n\n```\nsetup()  -->  handle_event() (called many times)  -->  finish()  -->  report()  -->  cleanup()\n```\n\n1. **`setup()`** - one-time initialization (validate config, download data, check API keys)\n2. **`handle_event(event)`** - called for each matching event\n3. **`finish()`** - called when the scan is finishing; can still emit events\n4. **`report()`** - called once after finish; for summary output\n5. **`cleanup()`** - called last; close files, delete temp data; **cannot** emit events\n\n---\n\n### Module Attributes Reference\n\n#### Event Configuration\n\n##### `watched_events` (list)\nEvent types this module wants to process. The module's `handle_event()` is only called for these types.\n\n```python\n# sslcert.py - watches for open ports to grab SSL certs from\nwatched_events = [\"OPEN_TCP_PORT\"]\n\n# newsletters.py - watches HTTP responses to scan HTML\nwatched_events = [\"HTTP_RESPONSE\"]\n\n# json.py (output module) - watches everything\nwatched_events = [\"*\"]\n```\n\n##### `produced_events` (list)\nEvent types this module may emit. Used for dependency resolution and documentation.\n\n```python\n# sslcert.py - can discover hostnames and emails from certificates\nproduced_events = [\"DNS_NAME\", \"EMAIL_ADDRESS\"]\n\n# portscan.py - finds open ports\nproduced_events = [\"OPEN_TCP_PORT\"]\n```\n\n##### `flags` (list)\nTags that describe the module's behavior. Must include at least one activity flag (`passive` or `active`). Must also include `safe`, `loud`, or `invasive` (or a combination of `loud` and `invasive`).\n\nCommon flags:\n- `passive` / `active` - whether the module touches the target directly\n- `safe` - non-intrusive and non-destructive\n- `loud` - generates a large amount of network traffic\n- `invasive` - intrusive or potentially destructive\n- `subdomain-enum` - participates in subdomain enumeration\n- `web` - basic web scanning\n- `email-enum` - email discovery\n\n```python\n# crt.py - queries a third-party API, never touches the target\nflags = [\"subdomain-enum\", \"passive\"]\n\n# sslcert.py - connects directly to target ports\nflags = [\"affiliates\", \"subdomain-enum\", \"email-enum\", \"active\", \"web\"]\n```\n\n##### `meta` (dict)\nModule metadata. Must include `description`, `created_date`, and `author`. Set `auth_required: True` if the module needs an API key.\n\n```python\nmeta = {\n    \"description\": \"Query crt.sh (certificate transparency) for subdomains\",\n    \"created_date\": \"2022-05-13\",\n    \"author\": \"@TheTechromancer\",\n}\n\n# For API-key modules:\nmeta = {\"description\": \"Query API for subdomains\", \"auth_required\": True}\n```\n\n---\n\n#### Options\n\n##### `options` / `options_desc` (dict)\nUser-configurable settings. Access them via `self.config.get(\"option_name\")`.\n\n```python\n# robots.py - configurable parsing options\noptions = {\"include_sitemap\": False, \"include_allow\": True, \"include_disallow\": True}\noptions_desc = {\n    \"include_sitemap\": \"Include 'sitemap' entries\",\n    \"include_allow\": \"Include 'Allow' Entries\",\n    \"include_disallow\": \"Include 'Disallow' Entries\",\n}\n\n# In handle_event():\nif self.config.get(\"include_sitemap\") is True:\n    ...\n```\n\n```python\n# sslcert.py - timeout and behavior options\noptions = {\"timeout\": 5.0, \"skip_non_ssl\": True}\noptions_desc = {\"timeout\": \"Socket connect timeout in seconds\", \"skip_non_ssl\": \"Don't try common non-SSL ports\"}\n```\n\n---\n\n#### Scope & Filtering\n\n##### `scope_distance_modifier` (int or None) -- default: `0`\nControls which events the module accepts based on how far they are from the target.\n\n- `0` (default) - accept events up to the scan's configured search distance\n- `1` - accept events up to search distance + 1\n- `None` - accept all events regardless of distance\n\n```python\n# sslcert.py - looks one hop beyond normal scope, because certificate names\n# found on in-scope hosts often reveal related infrastructure\nscope_distance_modifier = 1\n```\n\n##### `in_scope_only` (bool) -- default: `False`\nOnly accept events that are explicitly in-scope (distance == 0). More restrictive than `scope_distance_modifier = 0`.\n\n```python\n# robots.py - only fetch robots.txt for in-scope hosts\nin_scope_only = True\n```\n\n##### `target_only` (bool) -- default: `False`\nOnly accept the initial target/seed events. Useful for modules that should only run once against the original targets.\n\n##### `accept_seeds` (bool) -- default: `True` for passive, `False` for active\nWhether to process seed events (the initial targets provided to the scan).\n\n##### `accept_url_special` (bool) -- default: `False`\nWhether to accept \"special\" URLs (e.g. JavaScript files) that are not normally distributed to web modules.\n\n```python\n# http.py - needs to process all URLs including special ones\naccept_url_special = True\n```\n\n---\n\n#### Deduplication\n\n##### `accept_dupes` (bool) -- default: `False`\nWhether to accept the same event more than once. Most modules should leave this `False`.\n\n```python\n# Output modules set this True because they need to see every event\naccept_dupes = True\n```\n\n##### `suppress_dupes` (bool) -- default: `True`\nWhether to suppress duplicate *outgoing* events. Prevents the same event from being emitted twice.\n\n##### `per_host_only` (bool) -- default: `False`\nOnly process one event per unique host. After processing `1.2.3.4`, skip any future events for `1.2.3.4`.\n\n##### `per_hostport_only` (bool) -- default: `False`\nOnly process one event per unique host:port combination.\n\n```python\n# robots.py - only fetch robots.txt once per host:port\nper_hostport_only = True\n```\n\n##### `per_domain_only` (bool) -- default: `False`\nOnly process one event per unique root domain. After processing `www.example.com`, skip `api.example.com`.\n\n```python\n# emailformat.py - one API query per domain is enough\nper_domain_only = True\n```\n\n##### `_incoming_dedup_hash(self, event)` -- override for custom dedup\nOverride this to define custom deduplication logic. Return a hash (int) or `(hash, reason_string)`.\n\n```python\n# securitytxt.py - dedupe by parent domain so we only check security.txt once\n# per parent domain, not once per subdomain\ndef _incoming_dedup_hash(self, event):\n    parent_domain = self.helpers.parent_domain(event.data)\n    return hash(parent_domain), \"already processed parent domain\"\n```\n\n```python\n# subdomain_enum.py template - dedupe by highest or lowest parent domain\ndef _incoming_dedup_hash(self, event):\n    return hash(self.make_query(event)), f\"dedup_strategy={self.dedup_strategy}\"\n```\n\n---\n\n#### Concurrency & Batching\n\n##### `_module_threads` (int) -- default: `1`\nHow many `handle_event()` calls can run concurrently. Increase this for I/O-bound modules.\n\n```python\n# sslcert.py - connects to many hosts in parallel\n_module_threads = 25\n```\n\n##### `_batch_size` (int) -- default: `1`\nWhen > 1, events are collected into batches and passed to `handle_batch(*events)` instead of `handle_event()`. Useful for tools that work better with bulk input.\n\n```python\n# portscan.py - masscan is most efficient with all targets at once\nbatch_size = 1000000\n\nasync def handle_batch(self, *events):\n    targets, correlator = await self.make_targets(events, self.syn_scanned)\n    async for ip, port, parent_event in self.masscan(targets, correlator):\n        await self.emit_open_port(ip, port, parent_event)\n```\n\n##### `_shuffle_incoming_queue` (bool) -- default: `True`\nWhether to randomize the order of incoming events. Set to `False` when order matters.\n\n```python\n# portscan.py - processes all events together, order doesn't matter but\n# we disable shuffle because batch_size is huge\n_shuffle_incoming_queue = False\n```\n\n---\n\n#### Dependencies\n\n##### `deps_pip` (list)\nPython packages to install.\n\n```python\n# sslcert.py\ndeps_pip = [\"pyOpenSSL~=25.3.0\"]\n```\n\n##### `deps_apt` (list)\nSystem packages to install.\n\n```python\n# sslcert.py\ndeps_apt = [\"openssl\"]\n```\n\n##### `deps_modules` (list)\nOther BBOT modules that must be enabled for this module to work.\n\n##### `deps_shell` (list)\nShell commands to run for installation (uses Ansible's `shell` module).\n\n##### `deps_ansible` (list)\nAnsible tasks for complex dependency installation (downloading binaries, etc.).\n\n```python\n# fingerprintx.py - downloads a Go binary\ndeps_ansible = [\n    {\n        \"name\": \"Download fingerprintx\",\n        \"unarchive\": {\n            \"src\": \"https://github.com/.../fingerprintx_{version}_{platform}_{arch}.tar.gz\",\n            \"include\": \"fingerprintx\",\n            \"dest\": \"#{BBOT_TOOLS}\",\n            \"remote_src\": True,\n        },\n    },\n]\n```\n\n---\n\n#### Priority & Queue\n\n##### `_priority` (int) -- default: `3`\nModule priority from 1 (highest) to 5 (lowest). Lower-priority modules get events first.\n\n```python\n# sslcert.py - runs early because other modules depend on the hostnames it discovers\n_priority = 2\n```\n\n##### `_qsize` (int) -- default: `1000`\nOutgoing event queue size. A smaller queue creates backpressure that helps with rate limiting.\n\n```python\n# subdomain_enum.py template - small queue to combat API rate limiting\n_qsize = 10\n```\n\n##### `_preserve_graph` (bool) -- default: `False`\nAccept duplicate events that are needed for complete event chain construction. Only used by output modules.\n\n```python\n# json.py - needs complete event chains for accurate output\n_preserve_graph = True\n```\n\n##### `_stats_exclude` (bool) -- default: `False`\nExclude this module from scan statistics. Used by output and report modules.\n\n##### `_disable_auto_module_deps` (bool) -- default: `False`\nPrevent BBOT from automatically enabling dependency modules. For example, if your module watches `URL` events, BBOT normally auto-enables `http`. Set this to `True` to prevent that.\n\n---\n\n### Key Methods\n\n#### `handle_event(self, event)` -- the core method\n\nCalled once for each matching event. This is where your module does its work.\n\n```python\n# robots.py - fetch and parse robots.txt\nasync def handle_event(self, event):\n    host = f\"{event.parsed_url.scheme}://{event.parsed_url.netloc}/\"\n    url = f\"{host}robots.txt\"\n    result = await self.helpers.request(url)\n    if result:\n        body = result.text\n        if body:\n            for line in body.split(\"\\n\"):\n                if line.startswith(\"Disallow:\"):\n                    path = line.split(\": \", 1)[1].lstrip(\"/\")\n                    await self.emit_event(\n                        f\"{host}{path}\",\n                        \"URL_UNVERIFIED\",\n                        parent=event,\n                        tags=[\"spider-danger\"],\n                    )\n```\n\n#### `handle_batch(self, *events)` -- bulk processing\n\nUsed when `_batch_size > 1`. Receives multiple events at once.\n\n```python\n# portscan.py - bulk port scanning with masscan\nasync def handle_batch(self, *events):\n    targets, correlator = await self.make_targets(events, self.syn_scanned)\n    async for ip, port, parent_event in self.masscan(targets, correlator):\n        await self.emit_open_port(ip, port, parent_event)\n```\n\n#### `filter_event(self, event)` -- custom event filtering\n\nCalled before `handle_event()`. Return `True` to accept, `False` to reject, or `(False, \"reason\")` to reject with a logged reason.\n\n```python\n# sslcert.py - skip ports that don't typically use SSL\nasync def filter_event(self, event):\n    if self.skip_non_ssl and event.port in self.non_ssl_ports:\n        return False, f\"Port {event.port} doesn't typically use SSL\"\n    return True\n```\n\n```python\n# subdomain_enum.py template - reject wildcards and cloud resources\nasync def filter_event(self, event):\n    query = self.make_query(event)\n    is_wildcard = await self._is_wildcard(query)\n    if self.reject_wildcards and is_wildcard:\n        return False, \"Event is a wildcard domain\"\n    return True, \"\"\n```\n\n#### `setup(self)` -- one-time initialization\n\nReturn values:\n- `True` -- success\n- `(True, \"message\")` -- success with message\n- `None` or `(None, \"message\")` -- **soft fail**: module is disabled, scan continues\n- `False` or `(False, \"message\")` -- **hard fail**: scan aborts\n\n```python\n# portscan.py - validates config, checks masscan, checks IPv6 support\nasync def setup(self):\n    self.top_ports = self.config.get(\"top_ports\", 100)\n    self.rate = self.config.get(\"rate\", 300)\n    self.ports = self.config.get(\"ports\", \"\")\n    if self.ports:\n        try:\n            self.helpers.parse_port_string(self.ports)\n        except ValueError as e:\n            return False, f\"Error parsing ports '{self.ports}': {e}\"\n    # ...\n    return True\n```\n\n```python\n# subdomain_enum_apikey template - soft-fail if API key is missing\nasync def setup(self):\n    await super().setup()\n    return await self.require_api_key()\n    # Returns (None, \"No API key set\") if missing, disabling the module\n```\n\n#### `finish(self)` -- called when scan is finishing\n\nCan still emit events. May be called multiple times if new activity is detected.\n\n#### `report(self)` -- summary output\n\nCalled once after `finish()`. Use for generating tables or summary data.\n\n```python\n# asn.py - output ASN statistics\nasync def report(self):\n    self.log_table(table_data, headers=[\"ASN\", \"Subnet\", \"Count\"], table_name=\"asns\")\n```\n\n#### `cleanup(self)` -- resource cleanup\n\nCalled once at the very end. Close files, delete temp files. **Cannot emit events.**\n\n```python\n# json.py\nasync def cleanup(self):\n    if getattr(self, \"_file\", None) is not None:\n        with suppress(Exception):\n            self.file.close()\n```\n\n```python\n# portscan.py\nasync def cleanup(self):\n    with suppress(Exception):\n        self.exclude_file.unlink()\n```\n\n---\n\n### Emitting Events\n\n#### `emit_event(data, event_type, parent, **kwargs)`\n\nCreates and queues an event for processing by other modules.\n\n```python\n# Simple string event\nawait self.emit_event(\"sub.example.com\", \"DNS_NAME\", parent=event)\n\n# With context (used for discovery chain documentation)\nawait self.emit_event(\n    \"sub.example.com\",\n    \"DNS_NAME\",\n    parent=event,\n    context=f\"{{module}} queried crt.sh and found {{event.type}}: {{event.data}}\",\n)\n\n# With tags\nawait self.emit_event(url, \"URL_UNVERIFIED\", parent=event, tags=[\"spider-danger\"])\n\n# FINDING event (dict data)\nawait self.emit_event(\n    {\n        \"host\": str(event.host),\n        \"description\": \"Found something interesting\",\n        \"url\": event.data[\"url\"],\n        \"severity\": \"HIGH\",\n    },\n    \"FINDING\",\n    parent=event,\n)\n```\n\n#### `make_event(data, event_type, parent, **kwargs)`\n\nCreates an event without emitting it. Useful when you need to inspect or modify it first.\n\n```python\nssl_event = self.make_event(hostname, \"DNS_NAME\", parent=event, raise_error=True)\nif ssl_event:\n    await self.emit_event(ssl_event, tags=[\"affiliate\"])\n```\n\n---\n\n### API Helpers\n\n#### `api_request(url, **kwargs)`\n\nHTTP request with automatic retry, rate-limit handling (429), API key cycling, and failure tracking. After too many failures, the module enters error state.\n\n```python\nr = await self.api_request(\"https://api.example.com/search?q=test\")\nif r and r.status_code == 200:\n    data = r.json()\n```\n\n#### `require_api_key()`\n\nValidates that an API key is configured. Call in `setup()`.\n\n```python\nasync def setup(self):\n    return await self.require_api_key()\n```\n\n#### `api_page_iter(url, page_size=100, **kwargs)`\n\nAsync generator for paginated API results. URL can contain `{page}`, `{page_size}`, and `{offset}` placeholders.\n\n```python\nasync for page in self.api_page_iter(\n    \"https://api.example.com/search?q=test&page={page}&limit={page_size}\"\n):\n    if not page.get(\"results\"):\n        break\n    for result in page[\"results\"]:\n        await self.emit_event(result[\"hostname\"], \"DNS_NAME\", parent=event)\n```\n\n---\n\n### Running External Processes\n\n```python\n# Run a command and get the result\nresult = await self.run_process([\"nmap\", \"-p\", \"22,80\", target])\nif result.returncode == 0:\n    output = result.stdout\n\n# Stream output line-by-line (for long-running tools)\nasync for line in self.run_process_live([\"masscan\", \"-oJ\", \"-\", ...]):\n    data = json.loads(line)\n```\n\n---\n\n### Logging\n\n```python\nself.debug(\"Low-level detail\")        # only visible with -d flag\nself.verbose(\"Useful but not critical\") # visible with -v flag\nself.info(\"Standard info\")\nself.success(\"Something good happened\") # green\nself.warning(\"Something concerning\")    # orange\nself.error(\"Something failed\")          # red\n\n# \"Huge\" variants: entire line in bold color\nself.hugesuccess(\"Major discovery!\")\nself.hugewarning(\"Major concern!\")\n```\n\n---\n\n### Templates\n\nFor common patterns, inherit from a template instead of `BaseModule` directly. Templates live in `bbot/modules/templates/`:\n\n- **`subdomain_enum`** - passive subdomain enumeration via free API. Handles dedup, wildcard rejection, query building.\n- **`subdomain_enum_apikey`** - same as above but requires an API key.\n- **`shodan`** - Shodan API integration.\n- **`github`** - GitHub API integration.\n- **`censys`** - Censys API integration.\n- **`bucket`** - Cloud storage bucket enumeration.\n- **`webhook`** - Webhook output.\n\nExample: `crt.py` inherits from `subdomain_enum` and only needs to override the request/parse logic:\n\n```python\nfrom bbot.modules.templates.subdomain_enum import subdomain_enum\n\n\nclass crt(subdomain_enum):\n    flags = [\"subdomain-enum\", \"passive\"]\n    watched_events = [\"DNS_NAME\"]\n    produced_events = [\"DNS_NAME\"]\n    meta = {\n        \"description\": \"Query crt.sh (certificate transparency) for subdomains\",\n        \"created_date\": \"2022-05-13\",\n        \"author\": \"@TheTechromancer\",\n    }\n    base_url = \"https://crt.sh\"\n\n    async def request_url(self, query):\n        params = {\"q\": f\"%.{query}\", \"output\": \"json\"}\n        url = self.helpers.add_get_params(self.base_url, params).geturl()\n        return await self.api_request(url, timeout=self.http_timeout + 30)\n\n    async def parse_results(self, r, query):\n        results = set()\n        for cert_info in r.json():\n            domain = cert_info.get(\"name_value\")\n            if domain:\n                for d in domain.splitlines():\n                    results.add(d.lower())\n        return results\n```\n\n---\n\n### Module Types\n\n#### Scan Modules (default)\nNormal modules that watch events and produce new ones. This is what you'll write 95% of the time.\n\n#### Output Modules\nInherit from `BaseOutputModule`. Receive all events and write them somewhere.\n\n```python\nfrom bbot.modules.output.base import BaseOutputModule\n\nclass my_output(BaseOutputModule):\n    watched_events = [\"*\"]\n    meta = {\"description\": \"Custom output\"}\n    _preserve_graph = True  # maintain complete event chains\n\n    async def handle_event(self, event):\n        # write event to file, database, API, etc.\n        ...\n```\n\nOutput modules automatically get:\n- `accept_dupes = True`\n- `scope_distance_modifier = None` (see all events)\n- `_stats_exclude = True`\n\n#### Internal Modules\nInherit from `BaseInternalModule`. System-level modules that aren't exposed to users.\n\n#### Intercept Modules\nInherit from `BaseInterceptModule`. Special high-priority modules that can modify or reject events before they reach normal modules. Used for DNS resolution, cloud detection, etc. You probably don't need to write one.\n\n---\n\n### Writing Tests\n\nEvery module needs a test in `bbot/test/test_step_2/module_tests/`. The test file must be named `test_module_<name>.py`.\n\nTest classes inherit from `ModuleTestBase` and follow this pattern:\n\n```python\nfrom .base import ModuleTestBase\n\n\nclass TestMyModule(ModuleTestBase):\n    # Optional: override targets (default: [\"blacklanternsecurity.com\"])\n    targets = [\"http://127.0.0.1:8888\"]\n\n    # Optional: override which modules are enabled\n    modules_overrides = [\"http\", \"my_module\"]\n\n    # Optional: override config\n    config_overrides = {\"modules\": {\"my_module\": {\"some_option\": True}}}\n\n    async def setup_before_prep(self, module_test):\n        \"\"\"Called BEFORE the scan is prepared. Set up HTTP mocks here.\"\"\"\n        pass\n\n    async def setup_after_prep(self, module_test):\n        \"\"\"Called AFTER the scan is prepared. Modify modules, add mocks here.\"\"\"\n        # Mock an HTTP response\n        module_test.blasthttp_mock.add_response(\n            url=\"https://api.example.com/lookup?domain=blacklanternsecurity.com\",\n            json={\"results\": [\"sub.blacklanternsecurity.com\"]},\n        )\n\n        # Mock DNS\n        await module_test.mock_dns({\n            \"blacklanternsecurity.com\": {\"A\": [\"127.0.0.88\"]},\n        })\n\n        # Mock an HTTP server response\n        module_test.set_expect_requests(\n            expect_args={\"method\": \"GET\", \"uri\": \"/robots.txt\"},\n            respond_args={\"response_data\": \"Disallow: /secret/\"},\n        )\n\n    def check(self, module_test, events):\n        \"\"\"Verify the scan produced the expected events.\"\"\"\n        assert any(\n            e.data == \"sub.blacklanternsecurity.com\" and e.type == \"DNS_NAME\"\n            for e in events\n        ), \"Failed to find subdomain\"\n```\n\nThe test lifecycle runs:\n1. `setup_before_prep()` - set up mocks\n2. Scan `_prep()` - loads modules, config\n3. `setup_after_prep()` - modify scan state\n4. Scan runs and collects events\n5. `check()` - your assertions\n\n### Test Utilities\n\n- **`module_test.blasthttp_mock`** - mock HTTP responses \n- **`module_test.httpserver`** - real HTTP server on port 8888\n- **`module_test.httpserver_ssl`** - real HTTPS server on port 9999\n- **`module_test.mock_dns(data)`** - mock DNS responses\n- **`module_test.mock_interactsh(name)`** - mock out-of-band interactions\n- **`module_test.module`** - reference to the module instance being tested\n- **`module_test.scan`** - reference to the Scanner instance\n\nReal example -- `test_module_robots.py`:\n\n```python\nclass TestRobots(ModuleTestBase):\n    targets = [\"http://127.0.0.1:8888\"]\n    modules_overrides = [\"http\", \"robots\"]\n    config_overrides = {\"modules\": {\"robots\": {\"include_sitemap\": True}}}\n\n    async def setup_after_prep(self, module_test):\n        robots = \"Allow: /allow/\\nDisallow: /disallow/\\nSitemap: http://127.0.0.1:8888/sitemap.txt\"\n        module_test.set_expect_requests(\n            expect_args={\"method\": \"GET\", \"uri\": \"/robots.txt\"},\n            respond_args={\"response_data\": robots},\n        )\n\n    def check(self, module_test, events):\n        assert any(e.data == \"http://127.0.0.1:8888/allow/\" for e in events)\n        assert any(e.data == \"http://127.0.0.1:8888/disallow/\" for e in events)\n        assert any(e.data == \"http://127.0.0.1:8888/sitemap.txt\" for e in events)\n```\n","CLAUDE.md":"Treat @AGENTS.md the same way you'd treat CLAUDE.md."},"items":[{"name":"AGENTS.md","path":"AGENTS.md","title":"AGENTS.md","content":"# BBOT Developer Guide\n\n## Core Principles\n\n### Modularity Principle\nWhen writing a BBOT module, make sure all module-specific code lives in the module itself. Don't hard-code module-specific things in core or in helpers.\n\n### DRY Principle\nDon't Repeat Yourself -- and interpret this broadly. If two pieces of code aren't identical but follow a similar enough pattern that they could be generalized, they should be. Extract shared logic into a common abstraction rather than duplicating the pattern. Usually this means creating a shared helper, or a shared module template in `bbot/modules/templates`. When you notice structural similarity, unify it.\n\n### Engineering Principle\nEvery system that is implemented must be implemented properly. No hacks, no hardcoding, no shortcuts. If we implement one of something, we build a proper system for it. It's okay to take a step back from the current task, in order to do things right. This relates directly to the Modularity Principle above.\n\n### Testing Principle\nBBOT has extremely thorough tests, including **one or more individual tests for each module, with no exceptions**. This is critical to maintaining stability in a recursive tool, which by its nature flirts with race conditions and infinite loops. If you add a module, you write a test. If you change a module, you make sure its test still passes.\n\n---\n\n## Tooling\n\n- **Package manager**: [uv](https://docs.astral.sh/uv/)\n- **Linter/formatter**: [ruff](https://docs.astral.sh/ruff/) (pinned to 0.15.10)\n- **Test framework**: [pytest](https://docs.pytest.org/) with pytest-asyncio\n- **Python**: 3.10 - 3.14\n\n---\n\n## Dev Environment Setup\n\n```bash\n# 1. Fork and clone\ngit clone git@github.com:<you>/bbot.git\ncd bbot\n\n# 2. Switch to dev branch, then create a feature branch\ngit checkout dev\ngit checkout -b my-feature\n\n# 3. Install uv (if you haven't already)\ncurl -LsSf https://astral.sh/uv/install.sh | sh\n\n# 4. Install all dependencies (including dev)\nuv sync --group dev\n\n# 5. Install pre-commit hooks (ruff, file checks, etc.)\nuv run pre-commit install\n\n# 6. Activate the virtualenv\nsource .venv/bin/activate\n\n# 7. Verify\nbbot --help\n```\n\n### Running Tests\n\n```bash\n# Run the full suite\n./bbot/test/run_tests.sh\n\n# Run specific module tests\n./bbot/test/run_tests.sh robots,sslcert\n\n# Run a single test file directly\npytest bbot/test/test_step_2/module_tests/test_module_robots.py -x -vv\n```\n\n### Linting\n\n```bash\nruff check          # lint\nruff format          # auto-format\nruff format --check  # verify formatting without changes\n```\n\n### Git Workflow\n\n- `stable` - production releases\n- `dev` - active development, **almost all PRs should target this branch**\n- Feature branches should be created from `dev`\n\n---\n\n## AI Use Disclosure\n\nUse of AI is not prohibited -- and in many cases, encouraged. However, when reviewing a PR, it is helpful for the reviewer to know the extent to which AI was used, and which model.\nPlease add a small section at the bottom of the PR with the header: `### AI Use Disclosure`, followed by the following information:\n\n* Extent of the AI use. For example, was this fully autonomous by the AI, or was it a collaborative back-and-forth, or did the user just use the AI to review their work, etc.\n* Model Used\n\nThis should only apply to external contributors, not members of the blacklanternsecurity organization.\n\n---\n\n## Architecture Overview\n\n### How a Scan Works\n\nBBOT is an async, recursive OSINT tool. A scan starts with **seed events** (targets) and passes them through a pipeline of **modules**. Each module watches for specific event types, processes them, and may emit new events, which feed back into the pipeline. This continues until no module has anything left to do.\n\n```\n                     Seeds (targets)\n                          |\n                          v\n                   +--------------+\n                   | ScanIngress  |  dedup, blacklist, scope check\n                   +--------------+\n                          |\n                          v\n                   +--------------+\n                   |  Intercept   |  dns, cloud tagging\n                   |   Modules    |  (modify/tag events before distribution)\n                   +--------------+\n                          |\n                          v\n                   +--------------+\n                   |  ScanEgress  |  scope filtering, graph management\n                   +--------------+\n                          |\n                          v\n              +-----------+-----------+\n              |           |           |\n           Module A    Module B    Module C   ...\n              |           |           |\n              +-----------+-----------+\n                          |\n                          v\n                   Output Modules (json, csv, neo4j, ...)\n```\n\n### Events\n\nEvents are the currency of BBOT. Every piece of data -- a hostname, IP, URL, open port, finding -- is an event. Events have:\n\n- **type**: `DNS_NAME`, `IP_ADDRESS`, `URL`, `OPEN_TCP_PORT`, `HTTP_RESPONSE`, `FINDING`, `EMAIL_ADDRESS`, etc.\n- **data**: the actual data (a string, dict, etc.)\n- **parent**: the event that led to this one (forming a discovery chain)\n- **scope_distance**: how many hops from the original target (0 = in-scope)\n- **tags**: metadata like `in-scope`, `affiliate`, `cloud-azure`, `open-port`, etc.\n- **module**: which module discovered it\n\n### Scope Distance\n\nScope distance tracks how far an event is from the original target:\n- `0` = explicitly in-scope (matches target or discovered in-scope)\n- `1` = one hop away (e.g. a hostname found in an SSL cert of an in-scope host)\n- `2+` = further away\n\nThe scan's `scope.search_distance` (default 0) controls how far modules are allowed to look. A module's `scope_distance_modifier` adjusts this per-module.\n\n### Helpers\n\nBBOT has a helper for almost everything. **Please use them.** They're accessible via `self.helpers` inside any module.\n\nKey helpers:\n\n| Helper | What it does |\n|--------|-------------|\n| `self.helpers.request(url)` | Make an HTTP request (with retries, SSL handling, etc.) |\n| `self.helpers.blasthttp` | Shared blasthttp client (rate-limited via `web.http_rate_limit` config) |\n| `self.helpers.resolve(host)` | DNS resolution |\n| `self.helpers.is_ip(s)` | Check if string is an IP |\n| `self.helpers.is_dns_name(s)` | Check if string is a hostname |\n| `self.helpers.split_domain(host)` | Split into subdomain + root domain |\n| `self.helpers.domain_parents(domain)` | Get all parent domains |\n| `self.helpers.make_netloc(host, port)` | Format `host:port` (handles IPv6) |\n| `self.helpers.parent_domain(domain)` | Get immediate parent domain |\n| `self.helpers.beautifulsoup(html, parser)` | Parse HTML |\n| `self.helpers.validators.validate_host(h)` | Validate and normalize a hostname |\n| `self.helpers.tempfile(data, pipe=False)` | Create a temp file with content |\n| `self.helpers.run(command)` | Run a shell command |\n| `self.helpers.run_live(command)` | Run a shell command, stream output |\n| `self.helpers.as_completed(tasks)` | Async iteration of completed tasks |\n| `self.helpers.wordlist(url_or_path)` | Download/cache a wordlist |\n| `self.helpers.rand_string(n)` | Random string of length n |\n| `self.helpers.regexes.email_regex` | Pre-compiled email regex |\n| `self.helpers.add_get_params(url, params)` | Add query params to a URL |\n| `self.helpers.quote(s)` | URL-encode a string |\n| `self.helpers.make_ip_type(s)` | Convert string to `ipaddress` object |\n| `self.helpers.parse_port_string(s)` | Parse port range string (e.g. `\"80,443,8000-9000\"`) |\n| `self.helpers.top_tcp_ports(n)` | Get top N TCP ports |\n\nThere are hundreds more in `bbot/core/helpers/misc.py`. Browse them before writing utility code yourself.\n\n---\n\n## Writing a Module\n\n### Quick Start\n\n1. Create `bbot/modules/my_module.py`\n2. Define a class that inherits from `BaseModule`\n3. Set `watched_events`, `produced_events`, `flags`, and `meta`\n4. Implement `handle_event()`\n5. Create `bbot/test/test_step_2/module_tests/test_module_my_module.py`\n\nHere's a minimal module:\n\n```python\nfrom bbot.modules.base import BaseModule\n\n\nclass my_module(BaseModule):\n    watched_events = [\"DNS_NAME\"]\n    produced_events = [\"EMAIL_ADDRESS\"]\n    flags = [\"passive\", \"email-enum\"]\n    meta = {\n        \"description\": \"Query example.com for email addresses\",\n        \"created_date\": \"2025-01-01\",\n        \"author\": \"@you\",\n    }\n\n    async def handle_event(self, event):\n        url = f\"https://api.example.com/lookup?domain={event.data}\"\n        r = await self.helpers.request(url)\n        if r and r.status_code == 200:\n            for email in r.json().get(\"emails\", []):\n                await self.emit_event(\n                    email,\n                    \"EMAIL_ADDRESS\",\n                    parent=event,\n                    context=f\"{{module}} queried example.com and found {{event.type}}: {{event.data}}\",\n                )\n```\n\nAnd its test:\n\n```python\nfrom .base import ModuleTestBase\n\n\nclass TestMyModule(ModuleTestBase):\n    async def setup_after_prep(self, module_test):\n        module_test.blasthttp_mock.add_response(\n            url=\"https://api.example.com/lookup?domain=blacklanternsecurity.com\",\n            json={\"emails\": [\"info@blacklanternsecurity.com\"]},\n        )\n\n    def check(self, module_test, events):\n        assert any(\n            e.data == \"info@blacklanternsecurity.com\" and e.type == \"EMAIL_ADDRESS\"\n            for e in events\n        ), \"Failed to find email\"\n```\n\n### Module Lifecycle\n\n```\nsetup()  -->  handle_event() (called many times)  -->  finish()  -->  report()  -->  cleanup()\n```\n\n1. **`setup()`** - one-time initialization (validate config, download data, check API keys)\n2. **`handle_event(event)`** - called for each matching event\n3. **`finish()`** - called when the scan is finishing; can still emit events\n4. **`report()`** - called once after finish; for summary output\n5. **`cleanup()`** - called last; close files, delete temp data; **cannot** emit events\n\n---\n\n### Module Attributes Reference\n\n#### Event Configuration\n\n##### `watched_events` (list)\nEvent types this module wants to process. The module's `handle_event()` is only called for these types.\n\n```python\n# sslcert.py - watches for open ports to grab SSL certs from\nwatched_events = [\"OPEN_TCP_PORT\"]\n\n# newsletters.py - watches HTTP responses to scan HTML\nwatched_events = [\"HTTP_RESPONSE\"]\n\n# json.py (output module) - watches everything\nwatched_events = [\"*\"]\n```\n\n##### `produced_events` (list)\nEvent types this module may emit. Used for dependency resolution and documentation.\n\n```python\n# sslcert.py - can discover hostnames and emails from certificates\nproduced_events = [\"DNS_NAME\", \"EMAIL_ADDRESS\"]\n\n# portscan.py - finds open ports\nproduced_events = [\"OPEN_TCP_PORT\"]\n```\n\n##### `flags` (list)\nTags that describe the module's behavior. Must include at least one activity flag (`passive` or `active`). Must also include `safe`, `loud`, or `invasive` (or a combination of `loud` and `invasive`).\n\nCommon flags:\n- `passive` / `active` - whether the module touches the target directly\n- `safe` - non-intrusive and non-destructive\n- `loud` - generates a large amount of network traffic\n- `invasive` - intrusive or potentially destructive\n- `subdomain-enum` - participates in subdomain enumeration\n- `web` - basic web scanning\n- `email-enum` - email discovery\n\n```python\n# crt.py - queries a third-party API, never touches the target\nflags = [\"subdomain-enum\", \"passive\"]\n\n# sslcert.py - connects directly to target ports\nflags = [\"affiliates\", \"subdomain-enum\", \"email-enum\", \"active\", \"web\"]\n```\n\n##### `meta` (dict)\nModule metadata. Must include `description`, `created_date`, and `author`. Set `auth_required: True` if the module needs an API key.\n\n```python\nmeta = {\n    \"description\": \"Query crt.sh (certificate transparency) for subdomains\",\n    \"created_date\": \"2022-05-13\",\n    \"author\": \"@TheTechromancer\",\n}\n\n# For API-key modules:\nmeta = {\"description\": \"Query API for subdomains\", \"auth_required\": True}\n```\n\n---\n\n#### Options\n\n##### `options` / `options_desc` (dict)\nUser-configurable settings. Access them via `self.config.get(\"option_name\")`.\n\n```python\n# robots.py - configurable parsing options\noptions = {\"include_sitemap\": False, \"include_allow\": True, \"include_disallow\": True}\noptions_desc = {\n    \"include_sitemap\": \"Include 'sitemap' entries\",\n    \"include_allow\": \"Include 'Allow' Entries\",\n    \"include_disallow\": \"Include 'Disallow' Entries\",\n}\n\n# In handle_event():\nif self.config.get(\"include_sitemap\") is True:\n    ...\n```\n\n```python\n# sslcert.py - timeout and behavior options\noptions = {\"timeout\": 5.0, \"skip_non_ssl\": True}\noptions_desc = {\"timeout\": \"Socket connect timeout in seconds\", \"skip_non_ssl\": \"Don't try common non-SSL ports\"}\n```\n\n---\n\n#### Scope & Filtering\n\n##### `scope_distance_modifier` (int or None) -- default: `0`\nControls which events the module accepts based on how far they are from the target.\n\n- `0` (default) - accept events up to the scan's configured search distance\n- `1` - accept events up to search distance + 1\n- `None` - accept all events regardless of distance\n\n```python\n# sslcert.py - looks one hop beyond normal scope, because certificate names\n# found on in-scope hosts often reveal related infrastructure\nscope_distance_modifier = 1\n```\n\n##### `in_scope_only` (bool) -- default: `False`\nOnly accept events that are explicitly in-scope (distance == 0). More restrictive than `scope_distance_modifier = 0`.\n\n```python\n# robots.py - only fetch robots.txt for in-scope hosts\nin_scope_only = True\n```\n\n##### `target_only` (bool) -- default: `False`\nOnly accept the initial target/seed events. Useful for modules that should only run once against the original targets.\n\n##### `accept_seeds` (bool) -- default: `True` for passive, `False` for active\nWhether to process seed events (the initial targets provided to the scan).\n\n##### `accept_url_special` (bool) -- default: `False`\nWhether to accept \"special\" URLs (e.g. JavaScript files) that are not normally distributed to web modules.\n\n```python\n# http.py - needs to process all URLs including special ones\naccept_url_special = True\n```\n\n---\n\n#### Deduplication\n\n##### `accept_dupes` (bool) -- default: `False`\nWhether to accept the same event more than once. Most modules should leave this `False`.\n\n```python\n# Output modules set this True because they need to see every event\naccept_dupes = True\n```\n\n##### `suppress_dupes` (bool) -- default: `True`\nWhether to suppress duplicate *outgoing* events. Prevents the same event from being emitted twice.\n\n##### `per_host_only` (bool) -- default: `False`\nOnly process one event per unique host. After processing `1.2.3.4`, skip any future events for `1.2.3.4`.\n\n##### `per_hostport_only` (bool) -- default: `False`\nOnly process one event per unique host:port combination.\n\n```python\n# robots.py - only fetch robots.txt once per host:port\nper_hostport_only = True\n```\n\n##### `per_domain_only` (bool) -- default: `False`\nOnly process one event per unique root domain. After processing `www.example.com`, skip `api.example.com`.\n\n```python\n# emailformat.py - one API query per domain is enough\nper_domain_only = True\n```\n\n##### `_incoming_dedup_hash(self, event)` -- override for custom dedup\nOverride this to define custom deduplication logic. Return a hash (int) or `(hash, reason_string)`.\n\n```python\n# securitytxt.py - dedupe by parent domain so we only check security.txt once\n# per parent domain, not once per subdomain\ndef _incoming_dedup_hash(self, event):\n    parent_domain = self.helpers.parent_domain(event.data)\n    return hash(parent_domain), \"already processed parent domain\"\n```\n\n```python\n# subdomain_enum.py template - dedupe by highest or lowest parent domain\ndef _incoming_dedup_hash(self, event):\n    return hash(self.make_query(event)), f\"dedup_strategy={self.dedup_strategy}\"\n```\n\n---\n\n#### Concurrency & Batching\n\n##### `_module_threads` (int) -- default: `1`\nHow many `handle_event()` calls can run concurrently. Increase this for I/O-bound modules.\n\n```python\n# sslcert.py - connects to many hosts in parallel\n_module_threads = 25\n```\n\n##### `_batch_size` (int) -- default: `1`\nWhen > 1, events are collected into batches and passed to `handle_batch(*events)` instead of `handle_event()`. Useful for tools that work better with bulk input.\n\n```python\n# portscan.py - masscan is most efficient with all targets at once\nbatch_size = 1000000\n\nasync def handle_batch(self, *events):\n    targets, correlator = await self.make_targets(events, self.syn_scanned)\n    async for ip, port, parent_event in self.masscan(targets, correlator):\n        await self.emit_open_port(ip, port, parent_event)\n```\n\n##### `_shuffle_incoming_queue` (bool) -- default: `True`\nWhether to randomize the order of incoming events. Set to `False` when order matters.\n\n```python\n# portscan.py - processes all events together, order doesn't matter but\n# we disable shuffle because batch_size is huge\n_shuffle_incoming_queue = False\n```\n\n---\n\n#### Dependencies\n\n##### `deps_pip` (list)\nPython packages to install.\n\n```python\n# sslcert.py\ndeps_pip = [\"pyOpenSSL~=25.3.0\"]\n```\n\n##### `deps_apt` (list)\nSystem packages to install.\n\n```python\n# sslcert.py\ndeps_apt = [\"openssl\"]\n```\n\n##### `deps_modules` (list)\nOther BBOT modules that must be enabled for this module to work.\n\n##### `deps_shell` (list)\nShell commands to run for installation (uses Ansible's `shell` module).\n\n##### `deps_ansible` (list)\nAnsible tasks for complex dependency installation (downloading binaries, etc.).\n\n```python\n# fingerprintx.py - downloads a Go binary\ndeps_ansible = [\n    {\n        \"name\": \"Download fingerprintx\",\n        \"unarchive\": {\n            \"src\": \"https://github.com/.../fingerprintx_{version}_{platform}_{arch}.tar.gz\",\n            \"include\": \"fingerprintx\",\n            \"dest\": \"#{BBOT_TOOLS}\",\n            \"remote_src\": True,\n        },\n    },\n]\n```\n\n---\n\n#### Priority & Queue\n\n##### `_priority` (int) -- default: `3`\nModule priority from 1 (highest) to 5 (lowest). Lower-priority modules get events first.\n\n```python\n# sslcert.py - runs early because other modules depend on the hostnames it discovers\n_priority = 2\n```\n\n##### `_qsize` (int) -- default: `1000`\nOutgoing event queue size. A smaller queue creates backpressure that helps with rate limiting.\n\n```python\n# subdomain_enum.py template - small queue to combat API rate limiting\n_qsize = 10\n```\n\n##### `_preserve_graph` (bool) -- default: `False`\nAccept duplicate events that are needed for complete event chain construction. Only used by output modules.\n\n```python\n# json.py - needs complete event chains for accurate output\n_preserve_graph = True\n```\n\n##### `_stats_exclude` (bool) -- default: `False`\nExclude this module from scan statistics. Used by output and report modules.\n\n##### `_disable_auto_module_deps` (bool) -- default: `False`\nPrevent BBOT from automatically enabling dependency modules. For example, if your module watches `URL` events, BBOT normally auto-enables `http`. Set this to `True` to prevent that.\n\n---\n\n### Key Methods\n\n#### `handle_event(self, event)` -- the core method\n\nCalled once for each matching event. This is where your module does its work.\n\n```python\n# robots.py - fetch and parse robots.txt\nasync def handle_event(self, event):\n    host = f\"{event.parsed_url.scheme}://{event.parsed_url.netloc}/\"\n    url = f\"{host}robots.txt\"\n    result = await self.helpers.request(url)\n    if result:\n        body = result.text\n        if body:\n            for line in body.split(\"\\n\"):\n                if line.startswith(\"Disallow:\"):\n                    path = line.split(\": \", 1)[1].lstrip(\"/\")\n                    await self.emit_event(\n                        f\"{host}{path}\",\n                        \"URL_UNVERIFIED\",\n                        parent=event,\n                        tags=[\"spider-danger\"],\n                    )\n```\n\n#### `handle_batch(self, *events)` -- bulk processing\n\nUsed when `_batch_size > 1`. Receives multiple events at once.\n\n```python\n# portscan.py - bulk port scanning with masscan\nasync def handle_batch(self, *events):\n    targets, correlator = await self.make_targets(events, self.syn_scanned)\n    async for ip, port, parent_event in self.masscan(targets, correlator):\n        await self.emit_open_port(ip, port, parent_event)\n```\n\n#### `filter_event(self, event)` -- custom event filtering\n\nCalled before `handle_event()`. Return `True` to accept, `False` to reject, or `(False, \"reason\")` to reject with a logged reason.\n\n```python\n# sslcert.py - skip ports that don't typically use SSL\nasync def filter_event(self, event):\n    if self.skip_non_ssl and event.port in self.non_ssl_ports:\n        return False, f\"Port {event.port} doesn't typically use SSL\"\n    return True\n```\n\n```python\n# subdomain_enum.py template - reject wildcards and cloud resources\nasync def filter_event(self, event):\n    query = self.make_query(event)\n    is_wildcard = await self._is_wildcard(query)\n    if self.reject_wildcards and is_wildcard:\n        return False, \"Event is a wildcard domain\"\n    return True, \"\"\n```\n\n#### `setup(self)` -- one-time initialization\n\nReturn values:\n- `True` -- success\n- `(True, \"message\")` -- success with message\n- `None` or `(None, \"message\")` -- **soft fail**: module is disabled, scan continues\n- `False` or `(False, \"message\")` -- **hard fail**: scan aborts\n\n```python\n# portscan.py - validates config, checks masscan, checks IPv6 support\nasync def setup(self):\n    self.top_ports = self.config.get(\"top_ports\", 100)\n    self.rate = self.config.get(\"rate\", 300)\n    self.ports = self.config.get(\"ports\", \"\")\n    if self.ports:\n        try:\n            self.helpers.parse_port_string(self.ports)\n        except ValueError as e:\n            return False, f\"Error parsing ports '{self.ports}': {e}\"\n    # ...\n    return True\n```\n\n```python\n# subdomain_enum_apikey template - soft-fail if API key is missing\nasync def setup(self):\n    await super().setup()\n    return await self.require_api_key()\n    # Returns (None, \"No API key set\") if missing, disabling the module\n```\n\n#### `finish(self)` -- called when scan is finishing\n\nCan still emit events. May be called multiple times if new activity is detected.\n\n#### `report(self)` -- summary output\n\nCalled once after `finish()`. Use for generating tables or summary data.\n\n```python\n# asn.py - output ASN statistics\nasync def report(self):\n    self.log_table(table_data, headers=[\"ASN\", \"Subnet\", \"Count\"], table_name=\"asns\")\n```\n\n#### `cleanup(self)` -- resource cleanup\n\nCalled once at the very end. Close files, delete temp files. **Cannot emit events.**\n\n```python\n# json.py\nasync def cleanup(self):\n    if getattr(self, \"_file\", None) is not None:\n        with suppress(Exception):\n            self.file.close()\n```\n\n```python\n# portscan.py\nasync def cleanup(self):\n    with suppress(Exception):\n        self.exclude_file.unlink()\n```\n\n---\n\n### Emitting Events\n\n#### `emit_event(data, event_type, parent, **kwargs)`\n\nCreates and queues an event for processing by other modules.\n\n```python\n# Simple string event\nawait self.emit_event(\"sub.example.com\", \"DNS_NAME\", parent=event)\n\n# With context (used for discovery chain documentation)\nawait self.emit_event(\n    \"sub.example.com\",\n    \"DNS_NAME\",\n    parent=event,\n    context=f\"{{module}} queried crt.sh and found {{event.type}}: {{event.data}}\",\n)\n\n# With tags\nawait self.emit_event(url, \"URL_UNVERIFIED\", parent=event, tags=[\"spider-danger\"])\n\n# FINDING event (dict data)\nawait self.emit_event(\n    {\n        \"host\": str(event.host),\n        \"description\": \"Found something interesting\",\n        \"url\": event.data[\"url\"],\n        \"severity\": \"HIGH\",\n    },\n    \"FINDING\",\n    parent=event,\n)\n```\n\n#### `make_event(data, event_type, parent, **kwargs)`\n\nCreates an event without emitting it. Useful when you need to inspect or modify it first.\n\n```python\nssl_event = self.make_event(hostname, \"DNS_NAME\", parent=event, raise_error=True)\nif ssl_event:\n    await self.emit_event(ssl_event, tags=[\"affiliate\"])\n```\n\n---\n\n### API Helpers\n\n#### `api_request(url, **kwargs)`\n\nHTTP request with automatic retry, rate-limit handling (429), API key cycling, and failure tracking. After too many failures, the module enters error state.\n\n```python\nr = await self.api_request(\"https://api.example.com/search?q=test\")\nif r and r.status_code == 200:\n    data = r.json()\n```\n\n#### `require_api_key()`\n\nValidates that an API key is configured. Call in `setup()`.\n\n```python\nasync def setup(self):\n    return await self.require_api_key()\n```\n\n#### `api_page_iter(url, page_size=100, **kwargs)`\n\nAsync generator for paginated API results. URL can contain `{page}`, `{page_size}`, and `{offset}` placeholders.\n\n```python\nasync for page in self.api_page_iter(\n    \"https://api.example.com/search?q=test&page={page}&limit={page_size}\"\n):\n    if not page.get(\"results\"):\n        break\n    for result in page[\"results\"]:\n        await self.emit_event(result[\"hostname\"], \"DNS_NAME\", parent=event)\n```\n\n---\n\n### Running External Processes\n\n```python\n# Run a command and get the result\nresult = await self.run_process([\"nmap\", \"-p\", \"22,80\", target])\nif result.returncode == 0:\n    output = result.stdout\n\n# Stream output line-by-line (for long-running tools)\nasync for line in self.run_process_live([\"masscan\", \"-oJ\", \"-\", ...]):\n    data = json.loads(line)\n```\n\n---\n\n### Logging\n\n```python\nself.debug(\"Low-level detail\")        # only visible with -d flag\nself.verbose(\"Useful but not critical\") # visible with -v flag\nself.info(\"Standard info\")\nself.success(\"Something good happened\") # green\nself.warning(\"Something concerning\")    # orange\nself.error(\"Something failed\")          # red\n\n# \"Huge\" variants: entire line in bold color\nself.hugesuccess(\"Major discovery!\")\nself.hugewarning(\"Major concern!\")\n```\n\n---\n\n### Templates\n\nFor common patterns, inherit from a template instead of `BaseModule` directly. Templates live in `bbot/modules/templates/`:\n\n- **`subdomain_enum`** - passive subdomain enumeration via free API. Handles dedup, wildcard rejection, query building.\n- **`subdomain_enum_apikey`** - same as above but requires an API key.\n- **`shodan`** - Shodan API integration.\n- **`github`** - GitHub API integration.\n- **`censys`** - Censys API integration.\n- **`bucket`** - Cloud storage bucket enumeration.\n- **`webhook`** - Webhook output.\n\nExample: `crt.py` inherits from `subdomain_enum` and only needs to override the request/parse logic:\n\n```python\nfrom bbot.modules.templates.subdomain_enum import subdomain_enum\n\n\nclass crt(subdomain_enum):\n    flags = [\"subdomain-enum\", \"passive\"]\n    watched_events = [\"DNS_NAME\"]\n    produced_events = [\"DNS_NAME\"]\n    meta = {\n        \"description\": \"Query crt.sh (certificate transparency) for subdomains\",\n        \"created_date\": \"2022-05-13\",\n        \"author\": \"@TheTechromancer\",\n    }\n    base_url = \"https://crt.sh\"\n\n    async def request_url(self, query):\n        params = {\"q\": f\"%.{query}\", \"output\": \"json\"}\n        url = self.helpers.add_get_params(self.base_url, params).geturl()\n        return await self.api_request(url, timeout=self.http_timeout + 30)\n\n    async def parse_results(self, r, query):\n        results = set()\n        for cert_info in r.json():\n            domain = cert_info.get(\"name_value\")\n            if domain:\n                for d in domain.splitlines():\n                    results.add(d.lower())\n        return results\n```\n\n---\n\n### Module Types\n\n#### Scan Modules (default)\nNormal modules that watch events and produce new ones. This is what you'll write 95% of the time.\n\n#### Output Modules\nInherit from `BaseOutputModule`. Receive all events and write them somewhere.\n\n```python\nfrom bbot.modules.output.base import BaseOutputModule\n\nclass my_output(BaseOutputModule):\n    watched_events = [\"*\"]\n    meta = {\"description\": \"Custom output\"}\n    _preserve_graph = True  # maintain complete event chains\n\n    async def handle_event(self, event):\n        # write event to file, database, API, etc.\n        ...\n```\n\nOutput modules automatically get:\n- `accept_dupes = True`\n- `scope_distance_modifier = None` (see all events)\n- `_stats_exclude = True`\n\n#### Internal Modules\nInherit from `BaseInternalModule`. System-level modules that aren't exposed to users.\n\n#### Intercept Modules\nInherit from `BaseInterceptModule`. Special high-priority modules that can modify or reject events before they reach normal modules. Used for DNS resolution, cloud detection, etc. You probably don't need to write one.\n\n---\n\n### Writing Tests\n\nEvery module needs a test in `bbot/test/test_step_2/module_tests/`. The test file must be named `test_module_<name>.py`.\n\nTest classes inherit from `ModuleTestBase` and follow this pattern:\n\n```python\nfrom .base import ModuleTestBase\n\n\nclass TestMyModule(ModuleTestBase):\n    # Optional: override targets (default: [\"blacklanternsecurity.com\"])\n    targets = [\"http://127.0.0.1:8888\"]\n\n    # Optional: override which modules are enabled\n    modules_overrides = [\"http\", \"my_module\"]\n\n    # Optional: override config\n    config_overrides = {\"modules\": {\"my_module\": {\"some_option\": True}}}\n\n    async def setup_before_prep(self, module_test):\n        \"\"\"Called BEFORE the scan is prepared. Set up HTTP mocks here.\"\"\"\n        pass\n\n    async def setup_after_prep(self, module_test):\n        \"\"\"Called AFTER the scan is prepared. Modify modules, add mocks here.\"\"\"\n        # Mock an HTTP response\n        module_test.blasthttp_mock.add_response(\n            url=\"https://api.example.com/lookup?domain=blacklanternsecurity.com\",\n            json={\"results\": [\"sub.blacklanternsecurity.com\"]},\n        )\n\n        # Mock DNS\n        await module_test.mock_dns({\n            \"blacklanternsecurity.com\": {\"A\": [\"127.0.0.88\"]},\n        })\n\n        # Mock an HTTP server response\n        module_test.set_expect_requests(\n            expect_args={\"method\": \"GET\", \"uri\": \"/robots.txt\"},\n            respond_args={\"response_data\": \"Disallow: /secret/\"},\n        )\n\n    def check(self, module_test, events):\n        \"\"\"Verify the scan produced the expected events.\"\"\"\n        assert any(\n            e.data == \"sub.blacklanternsecurity.com\" and e.type == \"DNS_NAME\"\n            for e in events\n        ), \"Failed to find subdomain\"\n```\n\nThe test lifecycle runs:\n1. `setup_before_prep()` - set up mocks\n2. Scan `_prep()` - loads modules, config\n3. `setup_after_prep()` - modify scan state\n4. Scan runs and collects events\n5. `check()` - your assertions\n\n### Test Utilities\n\n- **`module_test.blasthttp_mock`** - mock HTTP responses \n- **`module_test.httpserver`** - real HTTP server on port 8888\n- **`module_test.httpserver_ssl`** - real HTTPS server on port 9999\n- **`module_test.mock_dns(data)`** - mock DNS responses\n- **`module_test.mock_interactsh(name)`** - mock out-of-band interactions\n- **`module_test.module`** - reference to the module instance being tested\n- **`module_test.scan`** - reference to the Scanner instance\n\nReal example -- `test_module_robots.py`:\n\n```python\nclass TestRobots(ModuleTestBase):\n    targets = [\"http://127.0.0.1:8888\"]\n    modules_overrides = [\"http\", \"robots\"]\n    config_overrides = {\"modules\": {\"robots\": {\"include_sitemap\": True}}}\n\n    async def setup_after_prep(self, module_test):\n        robots = \"Allow: /allow/\\nDisallow: /disallow/\\nSitemap: http://127.0.0.1:8888/sitemap.txt\"\n        module_test.set_expect_requests(\n            expect_args={\"method\": \"GET\", \"uri\": \"/robots.txt\"},\n            respond_args={\"response_data\": robots},\n        )\n\n    def check(self, module_test, events):\n        assert any(e.data == \"http://127.0.0.1:8888/allow/\" for e in events)\n        assert any(e.data == \"http://127.0.0.1:8888/disallow/\" for e in events)\n        assert any(e.data == \"http://127.0.0.1:8888/sitemap.txt\" for e in events)\n```\n","category":"root","tokens":7939},{"name":"CLAUDE.md","path":"CLAUDE.md","title":"CLAUDE.md","content":"Treat @AGENTS.md the same way you'd treat CLAUDE.md.","category":"root","tokens":13}]}