Files

166 lines
5.5 KiB
Python

"""
AUTHOR:
Khushal P Soonderji
DATE:
Sunday, 9th Feb, 2025.
OBJECTIVE:
To provide an easy way to configure MikroTik devices remotely.
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 io
import os
# My utils:
from utils_v2.string import json
# For SSH:
from paramiko import SSHClient, AutoAddPolicy
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
def ssh_exec_one(ssh: SSHClient, command: str):
# Start by assuming failure:
result = {
"stdin": None,
"stdout": None,
"stderr": None,
"returnCode": -999,
"exception": None
}
try:
# Run the command:
stdin, stdout, stderr = ssh.exec_command(command)
print(stdin)
# Extract the results:
result["stdin"] = stdin.read().decode("utf8")
result["stdout"] = stdout.read().decode("utf8")
result["stderr"] = stderr.read().decode("utf8")
result["returnCode"] = stdout.channel.recv_exit_status()
# Close the file-like objects:
stdin.close()
stdout.close()
stderr.close()
# If something goes wrong:
except Exception as exception:
result["exception"] = exception
raise exception
# Done here:
return result
# ---------------------------------------------------------------------------------------------------------------------
def ssh_test(config: dict):
# Initialize the object:
ssh = SSHClient()
# ssh.load_host_keys("~/.ssh/known_hosts")
ssh.load_system_host_keys()
ssh.set_missing_host_key_policy(AutoAddPolicy())
# Log in to the SSH client:
ssh.connect(
hostname = config["nasIp"],
port = 19991,
username = config["username"],
password = config["password"]
)
# Execute the needed commands:
result = ssh_exec_one(ssh, "hostname")
print("EXEC ONE JSON:", json.to_string(result, default = str))
# Close the connection:
ssh.close()
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
# The expected configuration structure from the database:
sample_config_json = {
"siteName": "Hirani Telcom",
# "nasIp": "102.210.175.251",
"nasIp": "wtt.ditscentre.in",
"radius": "102.210.175.242",
"secret": "abcdefgh",
# "username": "easyfi",
"username": "chaalu",
# "password": "easyfi",
"password": "hellochaalu",
"location": "Kenya",
"snmpCommunity": "abcdefgh"
}
ssh_test(config = sample_config_json)