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

(20250211) MikroTik VLAN set up and rollback also ready.

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