""" AUTHOR: Khushal P Soonderji DATE: Monday, 10th Feb., 2025. OBJECTIVE: To handle all MikroTik configuration from one place. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # System-level activities: import distro import socket import platform # My async utils: from utils_v2.string import json from utils_v2.date_time import date_time from utils_v2.database.async_mongo_v2 import AsyncMongo from utils_v2.api.async_quart import describe_exception # To make very controlled API calls: from utils_v2.rest.controllers.async_base import AsyncREST from utils_v2.rest.controllers.auth import BasicAuth from utils_v2.rest.models.api_call import ApiResponse # To work with IP Addresses: import ipaddress # To work with datatypes: from typing import List, Any # To make HTTP requests: import httpx # To work with date and time: import datetime # For asynchronous activities: import asyncio # Misc: import math import random import string # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # Info for logging that will stay constant during runtime: SERVER_HOSTNAME = str(socket.gethostname()) PLATFORM_INFO = platform.uname() HOST_OS = str(distro.name(True)) HOST_CPU = f"{PLATFORM_INFO.processor} ({PLATFORM_INFO.machine})" # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class AsyncMikroTik: # ┏┓┓ ┓┏ # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ ACTION_LOG_COLLECTION = "_mikrotikActions" # ┏┓ # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓ # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛ def __init__( self, config_by: str, mikrotik_ip: str, username: str, password: str, port: int = None, use_https: bool = True, http_client: httpx.AsyncClient = None, action_log_conn: AsyncMongo = None, debug: bool = True, debug_prefix: str = "MikroTik (C) | ", debug_only_errors: bool = True ): """ To control MikroTik devices in an async manner. :param http_client: The HTTP client to use to make REST-ful API calls. :param debug: Whether, or not, you would like to print debugging messages: :param debug_prefix: The prefix to print with the debugging messages. :param debug_only_errors: Whether you would like to print only error messages or all messages. :return: None. """ # Save the input values: self._config_by = config_by self._mikrotik_ip = mikrotik_ip self._username = username self._password = password self._port = port self._use_https = use_https # For logging actions to Mongo: self._action_log_conn = action_log_conn self._action_log_chain = None self.new_action_chain() # For controlled REST-ful calls: self._rest = AsyncREST( http_client = http_client, debug = debug, debug_prefix = debug_prefix, debug_only_errors = debug_only_errors ) self._basic_auth = BasicAuth( username = self._username, password = self._password ) # ┏┓ • # ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏ # ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛ # ┛ @property def action_chain(self): return self._action_log_chain # ┓┏ ┓ # ┣┫┏┓┃┏┓┏┓┏┓┏ # ┛┗┗ ┗┣┛┗ ┛ ┛ # ┛ def get_mikrotik_url( self, path: str, use_https: bool = None ) -> str: """ Simply creates the base URL for hitting the MikroTik server. :param path: The path of the REST API to hit. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: The Base URL string. """ use_https = self._use_https if use_https is None else use_https base_url = r"https://" if use_https else r"http://" base_url += self._mikrotik_ip if self._port is not None: base_url += f":{self._port}" base_url += "/rest" if not path.startswith("/"): path = "/" + path return base_url + path @staticmethod def split_ipv4_range_equally( start_ip: ipaddress.IPv4Address | str, end_ip: ipaddress.IPv4Address | str, targets: List ) -> List[dict]: """ Divides the IPv4 range over the list of targets (like VLAN ids) in a simple fashion :param start_ip: The first IP in the full pool (range). :param end_ip: The last IP in the full pool (range). :param targets: The list of targets (need not be sequential or ordered) like VLAN ids. :return: A list of dicts that describes the subnet for each target. """ # Parse the inputs: start_ip_obj = start_ip if isinstance(start_ip, ipaddress.IPv4Address) else ipaddress.IPv4Address(start_ip) end_ip_obj = end_ip if isinstance(end_ip, ipaddress.IPv4Address) else ipaddress.IPv4Address(end_ip) target_count = len(targets) # Calculate the total number of IPs in the range: total_ips = int(end_ip_obj) - int(start_ip_obj) + 1 # Figure out the no. of IPs in each block: ips_per_target = math.ceil(total_ips / target_count) # Iterate through the list of targets, and give them the IPs: subnets = [] current_ip = start_ip_obj for index, target in enumerate(targets): # Figure out the first and last IPs for this object: subnet_start_ip_obj = current_ip if index == target_count - 1: subnet_end_ip_obj = end_ip_obj else: subnet_end_ip_obj = subnet_start_ip_obj + ips_per_target - 1 # Add the calculated subnet to the list: subnets.append({ "target": target, "network": str(subnet_start_ip_obj) + "-" + str(subnet_end_ip_obj), "size": int(subnet_end_ip_obj) - int(subnet_start_ip_obj) + 1, "startIp": str(subnet_start_ip_obj), "endIp": str(subnet_end_ip_obj) }) # Update the current IP for the next subnet: current_ip += ips_per_target # Done here: return subnets @staticmethod def split_ipv4_range_in_powers_of_two( start_ip: ipaddress.IPv4Address | str, end_ip: ipaddress.IPv4Address | str, targets: List, consider_reserved_ips: bool = True, strict: bool = False ) -> List[dict]: """ Divides the IPv4 range over the list of targets (like VLAN ids) in blocks whose sizes are in powers of 2. This means that you won't have a block which has an unusual size like 7. :param start_ip: The first IP in the full pool (range). :param end_ip: The last IP in the full pool (range). :param targets: The list of targets (need not be sequential or ordered) like VLAN ids. :param consider_reserved_ips: Whether, or not, to add 3 IPs per block (network, gateway, and broadcast) when splitting the range. :param strict: Whether, or not, you want strict dvision of blocks such that the network address is always x.x.x.0 and the broadcast address is always x.x.x.255. :return: A list of dicts that describes the subnet for each target. """ # Parse the inputs: start_ip_obj = start_ip if isinstance(start_ip, ipaddress.IPv4Address) else ipaddress.IPv4Address(start_ip) end_ip_obj = end_ip if isinstance(end_ip, ipaddress.IPv4Address) else ipaddress.IPv4Address(end_ip) target_count = len(targets) # Calculate the total number of IPs in the range: total_ips = int(end_ip_obj) - int(start_ip_obj) + 1 # Calculate the no. of IPs each VLAN gets (subtract one because counting starts from 0), # and add three (if asked) because we need IPs for network, gateway and broadcast: ips_per_target = math.ceil(total_ips / target_count) - 1 if consider_reserved_ips: ips_per_target += 3 # Generate subnets for each VLAN: subnets = [] current_ip = start_ip_obj for target in targets: # Calculate the network address for the current VLAN: subnet_network = ipaddress.IPv4Network(f"{current_ip}/{32 - ips_per_target.bit_length()}", strict = strict) # Check for overlap with the parent network: subnet_start_ip_obj = subnet_network.network_address subnet_end_ip_obj = subnet_network.broadcast_address if subnet_start_ip_obj <= end_ip_obj: # Add the calculated subnet to the list: subnets.append({ "target": target, "network": str(subnet_network), "size": int(2 ** (32 - subnet_network.prefixlen)), "startIp": str(subnet_start_ip_obj), "endIp": str(subnet_end_ip_obj) }) # Update the current IP for the next subnet: current_ip = ipaddress.IPv4Address(int(subnet_network.broadcast_address) + 1) # Done here: return subnets def create_comment_json( self, config_ts: datetime.datetime = None, roll_back_config: dict = None ) -> str: """ To create a JSON string that can be put as a comment in any step to later identify the work that was done and have some hint about how to roll it back if needed. :param config_ts: The datetime (preferably UTC) when this comment was created. :param roll_back_config: The original config that can be used when rolling back. :return: The comment string to be used. """ # Create the JSON: comment_json = { "configBy": self._config_by, "configTs": (config_ts or date_time.get_current_utc_date_time(as_string = False)).timestamp(), "rollbackConfig": roll_back_config } # Return a string: return json.to_string(comment_json, no_space = True) @staticmethod def parse_comment_json(comment: str) -> dict | list | None: """ Tries to parse a comment as if it were a JSON string. :param comment: The comment string. :return: The dict/list parsed form the JSON, or a null value of the input was not a valid JSON string. """ # Start with a null value: parsed = None # Try to parse the comment as if it were a JSON string: try: parsed = json.from_string(comment) except: parsed = None # Done here: return parsed @staticmethod def get_original_config(comment_json: dict) -> dict | None: """ Once you have the comment's JSON unpacked successfully, pass that dict here to get the original config back. :param comment_json: The dict received from parsing the comment's JSON string. :return: The dict that describes the previous config, or None if no previous config was found. """ return comment_json.get("rollbackConfig") if isinstance(comment_json, dict) else None def is_my_config( self, config_json: dict, name_substring: str = None ) -> bool: """ Looks at a particular resource's JSON and tells if it was created by this class. :param config_json: The JSON of the resource whose config needs to be checked. :param name_substring: A substring (case-sensitive) to look for in the 'name' field. :return: True if the config was created by this class, else False. """ # Extract needed values: name = config_json.get("name", "???") comment_json = self.parse_comment_json(config_json.get("comment")) or {} # Perform needed checks: if ( name.find(name_substring or "bhopli-muchhi-cookie") >= 0 or comment_json.get("configBy", "???") == self._config_by ): return True else: return False def new_action_chain(self, action_chain: str | int = None) -> None: """ Generates a new random string to mark a new action chain and saves it in the local variable. :param action_chain: A custom action chain to start using. A random string will be created if not provided. :return: None """ if action_chain: self._action_log_chain = action_chain else: self._action_log_chain = "".join(random.choices(string.ascii_letters + string.digits, k = 8)) async def log_action( self, api_response: ApiResponse, ) -> None: """ Invoke this to record actions automatically. :param api_response: The response received from invoking an API provided by MikroTik. :return: None. """ # Proceed only if we have a logging connection: if self._action_log_conn is not None: # Prepare the final log: action_log_json = { "hostname": SERVER_HOSTNAME, "hostOs": HOST_OS, "hostCpu": HOST_CPU, "actionChain": self._action_log_chain, "configBy": self._config_by, "mikrotikIp": self._mikrotik_ip, "url": api_response.url, "method": api_response.method, "action": api_response.action, "config": api_response.requestJson, "httpCode": api_response.httpCode, "success": api_response.success, "message": api_response.message, "exception": describe_exception(api_response.exception), "ts": date_time.get_current_utc_date_time(as_string = False) } # Record the action: asyncio.create_task(self._action_log_conn.insert_one( collection = self.ACTION_LOG_COLLECTION, document = action_log_json )) @staticmethod async def get_failure_message(api_response: ApiResponse) -> str: # On successful API calls, MikroTik doesn't give messages: if api_response.success: message = "Success" # On failure, the message is available in the JSON payload: else: response_json = await api_response.get_json() message = f"{response_json.get('message')} ({response_json.get('detail')})" # Done here: return message # ┏┓ # ┗┓┓┏┏╋┏┓┏┳┓ # ┗┛┗┫┛┗┗ ┛┗┗ # ┛ async def list_system_resources( self, use_https: bool = None ) -> ApiResponse: """ To get a summary of the hardware resources available in the MikroTik device. This also becomes a great way to quickly check if any given device is valid, and up and running. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/system/resource", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() # Done here: await self.log_action(api_response) return api_response # ┳ ┏ # ┃┏┓╋┏┓┏┓╋┏┓┏┏┓┏ # ┻┛┗┗┗ ┛ ┛┗┻┗┗ ┛ async def list_interfaces( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the physical connectivity interfaces available on the MikroTik device. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/interface", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_interface( self, dot_id: str, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To update an interface's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :param json_payload: The JSON to send in the body of the request. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url( path = f"/interface/{dot_id}", use_https = use_https ), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┳┏┓ ┏┓ ┓ ┓ ┳┓• ┓• # ┃┃┃ ┣┫┏┫┏┫┏┓┏┓┏┏ ┣┫┓┏┓┏┫┓┏┓┏┓┏ # ┻┣┛ ┛┗┗┻┗┻┛ ┗ ┛┛ ┻┛┗┛┗┗┻┗┛┗┗┫┛ # ┛ async def list_ip_address_bindings( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the IP Address bindings to Interfaces that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/ip/address", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_ip_address_binding( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new IP Address binding to an Interface to the MikroTik's configuration. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/ip/address", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_ip_address_binding( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing IP Address binding to an Interface in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/ip/address/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_ip_address_binding( self, dot_id: str, use_https: bool = None, ): """ To delete an existing IP Address to an Interface from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/ip/address/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┓┏┓ ┏┓┳┓ # ┃┃┃ ┣┫┃┃┏ # ┗┛┗┛┛┗┛┗┛ async def list_vlans( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the physical connectivity interfaces available on the MikroTik device. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/interface/vlan", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_vlan( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new VLAN to the MikroTik's configuration. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/interface/vlan", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_vlan( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing VLAN in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/interface/vlan/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_vlan( self, dot_id: str, use_https: bool = None, ): """ To delete an existing VLAN from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/interface/vlan/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┳┏┓ ┏┓ ┓ # ┃┃┃ ┃┃┏┓┏┓┃┏ # ┻┣┛ ┣┛┗┛┗┛┗┛ async def list_ip_pools( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the IP Pools that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/ip/pool", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_ip_pool( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new IP Pool to the MikroTik's configuration. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/ip/pool", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_ip_pool( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing IP Pool in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/ip/pool/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_ip_pool( self, dot_id: str, use_https: bool = None, ): """ To delete an existing IP Pool from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/ip/pool/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┏┓┏┓┏┓ ┏┓ ┏•┓ # ┃┃┃┃┃┃ ┃┃┏┓┏┓╋┓┃┏┓ # ┣┛┣┛┣┛ ┣┛┛ ┗┛┛┗┗┗ async def list_ppp_profiles( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the PPP Profiles that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/ppp/profile", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_ppp_profile( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new PPP Profile to the MikroTik's configuration. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/ppp/profile", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_ppp_profile( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing PPP Profile in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/ppp/profile/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_ppp_profile( self, dot_id: str, use_https: bool = None, ): """ To delete an existing PPP Profile from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/ppp/profile/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┏┓┏┓┏┓ ┏┓ # ┃┃┃┃┃┃ ┗┓┏┓┏┓┓┏┏┓┏┓ # ┣┛┣┛┣┛ ┗┛┗ ┛ ┗┛┗ ┛ async def list_ppp_servers( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the PPP Servers that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/interface/pppoe-server/server", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_ppp_server( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new PPP Server to the MikroTik's configuration. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/interface/pppoe-server/server", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_ppp_server( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing PPP Server in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/interface/pppoe-server/server/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_ppp_server( self, dot_id: str, use_https: bool = None, ): """ To delete an existing PPP Server from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/interface/pppoe-server/server/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┏┓┏┓┏┓ ┏┓┏┓┏┓ # ┃┃┃┃┃┃ ┣┫┣┫┣┫ # ┣┛┣┛┣┛ ┛┗┛┗┛┗ async def set_ppp_aaa( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To set the configuration for AAA in PPP. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.post( url = self.get_mikrotik_url(path = r"/ppp/aaa/set", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┓┏ ┏┓ ┏•┓ # ┣┫┏┓╋┏┏┓┏┓╋ ┃┃┏┓┏┓╋┓┃┏┓ # ┛┗┗┛┗┛┣┛┗┛┗ ┣┛┛ ┗┛┛┗┗┗ # ┛ async def list_hotspot_profiles( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the Hotspot Profiles that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/ip/hotspot/profile", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_hotspot_profile( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new Hotspot Profile to the MikroTik's configuration. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/ip/hotspot/profile", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_hotspot_profile( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing Hotspot Profile in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/ip/hotspot/profile/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_hotspot_profile( self, dot_id: str, use_https: bool = None, ): """ To delete an existing Hotspot Profile from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/ip/hotspot/profile/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┓┏ ┏┓ # ┣┫┏┓╋┏┏┓┏┓╋ ┗┓┏┓┏┓┓┏┏┓┏┓ # ┛┗┗┛┗┛┣┛┗┛┗ ┗┛┗ ┛ ┗┛┗ ┛ # ┛ async def list_hotspot_servers( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the Hotspot Servers that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/ip/hotspot", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_hotspot_server( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new Hotspot Server to the MikroTik's configuration. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/ip/hotspot", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_hotspot_server( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing Hotspot Server in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/ip/hotspot/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_hotspot_server( self, dot_id: str, use_https: bool = None, ): """ To delete an existing Hotspot Server from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/ip/hotspot/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┓┏ ┓ ┏ ┓┓ ┓ ┏┓ ┓ # ┣┫┏┓╋┏┏┓┏┓╋ ┃┃┃┏┓┃┃┏┓┏┫━━┃┓┏┓┏┓┏┫┏┓┏┓ # ┛┗┗┛┗┛┣┛┗┛┗ ┗┻┛┗┻┗┗┗ ┗┻ ┗┛┗┻┛ ┗┻┗ ┛┗ # ┛ async def list_hotspot_walled_gardens( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the Hotspot Walled-Gardens that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/ip/hotspot/walled-garden", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_hotspot_walled_garden( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new Hotspot Walled-Garden to the MikroTik's configuration. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/ip/hotspot/walled-garden", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_hotspot_walled_garden( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing Hotspot Walled-Garden in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/ip/hotspot/walled-garden/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_hotspot_walled_garden( self, dot_id: str, use_https: bool = None, ): """ To delete an existing Hotspot Walled-Garden from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/ip/hotspot/walled-garden/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┓┏ ┓ ┏ ┓┓ ┓ ┏┓ ┓ ┳┏┓ # ┣┫┏┓╋┏┏┓┏┓╋ ┃┃┃┏┓┃┃┏┓┏┫━━┃┓┏┓┏┓┏┫┏┓┏┓ ┃┃┃ # ┛┗┗┛┗┛┣┛┗┛┗ ┗┻┛┗┻┗┗┗ ┗┻ ┗┛┗┻┛ ┗┻┗ ┛┗ ┻┣┛ # ┛ async def list_hotspot_walled_garden_ips( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the Hotspot Walled-Garden IPs that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/ip/hotspot/walled-garden/ip", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_hotspot_walled_garden_ip( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new Hotspot Walled-Garden IP to the MikroTik's configuration. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/ip/hotspot/walled-garden/ip", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_hotspot_walled_garden_ip( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing Hotspot Walled-Garden IP in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/ip/hotspot/walled-garden/ip/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_hotspot_walled_garden_ip( self, dot_id: str, use_https: bool = None, ): """ To delete an existing Hotspot Walled-Garden IP from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/ip/hotspot/walled-garden/ip/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┳┓┏┓┳┓┳┳┳┏┓ # ┣┫┣┫┃┃┃┃┃┗┓ # ┛┗┛┗┻┛┻┗┛┗┛ async def set_radius_incoming( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To set the parameters for the way the MikroTik devices receive incoming hits from RADIUS. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.post( url = self.get_mikrotik_url(path = r"/radius/incoming/set", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def list_radius_servers( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the RADIUS Servers that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/radius", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_radius_server( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new RADIUS Server to the MikroTik's configuration. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/radius", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_radius_server( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing RADIUS Server in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/radius/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_radius_server( self, dot_id: str, use_https: bool = None, ): """ To delete an existing RADIUS Server from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/radius/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┳┓┓┏┏┓┏┓ ┏┓ # ┃┃┣┫┃ ┃┃ ┗┓┏┓┏┓┓┏┏┓┏┓ # ┻┛┛┗┗┛┣┛ ┗┛┗ ┛ ┗┛┗ ┛ async def list_dhcp_servers( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the DHCP Servers that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/ip/dhcp-server", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_dhcp_server( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new DHCP Server to the MikroTik's configuration. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/ip/dhcp-server", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_dhcp_server( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing DHCP Server in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/ip/dhcp-server/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_dhcp_server( self, dot_id: str, use_https: bool = None, ): """ To delete an existing DHCP Server from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/ip/dhcp-server/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┳┓┓┏┏┓┏┓ ┳┓ ┓ # ┃┃┣┫┃ ┃┃ ┃┃┏┓╋┓┏┏┏┓┏┓┃┏ # ┻┛┛┗┗┛┣┛ ┛┗┗ ┗┗┻┛┗┛┛ ┛┗ async def list_dhcp_networks( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the DHCP Networks that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/ip/dhcp-server/network", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_dhcp_network( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new DHCP Network to the MikroTik's configuration. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/ip/dhcp-server/network", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_dhcp_network( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing DHCP Network in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/ip/dhcp-server/network/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_dhcp_network( self, dot_id: str, use_https: bool = None, ): """ To delete an existing DHCP Network from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/ip/dhcp-server/network/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┏┓┳┓┳┳┓┏┓ ┏┓ • # ┗┓┃┃┃┃┃┃┃ ┃ ┏┓┏┳┓┏┳┓┓┏┏┓┓╋┓┏ # ┗┛┛┗┛ ┗┣┛ ┗┛┗┛┛┗┗┛┗┗┗┻┛┗┗┗┗┫ # ┛ async def list_snmp_communities( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the SNMP Communities that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/snmp/community", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_snmp_community( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new SNMP Community to the MikroTik's configuration. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/snmp/community", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_snmp_community( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing SNMP Community in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/snmp/community/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_snmp_community( self, dot_id: str, use_https: bool = None, ): """ To delete an existing SNMP Community from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/snmp/community/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┳┓┏┓┏┳┓ • # ┃┃┣┫ ┃ ━━┓┏┓┏┓ # ┛┗┛┗ ┻ ┗┛┗┗┫ # ┛ async def list_firewall_nat( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the Firewall NAT that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/ip/firewall/nat", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_firewall_nat( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new Firewall NAT to the MikroTik's configuration. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/ip/firewall/nat", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_firewall_nat( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing Firewall NAT in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/ip/firewall/nat/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_firewall_nat( self, dot_id: str, use_https: bool = None, ): """ To delete an existing Firewall NAT from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/ip/firewall/nat/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┳┓┳┓┏┓ # ┃┃┃┃┗┓ # ┻┛┛┗┗┛ async def set_dns( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To set the DNS parameters for the MikroTik. :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.post( url = self.get_mikrotik_url(path = r"/ip/dns/set", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": # Make an instance of an HTTP client to use to make API calls: my_http_client = httpx.AsyncClient( limits = httpx.Limits( max_connections = 100, # ............ Maximum number of connections allowed in the pool. max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive. ), timeout = httpx.Timeout( pool = 120.0, # .... Time to wait for a free connection from the pool. connect = 2.5, # ... Time to wait for establishing a connection to the server. write = 10.0, # .... Time to wait for sending data. read = 9.9 # ....... Time to wait for receiving data. ) ) # Create an instance of a Mongo conn. to write action logs: logs_mongo = AsyncMongo( connection_string = r"mongodb://del.ditscentre.in:27017,wtt.ditscentre.in:27017,mum.arh.001.ditscentre.in:27017/admin?tls=true&tlsCAFile=%2Fetc%2Fssl%2Fdbu%2Fmongo_data_ca.pem&tlsCertificateKeyFile=%2Fetc%2Fssl%2Fdbu%2Fmongo_data_cert.pem&replicaSet=dits_mongod_rep&readPreference=primary&authMechanism=MONGODB-X509&authSource=%24external", database_name = "converse", max_connections = 10, ) # Create an instance of a MikroTik client: my_mikrotik = AsyncMikroTik( http_client = my_http_client, mikrotik_ip = "102.210.173.150", username = "easyfi", password = "easyfi", use_https = False, config_by = "easyfi", action_log_conn = logs_mongo ) async def main(): await logs_mongo.connect() api_response = await my_mikrotik.list_system_resources() print("SYS RESOURCES:", json.to_string(api_response.data)) api_response = await my_mikrotik.list_interfaces() print(f"INTERFACES ({len(api_response.data)}):", json.to_string(api_response.data)) api_response = await my_mikrotik.list_vlans() print(f"VLANS ({len(api_response.data)}):", json.to_string(api_response.data)) api_response = await my_mikrotik.list_ip_pools() print(f"IP POOLS ({len(api_response.data)}):", json.to_string(api_response.data)) api_response = await my_mikrotik.list_ppp_profiles() print(f"PPP PROFILES ({len(api_response.data)}):", json.to_string(api_response.data)) api_response = await my_mikrotik.list_ppp_servers() print(f"PPP SERVERS ({len(api_response.data)}):", json.to_string(api_response.data)) api_response = await my_mikrotik.list_hotspot_profiles() print(f"HOTSPOT PROFILES ({len(api_response.data)}):", json.to_string(api_response.data)) api_response = await my_mikrotik.list_hotspot_servers() print(f"HOTSPOT SERVERS ({len(api_response.data)}):", json.to_string(api_response.data)) api_response = await my_mikrotik.list_hotspot_walled_gardens() print(f"HOTSPOT WALLED GARDENS ({len(api_response.data)}):", json.to_string(api_response.data)) api_response = await my_mikrotik.list_hotspot_walled_garden_ips() print(f"HOTSPOT WALLED GARDEN IPS ({len(api_response.data)}):", json.to_string(api_response.data)) api_response = await my_mikrotik.list_radius_servers() print(f"RADIUS SERVERS ({len(api_response.data)}):", json.to_string(api_response.data)) api_response = await my_mikrotik.list_firewall_nat() print(f"FIREWALL NAT ({len(api_response.data)}):") async def hotspot_steps(): HS_VLAN_ID = "2001" HS_INTERFACE_NAME = "easyfi-hs-if" HS_VLAN_NAME = "easyfi-hs-vlan-" + HS_VLAN_ID HS_IP_POOL_NAME = "easyfi-hs-pool-" + HS_VLAN_ID HS_PROFILE_NAME = "easyfi-hs-prf-" + HS_VLAN_ID HS_SERVER_NAME = "easyfi-hs-srv-" + HS_VLAN_ID HS_DHCP_SERVER_NAME = "easyfi-hs-dhcp-" + HS_VLAN_ID HS_SNMP_COMMUNITY_NAME = "lkjhgfdsa1234567" HS_PUBLIC_IP_SUBNET = "102.210.173.150/26" HS_FIRST_PUBLIC_IP = "102.210.173.128" HS_LAST_PUBLIC_IP = "102.210.173.191" HS_PRIVATE_IP_SUBNET = "100.65.0.0/22" HS_FIRST_PRIVATE_IP = "100.65.0.0" HS_LAST_PRIVATE_IP = "100.65.3.255" HS_GATEWAY_PRIVATE_IP = "100.65.0.1" HS_RADIUS_SERVER_IP = "154.84.218.42" HS_RADIUS_SERVER_SECRET = "asdfghjkl" HS_COMMENT = my_mikrotik.create_comment_json() # # STEP 1: # # Set up the interface: # api_response = await my_mikrotik.update_interface( # dot_id = "*3", # json_payload = { # "name": HS_INTERFACE_NAME, # "comment": HS_COMMENT # } # ) # print("STEP 1:", api_response.success, f"({api_response.action})") # print("MESSAGE:", api_response.message) # print("JSON:", api_response.data) # print("\n\n") # # STEP 2: # # Create the VLAN: # api_response = await my_mikrotik.add_vlan( # json_payload = { # "name": HS_VLAN_NAME, # "interface": HS_INTERFACE_NAME, # "vlan-id": HS_VLAN_ID, # "disabled": "false", # "comment": HS_COMMENT # } # ) # print("STEP 2:", api_response.success, f"({api_response.action})") # print("MESSAGE:", api_response.message) # print("JSON:", api_response.data) # print("\n\n") # # STEP 3: # # Bind a new IP address/network to the VLAN: # api_response = await my_mikrotik.add_ip_address_binding( # json_payload = { # "address": HS_PRIVATE_IP_SUBNET, # "interface": HS_VLAN_NAME, # "disabled": "false", # "comment": HS_COMMENT # } # ) # print("STEP 3:", api_response.success, f"({api_response.action})") # print("MESSAGE:", api_response.message) # print("JSON:", api_response.data) # print("\n\n") # # STEP 4: # # Add a new IP Pool for the Hotspot users: # api_response = await my_mikrotik.add_ip_pool( # json_payload = { # "name": HS_IP_POOL_NAME, # "ranges": HS_PRIVATE_IP_SUBNET, # "comment": HS_COMMENT # } # ) # print("STEP 4:", api_response.success, f"({api_response.action})") # print("MESSAGE:", api_response.message) # print("JSON:", api_response.data) # print("\n\n") # STEP 5: # Add a new Hotspot Profile: api_response = await my_mikrotik.add_hotspot_profile( json_payload = { "dns-name": "hs01.easyfi.net.in", "hotspot-address": HS_GATEWAY_PRIVATE_IP, "html-directory": "hotspot", "html-directory-override": "", "http-cookie-lifetime": "3d", "http-proxy": "0.0.0.0:0", "install-hotspot-queue": "false", "login-by": "cookie,http-chap", # "name": "hs01.easyfi.net.in", "name": HS_PROFILE_NAME, "split-user-domain": "false", "use-radius": "true", "nas-port-type": "wireless-802.11", "radius-accounting": "true", "radius-default-domain": "", "radius-interim-update": "received", "radius-location-id": "", "radius-location-name": "", "radius-mac-format": "XX:XX:XX:XX:XX:XX", # "comment": HS_COMMENT } ) print("STEP 5:", api_response.success, f"({api_response.action})") print("MESSAGE:", api_response.message) print("JSON:", api_response.data) print("\n\n") # STEP 6: # Add a new Hotspot Server: api_response = await my_mikrotik.add_hotspot_server( json_payload = { "address-pool": HS_IP_POOL_NAME, "addresses-per-mac": "2", "idle-timeout": "5m", "interface": HS_VLAN_NAME, "keepalive-timeout": "none", "login-timeout": "none", # "name": "hs01.easyfi.net.in", "name": HS_SERVER_NAME, # "profile": "hs01.easyfi.net.in", "profile": HS_PROFILE_NAME, "disabled": "false", # "comment": HS_COMMENT } ) print("STEP 6:", api_response.success, f"({api_response.action})") print("MESSAGE:", api_response.message) print("JSON:", api_response.data) print("\n\n") # # STEP 7: # # Add a new DHCP Server: # api_response = await my_mikrotik.add_dhcp_server( # json_payload = { # "address-pool": HS_IP_POOL_NAME, # "authoritative": "yes", # "disabled": "false", # "interface": HS_VLAN_NAME, # "lease-time": "30m", # "name": HS_DHCP_SERVER_NAME, # "use-radius": "no", # "comment": HS_COMMENT # } # ) # print("STEP 7:", api_response.success, f"({api_response.action})") # print("MESSAGE:", api_response.message) # print("JSON:", api_response.data) # print("\n\n") # STEP 8: # Add a new DHCP Network: api_response = await my_mikrotik.add_dhcp_network( json_payload = { "address": HS_PRIVATE_IP_SUBNET, "gateway": HS_GATEWAY_PRIVATE_IP, "netmask": HS_PRIVATE_IP_SUBNET.split("/")[-1], "dns-server": HS_GATEWAY_PRIVATE_IP, "comment": HS_COMMENT } ) print("STEP 8:", api_response.success, f"({api_response.action})") print("MESSAGE:", api_response.message) print("JSON:", api_response.data) print("\n\n") # # STEP 9: # # Set up the Walled-Garden IP: # api_response = await my_mikrotik.add_hotspot_walled_garden_ip( # json_payload = { # "action": "accept", # "dst-address": HS_RADIUS_SERVER_IP, # "comment": HS_COMMENT # } # ) # print("STEP 9:", api_response.success, f"({api_response.action})") # print("MESSAGE:", api_response.message) # print("JSON:", api_response.data) # print("\n\n") # # # STEP 10: # # Add a new RADIUS Server: # api_response = await my_mikrotik.add_radius_server( # json_payload = { # "accounting-port": "1813", # "address": HS_RADIUS_SERVER_IP, # "authentication-port": "1812", # "disabled": "false", # "protocol": "udp", # "secret": HS_RADIUS_SERVER_SECRET, # "service": "ppp,login,hotspot,dhcp", # "timeout": "300ms", # "comment": HS_COMMENT # } # ) # print("STEP 10:", api_response.success, f"({api_response.action})") # print("MESSAGE:", api_response.message) # print("JSON:", api_response.data) # print("\n\n") # # # STEP 11: # # Allow incoming traffic from RADIUS: # api_response = await my_mikrotik.set_radius_incoming( # json_payload = { # "accept": "yes", # "port": "3799", # # "comment": HS_COMMENT # } # ) # print("STEP 11:", api_response.success, f"({api_response.action})") # print("MESSAGE:", api_response.message) # print("JSON:", api_response.data) # print("\n\n") # # # STEP 12: # # Set up the DNS: # api_response = await my_mikrotik.set_dns( # json_payload = { # "allow-remote-requests": "true", # "servers": "8.8.8.8,8.8.4.4", # # "comment": HS_COMMENT # } # ) # print("STEP 12:", api_response.success, f"({api_response.action})") # print("MESSAGE:", api_response.message) # print("JSON:", api_response.data) # print("\n\n") # # STEP 13: # # Map the private IPs to the public IPs for NAT-ing: # private_ips = [ # str(ipaddress.IPv4Address(_)) # for _ in range( # int(ipaddress.IPv4Address(HS_FIRST_PRIVATE_IP)), # int(ipaddress.IPv4Address(HS_LAST_PRIVATE_IP)) + 1 # ) # ] # public_ips = [ # str(ipaddress.IPv4Address(_)) # for _ in range( # int(ipaddress.IPv4Address(HS_FIRST_PUBLIC_IP)), # int(ipaddress.IPv4Address(HS_LAST_PUBLIC_IP)) + 1 # ) # ] # nat_map = my_mikrotik.split_ipv4_range_in_powers_of_two( # start_ip = HS_FIRST_PRIVATE_IP, # end_ip = HS_LAST_PRIVATE_IP, # targets = public_ips, # consider_reserved_ips = False # ) # print("NAT MAP:", json.to_string(nat_map, default=str)) # print("PRIVATE IP COUNT:", len(private_ips)) # print("PUBLIC IP COUNT:", len(public_ips)) # print("RULE COUNT:", len(nat_map)) # for index, nat_rule in enumerate(nat_map): # api_response = await my_mikrotik.add_firewall_nat( # json_payload = { # "action": "src-nat", # "chain": "srcnat", # "disabled": "false", # "src-address": nat_rule["network"], # "to-addresses": nat_rule["target"], # "comment": HS_COMMENT # } # ) # print(f"STEP 13 ({index+1}/{len(nat_map)}):", api_response.success, f"({api_response.action})") # print("MESSAGE:", api_response.message) # print("JSON:", api_response.data) # print("\n\n") # # # STEP 14: # # Set up the SNMP Community: # api_response = await my_mikrotik.add_snmp_community( # json_payload = { # "addresses": "::/0", # "authentication-password": "", # "authentication-protocol": "MD5", # "disabled": "false", # "encryption-password": "", # "encryption-protocol": "DES", # "name": HS_SNMP_COMMUNITY_NAME, # "read-access": "true", # "security": "none", # "write-access": "false", # "comment": HS_COMMENT # } # ) # print("STEP 14:", api_response.success, f"({api_response.action})") # print("MESSAGE:", api_response.message) # print("JSON:", api_response.data) # print("\n\n") asyncio.run(main())