239 lines
7.7 KiB
Python
239 lines
7.7 KiB
Python
import socket
|
|
import struct
|
|
import time
|
|
|
|
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))
|
|
|
|
# Login
|
|
self._write_sentence('/login', f'=name={username}', f'=password={password}')
|
|
resp = self._read_response()
|
|
|
|
if '!trap' in resp:
|
|
# Try v6.43+ challenge-response auth
|
|
challenge = None
|
|
for item in resp:
|
|
if isinstance(item, dict) and 'ret' in item:
|
|
challenge = item['ret']
|
|
|
|
if challenge and len(challenge) == 32:
|
|
import hashlib
|
|
response = '00' + hashlib.md5(b'\x00' + password.encode() + bytes.fromhex(challenge)).hexdigest()
|
|
self._write_sentence('/login', f'=name={username}', f'=response={response}')
|
|
resp = self._read_response()
|
|
if '!trap' in resp:
|
|
raise Exception(f"Login failed: {resp}")
|
|
|
|
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):
|
|
data = word.encode('utf-8')
|
|
self.sock.sendall(self._encode_length(len(data)) + data)
|
|
|
|
def _write_sentence(self, cmd, *words):
|
|
self._write_word(cmd)
|
|
for w in words:
|
|
self._write_word(w)
|
|
self.sock.sendall(b'\x00') # End-of-sentence
|
|
|
|
def _read_byte(self):
|
|
return self.sock.recv(1)
|
|
|
|
def _read_word(self):
|
|
first = self._read_byte()
|
|
if first == b'\x00':
|
|
return None
|
|
|
|
length_byte = first[0]
|
|
if length_byte & 0x80:
|
|
if (length_byte & 0xC0) == 0x80:
|
|
length = ((length_byte & 0x3F) << 8) + self._read_byte()[0]
|
|
elif (length_byte & 0xE0) == 0xC0:
|
|
length = ((length_byte & 0x1F) << 16) + (self._read_byte()[0] << 8) + self._read_byte()[0]
|
|
elif (length_byte & 0xF0) == 0xE0:
|
|
b2 = self._read_byte()[0]; b3 = self._read_byte()[0]; b4 = self._read_byte()[0]
|
|
length = ((length_byte & 0x0F) << 24) + (b2 << 16) + (b3 << 8) + b4
|
|
else:
|
|
length = (self._read_byte()[0] << 24) + (self._read_byte()[0] << 16) + (self._read_byte()[0] << 8) + self._read_byte()[0]
|
|
else:
|
|
length = length_byte
|
|
|
|
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:
|
|
word = self._read_word()
|
|
if word is None:
|
|
break
|
|
words.append(word)
|
|
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)
|
|
if len(parts) >= 3:
|
|
d[parts[1]] = parts[2]
|
|
elif len(parts) == 2:
|
|
d[parts[0]] = parts[1]
|
|
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
|
|
|
|
# ===== MAIN FIX =====
|
|
api = RouterOSApi('192.168.100.2')
|
|
api.connect('mi4', 'm14')
|
|
|
|
schedulers = ["Bulanan2", "Bulanan4", "vc-2K3J", "vc-5K10J", "Family", "DevB2"]
|
|
|
|
print("=== Fixing scheduler on-event ===")
|
|
for s in schedulers:
|
|
print(f"\n--- {s} ---")
|
|
|
|
resp = api.cmd('/system/scheduler/print', **{'?name': s})
|
|
if not resp:
|
|
print(f" Not found")
|
|
continue
|
|
|
|
oe = resp[0].get('on-event', '')
|
|
oid = resp[0].get('.id', '')
|
|
|
|
if ':pic' not in oe:
|
|
print(f" No :pic found, skipping")
|
|
continue
|
|
|
|
fixed = oe.replace(':pic', ':pick')
|
|
count = oe.count(':pic')
|
|
print(f" Fixed {count} :pic -> :pick (len={len(fixed)})")
|
|
|
|
# Set on-event
|
|
set_resp = api.cmd('/system/scheduler/set', **{'.id': oid, 'on-event': fixed})
|
|
|
|
# Check for traps
|
|
is_trap = any(isinstance(r, tuple) and r[0] == '!trap' for r in set_resp)
|
|
if is_trap:
|
|
for r in set_resp:
|
|
if isinstance(r, tuple) and r[0] == '!trap':
|
|
print(f" !trap: {r[1]}")
|
|
else:
|
|
print(f" Set OK (no traps)")
|
|
|
|
# Verify
|
|
verify = api.cmd('/system/scheduler/print', **{'?name': s})
|
|
if verify:
|
|
oe2 = verify[0].get('on-event', '')
|
|
has_pic = ':pic' in oe2
|
|
has_pick = ':pick' in oe2
|
|
print(f" Verify: pic={has_pic} pick={has_pick}")
|
|
if not has_pic and has_pick:
|
|
print(f" FIXED!")
|
|
elif has_pic:
|
|
print(f" STILL BROKEN")
|
|
|
|
# Fix profile on-login
|
|
profiles = ["Bulanan2", "Bulanan4", "vc-2K3J", "vc-5K10J", "Family", "home", "DevB2"]
|
|
print("\n\n=== Fixing profile on-login ===")
|
|
for p in profiles:
|
|
print(f"\n--- {p} ---")
|
|
|
|
resp = api.cmd('/ip/hotspot/user/profile/print', **{'?name': p})
|
|
if not resp:
|
|
print(f" Not found")
|
|
continue
|
|
|
|
ol = resp[0].get('on-login', '')
|
|
oid = resp[0].get('.id', '')
|
|
|
|
if ':pic' not in ol:
|
|
print(f" No :pic found, skipping")
|
|
continue
|
|
|
|
fixed = ol.replace(':pic', ':pick')
|
|
count = ol.count(':pic')
|
|
print(f" Fixed {count} :pic -> :pick")
|
|
|
|
set_resp = api.cmd('/ip/hotspot/user/profile/set', **{'.id': oid, 'on-login': fixed})
|
|
|
|
is_trap = any(isinstance(r, tuple) and r[0] == '!trap' for r in set_resp)
|
|
if is_trap:
|
|
for r in set_resp:
|
|
if isinstance(r, tuple) and r[0] == '!trap':
|
|
print(f" !trap: {r[1]}")
|
|
else:
|
|
print(f" Set OK")
|
|
|
|
verify = api.cmd('/ip/hotspot/user/profile/print', **{'?name': p})
|
|
if verify:
|
|
ol2 = verify[0].get('on-login', '')
|
|
hp = ':pic' in ol2; hk = ':pick' in ol2
|
|
print(f" Verify: pic={hp} pick={hk}")
|
|
print(f" {'FIXED!' if not hp and hk else 'STILL BROKEN'}")
|
|
|
|
api.close()
|
|
print("\nALL DONE")
|