From 962dc45f62c31d1673a97b280d569a2ebe704792 Mon Sep 17 00:00:00 2001 From: khushal Date: Wed, 19 Feb 2025 12:48:18 +0530 Subject: [PATCH] (20250219) Added a test API to quickly check if a mikrotik device has it's API access enabled. --- api/blueprints/common/session_token.py | 209 ++++++++++++++++++ .../test/mikrotik_api_access_check.py | 170 ++++++++++++++ api/main.py | 4 + models/api/common/session_token.py | 138 ++++++++++++ 4 files changed, 521 insertions(+) create mode 100644 api/blueprints/common/session_token.py create mode 100644 api/blueprints/test/mikrotik_api_access_check.py create mode 100644 models/api/common/session_token.py diff --git a/api/blueprints/common/session_token.py b/api/blueprints/common/session_token.py new file mode 100644 index 0000000..ee496b0 --- /dev/null +++ b/api/blueprints/common/session_token.py @@ -0,0 +1,209 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Wednesday, 19th Feb., 2025. + + OBJECTIVE: + + To generate a session token when requested by admin accounts. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + + NOTES: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# For using Quart: +from quart import Blueprint, current_app, request + +# My utils: +from utils_v2.string import json +from utils_v2.api.codes import StatusCodes, HttpCodes +from utils_v2.api.response import ResponseModel +from utils_v2.api.async_quart import ( + set_api_version, + read_input, + get_session_info, + log_request_to_mongo, + log_chain_to_mongo, + should_not_be_under_maintenance, + only_whitelisted_ips, + limit_rate, + validate_input, + handle_cancelled_request +) + +# Common: +from shared import constants + +# Data Models: +from models.core.user import CoreUserInfoModel +from models.api.common.session_token import GenerateSessionTokenRequestHeaders, GenerateSessionTokenRequestData + +# Helpers: +from api.helpers.user import token_check + +# To work with MongoDB: +from bson import ObjectId + +# To generate UUIDs: +import uuid + +# For asynchronous activities: +import asyncio + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# Related to Quart: +session_token_bp = Blueprint("session_tok_gen", __name__) + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +@session_token_bp.record_once +def init(blueprint_setup_state): + + # This gets called when the blueprint is registered. + # Consider this to be a one-time setup for the whole blueprint: + pass + + +# --------------------------------------------------------------------------------------------------------------------- + + +@session_token_bp.route("/session", methods = ["POST"]) +@set_api_version(api_version = "1.0.0") +@read_input(sanitize_headers = False, sanitize_data = False) +@get_session_info(key = "X-Session-Token", session_coro = "get_session") +@log_request_to_mongo( + attr_name = "logs_mongo", + project = constants.PROJECT_NAME, + log_type = constants.MODULE_NAME, + operation = "sTokApi", + log_input = True, + log_output = True, + sensitive_keys = ["sessionToken", "X-Session-Token", "tokenKey"] +) +@log_chain_to_mongo(attr_name = "logs_mongo") +@should_not_be_under_maintenance(attr_name = "is_under_maintenance") +@validate_input( + header_validator = lambda x: GenerateSessionTokenRequestHeaders(**x).model_dump(), + data_validator = lambda x: GenerateSessionTokenRequestData(**x) +) +@handle_cancelled_request() +async def disable_auth_token( + inbound_headers: dict | GenerateSessionTokenRequestHeaders = None, + inbound_data: dict | GenerateSessionTokenRequestData = None, + inbound_files: dict = None, + **kwargs +): + + """ + Use this for God knows what. + :param inbound_headers: auto-extracted by the decorators. + :param inbound_data: auto-extracted by the decorators. + :param inbound_files: auto-extracted by the decorators. + :param kwargs: Any number of extra inputs supplied by the decorators. + :return: A standard response structure. + """ + + # ┏┓ ┓ ┏┓┓ ┓ + # ┣┫┓┏╋┣┓ ┃ ┣┓┏┓┏┃┏ + # ┛┗┗┻┗┛┗ ┗┛┛┗┗ ┗┛┗ + + # If the session token is invalid/expired: + if kwargs.get("session_info") is None: + return constants.API_RESPONSE_UNAUTHORIZED + + # Get the user's info: + user_info = CoreUserInfoModel(**kwargs.get("session_info")) + + # The user requesting this action needs to be an "Owner": + if user_info.role.strip().lower() != "owner": + return ResponseModel( + status_code = StatusCodes.FAILED, + http_code = HttpCodes.UNAUTHORIZED, + message = "Only owners can perform this action." + ) + + # ┏┓ ┏┓ • + # ┃┓┏┓┏┓┏┓┏┓┏┓╋┏┓ ┗┓┏┓┏┏┓┏┓┏┓ + # ┗┛┗ ┛┗┗ ┛ ┗┻┗┗ ┗┛┗ ┛┛┗┗┛┛┗ + + # Generate a UUID: + session_token = str(uuid.uuid4()) + + # Save it to Redis: + success = await current_app.module_cache.set( + key = session_token, + value = None, + expiry = inbound_data.expiry or 86_400 + ) + + # Done here: + return ResponseModel( + status_code = StatusCodes.OK if success else StatusCodes.FAILED, + http_code = HttpCodes.SUCCESS if success else HttpCodes.INTERNAL_SERVER_ERROR, + data = { + "sessionToken": session_token + } + ) + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/api/blueprints/test/mikrotik_api_access_check.py b/api/blueprints/test/mikrotik_api_access_check.py new file mode 100644 index 0000000..3e00b9d --- /dev/null +++ b/api/blueprints/test/mikrotik_api_access_check.py @@ -0,0 +1,170 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Wednesday, 19th Feb., 2025. + + OBJECTIVE: + + For simply testing whether, or not, a MikroTik device can be accessed via API. + NOT TO BE USED FOR LONG. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + + NOTES: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# For using Quart: +from quart import Blueprint, current_app, request, render_template + +# My utils: +from utils_v2.string import json +from utils_v2.mikrotik.controllers.async_mikrotik import AsyncMikroTik +from utils_v2.api.codes import StatusCodes, HttpCodes +from utils_v2.api.response import ResponseModel +from utils_v2.api.async_quart import ( + set_api_version, + read_input, + get_session_info, + log_request_to_mongo, + log_chain_to_mongo, + should_not_be_under_maintenance, + only_whitelisted_ips, + limit_rate, + validate_input, + handle_cancelled_request +) + +# Common: +from shared import constants + +# For asynchronous activities: +import asyncio + +# For random choices: +import random + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# Related to Quart: +test_mikrotik_api_bp = Blueprint("test_mt_api", __name__) + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +@test_mikrotik_api_bp.record_once +def init(blueprint_setup_state): + + # This gets called when the blueprint is registered. + # Consider this to be a one-time setup for the whole blueprint: + pass + + +# --------------------------------------------------------------------------------------------------------------------- + + +@test_mikrotik_api_bp.route("/mikrotik/api", methods = ["GET", "POST", "PUT", "PATCH", "DELETE"]) +@set_api_version(api_version = "1.0.0") +@read_input(sanitize_headers = False, sanitize_data = False) +@get_session_info(key = "X-Session-Token", session_coro = "get_session") +@log_request_to_mongo( + attr_name = "logs_mongo", + project = constants.PROJECT_NAME, + log_type = constants.MODULE_NAME, + operation = "testMikroTikRestApi", + log_input = True, + log_output = True, + sensitive_keys = ["sessionToken", "X-Session-Token"] +) +@log_chain_to_mongo(attr_name = "logs_mongo") +@should_not_be_under_maintenance(attr_name = "is_under_maintenance") +@handle_cancelled_request() +async def callback_test( + inbound_headers: dict = None, + inbound_data: dict = None, + inbound_files: dict = None, + **kwargs +): + + # Create a client to communicate with the MikrotikDevice + mikrotik_client = AsyncMikroTik( + config_by = "easyfi", + mikrotik_ip = inbound_data.get("ip"), + username = inbound_data.get("u", "easyfi"), + password = inbound_data.get("p", "easyfi"), + port = None, + use_https = False, + http_client = None, + action_log_conn = current_app.data_mongo + ) + + # Check if the system's resources can get listed. + # This becomes a great quick test for API access: + response = await mikrotik_client.list_system_resources(use_https = False) + + # Done here: + return ResponseModel( + status_code = StatusCodes.OK if response.success else StatusCodes.FAILED, + http_code = HttpCodes.SUCCESS if response.success else HttpCodes.INTERNAL_SERVER_ERROR, + message = response.message, + data = response.data + ) + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/api/main.py b/api/main.py index 10ab2ac..96f4b3b 100644 --- a/api/main.py +++ b/api/main.py @@ -151,11 +151,13 @@ from api.blueprints.ai.llm.invoke import llm_invoke_bp # Common Blueprints: from api.blueprints.common.disable import auth_token_disable_bp +from api.blueprints.common.session_token import session_token_bp # Tech and Testing Blueprints: from api.blueprints.tech.chat_alerts import tech_chat_alert_bp from api.blueprints.test.callback import test_callback_bp from api.blueprints.test.mikrotik_roll_back import test_mikrotik_rollback_bp +from api.blueprints.test.mikrotik_api_access_check import test_mikrotik_api_bp # All the helpers: from api.helpers.user import session @@ -224,6 +226,7 @@ app.register_blueprint(trading_symbols_list_bp, url_prefix = f"/{MODULE_BASE}/fi # Common Blueprints: app.register_blueprint(auth_token_disable_bp, url_prefix = f"/{MODULE_BASE}") +app.register_blueprint(session_token_bp, url_prefix = f"/{MODULE_BASE}") # AI Blueprints: app.register_blueprint(llm_invoke_bp, url_prefix = f"/{MODULE_BASE}/ai") @@ -232,6 +235,7 @@ app.register_blueprint(llm_invoke_bp, url_prefix = f"/{MODULE_BASE}/ai") app.register_blueprint(tech_chat_alert_bp, url_prefix = f"/{MODULE_BASE}/tech/alert") app.register_blueprint(test_callback_bp, url_prefix = f"/{MODULE_BASE}/test") app.register_blueprint(test_mikrotik_rollback_bp, url_prefix = f"/{MODULE_BASE}/test") +app.register_blueprint(test_mikrotik_api_bp, url_prefix = f"/{MODULE_BASE}/test") # ***************************************************************************************************************** diff --git a/models/api/common/session_token.py b/models/api/common/session_token.py new file mode 100644 index 0000000..fe1bb7b --- /dev/null +++ b/models/api/common/session_token.py @@ -0,0 +1,138 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Wednesday, 19th Feb., 2025. + + OBJECTIVE: + + To provide a structure to allow admin users to generate a session token to be used elsewhere. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# For making data behaviour_models: +from pydantic import BaseModel, Field, field_validator, PastDatetime +from typing import Optional, Literal, List, Any + +# My utils: +from utils_v2.string import regex +from utils_v2.date_time import date_time + +# To work with MongoDB: +from bson.objectid import ObjectId + +# To work with date and time: +import datetime + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# RegEx Patterns: +REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$" + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +class GenerateSessionTokenRequestHeaders(BaseModel): + + sessionToken: str = Field( + description = "The session token of the user who is requesting the service.", + pattern = REGEX_SESSION_TOKEN, + frozen = True, + alias = "X-Session-Token" + ) + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "allow" + + def model_dump(self, *args, **kwargs): + return super().model_dump(*args, by_alias = True, **kwargs) + + +# --------------------------------------------------------------------------------------------------------------------- + + +class GenerateSessionTokenRequestData(BaseModel): + + expiry: int | float = Field( + description = "the no. of seconds till which the session must remain valid.", + frozen = True, + default = 86_400 + ) + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "forbid" + arbitrary_types_allowed = True + + # ┓┏ ┓• ┓ • + # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ + # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ + + pass + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass