2657 lines
97 KiB
Python
2657 lines
97 KiB
Python
"""
|
|
|
|
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())
|