252 lines
No EOL
8.4 KiB
Python
Executable file
252 lines
No EOL
8.4 KiB
Python
Executable file
#!/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() |