khushalps/api_internal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b53ef86ef83302f2663e591baf8e52b9de3caa58
api_internal/mikrotik/controllers/async_mikrotik.py
T
khushalps b53ef86ef8 Squashed 'utils_v2/' content from commit 7f27356 
git-subtree-dir: utils_v2
git-subtree-split: 7f273565196085feb05ee3328aa2e80d3d721fc3
2025-06-12 10:54:05 +05:30

2657 lines
97 KiB
Python
Raw Blame History

"""
AUTHOR:
Khushal P Soonderji
DATE:
Monday, 10th Feb., 2025.
OBJECTIVE:
To handle all MikroTik configuration from one place.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# System-level activities:
import distro
import socket
import platform
# My async utils:
from utils_v2.string import json
from utils_v2.date_time import date_time
from utils_v2.database.async_mongo_v2 import AsyncMongo
from utils_v2.api.async_quart import describe_exception
# To make very controlled API calls:
from utils_v2.rest.controllers.async_base import AsyncREST
from utils_v2.rest.controllers.auth import BasicAuth
from utils_v2.rest.models.api_call import ApiResponse
# To work with IP Addresses:
import ipaddress
# To work with datatypes:
from typing import List, Any
# To make HTTP requests:
import httpx
# To work with date and time:
import datetime
# For asynchronous activities:
import asyncio
# Misc:
import math
import random
import string
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# Info for logging that will stay constant during runtime:
SERVER_HOSTNAME = str(socket.gethostname())
PLATFORM_INFO = platform.uname()
HOST_OS = str(distro.name(True))
HOST_CPU = f"{PLATFORM_INFO.processor} ({PLATFORM_INFO.machine})"
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class AsyncMikroTik:
# ┏┓┓ ┓┏
# ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏
# ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛
ACTION_LOG_COLLECTION = "_mikrotikActions"
# ┏┓
# ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓
# ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛
def __init__(
self,
config_by: str,
mikrotik_ip: str,
username: str,
password: str,
port: int = None,
use_https: bool = True,
http_client: httpx.AsyncClient = None,
action_log_conn: AsyncMongo = None,
debug: bool = True,
debug_prefix: str = "MikroTik (C) | ",
debug_only_errors: bool = True
):
"""
To control MikroTik devices in an async manner.
:param http_client: The HTTP client to use to make REST-ful API calls.
:param debug: Whether, or not, you would like to print debugging messages:
:param debug_prefix: The prefix to print with the debugging messages.
:param debug_only_errors: Whether you would like to print only error messages or all messages.
:return: None.
"""
# Save the input values:
self._config_by = config_by
self._mikrotik_ip = mikrotik_ip
self._username = username
self._password = password
self._port = port
self._use_https = use_https
# For logging actions to Mongo:
self._action_log_conn = action_log_conn
self._action_log_chain = None
self.new_action_chain()
# For controlled REST-ful calls:
self._rest = AsyncREST(
http_client = http_client,
debug = debug,
debug_prefix = debug_prefix,
debug_only_errors = debug_only_errors
)
self._basic_auth = BasicAuth(
username = self._username,
password = self._password
)
# ┏┓ •
# ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
# ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
# ┛
@property
def action_chain(self):
return self._action_log_chain
# ┓┏ ┓
# ┣┫┏┓┃┏┓┏┓┏┓┏
# ┛┗┗ ┗┣┛┗ ┛ ┛
# ┛
def get_mikrotik_url(
self,
path: str,
use_https: bool = None
) -> str:
"""
Simply creates the base URL for hitting the MikroTik server.
:param path: The path of the REST API to hit.
:param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting.
:return: The Base URL string.
"""
use_https = self._use_https if use_https is None else use_https
base_url = r"https://" if use_https else r"http://"
base_url += self._mikrotik_ip
if self._port is not None: base_url += f":{self._port}"
base_url += "/rest"
if not path.startswith("/"): path = "/" + path
return base_url + path
@staticmethod
def split_ipv4_range_equally(
start_ip: ipaddress.IPv4Address | str,
end_ip: ipaddress.IPv4Address | str,
targets: List
) -> List[dict]:
"""
Divides the IPv4 range over the list of targets (like VLAN ids) in a simple fashion
: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
# Figure out the no. of IPs in each block:
ips_per_target = math.ceil(total_ips / target_count)
# Iterate through the list of targets, and give them the IPs:
subnets = []
current_ip = start_ip_obj
for index, target in enumerate(targets):
# Figure out the first and last IPs for this object:
subnet_start_ip_obj = current_ip
if index == target_count - 1: subnet_end_ip_obj = end_ip_obj
else: subnet_end_ip_obj = subnet_start_ip_obj + ips_per_target - 1
# Add the calculated subnet to the list:
subnets.append({
"target": target,
"network": str(subnet_start_ip_obj) + "-" + str(subnet_end_ip_obj),
"size": int(subnet_end_ip_obj) - int(subnet_start_ip_obj) + 1,
"startIp": str(subnet_start_ip_obj),
"endIp": str(subnet_end_ip_obj)
})
# Update the current IP for the next subnet:
current_ip += ips_per_target
# Done here:
return subnets
@staticmethod
def split_ipv4_range_in_powers_of_two(
start_ip: ipaddress.IPv4Address | str,
end_ip: ipaddress.IPv4Address | str,
targets: List,
consider_reserved_ips: bool = True,
strict: bool = False
) -> List[dict]:
"""
Divides the IPv4 range over the list of targets (like VLAN ids) in blocks whose sizes are in powers of 2. This
means that you won't have a block which has an unusual size like 7.
: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.
:param consider_reserved_ips: Whether, or not, to add 3 IPs per block (network, gateway, and broadcast) when
splitting the range.
:param strict: Whether, or not, you want strict dvision of blocks such that the network address is always
x.x.x.0 and the broadcast address is always x.x.x.255.
: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 (subtract one because counting starts from 0),
# and add three (if asked) because we need IPs for network, gateway and broadcast:
ips_per_target = math.ceil(total_ips / target_count) - 1
if consider_reserved_ips: ips_per_target += 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 = strict)
# 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
def create_comment_json(
self,
config_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 config_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 = {
"configBy": self._config_by,
"configTs": (config_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
@staticmethod
def get_original_config(comment_json: dict) -> dict | None:
"""
Once you have the comment's JSON unpacked successfully, pass that dict here to get the original config back.
:param comment_json: The dict received from parsing the comment's JSON string.
:return: The dict that describes the previous config, or None if no previous config was found.
"""
return comment_json.get("rollbackConfig") if isinstance(comment_json, dict) else None
def is_my_config(
self,
config_json: dict,
name_substring: str = None
) -> bool:
"""
Looks at a particular resource's JSON and tells if it was created by this class.
:param config_json: The JSON of the resource whose config needs to be checked.
:param name_substring: A substring (case-sensitive) to look for in the 'name' field.
:return: True if the config was created by this class, else False.
"""
# Extract needed values:
name = config_json.get("name", "???")
comment_json = self.parse_comment_json(config_json.get("comment")) or {}
# Perform needed checks:
if (
name.find(name_substring or "bhopli-muchhi-cookie") >= 0 or
comment_json.get("configBy", "???") == self._config_by
): return True
else: return False
def new_action_chain(self, action_chain: str | int = None) -> None:
"""
Generates a new random string to mark a new action chain and saves it in the local variable.
:param action_chain: A custom action chain to start using. A random string will be created if not provided.
:return: None
"""
if action_chain: self._action_log_chain = action_chain
else: self._action_log_chain = "".join(random.choices(string.ascii_letters + string.digits, k = 8))
async def log_action(
self,
api_response: ApiResponse,
) -> None:
"""
Invoke this to record actions automatically.
:param api_response: The response received from invoking an API provided by MikroTik.
:return: None.
"""
# Proceed only if we have a logging connection:
if self._action_log_conn is not None:
# Prepare the final log:
action_log_json = {
"hostname": SERVER_HOSTNAME,
"hostOs": HOST_OS,
"hostCpu": HOST_CPU,
"actionChain": self._action_log_chain,
"configBy": self._config_by,
"mikrotikIp": self._mikrotik_ip,
"url": api_response.url,
"method": api_response.method,
"action": api_response.action,
"config": api_response.requestJson,
"httpCode": api_response.httpCode,
"success": api_response.success,
"message": api_response.message,
"exception": describe_exception(api_response.exception),
"ts": date_time.get_current_utc_date_time(as_string = False)
}
# Record the action:
asyncio.create_task(self._action_log_conn.insert_one(
collection = self.ACTION_LOG_COLLECTION,
document = action_log_json
))
@staticmethod
async def get_failure_message(api_response: ApiResponse) -> str:
# On successful API calls, MikroTik doesn't give messages:
if api_response.success: message = "Success"
# On failure, the message is available in the JSON payload:
else:
response_json = await api_response.get_json()
message = f"{response_json.get('message')} ({response_json.get('detail')})"
# Done here:
return message
# ┏┓
# ┗┓┓┏┏╋┏┓┏┳┓
# ┗┛┗┫┛┗┗ ┛┗┗
# ┛
async def list_system_resources(
self,
use_https: bool = None
) -> 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 use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. 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"/system/resource", use_https = use_https),
auth = self._basic_auth
)
# Extract the needed values:
if api_response.success: api_response.data = await api_response.get_json()
# Done here:
await self.log_action(api_response)
return api_response
# ┳ ┏
# ┃┏┓╋┏┓┏┓╋┏┓┏┏┓┏
# ┻┛┗┗┗ ┛ ┛┗┻┗┗ ┛
async def list_interfaces(
self,
use_https: bool = None
) -> ApiResponse:
"""
To enlist all the physical connectivity interfaces available on the MikroTik device.
:param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. 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"/interface", 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 update_interface(
self,
dot_id: str,
json_payload: dict,
use_https: bool = None
) -> ApiResponse:
"""
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 of the request.
:return: A structured API response.
"""
# Make the API call:
api_response = await self._rest.patch(
url = self.get_mikrotik_url(
path = f"/interface/{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 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
# ┓┏┓ ┏┓┳┓
# ┃┃┃ ┣┫┃┃┏
# ┗┛┗┛┛┗┛┗┛
async def list_vlans(
self,
use_https: bool = None
) -> ApiResponse:
"""
To enlist all the physical connectivity interfaces available on the MikroTik device.
: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"/interface/vlan", 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_vlan(
self,
json_payload: dict,
use_https: bool = None
) -> ApiResponse:
"""
To add a new VLAN 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"/interface/vlan", 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_vlan(
self,
dot_id: str,
json_payload: dict,
use_https: bool = None,
) -> ApiResponse:
"""
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 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"/interface/vlan/{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_vlan(
self,
dot_id: str,
use_https: bool = None,
):
"""
To delete an existing VLAN 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"/interface/vlan/{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_ip_pools(
self,
use_https: bool = None
) -> ApiResponse:
"""
To enlist all the IP Pools 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/pool", 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_pool(
self,
json_payload: dict,
use_https: bool = None
) -> ApiResponse:
"""
To add a new IP Pool 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/pool", 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_pool(
self,
dot_id: str,
json_payload: dict,
use_https: bool = None,
) -> ApiResponse:
"""
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 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/pool/{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_pool(
self,
dot_id: str,
use_https: bool = None,
):
"""
To delete an existing IP Pool 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/pool/{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_ppp_profiles(
self,
use_https: bool = None
) -> ApiResponse:
"""
To enlist all the PPP 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"/ppp/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_ppp_profile(
self,
json_payload: dict,
use_https: bool = None
) -> ApiResponse:
"""
To add a new PPP 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"/ppp/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_ppp_profile(
self,
dot_id: str,
json_payload: dict,
use_https: bool = None,
) -> ApiResponse:
"""
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 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"/ppp/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_ppp_profile(
self,
dot_id: str,
use_https: bool = None,
):
"""
To delete an existing PPP 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"/ppp/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_ppp_servers(
self,
use_https: bool = None
) -> ApiResponse:
"""
To enlist all the PPP 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"/interface/pppoe-server/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_ppp_server(
self,
json_payload: dict,
use_https: bool = None
) -> ApiResponse:
"""
To add a new PPP 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"/interface/pppoe-server/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_ppp_server(
self,
dot_id: str,
json_payload: dict,
use_https: bool = None,
) -> ApiResponse:
"""
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 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"/interface/pppoe-server/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_ppp_server(
self,
dot_id: str,
use_https: bool = None,
):
"""
To delete an existing PPP 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"/interface/pppoe-server/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 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
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
# Make an instance of an HTTP client to use to make API calls:
my_http_client = httpx.AsyncClient(
limits = httpx.Limits(
max_connections = 100, # ............ Maximum number of connections allowed in the pool.
max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive.
),
timeout = httpx.Timeout(
pool = 120.0, # .... Time to wait for a free connection from the pool.
connect = 2.5, # ... Time to wait for establishing a connection to the server.
write = 10.0, # .... Time to wait for sending data.
read = 9.9 # ....... Time to wait for receiving data.
)
)
# Create an instance of a Mongo conn. to write action logs:
logs_mongo = AsyncMongo(
connection_string = r"mongodb://del.ditscentre.in:27017,wtt.ditscentre.in:27017,mum.arh.001.ditscentre.in:27017/admin?tls=true&tlsCAFile=%2Fetc%2Fssl%2Fdbu%2Fmongo_data_ca.pem&tlsCertificateKeyFile=%2Fetc%2Fssl%2Fdbu%2Fmongo_data_cert.pem&replicaSet=dits_mongod_rep&readPreference=primary&authMechanism=MONGODB-X509&authSource=%24external",
database_name = "converse",
max_connections = 10,
)
# Create an instance of a MikroTik client:
my_mikrotik = AsyncMikroTik(
http_client = my_http_client,
mikrotik_ip = "102.210.173.150",
username = "easyfi",
password = "easyfi",
use_https = False,
config_by = "easyfi",
action_log_conn = logs_mongo
)
async def 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_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_hotspot_profiles()
# print(f"HOTSPOT PROFILES ({len(api_response.data)}):", json.to_string(api_response.data))
#
# api_response = await my_mikrotik.list_hotspot_servers()
# print(f"HOTSPOT SERVERS ({len(api_response.data)}):", json.to_string(api_response.data))
#
api_response = await my_mikrotik.list_hotspot_walled_gardens()
print(f"HOTSPOT WALLED GARDENS ({len(api_response.data)}):", json.to_string(api_response.data))
#
# api_response = await my_mikrotik.list_hotspot_walled_garden_ips()
# print(f"HOTSPOT WALLED GARDEN IPS ({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))
#
# api_response = await my_mikrotik.list_firewall_nat()
# print(f"FIREWALL NAT ({len(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_PROFILE_NAME = "easyfi-hs-prf-" + HS_VLAN_ID
HS_SERVER_NAME = "easyfi-hs-srv-" + HS_VLAN_ID
HS_DHCP_SERVER_NAME = "easyfi-hs-dhcp-" + HS_VLAN_ID
HS_SNMP_COMMUNITY_NAME = "lkjhgfdsa1234567"
HS_PUBLIC_IP_SUBNET = "102.210.173.150/26"
HS_FIRST_PUBLIC_IP = "102.210.173.128"
HS_LAST_PUBLIC_IP = "102.210.173.191"
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",
# "name": HS_PROFILE_NAME,
# "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",
# "name": HS_SERVER_NAME,
# # "profile": "hs01.easyfi.net.in",
# "profile": HS_PROFILE_NAME,
# "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-A
# Set up the Walled-Garden:
for index, url in enumerate([
"api.thecaoffice.com",
"cdn.jsdelivr.net"
]):
api_response = await my_mikrotik.add_hotspot_walled_garden(
json_payload = {
"action": "allow",
"dst-host": url,
"comment": HS_COMMENT
}
)
print(f"STEP 9-A.{index}:", api_response.success, f"({api_response.action})")
print("MESSAGE:", api_response.message)
print("JSON:", api_response.data)
print("\n\n")
# # STEP 9-B:
# # 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-B:", 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:
# private_ips = [
# str(ipaddress.IPv4Address(_))
# for _ in range(
# int(ipaddress.IPv4Address(HS_FIRST_PRIVATE_IP)),
# int(ipaddress.IPv4Address(HS_LAST_PRIVATE_IP)) + 1
# )
# ]
# 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,
# consider_reserved_ips = False
# )
# print("NAT MAP:", json.to_string(nat_map, default=str))
# print("PRIVATE IP COUNT:", len(private_ips))
# print("PUBLIC IP COUNT:", len(public_ips))
# print("RULE COUNT:", len(nat_map))
# 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())
Reference in New Issue View Git Blame Copy Permalink