diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml index 6127df218f..5cb5a75167 100644 --- a/.github/workflows/benchmark.yml +++ b/.github/workflows/benchmark.yml @@ -61,8 +61,13 @@ jobs: current_benchmark_results.json retention-days: 30 - # Comment on PR with benchmark results + # Comment on PR with benchmark results. + # A pull_request run from a fork gets a read-only GITHUB_TOKEN no matter what + # the permissions block above says, so commenting is impossible there and the + # attempt fails the whole job with a 403. The report is still in the uploaded + # artifact either way. - name: Comment benchmark results on PR + if: github.event.pull_request.head.repo.full_name == github.repository uses: actions/github-script@v9 with: script: | diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 12225e1b9e..d81333f76b 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -38,7 +38,7 @@ jobs: env: BBOT_IO_API_KEY: ${{ secrets.BBOT_IO_API_KEY }} run: | - uv run pytest -vv --reruns 2 -o timeout_func_only=true --timeout 1200 --disable-warnings --log-cli-level=INFO --cov-config=bbot/test/coverage.cfg --cov-report xml:cov.xml --cov=bbot . + uv run pytest -vv -n $(python bbot/test/worker_count.py) --dist loadgroup --reruns 2 -o timeout_func_only=true --timeout 1200 --disable-warnings --log-cli-level=INFO --cov-config=bbot/test/coverage.cfg --cov-report xml:cov.xml --cov=bbot . - name: Upload Debug Logs if: always() uses: actions/upload-artifact@v7 diff --git a/bbot/test/bbot_fixtures.py b/bbot/test/bbot_fixtures.py index df8a7cb547..f296a65c4c 100644 --- a/bbot/test/bbot_fixtures.py +++ b/bbot/test/bbot_fixtures.py @@ -15,12 +15,15 @@ from bbot.core.config.merge import deep_merge from bbot.scanner import Preset from bbot.core.helpers.misc import mkdir, rand_string +from bbot.test.worker import BBOT_TEST_DIR log = logging.getLogger("bbot.test.fixtures") -bbot_test_dir = Path("/tmp/.bbot_test") +# Per-xdist-worker so parallel workers don't share caches, scan output, or the +# sessionfinish cleanup. Serial runs still get /tmp/.bbot_test. +bbot_test_dir = BBOT_TEST_DIR mkdir(bbot_test_dir) diff --git a/bbot/test/conftest.py b/bbot/test/conftest.py index d82a982db4..d7caf4aac5 100644 --- a/bbot/test/conftest.py +++ b/bbot/test/conftest.py @@ -10,6 +10,13 @@ from contextlib import suppress from pytest_httpserver import HTTPServer +from bbot.test.worker import ( + BBOT_TEST_DIR, + HTTPSERVER_ALLINTERFACES_PORT, + HTTPSERVER_PORT, + HTTPSERVER_SSL_PORT, +) + 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 @@ -26,12 +33,28 @@ with open(Path(__file__).parent / "test.conf") as _f: test_config = yaml.safe_load(_f) or {} +# Give each xdist worker its own BBOT home. test.conf carries the serial default; +# under -n the workers would otherwise share caches, scan output and temp files, +# and the sessionfinish cleanup below would delete a directory still in use. +test_config["home"] = str(BBOT_TEST_DIR) + os.environ["BBOT_DEBUG"] = "True" CORE.logger.log_level = logging.DEBUG # silence all stderr output: stderr_handler = CORE.logger.log_handlers["stderr"] stderr_handler.setLevel(logging.CRITICAL) +# worker.py clears _BBOT_LOGGING_SETUP for xdist workers, so every process that +# reaches here (serial or worker) owns a real QueueListener. If it is missing, +# logging never got set up and debug.log would silently stay empty, so fail +# loudly instead of continuing with logging quietly broken. +if CORE.logger.listener is None: + raise RuntimeError( + "BBOT logging was not initialized in this process " + f"(PYTEST_XDIST_WORKER={os.environ.get('PYTEST_XDIST_WORKER', '')!r}). " + "debug.log would be empty and log-reading tests would fail with " + "confusing assertion errors." + ) handlers = list(CORE.logger.listener.handlers) handlers.remove(stderr_handler) CORE.logger.listener.handlers = tuple(handlers) @@ -96,7 +119,7 @@ def stop_server(server): @pytest.fixture def bbot_httpserver(): - server = HTTPServer(host="127.0.0.1", port=8888, threaded=True) + server = HTTPServer(host="127.0.0.1", port=HTTPSERVER_PORT, threaded=True) server.start() yield server @@ -115,7 +138,7 @@ def bbot_httpserver_ssl(): keyfile = str(current_dir / "testsslkey.pem") certfile = str(current_dir / "testsslcert.pem") context.load_cert_chain(certfile, keyfile) - server = HTTPServer(host="127.0.0.1", port=9999, ssl_context=context, threaded=True) + server = HTTPServer(host="127.0.0.1", port=HTTPSERVER_SSL_PORT, ssl_context=context, threaded=True) server.start() yield server @@ -222,7 +245,7 @@ async def patched_request_batch_stream(self, urls, threads=10, **kwargs): @pytest.fixture def bbot_httpserver_allinterfaces(): - server = HTTPServer(host="0.0.0.0", port=5556, threaded=True) + server = HTTPServer(host="0.0.0.0", port=HTTPSERVER_ALLINTERFACES_PORT, threaded=True) server.start() yield server @@ -334,6 +357,20 @@ def proxy_server(): server_thread.join() +def pytest_collection_modifyitems(config, items): + """Pin docker-backed tests to one xdist worker. + + They start real containers with fixed names and fixed host port bindings + (kafka, elastic, mongo, mysql, nats, postgres, rabbitmq), so two workers + running them at once fight over both. They already mark themselves with + ``skip_distro_tests``, so reuse that as the signal. + """ + for item in items: + cls = getattr(item, "cls", None) + if cls is not None and getattr(cls, "skip_distro_tests", False): + item.add_marker(pytest.mark.xdist_group("docker")) + + def pytest_terminal_summary(terminalreporter, exitstatus, config): # pragma: no cover RED = "\033[1;31m" GREEN = "\033[1;32m" @@ -460,8 +497,10 @@ def pytest_sessionfinish(session, exitstatus): if child.is_alive(): child.kill() - # Wipe out BBOT home dir - shutil.rmtree("/tmp/.bbot_test", ignore_errors=True) + # Wipe out BBOT home dir. Scoped to this worker: under xdist the first + # worker to finish would otherwise delete the directory out from under + # every worker still running. + shutil.rmtree(BBOT_TEST_DIR, ignore_errors=True) # Ensure stdout/stderr are blocking before pytest writes summaries try: diff --git a/bbot/test/test_step_1/test_bbot_fastapi.py b/bbot/test/test_step_1/test_bbot_fastapi.py index 2c122f3191..e00f39c328 100644 --- a/bbot/test/test_step_1/test_bbot_fastapi.py +++ b/bbot/test/test_step_1/test_bbot_fastapi.py @@ -7,6 +7,7 @@ from urllib.request import urlopen, Request from urllib.error import URLError from urllib.parse import urlencode +from bbot.test.worker import FASTAPI_PORT, FASTAPI_URL, HTTPSERVER_URL cwd = Path(__file__).parent.parent.parent @@ -14,7 +15,7 @@ def run_bbot_multiprocess(queue): from bbot.scanner import Scanner - scan = Scanner("http://127.0.0.1:8888", "blacklanternsecurity.com", modules=["http"]) + scan = Scanner(HTTPSERVER_URL, "blacklanternsecurity.com", modules=["http"]) events = [e.json() for e in scan.start()] queue.put(events) @@ -42,7 +43,7 @@ def test_bbot_fastapi(bbot_httpserver): start_time = time.time() while True: try: - response = urlopen("http://127.0.0.1:8978/ping") + response = urlopen(f"{FASTAPI_URL}/ping") response.read() break except (URLError, ConnectionError): @@ -52,8 +53,8 @@ def test_bbot_fastapi(bbot_httpserver): continue # run a scan - params = urlencode({"targets": ["http://127.0.0.1:8888", "blacklanternsecurity.com"]}, doseq=True) - req = Request(f"http://127.0.0.1:8978/start?{params}") + params = urlencode({"targets": [HTTPSERVER_URL, "blacklanternsecurity.com"]}, doseq=True) + req = Request(f"{FASTAPI_URL}/start?{params}") response = urlopen(req, timeout=100) events = json.loads(response.read()) assert len(events) >= 3 @@ -75,6 +76,8 @@ def start_fastapi_server(): del env["BBOT_TESTING"] python_executable = str(sys.executable) process = Popen( - [python_executable, "-m", "uvicorn", "bbot.test.fastapi_test:app", "--port", "8978"], cwd=cwd, env=env + [python_executable, "-m", "uvicorn", "bbot.test.fastapi_test:app", "--port", str(FASTAPI_PORT)], + cwd=cwd, + env=env, ) return process diff --git a/bbot/test/test_step_1/test_cli.py b/bbot/test/test_step_1/test_cli.py index c8321283cd..2db0037d12 100644 --- a/bbot/test/test_step_1/test_cli.py +++ b/bbot/test/test_step_1/test_cli.py @@ -871,6 +871,12 @@ def test_cli_presets(monkeypatch, capsys, caplog, clean_default_config): """ ) + # These are cleaned up at the end of the test, but an earlier failure (or an + # interrupted run) leaves them behind, and then these "not exists" assertions + # fail on every later run against the same BBOT home. Clear them up front so + # the test establishes its precondition instead of assuming it. + shutil.rmtree(output_dir, ignore_errors=True) + assert not output_dir.exists() assert not scan_dir.exists() assert not output_file.exists() diff --git a/bbot/test/test_step_1/test_command.py b/bbot/test/test_step_1/test_command.py index 54bbdaba25..f8789f8588 100644 --- a/bbot/test/test_step_1/test_command.py +++ b/bbot/test/test_step_1/test_command.py @@ -1,5 +1,7 @@ import time from ..bbot_fixtures import * + +from bbot.test.worker import BBOT_TEST_DIR from subprocess import CalledProcessError @@ -38,7 +40,7 @@ async def test_command(bbot_scanner): assert result.splitlines() == [b"some", b"random", b"stdin"] # test overflow - run - tmpfile_path = Path("/tmp/test_bigfile") + tmpfile_path = BBOT_TEST_DIR / "test_bigfile" with open(tmpfile_path, "w") as f: # write 2MB f.write("A" * 1024 * 1024 * 2) @@ -46,7 +48,7 @@ async def test_command(bbot_scanner): assert len(result) == 1024 * 1024 * 2 tmpfile_path.unlink(missing_ok=True) # test overflow - run_live - tmpfile_path = Path("/tmp/test_bigfile") + tmpfile_path = BBOT_TEST_DIR / "test_bigfile" with open(tmpfile_path, "w") as f: # write 2MB f.write("A" * 10 + "\n") @@ -119,30 +121,30 @@ async def test_command(bbot_scanner): # test sudo + existence of environment variables await scan1._prep() path_parts = os.environ.get("PATH", "").split(":") - assert "/tmp/.bbot_test/tools" in path_parts + assert f"{BBOT_TEST_DIR}/tools" in path_parts run_lines = (await scan1.helpers.run(["env"])).stdout.splitlines() 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 + assert f"{BBOT_TEST_DIR}/tools" in path_parts run_lines_sudo = (await scan1.helpers.run(["env"], sudo=True)).stdout.splitlines() 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 + assert f"{BBOT_TEST_DIR}/tools" in path_parts run_live_lines = [l async for l in scan1.helpers.run_live(["env"])] 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 + assert f"{BBOT_TEST_DIR}/tools" in path_parts run_live_lines_sudo = [l async for l in scan1.helpers.run_live(["env"], sudo=True)] 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(":") - assert "/tmp/.bbot_test/tools" in path_parts + assert f"{BBOT_TEST_DIR}/tools" in path_parts await scan1._cleanup() diff --git a/bbot/test/test_step_1/test_excavate_url_regexes.py b/bbot/test/test_step_1/test_excavate_url_regexes.py index d1a30dc50d..69369fa916 100644 --- a/bbot/test/test_step_1/test_excavate_url_regexes.py +++ b/bbot/test/test_step_1/test_excavate_url_regexes.py @@ -11,6 +11,7 @@ import yara from bbot.modules.internal.excavate import excavate +from bbot.test.worker import HTTPSERVER_URL full_url_regex = excavate.URLExtractor.full_url_regex @@ -30,7 +31,7 @@ NON_IPV6_URLS = [ "http://example.com/", "https://www.example.com:8080/path", - "http://127.0.0.1:8888/", + f"{HTTPSERVER_URL}/", "https://asdffoo.test.notreal/some/path", ] diff --git a/bbot/test/test_step_1/test_manager_scope_accuracy.py b/bbot/test/test_step_1/test_manager_scope_accuracy.py index fb6b5c4c95..66d80773bf 100644 --- a/bbot/test/test_step_1/test_manager_scope_accuracy.py +++ b/bbot/test/test_step_1/test_manager_scope_accuracy.py @@ -12,12 +12,27 @@ from ..bbot_fixtures import * # noqa: F401 from pytest_httpserver import HTTPServer +from bbot.test.worker import ( + HTTPSERVER_HOSTPORT, + HTTPSERVER_PORT, + HTTPSERVER_PORT_ALT, + HTTPSERVER_SSL_HOSTPORT, + HTTPSERVER_SSL_PORT, + HTTPSERVER_URL, +) @pytest.fixture def bbot_other_httpservers(): - server_hosts = [("127.0.0.77", 8888), ("127.0.0.88", 8888), ("127.0.0.99", 8888), ("127.0.0.111", 8888), ("127.0.0.222", 8889), ("127.0.0.33", 8889)] + server_hosts = [ + ("127.0.0.77", HTTPSERVER_PORT), + ("127.0.0.88", HTTPSERVER_PORT), + ("127.0.0.99", HTTPSERVER_PORT), + ("127.0.0.111", HTTPSERVER_PORT), + ("127.0.0.222", HTTPSERVER_PORT_ALT), + ("127.0.0.33", HTTPSERVER_PORT_ALT), + ] servers = [HTTPServer(host=host, port=port, threaded=True) for host, port in server_hosts] for server in servers: @@ -47,13 +62,13 @@ async def test_manager_scope_accuracy_correct(bbot_scanner, bbot_httpserver, bbo server_77, server_88, server_99, server_111, server_222, server_33 = bbot_other_httpservers - bbot_httpserver.expect_request(uri="/").respond_with_data(response_data="") - server_77.expect_request(uri="/").respond_with_data(response_data="") - server_88.expect_request(uri="/").respond_with_data(response_data="") - server_99.expect_request(uri="/").respond_with_data(response_data="") - server_111.expect_request(uri="/").respond_with_data(response_data="") - server_222.expect_request(uri="/").respond_with_data(response_data="") - server_33.expect_request(uri="/").respond_with_data(response_data="") + bbot_httpserver.expect_request(uri="/").respond_with_data(response_data=f"") + server_77.expect_request(uri="/").respond_with_data(response_data=f"") + server_88.expect_request(uri="/").respond_with_data(response_data=f"") + server_99.expect_request(uri="/").respond_with_data(response_data=f"") + server_111.expect_request(uri="/").respond_with_data(response_data=f"") + server_222.expect_request(uri="/").respond_with_data(response_data=f"") + server_33.expect_request(uri="/").respond_with_data(response_data=f"") class DummyModule(BaseModule): _name = "dummy_module" @@ -295,7 +310,7 @@ def custom_setup(scan): "scope": {"report_distance": 1, "search_distance": 0}, "speculate": True, "excavate": True, - "modules": {"speculate": {"ports": "8888"}}, + "modules": {"speculate": {"ports": f"{HTTPSERVER_PORT}"}}, "omit_event_types": ["HTTP_RESPONSE", "URL_UNVERIFIED"], }, _dns_mock={}, @@ -305,54 +320,54 @@ def custom_setup(scan): assert 1 == len([e for e in events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 0 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0"]) assert 1 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:8888"]) - assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in events if e.type == "URL" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.1:8888"]) - assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.1:8888/"]) - assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.77:8888/"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in events if e.type == "URL" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["input"] == HTTPSERVER_HOSTPORT]) + assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == f"{HTTPSERVER_URL}/"]) + assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/"]) assert 1 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is False and e.scope_distance == 1]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.77:8888"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.77:{HTTPSERVER_PORT}"]) assert len(all_events) == 14 assert 1 == len([e for e in all_events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0" and e.internal is True and e.scope_distance == 0]) assert 2 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:8888" and e.internal is True and e.scope_distance == 0]) - assert 2 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1 and "spider-danger" in e.tags]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 0]) + assert 2 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["input"] == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1 and "spider-danger" in e.tags]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is False and e.scope_distance == 1]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.77:8888" and e.internal is True and e.scope_distance == 1]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.77:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 1]) assert len(all_events_nodups) == 12 assert 1 == len([e for e in all_events_nodups if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.0" and e.internal is True and e.scope_distance == 0]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:8888" and e.internal is True and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1 and "spider-danger" in e.tags]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["input"] == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1 and "spider-danger" in e.tags]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is False and e.scope_distance == 1]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.77:8888" and e.internal is True and e.scope_distance == 1]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.77:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 1]) for _graph_output_events in (graph_output_events, graph_output_batch_events): assert len(_graph_output_events) == 7 assert 1 == len([e for e in _graph_output_events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 0 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0"]) assert 1 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:8888"]) - assert 1 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.1:8888"]) - assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.1:8888/"]) - assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.77:8888/"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["input"] == HTTPSERVER_HOSTPORT]) + assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == f"{HTTPSERVER_URL}/"]) + assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/"]) assert 1 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is False and e.scope_distance == 1]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.77:8888"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.77:{HTTPSERVER_PORT}"]) # http/speculate IP_RANGE --> IP_ADDRESS --> OPEN_TCP_PORT --> URL, search distance = 0, in_scope_only = False events, all_events, all_events_nodups, graph_output_events, graph_output_batch_events = await do_scan( @@ -363,7 +378,7 @@ def custom_setup(scan): "scope": {"search_distance": 0, "report_distance": 1}, "excavate": True, "speculate": True, - "modules": {"http": {"in_scope_only": False}, "speculate": {"ports": "8888"}}, + "modules": {"http": {"in_scope_only": False}, "speculate": {"ports": f"{HTTPSERVER_PORT}"}}, "omit_event_types": ["HTTP_RESPONSE", "URL_UNVERIFIED"], }, ) @@ -377,70 +392,70 @@ def custom_setup(scan): assert 1 == len([e for e in events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 0 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0"]) assert 1 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:8888"]) - assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in events if e.type == "URL" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.1:8888"]) - assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.1:8888/"]) - assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.77:8888/"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in events if e.type == "URL" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["input"] == HTTPSERVER_HOSTPORT]) + assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == f"{HTTPSERVER_URL}/"]) + assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/"]) assert 1 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is False and e.scope_distance == 1]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.77:8888"]) - assert 1 == len([e for e in events if e.type == "URL" and e.url == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1]) - assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.77:8888"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.77:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in events if e.type == "URL" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1]) + assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["input"] == f"127.0.0.77:{HTTPSERVER_PORT}"]) assert 0 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.88"]) - assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.77:8888/"]) + assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/"]) assert len(all_events) == 18 assert 1 == len([e for e in all_events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0" and e.internal is True and e.scope_distance == 0]) assert 2 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:8888" and e.internal is True and e.scope_distance == 0]) - assert 2 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 0]) + assert 2 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["input"] == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is False and e.scope_distance == 1]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.77:8888" and e.internal is True and e.scope_distance == 1]) - assert 1 == len([e for e in all_events if e.type == "URL" and e.url == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1]) - assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["url"] == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.77:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 1]) + assert 1 == len([e for e in all_events if e.type == "URL" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1]) + assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["url"] == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.88" and e.internal is True and e.scope_distance == 2]) - assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.88:8888/" and e.internal is True and e.scope_distance == 2]) + assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.88:{HTTPSERVER_PORT}/" and e.internal is True and e.scope_distance == 2]) assert len(all_events_nodups) == 16 assert 1 == len([e for e in all_events_nodups if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.0" and e.internal is True and e.scope_distance == 0]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:8888" and e.internal is True and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1 and "spider-danger" in e.tags]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["input"] == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1 and "spider-danger" in e.tags]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is False and e.scope_distance == 1]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.77:8888" and e.internal is True and e.scope_distance == 1]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1]) - assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["url"] == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.77:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 1]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1]) + assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["url"] == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.88" and e.internal is True and e.scope_distance == 2]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.88:8888/" and e.internal is True and e.scope_distance == 2]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.88:{HTTPSERVER_PORT}/" and e.internal is True and e.scope_distance == 2]) for _graph_output_events in (graph_output_events, graph_output_batch_events): assert len(_graph_output_events) == 8 assert 1 == len([e for e in _graph_output_events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 0 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0"]) assert 1 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:8888"]) - assert 1 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.1:8888"]) - assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.1:8888/"]) - assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.77:8888/" and "spider-danger" in e.tags]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["input"] == HTTPSERVER_HOSTPORT]) + assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == f"{HTTPSERVER_URL}/"]) + assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and "spider-danger" in e.tags]) assert 1 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is False and e.scope_distance == 1]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.77:8888"]) - assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1]) - assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["url"] == "http://127.0.0.77:8888/"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.77:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1]) + assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["url"] == f"http://127.0.0.77:{HTTPSERVER_PORT}/"]) assert 0 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.88"]) - assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.88:8888/"]) + assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.88:{HTTPSERVER_PORT}/"]) # http/speculate IP_RANGE --> IP_ADDRESS --> OPEN_TCP_PORT --> URL, search distance = 1 events, all_events, all_events_nodups, graph_output_events, graph_output_batch_events = await do_scan( @@ -451,7 +466,7 @@ def custom_setup(scan): "scope": {"report_distance": 1, "search_distance": 1}, "excavate": True, "speculate": True, - "modules": {"http": {"in_scope_only": False}, "speculate": {"ports": "8888"}}, + "modules": {"http": {"in_scope_only": False}, "speculate": {"ports": f"{HTTPSERVER_PORT}"}}, "omit_event_types": ["HTTP_RESPONSE", "URL_UNVERIFIED"], }, ) @@ -460,78 +475,78 @@ def custom_setup(scan): assert 1 == len([e for e in events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 0 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0"]) assert 1 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:8888"]) - assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in events if e.type == "URL" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.1:8888"]) - assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.1:8888/"]) - assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.77:8888/"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in events if e.type == "URL" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["input"] == HTTPSERVER_HOSTPORT]) + assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == f"{HTTPSERVER_URL}/"]) + assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/"]) assert 1 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is False and e.scope_distance == 1]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.77:8888"]) - assert 1 == len([e for e in events if e.type == "URL" and e.url == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1]) - assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["url"] == "http://127.0.0.77:8888/"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.77:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in events if e.type == "URL" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1]) + assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["url"] == f"http://127.0.0.77:{HTTPSERVER_PORT}/"]) assert 0 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.88"]) - assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.77:8888/"]) + assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/"]) assert len(all_events) == 22 assert 1 == len([e for e in all_events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0" and e.internal is True and e.scope_distance == 0]) assert 2 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:8888" and e.internal is True and e.scope_distance == 0]) - assert 2 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1 and "spider-danger" in e.tags]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 0]) + assert 2 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["input"] == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1 and "spider-danger" in e.tags]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is False and e.scope_distance == 1]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.77:8888" and e.internal is True and e.scope_distance == 1]) - assert 1 == len([e for e in all_events if e.type == "URL" and e.url == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1]) - assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["url"] == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1]) - assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.88:8888/" and e.internal is True and e.scope_distance == 2]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.77:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 1]) + assert 1 == len([e for e in all_events if e.type == "URL" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1]) + assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["url"] == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1]) + assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.88:{HTTPSERVER_PORT}/" and e.internal is True and e.scope_distance == 2]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.88" and e.internal is True and e.scope_distance == 2]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.88:8888" and e.internal is True and e.scope_distance == 2]) - assert 1 == len([e for e in all_events if e.type == "URL" and e.url == "http://127.0.0.88:8888/" and e.internal is True and e.scope_distance == 2]) - assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["url"] == "http://127.0.0.88:8888/" and e.internal is True and e.scope_distance == 2]) - assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.99:8888/" and e.internal is True and e.scope_distance == 3]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.88:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 2]) + assert 1 == len([e for e in all_events if e.type == "URL" and e.url == f"http://127.0.0.88:{HTTPSERVER_PORT}/" and e.internal is True and e.scope_distance == 2]) + assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["url"] == f"http://127.0.0.88:{HTTPSERVER_PORT}/" and e.internal is True and e.scope_distance == 2]) + assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.99:{HTTPSERVER_PORT}/" and e.internal is True and e.scope_distance == 3]) assert len(all_events_nodups) == 20 assert 1 == len([e for e in all_events_nodups if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.0" and e.internal is True and e.scope_distance == 0]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:8888" and e.internal is True and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1 and "spider-danger" in e.tags]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["input"] == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1 and "spider-danger" in e.tags]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is False and e.scope_distance == 1]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.77:8888" and e.internal is True and e.scope_distance == 1]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1]) - assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["url"] == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.88:8888/" and e.internal is True and e.scope_distance == 2]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.77:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 1]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1]) + assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["url"] == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.88:{HTTPSERVER_PORT}/" and e.internal is True and e.scope_distance == 2]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.88" and e.internal is True and e.scope_distance == 2]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.88:8888" and e.internal is True and e.scope_distance == 2]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == "http://127.0.0.88:8888/" and e.internal is True and e.scope_distance == 2]) - assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["url"] == "http://127.0.0.88:8888/" and e.internal is True and e.scope_distance == 2]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.99:8888/" and e.internal is True and e.scope_distance == 3]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.88:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 2]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == f"http://127.0.0.88:{HTTPSERVER_PORT}/" and e.internal is True and e.scope_distance == 2]) + assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["url"] == f"http://127.0.0.88:{HTTPSERVER_PORT}/" and e.internal is True and e.scope_distance == 2]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.99:{HTTPSERVER_PORT}/" and e.internal is True and e.scope_distance == 3]) for _graph_output_events in (graph_output_events, graph_output_batch_events): assert len(_graph_output_events) == 8 assert 1 == len([e for e in _graph_output_events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 0 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0"]) assert 1 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:8888"]) - assert 1 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == "http://127.0.0.1:8888/" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.1:8888"]) - assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.1:8888/"]) - assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.77:8888/"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_HOSTPORT and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == f"{HTTPSERVER_URL}/" and e.internal is False and e.scope_distance == 0]) + assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["input"] == HTTPSERVER_HOSTPORT]) + assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == f"{HTTPSERVER_URL}/"]) + assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/"]) assert 1 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.77" and e.internal is False and e.scope_distance == 1]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.77:8888"]) - assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == "http://127.0.0.77:8888/" and e.internal is False and e.scope_distance == 1]) - assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["url"] == "http://127.0.0.77:8888/"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.77:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == f"http://127.0.0.77:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 1]) + assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["url"] == f"http://127.0.0.77:{HTTPSERVER_PORT}/"]) assert 0 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.88"]) - assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.88:8888/"]) + assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.88:{HTTPSERVER_PORT}/"]) # 2 events from a single HTTP_RESPONSE events, all_events, all_events_nodups, graph_output_events, graph_output_batch_events = await do_scan( @@ -546,7 +561,7 @@ def custom_setup(scan): "scope": {"search_distance": 0, "report_distance": 0}, "excavate": True, "speculate": True, - "modules": {"speculate": {"ports": "8888"}}, + "modules": {"speculate": {"ports": f"{HTTPSERVER_PORT}"}}, "omit_event_types": ["HTTP_RESPONSE", "URL_UNVERIFIED"], }, ) @@ -555,120 +570,120 @@ def custom_setup(scan): assert 1 == len([e for e in events if e.type == "IP_RANGE" and e.data == "127.0.0.110/31" and e.internal is False and e.scope_distance == 0]) assert 0 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.110"]) assert 1 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.111" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.110:8888"]) - assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.111:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in events if e.type == "URL" and e.url == "http://127.0.0.111:8888/" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.111:8888"]) - assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.111:8888/"]) - assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.222:8889/"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.110:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.111:{HTTPSERVER_PORT}" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in events if e.type == "URL" and e.url == f"http://127.0.0.111:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 0]) + assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["input"] == f"127.0.0.111:{HTTPSERVER_PORT}"]) + assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.111:{HTTPSERVER_PORT}/"]) + assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.222:{HTTPSERVER_PORT_ALT}/"]) assert 1 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.222" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.33:8889/"]) + assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.33:{HTTPSERVER_PORT_ALT}/"]) assert 1 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.33" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.222:8888"]) - assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.222:8889"]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.33:8888"]) - assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.33:8889"]) - assert 1 == len([e for e in events if e.type == "URL" and e.url == "http://127.0.0.222:8889/" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.222:8889"]) - assert 1 == len([e for e in events if e.type == "URL" and e.url == "http://127.0.0.33:8889/" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.33:8889"]) - assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.44:8888/"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.222:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.222:{HTTPSERVER_PORT_ALT}"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.33:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.33:{HTTPSERVER_PORT_ALT}"]) + assert 1 == len([e for e in events if e.type == "URL" and e.url == f"http://127.0.0.222:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) + assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["input"] == f"127.0.0.222:{HTTPSERVER_PORT_ALT}"]) + assert 1 == len([e for e in events if e.type == "URL" and e.url == f"http://127.0.0.33:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) + assert 0 == len([e for e in events if e.type == "HTTP_RESPONSE" and e.data["input"] == f"127.0.0.33:{HTTPSERVER_PORT_ALT}"]) + assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.44:{HTTPSERVER_PORT}/"]) assert 0 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.44"]) - assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.55:8888/"]) + assert 0 == len([e for e in events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.55:{HTTPSERVER_PORT}/"]) assert 0 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.55"]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.44:8888"]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.55:8888"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.44:{HTTPSERVER_PORT}"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.55:{HTTPSERVER_PORT}"]) assert len(all_events) == 31 assert 1 == len([e for e in all_events if e.type == "IP_RANGE" and e.data == "127.0.0.110/31" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.110" and e.internal is True and e.scope_distance == 0]) assert 2 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.111" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.110:8888" and e.internal is True and e.scope_distance == 0]) - assert 2 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.111:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL" and e.url == "http://127.0.0.111:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.111:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.111:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.222:8889/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.110:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 0]) + assert 2 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.111:{HTTPSERVER_PORT}" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL" and e.url == f"http://127.0.0.111:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["input"] == f"127.0.0.111:{HTTPSERVER_PORT}" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.111:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.222:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.222" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.33:8889/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.33:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.33" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.222:8888" and e.internal is True and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.222:8889" and e.internal is True and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.222:8889" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.33:8888" and e.internal is True and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.33:8889" and e.internal is True and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.33:8889" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL" and e.url == "http://127.0.0.222:8889/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["url"] == "http://127.0.0.222:8889/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL" and e.url == "http://127.0.0.33:8889/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["url"] == "http://127.0.0.33:8889/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.44:8888/" and e.internal is True and e.scope_distance == 1]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.222:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.222:{HTTPSERVER_PORT_ALT}" and e.internal is True and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.222:{HTTPSERVER_PORT_ALT}" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.33:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.33:{HTTPSERVER_PORT_ALT}" and e.internal is True and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.33:{HTTPSERVER_PORT_ALT}" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL" and e.url == f"http://127.0.0.222:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["url"] == f"http://127.0.0.222:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL" and e.url == f"http://127.0.0.33:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and e.data["url"] == f"http://127.0.0.33:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.44:{HTTPSERVER_PORT}/" and e.internal is True and e.scope_distance == 1]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.44" and e.internal is True and e.scope_distance == 1]) - assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.55:8888/" and e.internal is True and e.scope_distance == 1]) + assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.55:{HTTPSERVER_PORT}/" and e.internal is True and e.scope_distance == 1]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.55" and e.internal is True and e.scope_distance == 1]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.44:8888" and e.internal is True and e.scope_distance == 1]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.55:8888" and e.internal is True and e.scope_distance == 1]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.44:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 1]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.55:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 1]) assert len(all_events_nodups) == 27 assert 1 == len([e for e in all_events_nodups if e.type == "IP_RANGE" and e.data == "127.0.0.110/31" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.110" and e.internal is True and e.scope_distance == 0]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.111" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.110:8888" and e.internal is True and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.111:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == "http://127.0.0.111:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.111:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.111:8888/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.222:8889/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.110:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.111:{HTTPSERVER_PORT}" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == f"http://127.0.0.111:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["input"] == f"127.0.0.111:{HTTPSERVER_PORT}" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.111:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.222:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.222" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.33:8889/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.33:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.33" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.222:8888" and e.internal is True and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.222:8889" and e.internal is True and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.33:8888" and e.internal is True and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.33:8889" and e.internal is True and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == "http://127.0.0.222:8889/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["url"] == "http://127.0.0.222:8889/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == "http://127.0.0.33:8889/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["url"] == "http://127.0.0.33:8889/" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.44:8888/" and e.internal is True and e.scope_distance == 1]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.222:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.222:{HTTPSERVER_PORT_ALT}" and e.internal is True and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.33:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.33:{HTTPSERVER_PORT_ALT}" and e.internal is True and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == f"http://127.0.0.222:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["url"] == f"http://127.0.0.222:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL" and e.url == f"http://127.0.0.33:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and e.data["url"] == f"http://127.0.0.33:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.44:{HTTPSERVER_PORT}/" and e.internal is True and e.scope_distance == 1]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.44" and e.internal is True and e.scope_distance == 1]) - assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.55:8888/" and e.internal is True and e.scope_distance == 1]) + assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.55:{HTTPSERVER_PORT}/" and e.internal is True and e.scope_distance == 1]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.55" and e.internal is True and e.scope_distance == 1]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.44:8888" and e.internal is True and e.scope_distance == 1]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.55:8888" and e.internal is True and e.scope_distance == 1]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.44:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 1]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.55:{HTTPSERVER_PORT}" and e.internal is True and e.scope_distance == 1]) for _graph_output_events in (graph_output_events, graph_output_batch_events): assert len(_graph_output_events) == 12 assert 1 == len([e for e in _graph_output_events if e.type == "IP_RANGE" and e.data == "127.0.0.110/31" and e.internal is False and e.scope_distance == 0]) assert 0 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.110"]) assert 1 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.111" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.110:8888"]) - assert 1 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.111:8888" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == "http://127.0.0.111:8888/" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.111:8888"]) - assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.111:8888/"]) - assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.222:8889/"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.110:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.111:{HTTPSERVER_PORT}" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == f"http://127.0.0.111:{HTTPSERVER_PORT}/" and e.internal is False and e.scope_distance == 0]) + assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["input"] == f"127.0.0.111:{HTTPSERVER_PORT}"]) + assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.111:{HTTPSERVER_PORT}/"]) + assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.222:{HTTPSERVER_PORT_ALT}/"]) assert 1 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.222"]) - assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.33:8889/"]) + assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.33:{HTTPSERVER_PORT_ALT}/"]) assert 1 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.33"]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.222:8888"]) - assert 1 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.222:8889"]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.33:8888"]) - assert 1 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.33:8889"]) - assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == "http://127.0.0.222:8889/" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.222:8889"]) - assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == "http://127.0.0.33:8889/" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["input"] == "127.0.0.33:8889"]) - assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.44:8888/"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.222:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.222:{HTTPSERVER_PORT_ALT}"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.33:{HTTPSERVER_PORT}"]) + assert 1 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.33:{HTTPSERVER_PORT_ALT}"]) + assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == f"http://127.0.0.222:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) + assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["input"] == f"127.0.0.222:{HTTPSERVER_PORT_ALT}"]) + assert 1 == len([e for e in _graph_output_events if e.type == "URL" and e.url == f"http://127.0.0.33:{HTTPSERVER_PORT_ALT}/" and e.internal is False and e.scope_distance == 0]) + assert 0 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and e.data["input"] == f"127.0.0.33:{HTTPSERVER_PORT_ALT}"]) + assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.44:{HTTPSERVER_PORT}/"]) assert 0 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.44"]) - assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.55:8888/"]) + assert 0 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and e.url == f"http://127.0.0.55:{HTTPSERVER_PORT}/"]) assert 0 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.55"]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.44:8888"]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.55:8888"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.44:{HTTPSERVER_PORT}"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.55:{HTTPSERVER_PORT}"]) # sslcert with in-scope chain events, all_events, all_events_nodups, graph_output_events, graph_output_batch_events = await do_scan( - "127.0.0.0/31", modules=["sslcert"], _config={"scope": {"report_distance": 0}, "speculate": True, "modules": {"speculate": {"ports": "9999"}}}, _dns_mock={"www.bbottest.notreal": {"A": ["127.0.1.0"]}, "test.notreal": {"A": ["127.0.0.1"]}} + "127.0.0.0/31", modules=["sslcert"], _config={"scope": {"report_distance": 0}, "speculate": True, "modules": {"speculate": {"ports": f"{HTTPSERVER_SSL_PORT}"}}}, _dns_mock={"www.bbottest.notreal": {"A": ["127.0.1.0"]}, "test.notreal": {"A": ["127.0.0.1"]}} ) # sslcert now watches HTTP_RESPONSE, which auto-enables the http module. @@ -677,68 +692,68 @@ def custom_setup(scan): assert 1 == len([e for e in events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 0 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0"]) assert 1 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1"]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:9999"]) - assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:9999"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_SSL_PORT}"]) + assert 1 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_SSL_HOSTPORT]) assert 1 == len([e for e in events if e.type == "HTTP_RESPONSE" and str(e.module) == "http"]) assert 1 == len([e for e in events if e.type == "URL" and str(e.module) == "http"]) assert 1 == len([e for e in events if e.type == "URL_UNVERIFIED" and str(e.module) == "speculate"]) assert 1 == len([e for e in events if e.type == "DNS_NAME" and e.data == "test.notreal" and e.internal is False and e.scope_distance == 0 and str(e.module) == "sslcert"]) assert 1 == len([e for e in events if e.type == "DNS_NAME" and e.data == "www.bbottest.notreal" and e.internal is False and e.scope_distance == 1 and str(e.module) == "sslcert" and "affiliate" in e.tags]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "test.notreal:9999"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"test.notreal:{HTTPSERVER_SSL_PORT}"]) assert 0 == len([e for e in events if e.type == "DNS_NAME_UNRESOLVED" and e.data == "notreal"]) assert len(all_events) == 16 assert 1 == len([e for e in all_events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0" and e.internal is True and e.scope_distance == 0]) assert 2 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:9999" and e.internal is True and e.scope_distance == 0]) - assert 2 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:9999" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_SSL_PORT}" and e.internal is True and e.scope_distance == 0]) + assert 2 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_SSL_HOSTPORT and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events if e.type == "HTTP_RESPONSE" and str(e.module) == "http"]) assert 1 == len([e for e in all_events if e.type == "URL" and str(e.module) == "http"]) assert 1 == len([e for e in all_events if e.type == "URL_UNVERIFIED" and str(e.module) == "speculate"]) assert 1 == len([e for e in all_events if e.type == "DNS_NAME" and e.data == "test.notreal" and e.internal is False and e.scope_distance == 0 and str(e.module) == "sslcert"]) assert 1 == len([e for e in all_events if e.type == "DNS_NAME" and e.data == "www.bbottest.notreal" and e.internal is False and e.scope_distance == 1 and str(e.module) == "sslcert"]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "www.bbottest.notreal:9999" and e.internal is True and e.scope_distance == 1 and str(e.module) == "speculate"]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"www.bbottest.notreal:{HTTPSERVER_SSL_PORT}" and e.internal is True and e.scope_distance == 1 and str(e.module) == "speculate"]) assert 1 == len([e for e in all_events if e.type == "DNS_NAME_UNRESOLVED" and e.data == "bbottest.notreal" and e.internal is True and e.scope_distance == 2 and str(e.module) == "speculate"]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "test.notreal:9999" and e.internal is True and e.scope_distance == 0 and str(e.module) == "speculate"]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"test.notreal:{HTTPSERVER_SSL_PORT}" and e.internal is True and e.scope_distance == 0 and str(e.module) == "speculate"]) assert len(all_events_nodups) == 14 assert 1 == len([e for e in all_events_nodups if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.0" and e.internal is True and e.scope_distance == 0]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:9999" and e.internal is True and e.scope_distance == 0]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:9999" and e.internal is False and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_SSL_PORT}" and e.internal is True and e.scope_distance == 0]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_SSL_HOSTPORT and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in all_events_nodups if e.type == "HTTP_RESPONSE" and str(e.module) == "http"]) assert 1 == len([e for e in all_events_nodups if e.type == "URL" and str(e.module) == "http"]) assert 1 == len([e for e in all_events_nodups if e.type == "URL_UNVERIFIED" and str(e.module) == "speculate"]) assert 1 == len([e for e in all_events_nodups if e.type == "DNS_NAME" and e.data == "test.notreal" and e.internal is False and e.scope_distance == 0 and str(e.module) == "sslcert"]) assert 1 == len([e for e in all_events_nodups if e.type == "DNS_NAME" and e.data == "www.bbottest.notreal" and e.internal is False and e.scope_distance == 1 and str(e.module) == "sslcert"]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "www.bbottest.notreal:9999" and e.internal is True and e.scope_distance == 1 and str(e.module) == "speculate"]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"www.bbottest.notreal:{HTTPSERVER_SSL_PORT}" and e.internal is True and e.scope_distance == 1 and str(e.module) == "speculate"]) assert 1 == len([e for e in all_events_nodups if e.type == "DNS_NAME_UNRESOLVED" and e.data == "bbottest.notreal" and e.internal is True and e.scope_distance == 2 and str(e.module) == "speculate"]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "test.notreal:9999" and e.internal is True and e.scope_distance == 0 and str(e.module) == "speculate"]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"test.notreal:{HTTPSERVER_SSL_PORT}" and e.internal is True and e.scope_distance == 0 and str(e.module) == "speculate"]) for _graph_output_events in (graph_output_events, graph_output_batch_events): assert len(_graph_output_events) == 10 assert 1 == len([e for e in _graph_output_events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 0]) assert 0 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0"]) assert 1 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is False and e.scope_distance == 0]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:9999"]) - assert 1 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:9999" and e.internal is False and e.scope_distance == 0]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_SSL_PORT}"]) + assert 1 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_SSL_HOSTPORT and e.internal is False and e.scope_distance == 0]) assert 1 == len([e for e in _graph_output_events if e.type == "HTTP_RESPONSE" and str(e.module) == "http"]) assert 1 == len([e for e in _graph_output_events if e.type == "URL" and str(e.module) == "http"]) assert 1 == len([e for e in _graph_output_events if e.type == "URL_UNVERIFIED" and str(e.module) == "speculate"]) assert 1 == len([e for e in _graph_output_events if e.type == "DNS_NAME" and e.data == "test.notreal" and e.internal is False and e.scope_distance == 0 and str(e.module) == "sslcert"]) assert 1 == len([e for e in _graph_output_events if e.type == "DNS_NAME" and e.data == "www.bbottest.notreal" and e.internal is False and e.scope_distance == 1 and str(e.module) == "sslcert"]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "www.bbottest.notreal:9999"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"www.bbottest.notreal:{HTTPSERVER_SSL_PORT}"]) assert 0 == len([e for e in _graph_output_events if e.type == "DNS_NAME_UNRESOLVED" and e.data == "bbottest.notreal"]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "test.notreal:9999"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"test.notreal:{HTTPSERVER_SSL_PORT}"]) # sslcert with out-of-scope chain events, all_events, all_events_nodups, graph_output_events, graph_output_batch_events = await do_scan( "127.0.1.0", seeds=["127.0.0.0/31"], modules=["sslcert"], - _config={"scope": {"search_distance": 1, "report_distance": 0}, "speculate": True, "modules": {"speculate": {"ports": "9999"}}}, + _config={"scope": {"search_distance": 1, "report_distance": 0}, "speculate": True, "modules": {"speculate": {"ports": f"{HTTPSERVER_SSL_PORT}"}}}, _dns_mock={"www.bbottest.notreal": {"A": ["127.0.0.1"]}, "test.notreal": {"A": ["127.0.1.0"]}}, ) @@ -748,36 +763,36 @@ def custom_setup(scan): assert 1 == len([e for e in events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 1]) assert 0 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0"]) assert 0 == len([e for e in events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1"]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:9999"]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:9999"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_SSL_PORT}"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_SSL_HOSTPORT]) assert 0 == len([e for e in events if e.type == "DNS_NAME" and e.data == "test.notreal"]) assert 0 == len([e for e in events if e.type == "DNS_NAME" and e.data == "www.bbottest.notreal"]) - assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == "test.notreal:9999"]) + assert 0 == len([e for e in events if e.type == "OPEN_TCP_PORT" and e.data == f"test.notreal:{HTTPSERVER_SSL_PORT}"]) assert len(all_events) == 6 assert 1 == len([e for e in all_events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 1]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0" and e.internal is True and e.scope_distance == 2]) assert 1 == len([e for e in all_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is True and e.scope_distance == 2]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:9999" and e.internal is True and e.scope_distance == 2]) - assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:9999" and e.internal is True and e.scope_distance == 2]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_SSL_PORT}" and e.internal is True and e.scope_distance == 2]) + assert 1 == len([e for e in all_events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_SSL_HOSTPORT and e.internal is True and e.scope_distance == 2]) assert len(all_events_nodups) == 6 assert 1 == len([e for e in all_events_nodups if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 1]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.0" and e.internal is True and e.scope_distance == 2]) assert 1 == len([e for e in all_events_nodups if e.type == "IP_ADDRESS" and e.data == "127.0.0.1" and e.internal is True and e.scope_distance == 2]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:9999" and e.internal is True and e.scope_distance == 2]) - assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:9999" and e.internal is True and e.scope_distance == 2]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_SSL_PORT}" and e.internal is True and e.scope_distance == 2]) + assert 1 == len([e for e in all_events_nodups if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_SSL_HOSTPORT and e.internal is True and e.scope_distance == 2]) for _graph_output_events in (graph_output_events, graph_output_batch_events): assert len(_graph_output_events) == 3 assert 1 == len([e for e in _graph_output_events if e.type == "IP_RANGE" and e.data == "127.0.0.0/31" and e.internal is False and e.scope_distance == 1]) assert 0 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.0"]) assert 0 == len([e for e in _graph_output_events if e.type == "IP_ADDRESS" and e.data == "127.0.0.1"]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.0:9999"]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "127.0.0.1:9999"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"127.0.0.0:{HTTPSERVER_SSL_PORT}"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == HTTPSERVER_SSL_HOSTPORT]) assert 0 == len([e for e in _graph_output_events if e.type == "DNS_NAME" and e.data == "test.notreal"]) assert 0 == len([e for e in _graph_output_events if e.type == "DNS_NAME" and e.data == "www.bbottest.notreal"]) - assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == "test.notreal:9999"]) + assert 0 == len([e for e in _graph_output_events if e.type == "OPEN_TCP_PORT" and e.data == f"test.notreal:{HTTPSERVER_SSL_PORT}"]) @pytest.mark.asyncio @@ -786,7 +801,7 @@ async def test_manager_blacklist(bbot_scanner, bbot_httpserver, caplog): bbot_httpserver.expect_request(uri="/").respond_with_data(response_data="") # dns search distance = 1, report distance = 0 - scan = bbot_scanner("127.0.0.0/29", "test.notreal", seeds=["http://127.0.0.1:8888"], modules=["http"], config={"excavate": True, "dns": {"minimal": False, "search_distance": 1}, "scope": {"report_distance": 0}}, blacklist=["127.0.0.64/29"]) + scan = bbot_scanner("127.0.0.0/29", "test.notreal", seeds=[HTTPSERVER_URL], modules=["http"], config={"excavate": True, "dns": {"minimal": False, "search_distance": 1}, "scope": {"report_distance": 0}}, blacklist=["127.0.0.64/29"]) await scan._prep() await scan.helpers.dns._mock_dns({"www-prod.test.notreal": {"A": ["127.0.0.66"]}, "www-dev.test.notreal": {"A": ["127.0.0.22"]}}) @@ -835,13 +850,13 @@ async def test_scope_accuracy_with_special_urls(bbot_scanner, bbot_httpserver): was causing special URLs to be rejected by critical internal modules like `_scan_egress`, leading to the output of unwanted URLs. """ bbot_httpserver.expect_request(uri="/v2/users/spacex").respond_with_data(response_data="") - bbot_httpserver.expect_request(uri="/u/spacex").respond_with_data(response_data="") + bbot_httpserver.expect_request(uri="/u/spacex").respond_with_data(response_data=f"") scan = bbot_scanner("ORG:spacex", modules=["http", "social", "dockerhub"], config={"speculate": True, "excavate": True}) await scan._prep() - scan.modules["dockerhub"].site_url = "http://127.0.0.1:8888" - scan.modules["dockerhub"].api_url = "http://127.0.0.1:8888/v2" + scan.modules["dockerhub"].site_url = HTTPSERVER_URL + scan.modules["dockerhub"].api_url = f"{HTTPSERVER_URL}/v2" async def mock_wildcard(*args, **kwargs): return False @@ -868,7 +883,7 @@ async def handle_event(self, event): # there are actually 2 URL events. They are both from the same URL, but one was extracted by the full URL regex, and the other by the src/href= regex. # however, they should be deduped by scan_ingress. - bad_url_events = [e for e in dummy_module.events if e.type == "URL_UNVERIFIED" and e.url == "http://127.0.0.1:8888/asdf.js"] + bad_url_events = [e for e in dummy_module.events if e.type == "URL_UNVERIFIED" and e.url == f"{HTTPSERVER_URL}/asdf.js"] assert len(bad_url_events) == 1 # they should both be internal assert all(e.internal is True for e in bad_url_events) diff --git a/bbot/test/test_step_1/test_presets.py b/bbot/test/test_step_1/test_presets.py index d1a4117866..49db7b78a8 100644 --- a/bbot/test/test_step_1/test_presets.py +++ b/bbot/test/test_step_1/test_presets.py @@ -4,6 +4,7 @@ from ..bbot_fixtures import * # noqa F401 from bbot.scanner import Scanner, Preset +from bbot.test.worker import worker_dir # FUTURE TODO: @@ -819,7 +820,7 @@ def test_preset_include(): custom_preset_dir_1 = bbot_test_dir / "custom_preset_dir" custom_preset_dir_2 = custom_preset_dir_1 / "preset_subdir" custom_preset_dir_3 = custom_preset_dir_2 / "subsubdir" - custom_preset_dir_4 = Path("/tmp/.bbot_preset_test") + custom_preset_dir_4 = worker_dir("/tmp/.bbot_preset_test") custom_preset_dir_5 = custom_preset_dir_4 / "subdir" mkdir(custom_preset_dir_1) mkdir(custom_preset_dir_2) diff --git a/bbot/test/test_step_1/test_python_api.py b/bbot/test/test_step_1/test_python_api.py index 1b3ce237fc..cb685980ea 100644 --- a/bbot/test/test_step_1/test_python_api.py +++ b/bbot/test/test_step_1/test_python_api.py @@ -1,5 +1,7 @@ from ..bbot_fixtures import * +from bbot.test.worker import worker_dir + @pytest.mark.asyncio async def test_python_api(clean_default_config): @@ -44,7 +46,7 @@ async def test_python_api(clean_default_config): assert "scan_logging_test" in open(debug_log).read() # make sure config loads properly - bbot_home = "/tmp/.bbot_python_api_test" + bbot_home = str(worker_dir("/tmp/.bbot_python_api_test")) scan4 = Scanner("127.0.0.1", config={"home": bbot_home}) await scan4._prep() assert os.environ["BBOT_TOOLS"] == str(Path(bbot_home) / "tools") @@ -88,7 +90,7 @@ async def test_python_api_sync(clean_default_config): out_file = scan2.helpers.scans_dir / "python_api_test" / "output.json" assert list(scan2.helpers.read_file(out_file)) # make sure config loads properly - bbot_home = "/tmp/.bbot_python_api_test" + bbot_home = str(worker_dir("/tmp/.bbot_python_api_test")) scan3 = Scanner("127.0.0.1", config={"home": bbot_home}) await scan3._prep() assert os.environ["BBOT_TOOLS"] == str(Path(bbot_home) / "tools") diff --git a/bbot/test/test_step_1/test_scope.py b/bbot/test/test_step_1/test_scope.py index 52ee4462a1..3b094e4d25 100644 --- a/bbot/test/test_step_1/test_scope.py +++ b/bbot/test/test_step_1/test_scope.py @@ -1,9 +1,10 @@ from ..bbot_fixtures import * # noqa: F401 from ..test_step_2.module_tests.base import ModuleTestBase +from bbot.test.worker import HTTPSERVER_PORT, HTTPSERVER_URL class TestScopeBaseline(ModuleTestBase): - targets = ["http://127.0.0.1:8888"] + targets = [HTTPSERVER_URL] modules_overrides = ["http"] config_overrides = {"omit_event_types": []} @@ -42,7 +43,7 @@ def check(self, module_test, events): for e in events if e.type == "HTTP_RESPONSE" and str(e.host) == "127.0.0.1" - and e.port == 8888 + and e.port == HTTPSERVER_PORT and e.scope_distance == 0 ] ) @@ -50,7 +51,10 @@ def check(self, module_test, events): [ e for e in events - if e.type == "URL" and str(e.host) == "127.0.0.1" and e.port == 8888 and e.scope_distance == 0 + if e.type == "URL" + and str(e.host) == "127.0.0.1" + and e.port == HTTPSERVER_PORT + and e.scope_distance == 0 ] ) diff --git a/bbot/test/test_step_1/test_target.py b/bbot/test/test_step_1/test_target.py index 2df98df652..4a09a24372 100644 --- a/bbot/test/test_step_1/test_target.py +++ b/bbot/test/test_step_1/test_target.py @@ -1,4 +1,5 @@ from ..bbot_fixtures import * # noqa: F401 +from bbot.test.worker import HTTPSERVER_URL @pytest.mark.asyncio @@ -665,26 +666,26 @@ async def test_blacklist_regex(bbot_scanner, bbot_httpserver): assert "http://test.com/asdf/123456.aspx" in blacklist bbot_httpserver.expect_request(uri="/").respond_with_data( - """ - - + f""" + + """ ) bbot_httpserver.expect_request(uri="/asdfevilasdf").respond_with_data("") bbot_httpserver.expect_request(uri="/logout.aspx").respond_with_data("") # make sure URL is detected normally - scan = bbot_scanner("http://127.0.0.1:8888/", presets=["spider"], config={"excavate": True}, debug=True) + scan = bbot_scanner(f"{HTTPSERVER_URL}/", presets=["spider"], config={"excavate": True}, debug=True) await scan._prep() assert {r.pattern for r in scan.target.blacklist.blacklist_regexes} == {r"/.*(sign|log)[_-]?out"} events = [e async for e in scan.async_start()] urls = [e.url for e in events if e.type == "URL"] assert len(urls) == 2 - assert set(urls) == {"http://127.0.0.1:8888/", "http://127.0.0.1:8888/asdfevil333asdf"} + assert set(urls) == {f"{HTTPSERVER_URL}/", f"{HTTPSERVER_URL}/asdfevil333asdf"} # same scan again but with blacklist regex scan = bbot_scanner( - "http://127.0.0.1:8888/", + f"{HTTPSERVER_URL}/", blacklist=[r"RE:evil[0-9]{3}"], presets=["spider"], config={"excavate": True}, @@ -700,7 +701,7 @@ async def test_blacklist_regex(bbot_scanner, bbot_httpserver): events = [e async for e in scan.async_start()] urls = [e.url for e in events if e.type == "URL"] assert len(urls) == 1 - assert set(urls) == {"http://127.0.0.1:8888/"} + assert set(urls) == {f"{HTTPSERVER_URL}/"} def test_blacklist_get_invalid_host(): diff --git a/bbot/test/test_step_1/test_web.py b/bbot/test/test_step_1/test_web.py index d3a0a7bdf6..530cc66b97 100644 --- a/bbot/test/test_step_1/test_web.py +++ b/bbot/test/test_step_1/test_web.py @@ -4,6 +4,8 @@ from ..bbot_fixtures import * +from bbot.test.worker import BBOT_TEST_DIR + @pytest.mark.asyncio async def test_web(bbot_scanner, bbot_httpserver, blasthttp_mock): @@ -204,7 +206,7 @@ async def test_web_helpers(bbot_scanner, bbot_httpserver, blasthttp_mock): assert scan1.helpers.is_cached(url) with open(filename) as f: assert f.read() == download_content - filename = Path("/tmp/bbot_download_test_file") + filename = BBOT_TEST_DIR / "bbot_download_test_file" filename.unlink(missing_ok=True) filename2 = await scan1.helpers.download(url, filename=filename) assert filename2 == filename diff --git a/bbot/test/test_step_1/test_worker_count.py b/bbot/test/test_step_1/test_worker_count.py new file mode 100644 index 0000000000..e6c50f231f --- /dev/null +++ b/bbot/test/test_step_1/test_worker_count.py @@ -0,0 +1,81 @@ +from unittest.mock import patch + +from bbot.test.worker_count import RESERVE_MB, MB_PER_WORKER, cpu_count, worker_count + + +def _sysconf(total_mb): + def inner(name): + if name == "SC_PAGE_SIZE": + return 4096 + if name == "SC_PHYS_PAGES": + return total_mb * 1024 * 1024 // 4096 + raise ValueError(name) + + return inner + + +def test_worker_count_env_override_wins(): + with patch.dict("os.environ", {"BBOT_TEST_WORKERS": "3"}): + assert worker_count() == 3 + with patch.dict("os.environ", {"BBOT_TEST_WORKERS": "999"}): + assert worker_count() == 999 + + +def test_worker_count_env_override_is_sanitized(): + with patch.dict("os.environ", {"BBOT_TEST_WORKERS": " 4 "}): + assert worker_count() == 4 + for bogus in ("0", "-1"): + with patch.dict("os.environ", {"BBOT_TEST_WORKERS": bogus}): + assert worker_count() == 1 + + +def test_worker_count_ignores_blank_override(): + with patch.dict("os.environ", {"BBOT_TEST_WORKERS": " "}): + with patch("bbot.test.worker_count.cpu_count", return_value=4): + with patch("os.sysconf", _sysconf(RESERVE_MB + 4 * MB_PER_WORKER)): + assert worker_count() == 4 + + +def test_worker_count_is_capped_by_memory(): + total = RESERVE_MB + 2 * MB_PER_WORKER + with patch.dict("os.environ", {}, clear=True): + with patch("bbot.test.worker_count.cpu_count", return_value=64): + with patch("os.sysconf", _sysconf(total)): + assert worker_count() == 2 + + +def test_worker_count_is_capped_by_cores(): + with patch.dict("os.environ", {}, clear=True): + with patch("bbot.test.worker_count.cpu_count", return_value=4): + with patch("os.sysconf", _sysconf(RESERVE_MB + 64 * MB_PER_WORKER)): + assert worker_count() == 4 + + +def test_worker_count_never_returns_zero_on_small_machines(): + with patch.dict("os.environ", {}, clear=True): + with patch("bbot.test.worker_count.cpu_count", return_value=8): + for total in (RESERVE_MB // 2, RESERVE_MB, RESERVE_MB + 1): + with patch("os.sysconf", _sysconf(total)): + assert worker_count() == 1 + + +def test_worker_count_falls_back_to_cpus_when_memory_unknown(): + for exc in (ValueError, OSError, AttributeError): + with patch.dict("os.environ", {}, clear=True): + with patch("bbot.test.worker_count.cpu_count", return_value=6): + with patch("os.sysconf", side_effect=exc): + assert worker_count() == 6 + + +def test_cpu_count_prefers_affinity(): + with patch("os.sched_getaffinity", return_value={0, 1}): + assert cpu_count() == 2 + + +def test_cpu_count_falls_back_without_affinity(): + with patch("os.sched_getaffinity", side_effect=AttributeError): + with patch("os.cpu_count", return_value=12): + assert cpu_count() == 12 + with patch("os.sched_getaffinity", side_effect=AttributeError): + with patch("os.cpu_count", return_value=None): + assert cpu_count() == 1 diff --git a/bbot/test/test_step_1/test_worker_isolation.py b/bbot/test/test_step_1/test_worker_isolation.py new file mode 100644 index 0000000000..d7d887c583 --- /dev/null +++ b/bbot/test/test_step_1/test_worker_isolation.py @@ -0,0 +1,85 @@ +from pathlib import Path +from unittest.mock import patch + +from bbot.test.worker import ( + BASE_BBOT_TEST_DIR, + WORKER_PORT_STRIDE, + _worker_index, + worker_dir, + worker_id, + worker_port, +) + + +def _env(value): + if value is None: + return patch.dict("os.environ", {}, clear=True) + return patch.dict("os.environ", {"PYTEST_XDIST_WORKER": value}) + + +def test_worker_id_serial_is_empty(): + with _env(None): + assert worker_id() == "" + + +def test_worker_id_recognizes_gw_workers(): + for name in ("gw0", "gw1", "gw15"): + with _env(name): + assert worker_id() == name + + +def test_worker_id_rejects_non_gw_values(): + for name in ("master", "", "controller"): + with _env(name): + assert worker_id() == "" + + +def test_worker_index_parses_the_number(): + for name, expected in (("gw0", 0), ("gw1", 1), ("gw15", 15)): + with _env(name): + assert _worker_index() == expected + + +def test_worker_index_defaults_to_zero(): + with _env(None): + assert _worker_index() == 0 + with _env("gwX"): + assert _worker_index() == 0 + + +def test_worker_port_is_unchanged_when_serial(): + with _env(None): + assert worker_port(8888) == 8888 + + +def test_worker_port_blocks_do_not_overlap(): + bases = [8888, 9999, 5556, 8978, 8765] + seen = {} + for i in range(16): + with _env(f"gw{i}"): + for base in bases: + port = worker_port(base) + assert port not in seen, f"port {port} collides: gw{i}/{base} vs {seen[port]}" + seen[port] = f"gw{i}/{base}" + + +def test_worker_port_uses_the_stride(): + with _env("gw3"): + assert worker_port(8888) == 8888 + 3 * WORKER_PORT_STRIDE + + +def test_worker_dir_is_unchanged_when_serial(): + with _env(None): + assert worker_dir() == BASE_BBOT_TEST_DIR + assert worker_dir("/tmp/.bbot_example") == Path("/tmp/.bbot_example") + + +def test_worker_dir_is_unique_per_worker(): + seen = set() + for i in range(16): + with _env(f"gw{i}"): + d = worker_dir() + assert d not in seen + seen.add(d) + with _env("gw2"): + assert worker_dir("/tmp/.bbot_example") == Path("/tmp/.bbot_example_gw2") diff --git a/bbot/test/test_step_2/module_tests/test_module_ajaxpro.py b/bbot/test/test_step_2/module_tests/test_module_ajaxpro.py index 21c44f8519..1c3fb4dc9f 100644 --- a/bbot/test/test_step_2/module_tests/test_module_ajaxpro.py +++ b/bbot/test/test_step_2/module_tests/test_module_ajaxpro.py @@ -1,8 +1,9 @@ from .base import ModuleTestBase +from bbot.test.worker import HTTPSERVER_URL class TestAjaxpro(ModuleTestBase): - targets = ["http://127.0.0.1:8888"] + targets = [HTTPSERVER_URL] modules_overrides = ["http", "ajaxpro"] exploit_headers = {"X-Ajaxpro-Method": "AddItem", "Content-Type": "text/json; charset=UTF-8"} exploit_response = """ @@ -37,8 +38,7 @@ def check(self, module_test, events): if ( e.type == "FINDING" and "Ajaxpro Deserialization RCE (CVE-2021-23758)" in e.data["description"] - and "http://127.0.0.1:8888/ajaxpro/AjaxPro.Services.ICartService,AjaxPro.2.ashx" - in e.data["description"] + and f"{HTTPSERVER_URL}/ajaxpro/AjaxPro.Services.ICartService,AjaxPro.2.ashx" in e.data["description"] ): ajaxpro_exploit_detection = True diff --git a/bbot/test/test_step_2/module_tests/test_module_aspnet_bin_exposure.py b/bbot/test/test_step_2/module_tests/test_module_aspnet_bin_exposure.py index 2af603865f..b51f6fc97a 100644 --- a/bbot/test/test_step_2/module_tests/test_module_aspnet_bin_exposure.py +++ b/bbot/test/test_step_2/module_tests/test_module_aspnet_bin_exposure.py @@ -1,5 +1,6 @@ from .base import ModuleTestBase import re +from bbot.test.worker import HTTPSERVER_URL class TestAspnetBinExposure(ModuleTestBase): @@ -11,7 +12,7 @@ class TestAspnetBinExposure(ModuleTestBase): assertion on the Detection Url pattern would fail. """ - targets = ["http://127.0.0.1:8888"] + targets = [HTTPSERVER_URL] modules_overrides = ["http", "aspnet_bin_exposure"] async def setup_before_prep(self, module_test): @@ -52,7 +53,7 @@ def check(self, module_test, events): class TestAspnetBinExposure_DeadHost(ModuleTestBase): """Dead host returns 404 for everything -- no FINDING should be emitted.""" - targets = ["http://127.0.0.1:8888"] + targets = [HTTPSERVER_URL] modules_overrides = ["http", "aspnet_bin_exposure"] async def setup_before_prep(self, module_test): @@ -68,7 +69,7 @@ def check(self, module_test, events): class TestAspnetBinExposure_FalsePositive(ModuleTestBase): """Host serves DLLs for everything (including the fake DLL) -- no FINDING should be emitted.""" - targets = ["http://127.0.0.1:8888"] + targets = [HTTPSERVER_URL] modules_overrides = ["http", "aspnet_bin_exposure"] async def setup_before_prep(self, module_test): diff --git a/bbot/test/test_step_2/module_tests/test_module_asset_inventory.py b/bbot/test/test_step_2/module_tests/test_module_asset_inventory.py index 39aca71341..c066832bea 100644 --- a/bbot/test/test_step_2/module_tests/test_module_asset_inventory.py +++ b/bbot/test/test_step_2/module_tests/test_module_asset_inventory.py @@ -1,13 +1,16 @@ from .base import ModuleTestBase +from bbot.test.worker import HTTPSERVER_SSL_HOSTPORT, HTTPSERVER_SSL_PORT class TestAsset_Inventory(ModuleTestBase): targets = ["127.0.0.1", "bbottest.notreal"] scan_name = "asset_inventory_test" - config_overrides = {"dns": {"minimal": False}, "modules": {"portscan": {"ports": "9999"}}} + config_overrides = {"dns": {"minimal": False}, "modules": {"portscan": {"ports": f"{HTTPSERVER_SSL_PORT}"}}} modules_overrides = ["asset_inventory", "portscan", "sslcert"] - masscan_output = """{ "ip": "127.0.0.1", "timestamp": "1680197558", "ports": [ {"port": 9999, "proto": "tcp", "status": "open", "reason": "syn-ack", "ttl": 54} ] }""" + # Mocked masscan output. The port here is what the module "discovers", so it + # has to match the port the assertions expect, which moves per xdist worker. + masscan_output = f"""{{ "ip": "127.0.0.1", "timestamp": "1680197558", "ports": [ {{"port": {HTTPSERVER_SSL_PORT}, "proto": "tcp", "status": "open", "reason": "syn-ack", "ttl": 54}} ] }}""" async def setup_after_prep(self, module_test): async def run_masscan(command, *args, **kwargs): @@ -32,7 +35,7 @@ async def run_masscan(command, *args, **kwargs): ) def check(self, module_test, events): - assert any(e.data == "127.0.0.1:9999" for e in events), "No open port found" + assert any(e.data == HTTPSERVER_SSL_HOSTPORT for e in events), "No open port found" assert any(e.data == "www.bbottest.notreal" for e in events), "No DNS name found" filename = next(module_test.scan.home.glob("asset-inventory.csv")) with open(filename) as f: @@ -51,8 +54,21 @@ class TestAsset_InventoryEmitPrevious(TestAsset_Inventory): config_overrides = {"dns": {"minimal": False}, "modules": {"asset_inventory": {"use_previous": True}}} modules_overrides = ["asset_inventory"] + async def setup_before_prep(self, module_test): + await super().setup_before_prep(module_test) + # use_previous=True reads the asset-inventory.csv left behind by a + # prior scan of the same name. Relying on TestAsset_Inventory to have + # produced it makes this test order-dependent, which does not survive + # xdist: sibling tests can run in any order, and on any worker. Seed + # the file instead so the test stands on its own. + scan_home = module_test.scan.home + scan_home.mkdir(parents=True, exist_ok=True) + with open(scan_home / "asset-inventory.csv", "w") as f: + f.write("Host,Provider,IP(s),Status,Open Ports,Risk Rating,Findings,Description\n") + f.write(f"www.bbottest.notreal,,127.0.0.1,Active,{HTTPSERVER_SSL_PORT},,,\n") + def check(self, module_test, events): - assert any(e.data == "www.bbottest.notreal:9999" for e in events), "No open port found" + assert any(e.data == f"www.bbottest.notreal:{HTTPSERVER_SSL_PORT}" for e in events), "No open port found" assert any(e.data == "www.bbottest.notreal" for e in events), "No DNS name found" filename = next(module_test.scan.home.glob("asset-inventory.csv")) with open(filename) as f: diff --git a/bbot/test/test_step_2/module_tests/test_module_baddns.py b/bbot/test/test_step_2/module_tests/test_module_baddns.py index ecda4f8824..2ef77da8df 100644 --- a/bbot/test/test_step_2/module_tests/test_module_baddns.py +++ b/bbot/test/test_step_2/module_tests/test_module_baddns.py @@ -7,6 +7,7 @@ from bbot.modules.baddns import SUBMODULE_MAX_SEVERITY, SUBMODULE_MAX_CONFIDENCE from .base import ModuleTestBase +from bbot.test.worker import HTTPSERVER_HOSTPORT, HTTPSERVER_PORT def _extract_finding_values(module_cls): @@ -183,7 +184,7 @@ def check(self, module_test, events): class TestBaddns_cname_signature(BaseTestBaddns): - targets = ["bad.dns:8888"] + targets = [f"bad.dns:{HTTPSERVER_PORT}"] modules_overrides = ["baddns", "speculate"] async def setup_after_prep(self, module_test): @@ -192,7 +193,7 @@ async def setup_after_prep(self, module_test): from baddns.lib.whoismanager import WhoisManager def set_target(self, target): - return "127.0.0.1:8888" + return HTTPSERVER_HOSTPORT expect_args = {"method": "GET", "uri": "/"} respond_args = {"response_data": "
Links
href -