From 6b8b613f320124267b8d9006600865538bc55404 Mon Sep 17 00:00:00 2001 From: khushal Date: Tue, 11 Feb 2025 20:06:17 +0530 Subject: [PATCH] (20250211) MikroTik VLAN set up and rollback also ready. --- controllers_v2/software/mikrotik/base.py | 189 ++++++++- .../software/mikrotik/mikrotik_pppoe_1000.py | 395 +++++++++++++++--- readme/MikroTik-PPPoE-1000.md | 2 +- utils_v2/rest/controllers/async_base.py | 2 +- 4 files changed, 514 insertions(+), 74 deletions(-) diff --git a/controllers_v2/software/mikrotik/base.py b/controllers_v2/software/mikrotik/base.py index c022617..ce58416 100644 --- a/controllers_v2/software/mikrotik/base.py +++ b/controllers_v2/software/mikrotik/base.py @@ -361,7 +361,7 @@ class MikroTikController(CoreSoftwareController, ABC): password: str, port_no: int | str = None, use_https: bool = True - ): + ) -> ApiResponse: """ To enlist all the physical connectivity interfaces available on the MikroTik device. @@ -395,42 +395,29 @@ class MikroTikController(CoreSoftwareController, ABC): mikrotik_ip: str, username: str, password: str, - interface_id: str, + dot_id: str, + json_payload: dict, port_no: int | str = None, - use_https: bool = True, - name: str = None, - disabled: bool = False, - comment: str = None - ): + use_https: bool = True + ) -> ApiResponse: """ To update an interface's configuration. :param mikrotik_ip: The IP address of the MikroTik device. :param username: The username to get access to the MikroTik device. :param password: The password to get access to the MikroTik device. - :param interface_id: The id of the interface that you'd like to modify. + :param dot_id: The value of the '.id' field in the list. :param port_no: The port no. to hit the MikroTik device on. :param use_https: Whether to use HTTPS, or HTTP. - :param name: The new name for the interface. - :param disabled: Whether, or not, you'd like to disable the interface. - :param comment: A note that you'd like to attach to the entry. It'll then be available in the listing API. + :param json_payload: The JSON to send in the body pf the request. :return: A structured API response. """ - # Prepare the JSON to give to the API: - input_json = { - k: v for k, v in { - "name": name, - "disabled": disabled, - "comment": comment - }.items() if v is not None - } - # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url( mikrotik_ip = mikrotik_ip, - path = f"/interface/{interface_id}", + path = f"/interface/{dot_id}", port_no = port_no, use_https = use_https ), @@ -438,7 +425,165 @@ class MikroTikController(CoreSoftwareController, ABC): username = username, password = password ), - json = input_json + json = json_payload + ) + + # Done here: + return api_response + + # ┓┏┓ ┏┓┳┓ + # ┃┃┃ ┣┫┃┃┏ + # ┗┛┗┛┛┗┛┗┛ + + async def list_vlans( + self, + mikrotik_ip: str, + username: str, + password: str, + port_no: int | str = None, + use_https: bool = True + ) -> ApiResponse: + + """ + To enlist all the physical connectivity interfaces available on the MikroTik device. + :param mikrotik_ip: The IP address of the MikroTik device. + :param username: The username to get access to the MikroTik device. + :param password: The password to get access to the MikroTik device. + :param port_no: The port no. to hit the MikroTik device on. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.get( + url = self.get_mikrotik_url( + mikrotik_ip = mikrotik_ip, + path = r"/interface/vlan", + port_no = port_no, + use_https = use_https + ), + auth = httpx.BasicAuth( + username = username, + password = password + ) + ) + + # Done here: + return api_response + + async def add_vlan( + self, + mikrotik_ip: str, + username: str, + password: str, + json_payload: dict, + port_no: int | str = None, + use_https: bool = True + ) -> ApiResponse: + + """ + To add a new VLAN to the MikroTik's configuration. + :param mikrotik_ip: The IP address of the MikroTik device. + :param username: The username to get access to the MikroTik device. + :param password: The password to get access to the MikroTik device. + :param json_payload: The JSON to send in the body pf the request. + :param port_no: The port no. to hit the MikroTik device on. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.put( + url = self.get_mikrotik_url( + mikrotik_ip = mikrotik_ip, + path = r"/interface/vlan", + port_no = port_no, + use_https = use_https + ), + auth = httpx.BasicAuth( + username = username, + password = password + ), + json = json_payload + ) + + # Done here: + return api_response + + async def update_vlan( + self, + mikrotik_ip: str, + username: str, + password: str, + dot_id: str, + json_payload: dict, + port_no: int | str = None, + use_https: bool = True, + ) -> ApiResponse: + + """ + To update an existing VLAN in the MikroTik's configuration. + :param mikrotik_ip: The IP address of the MikroTik device. + :param username: The username to get access to the MikroTik device. + :param password: The password to get access to the MikroTik device. + :param dot_id: The value of the '.id' field in the list. + :param json_payload: The JSON to send in the body pf the request. + :param port_no: The port no. to hit the MikroTik device on. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.patch( + url = self.get_mikrotik_url( + mikrotik_ip = mikrotik_ip, + path = f"/interface/vlan/{dot_id}", + port_no = port_no, + use_https = use_https + ), + auth = httpx.BasicAuth( + username = username, + password = password + ), + json = json_payload + ) + + # Done here: + return api_response + + async def remove_vlan( + self, + mikrotik_ip: str, + username: str, + password: str, + dot_id: str, + port_no: int | str = None, + use_https: bool = True, + ): + + """ + To delete an existing VLAN from the MikroTik's configuration. + :param mikrotik_ip: The IP address of the MikroTik device. + :param username: The username to get access to the MikroTik device. + :param password: The password to get access to the MikroTik device. + :param dot_id: The value of the '.id' field in the list. + :param port_no: The port no. to hit the MikroTik device on. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.delete( + url = self.get_mikrotik_url( + mikrotik_ip = mikrotik_ip, + path = f"/interface/vlan/{dot_id}", + port_no = port_no, + use_https = use_https + ), + auth = httpx.BasicAuth( + username = username, + password = password + ) ) # Done here: diff --git a/controllers_v2/software/mikrotik/mikrotik_pppoe_1000.py b/controllers_v2/software/mikrotik/mikrotik_pppoe_1000.py index 4efe1c1..fc73e86 100644 --- a/controllers_v2/software/mikrotik/mikrotik_pppoe_1000.py +++ b/controllers_v2/software/mikrotik/mikrotik_pppoe_1000.py @@ -162,18 +162,45 @@ class MikroTikPPPoE1000Controller(MikroTikController): # Init a variable in a parent: self._client = self.CLIENT_NAME + # ┓┏ ┓ + # ┣┫┏┓┃┏┓┏┓┏┓┏ + # ┛┗┗ ┗┣┛┗ ┛ ┛ + # ┛ + + def created_by_easyfi( + self, + resource_json: dict = None + ) -> bool: + + """ + Checks if the resource was created/modified by this class. + :param resource_json; The JSON of the resource that needs to be checked. + :return: True if created by this class, else False. + """ + + # Extract values: + resource_name = resource_json.get("name") or "" + comment_json = self.parse_comment_json(resource_json.get("comment")) or {} + + # Perform checks: + if ( + comment_json.get("createdBy", "???") == self.CREATED_BY_NAME or + resource_name.lower().find("easyfi") >= 0 + ): return True + else: return False + # ┳ ┏ # ┃┏┓╋┏┓┏┓╋┏┓┏┏┓┏ # ┻┛┗┗┗ ┛ ┛┗┻┗┗ ┛ - async def interface_setup( + async def set_up_interface( self, mikrotik_auth: MikroTikPPPoE1000Auth, use_https: bool = True ) -> MikroTikConfigAttemptResponse: """ - To find the first available interface. + To find the first available interface and set it up for 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. @@ -217,16 +244,18 @@ class MikroTikPPPoE1000Controller(MikroTikController): mikrotik_ip = mikrotik_auth.nasIp, username = mikrotik_auth.username, password = mikrotik_auth.password, - interface_id = unused_if_json.get(".id"), + dot_id = unused_if_json.get(".id"), port_no = mikrotik_auth.nasPort, use_https = use_https, - name = self.HW_INTERFACE_NAME, - disabled = False, - comment = self.create_comment_json( - created_by = self.CREATED_BY_NAME, - created_ts = date_time.get_current_utc_date_time(as_string = False), - roll_back_config = unused_if_json - ) + json_payload = { + "name": self.HW_INTERFACE_NAME, + "disabled": "false", + "comment": self.create_comment_json( + created_by = self.CREATED_BY_NAME, + created_ts = date_time.get_current_utc_date_time(as_string = False), + roll_back_config = unused_if_json + ) + } ) # Check if the attempt was successful: @@ -242,7 +271,7 @@ class MikroTikPPPoE1000Controller(MikroTikController): # Done here: return step_response - async def interface_roll_back( + async def roll_back_interface( self, mikrotik_auth: MikroTikPPPoE1000Auth, use_https: bool = True @@ -296,12 +325,14 @@ class MikroTikPPPoE1000Controller(MikroTikController): mikrotik_ip = mikrotik_auth.nasIp, username = mikrotik_auth.username, password = mikrotik_auth.password, - interface_id = if_json.get(".id"), + dot_id = if_json.get(".id"), port_no = mikrotik_auth.nasPort, use_https = use_https, - name = original_config.get("name", self.HW_INTERFACE_NAME), - disabled = False, - comment = original_config.get("comment", "") + json_payload={ + "name": original_config.get("name", self.HW_INTERFACE_NAME), + "disabled": "false", + "comment": original_config.get("comment", "") + } ) total_count += 1 if api_response.success: rolled_back_count += 1 @@ -314,6 +345,275 @@ class MikroTikPPPoE1000Controller(MikroTikController): # Done here: return step_response + # ┓┏┓ ┏┓┳┓ + # ┃┃┃ ┣┫┃┃┏ + # ┗┛┗┛┛┗┛┗┛ + + async def set_up_one_vlan( + self, + mikrotik_auth: MikroTikPPPoE1000Auth, + vlan_id: int, + vlan_list: List[dict], + use_https: bool = True + ): + + """ + 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_auth: The set of credentials as received from the UI/API. + :param vlan_id: The id of the VLAN you want to create. + :param vlan_list: 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 you can find the VLAN id in the existing VLANs list: + existing_dot_id = None + existing_vlan_id = None + existing_vlan_json = None + for vlan_json in vlan_list: + if vlan_json.get("vlan-id") == str(vlan_id): + existing_dot_id = vlan_json.get(".id") + existing_vlan_id = vlan_json.get("vlan-id") + existing_vlan_json = vlan_json + break + + # If the VLAN id is unused: + if existing_dot_id is None: + + # We add the VLAN: + api_response = await self.add_vlan( + mikrotik_ip = mikrotik_auth.nasIp, + username = mikrotik_auth.username, + password = mikrotik_auth.password, + json_payload = { + "name": self.VLAN_NAME.format(vlan_id), + "vlan-id": str(vlan_id), + "interface": self.HW_INTERFACE_NAME, + "disabled": "false", + "comment": self.create_comment_json( + created_by = self.CREATED_BY_NAME, + created_ts = date_time.get_current_utc_date_time(as_string = False), + roll_back_config = existing_vlan_json + ) + }, + port_no = mikrotik_auth.nasPort, + 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 {vlan_id}." + step_response.exception = None + else: + step_response.success = False + step_response.message = f"Failed to add new VLAN with id {vlan_id}." + step_response.exception = api_response.exception + + # If the VLAN id is already used: + else: + + # We update it: + api_response = await self.update_vlan( + mikrotik_ip = mikrotik_auth.nasIp, + username = mikrotik_auth.username, + password = mikrotik_auth.password, + dot_id = existing_dot_id, + json_payload = { + "name": self.VLAN_NAME.format(vlan_id), + "interface": self.HW_INTERFACE_NAME, + "disabled": "false", + "comment": self.create_comment_json( + created_by = self.CREATED_BY_NAME, + created_ts = date_time.get_current_utc_date_time(as_string = False), + roll_back_config = existing_vlan_json + ) + }, + port_no = mikrotik_auth.nasPort, + use_https = use_https + ) + + # And assess the result: + if api_response.success: + step_response.success = True + step_response.message = f"Updated existing VLAN ({existing_vlan_id}) to id {vlan_id}." + step_response.exception = None + else: + step_response.success = False + step_response.message = f"Failed to update existing VLAN ({existing_vlan_id}) to id {vlan_id}." + step_response.exception = api_response.exception + + # Done here: + return step_response + + async def set_up_vlans( + self, + mikrotik_auth: MikroTikPPPoE1000Auth, + use_https: bool = True + ) -> ApiResponse: + + """ + To set up all the needed VLANs: + :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 self.list_vlans( + mikrotik_ip = mikrotik_auth.nasIp, + port_no = mikrotik_auth.nasPort, + username = mikrotik_auth.username, + password = mikrotik_auth.password, + 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: + vlan_list = await api_response.get_json() + + # Create a task for each VLAN that you need to set up, + # then fire them all asynchronously: + tasks = [ + self.set_up_one_vlan( + mikrotik_auth = mikrotik_auth, + vlan_id = vlan_id, + vlan_list = vlan_list, + use_https = use_https + ) for vlan_id in mikrotik_auth.vlanRange + ] + results = await asyncio.gather(*tasks) + + # Assess the results: + total_count = len(results) + success_count = 0 + all_messages = [] + for result in results: + if result.success: success_count += 1 + all_messages.append(result.message) + step_response.message = " -> ".join(all_messages) + step_response.success = True if success_count == total_count else False + + # Done here: + return step_response + + async def roll_back_vlans( + self, + mikrotik_auth: MikroTikPPPoE1000Auth, + use_https: bool = True + ) -> MikroTikRollBackAttemptResponse: + + """ + To reset VLANs to their original state (if they already existed) or remove them if this class added them. + :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: + total_count = 0 + success_count = 0 + all_messages = [] + step_response = MikroTikConfigAttemptResponse() + + # Enlist all the existing VLANs: + api_response = await self.list_vlans( + mikrotik_ip = mikrotik_auth.nasIp, + port_no = mikrotik_auth.nasPort, + username = mikrotik_auth.username, + password = mikrotik_auth.password, + 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 await api_response.get_json(): + + # Check if the VLAN was created by this class: + if not self.created_by_easyfi(vlan_json): continue + + # Extract the rollback information: + comment = self.parse_comment_json(vlan_json.get("comment")) or {} + roll_back_json = comment.get("rollbackConfig") + 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 self.remove_vlan( + mikrotik_ip = mikrotik_auth.nasIp, + username = mikrotik_auth.username, + password = mikrotik_auth.password, + dot_id = dot_id, + port_no = mikrotik_auth.nasPort, + 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 self.update_vlan( + mikrotik_ip = mikrotik_auth.nasIp, + username = mikrotik_auth.username, + password = mikrotik_auth.password, + 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": "" + }, + port_no = mikrotik_auth.nasPort, + 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: + all_messages.append(f"Removed/rolled-back {success_count}/{total_count} VLANs.") + if success_count == total_count: step_response.success = True + step_response.message = " -> ".join(all_messages) + + # Done here: + return step_response + # ┏┓ ┓ # ┣┫┓┏╋┣┓ # ┛┗┗┻┗┛┗ @@ -334,43 +634,25 @@ class MikroTikPPPoE1000Controller(MikroTikController): :return: A structured response to indicate what happened during authorization. """ - # print("IN-AUTH:", json.to_string(mikrotik_auth.model_dump(), default = str)) - print("VLAN SUBNETS:", json.to_string( - self.split_ipv4_range_among_targets( - start_ip = mikrotik_auth.firstPrivateIp, - end_ip = mikrotik_auth.lastPrivateIp, - targets = mikrotik_auth.vlanRange - ), - default = str - )) + # # print("IN-AUTH:", json.to_string(mikrotik_auth.model_dump(), default = str)) + # print("VLAN SUBNETS:", json.to_string( + # self.split_ipv4_range_among_targets( + # start_ip = mikrotik_auth.firstPrivateIp, + # end_ip = mikrotik_auth.lastPrivateIp, + # targets = mikrotik_auth.vlanRange + # ), + # default = str + # )) # Start with a blank response: auth_response = MikroTikAuthResponse() - # # Try connecting to the server to check if the credentials are valid, or not: - # api_response = await self.get_system_resource( - # mikrotik_ip = mikrotik_auth.nasIp, - # port_no = mikrotik_auth.nasPort, - # username = mikrotik_auth.username, - # password = mikrotik_auth.password, - # use_https = False - # ) - # - # # If the connection attempt failed: - # if not api_response.success: - # auth_response.exception = api_response.exception - # if api_response.httpCode is None: - # auth_response.message = "Exception: " + api_response.exception.__class__.__name__ - # if exception_str := str(auth_response.exception): auth_response.message += f" ({exception_str})" - # elif api_response.httpCode in [401]: auth_response.message = "Invalid credentials passed." - # elif api_response.httpCode in [502]: auth_response.message = "Could not ." - # else: auth_response.message = "Unknown error." + # Try to configure the system: + config_response = await self.configure(mikrotik_auth) + print("CONFIG RESPONSE:", config_response) + auth_response = config_response - # # Try to configure the system: - # config_response = await self.configure(mikrotik_auth) - # print("CONFIG RESPONSE:", config_response) - # - # await asyncio.sleep(5.0) + await asyncio.sleep(10.0) # Try to roll all configuration back: roll_back_response = await self.roll_back(mikrotik_auth) @@ -401,9 +683,15 @@ class MikroTikPPPoE1000Controller(MikroTikController): all_messages = [] roll_back_response = MikroTikConfigAttemptResponse() - # First, we arrange an interface: + # Next we remove all the VLANs: if keep_going: - step_response = await self.interface_roll_back(mikrotik_auth, use_https = False) + step_response = await self.roll_back_vlans(mikrotik_auth, 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_auth, use_https = False) all_messages.append(step_response.message) keep_going = step_response.success @@ -430,7 +718,14 @@ class MikroTikPPPoE1000Controller(MikroTikController): # First, we arrange an interface: if keep_going: - step_response = await self.interface_setup(mikrotik_auth, use_https = False) + step_response = await self.set_up_interface(mikrotik_auth, 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_auth, use_https = False) all_messages.append(step_response.message) keep_going = step_response.success diff --git a/readme/MikroTik-PPPoE-1000.md b/readme/MikroTik-PPPoE-1000.md index 3ed36fb..c24697c 100644 --- a/readme/MikroTik-PPPoE-1000.md +++ b/readme/MikroTik-PPPoE-1000.md @@ -127,7 +127,7 @@ Use `PUT` or `PATCH` method on the path `/interface/pppoe-server/server`. Consid **ROLL-BACK:** Save the original configuration as a JSON string in the `comment` field. If there was no original configuration, simply delete your record. -### 6. Set-up AAA to use RADIUS +### 6. Set up AAA to use RADIUS Now we need to tell the MikroTik device to use a RADIUS server for AAA. diff --git a/utils_v2/rest/controllers/async_base.py b/utils_v2/rest/controllers/async_base.py index 5304961..382ea67 100644 --- a/utils_v2/rest/controllers/async_base.py +++ b/utils_v2/rest/controllers/async_base.py @@ -338,7 +338,7 @@ class AsyncREST: api_response = ApiResponse( action = inspect.stack()[1].function, url = url, - method = "POST" + method = "PATCH" ) try: