diff --git a/controllers_v2/software/mikrotik/mikrotik_pppoe_1000.py b/controllers_v2/software/mikrotik/mikrotik_pppoe_1000.py index 5689952..1e086b0 100644 --- a/controllers_v2/software/mikrotik/mikrotik_pppoe_1000.py +++ b/controllers_v2/software/mikrotik/mikrotik_pppoe_1000.py @@ -322,8 +322,8 @@ class MikroTikPPPoE1000Controller(MikroTikController): async def set_up_one_vlan( self, mikrotik_client: AsyncMikroTik, - vlan_id: int, - vlan_list: List[dict], + new_resource: int, + existing_resources: List[dict], use_https: bool = True ) -> MikroTikConfigAttemptResponse: @@ -331,8 +331,8 @@ class MikroTikPPPoE1000Controller(MikroTikController): 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 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 + :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. @@ -341,15 +341,13 @@ class MikroTikPPPoE1000Controller(MikroTikController): # Start by assuming failure: step_response = MikroTikConfigAttemptResponse() - # Check if you can find the VLAN id in the existing VLANs list: + # Check if the configuration already exists: 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 + 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: @@ -358,11 +356,11 @@ class MikroTikPPPoE1000Controller(MikroTikController): # We add the VLAN: api_response = await mikrotik_client.add_vlan( json_payload = { - "name": self.VLAN_NAME.format(vlan_id), - "vlan-id": str(vlan_id), + "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_vlan_json) + "comment": mikrotik_client.create_comment_json(roll_back_config = existing_resource) }, use_https = use_https ) @@ -370,11 +368,11 @@ class MikroTikPPPoE1000Controller(MikroTikController): # 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.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 {vlan_id}." + 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: @@ -384,10 +382,10 @@ class MikroTikPPPoE1000Controller(MikroTikController): api_response = await mikrotik_client.update_vlan( dot_id = existing_dot_id, json_payload = { - "name": self.VLAN_NAME.format(vlan_id), + "name": self.VLAN_NAME.format(new_resource), "interface": self.HW_INTERFACE_NAME, "disabled": "false", - "comment": mikrotik_client.create_comment_json(roll_back_config = existing_vlan_json) + "comment": mikrotik_client.create_comment_json(roll_back_config = existing_resource) }, use_https = use_https ) @@ -395,11 +393,11 @@ class MikroTikPPPoE1000Controller(MikroTikController): # 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.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_vlan_id}) to id {vlan_id}." + step_response.message = f"Failed to update existing VLAN ({existing_dot_id}) to id {new_resource}." step_response.exception = api_response.exception # Done here: @@ -434,15 +432,15 @@ class MikroTikPPPoE1000Controller(MikroTikController): return step_response # Extract the list here: - vlan_list = api_response.data + 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, - vlan_id = vlan_id, - vlan_list = vlan_list, + new_resource = vlan_id, + existing_resources = existing_resources, use_https = use_https ) for vlan_id in mikrotik_auth.vlanIds ] @@ -554,16 +552,16 @@ class MikroTikPPPoE1000Controller(MikroTikController): async def set_up_one_ip_pool( self, mikrotik_client: AsyncMikroTik, - ip_pool: dict, - existing_ip_pools: List[dict], + 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 ip_pool: The IP pool you want configured by the end of the process. - :param existing_ip_pools: A list of the existing IP pools. Helps to decide between 'PUT' and 'PATCH' methods. + :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. """ @@ -573,11 +571,11 @@ class MikroTikPPPoE1000Controller(MikroTikController): # Check if the configuration already exists: existing_dot_id = None - existing_ip_pool_json = None - for existing_ip_pool in existing_ip_pools: - if existing_ip_pool.get("name", "???") == ip_pool["name"]: - existing_dot_id = existing_ip_pool[".id"] - existing_ip_pool_json = existing_ip_pool + 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: @@ -586,8 +584,8 @@ class MikroTikPPPoE1000Controller(MikroTikController): # Add the configuration: api_response = await mikrotik_client.add_ip_pool( json_payload = { - "name": ip_pool["name"], - "ranges": ip_pool["network"], + "name": new_resource["name"], + "ranges": new_resource["network"], "comment": mikrotik_client.create_comment_json() }, use_https = use_https @@ -596,11 +594,11 @@ class MikroTikPPPoE1000Controller(MikroTikController): # And assess the result: if api_response.success: step_response.success = True - step_response.message = f"Added new IP pool '{ip_pool['name']}'." + 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 '{ip_pool['name']}'." + 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: @@ -610,9 +608,9 @@ class MikroTikPPPoE1000Controller(MikroTikController): api_response = await mikrotik_client.update_ip_pool( dot_id = existing_dot_id, json_payload = { - "name": ip_pool["name"], - "ranges": ip_pool["network"], - "comment": mikrotik_client.create_comment_json(roll_back_config = existing_ip_pool_json) + "name": new_resource["name"], + "ranges": new_resource["network"], + "comment": mikrotik_client.create_comment_json(roll_back_config = existing_resource) }, use_https = use_https ) @@ -620,11 +618,11 @@ class MikroTikPPPoE1000Controller(MikroTikController): # And assess the result: if api_response.success: step_response.success = True - step_response.message = f"Updated existing IP pool '{ip_pool['name']}'." + 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 '{ip_pool['name']}'." + step_response.message = f"Failed to update existing IP pool '{new_resource['name']}'." step_response.exception = api_response.exception # Done here: @@ -654,16 +652,16 @@ class MikroTikPPPoE1000Controller(MikroTikController): # 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.message = "Failed to enlist existing IP Pools during configuration." step_response.exception = step_response.exception return step_response # Extract the list here: - existing_ip_pools = api_response.data + existing_resources = api_response.data # First we split the whole private IP range # into the no. of VLANs we had been asked to make: - ip_pools = mikrotik_client.split_ipv4_range_equally( + new_resources = mikrotik_client.split_ipv4_range_equally( start_ip = mikrotik_auth.firstPrivateIp, end_ip = mikrotik_auth.lastPrivateIp, targets = mikrotik_auth.vlanIds @@ -674,13 +672,13 @@ class MikroTikPPPoE1000Controller(MikroTikController): tasks = [ self.set_up_one_ip_pool( mikrotik_client = mikrotik_client, - ip_pool = { - "name": self.PRIVATE_IP_POOL_NAME.format(ip_pool["target"]), - "network": ip_pool["network"], + new_resource = { + "name": self.PRIVATE_IP_POOL_NAME.format(new_resource["target"]), + "network": new_resource["network"], }, - existing_ip_pools = existing_ip_pools, + existing_resources = existing_resources, use_https = use_https - ) for ip_pool in ip_pools + ) for new_resource in new_resources ] results = await asyncio.gather(*tasks) @@ -689,7 +687,7 @@ class MikroTikPPPoE1000Controller(MikroTikController): 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.message = f"{success_count}/{total_count} IP Pool(s) configured." step_response.success = True if success_count == total_count else False # Done here: @@ -719,7 +717,7 @@ class MikroTikPPPoE1000Controller(MikroTikController): # 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.message = "Failed to enlist existing IP Pools during rollback." step_response.exception = step_response.exception return step_response @@ -769,11 +767,520 @@ class MikroTikPPPoE1000Controller(MikroTikController): # 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)." + 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 + # ┏┓ ┓ # ┣┫┓┏╋┣┓ # ┛┗┗┻┗┛┗ @@ -813,7 +1320,8 @@ class MikroTikPPPoE1000Controller(MikroTikController): config_response = await self.configure(mikrotik_client, mikrotik_auth) print("CONFIG RESPONSE:", config_response) - await asyncio.sleep(2.0) + print("Waiting...") + await asyncio.sleep(10.0) # Try to roll all configuration back: roll_back_response = await self.roll_back(mikrotik_client, mikrotik_auth) @@ -848,6 +1356,18 @@ class MikroTikPPPoE1000Controller(MikroTikController): 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) @@ -909,6 +1429,20 @@ class MikroTikPPPoE1000Controller(MikroTikController): 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) diff --git a/utils_v2/mikrotik/controllers/async_mikrotik.py b/utils_v2/mikrotik/controllers/async_mikrotik.py index 0dd4bf0..3e666cb 100644 --- a/utils_v2/mikrotik/controllers/async_mikrotik.py +++ b/utils_v2/mikrotik/controllers/async_mikrotik.py @@ -533,7 +533,7 @@ class AsyncMikroTik: To update an interface's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. - :param json_payload: The JSON to send in the body pf the request. + :param json_payload: The JSON to send in the body of the request. :return: A structured API response. """ @@ -555,6 +555,121 @@ class AsyncMikroTik: await self.log_action(api_response) return api_response + # ┳┏┓ ┏┓ ┓ ┓ ┳┓• ┓• + # ┃┃┃ ┣┫┏┫┏┫┏┓┏┓┏┏ ┣┫┓┏┓┏┫┓┏┓┏┓┏ + # ┻┣┛ ┛┗┗┻┗┻┛ ┗ ┛┛ ┻┛┗┛┗┗┻┗┛┗┗┫┛ + # ┛ + + async def list_ip_address_bindings( + self, + use_https: bool = None + ) -> ApiResponse: + + """ + To enlist all the IP Address bindings to Interfaces that have been configured. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.get( + url = self.get_mikrotik_url(path = r"/ip/address", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def add_ip_address_binding( + self, + json_payload: dict, + use_https: bool = None + ) -> ApiResponse: + + """ + To add a new IP Address binding to an Interface to the MikroTik's configuration. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.put( + url = self.get_mikrotik_url(path = r"/ip/address", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def update_ip_address_binding( + self, + dot_id: str, + json_payload: dict, + use_https: bool = None, + ) -> ApiResponse: + + """ + To update an existing IP Address binding to an Interface in the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.patch( + url = self.get_mikrotik_url(path = f"/ip/address/{dot_id}", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def remove_ip_address_binding( + self, + dot_id: str, + use_https: bool = None, + ): + + """ + To delete an existing IP Address to an Interface from the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.delete( + url = self.get_mikrotik_url(path = f"/ip/address/{dot_id}", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + # ┓┏┓ ┏┓┳┓ # ┃┃┃ ┣┫┃┃┏ # ┗┛┗┛┛┗┛┗┛ @@ -592,7 +707,7 @@ class AsyncMikroTik: """ To add a new VLAN to the MikroTik's configuration. - :param json_payload: The JSON to send in the body pf the request. + :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ @@ -622,7 +737,7 @@ class AsyncMikroTik: """ To update an existing VLAN in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. - :param json_payload: The JSON to send in the body pf the request. + :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ @@ -706,7 +821,7 @@ class AsyncMikroTik: """ To add a new IP Pool to the MikroTik's configuration. - :param json_payload: The JSON to send in the body pf the request. + :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ @@ -736,7 +851,7 @@ class AsyncMikroTik: """ To update an existing IP Pool in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. - :param json_payload: The JSON to send in the body pf the request. + :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ @@ -820,7 +935,7 @@ class AsyncMikroTik: """ To add a new PPP Profile to the MikroTik's configuration. - :param json_payload: The JSON to send in the body pf the request. + :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ @@ -850,7 +965,7 @@ class AsyncMikroTik: """ To update an existing PPP Profile in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. - :param json_payload: The JSON to send in the body pf the request. + :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ @@ -934,7 +1049,7 @@ class AsyncMikroTik: """ To add a new PPP Server to the MikroTik's configuration. - :param json_payload: The JSON to send in the body pf the request. + :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ @@ -964,7 +1079,7 @@ class AsyncMikroTik: """ To update an existing PPP Server in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. - :param json_payload: The JSON to send in the body pf the request. + :param json_payload: The JSON to send in the body of the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ @@ -1011,6 +1126,1130 @@ class AsyncMikroTik: await self.log_action(api_response) return api_response + # ┏┓┏┓┏┓ ┏┓┏┓┏┓ + # ┃┃┃┃┃┃ ┣┫┣┫┣┫ + # ┣┛┣┛┣┛ ┛┗┛┗┛┗ + + async def set_ppp_aaa( + self, + json_payload: dict, + use_https: bool = None + ) -> ApiResponse: + + """ + To set the configuration for AAA in PPP. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.post( + url = self.get_mikrotik_url(path = r"/ppp/aaa/set", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + # ┓┏ ┏┓ ┏•┓ + # ┣┫┏┓╋┏┏┓┏┓╋ ┃┃┏┓┏┓╋┓┃┏┓ + # ┛┗┗┛┗┛┣┛┗┛┗ ┣┛┛ ┗┛┛┗┗┗ + # ┛ + + async def list_hotspot_profiles( + self, + use_https: bool = None + ) -> ApiResponse: + + """ + To enlist all the Hotspot Profiles that have been configured. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.get( + url = self.get_mikrotik_url(path = r"/ip/hotspot/profile", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def add_hotspot_profile( + self, + json_payload: dict, + use_https: bool = None + ) -> ApiResponse: + + """ + To add a new Hotspot Profile to the MikroTik's configuration. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.put( + url = self.get_mikrotik_url(path = r"/ip/hotspot/profile", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def update_hotspot_profile( + self, + dot_id: str, + json_payload: dict, + use_https: bool = None, + ) -> ApiResponse: + + """ + To update an existing Hotspot Profile in the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.patch( + url = self.get_mikrotik_url(path = f"/ip/hotspot/profile/{dot_id}", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def remove_hotspot_profile( + self, + dot_id: str, + use_https: bool = None, + ): + + """ + To delete an existing Hotspot Profile from the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.delete( + url = self.get_mikrotik_url(path = f"/ip/hotspot/profile/{dot_id}", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + # ┓┏ ┏┓ + # ┣┫┏┓╋┏┏┓┏┓╋ ┗┓┏┓┏┓┓┏┏┓┏┓ + # ┛┗┗┛┗┛┣┛┗┛┗ ┗┛┗ ┛ ┗┛┗ ┛ + # ┛ + + async def list_hotspot_servers( + self, + use_https: bool = None + ) -> ApiResponse: + + """ + To enlist all the Hotspot Servers that have been configured. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.get( + url = self.get_mikrotik_url(path = r"/ip/hotspot", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def add_hotspot_server( + self, + json_payload: dict, + use_https: bool = None + ) -> ApiResponse: + + """ + To add a new Hotspot Server to the MikroTik's configuration. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.put( + url = self.get_mikrotik_url(path = r"/ip/hotspot", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def update_hotspot_server( + self, + dot_id: str, + json_payload: dict, + use_https: bool = None, + ) -> ApiResponse: + + """ + To update an existing Hotspot Server in the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.patch( + url = self.get_mikrotik_url(path = f"/ip/hotspot/{dot_id}", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def remove_hotspot_server( + self, + dot_id: str, + use_https: bool = None, + ): + + """ + To delete an existing Hotspot Server from the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.delete( + url = self.get_mikrotik_url(path = f"/ip/hotspot/{dot_id}", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + # ┓┏ ┓ ┏ ┓┓ ┓ ┏┓ ┓ + # ┣┫┏┓╋┏┏┓┏┓╋ ┃┃┃┏┓┃┃┏┓┏┫━━┃┓┏┓┏┓┏┫┏┓┏┓ + # ┛┗┗┛┗┛┣┛┗┛┗ ┗┻┛┗┻┗┗┗ ┗┻ ┗┛┗┻┛ ┗┻┗ ┛┗ + # ┛ + + async def list_hotspot_walled_gardens( + self, + use_https: bool = None + ) -> ApiResponse: + + """ + To enlist all the Hotspot Walled-Gardens that have been configured. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.get( + url = self.get_mikrotik_url(path = r"/ip/hotspot/walled-garden", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def add_hotspot_walled_garden( + self, + json_payload: dict, + use_https: bool = None + ) -> ApiResponse: + + """ + To add a new Hotspot Walled-Garden to the MikroTik's configuration. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.put( + url = self.get_mikrotik_url(path = r"/ip/hotspot/walled-garden", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def update_hotspot_walled_garden( + self, + dot_id: str, + json_payload: dict, + use_https: bool = None, + ) -> ApiResponse: + + """ + To update an existing Hotspot Walled-Garden in the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.patch( + url = self.get_mikrotik_url(path = f"/ip/hotspot/walled-garden/{dot_id}", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def remove_hotspot_walled_garden( + self, + dot_id: str, + use_https: bool = None, + ): + + """ + To delete an existing Hotspot Walled-Garden from the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.delete( + url = self.get_mikrotik_url(path = f"/ip/hotspot/walled-garden/{dot_id}", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + # ┓┏ ┓ ┏ ┓┓ ┓ ┏┓ ┓ ┳┏┓ + # ┣┫┏┓╋┏┏┓┏┓╋ ┃┃┃┏┓┃┃┏┓┏┫━━┃┓┏┓┏┓┏┫┏┓┏┓ ┃┃┃ + # ┛┗┗┛┗┛┣┛┗┛┗ ┗┻┛┗┻┗┗┗ ┗┻ ┗┛┗┻┛ ┗┻┗ ┛┗ ┻┣┛ + # ┛ + + async def list_hotspot_walled_garden_ips( + self, + use_https: bool = None + ) -> ApiResponse: + + """ + To enlist all the Hotspot Walled-Garden IPs that have been configured. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.get( + url = self.get_mikrotik_url(path = r"/ip/hotspot/walled-garden/ip", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def add_hotspot_walled_garden_ip( + self, + json_payload: dict, + use_https: bool = None + ) -> ApiResponse: + + """ + To add a new Hotspot Walled-Garden IP to the MikroTik's configuration. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.put( + url = self.get_mikrotik_url(path = r"/ip/hotspot/walled-garden/ip", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def update_hotspot_walled_garden_ip( + self, + dot_id: str, + json_payload: dict, + use_https: bool = None, + ) -> ApiResponse: + + """ + To update an existing Hotspot Walled-Garden IP in the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.patch( + url = self.get_mikrotik_url(path = f"/ip/hotspot/walled-garden/ip/{dot_id}", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def remove_hotspot_walled_garden_ip( + self, + dot_id: str, + use_https: bool = None, + ): + + """ + To delete an existing Hotspot Walled-Garden IP from the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.delete( + url = self.get_mikrotik_url(path = f"/ip/hotspot/walled-garden/ip/{dot_id}", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + # ┳┓┏┓┳┓┳┳┳┏┓ + # ┣┫┣┫┃┃┃┃┃┗┓ + # ┛┗┛┗┻┛┻┗┛┗┛ + + async def set_radius_incoming( + self, + json_payload: dict, + use_https: bool = None + ) -> ApiResponse: + + """ + To set the parameters for the way the MikroTik devices receive incoming hits from RADIUS. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.post( + url = self.get_mikrotik_url(path = r"/radius/incoming/set", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def list_radius_servers( + self, + use_https: bool = None + ) -> ApiResponse: + + """ + To enlist all the RADIUS Servers that have been configured. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.get( + url = self.get_mikrotik_url(path = r"/radius", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def add_radius_server( + self, + json_payload: dict, + use_https: bool = None + ) -> ApiResponse: + + """ + To add a new RADIUS Server to the MikroTik's configuration. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.put( + url = self.get_mikrotik_url(path = r"/radius", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def update_radius_server( + self, + dot_id: str, + json_payload: dict, + use_https: bool = None, + ) -> ApiResponse: + + """ + To update an existing RADIUS Server in the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.patch( + url = self.get_mikrotik_url(path = f"/radius/{dot_id}", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def remove_radius_server( + self, + dot_id: str, + use_https: bool = None, + ): + + """ + To delete an existing RADIUS Server from the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.delete( + url = self.get_mikrotik_url(path = f"/radius/{dot_id}", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + # ┳┓┓┏┏┓┏┓ ┏┓ + # ┃┃┣┫┃ ┃┃ ┗┓┏┓┏┓┓┏┏┓┏┓ + # ┻┛┛┗┗┛┣┛ ┗┛┗ ┛ ┗┛┗ ┛ + + async def list_dhcp_servers( + self, + use_https: bool = None + ) -> ApiResponse: + + """ + To enlist all the DHCP Servers that have been configured. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.get( + url = self.get_mikrotik_url(path = r"/ip/dhcp-server", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def add_dhcp_server( + self, + json_payload: dict, + use_https: bool = None + ) -> ApiResponse: + + """ + To add a new DHCP Server to the MikroTik's configuration. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.put( + url = self.get_mikrotik_url(path = r"/ip/dhcp-server", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def update_dhcp_server( + self, + dot_id: str, + json_payload: dict, + use_https: bool = None, + ) -> ApiResponse: + + """ + To update an existing DHCP Server in the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.patch( + url = self.get_mikrotik_url(path = f"/ip/dhcp-server/{dot_id}", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def remove_dhcp_server( + self, + dot_id: str, + use_https: bool = None, + ): + + """ + To delete an existing DHCP Server from the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.delete( + url = self.get_mikrotik_url(path = f"/ip/dhcp-server/{dot_id}", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + # ┳┓┓┏┏┓┏┓ ┳┓ ┓ + # ┃┃┣┫┃ ┃┃ ┃┃┏┓╋┓┏┏┏┓┏┓┃┏ + # ┻┛┛┗┗┛┣┛ ┛┗┗ ┗┗┻┛┗┛┛ ┛┗ + + async def list_dhcp_networks( + self, + use_https: bool = None + ) -> ApiResponse: + + """ + To enlist all the DHCP Networks that have been configured. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.get( + url = self.get_mikrotik_url(path = r"/ip/dhcp-server/network", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def add_dhcp_network( + self, + json_payload: dict, + use_https: bool = None + ) -> ApiResponse: + + """ + To add a new DHCP Network to the MikroTik's configuration. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.put( + url = self.get_mikrotik_url(path = r"/ip/dhcp-server/network", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def update_dhcp_network( + self, + dot_id: str, + json_payload: dict, + use_https: bool = None, + ) -> ApiResponse: + + """ + To update an existing DHCP Network in the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.patch( + url = self.get_mikrotik_url(path = f"/ip/dhcp-server/network/{dot_id}", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def remove_dhcp_network( + self, + dot_id: str, + use_https: bool = None, + ): + + """ + To delete an existing DHCP Network from the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.delete( + url = self.get_mikrotik_url(path = f"/ip/dhcp-server/network/{dot_id}", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + # ┏┓┳┓┳┳┓┏┓ ┏┓ • + # ┗┓┃┃┃┃┃┃┃ ┃ ┏┓┏┳┓┏┳┓┓┏┏┓┓╋┓┏ + # ┗┛┛┗┛ ┗┣┛ ┗┛┗┛┛┗┗┛┗┗┗┻┛┗┗┗┗┫ + # ┛ + + async def list_snmp_communities( + self, + use_https: bool = None + ) -> ApiResponse: + + """ + To enlist all the SNMP Communities that have been configured. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.get( + url = self.get_mikrotik_url(path = r"/snmp/community", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def add_snmp_community( + self, + json_payload: dict, + use_https: bool = None + ) -> ApiResponse: + + """ + To add a new SNMP Community to the MikroTik's configuration. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.put( + url = self.get_mikrotik_url(path = r"/snmp/community", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def update_snmp_community( + self, + dot_id: str, + json_payload: dict, + use_https: bool = None, + ) -> ApiResponse: + + """ + To update an existing SNMP Community in the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.patch( + url = self.get_mikrotik_url(path = f"/snmp/community/{dot_id}", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def remove_snmp_community( + self, + dot_id: str, + use_https: bool = None, + ): + + """ + To delete an existing SNMP Community from the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.delete( + url = self.get_mikrotik_url(path = f"/snmp/community/{dot_id}", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + # ┳┓┏┓┏┳┓ • + # ┃┃┣┫ ┃ ━━┓┏┓┏┓ + # ┛┗┛┗ ┻ ┗┛┗┗┫ + # ┛ + + async def list_firewall_nat( + self, + use_https: bool = None + ) -> ApiResponse: + + """ + To enlist all the Firewall NAT that have been configured. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.get( + url = self.get_mikrotik_url(path = r"/ip/firewall/nat", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def add_firewall_nat( + self, + json_payload: dict, + use_https: bool = None + ) -> ApiResponse: + + """ + To add a new Firewall NAT to the MikroTik's configuration. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.put( + url = self.get_mikrotik_url(path = r"/ip/firewall/nat", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def update_firewall_nat( + self, + dot_id: str, + json_payload: dict, + use_https: bool = None, + ) -> ApiResponse: + + """ + To update an existing Firewall NAT in the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.patch( + url = self.get_mikrotik_url(path = f"/ip/firewall/nat/{dot_id}", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + async def remove_firewall_nat( + self, + dot_id: str, + use_https: bool = None, + ): + + """ + To delete an existing Firewall NAT from the MikroTik's configuration. + :param dot_id: The value of the '.id' field in the list. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.delete( + url = self.get_mikrotik_url(path = f"/ip/firewall/nat/{dot_id}", use_https = use_https), + auth = self._basic_auth + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + + # ┳┓┳┓┏┓ + # ┃┃┃┃┗┓ + # ┻┛┛┗┗┛ + + async def set_dns( + self, + json_payload: dict, + use_https: bool = None + ) -> ApiResponse: + + """ + To set the DNS parameters for the MikroTik. + :param json_payload: The JSON to send in the body of the request. + :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. + :return: A structured API response. + """ + + # Make the API call: + api_response = await self._rest.post( + url = self.get_mikrotik_url(path = r"/ip/dns/set", use_https = use_https), + auth = self._basic_auth, + json = json_payload + ) + + # Extract the needed values: + if api_response.success: api_response.data = await api_response.get_json() + else: api_response.message = await self.get_failure_message(api_response) + + # Done here: + await self.log_action(api_response) + return api_response + # ***************************************************************************************************************** # ***** **** @@ -1049,7 +2288,7 @@ if __name__ == "__main__": username = "easyfi", password = "easyfi", use_https = False, - config_by = "bhopli", + config_by = "easyfi", action_log_conn = logs_mongo ) @@ -1057,22 +2296,308 @@ if __name__ == "__main__": await logs_mongo.connect() - # api_response = await my_mikrotik.list_system_resources() - # print("SYS RESOURCES:", json.to_string(api_response.data)) - # - # api_response = await my_mikrotik.list_interfaces() - # print(f"INTERFACES ({len(api_response.data)}):", json.to_string(api_response.data)) - # - # api_response = await my_mikrotik.list_vlans() - # print(f"VLANS (({len(api_response.data)})):", json.to_string(api_response.data)) + api_response = await my_mikrotik.list_system_resources() + print("SYS RESOURCES:", json.to_string(api_response.data)) + + api_response = await my_mikrotik.list_interfaces() + print(f"INTERFACES ({len(api_response.data)}):", json.to_string(api_response.data)) + + api_response = await my_mikrotik.list_vlans() + print(f"VLANS ({len(api_response.data)}):", json.to_string(api_response.data)) api_response = await my_mikrotik.list_ip_pools() print(f"IP POOLS ({len(api_response.data)}):", json.to_string(api_response.data)) - # api_response = await my_mikrotik.list_ppp_profiles() - # print(f"PPP PROFILES ({len(api_response.data)}):", json.to_string(api_response.data)) - # - # api_response = await my_mikrotik.list_ppp_servers() - # print(f"PPP SERVERS ({len(api_response.data)}):", json.to_string(api_response.data)) + api_response = await my_mikrotik.list_ppp_profiles() + print(f"PPP PROFILES ({len(api_response.data)}):", json.to_string(api_response.data)) - asyncio.run(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_radius_servers() + print(f"RADIUS SERVERS ({len(api_response.data)}):", json.to_string(api_response.data)) + + async def hotspot_steps(): + + HS_VLAN_ID = "2001" + + HS_INTERFACE_NAME = "easyfi-hs-if" + HS_VLAN_NAME = "easyfi-hs-vlan-" + HS_VLAN_ID + HS_IP_POOL_NAME = "easyfi-hs-pool-" + HS_VLAN_ID + HS_DHCP_SERVER_NAME = "easyfi-hs-dhcp-" + HS_VLAN_ID + HS_SNMP_COMMUNITY_NAME = "lkjhgfdsa1234567" + + HS_PUBLIC_IP_SUBNET = "154.84.218.42/25" + HS_FIRST_PUBLIC_IP = "154.84.218.0" + HS_LAST_PUBLIC_IP = "154.84.218.127" + HS_PRIVATE_IP_SUBNET = "100.65.0.0/22" + HS_FIRST_PRIVATE_IP = "100.65.0.0" + HS_LAST_PRIVATE_IP = "100.65.3.255" + HS_GATEWAY_PRIVATE_IP = "100.65.0.1" + + HS_RADIUS_SERVER_IP = "154.84.218.42" + HS_RADIUS_SERVER_SECRET = "asdfghjkl" + + HS_COMMENT = my_mikrotik.create_comment_json() + + # # STEP 1: + # # Set up the interface: + # api_response = await my_mikrotik.update_interface( + # dot_id = "*3", + # json_payload = { + # "name": HS_INTERFACE_NAME, + # "comment": HS_COMMENT + # } + # ) + # print("STEP 1:", api_response.success, f"({api_response.action})") + # print("MESSAGE:", api_response.message) + # print("JSON:", api_response.data) + # print("\n\n") + # + # # STEP 2: + # # Create the VLAN: + # api_response = await my_mikrotik.add_vlan( + # json_payload = { + # "name": HS_VLAN_NAME, + # "interface": HS_INTERFACE_NAME, + # "vlan-id": HS_VLAN_ID, + # "disabled": "false", + # "comment": HS_COMMENT + # } + # ) + # print("STEP 2:", api_response.success, f"({api_response.action})") + # print("MESSAGE:", api_response.message) + # print("JSON:", api_response.data) + # print("\n\n") + # + # # STEP 3: + # # Bind a new IP address/network to the VLAN: + # api_response = await my_mikrotik.add_ip_address_binding( + # json_payload = { + # "address": HS_PRIVATE_IP_SUBNET, + # "interface": HS_VLAN_NAME, + # "disabled": "false", + # "comment": HS_COMMENT + # } + # ) + # print("STEP 3:", api_response.success, f"({api_response.action})") + # print("MESSAGE:", api_response.message) + # print("JSON:", api_response.data) + # print("\n\n") + # + # # STEP 4: + # # Add a new IP Pool for the Hotspot users: + # api_response = await my_mikrotik.add_ip_pool( + # json_payload = { + # "name": HS_IP_POOL_NAME, + # "ranges": HS_PRIVATE_IP_SUBNET, + # "comment": HS_COMMENT + # } + # ) + # print("STEP 4:", api_response.success, f"({api_response.action})") + # print("MESSAGE:", api_response.message) + # print("JSON:", api_response.data) + # print("\n\n") + # + # # STEP 5: + # # Add a new Hotspot Profile: + # api_response = await my_mikrotik.add_hotspot_profile( + # json_payload = { + # "dns-name": "hs01.easyfi.net.in", + # "hotspot-address": HS_GATEWAY_PRIVATE_IP, + # "html-directory": "hotspot", + # "html-directory-override": "", + # "http-cookie-lifetime": "3d", + # "http-proxy": "0.0.0.0:0", + # "install-hotspot-queue": "false", + # "login-by": "cookie,http-chap", + # "name": "hs01.easyfi.net.in", + # "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 7: + # # Add a new DHCP Server: + # api_response = await my_mikrotik.add_dhcp_server( + # json_payload = { + # "address-pool": HS_IP_POOL_NAME, + # "authoritative": "yes", + # "disabled": "false", + # "interface": HS_VLAN_NAME, + # "lease-time": "30m", + # "name": HS_DHCP_SERVER_NAME, + # "use-radius": "no", + # "comment": HS_COMMENT + # } + # ) + # print("STEP 7:", api_response.success, f"({api_response.action})") + # print("MESSAGE:", api_response.message) + # print("JSON:", api_response.data) + # print("\n\n") + + # STEP 8: + # Add a new DHCP Network: + api_response = await my_mikrotik.add_dhcp_network( + json_payload = { + "address": HS_PRIVATE_IP_SUBNET, + "gateway": HS_GATEWAY_PRIVATE_IP, + "netmask": HS_PRIVATE_IP_SUBNET.split("/")[-1], + "dns-server": HS_GATEWAY_PRIVATE_IP, + "comment": HS_COMMENT + } + ) + print("STEP 8:", api_response.success, f"({api_response.action})") + print("MESSAGE:", api_response.message) + print("JSON:", api_response.data) + print("\n\n") + + # # STEP 9: + # # Set up the Walled-Garden IP: + # api_response = await my_mikrotik.add_hotspot_walled_garden_ip( + # json_payload = { + # "action": "accept", + # "dst-address": HS_RADIUS_SERVER_IP, + # "comment": HS_COMMENT + # } + # ) + # print("STEP 9:", api_response.success, f"({api_response.action})") + # print("MESSAGE:", api_response.message) + # print("JSON:", api_response.data) + # print("\n\n") + # + # # STEP 10: + # # Add a new RADIUS Server: + # api_response = await my_mikrotik.add_radius_server( + # json_payload = { + # "accounting-port": "1813", + # "address": HS_RADIUS_SERVER_IP, + # "authentication-port": "1812", + # "disabled": "false", + # "protocol": "udp", + # "secret": HS_RADIUS_SERVER_SECRET, + # "service": "ppp,login,hotspot,dhcp", + # "timeout": "300ms", + # "comment": HS_COMMENT + # } + # ) + # print("STEP 10:", api_response.success, f"({api_response.action})") + # print("MESSAGE:", api_response.message) + # print("JSON:", api_response.data) + # print("\n\n") + # + # # STEP 11: + # # Allow incoming traffic from RADIUS: + # api_response = await my_mikrotik.set_radius_incoming( + # json_payload = { + # "accept": "yes", + # "port": "3799", + # # "comment": HS_COMMENT + # } + # ) + # print("STEP 11:", api_response.success, f"({api_response.action})") + # print("MESSAGE:", api_response.message) + # print("JSON:", api_response.data) + # print("\n\n") + # + # # STEP 12: + # # Set up the DNS: + # api_response = await my_mikrotik.set_dns( + # json_payload = { + # "allow-remote-requests": "true", + # "servers": "8.8.8.8,8.8.4.4", + # # "comment": HS_COMMENT + # } + # ) + # print("STEP 12:", api_response.success, f"({api_response.action})") + # print("MESSAGE:", api_response.message) + # print("JSON:", api_response.data) + # print("\n\n") + # + # # STEP 13: + # # Map the private IPs to the public IPs for NAT-ing: + # 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 + # ) + # for index, nat_rule in enumerate(nat_map): + # api_response = await my_mikrotik.add_firewall_nat( + # json_payload = { + # "action": "src-nat", + # "chain": "srcnat", + # "disabled": "false", + # "src-address": nat_rule["network"], + # "to-addresses": nat_rule["target"], + # "comment": HS_COMMENT + # } + # ) + # print(f"STEP 13 ({index+1}/{len(nat_map)}):", api_response.success, f"({api_response.action})") + # print("MESSAGE:", api_response.message) + # print("JSON:", api_response.data) + # print("\n\n") + # + # # STEP 14: + # # Set up the SNMP Community: + # api_response = await my_mikrotik.add_snmp_community( + # json_payload = { + # "addresses": "::/0", + # "authentication-password": "", + # "authentication-protocol": "MD5", + # "disabled": "false", + # "encryption-password": "", + # "encryption-protocol": "DES", + # "name": HS_SNMP_COMMUNITY_NAME, + # "read-access": "true", + # "security": "none", + # "write-access": "false", + # "comment": HS_COMMENT + # } + # ) + # print("STEP 14:", api_response.success, f"({api_response.action})") + # print("MESSAGE:", api_response.message) + # print("JSON:", api_response.data) + # print("\n\n") + + + asyncio.run(hotspot_steps())