Files
iix/sync_routing.py

123 lines
4.5 KiB
Python

import os
import requests
import json
import ipaddress
from dotenv import load_dotenv
# Load configuration dari file .env
load_dotenv()
BGP_ROUTER_IP = os.getenv("BGP_ROUTER_IP")
BGP_ROUTER_USER = os.getenv("BGP_ROUTER_USER")
BGP_ROUTER_PASSWORD = os.getenv("BGP_ROUTER_PASSWORD")
# URL API RouterOS v7 untuk membaca tabel routing
API_URL = f"http://{BGP_ROUTER_IP}/rest/routing/route"
# Kredensial basic auth
auth = (BGP_ROUTER_USER, BGP_ROUTER_PASSWORD)
def get_local_bgp_routes():
"""Download file bgp_lokal_export.txt dari router BGP.
File ini diekspor secara berkala oleh scheduler MikroTik (bgp_lokal_scheduler)
yang menjalankan script bgp_lokal_export setiap hari jam 04:00.
File berisi daftar dst-address dari rute CDN (distance=15) dan NIX (distance=200).
"""
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
print(f"Menghubungi router {BGP_ROUTER_IP} via REST API...")
session = requests.Session()
session.auth = auth
session.verify = False
api_url = f"http://{BGP_ROUTER_IP}/rest"
export_file = "bgp_lokal_export.txt"
# 1. Cek apakah file export ada di router
print(f"Mengecek file {export_file} di router...")
try:
res = session.get(f"{api_url}/file", params={"name": export_file}, timeout=10)
if res.status_code == 200:
files = res.json()
if not files:
print(f" -> File {export_file} tidak ditemukan di router!")
print(" Pastikan script bgp_lokal_export sudah dibuat dan dijalankan di router.")
print(" Lihat instruksi di setup_scheduler.md")
return []
size = files[0].get("size", "0")
print(f" -> File ditemukan ({size} bytes)")
else:
print(f" -> Error cek file: {res.status_code}")
return []
except Exception as e:
print(f" -> Gagal koneksi ke router: {e}")
return []
# 2. Download isi file via /execute :put
print(f"Mengunduh isi {export_file}...")
try:
res = session.post(f"{api_url}/execute",
json={"script": f':put [/file get {export_file} contents]'},
timeout=30)
if res.status_code == 200:
content = res.json().get("ret", "")
if content:
# Parse: setiap baris adalah dst-address (contoh: 103.10.0.0/16)
routes = [line.strip() for line in content.split("\n")
if line.strip() and "/" in line.strip()]
# Hapus default routes
routes = [r for r in routes if r not in ("0.0.0.0/0", "::/0")]
print(f" -> Berhasil: {len(routes)} rute lokal diunduh")
return list(set(routes))
else:
print(" -> File kosong!")
return []
else:
print(f" -> Error download: {res.status_code} - {res.text[:100]}")
return []
except Exception as e:
print(f" -> Exception: {e}")
return []
def generate_address_list_script(route_list, list_name, filename):
"""Membentuk file .rsc yang berisi perintah create address-list MikroTik"""
if not route_list:
print(f"List kosong, tidak ada rute LOKAL yang ditemukan.")
return
print(f"Membuat file {filename} dengan {len(route_list)} subnet IP...")
with open(filename, "w") as f:
f.write(f"# Dibuat otomatis via Script API Python\n")
f.write(f"/ip firewall address-list\n")
# Bersihkan list lama sebelum memasukkan yang baru
f.write(f"remove [find list=\"{list_name}\"]\n")
for ip in route_list:
f.write(f"add list={list_name} address=\"{ip}\"\n")
print(f"File {filename} berhasil dibuat.")
if __name__ == "__main__":
if not all([BGP_ROUTER_IP, BGP_ROUTER_USER, BGP_ROUTER_PASSWORD]):
print("Error: Variabel kredensial/BGP_ROUTER_IP belum diset dengan benar di file .env")
exit(1)
print("--- Memulai Proses Ekstraksi Ringan Routing LOKAL ---")
local_ips = get_local_bgp_routes()
if local_ips:
print(f"\nRingkasan:")
print(f"Total Subnet IP Lokal & CDN: {len(local_ips)}")
# Generate script address-list khusus untuk trafik Lokal
generate_address_list_script(local_ips, "ip-lokal", "routing-lokal.rsc")
print("\nSelesai! File routing-lokal.rsc siap di-upload ke router distribusi.")
print("Di Router Distribusi jalankan perintah: /import file-name=routing-lokal.rsc")