From b7823f4c48db3551f13c3dd4908bd1ae9e7f12e3 Mon Sep 17 00:00:00 2001 From: khushal Date: Tue, 18 Feb 2025 15:17:23 +0530 Subject: [PATCH] (20250218) ... --- .../mikrotik/mikrotik_hostpot_1000.py | 1461 +++++++++++++++++ playground/switch/__init__.py | 0 playground/switch/json.py | 202 +++ playground/switch/switch_config.py | 105 ++ .../mikrotik/controllers/async_mikrotik.py | 220 +-- 5 files changed, 1888 insertions(+), 100 deletions(-) create mode 100644 controllers_v2/software/mikrotik/mikrotik_hostpot_1000.py create mode 100644 playground/switch/__init__.py create mode 100644 playground/switch/json.py create mode 100644 playground/switch/switch_config.py diff --git a/controllers_v2/software/mikrotik/mikrotik_hostpot_1000.py b/controllers_v2/software/mikrotik/mikrotik_hostpot_1000.py new file mode 100644 index 0000000..1e086b0 --- /dev/null +++ b/controllers_v2/software/mikrotik/mikrotik_hostpot_1000.py @@ -0,0 +1,1461 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Monday, 10th Feb., 2025. + + OBJECTIVE: + + To handle configuration for MikroTik servers such that they work in PPPoE mode with support for 1,000 clients. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# My async utils: +from utils_v2.string import json +from utils_v2.date_time import date_time +from utils_v2.mikrotik.controllers.async_mikrotik import AsyncMikroTik +from utils_v2.database.async_mysql_v2 import AsyncMySQL +from utils_v2.database.async_mongo_v2 import AsyncMongo +from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache + +# Controllers: +from controllers_v2.software.mikrotik.base import MikroTikController + +# To make very controlled API calls: +from utils_v2.rest.controllers.async_base import AsyncREST +from utils_v2.rest.models.api_call import ApiResponse + +# Models: +from models.software.mikrotik.auth import ( + MikroTikPPPoE1000Auth, + MikroTikHotspot1000Auth, + MikroTikAuthResponse +) +from models.software.mikrotik.configure import MikroTikConfigAttemptResponse + +# To work with datatypes: +from typing import List, Any + +# To make HTTP requests: +import httpx + +# to work with MongoDB: +from bson.objectid import ObjectId + +# To make abstract classes: +from abc import ABC, abstractmethod + +# For asynchronous activities: +import asyncio + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** CLASSES *** +# ***** **** +# ***************************************************************************************************************** + + +class MikroTikPPPoE1000Controller(MikroTikController): + + # ┏┓┓ ┓┏ + # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ + # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ + + CLIENT_NAME = "mikrotikPPPoE1000" + + # For automated configuration, and identification of automated configuration: + NAME_PREFIX = "easyfi" + CREATED_BY_NAME = NAME_PREFIX + "-pppoe-1000" + HW_INTERFACE_NAME = NAME_PREFIX + "-pppoe" + VLAN_NAME = NAME_PREFIX + "-vlan-{}" # .................... Substitute the VLAN's id here. + PRIVATE_IP_POOL_NAME = NAME_PREFIX + "-pppoe-pool-{}" # ... Substitute the VLAN's id here. + PPP_PROFILE_NAME = NAME_PREFIX + "-pppoe-prf-{}" # ........ Substitute the VLAN's id here. + PPP_SERVER_NAME = NAME_PREFIX + "-pppoe-srv-{}" # ......... Substitute the VLAN's id here. + RADIUS_SERVER_NAME = NAME_PREFIX + "-radius" # ............ Substitute the VLAN's id here. + NAT_RULE_NAME = NAME_PREFIX + "-pppoe-nat-{}" # ........... Substitute rule no. here. + SNMP_COMMUNITY_NAME = NAME_PREFIX + "-snmp" + + # ┏┓ + # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓ + # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛ + + def __init__( + self, + cache: AsyncRedisCache = None, + http_client: httpx.AsyncClient = None, + alert_url: str = None, + debug: bool = True, + debug_prefix: str = "MTik. PPPoE 1K (C) | ", + debug_only_errors: bool = True + ): + + """ + This is a special class that specifically handles configuration for MikroTik servers that work in PPPoE mode and + have support for upto 1,000 active clients. + :param cache: The object to use for caching results from database calls. + :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. + """ + + # Invoke the parent's constructor: + MikroTikController.__init__( + self, + cache = cache, + alert_url = alert_url, + http_client = http_client, + base_filter = {"client": self.CLIENT_NAME}, + debug = debug, + debug_prefix = debug_prefix, + debug_only_errors = debug_only_errors + ) + + # Init a variable in a parent: + self._client = self.CLIENT_NAME + + # ┳ ┏ + # ┃┏┓╋┏┓┏┓╋┏┓┏┏┓┏ + # ┻┛┗┗┗ ┛ ┛┗┻┗┗ ┛ + + async def set_up_interface( + self, + mikrotik_client: AsyncMikroTik, + dot_id: str = None, + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To find the first available interface and set it up for use. + :param mikrotik_client: The client to use. + :param dot_id: If you already have an interface in mind, send its id here. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + step_response = MikroTikConfigAttemptResponse() + + # Enlist all the interfaces: + api_response = await mikrotik_client.list_interfaces(use_https = use_https) + + # If the listing failed: + if not api_response.success: + step_response.success = False + step_response.message = "Failed to enlist h/w interfaces during configuration." + step_response.exception = step_response.exception + return step_response + + # We find/select an interface: + selected_if_json = None + for if_json in api_response.data: + if ( + str(if_json.get("type")).lower() == "ether" and + ( + ( + dot_id is not None and + str(if_json.get(".id")).lower() == str(dot_id) + ) or + ( + dot_id is None and + str(if_json.get("running")).lower() == "false" + ) + ) + ): + selected_if_json = if_json + break + + # If we found no available interface: + if selected_if_json is None: + step_response.success = False + step_response.message = "Failed to select a h/w interface." + step_response.exception = None + return step_response + + # Reconfigure the idle interface: + api_response = await mikrotik_client.update_interface( + dot_id = selected_if_json.get(".id"), + json_payload = { + "name": self.HW_INTERFACE_NAME, + "disabled": "false", + "comment": mikrotik_client.create_comment_json(roll_back_config = selected_if_json) + }, + use_https = use_https, + ) + + # Check if the attempt was successful: + if api_response.success: + step_response.success = True + step_response.message = f"H/w interface ('{selected_if_json.get('.id')}') configured." + step_response.exception = None + else: + step_response.success = False + step_response.message = f"H/w interface ('{selected_if_json.get('.id')}') found, but could NOT be configured." + step_response.exception = api_response.exception + + # Done here: + return step_response + + async def roll_back_interface( + self, + mikrotik_client: AsyncMikroTik, + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To find the first available interface. + :param mikrotik_client: The client to use. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + total_count = 0 + rolled_back_count = 0 + step_response = MikroTikConfigAttemptResponse() + + # Enlist all the interfaces: + api_response = await mikrotik_client.list_interfaces(use_https = use_https) + + # If the listing failed: + if not api_response.success: + step_response.success = False + step_response.message = "Failed to enlist h/w interfaces during rollback." + step_response.exception = step_response.exception + return step_response + + # Loop through all the interfaces, + # identify which ones were configured by this class, + # and perform rollback on those: + for if_json in api_response.data: + + # Extract the needed values: + comment_json = mikrotik_client.parse_comment_json(if_json.get("comment")) + original_config = mikrotik_client.get_original_config(comment_json) + + # If this was indeed set up by this class, + # and if rollback data is available: + if mikrotik_client.is_my_config( + config_json = if_json, + name_substring = self.NAME_PREFIX + ): + api_response = await mikrotik_client.update_interface( + dot_id = if_json.get(".id"), + json_payload = { + "name": original_config.get("name"), + "disabled": "false", + "comment": original_config.get("comment") + }, + use_https = use_https, + ) + total_count += 1 + if api_response.success: rolled_back_count += 1 + + # Construct the final response: + step_response.message = f"Rolled back {rolled_back_count}/{total_count} h/w interface configuration(s)." + if rolled_back_count == total_count: step_response.success = True + + # Done here: + return step_response + + # ┓┏┓ ┏┓┳┓ + # ┃┃┃ ┣┫┃┃┏ + # ┗┛┗┛┛┗┛┗┛ + + async def set_up_one_vlan( + self, + mikrotik_client: AsyncMikroTik, + new_resource: int, + existing_resources: List[dict], + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To set up just one of the needed VLANs. If the VLAN id is not taken, a new VLAN will be created, otherwise the + existing one will be updated. + :param mikrotik_client: The client to use. + :param new_resource: The id of the VLAN you want to create. + :param existing_resources: The list of existing VLANs already configured in the MikroTik device. Helps us decide between + the use of PUT and PATCH methods. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + step_response = MikroTikConfigAttemptResponse() + + # Check if the configuration already exists: + existing_dot_id = None + existing_resource = None + for _ in existing_resources: + if _.get("vlan-id", "???") == str(new_resource): + existing_dot_id = _.get(".id") + existing_resource = _ + break + + # If the VLAN id is unused: + if existing_dot_id is None: + + # We add the VLAN: + api_response = await mikrotik_client.add_vlan( + json_payload = { + "name": self.VLAN_NAME.format(new_resource), + "vlan-id": str(new_resource), + "interface": self.HW_INTERFACE_NAME, + "disabled": "false", + "comment": mikrotik_client.create_comment_json(roll_back_config = existing_resource) + }, + use_https = use_https + ) + + # And assess the result: + if api_response.success: + step_response.success = True + step_response.message = f"Added new VLAN with id {new_resource}." + step_response.exception = None + else: + step_response.success = False + step_response.message = f"Failed to add new VLAN with id {new_resource}." + step_response.exception = api_response.exception + + # If the VLAN id is already used: + else: + + # We update it: + api_response = await mikrotik_client.update_vlan( + dot_id = existing_dot_id, + json_payload = { + "name": self.VLAN_NAME.format(new_resource), + "interface": self.HW_INTERFACE_NAME, + "disabled": "false", + "comment": mikrotik_client.create_comment_json(roll_back_config = existing_resource) + }, + use_https = use_https + ) + + # And assess the result: + if api_response.success: + step_response.success = True + step_response.message = f"Updated existing VLAN ({existing_dot_id}) to id {new_resource}." + step_response.exception = None + else: + step_response.success = False + step_response.message = f"Failed to update existing VLAN ({existing_dot_id}) to id {new_resource}." + step_response.exception = api_response.exception + + # Done here: + return step_response + + async def set_up_vlans( + self, + mikrotik_client: AsyncMikroTik, + mikrotik_auth: MikroTikPPPoE1000Auth, + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To set up all the needed VLANs: + :param mikrotik_client: The client to use. + :param mikrotik_auth: The set of credentials as received from the UI/API. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + step_response = MikroTikConfigAttemptResponse() + + # Enlist all the existing VLANs: + api_response = await mikrotik_client.list_vlans(use_https = use_https) + + # If the listing failed: + if not api_response.success: + step_response.success = False + step_response.message = "Failed to enlist existing VLANs during configuration." + step_response.exception = step_response.exception + return step_response + + # Extract the list here: + existing_resources = api_response.data + + # Create a task for each VLAN that you need to set up, + # then fire them all asynchronously: + tasks = [ + self.set_up_one_vlan( + mikrotik_client = mikrotik_client, + new_resource = vlan_id, + existing_resources = existing_resources, + use_https = use_https + ) for vlan_id in mikrotik_auth.vlanIds + ] + results = await asyncio.gather(*tasks) + + # Assess the results: + total_count = len(results) + success_count = 0 + for result in results: + if result.success: success_count += 1 + step_response.message = f"{success_count}/{total_count} VLAN(s) configured." + step_response.success = True if success_count == total_count else False + + # Done here: + return step_response + + async def roll_back_vlans( + self, + mikrotik_client: AsyncMikroTik, + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To reset VLANs to their original state (if they already existed) or remove them if this class added them. + :param mikrotik_client: The client to use. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + total_count = 0 + success_count = 0 + all_messages = [] + step_response = MikroTikConfigAttemptResponse() + + # Enlist all the existing VLANs: + api_response = await mikrotik_client.list_vlans(use_https = use_https) + + # If the listing failed: + if not api_response.success: + step_response.success = False + step_response.message = "Failed to enlist existing VLANs during rollback." + step_response.exception = step_response.exception + return step_response + + # Loop through all the VLANs, + # and either reset them to their original state (if they already existed), + # or remove them entirely if they didn't exist before we came: + for vlan_json in api_response.data: + + # Check if the VLAN was created by this class: + if not mikrotik_client.is_my_config( + config_json = vlan_json, + name_substring = self.NAME_PREFIX + ): continue + + # Extract the rollback information: + comment_json = mikrotik_client.parse_comment_json(vlan_json.get("comment")) + roll_back_json = mikrotik_client.get_original_config(comment_json) + vlan_id = vlan_json.get("vlan-id") + dot_id = vlan_json.get(".id") + + # If the resource has no rollback information: + if roll_back_json is None: + + # We delete the resource: + await mikrotik_client.remove_vlan(dot_id = dot_id, use_https = use_https) + + # Assess the result: + total_count += 1 + if api_response.success: + all_messages.append(f"Removed VLAN with id {vlan_id}") + success_count += 1 + else: all_messages.append(f"Failed to remove VLAN with id {vlan_id}") + + # If the resource has rollback information: + else: + + # We roll back to the previous configuration: + await mikrotik_client.update_vlan( + dot_id = dot_id, + json_payload = { + "name": roll_back_json.get("name"), + "interface": roll_back_json.get("interface"), + "disabled": roll_back_json.get("disabled"), + "comment": "" + }, + use_https = use_https + ) + + # Assess the result: + total_count += 1 + if api_response.success: + all_messages.append(f"Rolled back VLAN with id {vlan_id}") + success_count += 1 + else: all_messages.append(f"Failed to roll back VLAN with id {vlan_id}") + + # Assess the overall results: + if success_count == total_count: step_response.success = True + step_response.message = f"Removed/rolled-back {success_count}/{total_count} VLAN(s)." + + # Done here: + return step_response + + # ┳┏┓ ┏┓ ┓ + # ┃┃┃ ┃┃┏┓┏┓┃┏ + # ┻┣┛ ┣┛┗┛┗┛┗┛ + + async def set_up_one_ip_pool( + self, + mikrotik_client: AsyncMikroTik, + new_resource: dict, + existing_resources: List[dict], + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To set up one of the needed IP pools. + :param mikrotik_client: The client to use. + :param new_resource: The IP pool you want configured by the end of the process. + :param existing_resources: A list of the existing IP pools. Helps to decide between 'PUT' and 'PATCH' methods. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + step_response = MikroTikConfigAttemptResponse() + + # Check if the configuration already exists: + existing_dot_id = None + existing_resource = None + for _ in existing_resources: + if _.get("name", "???") == new_resource["name"]: + existing_dot_id = _.get(".id") + existing_resource = _ + break + + # If the IP pool doesn't already exist: + if existing_dot_id is None: + + # Add the configuration: + api_response = await mikrotik_client.add_ip_pool( + json_payload = { + "name": new_resource["name"], + "ranges": new_resource["network"], + "comment": mikrotik_client.create_comment_json() + }, + use_https = use_https + ) + + # And assess the result: + if api_response.success: + step_response.success = True + step_response.message = f"Added new IP Pool '{new_resource['name']}'." + step_response.exception = None + else: + step_response.success = False + step_response.message = f"Failed to add new IP Pool '{new_resource['name']}'." + step_response.exception = api_response.exception + + # If the IP pool already exists: + else: + + # Update the configuration: + api_response = await mikrotik_client.update_ip_pool( + dot_id = existing_dot_id, + json_payload = { + "name": new_resource["name"], + "ranges": new_resource["network"], + "comment": mikrotik_client.create_comment_json(roll_back_config = existing_resource) + }, + use_https = use_https + ) + + # And assess the result: + if api_response.success: + step_response.success = True + step_response.message = f"Updated existing IP pool '{new_resource['name']}'." + step_response.exception = None + else: + step_response.success = False + step_response.message = f"Failed to update existing IP pool '{new_resource['name']}'." + step_response.exception = api_response.exception + + # Done here: + return step_response + + async def set_up_ip_pools( + self, + mikrotik_client: AsyncMikroTik, + mikrotik_auth: MikroTikPPPoE1000Auth, + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To set up ALL the needed IP pools. + :param mikrotik_client: The client to use. + :param mikrotik_auth: The set of credentials as received from the UI/API. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + step_response = MikroTikConfigAttemptResponse() + + # Enlist all the existing VLANs: + api_response = await mikrotik_client.list_ip_pools(use_https = use_https) + + # If the listing failed: + if not api_response.success: + step_response.success = False + step_response.message = "Failed to enlist existing IP Pools during configuration." + step_response.exception = step_response.exception + return step_response + + # Extract the list here: + existing_resources = api_response.data + + # First we split the whole private IP range + # into the no. of VLANs we had been asked to make: + new_resources = mikrotik_client.split_ipv4_range_equally( + start_ip = mikrotik_auth.firstPrivateIp, + end_ip = mikrotik_auth.lastPrivateIp, + targets = mikrotik_auth.vlanIds + ) + + # Create tasks to create IP Pools, + # and fire them asynchronously: + tasks = [ + self.set_up_one_ip_pool( + mikrotik_client = mikrotik_client, + new_resource = { + "name": self.PRIVATE_IP_POOL_NAME.format(new_resource["target"]), + "network": new_resource["network"], + }, + existing_resources = existing_resources, + use_https = use_https + ) for new_resource in new_resources + ] + results = await asyncio.gather(*tasks) + + # Assess the results: + total_count = len(results) + success_count = 0 + for result in results: + if result.success: success_count += 1 + step_response.message = f"{success_count}/{total_count} IP Pool(s) configured." + step_response.success = True if success_count == total_count else False + + # Done here: + return step_response + + async def roll_back_ip_pools( + self, + mikrotik_client: AsyncMikroTik, + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To reset IP pools to their original state (if they already existed) or remove them if this class added them. + :param mikrotik_client: The client to use. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + total_count = 0 + success_count = 0 + step_response = MikroTikConfigAttemptResponse() + + # Enlist all the existing VLANs: + api_response = await mikrotik_client.list_ip_pools(use_https = use_https) + + # If the listing failed: + if not api_response.success: + step_response.success = False + step_response.message = "Failed to enlist existing IP Pools during rollback." + step_response.exception = step_response.exception + return step_response + + # Loop through all the VLANs, + # and either reset them to their original state (if they already existed), + # or remove them entirely if they didn't exist before we came: + for resource_json in api_response.data: + + # Check if the VLAN was created by this class: + if not mikrotik_client.is_my_config( + config_json = resource_json, + name_substring = self.NAME_PREFIX + ): continue + + # Extract the rollback information: + comment_json = mikrotik_client.parse_comment_json(resource_json.get("comment")) + roll_back_json = mikrotik_client.get_original_config(comment_json) + dot_id = resource_json.get(".id") + + # If the resource has no rollback information: + if roll_back_json is None: + + # We delete the resource: + await mikrotik_client.remove_ip_pool(dot_id = dot_id, use_https = use_https) + + # Assess the result: + total_count += 1 + if api_response.success: success_count += 1 + + # If the resource has rollback information: + else: + + # We roll back to the previous configuration: + await mikrotik_client.update_ip_pool( + dot_id = dot_id, + json_payload = { + "name": roll_back_json.get("name"), + "ranges": roll_back_json.get("ranges"), + "comment": "" + }, + use_https = use_https + ) + + # Assess the result: + total_count += 1 + if api_response.success: success_count += 1 + + # Assess the overall results: + if success_count == total_count: step_response.success = True + step_response.message = f"Removed/rolled-back {success_count}/{total_count} IP Pool(s)." + + # Done here: + return step_response + + # ┏┓┏┓┏┓ ┏┓ ┏┓ ┏•┓ + # ┃┃┃┃┃┃┏┓┣ ┃┃┏┓┏┓╋┓┃┏┓┏ + # ┣┛┣┛┣┛┗┛┗┛ ┣┛┛ ┗┛┛┗┗┗ ┛ + + async def set_up_one_ppp_profile( + self, + mikrotik_client: AsyncMikroTik, + new_resource: dict, + existing_resources: List[dict], + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To set up one of the needed PPPoE Profiles. + :param mikrotik_client: The client to use. + :param new_resource: The PPPoE Profile you want configured by the end of the process. + :param existing_resources: A list of the existing PPPoE Profiles. Helps to decide between 'PUT' and 'PATCH' + methods. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + step_response = MikroTikConfigAttemptResponse() + + # Check if the configuration already exists: + existing_dot_id = None + existing_resource = None + for _ in existing_resources: + if _.get("name", "???") == new_resource["name"]: + existing_dot_id = _.get(".id") + existing_resource = _ + break + + # If the IP pool doesn't already exist: + if existing_dot_id is None: + + # Add the configuration: + api_response = await mikrotik_client.add_ppp_profile( + json_payload = { + "name": new_resource["name"], + "address-list": "", + "bridge-learning": "default", + "change-tcp-mss": "yes", + "dns-server": "8.8.8.8", + "local-address": new_resource["local-address"], + "only-one": "yes", + "remote-address": new_resource["remote-address"], + "use-compression": "default", + "use-encryption": "default", + "use-ipv6": "yes", + "use-mpls": "default", + "use-upnp": "default", + "comment": mikrotik_client.create_comment_json() + }, + use_https = use_https + ) + + # And assess the result: + if api_response.success: + step_response.success = True + step_response.message = f"Added new PPPoE Profile '{new_resource['name']}'." + step_response.exception = None + else: + step_response.success = False + step_response.message = f"Failed to add new PPPoE Profile '{new_resource['name']}'." + step_response.exception = api_response.exception + + # If the IP pool already exists: + else: + + # Update the configuration: + api_response = await mikrotik_client.update_ppp_profile( + dot_id = existing_dot_id, + json_payload = { + "name": new_resource["name"], + "address-list": "", + "bridge-learning": "default", + "change-tcp-mss": "yes", + "dns-server": "8.8.8.8", + "local-address": new_resource["local-address"], + "only-one": "yes", + "remote-address": new_resource["remote-address"], + "use-compression": "default", + "use-encryption": "default", + "use-ipv6": "yes", + "use-mpls": "default", + "use-upnp": "default", + "comment": mikrotik_client.create_comment_json(roll_back_config = existing_resource) + }, + use_https = use_https + ) + + # And assess the result: + if api_response.success: + step_response.success = True + step_response.message = f"Updated existing PPPoE Profile '{new_resource['name']}'." + step_response.exception = None + else: + step_response.success = False + step_response.message = f"Failed to update existing PPPoE Profile '{new_resource['name']}'." + step_response.exception = api_response.exception + + # Done here: + return step_response + + async def set_up_ppp_profiles( + self, + mikrotik_client: AsyncMikroTik, + mikrotik_auth: MikroTikPPPoE1000Auth, + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To set up ALL the needed PPPoE Profiles. + :param mikrotik_client: The client to use. + :param mikrotik_auth: The set of credentials as received from the UI/API. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + step_response = MikroTikConfigAttemptResponse() + + # Enlist all the existing VLANs: + api_response = await mikrotik_client.list_ppp_profiles(use_https = use_https) + + # If the listing failed: + if not api_response.success: + step_response.success = False + step_response.message = "Failed to enlist existing PPPoE Profiles during configuration." + step_response.exception = step_response.exception + return step_response + + # Extract the list here: + existing_resources = api_response.data + + # First we split the whole private IP range + # into the no. of VLANs we had been asked to make: + new_resources = mikrotik_client.split_ipv4_range_equally( + start_ip = mikrotik_auth.firstPrivateIp, + end_ip = mikrotik_auth.lastPrivateIp, + targets = mikrotik_auth.vlanIds + ) + + # Create tasks to create new PPPoE Profiles, + # and fire them asynchronously: + tasks = [ + self.set_up_one_ppp_profile( + mikrotik_client = mikrotik_client, + new_resource = { + "name": self.PPP_PROFILE_NAME.format(new_resource["target"]), + "local-address": mikrotik_auth.nasIp, + "remote-address": self.PRIVATE_IP_POOL_NAME.format(new_resource["target"]), + }, + existing_resources = existing_resources, + use_https = use_https + ) for new_resource in new_resources + ] + results = await asyncio.gather(*tasks) + + # Assess the results: + total_count = len(results) + success_count = 0 + for result in results: + if result.success: success_count += 1 + step_response.message = f"{success_count}/{total_count} PPPoE Profile(s) configured." + step_response.success = True if success_count == total_count else False + + # Done here: + return step_response + + async def roll_back_ppp_profiles( + self, + mikrotik_client: AsyncMikroTik, + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To reset PPPoE Profiles to their original state (if they already existed) or remove them if this class added + them. + :param mikrotik_client: The client to use. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + total_count = 0 + success_count = 0 + step_response = MikroTikConfigAttemptResponse() + + # Enlist all the existing VLANs: + api_response = await mikrotik_client.list_ppp_profiles(use_https = use_https) + + # If the listing failed: + if not api_response.success: + step_response.success = False + step_response.message = "Failed to enlist existing PPPoE Profiles during rollback." + step_response.exception = step_response.exception + return step_response + + # Loop through all the VLANs, + # and either reset them to their original state (if they already existed), + # or remove them entirely if they didn't exist before we came: + for resource_json in api_response.data: + + # Check if the VLAN was created by this class: + if not mikrotik_client.is_my_config( + config_json = resource_json, + name_substring = self.NAME_PREFIX + ): continue + + # Extract the rollback information: + dot_id = resource_json.get(".id") + + # We delete the resource: + api_response = await mikrotik_client.remove_ppp_profile(dot_id = dot_id, use_https = use_https) + + # Assess the result: + total_count += 1 + if api_response.success: success_count += 1 + + # Assess the overall results: + if success_count == total_count: step_response.success = True + step_response.message = f"Removed/rolled-back {success_count}/{total_count} PPPoE Profile(s)." + + # Done here: + return step_response + + # ┏┓┏┓┏┓ ┏┓ ┏┓ + # ┃┃┃┃┃┃┏┓┣ ┗┓┏┓┏┓┓┏┏┓┏┓┏ + # ┣┛┣┛┣┛┗┛┗┛ ┗┛┗ ┛ ┗┛┗ ┛ ┛ + + async def set_up_one_ppp_server( + self, + mikrotik_client: AsyncMikroTik, + new_resource: dict, + existing_resources: List[dict], + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To set up one of the needed PPPoE Servers. + :param mikrotik_client: The client to use. + :param new_resource: The PPPoE Server you want configured by the end of the process. + :param existing_resources: A list of the existing PPPoE Server. Helps to decide between 'PUT' and 'PATCH' + methods. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + step_response = MikroTikConfigAttemptResponse() + + # Check if the configuration already exists: + existing_dot_id = None + existing_resource = None + for _ in existing_resources: + if _.get("name", "???") == new_resource["name"]: + existing_dot_id = _.get(".id") + existing_resource = _ + break + + # If the IP pool doesn't already exist: + if existing_dot_id is None: + + # Add the configuration: + api_response = await mikrotik_client.add_ppp_server( + json_payload = { + "service-name": new_resource["service-name"], + "interface": new_resource["interface"], + "default-profile": new_resource["default-profile"], + "authentication": "pap,chap", + "disabled": "false", + "keepalive-timeout": "900", + "max-mru": "1500", + "max-mtu": "1500", + "max-sessions": "unlimited", + "mrru": "1500", + "one-session-per-host": "true", + "pado-delay": "0", + "comment": mikrotik_client.create_comment_json() + }, + use_https = use_https + ) + + # And assess the result: + if api_response.success: + step_response.success = True + step_response.message = f"Added new PPPoE Server '{new_resource['service-name']}'." + step_response.exception = None + else: + step_response.success = False + step_response.message = f"Failed to add new PPPoE Server '{new_resource['service-name']}'." + step_response.exception = api_response.exception + + # If the IP pool already exists: + else: + + # Update the configuration: + api_response = await mikrotik_client.update_ppp_server( + dot_id = existing_dot_id, + json_payload = { + "service-name": new_resource["service-name"], + "interface": new_resource["interface"], + "default-profile": new_resource["default-profile"], + "authentication": "pap,chap", + "disabled": "false", + "keepalive-timeout": "900", + "max-mru": "1500", + "max-mtu": "1500", + "max-sessions": "unlimited", + "mrru": "1500", + "one-session-per-host": "true", + "pado-delay": "0", + "comment": mikrotik_client.create_comment_json(roll_back_config = existing_resource) + }, + use_https = use_https + ) + + # And assess the result: + if api_response.success: + step_response.success = True + step_response.message = f"Updated existing PPPoE Server '{new_resource['service-name']}'." + step_response.exception = None + else: + step_response.success = False + step_response.message = f"Failed to update existing PPPoE Server '{new_resource['service-name']}'." + step_response.exception = api_response.exception + + # Done here: + return step_response + + async def set_up_ppp_servers( + self, + mikrotik_client: AsyncMikroTik, + mikrotik_auth: MikroTikPPPoE1000Auth, + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To set up ALL the needed PPPoE Servers. + :param mikrotik_client: The client to use. + :param mikrotik_auth: The set of credentials as received from the UI/API. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + step_response = MikroTikConfigAttemptResponse() + + # Enlist all the existing VLANs: + api_response = await mikrotik_client.list_ppp_servers(use_https = use_https) + + # If the listing failed: + if not api_response.success: + step_response.success = False + step_response.message = "Failed to enlist existing PPPoE Servers during configuration." + step_response.exception = step_response.exception + return step_response + + # Extract the list here: + existing_resources = api_response.data + + # First we split the whole private IP range + # into the no. of VLANs we had been asked to make: + new_resources = mikrotik_client.split_ipv4_range_equally( + start_ip = mikrotik_auth.firstPrivateIp, + end_ip = mikrotik_auth.lastPrivateIp, + targets = mikrotik_auth.vlanIds + ) + + # Create tasks to create new PPPoE Profiles, + # and fire them asynchronously: + tasks = [ + self.set_up_one_ppp_server( + mikrotik_client = mikrotik_client, + new_resource = { + "service-name": self.PPP_SERVER_NAME.format(new_resource["target"]), + "interface": self.VLAN_NAME.format(new_resource["target"]), + "default-profile": self.PPP_PROFILE_NAME.format(new_resource["target"]) + }, + existing_resources = existing_resources, + use_https = use_https + ) for new_resource in new_resources + ] + results = await asyncio.gather(*tasks) + + # Assess the results: + total_count = len(results) + success_count = 0 + for result in results: + if result.success: success_count += 1 + step_response.message = f"{success_count}/{total_count} PPPoE Server(s) configured." + step_response.success = True if success_count == total_count else False + + # Done here: + return step_response + + async def roll_back_ppp_servers( + self, + mikrotik_client: AsyncMikroTik, + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To reset PPPoE Servers to their original state (if they already existed) or remove them if this class added + them. + :param mikrotik_client: The client to use. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + total_count = 0 + success_count = 0 + step_response = MikroTikConfigAttemptResponse() + + # Enlist all the existing VLANs: + api_response = await mikrotik_client.list_ppp_servers(use_https = use_https) + + # If the listing failed: + if not api_response.success: + step_response.success = False + step_response.message = "Failed to enlist existing PPPoE Servers during rollback." + step_response.exception = step_response.exception + return step_response + + # Loop through all the VLANs, + # and either reset them to their original state (if they already existed), + # or remove them entirely if they didn't exist before we came: + for resource_json in api_response.data: + + # Check if the VLAN was created by this class: + if not mikrotik_client.is_my_config( + config_json = resource_json, + name_substring = self.NAME_PREFIX + ): continue + + # Extract the rollback information: + dot_id = resource_json.get(".id") + + # We delete the resource: + await mikrotik_client.remove_ppp_server(dot_id = dot_id, use_https = use_https) + + # Assess the result: + total_count += 1 + if api_response.success: success_count += 1 + + # Assess the overall results: + if success_count == total_count: step_response.success = True + step_response.message = f"Removed/rolled-back {success_count}/{total_count} PPPoE Server(s)." + + # Done here: + return step_response + + # ┳┓┏┓┳┓┳┳┳┏┓ + # ┣┫┣┫┃┃┃┃┃┗┓ + # ┛┗┛┗┻┛┻┗┛┗┛ + + async def enable_radius( + self, + mikrotik_client: AsyncMikroTik, + mikrotik_auth: MikroTikPPPoE1000Auth, + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To set up AAA such that it uses RADIUS. + :param mikrotik_client: The client to use. + :param mikrotik_auth: The set of credentials as received from the UI/API. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming failure: + step_response = MikroTikConfigAttemptResponse() + + # We hit the API: + api_response = await mikrotik_client.set_ppp_aaa( + json_payload = { + "accounting": "true", + "interim-update": "3m", + "use-circuit-id-in-nas-port-id": "false", + "use-radius": "true" + }, + use_https = use_https + ) + + # Assess the results: + if api_response.success: + step_response.success = True + step_response.message = f"Enabled RADIUS." + step_response.exception = None + else: + step_response.success = False + step_response.message = f"Failed to enable RADIUS." + step_response.exception = api_response.exception + + # Done here: + return step_response + + async def add_radius_server( + self, + mikrotik_client: AsyncMikroTik, + mikrotik_auth: MikroTikPPPoE1000Auth, + use_https: bool = True + ): pass + + # ┏┓ ┓ + # ┣┫┓┏╋┣┓ + # ┛┗┗┻┗┛┗ + + async def save_auth( + self, + sql_conn: AsyncMySQL, + mongo_data_conn: AsyncMongo, + mikrotik_auth: MikroTikPPPoE1000Auth + ) -> MikroTikAuthResponse: + + """ + Checks if a particular set of incoming credentials give access to a valid server and then stores the + credentials. + :param sql_conn: The database connection to use to perform this task. + :param mongo_data_conn: The database connection to use to perform this task. + :param mikrotik_auth: The set of credentials as received from the UI/API. + :return: A structured response to indicate what happened during authorization. + """ + + # Start with a blank response: + auth_response = MikroTikAuthResponse() + + # Create a MikroTik client: + mikrotik_client = AsyncMikroTik( + config_by = self.CREATED_BY_NAME, + mikrotik_ip = mikrotik_auth.nasIp, + username = mikrotik_auth.username, + password = mikrotik_auth.password, + port = mikrotik_auth.nasPort, + use_https = True, + http_client = None, + action_log_conn = mongo_data_conn + ) + + # Try to configure the system: + config_response = await self.configure(mikrotik_client, mikrotik_auth) + print("CONFIG RESPONSE:", config_response) + + print("Waiting...") + await asyncio.sleep(10.0) + + # Try to roll all configuration back: + roll_back_response = await self.roll_back(mikrotik_client, mikrotik_auth) + print("ROLL-BACK RESPONSE:", roll_back_response) + config_response += roll_back_response + + # Done here: + auth_response = config_response + return auth_response + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + async def roll_back( + self, + mikrotik_client: AsyncMikroTik, + mikrotik_auth: MikroTikPPPoE1000Auth + ) -> MikroTikConfigAttemptResponse: + + """ + The rolling-back to the original state (as best as possible) in case the configurations fails midway after + completing some no. of steps. + :param mikrotik_client: The client to use. + :param mikrotik_auth: The set of credentials as received from the UI/API. + :return: A structured response to indicate what happened during the configuration attempt. + """ + + # Start with some variables: + keep_going = True + all_messages = ["STARTING ROLLBACK."] + roll_back_response = MikroTikConfigAttemptResponse() + + # Next, we roll back all the PPPoE Servers: + if keep_going: + step_response = await self.roll_back_ppp_servers(mikrotik_client, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next, we roll back all the PPPoE Profiles: + if keep_going: + step_response = await self.roll_back_ppp_profiles(mikrotik_client, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next, we roll back all the IP pools: + if keep_going: + step_response = await self.roll_back_ip_pools(mikrotik_client, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next we roll back all the VLANs: + if keep_going: + step_response = await self.roll_back_vlans(mikrotik_client, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next, we free-up the interface: + if keep_going: + step_response = await self.roll_back_interface(mikrotik_client, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Done here: + roll_back_response.success = keep_going + roll_back_response.message = " -> ".join(all_messages) + return roll_back_response + + async def configure( + self, + mikrotik_client: AsyncMikroTik, + mikrotik_auth: MikroTikPPPoE1000Auth + ) -> MikroTikConfigAttemptResponse: + + """ + Run the configuration steps for the system. + :param mikrotik_client: The client to use. + :param mikrotik_auth: The set of credentials as received from the UI/API. + :return: A structured response to indicate what happened during the configuration attempt. + """ + + # Start with some variables: + keep_going = True + all_messages = ["STARTING CONFIG."] + config_response = MikroTikConfigAttemptResponse(actionChain = mikrotik_client.action_chain) + + # First, we arrange an interface: + if keep_going: + step_response = await self.set_up_interface(mikrotik_client, dot_id = "*3", use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next, we create the needed VLANs: + if keep_going: + await asyncio.sleep(0.25) + step_response = await self.set_up_vlans(mikrotik_client, mikrotik_auth, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next, we create the Private IP Pools: + if keep_going: + await asyncio.sleep(0.25) + step_response = await self.set_up_ip_pools(mikrotik_client, mikrotik_auth, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next, we create the PPPoE Profiles: + if keep_going: + await asyncio.sleep(0.25) + step_response = await self.set_up_ppp_profiles(mikrotik_client, mikrotik_auth, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next, we create the PPPoE Servers: + if keep_going: + await asyncio.sleep(0.25) + step_response = await self.set_up_ppp_servers(mikrotik_client, mikrotik_auth, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Done here: + config_response.success = keep_going + config_response.message = " -> ".join(all_messages) + return config_response + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/playground/switch/__init__.py b/playground/switch/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/playground/switch/json.py b/playground/switch/json.py new file mode 100644 index 0000000..67d39f7 --- /dev/null +++ b/playground/switch/json.py @@ -0,0 +1,202 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Create: Saturday, 18th May, 2022 + Update: Thursday, 22nd Aug. 2024 + + OBJECTIVE: + + To provide an easy way to work with '.json' data and files. + + REFERENCES: + + 1) https://www.w3schools.com/python/python_json.asp + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# System-level activities: +import io + +# To work with the JSON standard: +import json + +# To work with files: +from utils_v2.system import files + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +def from_string(json_data): + + """ + Decodes a JSON string to a pythonic variable like a dict. + :param json_data: The JSON string to decode. + :return: The decoded pythonic variable. + """ + + python_data = json.loads(json_data) + return python_data + + +# --------------------------------------------------------------------------------------------------------------------- + + +def to_string( + python_data, + indent = 4, + default = None, + separators = None, + no_space = False +): + + """ + Converts the given pythonic data to a JSON string. + :param python_data: The input data like a dict. + :param indent: The tab-width for pretty presentation. + :param default: The function to use on something that cannot be directly parsed into a JSON string. + :param separators: Custom separators to use. + :param no_space: If you want a dense JSON string that saves memory by not using spaces or tabs or line-breaks. Not + good for human readability, very good for saving memory. WARNING: THIS OVERRIDES EVERY OTHER PARAMETER EXCEPT + 'default'. + :return: The JSON string representation of the input pythonic data. + """ + + if no_space: + json_data = json.dumps( + python_data, + default = default, + separators = (',', ':') + ) + + else: + json_data = json.dumps( + python_data, + indent = indent, + default = default, + separators = separators + ) + + return json_data + + +# --------------------------------------------------------------------------------------------------------------------- + + +def from_file(file): + + """ + Reads a JSON file and returns it as a pythonic variable like a dict. + :param file: The path to the file on the disk or a file held in RAM as a BytesIO object. + :return: The decoded pythonic variable. + """ + + if isinstance(file, io.BytesIO): + file.seek(0) + json_data = file.getvalue() + else: json_data = files.read_file(file) + python_data = from_string(json_data) + return python_data + + +# --------------------------------------------------------------------------------------------------------------------- + + +def to_file( + file, + python_data, + indent = 4, + default = None, + separators = None, + no_space = False +): + + """ + + :param file: Either a path to a file on disk, or a buffer in RAM in the form of a BytesIO object. + :param python_data: The pythonic data to be converted to the JSON string. + :param indent: The tab-width for pretty presentation. + :param default: The function to use on something that cannot be directly parsed into a JSON string. + :param separators: Custom separators to use. + :param no_space: If you want a dense JSON string that saves memory by not using spaces or tabs or line-breaks. Not + good for human readability, very good for saving memory. WARNING: THIS OVERRIDES EVERY OTHER PARAMETER EXCEPT + 'default'. + :return: True/False if a path was given, else the same BytesIO object with the written JSON data. + """ + + json_data = to_string( + python_data, + indent = indent, + default = default, + separators = separators, + no_space = no_space + ) + + if isinstance(file, io.BytesIO): + file.write(json_data.encode("utf-8")) + file.seek(0) + return file + + else: + try: + files.write_file(file, json_data, mode = "w") + return True + except: return False + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/playground/switch/switch_config.py b/playground/switch/switch_config.py new file mode 100644 index 0000000..b2b6fd9 --- /dev/null +++ b/playground/switch/switch_config.py @@ -0,0 +1,105 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Saturday, 15th Feb., 2025. + + OBJECTIVE: + + ??? + + 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 +import os + +# For SSH: +from paramiko import SSHClient, AutoAddPolicy + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + ssh_client = SSHClient() + # ssh_client.load_host_keys("~/.ssh/known_hosts") + ssh_client.load_system_host_keys() + ssh_client.set_missing_host_key_policy(AutoAddPolicy()) + + ssh_client.connect( + hostname = "x.x.x.x", + username = "user", + password = "pass" + ) + + std_in, std_out, std_err = ssh_client.exec_command("hostname") + print("IN :", std_in.read().decode("utf8")) + print("OUT:", std_out.read().decode("utf8")) + print("ERR:", std_err.read().decode("utf8")) + print("RET. CODE:", std_out.channel.recv_exit_status()) + std_in.close() + std_out.close() + std_err.close() + + ssh_client.close() diff --git a/utils_v2/mikrotik/controllers/async_mikrotik.py b/utils_v2/mikrotik/controllers/async_mikrotik.py index ae008ca..44eb177 100644 --- a/utils_v2/mikrotik/controllers/async_mikrotik.py +++ b/utils_v2/mikrotik/controllers/async_mikrotik.py @@ -2314,9 +2314,24 @@ if __name__ == "__main__": 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" @@ -2324,6 +2339,8 @@ if __name__ == "__main__": 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" @@ -2354,7 +2371,7 @@ if __name__ == "__main__": # 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( @@ -2370,7 +2387,7 @@ if __name__ == "__main__": # 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( @@ -2385,7 +2402,7 @@ if __name__ == "__main__": # 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( @@ -2399,58 +2416,61 @@ if __name__ == "__main__": # 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", - # "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", - # "profile": "hs01.easyfi.net.in", - # "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 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( @@ -2469,23 +2489,23 @@ if __name__ == "__main__": # 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 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( @@ -2547,33 +2567,33 @@ if __name__ == "__main__": # 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)) + + # # 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 = { @@ -2613,4 +2633,4 @@ if __name__ == "__main__": # print("\n\n") - asyncio.run(hotspot_steps()) + asyncio.run(main())