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")