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

(20250211) Completed documentation for MikroTik's PPPoE system as per my current understanding.

Browse Source
This commit is contained in:
khushal
2025-02-11 17:37:56 +05:30
parent 382c6c991d
commit cf6520d119
3 changed files with 556 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controllers_v2/software/mikrotik/base.py
+162 -19
View File
@@ -21,8 +21,7 @@
N/A
"""
import datetime
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
@@ -36,6 +35,8 @@ sys.path.append(".")
sys.path.append("..")
# My async utils:
from utils_v2.string import json
from utils_v2.date_time import date_time
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
@@ -185,7 +186,7 @@ class MikroTikController(CoreSoftwareController, ABC):
@staticmethod
def get_mikrotik_url(
nas_ip: str,
mikrotik_ip: str,
path: str,
port_no: int | str = None,
use_https: bool = True
@@ -193,7 +194,7 @@ class MikroTikController(CoreSoftwareController, ABC):
"""
Simply creates the base URL for hitting the MikroTik server.
:param nas_ip: The IP address of the MikroTik device.
: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.
@@ -201,46 +202,46 @@ class MikroTikController(CoreSoftwareController, ABC):
"""
base_url = r"https://" if use_https else r"http://"
base_url += nas_ip
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_vlans(
def split_ipv4_range_among_targets(
start_ip: ipaddress.IPv4Address | str,
end_ip: ipaddress.IPv4Address | str,
vlan_ids: List[int]
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 vlan_ids: The list of VLAN ids (need not be sequential or ordered).
:return: A list of dicts that describes each VLAN.
: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)
vlan_count = len(vlan_ids)
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_vlan = math.ceil(total_ips / vlan_count) + 3
ips_per_target = math.ceil(total_ips / target_count) + 3
# Generate subnets for each VLAN:
subnets = []
current_ip = start_ip_obj
for vlan_id in vlan_ids:
for target in targets:
# Calculate the network address for the current VLAN:
subnet_network = ipaddress.IPv4Network(f"{current_ip}/{32 - (ips_per_vlan).bit_length()}", strict = False)
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
@@ -249,7 +250,7 @@ class MikroTikController(CoreSoftwareController, ABC):
# Add the calculated subnet to the list:
subnets.append({
"vlanId": vlan_id,
"target": target,
"network": str(subnet_network),
"size": int(2 ** (32 - subnet_network.prefixlen)),
"startIp": str(subnet_start_ip_obj),
@@ -262,6 +263,51 @@ class MikroTikController(CoreSoftwareController, ABC):
# 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
# ┏┓
# ┗┓┓┏┏╋┏┓┏┳┓
# ┗┛┗┫┛┗┗ ┛┗┗
@@ -269,7 +315,7 @@ class MikroTikController(CoreSoftwareController, ABC):
async def get_system_resource(
self,
nas_ip: str,
mikrotik_ip: str,
username: str,
password: str,
port_no: int | str = None,
@@ -279,7 +325,7 @@ class MikroTikController(CoreSoftwareController, ABC):
"""
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 nas_ip: The IP address of 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.
@@ -287,10 +333,10 @@ class MikroTikController(CoreSoftwareController, ABC):
:return: A structured API response.
"""
# Make the API call and return the response:
return await self._rest.get(
# Make the API call:
api_response = await self._rest.get(
url = self.get_mikrotik_url(
nas_ip = nas_ip,
mikrotik_ip = mikrotik_ip,
path = r"/system/resource",
port_no = port_no,
use_https = use_https
@@ -301,6 +347,103 @@ class MikroTikController(CoreSoftwareController, ABC):
)
)
# 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
):
"""
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,
interface_id: str,
port_no: int | str = None,
use_https: bool = True,
name: str = None,
disabled: bool = False,
comment: str = None
):
"""
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 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.
: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}",
port_no = port_no,
use_https = use_https
),
auth = httpx.BasicAuth(
username = username,
password = password
),
json = input_json
)
# Done here:
return api_response
# ┏┓ ┓
# ┣┫┓┏╋┣┓
# ┛┗┗┻┗┛┗