+
+
+```yaml
+{loaded_preset._yaml_str}
+```
+
+
+"""
+ update_md_files(f"BBOT {loaded_preset.name.upper()} PRESET", preset_yaml)
+ update_md_files(f"BBOT {loaded_preset.name.upper()} PRESET EXPANDABLE", preset_yaml_expandable)
+
+ content = []
+ for yaml_file, (loaded_preset, category, preset_path, original_filename) in DEFAULT_PRESET.all_presets.items():
+ yaml_str = loaded_preset._yaml_str
+ indent = " " * 4
+ yaml_str = f"\n{indent}".join(yaml_str.splitlines())
+ filename = homedir_collapseuser(yaml_file)
+
+ num_modules = len(loaded_preset.scan_modules)
+ modules = ", ".join(sorted([f"`{m}`" for m in loaded_preset.scan_modules]))
+ category = f"Category: {category}" if category else ""
+
+ content.append(
+ f"""## **{loaded_preset.name}**
+
+{loaded_preset.description}
+
+??? note "`{filename.name}`"
+ ```yaml title="{filename}"
+ {yaml_str}
+ ```
+
+{category}
+
+Modules: [{num_modules:,}]("{modules}")"""
+ )
+ assert len(content) > 5
+ update_md_files("BBOT PRESET YAML", "\n\n".join(content))
+
# Default config
default_config_file = bbot_code_dir / "bbot" / "defaults.yml"
with open(default_config_file) as f:
@@ -154,5 +289,8 @@ def update_toc(section, level=0):
# assert len(bbot_docs_toc.splitlines()) == 2
update_md_files("BBOT DOCS TOC", bbot_docs_toc)
+ # generate data for chord graph
+ gen_chord_data()
+
update_docs()
diff --git a/bbot/test/bbot_fixtures.py b/bbot/test/bbot_fixtures.py
index 85af696cd..44cc7a3ba 100644
--- a/bbot/test/bbot_fixtures.py
+++ b/bbot/test/bbot_fixtures.py
@@ -1,17 +1,46 @@
-import os
-import dns
+import os # noqa
import sys
import pytest
+import shutil # noqa
import asyncio # noqa
import logging
import subprocess
import tldextract
import pytest_httpserver
from pathlib import Path
-from omegaconf import OmegaConf
+from omegaconf import OmegaConf # noqa
from werkzeug.wrappers import Request
+from bbot.errors import * # noqa: F401
+from bbot.core import CORE
+from bbot.scanner import Preset
+from bbot.core.helpers.misc import mkdir
+
+
+log = logging.getLogger(f"bbot.test.fixtures")
+
+
+bbot_test_dir = Path("/tmp/.bbot_test")
+mkdir(bbot_test_dir)
+
+
+DEFAULT_PRESET = Preset()
+
+available_modules = list(DEFAULT_PRESET.module_loader.configs(type="scan"))
+available_output_modules = list(DEFAULT_PRESET.module_loader.configs(type="output"))
+available_internal_modules = list(DEFAULT_PRESET.module_loader.configs(type="internal"))
+
+
+@pytest.fixture
+def clean_default_config(monkeypatch):
+ clean_config = OmegaConf.merge(
+ CORE.files_config.get_default_config(), {"modules": DEFAULT_PRESET.module_loader.configs()}
+ )
+ with monkeypatch.context() as m:
+ m.setattr("bbot.core.core.DEFAULT_CONFIG", clean_config)
+ yield
+
class SubstringRequestMatcher(pytest_httpserver.httpserver.RequestMatcher):
def match_data(self, request: Request) -> bool:
@@ -22,28 +51,12 @@ def match_data(self, request: Request) -> bool:
pytest_httpserver.httpserver.RequestMatcher = SubstringRequestMatcher
-
-test_config = OmegaConf.load(Path(__file__).parent / "test.conf")
-if test_config.get("debug", False):
- os.environ["BBOT_DEBUG"] = "True"
-
-from .bbot_fixtures import * # noqa: F401
-import bbot.core.logger # noqa: F401
-from bbot.core.errors import * # noqa: F401
-
# silence pytest_httpserver
log = logging.getLogger("werkzeug")
log.setLevel(logging.CRITICAL)
-# silence stdout
-root_logger = logging.getLogger()
-for h in root_logger.handlers:
- h.addFilter(lambda x: x.levelname not in ("STDOUT", "TRACE"))
-
tldextract.extract("www.evilcorp.com")
-log = logging.getLogger(f"bbot.test.fixtures")
-
@pytest.fixture
def bbot_scanner():
@@ -53,16 +66,10 @@ def bbot_scanner():
@pytest.fixture
-def scan(monkeypatch, bbot_config):
+def scan(monkeypatch):
from bbot.scanner import Scanner
- bbot_scan = Scanner("127.0.0.1", modules=["ipneighbor"], config=bbot_config)
-
- fallback_nameservers_file = bbot_scan.helpers.bbot_home / "fallback_nameservers.txt"
- with open(fallback_nameservers_file, "w") as f:
- f.write("8.8.8.8\n")
- monkeypatch.setattr(bbot_scan.helpers.dns, "fallback_nameservers_file", fallback_nameservers_file)
-
+ bbot_scan = Scanner("127.0.0.1", modules=["ipneighbor"])
return bbot_scan
@@ -121,47 +128,47 @@ def helpers(scan):
@pytest.fixture
def events(scan):
class bbot_events:
- localhost = scan.make_event("127.0.0.1", source=scan.root_event)
- ipv4 = scan.make_event("8.8.8.8", source=scan.root_event)
- netv4 = scan.make_event("8.8.8.8/30", source=scan.root_event)
- ipv6 = scan.make_event("2001:4860:4860::8888", source=scan.root_event)
- netv6 = scan.make_event("2001:4860:4860::8888/126", source=scan.root_event)
- domain = scan.make_event("publicAPIs.org", source=scan.root_event)
- subdomain = scan.make_event("api.publicAPIs.org", source=scan.root_event)
- email = scan.make_event("bob@evilcorp.co.uk", "EMAIL_ADDRESS", source=scan.root_event)
- open_port = scan.make_event("api.publicAPIs.org:443", source=scan.root_event)
+ localhost = scan.make_event("127.0.0.1", parent=scan.root_event)
+ ipv4 = scan.make_event("8.8.8.8", parent=scan.root_event)
+ netv4 = scan.make_event("8.8.8.8/30", parent=scan.root_event)
+ ipv6 = scan.make_event("2001:4860:4860::8888", parent=scan.root_event)
+ netv6 = scan.make_event("2001:4860:4860::8888/126", parent=scan.root_event)
+ domain = scan.make_event("publicAPIs.org", parent=scan.root_event)
+ subdomain = scan.make_event("api.publicAPIs.org", parent=scan.root_event)
+ email = scan.make_event("bob@evilcorp.co.uk", "EMAIL_ADDRESS", parent=scan.root_event)
+ open_port = scan.make_event("api.publicAPIs.org:443", parent=scan.root_event)
protocol = scan.make_event(
- {"host": "api.publicAPIs.org", "port": 443, "protocol": "HTTP"}, "PROTOCOL", source=scan.root_event
+ {"host": "api.publicAPIs.org", "port": 443, "protocol": "HTTP"}, "PROTOCOL", parent=scan.root_event
)
- ipv4_open_port = scan.make_event("8.8.8.8:443", source=scan.root_event)
- ipv6_open_port = scan.make_event("[2001:4860:4860::8888]:443", "OPEN_TCP_PORT", source=scan.root_event)
- url_unverified = scan.make_event("https://api.publicAPIs.org:443/hellofriend", source=scan.root_event)
- ipv4_url_unverified = scan.make_event("https://8.8.8.8:443/hellofriend", source=scan.root_event)
- ipv6_url_unverified = scan.make_event("https://[2001:4860:4860::8888]:443/hellofriend", source=scan.root_event)
+ ipv4_open_port = scan.make_event("8.8.8.8:443", parent=scan.root_event)
+ ipv6_open_port = scan.make_event("[2001:4860:4860::8888]:443", "OPEN_TCP_PORT", parent=scan.root_event)
+ url_unverified = scan.make_event("https://api.publicAPIs.org:443/hellofriend", parent=scan.root_event)
+ ipv4_url_unverified = scan.make_event("https://8.8.8.8:443/hellofriend", parent=scan.root_event)
+ ipv6_url_unverified = scan.make_event("https://[2001:4860:4860::8888]:443/hellofriend", parent=scan.root_event)
url = scan.make_event(
- "https://api.publicAPIs.org:443/hellofriend", "URL", tags=["status-200"], source=scan.root_event
+ "https://api.publicAPIs.org:443/hellofriend", "URL", tags=["status-200"], parent=scan.root_event
)
ipv4_url = scan.make_event(
- "https://8.8.8.8:443/hellofriend", "URL", tags=["status-200"], source=scan.root_event
+ "https://8.8.8.8:443/hellofriend", "URL", tags=["status-200"], parent=scan.root_event
)
ipv6_url = scan.make_event(
- "https://[2001:4860:4860::8888]:443/hellofriend", "URL", tags=["status-200"], source=scan.root_event
+ "https://[2001:4860:4860::8888]:443/hellofriend", "URL", tags=["status-200"], parent=scan.root_event
)
- url_hint = scan.make_event("https://api.publicAPIs.org:443/hello.ash", "URL_HINT", source=url)
+ url_hint = scan.make_event("https://api.publicAPIs.org:443/hello.ash", "URL_HINT", parent=url)
vulnerability = scan.make_event(
{"host": "evilcorp.com", "severity": "INFO", "description": "asdf"},
"VULNERABILITY",
- source=scan.root_event,
+ parent=scan.root_event,
)
- finding = scan.make_event({"host": "evilcorp.com", "description": "asdf"}, "FINDING", source=scan.root_event)
- vhost = scan.make_event({"host": "evilcorp.com", "vhost": "www.evilcorp.com"}, "VHOST", source=scan.root_event)
- http_response = scan.make_event(httpx_response, "HTTP_RESPONSE", source=scan.root_event)
+ finding = scan.make_event({"host": "evilcorp.com", "description": "asdf"}, "FINDING", parent=scan.root_event)
+ vhost = scan.make_event({"host": "evilcorp.com", "vhost": "www.evilcorp.com"}, "VHOST", parent=scan.root_event)
+ http_response = scan.make_event(httpx_response, "HTTP_RESPONSE", parent=scan.root_event)
storage_bucket = scan.make_event(
{"name": "storage", "url": "https://storage.blob.core.windows.net"},
"STORAGE_BUCKET",
- source=scan.root_event,
+ parent=scan.root_event,
)
- emoji = scan.make_event("💩", "WHERE_IS_YOUR_GOD_NOW", source=scan.root_event)
+ emoji = scan.make_event("💩", "WHERE_IS_YOUR_GOD_NOW", parent=scan.root_event)
bbot_events.all = [ # noqa: F841
bbot_events.localhost,
@@ -197,104 +204,9 @@ class bbot_events:
return bbot_events
-@pytest.fixture
-def agent(monkeypatch, bbot_config):
- from bbot import agent
-
- test_agent = agent.Agent(bbot_config)
- test_agent.setup()
- return test_agent
-
-
-# bbot config
-from bbot import config as default_config
-
-test_config = OmegaConf.load(Path(__file__).parent / "test.conf")
-test_config = OmegaConf.merge(default_config, test_config)
-
-if test_config.get("debug", False):
- logging.getLogger("bbot").setLevel(logging.DEBUG)
-
-
-@pytest.fixture
-def bbot_config():
- return test_config
-
-
-from bbot.modules import module_loader
-
-available_modules = list(module_loader.configs(type="scan"))
-available_output_modules = list(module_loader.configs(type="output"))
-available_internal_modules = list(module_loader.configs(type="internal"))
-
-
-@pytest.fixture(autouse=True)
+@pytest.fixture(scope="session", autouse=True)
def install_all_python_deps():
deps_pip = set()
- for module in module_loader.preloaded().values():
+ for module in DEFAULT_PRESET.module_loader.preloaded().values():
deps_pip.update(set(module.get("deps", {}).get("pip", [])))
subprocess.run([sys.executable, "-m", "pip", "install"] + list(deps_pip))
-
-
-class MockResolver:
- import dns
-
- def __init__(self, mock_data=None):
- self.mock_data = mock_data if mock_data else {}
- self.nameservers = ["127.0.0.1"]
-
- async def resolve_address(self, ipaddr, *args, **kwargs):
- modified_kwargs = {}
- modified_kwargs.update(kwargs)
- modified_kwargs["rdtype"] = "PTR"
- return await self.resolve(str(dns.reversename.from_address(ipaddr)), *args, **modified_kwargs)
-
- def create_dns_response(self, query_name, rdtype):
- query_name = query_name.strip(".")
- answers = self.mock_data.get(query_name, {}).get(rdtype, [])
- if not answers:
- raise self.dns.resolver.NXDOMAIN(f"No answer found for {query_name} {rdtype}")
-
- message_text = f"""id 1234
-opcode QUERY
-rcode NOERROR
-flags QR AA RD
-;QUESTION
-{query_name}. IN {rdtype}
-;ANSWER"""
- for answer in answers:
- message_text += f"\n{query_name}. 1 IN {rdtype} {answer}"
-
- message_text += "\n;AUTHORITY\n;ADDITIONAL\n"
- message = self.dns.message.from_text(message_text)
- return message
-
- async def resolve(self, query_name, rdtype=None):
- if rdtype is None:
- rdtype = "A"
- elif isinstance(rdtype, str):
- rdtype = rdtype.upper()
- else:
- rdtype = str(rdtype.name).upper()
-
- domain_name = self.dns.name.from_text(query_name)
- rdtype_obj = self.dns.rdatatype.from_text(rdtype)
-
- if "_NXDOMAIN" in self.mock_data and query_name in self.mock_data["_NXDOMAIN"]:
- # Simulate the NXDOMAIN exception
- raise self.dns.resolver.NXDOMAIN
-
- try:
- response = self.create_dns_response(query_name, rdtype)
- answer = self.dns.resolver.Answer(domain_name, rdtype_obj, self.dns.rdataclass.IN, response)
- return answer
- except self.dns.resolver.NXDOMAIN:
- return []
-
-
-@pytest.fixture()
-def mock_dns():
- def _mock_dns(scan, mock_data):
- scan.helpers.dns.resolver = MockResolver(mock_data)
-
- return _mock_dns
diff --git a/bbot/test/conftest.py b/bbot/test/conftest.py
index ee8a6b2f0..5ae7b028a 100644
--- a/bbot/test/conftest.py
+++ b/bbot/test/conftest.py
@@ -1,15 +1,34 @@
+import os
import ssl
import shutil
import pytest
import asyncio
import logging
from pathlib import Path
+from contextlib import suppress
+from omegaconf import OmegaConf
from pytest_httpserver import HTTPServer
+from bbot.core import CORE
from bbot.core.helpers.misc import execute_sync_or_async
from bbot.core.helpers.interactsh import server_list as interactsh_servers
+test_config = OmegaConf.load(Path(__file__).parent / "test.conf")
+if test_config.get("debug", False):
+ os.environ["BBOT_DEBUG"] = "True"
+
+if test_config.get("debug", False):
+ logging.getLogger("bbot").setLevel(logging.DEBUG)
+else:
+ # silence stdout + trace
+ root_logger = logging.getLogger()
+ for h in root_logger.handlers:
+ h.addFilter(lambda x: x.levelname not in ("STDOUT", "TRACE"))
+
+CORE.merge_default(test_config)
+
+
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_sessionfinish(session, exitstatus):
# Remove handlers from all loggers to prevent logging errors at exit
@@ -25,11 +44,6 @@ def pytest_sessionfinish(session, exitstatus):
yield
-@pytest.fixture
-def non_mocked_hosts() -> list:
- return ["127.0.0.1", "localhost", "raw.githubusercontent.com"] + interactsh_servers
-
-
@pytest.fixture
def assert_all_responses_were_requested() -> bool:
return False
@@ -76,6 +90,11 @@ def bbot_httpserver_ssl():
server.clear()
+@pytest.fixture
+def non_mocked_hosts() -> list:
+ return ["127.0.0.1", "localhost", "raw.githubusercontent.com"] + interactsh_servers
+
+
@pytest.fixture
def bbot_httpserver_allinterfaces():
server = HTTPServer(host="0.0.0.0", port=5556)
@@ -90,28 +109,32 @@ def bbot_httpserver_allinterfaces():
server.clear()
-@pytest.fixture
-def interactsh_mock_instance():
- interactsh_mock = Interactsh_mock()
- return interactsh_mock
-
-
class Interactsh_mock:
- def __init__(self):
+ def __init__(self, name):
+ self.name = name
+ self.log = logging.getLogger(f"bbot.interactsh.{self.name}")
self.interactions = []
self.correlation_id = "deadbeef-dead-beef-dead-beefdeadbeef"
self.stop = False
+ self.poll_task = None
- def mock_interaction(self, subdomain_tag):
+ def mock_interaction(self, subdomain_tag, msg=None):
+ self.log.info(f"Mocking interaction to subdomain tag: {subdomain_tag}")
+ if msg is not None:
+ self.log.info(msg)
self.interactions.append(subdomain_tag)
async def register(self, callback=None):
if callable(callback):
- asyncio.create_task(self.poll_loop(callback))
+ self.poll_task = asyncio.create_task(self.poll_loop(callback))
return "fakedomain.fakeinteractsh.com"
async def deregister(self, callback=None):
self.stop = True
+ if self.poll_task is not None:
+ self.poll_task.cancel()
+ with suppress(BaseException):
+ await self.poll_task
async def poll_loop(self, callback=None):
while not self.stop:
diff --git a/bbot/test/test.conf b/bbot/test/test.conf
index ba8367461..5b9f35b45 100644
--- a/bbot/test/test.conf
+++ b/bbot/test/test.conf
@@ -4,9 +4,6 @@ modules:
wordlist: https://raw.githubusercontent.com/danielmiessler/SecLists/master/Discovery/DNS/deepmagic.com-prefixes-top500.txt
ffuf:
prefix_busting: true
- ipneighbor:
- test_option: ipneighbor
-output_modules:
http:
url: http://127.0.0.1:11111
username: username
@@ -17,35 +14,34 @@ output_modules:
token: asdf
neo4j:
uri: bolt://127.0.0.1:11111
- python:
- test_option: asdf
-internal_modules:
- speculate:
- test_option: speculate
-http_proxy:
-http_headers: { "test": "header" }
-ssl_verify: false
-scope_search_distance: 0
-scope_report_distance: 0
-scope_dns_search_distance: 1
-plumbus: asdf
-dns_debug: false
-user_agent: "BBOT Test User-Agent"
-http_debug: false
+web:
+ http_proxy:
+ http_headers: { "test": "header" }
+ ssl_verify: false
+ user_agent: "BBOT Test User-Agent"
+ debug: false
+scope:
+ search_distance: 0
+ report_distance: 0
+dns:
+ disable: false
+ minimal: true
+ search_distance: 1
+ debug: false
+ timeout: 1
+ wildcard_ignore:
+ - blacklanternsecurity.com
+ - fakedomain
+ - notreal
+ - google
+ - google.com
+ - example.com
+ - evilcorp.com
agent_url: ws://127.0.0.1:8765
agent_token: test
-dns_resolution: false
-dns_timeout: 1
speculate: false
excavate: false
aggregate: false
+cloudcheck: false
omit_event_types: []
debug: true
-dns_wildcard_ignore:
- - blacklanternsecurity.com
- - fakedomain
- - notreal
- - google
- - google.com
- - example.com
- - evilcorp.com
diff --git a/bbot/test/test_step_1/test__module__tests.py b/bbot/test/test_step_1/test__module__tests.py
index 0d0855557..791e58f58 100644
--- a/bbot/test/test_step_1/test__module__tests.py
+++ b/bbot/test/test_step_1/test__module__tests.py
@@ -2,8 +2,8 @@
import importlib
from pathlib import Path
+from bbot import Preset
from ..test_step_2.module_tests.base import ModuleTestBase
-from bbot.modules import module_loader
log = logging.getLogger("bbot.test.modules")
@@ -15,8 +15,11 @@
def test__module__tests():
+
+ preset = Preset()
+
# make sure each module has a .py file
- for module_name in module_loader.preloaded():
+ for module_name in preset.module_loader.preloaded():
module_name = module_name.lower()
assert module_name in module_test_files, f'No test file found for module "{module_name}"'
diff --git a/bbot/test/test_step_1/test_agent.py b/bbot/test/test_step_1/test_agent.py
deleted file mode 100644
index 00d70a751..000000000
--- a/bbot/test/test_step_1/test_agent.py
+++ /dev/null
@@ -1,158 +0,0 @@
-import json
-import websockets
-from functools import partial
-
-from ..bbot_fixtures import * # noqa: F401
-
-
-_first_run = True
-success = False
-
-
-async def websocket_handler(websocket, path, scan_done=None):
- # whether this is the first run
- global _first_run
- first_run = int(_first_run)
- # whether the test succeeded
- global success
- # test phase
- phase = "ping"
- # control channel or event channel?
- control = True
-
- if path == "/control/" and first_run:
- # test ping
- await websocket.send(json.dumps({"conversation": "90196cc1-299f-4555-82a0-bc22a4247590", "command": "ping"}))
- _first_run = False
- else:
- control = False
-
- # Bearer token
- assert websocket.request_headers["Authorization"] == "Bearer test"
-
- async for message in websocket:
- log.debug(f"PHASE: {phase}, MESSAGE: {message}")
- if not control or not first_run:
- continue
- m = json.loads(message)
- # ping
- if phase == "ping":
- assert json.loads(message)["message_type"] == "pong"
- phase = "start_scan_bad"
- if phase == "start_scan_bad":
- await websocket.send(
- json.dumps(
- {
- "conversation": "90196cc1-299f-4555-82a0-bc22a4247590",
- "command": "start_scan",
- "arguments": {
- "scan_id": "90196cc1-299f-4555-82a0-bc22a4247590",
- "targets": ["127.0.0.2"],
- "modules": ["asdf"],
- "output_modules": ["human"],
- "name": "agent_test_scan_bad",
- },
- }
- )
- )
- phase = "success"
- continue
- # scan start success
- if phase == "success":
- assert m["message"]["success"] == "Started scan"
- phase = "cleaning_up"
- continue
- # CLEANING_UP status message
- if phase == "cleaning_up":
- assert m["message_type"] == "scan_status_change"
- assert m["status"] == "CLEANING_UP"
- phase = "failed"
- continue
- # FAILED status message
- if phase == "failed":
- assert m["message_type"] == "scan_status_change"
- assert m["status"] == "FAILED"
- phase = "start_scan"
- # start good scan
- if phase == "start_scan":
- await websocket.send(
- json.dumps(
- {
- "conversation": "90196cc1-299f-4555-82a0-bc22a4247590",
- "command": "start_scan",
- "arguments": {
- "scan_id": "90196cc1-299f-4555-82a0-bc22a4247590",
- "targets": ["127.0.0.2"],
- "modules": ["ipneighbor"],
- "output_modules": ["human"],
- "name": "agent_test_scan",
- },
- }
- )
- )
- phase = "success_2"
- continue
- # scan start success
- if phase == "success_2":
- assert m["message"]["success"] == "Started scan"
- phase = "starting"
- continue
- # STARTING status message
- if phase == "starting":
- assert m["message_type"] == "scan_status_change"
- assert m["status"] == "STARTING"
- phase = "running"
- continue
- # RUNNING status message
- if phase == "running":
- assert m["message_type"] == "scan_status_change"
- assert m["status"] == "RUNNING"
- phase = "finishing"
- continue
- # FINISHING status message
- if phase == "finishing":
- assert m["message_type"] == "scan_status_change"
- assert m["status"] == "FINISHING"
- phase = "cleaning_up_2"
- continue
- # CLEANING_UP status message
- if phase == "cleaning_up_2":
- assert m["message_type"] == "scan_status_change"
- assert m["status"] == "CLEANING_UP"
- phase = "finished_2"
- continue
- # FINISHED status message
- if phase == "finished_2":
- assert m["message_type"] == "scan_status_change"
- assert m["status"] == "FINISHED"
- success = True
- scan_done.set()
- break
-
-
-@pytest.mark.asyncio
-async def test_agent(agent):
- scan_done = asyncio.Event()
- scan_status = await agent.scan_status()
- assert scan_status["error"] == "Scan not in progress"
-
- _websocket_handler = partial(websocket_handler, scan_done=scan_done)
-
- global success
- async with websockets.serve(_websocket_handler, "127.0.0.1", 8765):
- agent_task = asyncio.create_task(agent.start())
- # wait for 90 seconds
- await asyncio.wait_for(scan_done.wait(), 60)
- assert success
-
- await agent.start_scan("scan_to_be_cancelled", targets=["127.0.0.1"], modules=["ipneighbor"])
- await agent.start_scan("scan_to_be_rejected", targets=["127.0.0.1"], modules=["ipneighbor"])
- await asyncio.sleep(0.1)
- await agent.stop_scan()
- tasks = [agent.task, agent_task]
- for task in tasks:
- try:
- task.cancel()
- await task
- except (asyncio.CancelledError, AttributeError):
- pass
diff --git a/bbot/test/test_step_1/test_bloom_filter.py b/bbot/test/test_step_1/test_bloom_filter.py
new file mode 100644
index 000000000..6d8e6918d
--- /dev/null
+++ b/bbot/test/test_step_1/test_bloom_filter.py
@@ -0,0 +1,65 @@
+import time
+import string
+import random
+
+
+def test_bloom_filter():
+
+ def generate_random_strings(n, length=10):
+ """Generate a list of n random strings."""
+ return ["".join(random.choices(string.ascii_letters + string.digits, k=length)) for _ in range(n)]
+
+ from bbot.scanner import Scanner
+
+ scan = Scanner()
+
+ n_items_to_add = 100000
+ n_items_to_test = 100000
+ bloom_filter_size = 8000000
+
+ # Initialize the simple bloom filter and the set
+ bloom_filter = scan.helpers.bloom_filter(size=bloom_filter_size)
+
+ test_set = set()
+
+ # Generate random strings to add
+ print(f"Generating {n_items_to_add:,} items to add")
+ items_to_add = set(generate_random_strings(n_items_to_add))
+
+ # Generate random strings to test
+ print(f"Generating {n_items_to_test:,} items to test")
+ items_to_test = generate_random_strings(n_items_to_test)
+
+ print("Adding items")
+ start = time.time()
+ for item in items_to_add:
+ bloom_filter.add(item)
+ test_set.add(hash(item))
+ end = time.time()
+ elapsed = end - start
+ print(f"elapsed: {elapsed:.2f} ({int(n_items_to_test/elapsed)}/s)")
+ # this shouldn't take longer than 5 seconds
+ assert elapsed < 5
+
+ # make sure we have 100% accuracy
+ start = time.time()
+ for item in items_to_add:
+ assert item in bloom_filter
+ end = time.time()
+ elapsed = end - start
+ print(f"elapsed: {elapsed:.2f} ({int(n_items_to_test/elapsed)}/s)")
+ # this shouldn't take longer than 5 seconds
+ assert elapsed < 5
+
+ print("Measuring false positives")
+ # Check for false positives
+ false_positives = 0
+ for item in items_to_test:
+ if bloom_filter.check(item) and hash(item) not in test_set:
+ false_positives += 1
+ false_positive_percent = false_positives / len(items_to_test) * 100
+
+ print(f"False positive rate: {false_positive_percent:.2f}% ({false_positives}/{len(items_to_test)})")
+
+ # ensure false positives are less than .02 percent
+ assert false_positive_percent < 0.02
diff --git a/bbot/test/test_step_1/test_cli.py b/bbot/test/test_step_1/test_cli.py
index 0ccc94887..52b3867fe 100644
--- a/bbot/test/test_step_1/test_cli.py
+++ b/bbot/test/test_step_1/test_cli.py
@@ -1,35 +1,118 @@
from ..bbot_fixtures import *
+from bbot import cli
+
@pytest.mark.asyncio
-async def test_cli(monkeypatch, bbot_config):
- from bbot import cli
+async def test_cli_scope(monkeypatch, capsys):
+ import json
monkeypatch.setattr(sys, "exit", lambda *args, **kwargs: True)
monkeypatch.setattr(os, "_exit", lambda *args, **kwargs: True)
- monkeypatch.setattr(cli, "config", bbot_config)
- old_sys_argv = sys.argv
+ # basic target without whitelist
+ monkeypatch.setattr(
+ "sys.argv",
+ ["bbot", "-t", "one.one.one.one", "-c", "scope.report_distance=10", "dns.minimal=false", "--json"],
+ )
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == True
+ lines = [json.loads(l) for l in out.splitlines()]
+ dns_events = [l for l in lines if l["type"] == "DNS_NAME" and l["data"] == "one.one.one.one"]
+ assert dns_events
+ assert all([l["scope_distance"] == 0 and "in-scope" in l["tags"] for l in dns_events])
+ assert 1 == len(
+ [
+ l
+ for l in dns_events
+ if l["module"] == "TARGET"
+ and l["scope_distance"] == 0
+ and "in-scope" in l["tags"]
+ and "target" in l["tags"]
+ ]
+ )
+ ip_events = [l for l in lines if l["type"] == "IP_ADDRESS" and l["data"] == "1.1.1.1"]
+ assert ip_events
+ assert all([l["scope_distance"] == 1 and "distance-1" in l["tags"] for l in ip_events])
+ ip_events = [l for l in lines if l["type"] == "IP_ADDRESS" and l["data"] == "1.0.0.1"]
+ assert ip_events
+ assert all([l["scope_distance"] == 1 and "distance-1" in l["tags"] for l in ip_events])
+
+ # with whitelist
+ monkeypatch.setattr(
+ "sys.argv",
+ [
+ "bbot",
+ "-t",
+ "one.one.one.one",
+ "-w",
+ "192.168.0.1",
+ "-c",
+ "scope.report_distance=10",
+ "dns.minimal=false",
+ "dns.search_distance=2",
+ "--json",
+ ],
+ )
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == True
+ lines = [json.loads(l) for l in out.splitlines()]
+ lines = [l for l in lines if l["type"] != "SCAN"]
+ assert lines
+ assert not any([l["scope_distance"] == 0 for l in lines])
+ dns_events = [l for l in lines if l["type"] == "DNS_NAME" and l["data"] == "one.one.one.one"]
+ assert dns_events
+ assert all([l["scope_distance"] == 1 and "distance-1" in l["tags"] for l in dns_events])
+ assert 1 == len(
+ [
+ l
+ for l in dns_events
+ if l["module"] == "TARGET"
+ and l["scope_distance"] == 1
+ and "distance-1" in l["tags"]
+ and "target" in l["tags"]
+ ]
+ )
+ ip_events = [l for l in lines if l["type"] == "IP_ADDRESS" and l["data"] == "1.1.1.1"]
+ assert ip_events
+ assert all([l["scope_distance"] == 2 and "distance-2" in l["tags"] for l in ip_events])
+ ip_events = [l for l in lines if l["type"] == "IP_ADDRESS" and l["data"] == "1.0.0.1"]
+ assert ip_events
+ assert all([l["scope_distance"] == 2 and "distance-2" in l["tags"] for l in ip_events])
+
+
+@pytest.mark.asyncio
+async def test_cli_scan(monkeypatch):
+ monkeypatch.setattr(sys, "exit", lambda *args, **kwargs: True)
+ monkeypatch.setattr(os, "_exit", lambda *args, **kwargs: True)
- home_dir = Path(bbot_config["home"])
- scans_home = home_dir / "scans"
+ scans_home = bbot_test_dir / "scans"
# basic scan
monkeypatch.setattr(
sys,
"argv",
- ["bbot", "-y", "-t", "127.0.0.1", "www.example.com", "-n", "test_cli_scan", "-c", "dns_resolution=False"],
+ ["bbot", "-y", "-t", "127.0.0.1", "www.example.com", "-n", "test_cli_scan", "-c", "dns.disable=true"],
)
- await cli._main()
+ result = await cli._main()
+ assert result == True
scan_home = scans_home / "test_cli_scan"
+ assert (scan_home / "preset.yml").is_file(), "preset.yml not found"
assert (scan_home / "wordcloud.tsv").is_file(), "wordcloud.tsv not found"
assert (scan_home / "output.txt").is_file(), "output.txt not found"
assert (scan_home / "output.csv").is_file(), "output.csv not found"
- assert (scan_home / "output.ndjson").is_file(), "output.ndjson not found"
+ assert (scan_home / "output.json").is_file(), "output.json not found"
+
+ with open(scan_home / "preset.yml") as f:
+ text = f.read()
+ assert " dns:\n disable: true" in text
+
with open(scan_home / "output.csv") as f:
lines = f.readlines()
- assert lines[0] == "Event type,Event data,IP Address,Source Module,Scope Distance,Event Tags\n"
+ assert lines[0] == "Event type,Event data,IP Address,Source Module,Scope Distance,Event Tags,Discovery Path\n"
assert len(lines) > 1, "output.csv is not long enough"
ip_success = False
@@ -44,146 +127,537 @@ async def test_cli(monkeypatch, bbot_config):
dns_success = True
assert ip_success and dns_success, "IP_ADDRESS and/or DNS_NAME are not present in output.txt"
+
+@pytest.mark.asyncio
+async def test_cli_args(monkeypatch, caplog, capsys, clean_default_config):
+ caplog.set_level(logging.INFO)
+
+ monkeypatch.setattr(sys, "exit", lambda *args, **kwargs: True)
+ monkeypatch.setattr(os, "_exit", lambda *args, **kwargs: True)
+
# show version
monkeypatch.setattr("sys.argv", ["bbot", "--version"])
- await cli._main()
-
- # start agent
- monkeypatch.setattr("sys.argv", ["bbot", "--agent-mode"])
- task = asyncio.create_task(cli._main())
- await asyncio.sleep(2)
- task.cancel()
- try:
- await task
- except asyncio.CancelledError:
- pass
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == None
+ assert len(out.splitlines()) == 1
+ assert out.count(".") > 1
+
+ # list modules
+ monkeypatch.setattr("sys.argv", ["bbot", "--list-modules"])
+ result = await cli._main()
+ assert result == None
+ out, err = capsys.readouterr()
+ # internal modules
+ assert "| excavate " in out
+ # output modules
+ assert "| csv " in out
+ # scan modules
+ assert "| wayback " in out
+
+ # output dir and scan name
+ output_dir = bbot_test_dir / "bbot_cli_args_output"
+ scan_name = "bbot_cli_args_scan_name"
+ scan_dir = output_dir / scan_name
+ assert not output_dir.exists()
+ monkeypatch.setattr("sys.argv", ["bbot", "-o", str(output_dir), "-n", scan_name, "-y"])
+ result = await cli._main()
+ assert result == True
+ assert output_dir.is_dir()
+ assert scan_dir.is_dir()
+ assert "[SCAN]" in open(scan_dir / "output.txt").read()
+ assert "[INFO]" in open(scan_dir / "scan.log").read()
+ shutil.rmtree(output_dir)
+
+ # list module options
+ monkeypatch.setattr("sys.argv", ["bbot", "--list-module-options"])
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == None
+ assert "| modules.wayback.urls" in out
+ assert "| bool" in out
+ assert "| emit URLs in addition to DNS_NAMEs" in out
+ assert "| False" in out
+ assert "| modules.dnsbrute.wordlist" in out
+ assert "| modules.robots.include_allow" in out
+
+ # list module options by flag
+ monkeypatch.setattr("sys.argv", ["bbot", "-f", "subdomain-enum", "--list-module-options"])
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == None
+ assert "| modules.wayback.urls" in out
+ assert "| bool" in out
+ assert "| emit URLs in addition to DNS_NAMEs" in out
+ assert "| False" in out
+ assert "| modules.dnsbrute.wordlist" in out
+ assert not "| modules.robots.include_allow" in out
+
+ # list module options by module
+ monkeypatch.setattr("sys.argv", ["bbot", "-m", "dnsbrute", "-lmo"])
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == None
+ assert out.count("modules.") == out.count("modules.dnsbrute.")
+ assert not "| modules.wayback.urls" in out
+ assert "| modules.dnsbrute.wordlist" in out
+ assert not "| modules.robots.include_allow" in out
+
+ # list output module options by module
+ monkeypatch.setattr("sys.argv", ["bbot", "-om", "stdout", "-lmo"])
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == None
+ assert out.count("modules.") == out.count("modules.stdout.")
+
+ # list flags
+ monkeypatch.setattr("sys.argv", ["bbot", "--list-flags"])
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == None
+ assert "| safe " in out
+ assert "| Non-intrusive, safe to run " in out
+ assert "| active " in out
+ assert "| passive " in out
+
+ # list only a single flag
+ monkeypatch.setattr("sys.argv", ["bbot", "-f", "active", "--list-flags"])
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == None
+ assert not "| safe " in out
+ assert "| active " in out
+ assert not "| passive " in out
+
+ # list multiple flags
+ monkeypatch.setattr("sys.argv", ["bbot", "-f", "active", "safe", "--list-flags"])
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == None
+ assert "| safe " in out
+ assert "| active " in out
+ assert not "| passive " in out
# no args
monkeypatch.setattr("sys.argv", ["bbot"])
- await cli._main()
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == None
+ assert "Target:\n -t TARGET [TARGET ...]" in out
+
+ # list modules
+ monkeypatch.setattr("sys.argv", ["bbot", "-l"])
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == None
+ assert "| dnsbrute " in out
+ assert "| httpx " in out
+ assert "| robots " in out
+
+ # list modules by flag
+ monkeypatch.setattr("sys.argv", ["bbot", "-f", "subdomain-enum", "-l"])
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == None
+ assert "| dnsbrute " in out
+ assert "| httpx " in out
+ assert not "| robots " in out
+
+ # list modules by flag + required flag
+ monkeypatch.setattr("sys.argv", ["bbot", "-f", "subdomain-enum", "-rf", "passive", "-l"])
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == None
+ assert "| dnsbrute " in out
+ assert not "| httpx " in out
- # enable module by flag
- monkeypatch.setattr("sys.argv", ["bbot", "-f", "report"])
- await cli._main()
+ # list modules by flag + excluded flag
+ monkeypatch.setattr("sys.argv", ["bbot", "-f", "subdomain-enum", "-ef", "active", "-l"])
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == None
+ assert "| dnsbrute " in out
+ assert not "| httpx " in out
+
+ # list modules by flag + excluded module
+ monkeypatch.setattr("sys.argv", ["bbot", "-f", "subdomain-enum", "-em", "dnsbrute", "-l"])
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == None
+ assert not "| dnsbrute " in out
+ assert "| httpx " in out
+
+ # output modules override
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr("sys.argv", ["bbot", "-om", "csv,json", "-y"])
+ result = await cli._main()
+ assert result == True
+ assert "Loaded 2/2 output modules, (csv,json)" in caplog.text
+ caplog.clear()
+ monkeypatch.setattr("sys.argv", ["bbot", "-em", "csv,json", "-y"])
+ result = await cli._main()
+ assert result == True
+ assert "Loaded 3/3 output modules, (python,stdout,txt)" in caplog.text
+
+ # output modules override
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr("sys.argv", ["bbot", "-om", "subdomains", "-y"])
+ result = await cli._main()
+ assert result == True
+ assert "Loaded 6/6 output modules, (csv,json,python,stdout,subdomains,txt)" in caplog.text
+
+ # internal modules override
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr("sys.argv", ["bbot", "-y"])
+ result = await cli._main()
+ assert result == True
+ assert "Loaded 5/5 internal modules (aggregate,cloudcheck,dnsresolve,excavate,speculate)" in caplog.text
+ caplog.clear()
+ monkeypatch.setattr("sys.argv", ["bbot", "-em", "excavate", "speculate", "-y"])
+ result = await cli._main()
+ assert result == True
+ assert "Loaded 3/3 internal modules (aggregate,cloudcheck,dnsresolve)" in caplog.text
+ caplog.clear()
+ monkeypatch.setattr("sys.argv", ["bbot", "-c", "speculate=false", "-y"])
+ result = await cli._main()
+ assert result == True
+ assert "Loaded 4/4 internal modules (aggregate,cloudcheck,dnsresolve,excavate)" in caplog.text
+
+ # custom target type
+ out, err = capsys.readouterr()
+ monkeypatch.setattr("sys.argv", ["bbot", "-t", "ORG:evilcorp", "-y"])
+ result = await cli._main()
+ out, err = capsys.readouterr()
+ assert result == True
+ assert "[ORG_STUB] evilcorp TARGET" in out
+
+ # activate modules by flag
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr("sys.argv", ["bbot", "-f", "passive"])
+ result = await cli._main()
+ assert result == True
# unconsoleable output module
monkeypatch.setattr("sys.argv", ["bbot", "-om", "web_report"])
- await cli._main()
-
- # install all deps
- monkeypatch.setattr("sys.argv", ["bbot", "--install-all-deps"])
- success = await cli._main()
- assert success, "--install-all-deps failed for at least one module"
+ result = await cli._main()
+ assert result == True
# unresolved dependency
monkeypatch.setattr("sys.argv", ["bbot", "-m", "wappalyzer"])
- await cli._main()
+ result = await cli._main()
+ assert result == True
- # resolved dependency, excluded module
+ # enable and exclude the same module
+ caplog.clear()
+ assert not caplog.text
monkeypatch.setattr("sys.argv", ["bbot", "-m", "ffuf_shortnames", "-em", "ffuf_shortnames"])
- await cli._main()
+ result = await cli._main()
+ assert result == None
+ assert 'Unable to add scan module "ffuf_shortnames" because the module has been excluded' in caplog.text
# require flags
monkeypatch.setattr("sys.argv", ["bbot", "-f", "active", "-rf", "passive"])
- await cli._main()
+ result = await cli._main()
+ assert result == True
# excluded flags
monkeypatch.setattr("sys.argv", ["bbot", "-f", "active", "-ef", "active"])
- await cli._main()
+ result = await cli._main()
+ assert result == True
# slow modules
- monkeypatch.setattr("sys.argv", ["bbot", "-m", "massdns"])
- await cli._main()
+ monkeypatch.setattr("sys.argv", ["bbot", "-m", "bucket_digitalocean"])
+ result = await cli._main()
+ assert result == True
# deadly modules
+ caplog.clear()
+ assert not caplog.text
monkeypatch.setattr("sys.argv", ["bbot", "-m", "nuclei"])
result = await cli._main()
assert result == False, "-m nuclei ran without --allow-deadly"
+ assert "Please specify --allow-deadly to continue" in caplog.text
# --allow-deadly
monkeypatch.setattr("sys.argv", ["bbot", "-m", "nuclei", "--allow-deadly"])
result = await cli._main()
- assert result != False, "-m nuclei failed to run with --allow-deadly"
+ assert result == True, "-m nuclei failed to run with --allow-deadly"
- # show current config
- monkeypatch.setattr("sys.argv", ["bbot", "-y", "--current-config"])
- await cli._main()
+ # install all deps
+ monkeypatch.setattr("sys.argv", ["bbot", "--install-all-deps"])
+ success = await cli._main()
+ assert success == True, "--install-all-deps failed for at least one module"
- # list modules
- monkeypatch.setattr("sys.argv", ["bbot", "-l"])
- await cli._main()
- # list module options
- monkeypatch.setattr("sys.argv", ["bbot", "--help-all"])
- await cli._main()
+@pytest.mark.asyncio
+async def test_cli_customheaders(monkeypatch, caplog, capsys):
+ monkeypatch.setattr(sys, "exit", lambda *args, **kwargs: True)
+ monkeypatch.setattr(os, "_exit", lambda *args, **kwargs: True)
+ import yaml
+
+ # test custom headers
+ monkeypatch.setattr(
+ "sys.argv", ["bbot", "--custom-headers", "foo=bar", "foo2=bar2", "foo3=bar=3", "--current-preset"]
+ )
+ success = await cli._main()
+ assert success == None, "setting custom headers on command line failed"
+ captured = capsys.readouterr()
+ stdout_preset = yaml.safe_load(captured.out)
+ assert stdout_preset["config"]["web"]["http_headers"] == {"foo": "bar", "foo2": "bar2", "foo3": "bar=3"}
- # unpatch sys.argv
- monkeypatch.setattr("sys.argv", old_sys_argv)
+ # test custom headers invalid (no "=")
+ monkeypatch.setattr("sys.argv", ["bbot", "--custom-headers", "justastring", "--current-preset"])
+ result = await cli._main()
+ assert result == None
+ assert "Custom headers not formatted correctly (missing '=')" in caplog.text
+ caplog.clear()
+ # test custom headers invalid (missing key)
+ monkeypatch.setattr("sys.argv", ["bbot", "--custom-headers", "=nokey", "--current-preset"])
+ result = await cli._main()
+ assert result == None
+ assert "Custom headers not formatted correctly (missing header name or value)" in caplog.text
+ caplog.clear()
+
+ # test custom headers invalid (missing value)
+ monkeypatch.setattr("sys.argv", ["bbot", "--custom-headers", "missingvalue=", "--current-preset"])
+ result = await cli._main()
+ assert result == None
+ assert "Custom headers not formatted correctly (missing header name or value)" in caplog.text
-def test_config_validation(monkeypatch, capsys, bbot_config):
- from bbot import cli
- from bbot.core.configurator import args
+def test_cli_config_validation(monkeypatch, caplog):
monkeypatch.setattr(sys, "exit", lambda *args, **kwargs: True)
monkeypatch.setattr(os, "_exit", lambda *args, **kwargs: True)
- monkeypatch.setattr(cli, "config", bbot_config)
-
- old_cli_config = args.cli_config
# incorrect module option
- monkeypatch.setattr(args, "cli_config", ["bbot", "-c", "modules.ipnegibhor.num_bits=4"])
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr("sys.argv", ["bbot", "-c", "modules.ipnegibhor.num_bits=4"])
cli.main()
- captured = capsys.readouterr()
- assert 'Could not find module option "modules.ipnegibhor.num_bits"' in captured.err
- assert 'Did you mean "modules.ipneighbor.num_bits"?' in captured.err
+ assert 'Could not find config option "modules.ipnegibhor.num_bits"' in caplog.text
+ assert 'Did you mean "modules.ipneighbor.num_bits"?' in caplog.text
# incorrect global option
- monkeypatch.setattr(args, "cli_config", ["bbot", "-c", "web_spier_distance=4"])
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr("sys.argv", ["bbot", "-c", "web_spier_distance=4"])
cli.main()
- captured = capsys.readouterr()
- assert 'Could not find module option "web_spier_distance"' in captured.err
- assert 'Did you mean "web_spider_distance"?' in captured.err
+ assert 'Could not find config option "web_spier_distance"' in caplog.text
+ assert 'Did you mean "web.spider_distance"?' in caplog.text
+
+
+def test_cli_module_validation(monkeypatch, caplog):
+ monkeypatch.setattr(sys, "exit", lambda *args, **kwargs: True)
+ monkeypatch.setattr(os, "_exit", lambda *args, **kwargs: True)
+
+ # incorrect module
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr("sys.argv", ["bbot", "-m", "dnsbrutes"])
+ cli.main()
+ assert 'Could not find scan module "dnsbrutes"' in caplog.text
+ assert 'Did you mean "dnsbrute"?' in caplog.text
+
+ # incorrect excluded module
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr("sys.argv", ["bbot", "-em", "dnsbrutes"])
+ cli.main()
+ assert 'Could not find module "dnsbrutes"' in caplog.text
+ assert 'Did you mean "dnsbrute"?' in caplog.text
+
+ # incorrect output module
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr("sys.argv", ["bbot", "-om", "neoo4j"])
+ cli.main()
+ assert 'Could not find output module "neoo4j"' in caplog.text
+ assert 'Did you mean "neo4j"?' in caplog.text
+
+ # output module setup failed
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr("sys.argv", ["bbot", "-om", "websocket", "-c", "modules.websocket.url=", "-y"])
+ cli.main()
+ lines = caplog.text.splitlines()
+ assert "Loaded 6/6 output modules, (csv,json,python,stdout,txt,websocket)" in caplog.text
+ assert 1 == len(
+ [
+ l
+ for l in lines
+ if l.startswith("WARNING bbot.scanner:scanner.py")
+ and l.endswith("Setup hard-failed for websocket: Must set URL")
+ ]
+ )
+ assert 1 == len(
+ [
+ l
+ for l in lines
+ if l.startswith("WARNING bbot.modules.output.websocket:base.py") and l.endswith("Setting error state")
+ ]
+ )
+ assert 1 == len(
+ [
+ l
+ for l in lines
+ if l.startswith("ERROR bbot.cli:cli.py")
+ and l.endswith("Setup hard-failed for 1 modules (websocket) (--force to run module anyway)")
+ ]
+ )
+
+ # only output module setup failed
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr(
+ "sys.argv",
+ ["bbot", "-om", "websocket", "-em", "python,stdout,csv,json,txt", "-c", "modules.websocket.url=", "-y"],
+ )
+ cli.main()
+ lines = caplog.text.splitlines()
+ assert "Loaded 1/1 output modules, (websocket)" in caplog.text
+ assert 1 == len(
+ [
+ l
+ for l in lines
+ if l.startswith("WARNING bbot.scanner:scanner.py")
+ and l.endswith("Setup hard-failed for websocket: Must set URL")
+ ]
+ )
+ assert 1 == len(
+ [
+ l
+ for l in lines
+ if l.startswith("WARNING bbot.modules.output.websocket:base.py") and l.endswith("Setting error state")
+ ]
+ )
+ assert 1 == len(
+ [
+ l
+ for l in lines
+ if l.startswith("ERROR bbot.cli:cli.py") and l.endswith("Failed to load output modules. Aborting.")
+ ]
+ )
- # unpatch cli_options
- monkeypatch.setattr(args, "cli_config", old_cli_config)
+ # incorrect flag
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr("sys.argv", ["bbot", "-f", "subdomainenum"])
+ cli.main()
+ assert 'Could not find flag "subdomainenum"' in caplog.text
+ assert 'Did you mean "subdomain-enum"?' in caplog.text
+ # incorrect excluded flag
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr("sys.argv", ["bbot", "-ef", "subdomainenum"])
+ cli.main()
+ assert 'Could not find flag "subdomainenum"' in caplog.text
+ assert 'Did you mean "subdomain-enum"?' in caplog.text
+
+ # incorrect required flag
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr("sys.argv", ["bbot", "-rf", "subdomainenum"])
+ cli.main()
+ assert 'Could not find flag "subdomainenum"' in caplog.text
+ assert 'Did you mean "subdomain-enum"?' in caplog.text
-def test_module_validation(monkeypatch, capsys, bbot_config):
- from bbot.core.configurator import args
+
+def test_cli_presets(monkeypatch, capsys, caplog):
+ import yaml
monkeypatch.setattr(sys, "exit", lambda *args, **kwargs: True)
monkeypatch.setattr(os, "_exit", lambda *args, **kwargs: True)
- old_sys_argv = sys.argv
+ # show current preset
+ monkeypatch.setattr("sys.argv", ["bbot", "-c", "web.http_proxy=currentpresettest", "--current-preset"])
+ cli.main()
+ captured = capsys.readouterr()
+ assert " http_proxy: currentpresettest" in captured.out
- # incorrect module
- monkeypatch.setattr(sys, "argv", ["bbot", "-m", "massdnss"])
- args.parser.parse_args()
+ # show current preset (full)
+ monkeypatch.setattr("sys.argv", ["bbot", "-c" "modules.c99.api_key=asdf", "--current-preset-full"])
+ cli.main()
+ captured = capsys.readouterr()
+ assert " api_key: asdf" in captured.out
+
+ preset_dir = bbot_test_dir / "test_cli_presets"
+ preset_dir.mkdir(exist_ok=True)
+
+ preset1_file = preset_dir / "cli_preset1.conf"
+ with open(preset1_file, "w") as f:
+ f.write(
+ """
+config:
+ web:
+ http_proxy: http://proxy1
+ """
+ )
+
+ preset2_file = preset_dir / "cli_preset2.yml"
+ with open(preset2_file, "w") as f:
+ f.write(
+ """
+config:
+ web:
+ http_proxy: http://proxy2
+ """
+ )
+
+ # test reading single preset
+ monkeypatch.setattr("sys.argv", ["bbot", "-p", str(preset1_file.resolve()), "--current-preset"])
+ cli.main()
captured = capsys.readouterr()
- assert 'Could not find module "massdnss"' in captured.err
- assert 'Did you mean "massdns"?' in captured.err
+ stdout_preset = yaml.safe_load(captured.out)
+ assert stdout_preset["config"]["web"]["http_proxy"] == "http://proxy1"
- # incorrect excluded module
- monkeypatch.setattr(sys, "argv", ["bbot", "-em", "massdnss"])
- args.parser.parse_args()
+ # preset overrides preset
+ monkeypatch.setattr(
+ "sys.argv", ["bbot", "-p", str(preset2_file.resolve()), str(preset1_file.resolve()), "--current-preset"]
+ )
+ cli.main()
captured = capsys.readouterr()
- assert 'Could not find module "massdnss"' in captured.err
- assert 'Did you mean "massdns"?' in captured.err
+ stdout_preset = yaml.safe_load(captured.out)
+ assert stdout_preset["config"]["web"]["http_proxy"] == "http://proxy1"
- # incorrect output module
- monkeypatch.setattr(sys, "argv", ["bbot", "-om", "neoo4j"])
- args.parser.parse_args()
+ # override other way
+ monkeypatch.setattr(
+ "sys.argv", ["bbot", "-p", str(preset1_file.resolve()), str(preset2_file.resolve()), "--current-preset"]
+ )
+ cli.main()
captured = capsys.readouterr()
- assert 'Could not find output module "neoo4j"' in captured.err
- assert 'Did you mean "neo4j"?' in captured.err
+ stdout_preset = yaml.safe_load(captured.out)
+ assert stdout_preset["config"]["web"]["http_proxy"] == "http://proxy2"
- # incorrect flag
- monkeypatch.setattr(sys, "argv", ["bbot", "-f", "subdomainenum"])
- args.parser.parse_args()
+ # cli config overrides all presets
+ monkeypatch.setattr(
+ "sys.argv",
+ [
+ "bbot",
+ "-p",
+ str(preset1_file.resolve()),
+ str(preset2_file.resolve()),
+ "-c",
+ "web.http_proxy=asdf",
+ "--current-preset",
+ ],
+ )
+ cli.main()
captured = capsys.readouterr()
- assert 'Could not find flag "subdomainenum"' in captured.err
- assert 'Did you mean "subdomain-enum"?' in captured.err
+ stdout_preset = yaml.safe_load(captured.out)
+ assert stdout_preset["config"]["web"]["http_proxy"] == "asdf"
+
+ # invalid preset
+ caplog.clear()
+ assert not caplog.text
+ monkeypatch.setattr("sys.argv", ["bbot", "-p", "asdfasdfasdf", "-y"])
+ cli.main()
+ assert "file does not exist. Use -lp to list available presets" in caplog.text
- # unpatch sys.argv
- monkeypatch.setattr("sys.argv", old_sys_argv)
+ preset1_file.unlink()
+ preset2_file.unlink()
diff --git a/bbot/test/test_step_1/test_cloud_helpers.py b/bbot/test/test_step_1/test_cloud_helpers.py
deleted file mode 100644
index b42da11a7..000000000
--- a/bbot/test/test_step_1/test_cloud_helpers.py
+++ /dev/null
@@ -1,86 +0,0 @@
-from ..bbot_fixtures import * # noqa: F401
-
-
-@pytest.mark.asyncio
-async def test_cloud_helpers(bbot_scanner, bbot_config):
- scan1 = bbot_scanner("127.0.0.1", config=bbot_config)
-
- provider_names = ("amazon", "google", "azure", "digitalocean", "oracle", "akamai", "cloudflare", "github")
- for provider_name in provider_names:
- assert provider_name in scan1.helpers.cloud.providers.providers
-
- for p in scan1.helpers.cloud.providers.providers.values():
- print(f"{p.name}: {p.domains} / {p.ranges}")
- amazon_ranges = list(scan1.helpers.cloud["amazon"].ranges)
- assert amazon_ranges
- amazon_range = next(iter(amazon_ranges))
- amazon_address = amazon_range.broadcast_address
-
- ip_event = scan1.make_event(amazon_address, source=scan1.root_event)
- aws_event1 = scan1.make_event("amazonaws.com", source=scan1.root_event)
- aws_event2 = scan1.make_event("asdf.amazonaws.com", source=scan1.root_event)
- aws_event3 = scan1.make_event("asdfamazonaws.com", source=scan1.root_event)
- aws_event4 = scan1.make_event("test.asdf.aws", source=scan1.root_event)
-
- other_event1 = scan1.make_event("cname.evilcorp.com", source=scan1.root_event)
- other_event2 = scan1.make_event("cname2.evilcorp.com", source=scan1.root_event)
- other_event3 = scan1.make_event("cname3.evilcorp.com", source=scan1.root_event)
- other_event2._resolved_hosts = {amazon_address}
- other_event3._resolved_hosts = {"asdf.amazonaws.com"}
-
- for event in (ip_event, aws_event1, aws_event2, aws_event4, other_event2, other_event3):
- await scan1.helpers.cloud.tag_event(event)
- assert "cloud-amazon" in event.tags, f"{event} was not properly cloud-tagged"
-
- for event in (aws_event3, other_event1):
- await scan1.helpers.cloud.tag_event(event)
- assert "cloud-amazon" not in event.tags, f"{event} was improperly cloud-tagged"
- assert not any(
- t for t in event.tags if t.startswith("cloud-") or t.startswith("cdn-")
- ), f"{event} was improperly cloud-tagged"
-
- google_event1 = scan1.make_event("asdf.googleapis.com", source=scan1.root_event)
- google_event2 = scan1.make_event("asdf.google", source=scan1.root_event)
- google_event3 = scan1.make_event("asdf.evilcorp.com", source=scan1.root_event)
- google_event3._resolved_hosts = {"asdf.storage.googleapis.com"}
-
- for event in (google_event1, google_event2, google_event3):
- await scan1.helpers.cloud.tag_event(event)
- assert "cloud-google" in event.tags, f"{event} was not properly cloud-tagged"
- assert "cloud-storage-bucket" in google_event3.tags
-
-
-@pytest.mark.asyncio
-async def test_cloud_helpers_excavate(bbot_scanner, bbot_config, bbot_httpserver):
- url = bbot_httpserver.url_for("/test_cloud_helpers_excavate")
- bbot_httpserver.expect_request(uri="/test_cloud_helpers_excavate").respond_with_data(
- ""
- )
- scan1 = bbot_scanner(url, modules=["httpx", "excavate"], config=bbot_config)
- events = [e async for e in scan1.async_start()]
- assert 1 == len(
- [
- e
- for e in events
- if e.type == "STORAGE_BUCKET"
- and e.data["name"] == "asdf"
- and "cloud-amazon" in e.tags
- and "cloud-storage-bucket" in e.tags
- ]
- )
-
-
-@pytest.mark.asyncio
-async def test_cloud_helpers_speculate(bbot_scanner, bbot_config):
- scan1 = bbot_scanner("asdf.s3.amazonaws.com", modules=["speculate"], config=bbot_config)
- events = [e async for e in scan1.async_start()]
- assert 1 == len(
- [
- e
- for e in events
- if e.type == "STORAGE_BUCKET"
- and e.data["name"] == "asdf"
- and "cloud-amazon" in e.tags
- and "cloud-storage-bucket" in e.tags
- ]
- )
diff --git a/bbot/test/test_step_1/test_command.py b/bbot/test/test_step_1/test_command.py
index 42e41e4da..31abef19b 100644
--- a/bbot/test/test_step_1/test_command.py
+++ b/bbot/test/test_step_1/test_command.py
@@ -4,8 +4,8 @@
@pytest.mark.asyncio
-async def test_command(bbot_scanner, bbot_config):
- scan1 = bbot_scanner(config=bbot_config)
+async def test_command(bbot_scanner):
+ scan1 = bbot_scanner()
# test timeouts
command = ["sleep", "3"]
@@ -78,10 +78,10 @@ async def test_command(bbot_scanner, bbot_config):
# test check=True
with pytest.raises(CalledProcessError) as excinfo:
- lines = [l async for line in scan1.helpers.run_live(["ls", "/aslkdjflasdkfsd"], check=True)]
+ lines = [line async for line in scan1.helpers.run_live(["ls", "/aslkdjflasdkfsd"], check=True)]
assert "No such file or directory" in excinfo.value.stderr
with pytest.raises(CalledProcessError) as excinfo:
- lines = [l async for line in scan1.helpers.run_live(["ls", "/aslkdjflasdkfsd"], check=True, text=False)]
+ lines = [line async for line in scan1.helpers.run_live(["ls", "/aslkdjflasdkfsd"], check=True, text=False)]
assert b"No such file or directory" in excinfo.value.stderr
with pytest.raises(CalledProcessError) as excinfo:
await scan1.helpers.run(["ls", "/aslkdjflasdkfsd"], check=True)
@@ -116,29 +116,29 @@ async def test_command(bbot_scanner, bbot_config):
assert not lines
# test sudo + existence of environment variables
- scan1.load_modules()
+ await scan1.load_modules()
path_parts = os.environ.get("PATH", "").split(":")
assert "/tmp/.bbot_test/tools" in path_parts
run_lines = (await scan1.helpers.run(["env"])).stdout.splitlines()
- assert "BBOT_PLUMBUS=asdf" in run_lines
+ assert "BBOT_WEB_USER_AGENT=BBOT Test User-Agent" in run_lines
for line in run_lines:
if line.startswith("PATH="):
path_parts = line.split("=", 1)[-1].split(":")
assert "/tmp/.bbot_test/tools" in path_parts
run_lines_sudo = (await scan1.helpers.run(["env"], sudo=True)).stdout.splitlines()
- assert "BBOT_PLUMBUS=asdf" in run_lines_sudo
+ assert "BBOT_WEB_USER_AGENT=BBOT Test User-Agent" in run_lines_sudo
for line in run_lines_sudo:
if line.startswith("PATH="):
path_parts = line.split("=", 1)[-1].split(":")
assert "/tmp/.bbot_test/tools" in path_parts
run_live_lines = [l async for l in scan1.helpers.run_live(["env"])]
- assert "BBOT_PLUMBUS=asdf" in run_live_lines
+ assert "BBOT_WEB_USER_AGENT=BBOT Test User-Agent" in run_live_lines
for line in run_live_lines:
if line.startswith("PATH="):
path_parts = line.strip().split("=", 1)[-1].split(":")
assert "/tmp/.bbot_test/tools" in path_parts
run_live_lines_sudo = [l async for l in scan1.helpers.run_live(["env"], sudo=True)]
- assert "BBOT_PLUMBUS=asdf" in run_live_lines_sudo
+ assert "BBOT_WEB_USER_AGENT=BBOT Test User-Agent" in run_live_lines_sudo
for line in run_live_lines_sudo:
if line.startswith("PATH="):
path_parts = line.strip().split("=", 1)[-1].split(":")
diff --git a/bbot/test/test_step_1/test_config.py b/bbot/test/test_step_1/test_config.py
index 2d9980a2c..b237fcae4 100644
--- a/bbot/test/test_step_1/test_config.py
+++ b/bbot/test/test_step_1/test_config.py
@@ -2,9 +2,21 @@
@pytest.mark.asyncio
-async def test_config(bbot_config, bbot_scanner):
- scan1 = bbot_scanner("127.0.0.1", modules=["ipneighbor", "speculate"], config=bbot_config)
+async def test_config(bbot_scanner):
+ config = OmegaConf.create(
+ {
+ "plumbus": "asdf",
+ "speculate": True,
+ "modules": {
+ "ipneighbor": {"test_option": "ipneighbor"},
+ "python": {"test_option": "asdf"},
+ "speculate": {"test_option": "speculate"},
+ },
+ }
+ )
+ scan1 = bbot_scanner("127.0.0.1", modules=["ipneighbor"], config=config)
await scan1.load_modules()
+ assert scan1.config.web.user_agent == "BBOT Test User-Agent"
assert scan1.config.plumbus == "asdf"
assert scan1.modules["ipneighbor"].config.test_option == "ipneighbor"
assert scan1.modules["python"].config.test_option == "asdf"
diff --git a/bbot/test/test_step_1/test_depsinstaller.py b/bbot/test/test_step_1/test_depsinstaller.py
index 39a56bf41..e3f80d5cf 100644
--- a/bbot/test/test_step_1/test_depsinstaller.py
+++ b/bbot/test/test_step_1/test_depsinstaller.py
@@ -1,11 +1,9 @@
from ..bbot_fixtures import *
-def test_depsinstaller(monkeypatch, bbot_config, bbot_scanner):
+def test_depsinstaller(monkeypatch, bbot_scanner):
scan = bbot_scanner(
"127.0.0.1",
- modules=["dnsresolve"],
- config=bbot_config,
)
# test shell
diff --git a/bbot/test/test_step_1/test_dns.py b/bbot/test/test_step_1/test_dns.py
index 91465507e..ae24c6c25 100644
--- a/bbot/test/test_step_1/test_dns.py
+++ b/bbot/test/test_step_1/test_dns.py
@@ -1,92 +1,162 @@
from ..bbot_fixtures import *
+from bbot.core.helpers.dns.helpers import extract_targets
+
+
+mock_records = {
+ "one.one.one.one": {
+ "A": ["1.1.1.1", "1.0.0.1"],
+ "AAAA": ["2606:4700:4700::1111", "2606:4700:4700::1001"],
+ "TXT": [
+ '"v=spf1 ip4:103.151.192.0/23 ip4:185.12.80.0/22 ip4:188.172.128.0/20 ip4:192.161.144.0/20 ip4:216.198.0.0/18 ~all"'
+ ],
+ },
+ "1.1.1.1.in-addr.arpa": {"PTR": ["one.one.one.one."]},
+}
+
@pytest.mark.asyncio
-async def test_dns(bbot_scanner, bbot_config, mock_dns):
- scan = bbot_scanner("1.1.1.1", config=bbot_config)
- helpers = scan.helpers
+async def test_dns_engine(bbot_scanner):
+ scan = bbot_scanner()
+ await scan.helpers._mock_dns(
+ {"one.one.one.one": {"A": ["1.1.1.1"]}, "1.1.1.1.in-addr.arpa": {"PTR": ["one.one.one.one"]}}
+ )
+ result = await scan.helpers.resolve("one.one.one.one")
+ assert "1.1.1.1" in result
+ assert not "2606:4700:4700::1111" in result
+
+ results = [_ async for _ in scan.helpers.resolve_batch(("one.one.one.one", "1.1.1.1"))]
+ pass_1 = False
+ pass_2 = False
+ for query, result in results:
+ if query == "one.one.one.one" and "1.1.1.1" in result:
+ pass_1 = True
+ elif query == "1.1.1.1" and "one.one.one.one" in result:
+ pass_2 = True
+ assert pass_1 and pass_2
+
+ results = [_ async for _ in scan.helpers.resolve_raw_batch((("one.one.one.one", "A"), ("1.1.1.1", "PTR")))]
+ pass_1 = False
+ pass_2 = False
+ for (query, rdtype), (result, errors) in results:
+ result = extract_targets(result)
+ _results = [r[1] for r in result]
+ if query == "one.one.one.one" and "1.1.1.1" in _results:
+ pass_1 = True
+ elif query == "1.1.1.1" and "one.one.one.one" in _results:
+ pass_2 = True
+ assert pass_1 and pass_2
+
+ from bbot.core.helpers.dns.mock import MockResolver
+
+ # ensure dns records are being properly cleaned
+ mockresolver = MockResolver({"evilcorp.com": {"MX": ["0 ."]}})
+ mx_records = await mockresolver.resolve("evilcorp.com", rdtype="MX")
+ results = set()
+ for r in mx_records:
+ results.update(extract_targets(r))
+ assert not results
+
+
+@pytest.mark.asyncio
+async def test_dns_resolution(bbot_scanner):
+ scan = bbot_scanner("1.1.1.1")
+
+ from bbot.core.helpers.dns.engine import DNSEngine
+
+ dnsengine = DNSEngine(None)
+ await dnsengine._mock_dns(mock_records)
# lowest level functions
- a_responses = await helpers._resolve_hostname("one.one.one.one")
- aaaa_responses = await helpers._resolve_hostname("one.one.one.one", rdtype="AAAA")
- ip_responses = await helpers._resolve_ip("1.1.1.1")
+ a_responses = await dnsengine._resolve_hostname("one.one.one.one")
+ aaaa_responses = await dnsengine._resolve_hostname("one.one.one.one", rdtype="AAAA")
+ ip_responses = await dnsengine._resolve_ip("1.1.1.1")
assert a_responses[0].response.answer[0][0].address in ("1.1.1.1", "1.0.0.1")
assert aaaa_responses[0].response.answer[0][0].address in ("2606:4700:4700::1111", "2606:4700:4700::1001")
assert ip_responses[0].response.answer[0][0].target.to_text() in ("one.one.one.one.",)
# mid level functions
- _responses, errors = await helpers.resolve_raw("one.one.one.one")
+ answers, errors = await dnsengine.resolve_raw("one.one.one.one", type="A")
responses = []
- for rdtype, response in _responses:
- for answers in response:
- responses += list(helpers.extract_targets(answers))
+ for answer in answers:
+ responses += list(extract_targets(answer))
assert ("A", "1.1.1.1") in responses
- _responses, errors = await helpers.resolve_raw("one.one.one.one", rdtype="AAAA")
+ assert not ("AAAA", "2606:4700:4700::1111") in responses
+ answers, errors = await dnsengine.resolve_raw("one.one.one.one", type="AAAA")
responses = []
- for rdtype, response in _responses:
- for answers in response:
- responses += list(helpers.extract_targets(answers))
+ for answer in answers:
+ responses += list(extract_targets(answer))
+ assert not ("A", "1.1.1.1") in responses
assert ("AAAA", "2606:4700:4700::1111") in responses
- _responses, errors = await helpers.resolve_raw("1.1.1.1")
+ answers, errors = await dnsengine.resolve_raw("1.1.1.1")
responses = []
- for rdtype, response in _responses:
- for answers in response:
- responses += list(helpers.extract_targets(answers))
+ for answer in answers:
+ responses += list(extract_targets(answer))
assert ("PTR", "one.one.one.one") in responses
# high level functions
- assert "1.1.1.1" in await helpers.resolve("one.one.one.one")
- assert "2606:4700:4700::1111" in await helpers.resolve("one.one.one.one", type="AAAA")
- assert "one.one.one.one" in await helpers.resolve("1.1.1.1")
+ dnsengine = DNSEngine(None)
+ assert "1.1.1.1" in await dnsengine.resolve("one.one.one.one")
+ assert "2606:4700:4700::1111" in await dnsengine.resolve("one.one.one.one", type="AAAA")
+ assert "one.one.one.one" in await dnsengine.resolve("1.1.1.1")
for rdtype in ("NS", "SOA", "MX", "TXT"):
- assert len(await helpers.resolve("google.com", type=rdtype)) > 0
+ assert len(await dnsengine.resolve("google.com", type=rdtype)) > 0
# batch resolution
- batch_results = [r async for r in helpers.resolve_batch(["1.1.1.1", "one.one.one.one"])]
+ batch_results = [r async for r in dnsengine.resolve_batch(["1.1.1.1", "one.one.one.one"])]
assert len(batch_results) == 2
batch_results = dict(batch_results)
assert any([x in batch_results["one.one.one.one"] for x in ("1.1.1.1", "1.0.0.1")])
assert "one.one.one.one" in batch_results["1.1.1.1"]
- # "any" type
- resolved = await helpers.resolve("google.com", type="any")
- assert any([helpers.is_subdomain(h) for h in resolved])
+ # custom batch resolution
+ batch_results = [r async for r in dnsengine.resolve_raw_batch([("1.1.1.1", "PTR"), ("one.one.one.one", "A")])]
+ batch_results = [(response.to_text(), response.rdtype.name) for query, (response, errors) in batch_results]
+ assert len(batch_results) == 3
+ assert any(answer == "1.0.0.1" and rdtype == "A" for answer, rdtype in batch_results)
+ assert any(answer == "one.one.one.one." and rdtype == "PTR" for answer, rdtype in batch_results)
# dns cache
- helpers.dns._dns_cache.clear()
- assert hash(f"1.1.1.1:PTR") not in helpers.dns._dns_cache
- assert hash(f"one.one.one.one:A") not in helpers.dns._dns_cache
- assert hash(f"one.one.one.one:AAAA") not in helpers.dns._dns_cache
- await helpers.resolve("1.1.1.1", use_cache=False)
- await helpers.resolve("one.one.one.one", use_cache=False)
- assert hash(f"1.1.1.1:PTR") not in helpers.dns._dns_cache
- assert hash(f"one.one.one.one:A") not in helpers.dns._dns_cache
- assert hash(f"one.one.one.one:AAAA") not in helpers.dns._dns_cache
-
- await helpers.resolve("1.1.1.1")
- assert hash(f"1.1.1.1:PTR") in helpers.dns._dns_cache
- await helpers.resolve("one.one.one.one")
- assert hash(f"one.one.one.one:A") in helpers.dns._dns_cache
- assert hash(f"one.one.one.one:AAAA") in helpers.dns._dns_cache
+ dnsengine._dns_cache.clear()
+ assert hash(f"1.1.1.1:PTR") not in dnsengine._dns_cache
+ assert hash(f"one.one.one.one:A") not in dnsengine._dns_cache
+ assert hash(f"one.one.one.one:AAAA") not in dnsengine._dns_cache
+ await dnsengine.resolve("1.1.1.1", use_cache=False)
+ await dnsengine.resolve("one.one.one.one", use_cache=False)
+ assert hash(f"1.1.1.1:PTR") not in dnsengine._dns_cache
+ assert hash(f"one.one.one.one:A") not in dnsengine._dns_cache
+ assert hash(f"one.one.one.one:AAAA") not in dnsengine._dns_cache
+
+ await dnsengine.resolve("1.1.1.1")
+ assert hash(f"1.1.1.1:PTR") in dnsengine._dns_cache
+ await dnsengine.resolve("one.one.one.one", type="A")
+ assert hash(f"one.one.one.one:A") in dnsengine._dns_cache
+ assert not hash(f"one.one.one.one:AAAA") in dnsengine._dns_cache
+ dnsengine._dns_cache.clear()
+ await dnsengine.resolve("one.one.one.one", type="AAAA")
+ assert hash(f"one.one.one.one:AAAA") in dnsengine._dns_cache
+ assert not hash(f"one.one.one.one:A") in dnsengine._dns_cache
# Ensure events with hosts have resolved_hosts attribute populated
- resolved_hosts_event1 = scan.make_event("one.one.one.one", "DNS_NAME", dummy=True)
- resolved_hosts_event2 = scan.make_event("http://one.one.one.one/", "URL_UNVERIFIED", dummy=True)
- event_tags1, event_whitelisted1, event_blacklisted1, children1 = await scan.helpers.resolve_event(
- resolved_hosts_event1
- )
- event_tags2, event_whitelisted2, event_blacklisted2, children2 = await scan.helpers.resolve_event(
- resolved_hosts_event2
- )
- assert "1.1.1.1" in [str(x) for x in children1["A"]]
- assert "1.1.1.1" in [str(x) for x in children2["A"]]
- assert set(children1.keys()) == set(children2.keys())
-
- dns_config = OmegaConf.create({"dns_resolution": True})
- dns_config = OmegaConf.merge(bbot_config, dns_config)
- scan2 = bbot_scanner("evilcorp.com", config=dns_config)
- mock_dns(
- scan2,
+ await scan._prep()
+ resolved_hosts_event1 = scan.make_event("one.one.one.one", "DNS_NAME", parent=scan.root_event)
+ resolved_hosts_event2 = scan.make_event("http://one.one.one.one/", "URL_UNVERIFIED", parent=scan.root_event)
+ dnsresolve = scan.modules["dnsresolve"]
+ assert hash(resolved_hosts_event1.host) not in dnsresolve._event_cache
+ assert hash(resolved_hosts_event2.host) not in dnsresolve._event_cache
+ await dnsresolve.handle_event(resolved_hosts_event1, {})
+ assert hash(resolved_hosts_event1.host) in dnsresolve._event_cache
+ assert hash(resolved_hosts_event2.host) in dnsresolve._event_cache
+ await dnsresolve.handle_event(resolved_hosts_event2, {})
+ assert "1.1.1.1" in resolved_hosts_event2.resolved_hosts
+ assert "1.1.1.1" in resolved_hosts_event2.dns_children["A"]
+ assert resolved_hosts_event1.resolved_hosts == resolved_hosts_event2.resolved_hosts
+ assert resolved_hosts_event1.dns_children == resolved_hosts_event2.dns_children
+ assert "a-record" in resolved_hosts_event1.tags
+ assert not "a-record" in resolved_hosts_event2.tags
+
+ scan2 = bbot_scanner("evilcorp.com", config={"dns": {"minimal": False}})
+ await scan2.helpers.dns._mock_dns(
{
"evilcorp.com": {"TXT": ['"v=spf1 include:cloudprovider.com ~all"']},
"cloudprovider.com": {"A": ["1.2.3.4"]},
@@ -99,48 +169,51 @@ async def test_dns(bbot_scanner, bbot_config, mock_dns):
@pytest.mark.asyncio
-async def test_wildcards(bbot_scanner, bbot_config):
- scan = bbot_scanner("1.1.1.1", config=bbot_config)
+async def test_wildcards(bbot_scanner):
+ scan = bbot_scanner("1.1.1.1")
helpers = scan.helpers
+ from bbot.core.helpers.dns.engine import DNSEngine
+
+ dnsengine = DNSEngine(None)
+
# wildcards
- wildcard_domains = await helpers.is_wildcard_domain("asdf.github.io")
- assert hash("github.io") in helpers.dns._wildcard_cache
- assert hash("asdf.github.io") in helpers.dns._wildcard_cache
+ wildcard_domains = await dnsengine.is_wildcard_domain("asdf.github.io")
+ assert hash("github.io") in dnsengine._wildcard_cache
+ assert hash("asdf.github.io") in dnsengine._wildcard_cache
assert "github.io" in wildcard_domains
assert "A" in wildcard_domains["github.io"]
assert "SRV" not in wildcard_domains["github.io"]
assert wildcard_domains["github.io"]["A"] and all(helpers.is_ip(r) for r in wildcard_domains["github.io"]["A"])
- helpers.dns._wildcard_cache.clear()
+ dnsengine._wildcard_cache.clear()
- wildcard_rdtypes = await helpers.is_wildcard("blacklanternsecurity.github.io")
+ wildcard_rdtypes = await dnsengine.is_wildcard("blacklanternsecurity.github.io")
assert "A" in wildcard_rdtypes
assert "SRV" not in wildcard_rdtypes
assert wildcard_rdtypes["A"] == (True, "github.io")
- assert hash("github.io") in helpers.dns._wildcard_cache
- assert len(helpers.dns._wildcard_cache[hash("github.io")]) > 0
- helpers.dns._wildcard_cache.clear()
+ assert hash("github.io") in dnsengine._wildcard_cache
+ assert len(dnsengine._wildcard_cache[hash("github.io")]) > 0
+ dnsengine._wildcard_cache.clear()
- wildcard_rdtypes = await helpers.is_wildcard("asdf.asdf.asdf.github.io")
+ wildcard_rdtypes = await dnsengine.is_wildcard("asdf.asdf.asdf.github.io")
assert "A" in wildcard_rdtypes
assert "SRV" not in wildcard_rdtypes
assert wildcard_rdtypes["A"] == (True, "github.io")
- assert hash("github.io") in helpers.dns._wildcard_cache
- assert not hash("asdf.github.io") in helpers.dns._wildcard_cache
- assert not hash("asdf.asdf.github.io") in helpers.dns._wildcard_cache
- assert not hash("asdf.asdf.asdf.github.io") in helpers.dns._wildcard_cache
- assert len(helpers.dns._wildcard_cache[hash("github.io")]) > 0
+ assert hash("github.io") in dnsengine._wildcard_cache
+ assert not hash("asdf.github.io") in dnsengine._wildcard_cache
+ assert not hash("asdf.asdf.github.io") in dnsengine._wildcard_cache
+ assert not hash("asdf.asdf.asdf.github.io") in dnsengine._wildcard_cache
+ assert len(dnsengine._wildcard_cache[hash("github.io")]) > 0
wildcard_event1 = scan.make_event("wat.asdf.fdsa.github.io", "DNS_NAME", dummy=True)
wildcard_event2 = scan.make_event("wats.asd.fdsa.github.io", "DNS_NAME", dummy=True)
wildcard_event3 = scan.make_event("github.io", "DNS_NAME", dummy=True)
# event resolution
- event_tags1, event_whitelisted1, event_blacklisted1, children1 = await scan.helpers.resolve_event(wildcard_event1)
- event_tags2, event_whitelisted2, event_blacklisted2, children2 = await scan.helpers.resolve_event(wildcard_event2)
- event_tags3, event_whitelisted3, event_blacklisted3, children3 = await scan.helpers.resolve_event(wildcard_event3)
- await helpers.handle_wildcard_event(wildcard_event1, children1)
- await helpers.handle_wildcard_event(wildcard_event2, children2)
- await helpers.handle_wildcard_event(wildcard_event3, children3)
+ await scan._prep()
+ dnsresolve = scan.modules["dnsresolve"]
+ await dnsresolve.handle_event(wildcard_event1, {})
+ await dnsresolve.handle_event(wildcard_event2, {})
+ await dnsresolve.handle_event(wildcard_event3, {})
assert "wildcard" in wildcard_event1.tags
assert "a-wildcard" in wildcard_event1.tags
assert "srv-wildcard" not in wildcard_event1.tags
@@ -149,6 +222,195 @@ async def test_wildcards(bbot_scanner, bbot_config):
assert "srv-wildcard" not in wildcard_event2.tags
assert wildcard_event1.data == "_wildcard.github.io"
assert wildcard_event2.data == "_wildcard.github.io"
- assert "wildcard-domain" in wildcard_event3.tags
- assert "a-wildcard-domain" in wildcard_event3.tags
- assert "srv-wildcard-domain" not in wildcard_event3.tags
+ assert wildcard_event3.data == "github.io"
+
+ # dns resolve distance
+ event_distance_0 = scan.make_event("8.8.8.8", module=scan._make_dummy_module_dns("PTR"), parent=scan.root_event)
+ assert event_distance_0.dns_resolve_distance == 0
+ event_distance_1 = scan.make_event(
+ "evilcorp.com", module=scan._make_dummy_module_dns("A"), parent=event_distance_0
+ )
+ assert event_distance_1.dns_resolve_distance == 1
+ event_distance_2 = scan.make_event("1.2.3.4", module=scan._make_dummy_module_dns("PTR"), parent=event_distance_1)
+ assert event_distance_2.dns_resolve_distance == 1
+ event_distance_3 = scan.make_event(
+ "evilcorp.org", module=scan._make_dummy_module_dns("A"), parent=event_distance_2
+ )
+ assert event_distance_3.dns_resolve_distance == 2
+
+ from bbot.scanner import Scanner
+
+ # test with full scan
+ scan2 = Scanner("asdfl.gashdgkjsadgsdf.github.io", whitelist=["github.io"], config={"dns": {"minimal": False}})
+ await scan2._prep()
+ other_event = scan2.make_event(
+ "lkjg.sdfgsg.jgkhajshdsadf.github.io", module=scan2.modules["dnsresolve"], parent=scan2.root_event
+ )
+ await scan2.ingress_module.queue_event(other_event, {})
+ events = [e async for e in scan2.async_start()]
+ assert len(events) == 3
+ assert 1 == len([e for e in events if e.type == "SCAN"])
+ unmodified_wildcard_events = [
+ e for e in events if e.type == "DNS_NAME" and e.data == "asdfl.gashdgkjsadgsdf.github.io"
+ ]
+ assert len(unmodified_wildcard_events) == 1
+ assert unmodified_wildcard_events[0].tags.issuperset(
+ {
+ "a-record",
+ "target",
+ "aaaa-wildcard",
+ "in-scope",
+ "subdomain",
+ "aaaa-record",
+ "wildcard",
+ "a-wildcard",
+ }
+ )
+ modified_wildcard_events = [e for e in events if e.type == "DNS_NAME" and e.data == "_wildcard.github.io"]
+ assert len(modified_wildcard_events) == 1
+ assert modified_wildcard_events[0].tags.issuperset(
+ {
+ "a-record",
+ "aaaa-wildcard",
+ "in-scope",
+ "subdomain",
+ "aaaa-record",
+ "wildcard",
+ "a-wildcard",
+ }
+ )
+ assert modified_wildcard_events[0].host_original == "lkjg.sdfgsg.jgkhajshdsadf.github.io"
+
+ # test with full scan (wildcard detection disabled for domain)
+ scan2 = Scanner(
+ "asdfl.gashdgkjsadgsdf.github.io",
+ whitelist=["github.io"],
+ config={"dns": {"wildcard_ignore": ["github.io"]}},
+ exclude_modules=["cloudcheck"],
+ )
+ await scan2._prep()
+ other_event = scan2.make_event(
+ "lkjg.sdfgsg.jgkhajshdsadf.github.io", module=scan2.modules["dnsresolve"], parent=scan2.root_event
+ )
+ await scan2.ingress_module.queue_event(other_event, {})
+ events = [e async for e in scan2.async_start()]
+ assert len(events) == 3
+ assert 1 == len([e for e in events if e.type == "SCAN"])
+ unmodified_wildcard_events = [e for e in events if e.type == "DNS_NAME" and "_wildcard" not in e.data]
+ assert len(unmodified_wildcard_events) == 2
+ assert 1 == len(
+ [
+ e
+ for e in unmodified_wildcard_events
+ if e.data == "asdfl.gashdgkjsadgsdf.github.io"
+ and e.tags.issuperset(
+ {
+ "target",
+ "a-record",
+ "in-scope",
+ "subdomain",
+ "aaaa-record",
+ }
+ )
+ ]
+ )
+ assert 1 == len(
+ [
+ e
+ for e in unmodified_wildcard_events
+ if e.data == "lkjg.sdfgsg.jgkhajshdsadf.github.io"
+ and e.tags.issuperset(
+ {
+ "a-record",
+ "in-scope",
+ "subdomain",
+ "aaaa-record",
+ }
+ )
+ ]
+ )
+ modified_wildcard_events = [e for e in events if e.type == "DNS_NAME" and e.data == "_wildcard.github.io"]
+ assert len(modified_wildcard_events) == 0
+
+
+@pytest.mark.asyncio
+async def test_dns_raw_records(bbot_scanner):
+
+ from bbot.modules.base import BaseModule
+
+ class DummyModule(BaseModule):
+ watched_events = ["*"]
+
+ async def setup(self):
+ self.events = []
+ return True
+
+ async def handle_event(self, event):
+ self.events.append(event)
+
+ # scan without omitted event type
+ scan = bbot_scanner("one.one.one.one", "1.1.1.1", config={"dns": {"minimal": False}, "omit_event_types": []})
+ await scan.helpers.dns._mock_dns(mock_records)
+ dummy_module = DummyModule(scan)
+ scan.modules["dummy_module"] = dummy_module
+ events = [e async for e in scan.async_start()]
+ assert 1 == len([e for e in events if e.type == "RAW_DNS_RECORD"])
+ assert 1 == len(
+ [
+ e
+ for e in events
+ if e.type == "RAW_DNS_RECORD"
+ and e.host == "one.one.one.one"
+ and e.data["host"] == "one.one.one.one"
+ and e.data["type"] == "TXT"
+ and e.data["answer"]
+ == '"v=spf1 ip4:103.151.192.0/23 ip4:185.12.80.0/22 ip4:188.172.128.0/20 ip4:192.161.144.0/20 ip4:216.198.0.0/18 ~all"'
+ and e.discovery_context == "TXT lookup on one.one.one.one produced RAW_DNS_RECORD"
+ ]
+ )
+ assert 1 == len(
+ [
+ e
+ for e in dummy_module.events
+ if e.type == "RAW_DNS_RECORD"
+ and e.host == "one.one.one.one"
+ and e.data["host"] == "one.one.one.one"
+ and e.data["type"] == "TXT"
+ and e.data["answer"]
+ == '"v=spf1 ip4:103.151.192.0/23 ip4:185.12.80.0/22 ip4:188.172.128.0/20 ip4:192.161.144.0/20 ip4:216.198.0.0/18 ~all"'
+ and e.discovery_context == "TXT lookup on one.one.one.one produced RAW_DNS_RECORD"
+ ]
+ )
+ # scan with omitted event type
+ scan = bbot_scanner("one.one.one.one", config={"dns": {"minimal": False}, "omit_event_types": ["RAW_DNS_RECORD"]})
+ await scan.helpers.dns._mock_dns(mock_records)
+ dummy_module = DummyModule(scan)
+ scan.modules["dummy_module"] = dummy_module
+ events = [e async for e in scan.async_start()]
+ # no raw records should be emitted
+ assert 0 == len([e for e in events if e.type == "RAW_DNS_RECORD"])
+ assert 0 == len([e for e in dummy_module.events if e.type == "RAW_DNS_RECORD"])
+
+ # scan with watching module
+ DummyModule.watched_events = ["RAW_DNS_RECORD"]
+ scan = bbot_scanner("one.one.one.one", config={"dns": {"minimal": False}, "omit_event_types": ["RAW_DNS_RECORD"]})
+ await scan.helpers.dns._mock_dns(mock_records)
+ dummy_module = DummyModule(scan)
+ scan.modules["dummy_module"] = dummy_module
+ events = [e async for e in scan.async_start()]
+ # no raw records should be ouptut
+ assert 0 == len([e for e in events if e.type == "RAW_DNS_RECORD"])
+ # but they should still make it to the module
+ assert 1 == len(
+ [
+ e
+ for e in dummy_module.events
+ if e.type == "RAW_DNS_RECORD"
+ and e.host == "one.one.one.one"
+ and e.data["host"] == "one.one.one.one"
+ and e.data["type"] == "TXT"
+ and e.data["answer"]
+ == '"v=spf1 ip4:103.151.192.0/23 ip4:185.12.80.0/22 ip4:188.172.128.0/20 ip4:192.161.144.0/20 ip4:216.198.0.0/18 ~all"'
+ and e.discovery_context == "TXT lookup on one.one.one.one produced RAW_DNS_RECORD"
+ ]
+ )
diff --git a/bbot/test/test_step_1/test_docs.py b/bbot/test/test_step_1/test_docs.py
index 6b00d2d0d..a86947ff0 100644
--- a/bbot/test/test_step_1/test_docs.py
+++ b/bbot/test/test_step_1/test_docs.py
@@ -1,5 +1,4 @@
-from bbot.scripts.docs import update_docs
-
-
def test_docs():
+ from bbot.scripts.docs import update_docs
+
update_docs()
diff --git a/bbot/test/test_step_1/test_engine.py b/bbot/test/test_step_1/test_engine.py
new file mode 100644
index 000000000..a8a4156d1
--- /dev/null
+++ b/bbot/test/test_step_1/test_engine.py
@@ -0,0 +1,146 @@
+from ..bbot_fixtures import *
+
+
+@pytest.mark.asyncio
+async def test_engine():
+ from bbot.core.engine import EngineClient, EngineServer
+
+ counter = 0
+ yield_cancelled = False
+ yield_errored = False
+ return_started = False
+ return_finished = False
+ return_cancelled = False
+ return_errored = False
+
+ class TestEngineServer(EngineServer):
+
+ CMDS = {
+ 0: "return_thing",
+ 1: "yield_stuff",
+ }
+
+ async def return_thing(self, n):
+ nonlocal return_started
+ nonlocal return_finished
+ nonlocal return_cancelled
+ nonlocal return_errored
+ try:
+ return_started = True
+ await asyncio.sleep(n)
+ return_finished = True
+ return f"thing{n}"
+ except asyncio.CancelledError:
+ return_cancelled = True
+ raise
+ except Exception:
+ return_errored = True
+ raise
+
+ async def yield_stuff(self, n):
+ nonlocal counter
+ nonlocal yield_cancelled
+ nonlocal yield_errored
+ try:
+ for i in range(n):
+ yield f"thing{i}"
+ counter += 1
+ await asyncio.sleep(0.1)
+ except asyncio.CancelledError:
+ yield_cancelled = True
+ raise
+ except Exception:
+ yield_errored = True
+ raise
+
+ class TestEngineClient(EngineClient):
+
+ SERVER_CLASS = TestEngineServer
+
+ async def return_thing(self, n):
+ return await self.run_and_return("return_thing", n)
+
+ async def yield_stuff(self, n):
+ async for _ in self.run_and_yield("yield_stuff", n):
+ yield _
+
+ test_engine = TestEngineClient()
+
+ # test return functionality
+ return_res = await test_engine.return_thing(1)
+ assert return_res == "thing1"
+
+ # test async generator
+ assert counter == 0
+ assert yield_cancelled == False
+ yield_res = [r async for r in test_engine.yield_stuff(13)]
+ assert yield_res == [f"thing{i}" for i in range(13)]
+ assert len(yield_res) == 13
+ assert counter == 13
+
+ # test async generator with cancellation
+ counter = 0
+ yield_cancelled = False
+ yield_errored = False
+ agen = test_engine.yield_stuff(1000)
+ async for r in agen:
+ if counter > 10:
+ await agen.aclose()
+ break
+ await asyncio.sleep(5)
+ assert yield_cancelled == True
+ assert yield_errored == False
+ assert counter < 15
+
+ # test async generator with error
+ yield_cancelled = False
+ yield_errored = False
+ agen = test_engine.yield_stuff(None)
+ with pytest.raises(BBOTEngineError):
+ async for _ in agen:
+ pass
+ assert yield_cancelled == False
+ assert yield_errored == True
+
+ # test return with cancellation
+ return_started = False
+ return_finished = False
+ return_cancelled = False
+ return_errored = False
+ task = asyncio.create_task(test_engine.return_thing(2))
+ await asyncio.sleep(1)
+ task.cancel()
+ with pytest.raises(asyncio.CancelledError):
+ await task
+ await asyncio.sleep(0.1)
+ assert return_started == True
+ assert return_finished == False
+ assert return_cancelled == True
+ assert return_errored == False
+
+ # test return with late cancellation
+ return_started = False
+ return_finished = False
+ return_cancelled = False
+ return_errored = False
+ task = asyncio.create_task(test_engine.return_thing(1))
+ await asyncio.sleep(2)
+ task.cancel()
+ result = await task
+ assert result == "thing1"
+ assert return_started == True
+ assert return_finished == True
+ assert return_cancelled == False
+ assert return_errored == False
+
+ # test return with error
+ return_started = False
+ return_finished = False
+ return_cancelled = False
+ return_errored = False
+ with pytest.raises(BBOTEngineError):
+ result = await test_engine.return_thing(None)
+ assert return_started == True
+ assert return_finished == False
+ assert return_cancelled == False
+ assert return_errored == True
diff --git a/bbot/test/test_step_1/test_events.py b/bbot/test/test_step_1/test_events.py
index b6846d6e2..8b7503d13 100644
--- a/bbot/test/test_step_1/test_events.py
+++ b/bbot/test/test_step_1/test_events.py
@@ -3,10 +3,15 @@
import ipaddress
from ..bbot_fixtures import *
+from bbot.scanner import Scanner
@pytest.mark.asyncio
-async def test_events(events, scan, helpers, bbot_config):
+async def test_events(events, helpers):
+
+ scan = Scanner()
+ await scan._prep()
+
assert events.ipv4.type == "IP_ADDRESS"
assert events.ipv6.type == "IP_ADDRESS"
assert events.netv4.type == "IP_RANGE"
@@ -100,7 +105,7 @@ async def test_events(events, scan, helpers, bbot_config):
# http response
assert events.http_response.host == "example.com"
assert events.http_response.port == 80
- assert events.http_response.parsed.scheme == "http"
+ assert events.http_response.parsed_url.scheme == "http"
assert events.http_response.with_port().geturl() == "http://example.com:80/"
http_response = scan.make_event(
@@ -159,8 +164,9 @@ async def test_events(events, scan, helpers, bbot_config):
assert events.ipv6_url_unverified.host == ipaddress.ip_address("2001:4860:4860::8888")
assert events.ipv6_url_unverified.port == 443
- javascript_event = scan.make_event("http://evilcorp.com/asdf/a.js?b=c#d", "URL_UNVERIFIED", dummy=True)
+ javascript_event = scan.make_event("http://evilcorp.com/asdf/a.js?b=c#d", "URL_UNVERIFIED", parent=scan.root_event)
assert "extension-js" in javascript_event.tags
+ await scan.ingress_module.handle_event(javascript_event, {})
assert "httpx-only" in javascript_event.tags
# scope distance
@@ -168,70 +174,76 @@ async def test_events(events, scan, helpers, bbot_config):
assert event1._scope_distance == -1
event1.scope_distance = 0
assert event1._scope_distance == 0
- event2 = scan.make_event("2.3.4.5", source=event1)
+ event2 = scan.make_event("2.3.4.5", parent=event1)
assert event2._scope_distance == 1
- event3 = scan.make_event("3.4.5.6", source=event2)
+ event3 = scan.make_event("3.4.5.6", parent=event2)
assert event3._scope_distance == 2
- event4 = scan.make_event("3.4.5.6", source=event3)
+ event4 = scan.make_event("3.4.5.6", parent=event3)
assert event4._scope_distance == 2
- event5 = scan.make_event("4.5.6.7", source=event4)
+ event5 = scan.make_event("4.5.6.7", parent=event4)
assert event5._scope_distance == 3
- url_1 = scan.make_event("https://127.0.0.1/asdf", "URL_UNVERIFIED", source=scan.root_event)
+ url_1 = scan.make_event("https://127.0.0.1/asdf", "URL_UNVERIFIED", parent=scan.root_event)
assert url_1.scope_distance == 1
- url_2 = scan.make_event("https://127.0.0.1/test", "URL_UNVERIFIED", source=url_1)
+ url_2 = scan.make_event("https://127.0.0.1/test", "URL_UNVERIFIED", parent=url_1)
assert url_2.scope_distance == 1
- url_3 = scan.make_event("https://127.0.0.2/asdf", "URL_UNVERIFIED", source=url_1)
+ url_3 = scan.make_event("https://127.0.0.2/asdf", "URL_UNVERIFIED", parent=url_1)
assert url_3.scope_distance == 2
- org_stub_1 = scan.make_event("STUB1", "ORG_STUB", source=scan.root_event)
+ org_stub_1 = scan.make_event("STUB1", "ORG_STUB", parent=scan.root_event)
org_stub_1.scope_distance == 1
- org_stub_2 = scan.make_event("STUB2", "ORG_STUB", source=org_stub_1)
+ org_stub_2 = scan.make_event("STUB2", "ORG_STUB", parent=org_stub_1)
org_stub_2.scope_distance == 2
# internal event tracking
root_event = scan.make_event("0.0.0.0", dummy=True)
- internal_event1 = scan.make_event("1.2.3.4", source=root_event, internal=True)
+ internal_event1 = scan.make_event("1.2.3.4", parent=root_event, internal=True)
assert internal_event1._internal == True
assert "internal" in internal_event1.tags
# tag inheritance
for tag in ("affiliate", "mutation-1"):
- affiliate_event = scan.make_event("1.2.3.4", source=root_event, tags=tag)
+ affiliate_event = scan.make_event("1.2.3.4", parent=root_event, tags=tag)
assert tag in affiliate_event.tags
- affiliate_event2 = scan.make_event("1.2.3.4:88", source=affiliate_event)
- affiliate_event3 = scan.make_event("4.3.2.1:88", source=affiliate_event)
+ affiliate_event2 = scan.make_event("1.2.3.4:88", parent=affiliate_event)
+ affiliate_event3 = scan.make_event("4.3.2.1:88", parent=affiliate_event)
assert tag in affiliate_event2.tags
assert tag not in affiliate_event3.tags
+ # discovery context
+ event = scan.make_event(
+ "127.0.0.1", parent=scan.root_event, context="something discovered {event.type}: {event.data}"
+ )
+ assert event.discovery_context == "something discovered IP_ADDRESS: 127.0.0.1"
+
# updating an already-created event with make_event()
# updating tags
- event1 = scan.make_event("127.0.0.1", source=scan.root_event)
+ event1 = scan.make_event("127.0.0.1", parent=scan.root_event)
updated_event = scan.make_event(event1, tags="asdf")
assert "asdf" not in event1.tags
assert "asdf" in updated_event.tags
- # updating source
- event2 = scan.make_event("127.0.0.1", source=scan.root_event)
- updated_event = scan.make_event(event2, source=event1)
- assert event2.source == scan.root_event
- assert updated_event.source == event1
+ # updating parent
+ event2 = scan.make_event("127.0.0.1", parent=scan.root_event)
+ updated_event = scan.make_event(event2, parent=event1)
+ assert event2.parent == scan.root_event
+ assert updated_event.parent == event1
# updating module
- event3 = scan.make_event("127.0.0.1", source=scan.root_event)
+ event3 = scan.make_event("127.0.0.1", parent=scan.root_event)
updated_event = scan.make_event(event3, internal=True)
assert event3.internal == False
assert updated_event.internal == True
# event sorting
- parent1 = scan.make_event("127.0.0.1", source=scan.root_event)
- parent2 = scan.make_event("127.0.0.1", source=scan.root_event)
- parent2_child1 = scan.make_event("127.0.0.1", source=parent2)
- parent1_child1 = scan.make_event("127.0.0.1", source=parent1)
- parent1_child2 = scan.make_event("127.0.0.1", source=parent1)
- parent1_child2_child1 = scan.make_event("127.0.0.1", source=parent1_child2)
- parent1_child2_child2 = scan.make_event("127.0.0.1", source=parent1_child2)
- parent1_child1_child1 = scan.make_event("127.0.0.1", source=parent1_child1)
- parent2_child2 = scan.make_event("127.0.0.1", source=parent2)
- parent1_child2_child1_child1 = scan.make_event("127.0.0.1", source=parent1_child2_child1)
+ parent1 = scan.make_event("127.0.0.1", parent=scan.root_event)
+ parent2 = scan.make_event("127.0.0.1", parent=scan.root_event)
+ parent2_child1 = scan.make_event("127.0.0.1", parent=parent2)
+ parent1_child1 = scan.make_event("127.0.0.1", parent=parent1)
+ parent1_child2 = scan.make_event("127.0.0.1", parent=parent1)
+ parent1_child2_child1 = scan.make_event("127.0.0.1", parent=parent1_child2)
+ parent1_child2_child2 = scan.make_event("127.0.0.1", parent=parent1_child2)
+ parent1_child1_child1 = scan.make_event("127.0.0.1", parent=parent1_child1)
+ parent2_child2 = scan.make_event("127.0.0.1", parent=parent2)
+ parent1_child2_child1_child1 = scan.make_event("127.0.0.1", parent=parent1_child2_child1)
sortable_events = {
"parent1": parent1,
@@ -392,64 +404,271 @@ async def test_events(events, scan, helpers, bbot_config):
# test event serialization
from bbot.core.event import event_from_json
- db_event = scan.make_event("evilcorp.com", dummy=True)
+ db_event = scan.make_event("evilcorp.com:80", parent=scan.root_event, context="test context")
db_event._resolved_hosts = {"127.0.0.1"}
db_event.scope_distance = 1
+ assert db_event.discovery_context == "test context"
+ assert db_event.discovery_path == ["test context"]
timestamp = db_event.timestamp.timestamp()
json_event = db_event.json()
assert json_event["scope_distance"] == 1
- assert json_event["data"] == "evilcorp.com"
- assert json_event["type"] == "DNS_NAME"
+ assert json_event["data"] == "evilcorp.com:80"
+ assert json_event["type"] == "OPEN_TCP_PORT"
+ assert json_event["host"] == "evilcorp.com"
assert json_event["timestamp"] == timestamp
+ assert json_event["discovery_context"] == "test context"
+ assert json_event["discovery_path"] == ["test context"]
reconstituted_event = event_from_json(json_event)
assert reconstituted_event.scope_distance == 1
assert reconstituted_event.timestamp.timestamp() == timestamp
- assert reconstituted_event.data == "evilcorp.com"
- assert reconstituted_event.type == "DNS_NAME"
+ assert reconstituted_event.data == "evilcorp.com:80"
+ assert reconstituted_event.type == "OPEN_TCP_PORT"
+ assert reconstituted_event.host == "evilcorp.com"
+ assert reconstituted_event.discovery_context == "test context"
+ assert reconstituted_event.discovery_path == ["test context"]
assert "127.0.0.1" in reconstituted_event.resolved_hosts
+ hostless_event = scan.make_event("asdf", "ASDF", dummy=True)
+ hostless_event_json = hostless_event.json()
+ assert hostless_event_json["type"] == "ASDF"
+ assert hostless_event_json["data"] == "asdf"
+ assert not "host" in hostless_event_json
# SIEM-friendly serialize/deserialize
json_event_siemfriendly = db_event.json(siem_friendly=True)
assert json_event_siemfriendly["scope_distance"] == 1
- assert json_event_siemfriendly["data"] == {"DNS_NAME": "evilcorp.com"}
- assert json_event_siemfriendly["type"] == "DNS_NAME"
+ assert json_event_siemfriendly["data"] == {"OPEN_TCP_PORT": "evilcorp.com:80"}
+ assert json_event_siemfriendly["type"] == "OPEN_TCP_PORT"
+ assert json_event_siemfriendly["host"] == "evilcorp.com"
assert json_event_siemfriendly["timestamp"] == timestamp
reconstituted_event2 = event_from_json(json_event_siemfriendly, siem_friendly=True)
assert reconstituted_event2.scope_distance == 1
assert reconstituted_event2.timestamp.timestamp() == timestamp
- assert reconstituted_event2.data == "evilcorp.com"
- assert reconstituted_event2.type == "DNS_NAME"
+ assert reconstituted_event2.data == "evilcorp.com:80"
+ assert reconstituted_event2.type == "OPEN_TCP_PORT"
+ assert reconstituted_event2.host == "evilcorp.com"
assert "127.0.0.1" in reconstituted_event2.resolved_hosts
- http_response = scan.make_event(httpx_response, "HTTP_RESPONSE", source=scan.root_event)
- assert http_response.source_id == scan.root_event.id
+ http_response = scan.make_event(httpx_response, "HTTP_RESPONSE", parent=scan.root_event)
+ assert http_response.parent_id == scan.root_event.id
assert http_response.data["input"] == "http://example.com:80"
json_event = http_response.json(mode="graph")
assert isinstance(json_event["data"], str)
json_event = http_response.json()
assert isinstance(json_event["data"], dict)
assert json_event["type"] == "HTTP_RESPONSE"
- assert json_event["source"] == scan.root_event.id
+ assert json_event["host"] == "example.com"
+ assert json_event["parent"] == scan.root_event.id
reconstituted_event = event_from_json(json_event)
assert isinstance(reconstituted_event.data, dict)
assert reconstituted_event.data["input"] == "http://example.com:80"
+ assert reconstituted_event.host == "example.com"
assert reconstituted_event.type == "HTTP_RESPONSE"
- assert reconstituted_event.source_id == scan.root_event.id
+ assert reconstituted_event.parent_id == scan.root_event.id
- event_1 = scan.make_event("127.0.0.1", source=scan.root_event)
- event_2 = scan.make_event("127.0.0.2", source=event_1)
- event_3 = scan.make_event("127.0.0.3", source=event_2)
+ event_1 = scan.make_event("127.0.0.1", parent=scan.root_event)
+ event_2 = scan.make_event("127.0.0.2", parent=event_1)
+ event_3 = scan.make_event("127.0.0.3", parent=event_2)
event_3._omit = True
- event_4 = scan.make_event("127.0.0.4", source=event_3)
- event_5 = scan.make_event("127.0.0.5", source=event_4)
- assert event_5.get_sources() == [event_4, event_3, event_2, event_1, scan.root_event]
- assert event_5.get_sources(omit=True) == [event_4, event_2, event_1, scan.root_event]
+ event_4 = scan.make_event("127.0.0.4", parent=event_3)
+ event_5 = scan.make_event("127.0.0.5", parent=event_4)
+ assert event_5.get_parents() == [event_4, event_3, event_2, event_1, scan.root_event]
+ assert event_5.get_parents(omit=True) == [event_4, event_2, event_1, scan.root_event]
+
+ # test host backup
+ host_event = scan.make_event("asdf.evilcorp.com", "DNS_NAME", parent=scan.root_event)
+ assert host_event.host_original == "asdf.evilcorp.com"
+ host_event.host = "_wildcard.evilcorp.com"
+ assert host_event.host == "_wildcard.evilcorp.com"
+ assert host_event.host_original == "asdf.evilcorp.com"
# test storage bucket validation
bucket_event = scan.make_event(
{"name": "ASDF.s3.amazonaws.com", "url": "https://ASDF.s3.amazonaws.com"},
"STORAGE_BUCKET",
- source=scan.root_event,
+ parent=scan.root_event,
)
assert bucket_event.data["name"] == "asdf.s3.amazonaws.com"
assert bucket_event.data["url"] == "https://asdf.s3.amazonaws.com/"
+
+ # test module sequence
+ module = scan._make_dummy_module("mymodule")
+ parent_event_1 = scan.make_event("127.0.0.1", module=module, parent=scan.root_event)
+ assert str(parent_event_1.module) == "mymodule"
+ assert str(parent_event_1.module_sequence) == "mymodule"
+ parent_event_2 = scan.make_event("127.0.0.2", module=module, parent=parent_event_1)
+ assert str(parent_event_2.module) == "mymodule"
+ assert str(parent_event_2.module_sequence) == "mymodule"
+ parent_event_3 = scan.make_event("127.0.0.3", module=module, parent=parent_event_2)
+ assert str(parent_event_3.module) == "mymodule"
+ assert str(parent_event_3.module_sequence) == "mymodule"
+
+ module = scan._make_dummy_module("mymodule")
+ parent_event_1 = scan.make_event("127.0.0.1", module=module, parent=scan.root_event)
+ parent_event_1._omit = True
+ assert str(parent_event_1.module) == "mymodule"
+ assert str(parent_event_1.module_sequence) == "mymodule"
+ parent_event_2 = scan.make_event("127.0.0.2", module=module, parent=parent_event_1)
+ parent_event_2._omit = True
+ assert str(parent_event_2.module) == "mymodule"
+ assert str(parent_event_2.module_sequence) == "mymodule->mymodule"
+ parent_event_3 = scan.make_event("127.0.0.3", module=module, parent=parent_event_2)
+ assert str(parent_event_3.module) == "mymodule"
+ assert str(parent_event_3.module_sequence) == "mymodule->mymodule->mymodule"
+
+
+@pytest.mark.asyncio
+async def test_event_discovery_context():
+
+ from bbot.modules.base import BaseModule
+
+ scan = Scanner("evilcorp.com")
+ await scan.helpers.dns._mock_dns(
+ {
+ "evilcorp.com": {"A": ["1.2.3.4"]},
+ "one.evilcorp.com": {"A": ["1.2.3.4"]},
+ "two.evilcorp.com": {"A": ["1.2.3.4"]},
+ "three.evilcorp.com": {"A": ["1.2.3.4"]},
+ "four.evilcorp.com": {"A": ["1.2.3.4"]},
+ }
+ )
+ await scan._prep()
+
+ dummy_module_1 = scan._make_dummy_module("module_1")
+ dummy_module_2 = scan._make_dummy_module("module_2")
+
+ class DummyModule(BaseModule):
+ watched_events = ["DNS_NAME"]
+ _name = "dummy_module"
+
+ async def handle_event(self, event):
+ new_event = None
+ if event.data == "evilcorp.com":
+ new_event = scan.make_event(
+ "one.evilcorp.com",
+ "DNS_NAME",
+ event,
+ context="{module} invoked forbidden magick to discover {event.type} {event.data}",
+ module=dummy_module_1,
+ )
+ elif event.data == "one.evilcorp.com":
+ new_event = scan.make_event(
+ "two.evilcorp.com",
+ "DNS_NAME",
+ event,
+ context="{module} pledged its allegiance to cthulu and was awarded {event.type} {event.data}",
+ module=dummy_module_1,
+ )
+ elif event.data == "two.evilcorp.com":
+ new_event = scan.make_event(
+ "three.evilcorp.com",
+ "DNS_NAME",
+ event,
+ context="{module} asked nicely and was given {event.type} {event.data}",
+ module=dummy_module_2,
+ )
+ elif event.data == "three.evilcorp.com":
+ new_event = scan.make_event(
+ "four.evilcorp.com",
+ "DNS_NAME",
+ event,
+ context="{module} used brute force to obtain {event.type} {event.data}",
+ module=dummy_module_2,
+ )
+ if new_event is not None:
+ await self.emit_event(new_event)
+
+ dummy_module = DummyModule(scan)
+
+ scan.modules["dummy_module"] = dummy_module
+
+ test_event = dummy_module.make_event("evilcorp.com", "DNS_NAME", parent=scan.root_event)
+ assert test_event.discovery_context == "dummy_module discovered DNS_NAME: evilcorp.com"
+
+ events = [e async for e in scan.async_start()]
+ assert len(events) == 6
+
+ assert 1 == len(
+ [
+ e
+ for e in events
+ if e.type == "DNS_NAME"
+ and e.data == "evilcorp.com"
+ and e.discovery_context == f"Scan {scan.name} seeded with DNS_NAME: evilcorp.com"
+ and e.discovery_path == [f"Scan {scan.name} seeded with DNS_NAME: evilcorp.com"]
+ ]
+ )
+ assert 1 == len(
+ [
+ e
+ for e in events
+ if e.type == "DNS_NAME"
+ and e.data == "one.evilcorp.com"
+ and e.discovery_context == "module_1 invoked forbidden magick to discover DNS_NAME one.evilcorp.com"
+ and e.discovery_path
+ == [
+ f"Scan {scan.name} seeded with DNS_NAME: evilcorp.com",
+ "module_1 invoked forbidden magick to discover DNS_NAME one.evilcorp.com",
+ ]
+ ]
+ )
+ assert 1 == len(
+ [
+ e
+ for e in events
+ if e.type == "DNS_NAME"
+ and e.data == "two.evilcorp.com"
+ and e.discovery_context
+ == "module_1 pledged its allegiance to cthulu and was awarded DNS_NAME two.evilcorp.com"
+ and e.discovery_path
+ == [
+ f"Scan {scan.name} seeded with DNS_NAME: evilcorp.com",
+ "module_1 invoked forbidden magick to discover DNS_NAME one.evilcorp.com",
+ "module_1 pledged its allegiance to cthulu and was awarded DNS_NAME two.evilcorp.com",
+ ]
+ ]
+ )
+ assert 1 == len(
+ [
+ e
+ for e in events
+ if e.type == "DNS_NAME"
+ and e.data == "three.evilcorp.com"
+ and e.discovery_context == "module_2 asked nicely and was given DNS_NAME three.evilcorp.com"
+ and e.discovery_path
+ == [
+ f"Scan {scan.name} seeded with DNS_NAME: evilcorp.com",
+ "module_1 invoked forbidden magick to discover DNS_NAME one.evilcorp.com",
+ "module_1 pledged its allegiance to cthulu and was awarded DNS_NAME two.evilcorp.com",
+ "module_2 asked nicely and was given DNS_NAME three.evilcorp.com",
+ ]
+ ]
+ )
+ final_path = [
+ f"Scan {scan.name} seeded with DNS_NAME: evilcorp.com",
+ "module_1 invoked forbidden magick to discover DNS_NAME one.evilcorp.com",
+ "module_1 pledged its allegiance to cthulu and was awarded DNS_NAME two.evilcorp.com",
+ "module_2 asked nicely and was given DNS_NAME three.evilcorp.com",
+ "module_2 used brute force to obtain DNS_NAME four.evilcorp.com",
+ ]
+ final_event = [
+ e
+ for e in events
+ if e.type == "DNS_NAME"
+ and e.data == "four.evilcorp.com"
+ and e.discovery_context == "module_2 used brute force to obtain DNS_NAME four.evilcorp.com"
+ and e.discovery_path == final_path
+ ]
+ assert 1 == len(final_event)
+ j = final_event[0].json()
+ assert j["discovery_path"] == final_path
+
+ # test to make sure this doesn't come back
+ # https://github.com/blacklanternsecurity/bbot/issues/1498
+ scan = Scanner("http://blacklanternsecurity.com", config={"dns": {"minimal": False}})
+ await scan.helpers.dns._mock_dns(
+ {"blacklanternsecurity.com": {"TXT": ["blsops.com"], "A": ["127.0.0.1"]}, "blsops.com": {"A": ["127.0.0.1"]}}
+ )
+ events = [e async for e in scan.async_start()]
+ blsops_event = [e for e in events if e.type == "DNS_NAME" and e.data == "blsops.com"]
+ assert len(blsops_event) == 1
+ assert blsops_event[0].discovery_path[1] == "URL_UNVERIFIED has host DNS_NAME: blacklanternsecurity.com"
diff --git a/bbot/test/test_step_1/test_files.py b/bbot/test/test_step_1/test_files.py
index be52b1cd2..ed9bc0a33 100644
--- a/bbot/test/test_step_1/test_files.py
+++ b/bbot/test/test_step_1/test_files.py
@@ -4,8 +4,8 @@
@pytest.mark.asyncio
-async def test_files(bbot_scanner, bbot_config):
- scan1 = bbot_scanner(config=bbot_config)
+async def test_files(bbot_scanner):
+ scan1 = bbot_scanner()
# tempfile
tempfile = scan1.helpers.tempfile(("line1", "line2"), pipe=False)
diff --git a/bbot/test/test_step_1/test_helpers.py b/bbot/test/test_step_1/test_helpers.py
index 0045c7652..a289f2ff2 100644
--- a/bbot/test/test_step_1/test_helpers.py
+++ b/bbot/test/test_step_1/test_helpers.py
@@ -6,7 +6,7 @@
@pytest.mark.asyncio
-async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_config, bbot_httpserver):
+async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_httpserver):
### URL ###
bad_urls = (
"http://e.co/index.html",
@@ -103,17 +103,7 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_config, bbot_https
assert helpers.domain_stem("evilcorp.co.uk") == "evilcorp"
assert helpers.domain_stem("www.evilcorp.co.uk") == "www.evilcorp"
- assert helpers.host_in_host("www.evilcorp.com", "evilcorp.com") == True
- assert helpers.host_in_host("asdf.www.evilcorp.com", "evilcorp.com") == True
- assert helpers.host_in_host("evilcorp.com", "www.evilcorp.com") == False
- assert helpers.host_in_host("evilcorp.com", "evilcorp.com") == True
- assert helpers.host_in_host("evilcorp.com", "eevilcorp.com") == False
- assert helpers.host_in_host("eevilcorp.com", "evilcorp.com") == False
- assert helpers.host_in_host("evilcorp.com", "evilcorp") == False
- assert helpers.host_in_host("evilcorp", "evilcorp.com") == False
- assert helpers.host_in_host("evilcorp.com", "com") == True
-
- assert tuple(helpers.extract_emails("asdf@asdf.com\nT@t.Com&a=a@a.com__ b@b.com")) == (
+ assert tuple(await helpers.re.extract_emails("asdf@asdf.com\nT@t.Com&a=a@a.com__ b@b.com")) == (
"asdf@asdf.com",
"t@t.com",
"a@a.com",
@@ -188,7 +178,11 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_config, bbot_https
assert helpers.subdomain_depth("a.evilcorp.com") == 1
assert helpers.subdomain_depth("a.s.d.f.evilcorp.notreal") == 4
+ assert helpers.split_host_port("http://evilcorp.co.uk") == ("evilcorp.co.uk", 80)
assert helpers.split_host_port("https://evilcorp.co.uk") == ("evilcorp.co.uk", 443)
+ assert helpers.split_host_port("ws://evilcorp.co.uk") == ("evilcorp.co.uk", 80)
+ assert helpers.split_host_port("wss://evilcorp.co.uk") == ("evilcorp.co.uk", 443)
+ assert helpers.split_host_port("WSS://evilcorp.co.uk") == ("evilcorp.co.uk", 443)
assert helpers.split_host_port("http://evilcorp.co.uk:666") == ("evilcorp.co.uk", 666)
assert helpers.split_host_port("evilcorp.co.uk:666") == ("evilcorp.co.uk", 666)
assert helpers.split_host_port("evilcorp.co.uk") == ("evilcorp.co.uk", None)
@@ -224,7 +218,7 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_config, bbot_https
assert helpers.get_file_extension("/etc/passwd") == ""
assert helpers.tagify("HttP -_Web Title-- ") == "http-web-title"
- tagged_event = scan.make_event("127.0.0.1", source=scan.root_event, tags=["HttP web -__- title "])
+ tagged_event = scan.make_event("127.0.0.1", parent=scan.root_event, tags=["HttP web -__- title "])
assert "http-web-title" in tagged_event.tags
tagged_event.remove_tag("http-web-title")
assert "http-web-title" not in tagged_event.tags
@@ -254,6 +248,12 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_config, bbot_https
"https://www.evilcorp.com/fdsa",
}
+ replaced = helpers.search_format_dict(
+ {"asdf": [{"wat": {"here": "#{replaceme}!"}}, {500: True}]}, replaceme="asdf"
+ )
+ assert replaced["asdf"][1][500] == True
+ assert replaced["asdf"][0]["wat"]["here"] == "asdf!"
+
filtered_dict = helpers.filter_dict(
{"modules": {"c99": {"api_key": "1234", "filterme": "asdf"}, "ipneighbor": {"test": "test"}}}, "api_key"
)
@@ -321,12 +321,6 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_config, bbot_https
assert "secret" not in cleaned_dict4["modules"]["ipneighbor"]
assert "asdf" in cleaned_dict4["modules"]["ipneighbor"]
- replaced = helpers.search_format_dict(
- {"asdf": [{"wat": {"here": "#{replaceme}!"}}, {500: True}]}, replaceme="asdf"
- )
- assert replaced["asdf"][1][500] == True
- assert replaced["asdf"][0]["wat"]["here"] == "asdf!"
-
assert helpers.split_list([1, 2, 3, 4, 5]) == [[1, 2], [3, 4, 5]]
assert list(helpers.grouper("ABCDEFG", 3)) == [["A", "B", "C"], ["D", "E", "F"], ["G"]]
@@ -393,6 +387,18 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_config, bbot_https
assert helpers.validators.soft_validate("!@#$", "port") == False
with pytest.raises(ValueError):
helpers.validators.validate_port("asdf")
+ # top tcp ports
+ top_tcp_ports = helpers.top_tcp_ports(100)
+ assert len(top_tcp_ports) == 100
+ assert len(set(top_tcp_ports)) == 100
+ top_tcp_ports = helpers.top_tcp_ports(800000)
+ assert top_tcp_ports[:10] == [80, 23, 443, 21, 22, 25, 3389, 110, 445, 139]
+ assert top_tcp_ports[-10:] == [65526, 65527, 65528, 65529, 65530, 65531, 65532, 65533, 65534, 65535]
+ assert len(top_tcp_ports) == 65535
+ assert len(set(top_tcp_ports)) == 65535
+ assert all([isinstance(i, int) for i in top_tcp_ports])
+ top_tcp_ports = helpers.top_tcp_ports(10, as_string=True)
+ assert top_tcp_ports == "80,23,443,21,22,25,3389,110,445,139"
# urls
assert helpers.validators.validate_url(" httP://evilcorP.com/asdf?a=b&c=d#e") == "http://evilcorp.com/asdf"
assert (
@@ -428,20 +434,30 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_config, bbot_https
assert helpers.smart_encode_punycode("ドメイン.テスト:80") == "xn--eckwd4c7c.xn--zckzah:80"
assert helpers.smart_decode_punycode("xn--eckwd4c7c.xn--zckzah:80") == "ドメイン.テスト:80"
- assert helpers.recursive_decode("Hello%20world%21") == "Hello world!"
- assert helpers.recursive_decode("Hello%20%5Cu041f%5Cu0440%5Cu0438%5Cu0432%5Cu0435%5Cu0442") == "Hello Привет"
- assert helpers.recursive_decode("%5Cu0020%5Cu041f%5Cu0440%5Cu0438%5Cu0432%5Cu0435%5Cu0442%5Cu0021") == " Привет!"
- assert helpers.recursive_decode("Hello%2520world%2521") == "Hello world!"
+ assert await helpers.re.recursive_decode("Hello%20world%21") == "Hello world!"
+ assert (
+ await helpers.re.recursive_decode("Hello%20%5Cu041f%5Cu0440%5Cu0438%5Cu0432%5Cu0435%5Cu0442") == "Hello Привет"
+ )
+ assert (
+ await helpers.re.recursive_decode("%5Cu0020%5Cu041f%5Cu0440%5Cu0438%5Cu0432%5Cu0435%5Cu0442%5Cu0021")
+ == " Привет!"
+ )
+ assert await helpers.re.recursive_decode("Hello%2520world%2521") == "Hello world!"
assert (
- helpers.recursive_decode("Hello%255Cu0020%255Cu041f%255Cu0440%255Cu0438%255Cu0432%255Cu0435%255Cu0442")
+ await helpers.re.recursive_decode(
+ "Hello%255Cu0020%255Cu041f%255Cu0440%255Cu0438%255Cu0432%255Cu0435%255Cu0442"
+ )
== "Hello Привет"
)
assert (
- helpers.recursive_decode("%255Cu0020%255Cu041f%255Cu0440%255Cu0438%255Cu0432%255Cu0435%255Cu0442%255Cu0021")
+ await helpers.re.recursive_decode(
+ "%255Cu0020%255Cu041f%255Cu0440%255Cu0438%255Cu0432%255Cu0435%255Cu0442%255Cu0021"
+ )
== " Привет!"
)
assert (
- helpers.recursive_decode(r"Hello\\nWorld\\\tGreetings\\\\nMore\nText") == "Hello\nWorld\tGreetings\nMore\nText"
+ await helpers.re.recursive_decode(r"Hello\\nWorld\\\tGreetings\\\\nMore\nText")
+ == "Hello\nWorld\tGreetings\nMore\nText"
)
### CACHE ###
@@ -496,11 +512,11 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_config, bbot_https
with pytest.raises(NTLMError):
helpers.ntlm.ntlmdecode("asdf")
- test_filesize = Path("/tmp/test_filesize")
+ test_filesize = bbot_test_dir / "test_filesize"
test_filesize.touch()
assert test_filesize.is_file()
assert helpers.filesize(test_filesize) == 0
- assert helpers.filesize("/tmp/glkasjdlgksadlkfsdf") == 0
+ assert helpers.filesize(bbot_test_dir / "glkasjdlgksadlkfsdf") == 0
# memory stuff
int(helpers.memory_status().available)
@@ -509,6 +525,23 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_config, bbot_https
assert helpers.bytes_to_human(459819198709) == "428.24GB"
assert helpers.human_to_bytes("428.24GB") == 459819198709
+ # ordinals
+ assert helpers.integer_to_ordinal(1) == "1st"
+ assert helpers.integer_to_ordinal(2) == "2nd"
+ assert helpers.integer_to_ordinal(3) == "3rd"
+ assert helpers.integer_to_ordinal(4) == "4th"
+ assert helpers.integer_to_ordinal(11) == "11th"
+ assert helpers.integer_to_ordinal(12) == "12th"
+ assert helpers.integer_to_ordinal(13) == "13th"
+ assert helpers.integer_to_ordinal(21) == "21st"
+ assert helpers.integer_to_ordinal(22) == "22nd"
+ assert helpers.integer_to_ordinal(23) == "23rd"
+ assert helpers.integer_to_ordinal(101) == "101st"
+ assert helpers.integer_to_ordinal(111) == "111th"
+ assert helpers.integer_to_ordinal(112) == "112th"
+ assert helpers.integer_to_ordinal(113) == "113th"
+ assert helpers.integer_to_ordinal(0) == "0th"
+
scan1 = bbot_scanner(modules="ipneighbor")
await scan1.load_modules()
assert int(helpers.get_size(scan1.modules["ipneighbor"])) > 0
@@ -529,8 +562,39 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_config, bbot_https
< first_frequencies["e"]
)
-
-def test_word_cloud(helpers, bbot_config, bbot_scanner):
+ # error handling helpers
+ test_ran = False
+ try:
+ try:
+ raise KeyboardInterrupt("asdf")
+ except KeyboardInterrupt:
+ raise ValueError("asdf")
+ except Exception as e:
+ assert len(helpers.get_exception_chain(e)) == 2
+ assert len([_ for _ in helpers.get_exception_chain(e) if isinstance(_, KeyboardInterrupt)]) == 1
+ assert len([_ for _ in helpers.get_exception_chain(e) if isinstance(_, ValueError)]) == 1
+ assert helpers.in_exception_chain(e, (KeyboardInterrupt, asyncio.CancelledError)) == True
+ assert helpers.in_exception_chain(e, (TypeError, OSError)) == False
+ test_ran = True
+ assert test_ran
+ test_ran = False
+ try:
+ try:
+ raise AttributeError("asdf")
+ except AttributeError:
+ raise ValueError("asdf")
+ except Exception as e:
+ assert len(helpers.get_exception_chain(e)) == 2
+ assert len([_ for _ in helpers.get_exception_chain(e) if isinstance(_, AttributeError)]) == 1
+ assert len([_ for _ in helpers.get_exception_chain(e) if isinstance(_, ValueError)]) == 1
+ assert helpers.in_exception_chain(e, (KeyboardInterrupt, asyncio.CancelledError)) == False
+ assert helpers.in_exception_chain(e, (KeyboardInterrupt, AttributeError)) == True
+ assert helpers.in_exception_chain(e, (AttributeError,)) == True
+ test_ran = True
+ assert test_ran
+
+
+def test_word_cloud(helpers, bbot_scanner):
number_mutations = helpers.word_cloud.get_number_mutations("base2_p013", n=5, padding=2)
assert "base0_p013" in number_mutations
assert "base7_p013" in number_mutations
@@ -546,7 +610,7 @@ def test_word_cloud(helpers, bbot_config, bbot_scanner):
assert ("dev", "_base") in permutations
# saving and loading
- scan1 = bbot_scanner("127.0.0.1", config=bbot_config)
+ scan1 = bbot_scanner("127.0.0.1")
word_cloud = scan1.helpers.word_cloud
word_cloud.add_word("lantern")
word_cloud.add_word("black")
@@ -741,3 +805,99 @@ def test_liststring_invalidfnchars(helpers):
with pytest.raises(ValueError) as e:
helpers.chain_lists("hello,world,bbot|test", validate=True)
assert str(e.value) == "Invalid character in string: bbot|test"
+
+
+# test parameter validation
+@pytest.mark.asyncio
+async def test_parameter_validation(helpers):
+
+ getparam_valid_params = {
+ "name",
+ "age",
+ "valid_name",
+ "valid-name",
+ "session_token",
+ "user.id",
+ "user-name",
+ "client.id",
+ "auth-token",
+ "access_token",
+ "abcd",
+ "jqueryget",
+ "
+
+
+
+ {yaml_file.name}
+
+```yaml
+{loaded_preset._yaml_str}
+```
+
+Simple GET Form
+Use the form below to submit a GET request:
+ +Simple POST Form
+Use the form below to submit a POST request:
+ +Links
+ href +