mirror of
https://github.com/tomaae/homeassistant-mikrotik_router.git
synced 2025-06-21 00:25:48 +02:00
117 lines
No EOL
3.2 KiB
Python
117 lines
No EOL
3.2 KiB
Python
"""Mikrotik API for Mikrotik Router."""
|
|
|
|
import ssl
|
|
import librouteros
|
|
import logging
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
#---------------------------
|
|
# MikrotikAPI
|
|
#---------------------------
|
|
class MikrotikAPI:
|
|
"""Handle all communication with the Mikrotik API."""
|
|
|
|
def __init__(self, host, username, password, port = 0, use_ssl = True, login_method = "plain", encoding = "utf-8"):
|
|
"""Initialize the Mikrotik Client."""
|
|
self._host = host
|
|
self._use_ssl = use_ssl
|
|
self._port = port
|
|
self._username = username
|
|
self._password = password
|
|
self._login_method = login_method
|
|
self._encoding = encoding
|
|
self._ssl_wrapper = None
|
|
|
|
self._connection = None
|
|
self._connected = False
|
|
self.error = ""
|
|
|
|
## Default ports
|
|
if not self._port:
|
|
self._port = 8729 if self._use_ssl else 8728
|
|
|
|
#---------------------------
|
|
# connect
|
|
#---------------------------
|
|
def connect(self):
|
|
"""Connect to Mikrotik device."""
|
|
self.error = ""
|
|
self._connected = False
|
|
|
|
kwargs = {
|
|
"encoding": self._encoding,
|
|
"login_methods": self._login_method,
|
|
"port": self._port,
|
|
}
|
|
|
|
if self._use_ssl:
|
|
if self._ssl_wrapper is None:
|
|
ssl_context = ssl.create_default_context()
|
|
ssl_context.check_hostname = False
|
|
ssl_context.verify_mode = ssl.CERT_NONE
|
|
self._ssl_wrapper = ssl_context.wrap_socket
|
|
kwargs["ssl_wrapper"] = self._ssl_wrapper
|
|
|
|
try:
|
|
self._connection = librouteros.connect(self._host, self._username, self._password, **kwargs)
|
|
except (
|
|
librouteros.exceptions.TrapError,
|
|
librouteros.exceptions.MultiTrapError,
|
|
librouteros.exceptions.ConnectionClosed,
|
|
librouteros.exceptions.ProtocolError,
|
|
librouteros.exceptions.FatalError
|
|
) as api_error:
|
|
_LOGGER.error("Mikrotik %s: %s", self._host, api_error)
|
|
self.errorToStrings("%s" % api_error)
|
|
self._connection = None
|
|
return False
|
|
else:
|
|
_LOGGER.info("Mikrotik Connected to %s", self._host)
|
|
self._connected = True
|
|
|
|
return self._connected
|
|
|
|
#---------------------------
|
|
# errorToStrings
|
|
#---------------------------
|
|
def errorToStrings(self, error):
|
|
self.error = "cannot_connect"
|
|
if error == "invalid user name or password (6)":
|
|
self.error = "wrong_login"
|
|
|
|
return
|
|
|
|
#---------------------------
|
|
# connected
|
|
#---------------------------
|
|
def connected(self):
|
|
"""Return connected boolean."""
|
|
return self._connected
|
|
|
|
#---------------------------
|
|
# path
|
|
#---------------------------
|
|
def path(self, path):
|
|
"""Retrieve data from Mikrotik API."""
|
|
if not self._connected or not self._connection:
|
|
if not self.connect():
|
|
return None
|
|
|
|
try:
|
|
response = self._connection.path(path)
|
|
tuple(response)
|
|
except librouteros.exceptions.ConnectionClosed:
|
|
_LOGGER.error("Mikrotik %s connection closed", self._host)
|
|
self._connected = False
|
|
self._connection = None
|
|
return None
|
|
except (
|
|
librouteros.exceptions.TrapError,
|
|
librouteros.exceptions.MultiTrapError,
|
|
librouteros.exceptions.ProtocolError,
|
|
librouteros.exceptions.FatalError
|
|
) as api_error:
|
|
_LOGGER.error("Mikrotik %s connection error %s", self._host, api_error)
|
|
return None
|
|
|
|
return response if response else None |