marco/ids.alfacom.it
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix connection issues with MikroTik routers

Browse Source 
Update the MikroTik manager to correctly use API ports (8728/8729) and SSL settings for establishing connections.

Replit-Commit-Author: Agent
Replit-Commit-Session-Id: 7a657272-55ba-4a79-9a2e-f1ed9bc7a528
Replit-Commit-Checkpoint-Type: intermediate_checkpoint
Replit-Commit-Event-Id: 84f094af-954b-41c6-893f-6ee7fd519235
Replit-Commit-Screenshot-Url: https://storage.googleapis.com/screenshot-production-us-central1/449cf7c4-c97a-45ae-8234-e5c5b8d6a84f/7a657272-55ba-4a79-9a2e-f1ed9bc7a528/jFtLBWL
This commit is contained in:
marco370 2025-11-25 17:49:26 +00:00
parent c4546f843f
commit a947ac8cea
1 changed files with 33 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
python_ml/mikrotik_manager.py
View File

@ -21,33 +21,39 @@ class MikroTikManager:
self.timeout = timeout
self.clients = {} # Cache di client HTTP per router
def _get_client(self, router_ip: str, username: str, password: str, port: int = 8728) -> httpx.AsyncClient:
def _get_client(self, router_ip: str, username: str, password: str, port: int = 8728, use_ssl: bool = False) -> httpx.AsyncClient:
"""Ottiene o crea client HTTP per un router"""
key = f"{router_ip}:{port}"
key = f"{router_ip}:{port}:{use_ssl}"
if key not in self.clients:
# API REST MikroTik usa porta HTTP/HTTPS (default 80/443)
# Per semplicità useremo richieste HTTP dirette
# API REST MikroTik:
# - Porta 8728: HTTP (default)
# - Porta 8729: HTTPS (SSL)
protocol = "https" if use_ssl or port == 8729 else "http"
auth = base64.b64encode(f"{username}:{password}".encode()).decode()
headers = {
"Authorization": f"Basic {auth}",
"Content-Type": "application/json"
}
self.clients[key] = httpx.AsyncClient(
base_url=f"http://{router_ip}",
base_url=f"{protocol}://{router_ip}:{port}",
headers=headers,
timeout=self.timeout
timeout=self.timeout,
verify=False # Disable SSL verification for self-signed certs
)
return self.clients[key]
async def test_connection(self, router_ip: str, username: str, password: str, port: int = 8728) -> bool:
async def test_connection(self, router_ip: str, username: str, password: str, port: int = 8728, use_ssl: bool = False) -> bool:
"""Testa connessione a un router"""
try:
client = self._get_client(router_ip, username, password, port)
# Auto-detect SSL: porta 8729 = SSL
if port == 8729:
use_ssl = True
client = self._get_client(router_ip, username, password, port, use_ssl)
# Prova a leggere system identity
response = await client.get("/rest/system/identity")
return response.status_code == 200
except Exception as e:
print(f"[ERROR] Connessione a {router_ip} fallita: {e}")
print(f"[ERROR] Connessione a {router_ip}:{port} fallita: {e}")
return False
async def add_address_list(
@ -59,14 +65,18 @@ class MikroTikManager:
list_name: str = "ddos_blocked",
comment: str = "",
timeout_duration: str = "1h",
port: int = 8728
port: int = 8728,
use_ssl: bool = False
) -> bool:
"""
Aggiunge IP alla address-list del router
timeout_duration: es. "1h", "30m", "1d"
"""
try:
client = self._get_client(router_ip, username, password, port)
# Auto-detect SSL: porta 8729 = SSL
if port == 8729:
use_ssl = True
client = self._get_client(router_ip, username, password, port, use_ssl)
# Controlla se IP già esiste
response = await client.get("/rest/ip/firewall/address-list")
@ -105,11 +115,15 @@ class MikroTikManager:
password: str,
ip_address: str,
list_name: str = "ddos_blocked",
port: int = 8728
port: int = 8728,
use_ssl: bool = False
) -> bool:
"""Rimuove IP dalla address-list del router"""
try:
client = self._get_client(router_ip, username, password, port)
# Auto-detect SSL: porta 8729 = SSL
if port == 8729:
use_ssl = True
client = self._get_client(router_ip, username, password, port, use_ssl)
# Trova ID dell'entry
response = await client.get("/rest/ip/firewall/address-list")
@ -139,11 +153,15 @@ class MikroTikManager:
username: str,
password: str,
list_name: Optional[str] = None,
port: int = 8728
port: int = 8728,
use_ssl: bool = False
) -> List[Dict]:
"""Ottiene address-list da router"""
try:
client = self._get_client(router_ip, username, password, port)
# Auto-detect SSL: porta 8729 = SSL
if port == 8729:
use_ssl = True
client = self._get_client(router_ip, username, password, port, use_ssl)
response = await client.get("/rest/ip/firewall/address-list")
if response.status_code == 200: