128 lines
5.3 KiB
Python
128 lines
5.3 KiB
Python
import socket, hashlib
|
|
|
|
class RouterOSApi:
|
|
def __init__(self, host, port=8728):
|
|
self.host = host; self.port = port; self.sock = None
|
|
def connect(self, username, password):
|
|
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
self.sock.settimeout(10)
|
|
self.sock.connect((self.host, self.port))
|
|
self._write_sentence('/login', f'=name={username}', f'=password={password}')
|
|
resp = self._read_response()
|
|
if any(isinstance(r, tuple) and r[0] == '!trap' for r in resp):
|
|
chal = None
|
|
for item in resp:
|
|
if isinstance(item, dict) and 'ret' in item: chal = item['ret']
|
|
if chal and len(chal) == 32:
|
|
response = '00' + hashlib.md5(b'\x00' + password.encode() + bytes.fromhex(chal)).hexdigest()
|
|
self._write_sentence('/login', f'=name={username}', f'=response={response}')
|
|
resp = self._read_response()
|
|
return self
|
|
def _encode_length(self, length):
|
|
if length < 0x80: return bytes([length])
|
|
elif length < 0x4000: length |= 0x8000; return bytes([(length>>8)&0xFF, length&0xFF])
|
|
elif length < 0x200000: length |= 0xC00000; return bytes([(length>>16)&0xFF, (length>>8)&0xFF, length&0xFF])
|
|
elif length < 0x10000000: length |= 0xE0000000; return bytes([(length>>24)&0xFF, (length>>16)&0xFF, (length>>8)&0xFF, length&0xFF])
|
|
else: return bytes([0xF0, (length>>24)&0xFF, (length>>16)&0xFF, (length>>8)&0xFF, length&0xFF])
|
|
def _write_word(self, word): self.sock.sendall(self._encode_length(len(word)) + word.encode('utf-8'))
|
|
def _write_sentence(self, cmd, *words):
|
|
self._write_word(cmd)
|
|
for w in words: self._write_word(w)
|
|
self.sock.sendall(b'\x00')
|
|
def _read_byte(self): return self.sock.recv(1)
|
|
def _read_word(self):
|
|
first = self._read_byte();
|
|
if first == b'\x00': return None
|
|
lb = first[0]
|
|
if lb & 0x80:
|
|
if (lb & 0xC0) == 0x80: length = ((lb & 0x3F) << 8) + self._read_byte()[0]
|
|
elif (lb & 0xE0) == 0xC0: length = ((lb & 0x1F) << 16) + (self._read_byte()[0] << 8) + self._read_byte()[0]
|
|
elif (lb & 0xF0) == 0xE0: length = ((lb & 0x0F) << 24) + (self._read_byte()[0] << 16) + (self._read_byte()[0] << 8) + self._read_byte()[0]
|
|
else: length = (self._read_byte()[0] << 24) + (self._read_byte()[0] << 16) + (self._read_byte()[0] << 8) + self._read_byte()[0]
|
|
else: length = lb
|
|
data = b''
|
|
while len(data) < length:
|
|
chunk = self.sock.recv(length - len(data))
|
|
if not chunk: break
|
|
data += chunk
|
|
return data.decode('utf-8', errors='replace')
|
|
def _read_sentence(self):
|
|
words = []
|
|
while True:
|
|
w = self._read_word()
|
|
if w is None: break
|
|
words.append(w)
|
|
return words
|
|
def _read_response(self):
|
|
responses = []
|
|
while True:
|
|
sentence = self._read_sentence()
|
|
if not sentence: break
|
|
reply = sentence[0]; rest = sentence[1:]
|
|
if reply == '!done': break
|
|
elif reply == '!re':
|
|
d = {}
|
|
for w in rest:
|
|
if '=' in w:
|
|
parts = w.split('=', 2)
|
|
d[parts[1]] = parts[2] if len(parts) >= 3 else ''
|
|
else: d[w] = ''
|
|
responses.append(d)
|
|
elif reply == '!trap': responses.append(('!trap', rest))
|
|
elif reply == '!fatal': raise Exception(f"Fatal: {rest}")
|
|
else: responses.append((reply, rest))
|
|
return responses
|
|
def cmd(self, command, **attrs):
|
|
words = [command]
|
|
for k, v in attrs.items():
|
|
if k.startswith('?'): words.append(f'{k}={v}')
|
|
else: words.append(f'={k}={v}')
|
|
self._write_sentence(*words)
|
|
return self._read_response()
|
|
def close(self):
|
|
if self.sock: self.sock.close(); self.sock = None
|
|
|
|
api = RouterOSApi('192.168.100.2')
|
|
api.connect('mi4', 'm14')
|
|
|
|
# Get Bulanan2 on-event
|
|
resp = api.cmd('/system/scheduler/print', **{'?name': 'Bulanan2'})
|
|
raw = resp[0]['on-event']
|
|
|
|
print(f"Raw length: {len(raw)}")
|
|
print(f"Raw first 200: {raw[:200]}")
|
|
print(f":pic count: {raw.count(':pic')}")
|
|
print(f":pick count: {raw.count(':pick')}")
|
|
|
|
# Find all :pic positions
|
|
idx = 0
|
|
positions = []
|
|
while True:
|
|
idx = raw.find(':pic', idx)
|
|
if idx == -1: break
|
|
positions.append(idx)
|
|
print(f" :pic at {idx}: ...{raw[max(0,idx-8):idx+12]}...")
|
|
idx += 4
|
|
|
|
print(f"\nReplacing :pic -> :pick...")
|
|
fixed = raw.replace(':pic', ':pick')
|
|
print(f"Fixed length: {len(fixed)}")
|
|
print(f"Fixed :pic count: {fixed.count(':pic')}")
|
|
print(f"Fixed :pick count: {fixed.count(':pick')}")
|
|
match_count = 0
|
|
for i, (a, b) in enumerate(zip(raw, fixed)):
|
|
if a != b:
|
|
match_count += 1
|
|
if match_count <= 10:
|
|
print(f" Diff at {i}: '{a}' -> '{b}' (ctx: ...{raw[max(0,i-5):i+5]}...)")
|
|
|
|
# Also try manual byte-level check
|
|
raw_bytes = raw.encode('utf-8')
|
|
fixed_bytes = fixed.encode('utf-8')
|
|
print(f"\nRaw bytes length: {len(raw_bytes)}")
|
|
print(f"Fixed bytes length: {len(fixed_bytes)}")
|
|
print(f"Expected diff: {raw.count(':pic')} chars * 1 = {raw.count(':pic')} diff")
|
|
print(f"Actual diff: {len(fixed_bytes) - len(raw_bytes)}")
|
|
|
|
api.close()
|