diff --git a/utils_v2/api/async_quart.py b/utils_v2/api/async_quart.py index 5ee2872..ff2b6d9 100644 --- a/utils_v2/api/async_quart.py +++ b/utils_v2/api/async_quart.py @@ -312,6 +312,8 @@ def describe_exception(exc): :return: The dict that explains the exception. """ + if exc is None: return None + exc_desc = { "type": type(exc).__name__, "msg": str(exc), diff --git a/utils_v2/mikrotik/__init__.py b/utils_v2/mikrotik/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils_v2/mikrotik/controllers/__init__.py b/utils_v2/mikrotik/controllers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils_v2/mikrotik/controllers/async_mikrotik.py b/utils_v2/mikrotik/controllers/async_mikrotik.py new file mode 100644 index 0000000..0dd4bf0 --- /dev/null +++ b/utils_v2/mikrotik/controllers/async_mikrotik.py @@ -0,0 +1,1078 @@ +""" + + 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, + # and add three (if asked) because we need IPs for network, gateway and broadcast: + ips_per_target = math.ceil(total_ips / target_count) + 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) -> None: + + """ + Generates a new random string to mark a new action chain and saves it in the local variable. + :return: None + """ + + 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 pf 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_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 pf 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 pf 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 pf 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 pf 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 pf 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 pf 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 pf 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 pf 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 + + +# ***************************************************************************************************************** +# ***** **** +# *** 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 = "bhopli", + 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)) + + asyncio.run(main()) diff --git a/utils_v2/mikrotik/models/__init__.py b/utils_v2/mikrotik/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils_v2/rest/controllers/async_base.py b/utils_v2/rest/controllers/async_base.py index 382ea67..7040d5d 100644 --- a/utils_v2/rest/controllers/async_base.py +++ b/utils_v2/rest/controllers/async_base.py @@ -39,6 +39,7 @@ sys.path.append("..") import io # The base model: +from utils_v2.rest.controllers.auth import RESTAuth from utils_v2.rest.models.api_call import ApiResponse # My utils: @@ -156,7 +157,8 @@ class AsyncREST: url: str, headers: dict = None, params: dict = None, - auth: httpx.Auth = None + auth: RESTAuth = None, + request_id: str | int = None ) -> ApiResponse: """ @@ -165,6 +167,7 @@ class AsyncREST: :param headers: The headers to pass. :param params: The params to send in the query string itself. :param auth: Any authentication credentials that need to be sent. + :param request_id: An id to give this request. Can be useful for debugging later. :return: A structured response that includes the raw response, the exception (if any), and so on. """ @@ -172,7 +175,10 @@ class AsyncREST: api_response = ApiResponse( action = inspect.stack()[1].function, url = url, - method = "GET" + method = "GET", + requestHeaders = headers, + requestParams = params, + requestId = request_id ) try: @@ -182,7 +188,7 @@ class AsyncREST: url = url, headers = headers, params = params, - auth = auth + auth = auth.encode() ) # Note down the results: @@ -209,7 +215,8 @@ class AsyncREST: json: dict = None, data: dict = None, content: str | bytes = None, - auth: httpx.Auth = None + auth: RESTAuth = None, + request_id: str | int = None ) -> ApiResponse: """ @@ -220,6 +227,7 @@ class AsyncREST: :param data: The params to send in the form-data in the body. :param content: The raw content to be sent in the body (typically as an octet-stream). :param auth: Any authentication credentials that need to be sent. + :param request_id: An id to give this request. Can be useful for debugging later. :return: A structured response that includes the raw response, the exception (if any), and so on. """ @@ -227,7 +235,12 @@ class AsyncREST: api_response = ApiResponse( action = inspect.stack()[1].function, url = url, - method = "POST" + method = "POST", + requestHeaders = headers, + requestJson = json, + requestData = data, + requestContent = content, + requestId = request_id ) try: @@ -239,7 +252,7 @@ class AsyncREST: json = json, data = data, content = content, - auth = auth + auth = auth.encode() ) # Note down the results: @@ -265,7 +278,8 @@ class AsyncREST: headers: dict = None, json: dict = None, data: dict = None, - auth: httpx.Auth = None + auth: RESTAuth = None, + request_id: str | int = None ) -> ApiResponse: """ @@ -275,6 +289,7 @@ class AsyncREST: :param json: The params to send in the JSON body. :param data: The params to send in the form-data in the body. :param auth: Any authentication credentials that need to be sent. + :param request_id: An id to give this request. Can be useful for debugging later. :return: A structured response that includes the raw response, the exception (if any), and so on. """ @@ -282,7 +297,11 @@ class AsyncREST: api_response = ApiResponse( action = inspect.stack()[1].function, url = url, - method = "PUT" + method = "PUT", + requestHeaders = headers, + requestJson = json, + requestData = data, + requestId = request_id ) try: @@ -293,7 +312,7 @@ class AsyncREST: headers = headers, json = json, data = data, - auth = auth + auth = auth.encode() ) # Note down the results: @@ -320,7 +339,8 @@ class AsyncREST: json: dict = None, data: dict = None, content: str | bytes = None, - auth: httpx.Auth = None + auth: RESTAuth = None, + request_id: str | int = None ) -> ApiResponse: """ @@ -331,6 +351,7 @@ class AsyncREST: :param data: The params to send in the form-data in the body. :param content: The raw content to be sent in the body (typically as an octet-stream). :param auth: Any authentication credentials that need to be sent. + :param request_id: An id to give this request. Can be useful for debugging later. :return: A structured response that includes the raw response, the exception (if any), and so on. """ @@ -338,7 +359,11 @@ class AsyncREST: api_response = ApiResponse( action = inspect.stack()[1].function, url = url, - method = "PATCH" + method = "PATCH", + requestHeaders = headers, + requestJson = json, + requestData = data, + requestId = request_id ) try: @@ -350,7 +375,7 @@ class AsyncREST: json = json, data = data, content = content, - auth = auth + auth = auth.encode() ) # Note down the results: @@ -374,7 +399,8 @@ class AsyncREST: self, url: str, headers: dict = None, - auth: httpx.Auth = None + auth: RESTAuth = None, + request_id: str | int = None ) -> ApiResponse: """ @@ -382,6 +408,7 @@ class AsyncREST: :param url: The URL to call. :param headers: The headers to pass. :param auth: Any authentication credentials that need to be sent. + :param request_id: An id to give this request. Can be useful for debugging later. :return: A structured response that includes the raw response, the exception (if any), and so on. """ @@ -389,7 +416,9 @@ class AsyncREST: api_response = ApiResponse( action = inspect.stack()[1].function, url = url, - method = "DELETE" + method = "DELETE", + requestHeaders = headers, + requestId = request_id ) try: @@ -398,7 +427,7 @@ class AsyncREST: response = await self._http_client.delete( url = url, headers = headers, - auth = auth + auth = auth.encode() ) # Note down the results: diff --git a/utils_v2/rest/controllers/auth.py b/utils_v2/rest/controllers/auth.py new file mode 100644 index 0000000..b60553d --- /dev/null +++ b/utils_v2/rest/controllers/auth.py @@ -0,0 +1,148 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Wednesday, 12th Feb., 2025. + + OBJECTIVE: + + To provide authentication abstraction for using with the Async REST class. + The idea is that if an input library changes, the rest of the code shouldn't change. + + 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 io + +# The base model: +from utils_v2.rest.models.api_call import ApiResponse + +# My utils: +from utils_v2.date_time import date_time + +# To make API calls: +import httpx + +# To work with date and time: +import datetime + +# For working with datatypes: +from typing import Literal, List + +# For creating abstract classes: +from abc import ABC, abstractmethod + +# For debugging: +from icecream import IceCreamDebugger +import inspect + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** CLASSES *** +# ***** **** +# ***************************************************************************************************************** + + +class RESTAuth(ABC): + + def __init__(self): + pass + + @staticmethod + def encode(self) -> httpx.Auth: + + """ + Use this method to give the desired output. For example, in Basic Auth, you convert the input username and + password to a Base64 string. If you are using, say, the HTTPX library, you would want to return its class from + this method. + :return: Whatever object is needed by the library you are using inside 'AsyncREST'. + """ + + pass + + +# --------------------------------------------------------------------------------------------------------------------- + + +class BasicAuth(RESTAuth): + + def __init__( + self, + username: str, + password: str + ): + + super().__init__() + self._username = username + self._password = password + + def encode(self) -> httpx.BasicAuth: + return httpx.BasicAuth(self._username, self._password) + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/utils_v2/rest/models/api_call.py b/utils_v2/rest/models/api_call.py index 954f30f..f12e03b 100644 --- a/utils_v2/rest/models/api_call.py +++ b/utils_v2/rest/models/api_call.py @@ -79,11 +79,19 @@ class ApiResponse(BaseModel): default = None ) + requestId: str | int | None = None + url: str = Field(frozen = True) method: str = Field(frozen = True) response: Any = None httpCode: int = None + requestHeaders: dict | None = None + requestParams: dict | None = None + requestJson: dict | None = None + requestData: Any | None = None + requestContent: Any | None = None + success: bool = False message: str = None data: Any = None