khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20250213) Many things added to the mikrotik util. Even added a sample hotspot config. code to it.

Browse Source
This commit is contained in:
khushal
2025-02-13 16:59:00 +05:30
parent 519aff21ca
commit b913f91fbd
2 changed files with 2137 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controllers_v2/software/mikrotik/mikrotik_pppoe_1000.py
+588 -54
View File
@@ -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)
utils_v2/mikrotik/controllers/async_mikrotik.py
+1549 -24
View File
File diff suppressed because it is too large Load Diff