diff --git a/bbot/modules/deadly/legba.py b/bbot/modules/deadly/legba.py new file mode 100644 index 0000000000..17a315e879 --- /dev/null +++ b/bbot/modules/deadly/legba.py @@ -0,0 +1,248 @@ +import json +from pathlib import Path +from bbot.errors import WordlistError +from bbot.modules.base import BaseModule + +# key: value: +# List with `legba -L` +PROTOCOL_LEGBA_PLUGIN_MAP = { + "postgresql": "pgsql", +} + + +# Maps common protocol names to Legba protocol plugin names +def map_protocol_to_legba_plugin_name(common_protocol_name: str) -> str: + return PROTOCOL_LEGBA_PLUGIN_MAP.get(common_protocol_name, common_protocol_name) + + +class legba(BaseModule): + watched_events = ["PROTOCOL"] + produced_events = ["FINDING"] + flags = ["active", "aggressive", "deadly"] + per_hostport_only = True + meta = { + "description": "Credential bruteforcing supporting various services.", + "created_date": "2025-07-18", + "author": "@christianfl, @fuzikowski", + } + _module_threads = 25 + scope_distance_modifier = None + + options = { + "ssh_wordlist": "https://raw.githubusercontent.com/danielmiessler/SecLists/refs/heads/master/Passwords/Default-Credentials/ssh-betterdefaultpasslist.txt", + "ftp_wordlist": "https://raw.githubusercontent.com/danielmiessler/SecLists/refs/heads/master/Passwords/Default-Credentials/ftp-betterdefaultpasslist.txt", + "telnet_wordlist": "https://raw.githubusercontent.com/danielmiessler/SecLists/refs/heads/master/Passwords/Default-Credentials/telnet-betterdefaultpasslist.txt", + "vnc_wordlist": "https://raw.githubusercontent.com/danielmiessler/SecLists/refs/heads/master/Passwords/Default-Credentials/vnc-betterdefaultpasslist.txt", + "mssql_wordlist": "https://raw.githubusercontent.com/danielmiessler/SecLists/refs/heads/master/Passwords/Default-Credentials/mssql-betterdefaultpasslist.txt", + "mysql_wordlist": "https://raw.githubusercontent.com/danielmiessler/SecLists/refs/heads/master/Passwords/Default-Credentials/mysql-betterdefaultpasslist.txt", + "postgresql_wordlist": "https://raw.githubusercontent.com/danielmiessler/SecLists/refs/heads/master/Passwords/Default-Credentials/postgres-betterdefaultpasslist.txt", + "concurrency": 3, + "rate_limit": 3, + "version": "1.1.1", + } + + options_desc = { + "ssh_wordlist": "Wordlist URL for SSH combined username:password wordlist, newline separated", + "ftp_wordlist": "Wordlist URL for FTP combined username:password wordlist, newline separated", + "telnet_wordlist": "Wordlist URL for TELNET combined username:password wordlist, newline separated", + "vnc_wordlist": "Wordlist URL for VNC password wordlist, newline separated", + "mssql_wordlist": "Wordlist URL for MSSQL combined username:password wordlist, newline separated", + "mysql_wordlist": "Wordlist URL for MySQL combined username:password wordlist, newline separated", + "postgresql_wordlist": "Wordlist URL for PostgreSQL combined username:password wordlist, newline separated", + "concurrency": "Number of concurrent workers, gets overridden for SSH", + "rate_limit": "Limit the number of requests per second, gets overridden for SSH", + "version": "legba version", + } + + deps_ansible = [ + { + "name": "Download legba (x86)", + "unarchive": { + "src": "https://github.com/evilsocket/legba/releases/download/#{BBOT_MODULES_LEGBA_VERSION}/legba-#{BBOT_MODULES_LEGBA_VERSION}-linux-x86_64.tar.gz", + "dest": "#{BBOT_TEMP}", + "include": "legba-#{BBOT_MODULES_LEGBA_VERSION}-linux-x86_64/legba", + "remote_src": True, + }, + "when": "ansible_facts['system'] == 'Linux' and ansible_facts['architecture'] == 'x86_64'", + }, + { + "name": "Install legba (x86)", + "copy": { + "src": "#{BBOT_TEMP}/legba-#{BBOT_MODULES_LEGBA_VERSION}-linux-x86_64/legba", + "dest": "#{BBOT_TOOLS}/", + "mode": "u+x,g+x,o+x", + }, + "when": "ansible_facts['system'] == 'Linux' and ansible_facts['architecture'] == 'x86_64'", + }, + { + "name": "Download legba (ARM64)", + "unarchive": { + "src": "https://github.com/evilsocket/legba/releases/download/#{BBOT_MODULES_LEGBA_VERSION}/legba-#{BBOT_MODULES_LEGBA_VERSION}-linux-arm64.tar.gz", + "dest": "#{BBOT_TEMP}", + "include": "legba-#{BBOT_MODULES_LEGBA_VERSION}-linux-arm64/legba", + "remote_src": True, + }, + "when": "ansible_facts['system'] == 'Linux' and ansible_facts['architecture'] == 'aarch64'", + }, + { + "name": "Install legba (ARM64)", + "copy": { + "src": "#{BBOT_TEMP}/legba-#{BBOT_MODULES_LEGBA_VERSION}-linux-arm64/legba", + "dest": "#{BBOT_TOOLS}/", + "mode": "u+x,g+x,o+x", + }, + "when": "ansible_facts['system'] == 'Linux' and ansible_facts['architecture'] == 'aarch64'", + }, + ] + + async def setup(self): + self.output_dir = Path(self.scan.temp_dir / "legba-output") + self.helpers.mkdir(self.output_dir) + + return True + + async def filter_event(self, event): + handled_protocols = ["ssh", "ftp", "telnet", "vnc", "mssql", "mysql", "postgresql"] + + protocol = event.data["protocol"].lower() + if not protocol in handled_protocols: + return False, f"service {protocol} is currently not supported or can't be bruteforced by Legba" + + return True + + async def handle_event(self, event): + host = str(event.host) + port = str(event.port) + protocol = event.data["protocol"].lower() + + command_data = await self.construct_command(host, port, protocol) + + if not command_data: + self.warning(f"Skipping {host}:{port} ({protocol}) due to errors while constructing the command") + return + + command, output_path = command_data + + await self.run_process(command) + + async for finding_event in self.parse_output(output_path, event): + await self.emit_event(finding_event) + + async def parse_output(self, output_filepath, event): + protocol = event.data["protocol"].lower() + + try: + with open(output_filepath) as file: + for line in file: + # example line (ssh): + # {"found_at":"2025-07-18T06:28:08.969812152+01:00","target":"localhost:22","plugin":"ssh","data":{"username":"user","password":"pass"},"partial":false} + line = line.strip() + + try: + data = json.loads(line)["data"] + username = data.get("username", "") + password = data.get("password", "") + + if username and password: + message_addition = f"{username}:{password}" + elif username: + message_addition = username + elif password: + message_addition = password + except Exception as e: + self.warning(f"Failed to parse Legba output ({line}), using raw output instead: {e}") + message_addition = f"raw output: {line}" + + yield self.make_event( + { + "severity": "CRITICAL", + "confidence": "CONFIRMED", + "host": str(event.host), + "port": str(event.port), + "description": f"Valid {protocol} credentials found - {message_addition}", + }, + "FINDING", + parent=event, + ) + except FileNotFoundError: + self.info( + f"Could not open Legba output file {output_filepath}. File is missing if no valid credentials could be found" + ) + except Exception as e: + self.warning(f"Error processing Legba output file {output_filepath}: {e}") + else: + self.helpers.delete_file(output_filepath) + + async def construct_command(self, host, port, protocol): + # -C Combo wordlist delimited by ':' + # -P Passwordlist + # --target Target (allowed: host, url, IP address, CIDR, @filename) + # --output-format Output file format + # --output Save results to this file + # -Q Do not report statistics + # + # --wait Wait time in milliseconds per login attempt + # --rate-limit Limit the number of requests per second + # --concurrency Number of concurrent workers + + # Example command to bruteforce SSH: + # + # legba ssh -C combolist.txt --target 127.0.0.1:22 --output-format jsonl --output out.txt -Q --wait 4000 --rate-limit 1 --concurrency 1 + + try: + wordlist_path = await self.helpers.wordlist(self.config.get(f"{protocol}_wordlist")) + except WordlistError as e: + self.warning(f"Error retrieving wordlist for protocol {protocol}: {e}") + return None + except Exception as e: + self.warning(f"Unexpected error during wordlist loading for protocol {protocol}: {e}") + return None + + protocol_plugin_name = map_protocol_to_legba_plugin_name(protocol) + output_path = Path(self.output_dir) / f"{host}_{port}.json" + + cmd = [ + "legba", + protocol_plugin_name, + ] + + if protocol == "vnc": + # use only passwords, not combinations + cmd += ["-P"] + + else: + # use combinations + cmd += ["-C"] + + # wrap IPv6 addresses in square brackets + if self.helpers.is_ip(host, version=6): + host = f"[{host}]" + + cmd += [ + wordlist_path, + "--target", + f"{host}:{port}", + "--output-format", + "jsonl", + "--output", + output_path, + "-Q", + ] + + if protocol == "ssh": + # With OpenSSH 9.8, the sshd_config option "PerSourcePenalties" was introduced (on by default) + # The penalty "authfail" defaults to 5 seconds, so bruteforcing fast will block access. + # Legba is not able to check that by itself, so the wait time is set to 5 s, rate limit to 1 and concurrency to 1 with SSH. + # See https://www.openssh.com/txt/release-9.8 + cmd += [ + "--wait", + "5000", + "--rate-limit", + "1", + "--concurrency", + "1", + ] + else: + cmd += ["--rate-limit", self.config.rate_limit, "--concurrency", self.config.concurrency] + + return cmd, output_path diff --git a/bbot/test/test_step_2/module_tests/test_module_legba.py b/bbot/test/test_step_2/module_tests/test_module_legba.py new file mode 100644 index 0000000000..96bda40fd8 --- /dev/null +++ b/bbot/test/test_step_2/module_tests/test_module_legba.py @@ -0,0 +1,100 @@ +from pathlib import Path +from .base import ModuleTestBase, tempwordlist +import pytest + + +@pytest.fixture(params=["ssh", "ftp", "telnet", "vnc", "mssql", "mysql", "postgresql"]) +def protocol(request): + return request.param + + +@pytest.fixture +def mock_legba_run_process(monkeypatch, request): + async def fake_run_process(self, cmd): + try: + # find index of `--output` in cmd + output_index = cmd.index("--output") + # output_path is directly after `--output` in cmd + output_path = Path(cmd[output_index + 1]) + except Exception as e: + raise Exception(f"Could not determine output file path from command {cmd}: {e}") + + protocol = request.getfixturevalue("protocol") + + expected_file_content_per_protocol = { + "ssh": '{"found_at":"2025-07-22T20:50:19.541305293+02:00","target":"127.0.0.1:2222","plugin":"ssh","data":{"username":"remnux","password":"malware"},"partial":false}', + "ftp": '{"found_at":"2025-07-22T20:51:19.541305293+02:00","target":"127.0.0.1:21","plugin":"ftp","data":{"username":"ftp_boot","password":"ftp_boot"},"partial":false}', + "telnet": '{"found_at":"2025-07-22T20:51:19.541305293+02:00","target":"127.0.0.1:23","plugin":"telnet","data":{"username":"guest","password":"guest"},"partial":false}', + "vnc": '{"found_at":"2025-07-22T20:51:19.541305293+02:00","target":"127.0.0.1:5900","plugin":"vnc","data":{"username":"Administrator","password":""},"partial":false}', + "mssql": '{"found_at":"2025-07-22T20:51:19.541305293+02:00","target":"127.0.0.1:1433","plugin":"mssql","data":{"username":"sa","password":"default"},"partial":false}', + "mysql": '{"found_at":"2025-07-22T20:51:19.541305293+02:00","target":"127.0.0.1:3306","plugin":"mysql","data":{"username":"root","password":"moves"},"partial":false}', + "postgresql": '{"found_at":"2025-07-22T20:51:19.541305293+02:00","target":"127.0.0.1:5432","plugin":"pgsql","data":{"username":"postgres","password":"postgres"},"partial":false}', + } + + output_path.write_text(expected_file_content_per_protocol[protocol]) + + from bbot.modules.base import BaseModule + + monkeypatch.setattr(BaseModule, "run_process", fake_run_process) + + +@pytest.mark.usefixtures("mock_legba_run_process") +class TestLegba(ModuleTestBase): + targets = ["127.0.0.1"] + + temp_ssh_wordlist = tempwordlist(["test:test", "admin:admin", "admin:password", "remnux:malware", "user:pass"]) + temp_ftp_wordlist = tempwordlist(["test:test", "ftp_boot:ftp_boot", "admin:password", "root:root", "user:pass"]) + temp_telnet_wordlist = tempwordlist(["test:test", "admin:admin", "admin:password", "root:root", "guest:guest"]) + temp_vnc_wordlist = tempwordlist(["test", "admin", "password", "Administrator", "pass"]) + temp_mssql_wordlist = tempwordlist(["sa:default", "admin:admin", "admin:password", "root:root", "user:pass"]) + temp_mysql_wordlist = tempwordlist(["test:test", "admin:admin", "root:moves", "root:root", "user:pass"]) + temp_postgresql_wordlist = tempwordlist(["postgres:postgres", "admin:admin", "admin:password", "user:pass"]) + + config_overrides = { + "modules": { + "legba": { + "ssh_wordlist": str(temp_ssh_wordlist), + "ftp_wordlist": str(temp_ftp_wordlist), + "telnet_wordlist": str(temp_telnet_wordlist), + "vnc_wordlist": str(temp_vnc_wordlist), + "mssql_wordlist": str(temp_mssql_wordlist), + "mysql_wordlist": str(temp_mysql_wordlist), + "postgresql_wordlist": str(temp_postgresql_wordlist), + } + } + } + + @pytest.fixture(autouse=True) + def _protocol_dependency(self, protocol): + # ensure pytest sees dependency and runs one test per protocol + self._protocol = protocol + + async def setup_after_prep(self, module_test): + protocol = module_test.request_fixture.getfixturevalue("protocol") + ports = {"ssh": 2222, "ftp": 21, "telnet": 23, "vnc": 5900, "mssql": 1433, "mysql": 3306, "postgresql": 5432} + event_data = {"host": str(self.targets[0]), "protocol": protocol.upper(), "port": ports[protocol]} + protocol_event = module_test.scan.make_event( + event_data, + "PROTOCOL", + parent=module_test.scan.root_event, + ) + + await module_test.module.emit_event(protocol_event) + + def check(self, module_test, events): + protocol = module_test.request_fixture.getfixturevalue("protocol") + finding_events = [e for e in events if e.type == "FINDING"] + + assert len(finding_events) == 1 + + expected_desc = { + "ssh": "Valid ssh credentials found - remnux:malware", + "ftp": "Valid ftp credentials found - ftp_boot:ftp_boot", + "telnet": "Valid telnet credentials found - guest:guest", + "vnc": "Valid vnc credentials found - Administrator", + "mssql": "Valid mssql credentials found - sa:default", + "mysql": "Valid mysql credentials found - root:moves", + "postgresql": "Valid postgresql credentials found - postgres:postgres", + } + + assert expected_desc[protocol] in finding_events[0].data["description"]