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

(20250210) Started working on MikroTik config automation.

Browse Source
This commit is contained in:
khushal
2025-02-10 19:35:28 +05:30
parent 95d90e807b
commit 94e5c81087
13 changed files with 789 additions and 836 deletions
Download Patch File Download Diff File Expand all files Collapse all files
models/api/software/auth.py
+24 -37
View File
@@ -36,9 +36,13 @@ sys.path.append(".")
sys.path.append("..")
# For making data behaviour_models:
from pydantic import BaseModel, Field, field_validator, PastDatetime
from pydantic import BaseModel, Field, field_validator, PastDatetime, model_validator
from typing import Optional, Literal, Union
# Other data models:
from models.software.tcaoff_ai.auth import TheCAOfficeAIAuth
from models.software.mikrotik.auth import MikroTikPPPoE1000Auth, MikroTikHotspot1000Auth
# My utils:
from utils_v2.string import regex
from utils_v2.date_time import date_time
@@ -75,40 +79,6 @@ REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]
# *****************************************************************************************************************
class TheCAOfficeAIAuth(BaseModel):
authorizeSoftware: bool = Field(
description = "god knows; don't ask",
frozen = True
)
authorizeMailMessaging: bool = Field(
description = "god knows; don't ask",
frozen = True
)
authorizeCompliancePortal: bool = Field(
description = "god knows; don't ask",
frozen = True
)
authorizeFinancialInstitution: bool = Field(
description = "god knows; don't ask",
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ---------------------------------------------------------------------------------------------------------------------
class SoftwareAuthRequestHeaders(BaseModel):
sessionToken: str = Field(
@@ -135,8 +105,8 @@ class SoftwareAuthRequestHeaders(BaseModel):
class SoftwareAuthRequestData(BaseModel):
softwareClient: Literal["theCaOfficeAi"] = Field(alias = "client")
auth: Union[TheCAOfficeAIAuth]
softwareClient: Literal["theCaOfficeAi", "mikrotikPPPoE1000", "mikrotikHotspot1000"] = Field(alias = "client")
auth: Union[TheCAOfficeAIAuth, MikroTikPPPoE1000Auth, MikroTikHotspot1000Auth]
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
@@ -146,6 +116,23 @@ class SoftwareAuthRequestData(BaseModel):
class Config:
extra = "forbid"
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@model_validator(mode = "after")
def ensure_harmony(cls, values):
client = values.softwareClient
auth = values.auth
harmony_map = {
"theCaOfficeAi": TheCAOfficeAIAuth,
"mikrotikPPPoE1000": MikroTikPPPoE1000Auth,
"mikrotikHotspot1000": MikroTikHotspot1000Auth
}
if not isinstance(auth, harmony_map[client]):
raise ValueError(f"incorrect 'auth' for selected client '{client}'")
return values
# *****************************************************************************************************************
# ***** ****
models/core/auth_token.py
+6 -6
View File
@@ -105,12 +105,12 @@ class CoreAuthTokenModel(BaseModel):
)
client: Literal[
"gmail", "outlook", # ............................. Mail Clients
"telegram", "whatsapp", "whatsappNimbus", # ....... Chat Clients
"nimbusSmsIndia", "savvyBulkSmsKenya", # .......... SMS Clients
"razorpay", "safaricomMPesaExpress", # ............ Payment Gateways
"zerodhaKite", "iciciBreeze", "paperTrading", # ... Stock Brokers
"theCaOfficeAi" # ................................. Software
"gmail", "outlook", # ........................................... Mail Clients
"telegram", "whatsapp", "whatsappNimbus", # ..................... Chat Clients
"nimbusSmsIndia", "savvyBulkSmsKenya", # ........................ SMS Clients
"razorpay", "safaricomMPesaExpress", # .......................... Payment Gateways
"zerodhaKite", "iciciBreeze", "paperTrading", # ................. Stock Brokers
"theCaOfficeAi", "mikrotikPPPoE1000", "mikrotikHotspot1000" # ... Software
] = Field(
description = "the third-part client that was used",
frozen = True
models/software/mikrotik/auth.py
+269 -15
View File
@@ -6,11 +6,11 @@
DATE:
Friday, 6th Dec., 2024.
Monday, 10th Feb., 2025.
OBJECTIVE:
To provide a structure to receive auth details of various chat apps (like Telegram and WhatsApp).
To provide a structure to receive auth details for MikroTik devices.
REFERENCES:
@@ -37,11 +37,12 @@ sys.path.append("..")
# For making data behaviour_models:
from pydantic import BaseModel, Field, field_validator, PastDatetime, model_validator
from typing import Optional, Literal, Union
from typing import Optional, Literal, Union, Any, List
# My utils:
from utils_v2.string import regex
from utils_v2.date_time import date_time
from utils_v2.network import ip
# To work with date and time:
import datetime
@@ -75,22 +76,118 @@ REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]
# *****************************************************************************************************************
class TelegramAuth(BaseModel):
def parse_vlans(vlan_range_str) -> List[int]:
botId: str = Field(
description = "The id of the bot (that can be invoked with '@').",
"""
Converts VLAN ids from a string representation to an array of integers.
:param vlan_range_str: A string like ""0,1,2,10-15,20".
:return: A list like [0, 1, 2, 10, 11, 12, 13, 14, 15, 20].
"""
# Start with an empty array:
result = []
# Split the string by commas:
parts = vlan_range_str.replace(" ", "").split(",")
# Loop through each part
for part in parts:
# Check if the part is a range (contains a hyphen):
if "-" in part:
# Add all numbers in the range to the result list:
start, end = part.split("-")
result.extend(range(int(start), int(end) + 1))
# Add the single number to the result list:
else: result.append(int(part))
# Done here:
result = sorted(result)
print(result)
return result
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class MikroTikPPPoE1000Auth(BaseModel):
siteName: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
botName: str = Field(
description = "The display name of the bot.",
nasIp: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
botToken: str = Field(
description = "the token granted by BotFather",
nasPort: int | None = Field(
description = "Don't know, and don't want to know.",
frozen = True,
default = None
)
radius: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
secret: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
username: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
password: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
location: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
snmpCommunity: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
vlanRange: List[int] = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
privateIpPool: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True,
default = "100.64.0.0/22",
validate_default = True
)
publicIpPool: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
@@ -103,20 +200,124 @@ class TelegramAuth(BaseModel):
class Config:
extra = "forbid"
# ┏┓ •
# ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
# ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
# ┛
@property
def firstPrivateIp(self) -> str:
return self.privateIpPool.split("-")[0]
@property
def lastPrivateIp(self) -> str:
return self.privateIpPool.split("-")[1]
@property
def firstPublicIp(self) -> str:
return self.publicIpPool.split("-")[0]
@property
def lastPublicIp(self) -> str:
return self.publicIpPool.split("-")[1]
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@field_validator("nasPort", mode = "before")
def validate_port_no(cls, value):
if isinstance(value, (str, float)): value = int(value)
return value
@field_validator("privateIpPool", "publicIpPool", mode = "before")
def validate_ip_range(cls, value):
return ip.to_hyphen_notation(value)
@field_validator("vlanRange", mode = "before")
def validate_vlans(cls, value):
return parse_vlans(value)
# ---------------------------------------------------------------------------------------------------------------------
class WhatsAppNimbusAuth(BaseModel):
class MikroTikHotspot1000Auth(BaseModel):
apiKey: str = Field(
description = "???",
siteName: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
senderId: str = Field(
description = "???",
nasIp: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
nasPort: int | None = Field(
description = "Don't know, and don't want to know.",
frozen = True,
default = None
)
radius: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
secret: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
username: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
password: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
location: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
snmpCommunity: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
vlanRange: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
privateIpPool: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
publicIpPool: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
domainName: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
@@ -129,6 +330,59 @@ class WhatsAppNimbusAuth(BaseModel):
class Config:
extra = "forbid"
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@field_validator("nasPort", mode = "before")
def validate_port_no(cls, value):
if isinstance(value, (str, float)): value = int(value)
return value
# ---------------------------------------------------------------------------------------------------------------------
class MikroTikAuthResponse(BaseModel):
success: bool = Field(
description = "To indicate whether or not, the action was a success",
frozen = False,
default = False
)
message: str = Field(
description = "To explain what happened in the process of handling the OAuth callback.",
frozen = False,
default = "ERR: Message not captured."
)
exception: Any = Field(
description = "To pass on any exception that occurred in the process.",
frozen = False,
default = None
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ┏┓ ┏┓
# ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏
# ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛
pass
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
pass
# *****************************************************************************************************************
# ***** ****
models/software/mikrotik/configure.py
+19 -159
View File
@@ -10,7 +10,7 @@
OBJECTIVE:
To provide a structure to receive auth details for MikroTik devices.
Maa chude, yaar. Ab fark nahin padtaa.
REFERENCES:
@@ -75,68 +75,26 @@ REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]
# *****************************************************************************************************************
class MikroTikPPPoE1000Auth(BaseModel):
class MikroTikConfigAttemptResponse(BaseModel):
siteName: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
success: bool = Field(
description = "To indicate whether or not, the action was a success",
frozen = False,
default = False
)
nasIp: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
message: str = Field(
description = "To explain what happened in the process of handling the OAuth callback.",
frozen = False,
default = "ERR: Message not captured."
)
nasPort: int | None = Field(
description = "Don't know, and don't want to know.",
frozen = True,
exception: Any = Field(
description = "To pass on any exception that occurred in the process.",
frozen = False,
default = None
)
radius: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
secret: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
username: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
password: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
location: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
snmpCommunity: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
publicIpPool: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
@@ -145,121 +103,23 @@ class MikroTikPPPoE1000Auth(BaseModel):
class Config:
extra = "forbid"
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
# ┏┓ ┏┓
# ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏
# ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛
@field_validator("nasPort", mode = "before")
def validate_port_no(cls, value):
if isinstance(value, (str, float)): value = int(value)
return value
# ---------------------------------------------------------------------------------------------------------------------
class MikroTikHotspot1000Auth(BaseModel):
siteName: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
nasIp: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
nasPort: int | None = Field(
description = "Don't know, and don't want to know.",
frozen = True,
default = None
)
radius: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
secret: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
username: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
password: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
location: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
snmpCommunity: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
vlanRange: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
privateIpPool: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
publicIpPool: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
domainName: str = Field(
description = "Don't know, and don't want to know.",
min_length = 1,
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
pass
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@field_validator("nasPort", mode = "before")
def validate_port_no(cls, value):
if isinstance(value, (str, float)): value = int(value)
return value
pass
# ---------------------------------------------------------------------------------------------------------------------
class MikroTikAuthResponse(BaseModel):
class MikroTikRollBackAttemptResponse(BaseModel):
success: bool = Field(
description = "To indicate whether or not, the action was a success",
models/software/tcaoff_ai/auth.py
+1 -33
View File
@@ -10,7 +10,7 @@
OBJECTIVE:
To provide a structure to receive auth details for MikroTik devices.
Don't ask. I don't know. We prefer speed over quality and certainty.
REFERENCES:
@@ -106,38 +106,6 @@ class TheCAOfficeAIAuth(BaseModel):
extra = "forbid"
# ---------------------------------------------------------------------------------------------------------------------
class MikroTikAuth(BaseModel):
botId: str = Field(
description = "The id of the bot (that can be invoked with '@').",
min_length = 1,
frozen = True
)
botName: str = Field(
description = "The display name of the bot.",
min_length = 1,
frozen = True
)
botToken: str = Field(
description = "the token granted by BotFather",
min_length = 1,
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***