mirror of
https://github.com/DaanSelen/meshbook.git
synced 2026-02-20 00:12:10 +00:00
Working version. Rewrite complete. Version 2 soon.
This commit is contained in:
Daan
31
examples/big_test.yaml
Normal file
31
examples/big_test.yaml
Normal file
@@ -0,0 +1,31 @@
|
||||
---
|
||||
name: Echo some text in the terminal of the device
|
||||
group: "Kubernetes"
|
||||
variables:
|
||||
- name: package_manager
|
||||
value: "apt"
|
||||
- name: google_dns
|
||||
value: "8.8.8.8"
|
||||
- name: "quad9_dns"
|
||||
value: "9.9.9.9"
|
||||
tasks:
|
||||
- name: refresh the cache
|
||||
command: "{{ package_manager }} update"
|
||||
|
||||
- name: display available upgrades
|
||||
command: "{{ package_manager }} list --upgradable"
|
||||
|
||||
- name: apply upgrades
|
||||
command: "{{ package_manager }} upgrade -y"
|
||||
|
||||
- name: cleanup remaining packages
|
||||
command: "{{ package_manager }} autoremove -y"
|
||||
|
||||
- name: run autoclean
|
||||
command: "{{ package_manager }} autoclean -y"
|
||||
|
||||
- name: Ping Google DNS
|
||||
command: "ping {{ google_dns }} -c 4"
|
||||
|
||||
- name: Ping Quad9 DNS
|
||||
command: "ping {{ quad9_dns }} -c 4"
|
||||
@@ -1,6 +1,6 @@
|
||||
---
|
||||
name: Ping Multiple Points
|
||||
group: "Temp-Agents"
|
||||
group: "Kubernetes"
|
||||
variables:
|
||||
- name: host1
|
||||
value: "1.1.1.1"
|
||||
|
||||
343
legacy/meshbook-legacy.py
Normal file
343
legacy/meshbook-legacy.py
Normal file
@@ -0,0 +1,343 @@
|
||||
#!/bin/python3
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
from base64 import b64encode
|
||||
from configparser import ConfigParser
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import yaml
|
||||
import websockets
|
||||
|
||||
sequence = 0
|
||||
response_counter = 0
|
||||
expected_responses = 0
|
||||
basic_ready_state = asyncio.Event()
|
||||
ready_for_next = asyncio.Event()
|
||||
global_list = []
|
||||
responses_dict = {}
|
||||
|
||||
class ScriptEndTrigger(Exception):
|
||||
"""Custom Exception to handle script termination events."""
|
||||
pass
|
||||
|
||||
class MeshbookUtilities:
|
||||
"""Helper utility functions for the Meshcaller application."""
|
||||
|
||||
@staticmethod
|
||||
def base64_encode(string: str) -> str:
|
||||
"""Encode a string in Base64 format."""
|
||||
return b64encode(string.encode('utf-8')).decode()
|
||||
|
||||
@staticmethod
|
||||
def get_target_ids(company: str = None, device: str = None) -> list:
|
||||
"""Retrieve target IDs based on company or device."""
|
||||
ids = []
|
||||
|
||||
for entry in global_list:
|
||||
nodes = entry.get('nodes', [])
|
||||
if company and not device:
|
||||
if entry.get('mesh_name') == company:
|
||||
ids.extend(node['node_id'] for node in nodes if node.get('powered_on'))
|
||||
elif device and not company:
|
||||
for node in nodes:
|
||||
if node['node_name'] == device and node.get('powered_on'):
|
||||
return [node['node_id']] # Immediate return for single device
|
||||
elif not company and not device:
|
||||
ids.extend(node['node_id'] for node in nodes if node.get('powered_on'))
|
||||
|
||||
return ids
|
||||
|
||||
|
||||
|
||||
@staticmethod
|
||||
def read_yaml(file_path: str) -> dict:
|
||||
"""Read a YAML file and return its content as a dictionary."""
|
||||
with open(file_path, 'r') as file:
|
||||
return yaml.safe_load(file)
|
||||
|
||||
@staticmethod
|
||||
def replace_placeholders(playbook) -> dict:
|
||||
# Convert 'variables' to a dictionary for quick lookup
|
||||
variables = {var["name"]: var["value"] for var in playbook.get("variables", [])}
|
||||
|
||||
# Traverse 'tasks' to replace placeholders
|
||||
for task in playbook.get("tasks", []):
|
||||
command = task.get("command", "")
|
||||
for var_name, var_value in variables.items():
|
||||
placeholder = f"{{{{ {var_name} }}}}" # Create the placeholder string like "{{ host1 }}"
|
||||
command = command.replace(placeholder, var_value) # Update the command string
|
||||
task["command"] = command # Save the updated command string
|
||||
return playbook
|
||||
|
||||
@staticmethod
|
||||
def translate_nodeids(batches_dict, global_list) -> dict:
|
||||
for batch_name, items in batches_dict.items():
|
||||
|
||||
for item in items: # Process each item in the batch
|
||||
node_id = item["nodeid"] # Get the nodeid field
|
||||
|
||||
real_name = None
|
||||
for company in global_list:
|
||||
for machine in company["nodes"]:
|
||||
if machine["node_id"] == node_id:
|
||||
real_name = machine["node_name"]
|
||||
break
|
||||
if real_name:
|
||||
break
|
||||
|
||||
# If found, replace the nodeid with the real node name, otherwise mark it as "Unknown Node"
|
||||
if real_name:
|
||||
item["nodeid"] = real_name
|
||||
|
||||
return batches_dict
|
||||
|
||||
class MeshbookWebsocket:
|
||||
"""Handles WebSocket connections and interactions."""
|
||||
|
||||
def __init__(self):
|
||||
self.meshsocket = None
|
||||
self.received_response_queue = asyncio.Queue()
|
||||
|
||||
async def ws_on_open(self):
|
||||
"""Called when WebSocket connection is established."""
|
||||
if not args.silent:
|
||||
print('Connection established.')
|
||||
|
||||
async def ws_on_close(self):
|
||||
"""Called when WebSocket connection is closed."""
|
||||
print('Connection closed by remote host.')
|
||||
raise ScriptEndTrigger("WebSocket connection closed.")
|
||||
|
||||
async def ws_on_message(self, message: str):
|
||||
"""Processes incoming WebSocket messages."""
|
||||
try:
|
||||
received_response = json.loads(message)
|
||||
await self.received_response_queue.put(received_response)
|
||||
except json.JSONDecodeError:
|
||||
print("Error processing:", message)
|
||||
raise ScriptEndTrigger("Failed to decode JSON message.")
|
||||
|
||||
async def ws_send_data(self, message: str):
|
||||
"""Send data to the WebSocket server."""
|
||||
if not self.meshsocket:
|
||||
raise ScriptEndTrigger("WebSocket connection not established. Unable to send data.")
|
||||
if not args.silent:
|
||||
print('Sending data to the server.')
|
||||
await self.meshsocket.send(message)
|
||||
|
||||
async def gen_simple_list(self):
|
||||
"""Send requests to retrieve mesh and node lists."""
|
||||
await self.ws_send_data(json.dumps({'action': 'meshes', 'responseid': 'meshctrl'}))
|
||||
await self.ws_send_data(json.dumps({'action': 'nodes', 'responseid': 'meshctrl'}))
|
||||
|
||||
async def ws_handler(self, uri: str, username: str, password: str):
|
||||
"""Main WebSocket connection handler."""
|
||||
login_string = f'{MeshbookUtilities.base64_encode(username)},{MeshbookUtilities.base64_encode(password)}'
|
||||
ws_headers = {
|
||||
'User-Agent': 'MeshCentral API client',
|
||||
'x-meshauth': login_string
|
||||
}
|
||||
if not args.silent:
|
||||
print("Attempting WebSocket connection...")
|
||||
|
||||
try:
|
||||
async with websockets.connect(uri, additional_headers=ws_headers) as meshsocket:
|
||||
self.meshsocket = meshsocket
|
||||
await self.ws_on_open()
|
||||
await self.gen_simple_list()
|
||||
|
||||
while True:
|
||||
try:
|
||||
message = await meshsocket.recv()
|
||||
await self.ws_on_message(message)
|
||||
except websockets.ConnectionClosed:
|
||||
await self.ws_on_close()
|
||||
break
|
||||
except ScriptEndTrigger as e:
|
||||
print(f"WebSocket handler terminated: {e}")
|
||||
except Exception as e:
|
||||
print(f"An error occurred: {e}")
|
||||
|
||||
|
||||
class MeshbookProcessor:
|
||||
"""Processes data received from the WebSocket."""
|
||||
|
||||
def __init__(self):
|
||||
self.basic_temp_list = []
|
||||
|
||||
def handle_basic_data(self, data):
|
||||
"""Handles basic data from the server."""
|
||||
if not args.silent:
|
||||
print("Processing received basic data...")
|
||||
|
||||
self.basic_temp_list.append(data)
|
||||
if len(self.basic_temp_list) < 2:
|
||||
return
|
||||
|
||||
temp_dict = {}
|
||||
for entry in self.basic_temp_list:
|
||||
if isinstance(entry, list):
|
||||
for mesh in entry:
|
||||
if mesh.get("type") == "mesh":
|
||||
mesh_id = mesh["_id"]
|
||||
temp_dict[mesh_id] = {
|
||||
"mesh_name": mesh.get("name", "Unknown Mesh"),
|
||||
"mesh_desc": mesh.get("desc", "No description"),
|
||||
"nodes": []
|
||||
}
|
||||
elif isinstance(entry, dict):
|
||||
for mesh_id, nodes in entry.items():
|
||||
if mesh_id in temp_dict:
|
||||
temp_dict[mesh_id]["nodes"].extend(nodes)
|
||||
else:
|
||||
temp_dict[mesh_id] = {
|
||||
"mesh_name": "Unknown Mesh",
|
||||
"mesh_desc": "No description",
|
||||
"nodes": nodes
|
||||
}
|
||||
|
||||
for mesh_id, details in temp_dict.items():
|
||||
global_list.append({
|
||||
"mesh_name": details["mesh_name"],
|
||||
"mesh_id": mesh_id,
|
||||
"nodes": [
|
||||
{
|
||||
"node_id": node["_id"],
|
||||
"node_name": node.get("name", "Unknown Node"),
|
||||
"powered_on": node.get("pwr") == 1
|
||||
}
|
||||
for node in details["nodes"]
|
||||
]
|
||||
})
|
||||
basic_ready_state.set()
|
||||
ready_for_next.set()
|
||||
|
||||
async def receive_processor(self, python_client: MeshbookWebsocket):
|
||||
"""Processes messages received from the WebSocket."""
|
||||
global response_counter
|
||||
temp_responses_list = []
|
||||
|
||||
while True:
|
||||
message = await python_client.received_response_queue.get()
|
||||
action_type = message.get('action')
|
||||
if action_type in ('meshes', 'nodes'):
|
||||
self.handle_basic_data(message[action_type])
|
||||
elif action_type == 'msg':
|
||||
temp_responses_list.append(message)
|
||||
|
||||
response_counter += 1 # Increment response counter
|
||||
|
||||
if not args.silent or args.information:
|
||||
print("Current Batch: {}".format(math.ceil(response_counter/len(target_ids))))
|
||||
print("Current response number: {}".format(response_counter))
|
||||
print("Current Calculation: {} % {} = {}".format(response_counter, len(target_ids), response_counter % len(target_ids)))
|
||||
|
||||
if response_counter % len(target_ids) == 0:
|
||||
batch_name = "Batch {}".format(math.ceil(response_counter / len(target_ids)))
|
||||
responses_dict[batch_name] = temp_responses_list
|
||||
temp_responses_list = []
|
||||
ready_for_next.set()
|
||||
elif action_type == 'close':
|
||||
print(message)
|
||||
elif not args.silent:
|
||||
print("Ignored action:", action_type)
|
||||
|
||||
|
||||
class MeshcallerActions:
|
||||
"""Processes playbook actions."""
|
||||
|
||||
@staticmethod
|
||||
async def process_arguments(python_client: MeshbookWebsocket, playbook_yaml: dict):
|
||||
"""Executes tasks defined in the playbook."""
|
||||
global response_counter, expected_responses, target_ids
|
||||
|
||||
await basic_ready_state.wait() # Wait for the basic data to be ready
|
||||
|
||||
target_ids = MeshbookUtilities.get_target_ids(
|
||||
company=playbook_yaml.get('company'),
|
||||
device=playbook_yaml.get('device')
|
||||
)
|
||||
if not target_ids:
|
||||
raise ScriptEndTrigger("No targets found.")
|
||||
|
||||
run_command_template = {
|
||||
'action': 'runcommands',
|
||||
'nodeids': target_ids,
|
||||
'type': 0,
|
||||
'cmds': None,
|
||||
'runAsUser': 0,
|
||||
'responseid': 'meshctrl',
|
||||
'reply': True
|
||||
}
|
||||
|
||||
expected_responses = len(playbook_yaml['tasks']) * len(target_ids) # Calculate the total expected responses: tasks x target nodes
|
||||
|
||||
# Send commands for all nodes at once
|
||||
for task in playbook_yaml['tasks']:
|
||||
await ready_for_next.wait()
|
||||
run_command_template["cmds"] = task['command']
|
||||
run_command_template["nodeids"] = target_ids # Send to all target IDs at once
|
||||
|
||||
if not args.silent or args.information:
|
||||
print("-=-" * 40)
|
||||
print("Running task:", task)
|
||||
print("-=-" * 40)
|
||||
|
||||
# Send the command to all nodes in one go
|
||||
await python_client.ws_send_data(json.dumps(run_command_template))
|
||||
ready_for_next.clear()
|
||||
|
||||
# Wait until all expected responses are received
|
||||
while response_counter < expected_responses:
|
||||
await asyncio.sleep(1)
|
||||
|
||||
# Exit gracefully
|
||||
if not args.silent or args.information:
|
||||
print("-=-" * 40)
|
||||
|
||||
updated_response_dict = MeshbookUtilities.translate_nodeids(responses_dict, global_list)
|
||||
|
||||
if not args.nojson:
|
||||
print(json.dumps(updated_response_dict,indent=4))
|
||||
raise ScriptEndTrigger("All tasks completed successfully: Expected {} Received {}".format(expected_responses, response_counter))
|
||||
|
||||
|
||||
async def main():
|
||||
parser = argparse.ArgumentParser(description="Process command-line arguments")
|
||||
parser.add_argument("-pb", "--playbook", type=str, help="Path to the playbook file.", required=True)
|
||||
|
||||
parser.add_argument("--conf", type=str, help="Path for the API configuration file (default: ./api.conf).")
|
||||
parser.add_argument("--nojson", action="store_true", help="Makes the program not output the JSON response data.")
|
||||
parser.add_argument("-s", "--silent", action="store_true", help="Suppress terminal output.")
|
||||
parser.add_argument("-i", "--information", action="store_true", help="Add the calculations and other informational data to the output.")
|
||||
|
||||
global args
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
credentials = MeshbookUtilities.load_config(args.conf)
|
||||
python_client = MeshbookWebsocket()
|
||||
processor = MeshbookProcessor()
|
||||
|
||||
websocket_task = asyncio.create_task(python_client.ws_handler(
|
||||
credentials['websocket_url'],
|
||||
credentials['username'],
|
||||
credentials['password']
|
||||
))
|
||||
processor_task = asyncio.create_task(processor.receive_processor(python_client))
|
||||
|
||||
playbook_yaml = MeshbookUtilities.read_yaml(args.playbook)
|
||||
translated_playbook = MeshbookUtilities.replace_placeholders(playbook_yaml)
|
||||
await MeshcallerActions.process_arguments(python_client, translated_playbook)
|
||||
|
||||
await asyncio.gather(websocket_task, processor_task)
|
||||
|
||||
except ScriptEndTrigger as e:
|
||||
if not args.silent or args.information:
|
||||
print(e)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -6,337 +6,177 @@ from base64 import b64encode
|
||||
from configparser import ConfigParser
|
||||
import json
|
||||
import math
|
||||
import meshctrl
|
||||
import os
|
||||
import yaml
|
||||
import websockets
|
||||
|
||||
sequence = 0
|
||||
response_counter = 0
|
||||
expected_responses = 0
|
||||
basic_ready_state = asyncio.Event()
|
||||
ready_for_next = asyncio.Event()
|
||||
global_list = []
|
||||
responses_dict = {}
|
||||
'''
|
||||
Script utilities are handled in the following section.
|
||||
'''
|
||||
|
||||
class ScriptEndTrigger(Exception):
|
||||
"""Custom Exception to handle script termination events."""
|
||||
pass
|
||||
|
||||
class MeshbookUtilities:
|
||||
"""Helper utility functions for the Meshcaller application."""
|
||||
def output_text(message: str, required=False):
|
||||
if required:
|
||||
print(message)
|
||||
elif not args.silent:
|
||||
print(message)
|
||||
|
||||
@staticmethod
|
||||
def base64_encode(string: str) -> str:
|
||||
"""Encode a string in Base64 format."""
|
||||
return b64encode(string.encode('utf-8')).decode()
|
||||
async def load_config(conf_file: str = './api.conf', segment: str = 'meshcentral-account') -> ConfigParser:
|
||||
if not os.path.exists(conf_file):
|
||||
raise ScriptEndTrigger(f'Missing config file {conf_file}. Provide an alternative path.')
|
||||
|
||||
@staticmethod
|
||||
def get_target_ids(company: str = None, device: str = None) -> list:
|
||||
"""Retrieve target IDs based on company or device."""
|
||||
ids = []
|
||||
config = ConfigParser()
|
||||
try:
|
||||
config.read(conf_file)
|
||||
except Exception as err:
|
||||
raise ScriptEndTrigger(f"Error reading configuration file '{conf_file}': {err}")
|
||||
|
||||
for entry in global_list:
|
||||
nodes = entry.get('nodes', [])
|
||||
if company and not device:
|
||||
if entry.get('mesh_name') == company:
|
||||
ids.extend(node['node_id'] for node in nodes if node.get('powered_on'))
|
||||
elif device and not company:
|
||||
for node in nodes:
|
||||
if node['node_name'] == device and node.get('powered_on'):
|
||||
return [node['node_id']] # Immediate return for single device
|
||||
elif not company and not device:
|
||||
ids.extend(node['node_id'] for node in nodes if node.get('powered_on'))
|
||||
if segment not in config:
|
||||
raise ScriptEndTrigger(f'Segment "{segment}" not found in config file {conf_file}.')
|
||||
|
||||
return ids
|
||||
return config[segment]
|
||||
|
||||
async def init_connection(credentials: dict) -> meshctrl.Session:
|
||||
session = meshctrl.Session(
|
||||
credentials['websocket_url'],
|
||||
user=credentials['username'],
|
||||
password=credentials['password']
|
||||
)
|
||||
await session.initialized.wait()
|
||||
return session
|
||||
|
||||
async def translate_id_to_name(target_id: str) -> str:
|
||||
for group in group_list:
|
||||
for device in group_list[group]:
|
||||
if device["device_id"] == target_id:
|
||||
return device["device_name"]
|
||||
|
||||
@staticmethod
|
||||
def read_yaml(file_path: str) -> dict:
|
||||
"""Read a YAML file and return its content as a dictionary."""
|
||||
with open(file_path, 'r') as file:
|
||||
return yaml.safe_load(file)
|
||||
'''
|
||||
Creation and compilation happends in the following section, where the yaml gets read in, and edited accordingly.
|
||||
'''
|
||||
|
||||
@staticmethod
|
||||
def replace_placeholders(playbook) -> dict:
|
||||
# Convert 'variables' to a dictionary for quick lookup
|
||||
async def compile_book(playbook_file: dict) -> dict:
|
||||
playbook = open(playbook_file, 'r')
|
||||
playbook = await replace_placeholders(yaml.safe_load(playbook))
|
||||
return playbook
|
||||
|
||||
async def replace_placeholders(playbook: dict) -> dict:
|
||||
variables = {var["name"]: var["value"] for var in playbook.get("variables", [])}
|
||||
|
||||
# Traverse 'tasks' to replace placeholders
|
||||
for task in playbook.get("tasks", []):
|
||||
command = task.get("command", "")
|
||||
for var_name, var_value in variables.items():
|
||||
placeholder = f"{{{{ {var_name} }}}}" # Create the placeholder string like "{{ host1 }}"
|
||||
command = command.replace(placeholder, var_value) # Update the command string
|
||||
task["command"] = command # Save the updated command string
|
||||
command = command.replace(placeholder, var_value)
|
||||
task["command"] = command
|
||||
return playbook
|
||||
|
||||
@staticmethod
|
||||
def translate_nodeids(batches_dict, global_list) -> dict:
|
||||
for batch_name, items in batches_dict.items():
|
||||
'''
|
||||
Creation and compilation of the MeshCentral nodes list (list of all nodes available to the user in the configuration) is handled in the following section.
|
||||
'''
|
||||
|
||||
for item in items: # Process each item in the batch
|
||||
node_id = item["nodeid"] # Get the nodeid field
|
||||
async def compile_group_list(session: meshctrl.Session) -> dict:
|
||||
devices_response = await session.list_devices(details=False, timeout=10)
|
||||
|
||||
real_name = None
|
||||
for company in global_list:
|
||||
for machine in company["nodes"]:
|
||||
if machine["node_id"] == node_id:
|
||||
real_name = machine["node_name"]
|
||||
break
|
||||
if real_name:
|
||||
break
|
||||
local_device_list = {}
|
||||
for device in devices_response:
|
||||
if device.meshname not in local_device_list:
|
||||
local_device_list[device.meshname] = []
|
||||
|
||||
# If found, replace the nodeid with the real node name, otherwise mark it as "Unknown Node"
|
||||
if real_name:
|
||||
item["nodeid"] = real_name
|
||||
|
||||
return batches_dict
|
||||
|
||||
class MeshbookWebsocket:
|
||||
"""Handles WebSocket connections and interactions."""
|
||||
|
||||
def __init__(self):
|
||||
self.meshsocket = None
|
||||
self.received_response_queue = asyncio.Queue()
|
||||
|
||||
async def ws_on_open(self):
|
||||
"""Called when WebSocket connection is established."""
|
||||
if not args.silent:
|
||||
print('Connection established.')
|
||||
|
||||
async def ws_on_close(self):
|
||||
"""Called when WebSocket connection is closed."""
|
||||
print('Connection closed by remote host.')
|
||||
raise ScriptEndTrigger("WebSocket connection closed.")
|
||||
|
||||
async def ws_on_message(self, message: str):
|
||||
"""Processes incoming WebSocket messages."""
|
||||
try:
|
||||
received_response = json.loads(message)
|
||||
await self.received_response_queue.put(received_response)
|
||||
except json.JSONDecodeError:
|
||||
print("Error processing:", message)
|
||||
raise ScriptEndTrigger("Failed to decode JSON message.")
|
||||
|
||||
async def ws_send_data(self, message: str):
|
||||
"""Send data to the WebSocket server."""
|
||||
if not self.meshsocket:
|
||||
raise ScriptEndTrigger("WebSocket connection not established. Unable to send data.")
|
||||
if not args.silent:
|
||||
print('Sending data to the server.')
|
||||
await self.meshsocket.send(message)
|
||||
|
||||
async def gen_simple_list(self):
|
||||
"""Send requests to retrieve mesh and node lists."""
|
||||
await self.ws_send_data(json.dumps({'action': 'meshes', 'responseid': 'meshctrl'}))
|
||||
await self.ws_send_data(json.dumps({'action': 'nodes', 'responseid': 'meshctrl'}))
|
||||
|
||||
async def ws_handler(self, uri: str, username: str, password: str):
|
||||
"""Main WebSocket connection handler."""
|
||||
login_string = f'{MeshbookUtilities.base64_encode(username)},{MeshbookUtilities.base64_encode(password)}'
|
||||
ws_headers = {
|
||||
'User-Agent': 'MeshCentral API client',
|
||||
'x-meshauth': login_string
|
||||
}
|
||||
if not args.silent:
|
||||
print("Attempting WebSocket connection...")
|
||||
|
||||
try:
|
||||
async with websockets.connect(uri, additional_headers=ws_headers) as meshsocket:
|
||||
self.meshsocket = meshsocket
|
||||
await self.ws_on_open()
|
||||
await self.gen_simple_list()
|
||||
|
||||
while True:
|
||||
try:
|
||||
message = await meshsocket.recv()
|
||||
await self.ws_on_message(message)
|
||||
except websockets.ConnectionClosed:
|
||||
await self.ws_on_close()
|
||||
break
|
||||
except ScriptEndTrigger as e:
|
||||
print(f"WebSocket handler terminated: {e}")
|
||||
except Exception as e:
|
||||
print(f"An error occurred: {e}")
|
||||
|
||||
|
||||
class MeshbookProcessor:
|
||||
"""Processes data received from the WebSocket."""
|
||||
|
||||
def __init__(self):
|
||||
self.basic_temp_list = []
|
||||
|
||||
def handle_basic_data(self, data):
|
||||
"""Handles basic data from the server."""
|
||||
if not args.silent:
|
||||
print("Processing received basic data...")
|
||||
|
||||
self.basic_temp_list.append(data)
|
||||
if len(self.basic_temp_list) < 2:
|
||||
return
|
||||
|
||||
temp_dict = {}
|
||||
for entry in self.basic_temp_list:
|
||||
if isinstance(entry, list):
|
||||
for mesh in entry:
|
||||
if mesh.get("type") == "mesh":
|
||||
mesh_id = mesh["_id"]
|
||||
temp_dict[mesh_id] = {
|
||||
"mesh_name": mesh.get("name", "Unknown Mesh"),
|
||||
"mesh_desc": mesh.get("desc", "No description"),
|
||||
"nodes": []
|
||||
}
|
||||
elif isinstance(entry, dict):
|
||||
for mesh_id, nodes in entry.items():
|
||||
if mesh_id in temp_dict:
|
||||
temp_dict[mesh_id]["nodes"].extend(nodes)
|
||||
else:
|
||||
temp_dict[mesh_id] = {
|
||||
"mesh_name": "Unknown Mesh",
|
||||
"mesh_desc": "No description",
|
||||
"nodes": nodes
|
||||
}
|
||||
|
||||
for mesh_id, details in temp_dict.items():
|
||||
global_list.append({
|
||||
"mesh_name": details["mesh_name"],
|
||||
"mesh_id": mesh_id,
|
||||
"nodes": [
|
||||
{
|
||||
"node_id": node["_id"],
|
||||
"node_name": node.get("name", "Unknown Node"),
|
||||
"powered_on": node.get("pwr") == 1
|
||||
}
|
||||
for node in details["nodes"]
|
||||
]
|
||||
local_device_list[device.meshname].append({
|
||||
"device_id": device.nodeid,
|
||||
"device_name": device.name,
|
||||
"device_os": device.os_description,
|
||||
"device_tags": device.tags,
|
||||
"reachable": device.connected
|
||||
})
|
||||
basic_ready_state.set()
|
||||
ready_for_next.set()
|
||||
|
||||
async def receive_processor(self, python_client: MeshbookWebsocket):
|
||||
"""Processes messages received from the WebSocket."""
|
||||
global response_counter
|
||||
temp_responses_list = []
|
||||
return local_device_list
|
||||
|
||||
while True:
|
||||
message = await python_client.received_response_queue.get()
|
||||
action_type = message.get('action')
|
||||
if action_type in ('meshes', 'nodes'):
|
||||
self.handle_basic_data(message[action_type])
|
||||
elif action_type == 'msg':
|
||||
temp_responses_list.append(message)
|
||||
async def gather_targets(playbook: dict) -> dict:
|
||||
target_list = []
|
||||
|
||||
response_counter += 1 # Increment response counter
|
||||
if "device" in playbook and "group" not in playbook:
|
||||
pseudo_target = playbook["device"]
|
||||
|
||||
if not args.silent or args.information:
|
||||
print("Current Batch: {}".format(math.ceil(response_counter/len(target_ids))))
|
||||
print("Current response number: {}".format(response_counter))
|
||||
print("Current Calculation: {} % {} = {}".format(response_counter, len(target_ids), response_counter % len(target_ids)))
|
||||
for group in group_list:
|
||||
for device in group_list[group]:
|
||||
if device["reachable"] and pseudo_target == device["device_name"]:
|
||||
target_list.append(device["device_id"])
|
||||
|
||||
if response_counter % len(target_ids) == 0:
|
||||
batch_name = "Batch {}".format(math.ceil(response_counter / len(target_ids)))
|
||||
responses_dict[batch_name] = temp_responses_list
|
||||
temp_responses_list = []
|
||||
ready_for_next.set()
|
||||
elif action_type == 'close':
|
||||
print(message)
|
||||
elif not args.silent:
|
||||
print("Ignored action:", action_type)
|
||||
elif "group" in playbook and "device" not in playbook:
|
||||
pseudo_target = playbook["group"]
|
||||
|
||||
for group in group_list:
|
||||
if pseudo_target == group:
|
||||
for device in group_list[group]:
|
||||
if device["reachable"]:
|
||||
target_list.append(device["device_id"])
|
||||
|
||||
class MeshcallerActions:
|
||||
"""Processes playbook actions."""
|
||||
return target_list
|
||||
|
||||
@staticmethod
|
||||
async def process_arguments(python_client: MeshbookWebsocket, playbook_yaml: dict):
|
||||
"""Executes tasks defined in the playbook."""
|
||||
global response_counter, expected_responses, target_ids
|
||||
async def execute_playbook(session: meshctrl.Session, targets: dict, playbook: dict):
|
||||
responses_list = {}
|
||||
round = 1
|
||||
for task in playbook["tasks"]:
|
||||
output_text(("\033[1m\033[92m" + str(round) + ". Running: " + task["name"] + "\033[0m"), False)
|
||||
response = await session.run_command(nodeids=targets, command=task["command"], timeout=300)
|
||||
|
||||
await basic_ready_state.wait() # Wait for the basic data to be ready
|
||||
task_batch = []
|
||||
for device in response:
|
||||
device_result = response[device]["result"]
|
||||
response[device]["result"] = device_result.replace("Run commands completed.", "")
|
||||
response[device]["device_id"] = device
|
||||
response[device]["device_name"] = await translate_id_to_name(device)
|
||||
task_batch.append(response[device])
|
||||
|
||||
target_ids = MeshbookUtilities.get_target_ids(
|
||||
company=playbook_yaml.get('company'),
|
||||
device=playbook_yaml.get('device')
|
||||
)
|
||||
if not target_ids:
|
||||
raise ScriptEndTrigger("No targets found.")
|
||||
|
||||
run_command_template = {
|
||||
'action': 'runcommands',
|
||||
'nodeids': target_ids,
|
||||
'type': 0,
|
||||
'cmds': None,
|
||||
'runAsUser': 0,
|
||||
'responseid': 'meshctrl',
|
||||
'reply': True
|
||||
}
|
||||
|
||||
expected_responses = len(playbook_yaml['tasks']) * len(target_ids) # Calculate the total expected responses: tasks x target nodes
|
||||
|
||||
# Send commands for all nodes at once
|
||||
for task in playbook_yaml['tasks']:
|
||||
await ready_for_next.wait()
|
||||
run_command_template["cmds"] = task['command']
|
||||
run_command_template["nodeids"] = target_ids # Send to all target IDs at once
|
||||
|
||||
if not args.silent or args.information:
|
||||
print("-=-" * 40)
|
||||
print("Running task:", task)
|
||||
print("-=-" * 40)
|
||||
|
||||
# Send the command to all nodes in one go
|
||||
await python_client.ws_send_data(json.dumps(run_command_template))
|
||||
ready_for_next.clear()
|
||||
|
||||
# Wait until all expected responses are received
|
||||
while response_counter < expected_responses:
|
||||
await asyncio.sleep(1)
|
||||
|
||||
# Exit gracefully
|
||||
if not args.silent or args.information:
|
||||
print("-=-" * 40)
|
||||
|
||||
updated_response_dict = MeshbookUtilities.translate_nodeids(responses_dict, global_list)
|
||||
|
||||
if not args.nojson:
|
||||
print(json.dumps(updated_response_dict,indent=4))
|
||||
raise ScriptEndTrigger("All tasks completed successfully: Expected {} Received {}".format(expected_responses, response_counter))
|
||||
responses_list[task["name"]] = task_batch
|
||||
round += 1
|
||||
|
||||
output_text(("-" * 40), False)
|
||||
output_text((json.dumps(responses_list,indent=4)), True)
|
||||
|
||||
async def main():
|
||||
parser = argparse.ArgumentParser(description="Process command-line arguments")
|
||||
parser.add_argument("-pb", "--playbook", type=str, help="Path to the playbook file.", required=True)
|
||||
|
||||
parser.add_argument("--conf", type=str, help="Path for the API configuration file (default: ./api.conf).")
|
||||
parser.add_argument("--nojson", action="store_true", help="Makes the program not output the JSON response data.")
|
||||
parser.add_argument("-s", "--silent", action="store_true", help="Suppress terminal output.")
|
||||
parser.add_argument("-i", "--information", action="store_true", help="Add the calculations and other informational data to the output.")
|
||||
parser.add_argument("--conf", type=str, help="Path for the API configuration file (default: ./api.conf).", required=False)
|
||||
parser.add_argument("--noout", action="store_true", help="Makes the program not output response data.", required=False)
|
||||
parser.add_argument("-s", "--silent", action="store_true", help="Suppress terminal output", required=False)
|
||||
|
||||
global args
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
credentials = MeshbookUtilities.load_config(args.conf)
|
||||
python_client = MeshbookWebsocket()
|
||||
processor = MeshbookProcessor()
|
||||
output_text(("-" * 40), False)
|
||||
output_text(("\x1B[3mTrying to load the MeshCentral account credential file...\x1B[0m"), False)
|
||||
output_text(("\x1B[3mTrying to load the Playbook yaml file and compile it into something workable...\x1B[0m"), False)
|
||||
|
||||
websocket_task = asyncio.create_task(python_client.ws_handler(
|
||||
credentials['websocket_url'],
|
||||
credentials['username'],
|
||||
credentials['password']
|
||||
))
|
||||
processor_task = asyncio.create_task(processor.receive_processor(python_client))
|
||||
credentials, playbook = await asyncio.gather(
|
||||
(load_config() if args.conf is None else load_config(args.conf)),
|
||||
(compile_book(args.playbook))
|
||||
)
|
||||
|
||||
playbook_yaml = MeshbookUtilities.read_yaml(args.playbook)
|
||||
translated_playbook = MeshbookUtilities.replace_placeholders(playbook_yaml)
|
||||
await MeshcallerActions.process_arguments(python_client, translated_playbook)
|
||||
output_text(("\x1B[3mConnecting to MeshCentral and establish a session using variables from previous credential file.\x1B[0m"), False)
|
||||
session = await init_connection(credentials)
|
||||
|
||||
await asyncio.gather(websocket_task, processor_task)
|
||||
output_text(("\x1B[3mGenerating group list with nodes and reference the targets from that.\x1B[0m"), False)
|
||||
global group_list
|
||||
group_list = await compile_group_list(session)
|
||||
targets_list = await gather_targets(playbook)
|
||||
|
||||
except ScriptEndTrigger as e:
|
||||
if not args.silent or args.information:
|
||||
print(e)
|
||||
output_text(("-" * 40), False)
|
||||
if len(targets_list) == 0:
|
||||
output_text(("\033[91mNo targets found or targets unreachable, quitting.\x1B[0m"), True)
|
||||
else:
|
||||
output_text(("\033[91mExecuting playbook on the targets.\x1B[0m"), False)
|
||||
await execute_playbook(session, targets_list, playbook)
|
||||
|
||||
await session.close()
|
||||
|
||||
except OSError as message:
|
||||
output_text(message, True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -1,164 +0,0 @@
|
||||
#!/bin/python3
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
from base64 import b64encode
|
||||
from configparser import ConfigParser
|
||||
import json
|
||||
import math
|
||||
import meshctrl
|
||||
import os
|
||||
import yaml
|
||||
|
||||
'''
|
||||
Script utilities are handled in the following section.
|
||||
'''
|
||||
|
||||
class ScriptEndTrigger(Exception):
|
||||
pass
|
||||
|
||||
async def load_config(conf_file: str = './api.conf', segment: str = 'meshcentral-account') -> ConfigParser:
|
||||
if not os.path.exists(conf_file):
|
||||
raise ScriptEndTrigger(f'Missing config file {conf_file}. Provide an alternative path.')
|
||||
|
||||
config = ConfigParser()
|
||||
try:
|
||||
config.read(conf_file)
|
||||
except Exception as err:
|
||||
raise ScriptEndTrigger(f"Error reading configuration file '{conf_file}': {err}")
|
||||
|
||||
if segment not in config:
|
||||
raise ScriptEndTrigger(f'Segment "{segment}" not found in config file {conf_file}.')
|
||||
|
||||
return config[segment]
|
||||
|
||||
async def init_connection(credentials: dict) -> meshctrl.Session:
|
||||
session = meshctrl.Session(
|
||||
credentials['websocket_url'],
|
||||
user=credentials['username'],
|
||||
password=credentials['password']
|
||||
)
|
||||
await session.initialized.wait()
|
||||
return session
|
||||
|
||||
def output(args: argparse.Namespace, message: str):
|
||||
if not args.silent or args.information:
|
||||
print(message)
|
||||
|
||||
'''
|
||||
Creation and compilation happends in the following section, where the yaml gets read in, and edited accordingly.
|
||||
'''
|
||||
|
||||
async def compile_book(playbook_file: dict) -> dict:
|
||||
playbook = open(playbook_file, 'r')
|
||||
playbook = await replace_placeholders(yaml.safe_load(playbook))
|
||||
return playbook
|
||||
|
||||
async def replace_placeholders(playbook: dict) -> dict:
|
||||
variables = {var["name"]: var["value"] for var in playbook.get("variables", [])}
|
||||
|
||||
for task in playbook.get("tasks", []):
|
||||
command = task.get("command", "")
|
||||
for var_name, var_value in variables.items():
|
||||
placeholder = f"{{{{ {var_name} }}}}" # Create the placeholder string like "{{ host1 }}"
|
||||
command = command.replace(placeholder, var_value)
|
||||
task["command"] = command
|
||||
return playbook
|
||||
|
||||
'''
|
||||
Creation and compilation of the MeshCentral nodes list (list of all nodes available to the user in the configuration) is handled in the following section.
|
||||
'''
|
||||
|
||||
async def compile_group_list(session: meshctrl.Session) -> dict:
|
||||
devices_response = await session.list_devices(details=False, timeout=10)
|
||||
|
||||
local_device_list = {}
|
||||
for device in devices_response:
|
||||
if device.meshname not in local_device_list:
|
||||
local_device_list[device.meshname] = []
|
||||
|
||||
local_device_list[device.meshname].append({
|
||||
"device_id": device.nodeid,
|
||||
"device_name": device.name,
|
||||
"device_os": device.os_description,
|
||||
"device_tags": device.tags,
|
||||
"reachable": device.connected
|
||||
})
|
||||
|
||||
return local_device_list
|
||||
|
||||
async def gather_targets(group_list: dict, playbook: dict) -> dict:
|
||||
target_list = []
|
||||
|
||||
if "device" in playbook and "group" not in playbook:
|
||||
pseudo_target = playbook["device"]
|
||||
|
||||
for group in group_list:
|
||||
for device in group_list[group]:
|
||||
if device["reachable"] and pseudo_target == device["device_name"]:
|
||||
target_list.append(device["device_id"])
|
||||
|
||||
elif "group" in playbook and "device" not in playbook:
|
||||
pseudo_target = playbook["group"]
|
||||
|
||||
for group in group_list:
|
||||
if pseudo_target == group:
|
||||
for device in group_list[group]:
|
||||
if device["reachable"]:
|
||||
target_list.append(device["device_id"])
|
||||
|
||||
return target_list
|
||||
|
||||
async def execute_playbook(session: meshctrl.Session, targets: dict, playbook: dict):
|
||||
|
||||
for task in playbook["tasks"]:
|
||||
print("Running:", task["name"])
|
||||
response = await session.run_command(nodeids=targets, command=task["command"], timeout=300)
|
||||
|
||||
print(json.dumps(response,indent=4))
|
||||
for device in response:
|
||||
device_result = str(response[device]["result"])
|
||||
device_result = device_result.replace("Run commands completed.", "")
|
||||
print("AFTER", device_result)
|
||||
|
||||
async def main():
|
||||
parser = argparse.ArgumentParser(description="Process command-line arguments")
|
||||
parser.add_argument("-pb", "--playbook", type=str, help="Path to the playbook file.", required=True)
|
||||
parser.add_argument("--conf", type=str, help="Path for the API configuration file (default: ./api.conf).", required=False)
|
||||
parser.add_argument("--noout", action="store_true", help="Makes the program not output response data.", required=False)
|
||||
parser.add_argument("-s", "--silent", action="store_true", help="Suppress terminal output.", required=False)
|
||||
parser.add_argument("-i", "--information", action="store_true", help="Add the calculations and other informational data to the output.", required=False)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
output(args, "Trying to load the MeshCentral account credential file...")
|
||||
output(args, "Trying to load the Playbook yaml file and compile it into something workable...")
|
||||
|
||||
credentials, playbook = await asyncio.gather(
|
||||
(load_config() if args.conf is None else load_config(args.conf)),
|
||||
(compile_book(args.playbook))
|
||||
)
|
||||
|
||||
output(args, "Connecting to MeshCentral and establish a session using variables from previous credential file.")
|
||||
session = await init_connection(credentials)
|
||||
|
||||
output(args, "Generating group list with nodes and reference the targets from that.")
|
||||
group_list = await compile_group_list(session)
|
||||
targets_list = await gather_targets(group_list, playbook)
|
||||
|
||||
if len(targets_list) == 0:
|
||||
output(args, "No targets found or targets unreachable, quitting.")
|
||||
else:
|
||||
output(args, "Executing playbook on the targets.")
|
||||
output(args, json.dumps(targets_list,indent=4))
|
||||
await execute_playbook(session, targets_list, playbook)
|
||||
|
||||
await session.close()
|
||||
|
||||
except ScriptEndTrigger as message:
|
||||
output(args, message)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user