diff --git a/api/blueprints/software/auth.py b/api/blueprints/software/auth.py index c1664f6..bac29ad 100644 --- a/api/blueprints/software/auth.py +++ b/api/blueprints/software/auth.py @@ -224,7 +224,16 @@ async def authorize_software_client( elif inbound_data.softwareClient == "mikrotikHotspot1000": - pass + # Make the client controller test and save the auth: + response = await current_app.mikrotik_hotspot_1000_controller.save_auth( + sql_conn = current_app.sql_writer, + mongo_data_conn = current_app.data_mongo, + mikrotik_auth = inbound_data.auth + ) + + # Note down the results: + success = response.success + message = f"Action Chain '{response.actionChain}': {response.message}" # ┳┓ # ┣┫┏┓┏┏┓┏┓┏┓┏┏┓ diff --git a/api/main.py b/api/main.py index cf4bc81..75b0106 100644 --- a/api/main.py +++ b/api/main.py @@ -100,6 +100,7 @@ from controllers_v2.finstitutions.payments.safaricom_mpesa_express import Safari # --- from controllers_v2.software.mikrotik.all_mikrotik import AllMikroTikController from controllers_v2.software.mikrotik.mikrotik_pppoe_1000 import MikroTikPPPoE1000Controller +from controllers_v2.software.mikrotik.mikrotik_hostpot_1000 import MikroTikHotspot1000Controller # To make REST API calls: import httpx @@ -573,6 +574,12 @@ async def app_startup(**kwargs): alert_url = current_app.script_data["alerts"]["url"], debug = enable_debugging ) + current_app.mikrotik_hotspot_1000_controller = MikroTikHotspot1000Controller( + cache = current_app.module_cache, + http_client = current_app.http_client, + alert_url = current_app.script_data["alerts"]["url"], + debug = enable_debugging + ) current_app.printer("Software/MikroTik (C) ready.") # ┏┓ ┓ ┏┓┓• diff --git a/controllers_v2/software/mikrotik/mikrotik_hostpot_1000.py b/controllers_v2/software/mikrotik/mikrotik_hostpot_1000.py index 1e086b0..f74ea1e 100644 --- a/controllers_v2/software/mikrotik/mikrotik_hostpot_1000.py +++ b/controllers_v2/software/mikrotik/mikrotik_hostpot_1000.py @@ -6,11 +6,11 @@ DATE: - Monday, 10th Feb., 2025. + Monday, 17th Feb., 2025. OBJECTIVE: - To handle configuration for MikroTik servers such that they work in PPPoE mode with support for 1,000 clients. + To handle configuration for MikroTik servers such that they work in Hotspot mode with support for 1,000 clients. REFERENCES: @@ -64,6 +64,9 @@ from typing import List, Any # To make HTTP requests: import httpx +# To work with IP addresses: +import ipaddress + # to work with MongoDB: from bson.objectid import ObjectId @@ -111,24 +114,25 @@ import asyncio # ***************************************************************************************************************** -class MikroTikPPPoE1000Controller(MikroTikController): +class MikroTikHotspot1000Controller(MikroTikController): # ┏┓┓ ┓┏ # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ - CLIENT_NAME = "mikrotikPPPoE1000" + CLIENT_NAME = "mikrotikHotspot1000" # 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. + CREATED_BY_NAME = NAME_PREFIX + "-hs-1000" + HW_INTERFACE_NAME = NAME_PREFIX + "-hs-if" + VLAN_NAME = NAME_PREFIX + "-vlan-{}" # ................. Substitute the VLAN's id here. + PRIVATE_IP_POOL_NAME = NAME_PREFIX + "-hs-pool-{}" # ... Substitute the VLAN's id here. + DHCP_SERVER_NAME = NAME_PREFIX + "-hs-dhcp-{}" # ....... Substitute the VLAN's id here. + HOTSPOT_PROFILE_NAME = NAME_PREFIX + "-hs-prf-{}" # .... Substitute the VLAN's id here. + HOTSPOT_SERVER_NAME = NAME_PREFIX + "-hs-srv-{}" # ..... Substitute the VLAN's id here. + RADIUS_SERVER_NAME = NAME_PREFIX + "-radius" # ......... Substitute the VLAN's id here. + NAT_RULE_NAME = NAME_PREFIX + "-hs-nat-{}" # ........... Substitute rule no. here. SNMP_COMMUNITY_NAME = NAME_PREFIX + "-snmp" # ┏┓ @@ -141,7 +145,7 @@ class MikroTikPPPoE1000Controller(MikroTikController): http_client: httpx.AsyncClient = None, alert_url: str = None, debug: bool = True, - debug_prefix: str = "MTik. PPPoE 1K (C) | ", + debug_prefix: str = "MTik. Hotspot 1K (C) | ", debug_only_errors: bool = True ): @@ -171,1115 +175,510 @@ class MikroTikPPPoE1000Controller(MikroTikController): # Init a variable in a parent: self._client = self.CLIENT_NAME - # ┳ ┏ - # ┃┏┓╋┏┓┏┓╋┏┓┏┏┓┏ - # ┻┛┗┗┗ ┛ ┛┗┻┗┗ ┛ + # ┏┓ • ┓ ┏┓ + # ┃┃┓┏┓┏┃┏ ┗┓┏┓╋┓┏┏┓ + # ┗┻┗┻┗┗┛┗ ┗┛┗ ┗┗┻┣┛ + # ┛ - async def set_up_interface( + async def quick_config( self, mikrotik_client: AsyncMikroTik, - dot_id: str = None, + mikrotik_auth: MikroTikHotspot1000Auth, use_https: bool = True ) -> MikroTikConfigAttemptResponse: """ - To find the first available interface and set it up for use. + To very quickly set up the MikroTik with just one VLAN. :param mikrotik_client: The client to use. - :param dot_id: If you already have an interface in mind, send its id here. + :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() + # Start by assuming values: + config_result = MikroTikConfigAttemptResponse() + all_success = True + all_messages = ["STARTING QUICK CONFIG."] - # Enlist all the interfaces: + # Prepare the needed variables: + HS_VLAN_ID = mikrotik_auth.vlanIds[0] + HS_INTERFACE_NAME = self.HW_INTERFACE_NAME + HS_VLAN_NAME = self.VLAN_NAME.format(HS_VLAN_ID) + HS_IP_POOL_NAME = self.PRIVATE_IP_POOL_NAME.format(HS_VLAN_ID) + HS_PROFILE_NAME = self.HOTSPOT_PROFILE_NAME.format(HS_VLAN_ID) + HS_SERVER_NAME = self.HOTSPOT_SERVER_NAME.format(HS_VLAN_ID) + HS_DHCP_SERVER_NAME = self.PRIVATE_IP_POOL_NAME.format(HS_VLAN_ID) + HS_SNMP_COMMUNITY_NAME = mikrotik_auth.snmpCommunity + # --- + HS_PUBLIC_IP_SUBNET = mikrotik_auth.publicIpPool + HS_FIRST_PUBLIC_IP = mikrotik_auth.firstPublicIp + HS_LAST_PUBLIC_IP = mikrotik_auth.lastPublicIp + # --- + HS_PRIVATE_IP_SUBNET = str(next(ipaddress.summarize_address_range( + ipaddress.IPv4Address(mikrotik_auth.firstPrivateIp), + ipaddress.IPv4Address(mikrotik_auth.lastPrivateIp) + ))) + HS_FIRST_PRIVATE_IP = mikrotik_auth.firstPrivateIp + HS_LAST_PRIVATE_IP = mikrotik_auth.lastPrivateIp + HS_GATEWAY_PRIVATE_IP = str(ipaddress.IPv4Address(mikrotik_auth.firstPrivateIp) + 1) + # --- + HS_RADIUS_SERVER_IP = mikrotik_auth.radius + HS_RADIUS_SERVER_SECRET = mikrotik_auth.secret + # --- + HS_COMMENT = mikrotik_client.create_comment_json() + + # STEP 1: + # Set up the interface: 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 + target_if_dot_id = None + for interface in api_response.data or []: + if interface["type"] == "ether": + target_if_dot_id = interface[".id"] 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"), + dot_id = target_if_dot_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" + "name": HS_INTERFACE_NAME, + "comment": HS_COMMENT }, use_https = use_https ) + all_messages.append(f"STEP 1 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False - # 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 + # STEP 2: + # Create the VLAN: + api_response = await mikrotik_client.add_vlan( + json_payload = { + "name": HS_VLAN_NAME, + "interface": HS_INTERFACE_NAME, + "vlan-id": HS_VLAN_ID, + "disabled": "false", + "comment": HS_COMMENT + } + ) + all_messages.append(f"STEP 2 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 3: + # Bind a new IP address/network to the VLAN: + api_response = await mikrotik_client.add_ip_address_binding( + json_payload = { + "address": HS_GATEWAY_PRIVATE_IP, + "interface": HS_VLAN_NAME, + "disabled": "false", + "comment": HS_COMMENT + } + ) + all_messages.append(f"STEP 3 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 4: + # Add a new IP Pool for the Hotspot users: + api_response = await mikrotik_client.add_ip_pool( + json_payload = { + "name": HS_IP_POOL_NAME, + "ranges": HS_PRIVATE_IP_SUBNET, + "comment": HS_COMMENT + } + ) + all_messages.append(f"STEP 4 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 5: + # Add a new Hotspot Profile: + api_response = await mikrotik_client.add_hotspot_profile( + json_payload = { + "dns-name": mikrotik_auth.domainName, + "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": 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 # ... NOT SUPPORTED! + } + ) + all_messages.append(f"STEP 5 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 6: + # Add a new Hotspot Server: + api_response = await mikrotik_client.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": HS_SERVER_NAME, + "profile": HS_PROFILE_NAME, + "disabled": "false", + # "comment": HS_COMMENT # ... NOT SUPPORTED! + } + ) + all_messages.append(f"STEP 6 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 7: + # Add a new DHCP Server: + api_response = await mikrotik_client.add_dhcp_server( + json_payload = { + "address-pool": HS_IP_POOL_NAME, + "authoritative": "yes", + "disabled": "false", + "interface": HS_VLAN_NAME, + "lease-time": "30m", + "name": HS_DHCP_SERVER_NAME, + "use-radius": "no", + "comment": HS_COMMENT + } + ) + all_messages.append(f"STEP 7 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 8: + # Add a new DHCP Network: + api_response = await mikrotik_client.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 + } + ) + all_messages.append(f"STEP 8 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 9: + # Set up the Walled-Garden IP: + api_response = await mikrotik_client.add_hotspot_walled_garden_ip( + json_payload = { + "action": "accept", + "dst-address": HS_RADIUS_SERVER_IP, + "comment": HS_COMMENT + } + ) + all_messages.append(f"STEP 9 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 10: + # Add a new RADIUS Server: + api_response = await mikrotik_client.add_radius_server( + json_payload = { + "accounting-port": "1813", + "address": HS_RADIUS_SERVER_IP, + "authentication-port": "1812", + "disabled": "false", + "protocol": "udp", + "secret": HS_RADIUS_SERVER_SECRET, + "service": "ppp,login,hotspot,dhcp", + "timeout": "300ms", + "comment": HS_COMMENT + } + ) + all_messages.append(f"STEP 10 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 11: + # Allow incoming traffic from RADIUS: + api_response = await mikrotik_client.set_radius_incoming( + json_payload = { + "accept": "yes", + "port": "3799", + # "comment": HS_COMMENT # ... NOT SUPPORTED! + } + ) + all_messages.append(f"STEP 11 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 12: + # Set up the DNS: + api_response = await mikrotik_client.set_dns( + json_payload = { + "allow-remote-requests": "true", + "servers": "8.8.8.8,8.8.4.4", + # "comment": HS_COMMENT # ... NOT SUPPORTED! + } + ) + all_messages.append(f"STEP 12 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # 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 = mikrotik_client.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 + ) + total_count = len(nat_map) + success_count = 0 + for index, nat_rule in enumerate(nat_map): + api_response = await mikrotik_client.add_firewall_nat( + json_payload = { + "action": "src-nat", + "chain": "srcnat", + "disabled": "false", + "src-address": nat_rule["network"], + "to-addresses": nat_rule["target"], + "comment": HS_COMMENT + } + ) + if api_response.success: success_count += 1 + all_messages.append(f"STEP 13 ({api_response.action}): {success_count}/{total_count}") + if not api_response.success: all_success = False + + # STEP 14: + # Set up the SNMP Community: + api_response = await mikrotik_client.add_snmp_community( + json_payload = { + "addresses": "::/0", + "authentication-password": "", + "authentication-protocol": "MD5", + "disabled": "false", + "encryption-password": "", + "encryption-protocol": "DES", + "name": HS_SNMP_COMMUNITY_NAME, + "read-access": "true", + "security": "none", + "write-access": "false", + "comment": HS_COMMENT + } + ) + all_messages.append(f"STEP 14 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False # Done here: - return step_response + config_result.message = " -> ".join(all_messages) + config_result.success = True if all_success else False + return config_result - async def add_radius_server( + async def quick_roll_back( self, mikrotik_client: AsyncMikroTik, - mikrotik_auth: MikroTikPPPoE1000Auth, use_https: bool = True - ): pass + ) -> MikroTikConfigAttemptResponse: + + """ + To very quickly roll back the setup done with the quick config method. + :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 values: + config_result = MikroTikConfigAttemptResponse() + all_success = True + all_messages = ["STARTING QUICK ROLL-BACK."] + + # STEP 1: + # Discard the SNMP community: + api_response = await mikrotik_client.list_snmp_communities(use_https = use_https) + if not api_response.success: all_success = False + resources = api_response.data or [] + total_count, success_count = 0, 0 + for resource in resources: + if mikrotik_client.is_my_config(resource, name_substring = self.NAME_PREFIX): + total_count += 1 + api_response = await mikrotik_client.remove_snmp_community(resource[".id"], use_https = use_https) + if api_response.success: success_count += 1 + else: all_success = False + all_messages.append(f"STEP 1 ({api_response.action}): {success_count}/{total_count}") + + # STEP 2: + # Discard all the NAT-ing rules: + api_response = await mikrotik_client.list_firewall_nat(use_https = use_https) + if not api_response.success: all_success = False + resources = api_response.data or [] + total_count, success_count = 0, 0 + for resource in resources: + print("NAT RES:", resource) + if mikrotik_client.is_my_config(resource, name_substring = self.NAME_PREFIX): + print("IS MINE!") + total_count += 1 + api_response = await mikrotik_client.remove_firewall_nat(resource[".id"], use_https = use_https) + if api_response.success: success_count += 1 + else: all_success = False + all_messages.append(f"STEP 2 ({api_response.action}): {success_count}/{total_count}") + + # STEP 3: + # Reset the DNS setting: + all_messages.append(f"STEP 3 (DNS): ??") + + # STEP 4: + # Reset the RADIUS incoming setting: + all_messages.append(f"STEP 4 (RADIUS Incoming): ??") + + # STEP 5: + # Discard all the RADIUS Servers: + api_response = await mikrotik_client.list_radius_servers(use_https = use_https) + if not api_response.success: all_success = False + resources = api_response.data or [] + total_count, success_count = 0, 0 + for resource in resources: + if mikrotik_client.is_my_config(resource, name_substring = self.NAME_PREFIX): + total_count += 1 + api_response = await mikrotik_client.remove_radius_server(resource[".id"], use_https = use_https) + if api_response.success: success_count += 1 + else: all_success = False + all_messages.append(f"STEP 5 ({api_response.action}): {success_count}/{total_count}") + + # STEP 6: + # Reset the Hotspot Walled Garden IP setting: + api_response = await mikrotik_client.list_hotspot_walled_garden_ips(use_https = use_https) + if not api_response.success: all_success = False + resources = api_response.data or [] + total_count, success_count = 0, 0 + for resource in resources: + if mikrotik_client.is_my_config(resource, name_substring = self.NAME_PREFIX): + total_count += 1 + api_response = await mikrotik_client.remove_hotspot_walled_garden_ip(resource[".id"], use_https = use_https) + if api_response.success: success_count += 1 + else: all_success = False + all_messages.append(f"STEP 6 ({api_response.action}): {success_count}/{total_count}") + + # STEP 7: + # Reset the DHCP Network: + all_messages.append(f"STEP 7 (DHCP Network): ??") + + # STEP 8: + # Discard all the DHCP Servers: + api_response = await mikrotik_client.list_dhcp_servers(use_https = use_https) + if not api_response.success: all_success = False + resources = api_response.data or [] + total_count, success_count = 0, 0 + for resource in resources: + if mikrotik_client.is_my_config(resource, name_substring = self.NAME_PREFIX): + total_count += 1 + api_response = await mikrotik_client.remove_dhcp_server(resource[".id"], use_https = use_https) + if api_response.success: success_count += 1 + else: all_success = False + all_messages.append(f"STEP 8 ({api_response.action}): {success_count}/{total_count}") + + # STEP 9: + # Discard all the Hotspot Servers: + api_response = await mikrotik_client.list_hotspot_servers(use_https = use_https) + if not api_response.success: all_success = False + resources = api_response.data or [] + total_count, success_count = 0, 0 + for resource in resources: + if mikrotik_client.is_my_config(resource, name_substring = self.NAME_PREFIX): + total_count += 1 + api_response = await mikrotik_client.remove_hotspot_server(resource[".id"], use_https = use_https) + if api_response.success: success_count += 1 + else: all_success = False + all_messages.append(f"STEP 9 ({api_response.action}): {success_count}/{total_count}") + + # STEP 10: + # Discard all the Hotspot Servers: + api_response = await mikrotik_client.list_hotspot_profiles(use_https = use_https) + if not api_response.success: all_success = False + resources = api_response.data or [] + total_count, success_count = 0, 0 + for resource in resources: + if mikrotik_client.is_my_config(resource, name_substring = self.NAME_PREFIX): + total_count += 1 + api_response = await mikrotik_client.remove_hotspot_profile(resource[".id"], use_https = use_https) + if api_response.success: success_count += 1 + else: all_success = False + all_messages.append(f"STEP 10 ({api_response.action}): {success_count}/{total_count}") + + # STEP 11: + # Discard all the IP Pools: + api_response = await mikrotik_client.list_ip_pools(use_https = use_https) + if not api_response.success: all_success = False + resources = api_response.data or [] + total_count, success_count = 0, 0 + for resource in resources: + if mikrotik_client.is_my_config(resource, name_substring = self.NAME_PREFIX): + total_count += 1 + api_response = await mikrotik_client.remove_ip_pool(resource[".id"], use_https = use_https) + if api_response.success: success_count += 1 + else: all_success = False + all_messages.append(f"STEP 11 ({api_response.action}): {success_count}/{total_count}") + + # STEP 12: + # Discard all the IP Addr. Bindings: + api_response = await mikrotik_client.list_ip_address_bindings(use_https = use_https) + if not api_response.success: all_success = False + resources = api_response.data or [] + total_count, success_count = 0, 0 + for resource in resources: + if mikrotik_client.is_my_config(resource, name_substring = self.NAME_PREFIX): + total_count += 1 + api_response = await mikrotik_client.remove_ip_address_binding(resource[".id"], use_https = use_https) + if api_response.success: success_count += 1 + else: all_success = False + all_messages.append(f"STEP 12 ({api_response.action}): {success_count}/{total_count}") + + # STEP 13: + # Discard all the VLANs: + api_response = await mikrotik_client.list_vlans(use_https = use_https) + if not api_response.success: all_success = False + resources = api_response.data or [] + total_count, success_count = 0, 0 + for resource in resources: + if mikrotik_client.is_my_config(resource, name_substring = self.NAME_PREFIX): + total_count += 1 + api_response = await mikrotik_client.remove_vlan(resource[".id"], use_https = use_https) + if api_response.success: success_count += 1 + else: all_success = False + all_messages.append(f"STEP 13 ({api_response.action}): {success_count}/{total_count}") + + # STEP 14: + # Reset the Interface that had been selected: + api_response = await mikrotik_client.list_interfaces(use_https = use_https) + if not api_response.success: all_success = False + resources = api_response.data or [] + total_count, success_count = 0, 0 + for resource in resources: + if mikrotik_client.is_my_config(resource, name_substring = self.NAME_PREFIX): + total_count += 1 + rb_json = mikrotik_client.parse_comment_json(resource.get("comment")) or {} + api_response = await mikrotik_client.update_interface( + dot_id = resource[".id"], + json_payload = { + "name": rb_json.get("name", resource["default-name"]), + "comment": "" + }, + use_https = use_https + ) + if api_response.success: success_count += 1 + else: all_success = False + all_messages.append(f"STEP 14 ({api_response.action}): {success_count}/{total_count}") + + # Done here: + config_result.message = " -> ".join(all_messages) + config_result.success = True if all_success else False + return config_result # ┏┓ ┓ # ┣┫┓┏╋┣┓ @@ -1289,7 +688,7 @@ class MikroTikPPPoE1000Controller(MikroTikController): self, sql_conn: AsyncMySQL, mongo_data_conn: AsyncMongo, - mikrotik_auth: MikroTikPPPoE1000Auth + mikrotik_auth: MikroTikHotspot1000Auth ) -> MikroTikAuthResponse: """ @@ -1311,26 +710,15 @@ class MikroTikPPPoE1000Controller(MikroTikController): username = mikrotik_auth.username, password = mikrotik_auth.password, port = mikrotik_auth.nasPort, - use_https = True, + use_https = False, 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 + # Try to configure the device: + config_response = await self.quick_config(mikrotik_client, mikrotik_auth, use_https = False) + config_response.actionChain = mikrotik_client.action_chain + return config_response # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ @@ -1340,7 +728,7 @@ class MikroTikPPPoE1000Controller(MikroTikController): async def roll_back( self, mikrotik_client: AsyncMikroTik, - mikrotik_auth: MikroTikPPPoE1000Auth + mikrotik_auth: MikroTikHotspot1000Auth ) -> MikroTikConfigAttemptResponse: """ @@ -1351,50 +739,12 @@ class MikroTikPPPoE1000Controller(MikroTikController): :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 + raise NotImplementedError async def configure( self, mikrotik_client: AsyncMikroTik, - mikrotik_auth: MikroTikPPPoE1000Auth + mikrotik_auth: MikroTikHotspot1000Auth ) -> MikroTikConfigAttemptResponse: """ @@ -1404,49 +754,7 @@ class MikroTikPPPoE1000Controller(MikroTikController): :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 + raise NotImplementedError # ***************************************************************************************************************** diff --git a/models/software/mikrotik/auth.py b/models/software/mikrotik/auth.py index 06e678f..64f25d6 100644 --- a/models/software/mikrotik/auth.py +++ b/models/software/mikrotik/auth.py @@ -298,7 +298,7 @@ class MikroTikHotspot1000Auth(BaseModel): frozen = True ) - vlanRange: str = Field( + vlanIds: List[int] = Field( description = "Don't know, and don't want to know.", min_length = 1, frozen = True @@ -330,6 +330,27 @@ class MikroTikHotspot1000Auth(BaseModel): class Config: extra = "forbid" + # ┏┓ • + # ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏ + # ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛ + # ┛ + + @property + def firstPrivateIp(self) -> str: + return self.privateIpPool.split("-")[0] + + @property + def lastPrivateIp(self) -> str: + return self.privateIpPool.split("-")[1] + + @property + def firstPublicIp(self) -> str: + return self.publicIpPool.split("-")[0] + + @property + def lastPublicIp(self) -> str: + return self.publicIpPool.split("-")[1] + # ┓┏ ┓• ┓ • # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ @@ -339,6 +360,14 @@ class MikroTikHotspot1000Auth(BaseModel): if isinstance(value, (str, float)): value = int(value) return value + @field_validator("privateIpPool", "publicIpPool", mode = "before") + def validate_ip_range(cls, value): + return ip.to_hyphen_notation(value) + + @field_validator("vlanIds", mode = "before") + def validate_vlans(cls, value): + return parse_vlans(value) + # --------------------------------------------------------------------------------------------------------------------- diff --git a/playground/switch/json.py b/playground/switch/json.py deleted file mode 100644 index 67d39f7..0000000 --- a/playground/switch/json.py +++ /dev/null @@ -1,202 +0,0 @@ -""" - - 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