#!/usr/bin/env python3 import ssl import socket import hashlib import binascii import sys import json class MikrotikAPI: def __init__(self, host, username, password, port=8729): self.host = host self.port = port self.username = username self.password = password self.sock = None self.ssl_sock = None def connect(self): """Establish SSL connection to MikroTik router""" try: # Create socket and SSL context context = ssl.create_default_context() context.check_hostname = False context.verify_mode = ssl.CERT_NONE # Allow older SSL/TLS versions for compatibility context.set_ciphers('DEFAULT:@SECLEVEL=0') # Connect to router self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.settimeout(30) # 30 second timeout self.sock.connect((self.host, self.port)) self.ssl_sock = context.wrap_socket(self.sock) print(f"Connected to {self.host}:{self.port}") return True except Exception as e: print(f"Connection failed: {e}") return False def disconnect(self): """Close the connection""" if self.ssl_sock: self.ssl_sock.close() if self.sock: self.sock.close() def encode_length(self, length): """Encode length for MikroTik API protocol""" if length <= 0x7F: return bytes([length]) elif length <= 0x3FFF: return bytes([((length >> 8) & 0xFF) | 0x80, length & 0xFF]) elif length <= 0x1FFFFF: return bytes([((length >> 16) & 0xFF) | 0xC0, (length >> 8) & 0xFF, length & 0xFF]) elif length <= 0xFFFFFFF: return bytes([((length >> 24) & 0xFF) | 0xE0, (length >> 16) & 0xFF, (length >> 8) & 0xFF, length & 0xFF]) else: return bytes([0xF0, (length >> 24) & 0xFF, (length >> 16) & 0xFF, (length >> 8) & 0xFF, length & 0xFF]) def decode_length(self): """Decode length from MikroTik API protocol""" c = self.ssl_sock.recv(1)[0] if (c & 0x80) == 0x00: return c elif (c & 0xC0) == 0x80: return ((c & ~0xC0) << 8) + self.ssl_sock.recv(1)[0] elif (c & 0xE0) == 0xC0: data = self.ssl_sock.recv(2) return ((c & ~0xE0) << 16) + (data[0] << 8) + data[1] elif (c & 0xF0) == 0xE0: data = self.ssl_sock.recv(3) return ((c & ~0xF0) << 24) + (data[0] << 16) + (data[1] << 8) + data[2] elif (c & 0xF8) == 0xF0: data = self.ssl_sock.recv(4) return (data[0] << 24) + (data[1] << 16) + (data[2] << 8) + data[3] def write_word(self, word): """Send a word to the router""" word_bytes = word.encode('utf-8') self.ssl_sock.send(self.encode_length(len(word_bytes))) self.ssl_sock.send(word_bytes) def read_word(self): """Read a word from the router""" length = self.decode_length() if length == 0: return "" return self.ssl_sock.recv(length).decode('utf-8', 'ignore') def write_sentence(self, words): """Send a sentence (list of words) to the router""" for word in words: self.write_word(word) self.write_word("") def read_sentence(self): """Read a sentence from the router""" sentence = [] while True: word = self.read_word() if word == "": break sentence.append(word) return sentence def login(self): """Login to the router using plain password method""" # Send login command self.write_sentence(["/login", f"=name={self.username}", f"=password={self.password}"]) # Read response response = self.read_sentence() if response and response[0] == "!done": print("Login successful") return True else: print(f"Login failed: {response}") return False def command(self, cmd, params=None): """Execute a command on the router""" if params is None: params = [] # Send command sentence = [cmd] + params self.write_sentence(sentence) # Read response results = [] while True: sentence = self.read_sentence() if not sentence: break if sentence[0] == "!done": break elif sentence[0] == "!re": # Parse response data result = {} for item in sentence[1:]: if item.startswith("="): parts = item[1:].split("=", 1) if len(parts) == 2: result[parts[0]] = parts[1] else: result[parts[0]] = "" results.append(result) elif sentence[0] == "!trap": print(f"Error: {sentence}") break return results def main(): # Router connection details host = "10.254.254.101" username = "grahamro" password = "cFKhz8q5gPLoucMbcT1Iy58r3IXgc3" # Create API instance api = MikrotikAPI(host, username, password) try: # Connect and login if not api.connect(): return if not api.login(): return print("\nRetrieving IP addresses and subnets...") # Get all IP addresses addresses = api.command("/ip/address/print") print("\n=== IP Addresses and Subnets ===") for addr in addresses: interface = addr.get('interface', 'unknown') address = addr.get('address', 'unknown') network = addr.get('network', '') actual_interface = addr.get('actual-interface', interface) disabled = addr.get('disabled', 'false') dynamic = addr.get('dynamic', 'false') comment = addr.get('comment', '') status = [] if disabled == 'true': status.append('disabled') if dynamic == 'true': status.append('dynamic') status_str = f" ({', '.join(status)})" if status else "" comment_str = f" - {comment}" if comment else "" print(f"\nInterface: {interface} ({actual_interface}){status_str}") print(f" Address: {address}") print(f" Network: {network}{comment_str}") # Get routing table for additional subnet information print("\n\n=== Routing Table ===") routes = api.command("/ip/route/print") # Filter and display relevant routes for route in routes: dst = route.get('dst-address', '') gateway = route.get('gateway', '') distance = route.get('distance', '') scope = route.get('scope', '') active = route.get('active', 'false') dynamic = route.get('dynamic', 'false') comment = route.get('comment', '') # Skip default routes for clarity if dst == '0.0.0.0/0': continue status = [] if active != 'true': status.append('inactive') if dynamic == 'true': status.append('dynamic') status_str = f" ({', '.join(status)})" if status else "" comment_str = f" - {comment}" if comment else "" print(f"\nDestination: {dst}{status_str}") print(f" Gateway: {gateway}") if distance: print(f" Distance: {distance}") if scope and scope != '30': print(f" Scope: {scope}") if comment: print(f" Comment: {comment}") finally: api.disconnect() print("\nDisconnected from router") if __name__ == "__main__": main()