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

(20250212) Started a new class (util) that handles MikroTik config. in async mode, and started implementing that in the PPPoE-1000 controller. I can now configure and roll back the first three steps. Long way to go :')

Browse Source
This commit is contained in:
khushal
2025-02-12 19:05:29 +05:30
parent 6e65de9dfc
commit 02dd675f83
8 changed files with 409 additions and 607 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controllers_v2/software/mikrotik/base.py
+5 -410
View File
@@ -37,6 +37,7 @@ sys.path.append("..")
# My async utils:
from utils_v2.string import json
from utils_v2.date_time import date_time
from utils_v2.mikrotik.controllers.async_mikrotik import AsyncMikroTik
from utils_v2.database.async_mysql_v2 import AsyncMySQL
from utils_v2.database.async_mongo_v2 import AsyncMongo
from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache
@@ -179,416 +180,6 @@ class MikroTikController(CoreSoftwareController, ABC):
debug_only_errors = debug_only_errors
)
# ┓┏ ┓
# ┣┫┏┓┃┏┓┏┓┏┓┏
# ┛┗┗ ┗┣┛┗ ┛ ┛
# ┛
@staticmethod
def get_mikrotik_url(
mikrotik_ip: str,
path: str,
port_no: int | str = None,
use_https: bool = True
) -> str:
"""
Simply creates the base URL for hitting the MikroTik server.
:param mikrotik_ip: The IP address of the MikroTik device.
:param path: The path of the REST API to hit.
:param port_no: The port no. to hit the MikroTik device on.
:param use_https: Whether to use HTTPS, or HTTP.
:return: The Base URL string.
"""
base_url = r"https://" if use_https else r"http://"
base_url += mikrotik_ip
if port_no is not None: base_url += f":{port_no}"
base_url += "/rest"
if not path.startswith("/"): path = "/" + path
return base_url + path
@staticmethod
def split_ipv4_range_among_targets(
start_ip: ipaddress.IPv4Address | str,
end_ip: ipaddress.IPv4Address | str,
targets: List
) -> List[dict]:
"""
Divides the IPv4 range over the list of VLAN ids.
:param start_ip: The first IP in the full pool (range).
:param end_ip: The last IP in the full pool (range).
:param targets: The list of targets (need not be sequential or ordered) like VLAN ids.
:return: A list of dicts that describes the subnet for each target.
"""
# Parse the inputs:
start_ip_obj = start_ip if isinstance(start_ip, ipaddress.IPv4Address) else ipaddress.IPv4Address(start_ip)
end_ip_obj = end_ip if isinstance(end_ip, ipaddress.IPv4Address) else ipaddress.IPv4Address(end_ip)
target_count = len(targets)
# Calculate the total number of IPs in the range:
total_ips = int(end_ip_obj) - int(start_ip_obj) + 1
# Calculate the no. of IPs each VLAN gets,
# and add three because we need IPs for network, gateway and broadcast:
ips_per_target = math.ceil(total_ips / target_count) + 3
# Generate subnets for each VLAN:
subnets = []
current_ip = start_ip_obj
for target in targets:
# Calculate the network address for the current VLAN:
subnet_network = ipaddress.IPv4Network(f"{current_ip}/{32 - ips_per_target.bit_length()}", strict = False)
# Check for overlap with the parent network:
subnet_start_ip_obj = subnet_network.network_address
subnet_end_ip_obj = subnet_network.broadcast_address
if subnet_start_ip_obj <= end_ip_obj:
# Add the calculated subnet to the list:
subnets.append({
"target": target,
"network": str(subnet_network),
"size": int(2 ** (32 - subnet_network.prefixlen)),
"startIp": str(subnet_start_ip_obj),
"endIp": str(subnet_end_ip_obj)
})
# Update the current IP for the next subnet
current_ip = ipaddress.IPv4Address(int(subnet_network.broadcast_address) + 1)
# Done here:
return subnets
@staticmethod
def create_comment_json(
created_by: str,
created_ts: datetime.datetime = None,
roll_back_config: dict = None
) -> str:
"""
To create a JSON string that can be put as a comment in any step to later identify the work that was done and
have some hint about how to roll it back if needed.
:param created_by: A hint to put to identify which process/tool created this comment/config/process.
:param created_ts: The datetime (preferably UTC) when this comment was created.
:param roll_back_config: The original config that can be used when rolling back.
:return: The comment string to be used.
"""
# Create the JSON:
comment_json = {
"createdBy": created_by,
"createdTs": (created_ts or date_time.get_current_utc_date_time(as_string = False)).timestamp(),
"rollbackConfig": roll_back_config
}
# Return a string:
return json.to_string(comment_json, no_space = True)
@staticmethod
def parse_comment_json(comment: str) -> dict | list | None:
"""
Tries to parse a comment as if it were a JSON string.
:param comment: The comment string.
:return: The dict/list parsed form the JSON, or a null value of the input was not a valid JSON string.
"""
# Start with a null value:
parsed = None
# Try to parse the comment as if it were a JSON string:
try: parsed = json.from_string(comment)
except: parsed = None
# Done here:
return parsed
# ┏┓
# ┗┓┓┏┏╋┏┓┏┳┓
# ┗┛┗┫┛┗┗ ┛┗┗
# ┛
async def get_system_resource(
self,
mikrotik_ip: str,
username: str,
password: str,
port_no: int | str = None,
use_https: bool = True
) -> ApiResponse:
"""
To get a summary of the hardware resources available in the MikroTik device. This also becomes a great way to
quickly check if any given device is valid, and up and running.
: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"/system/resource",
port_no = port_no,
use_https = use_https
),
auth = httpx.BasicAuth(
username = username,
password = password
)
)
# Done here:
return api_response
# ┳ ┏
# ┃┏┓╋┏┓┏┓╋┏┓┏┏┓┏
# ┻┛┗┗┗ ┛ ┛┗┻┗┗ ┛
async def list_interfaces(
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",
port_no = port_no,
use_https = use_https
),
auth = httpx.BasicAuth(
username = username,
password = password
)
)
# Done here:
return api_response
async def update_interface(
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 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 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 json_payload: The JSON to send in the body pf the request.
: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/{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 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:
return api_response
# ┏┓ ┓
# ┣┫┓┏╋┣┓
# ┛┗┗┻┗┛┗
@@ -620,12 +211,14 @@ class MikroTikController(CoreSoftwareController, ABC):
@abstractmethod
async def roll_back(
self,
mikrotik_client: AsyncMikroTik,
mikrotik_auth: MikroTikPPPoE1000Auth | MikroTikHotspot1000Auth
) -> MikroTikRollBackAttemptResponse:
"""
The rolling-back to the original state (as best as possible) in case the configurations fails midway after
completing some no. of steps.
:param mikrotik_client: The client to use.
:param mikrotik_auth: The set of credentials as received from the UI/API.
:return: A structured response to indicate what happened during the configuration attempt.
"""
@@ -635,11 +228,13 @@ class MikroTikController(CoreSoftwareController, ABC):
@abstractmethod
async def configure(
self,
mikrotik_client: AsyncMikroTik,
mikrotik_auth: MikroTikPPPoE1000Auth | MikroTikHotspot1000Auth
) -> MikroTikConfigAttemptResponse:
"""
Run the configuration steps for the system.
:param mikrotik_client: The client to use.
:param mikrotik_auth: The set of credentials as received from the UI/API.
:return: A structured response to indicate what happened during the configuration attempt.
"""