daanselen/meshbook
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #4 from DaanSelen/dev

Browse Source 
Tidying up everything and adding slight QoL changes.
This commit is contained in:
dselen
2025-02-13 15:37:29 +01:00
committed by GitHub
parent 03683976a8 7f0159a8fa
commit bd833456d0
2 changed files with 128 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/apt_update_example.yaml
+2 -2
View File
@@ -6,8 +6,8 @@ variables:
- name: package_manager - name: package_manager
value: "apt" value: "apt"
tasks: tasks:
- name: refresh the cache - name: refresh the {{ package_manager }} cache
command: "{{ package_manager }} update" command: "{{ package_manager }} update"
- name: display available upgrades - name: display available upgrades with {{ package_manager }}
command: "{{ package_manager }} list --upgradable" command: "{{ package_manager }} list --upgradable"
meshbook.py
+125 -59
View File
@@ -5,11 +5,12 @@ import asyncio
from base64 import b64encode from base64 import b64encode
from configparser import ConfigParser from configparser import ConfigParser
import json import json
import math
import meshctrl import meshctrl
import os import os
import yaml import yaml
grace_period = 3 # Grace period will last for x (by default 3) second(s).
''' '''
Script utilities are handled in the following section. Script utilities are handled in the following section.
''' '''
@@ -17,13 +18,32 @@ Script utilities are handled in the following section.
class ScriptEndTrigger(Exception): class ScriptEndTrigger(Exception):
pass pass
def console(message: str, required: bool=False): class text_color:
if required: black = "\033[30m"
print(message) red = "\033[31m"
elif not args.silent: green = "\033[32m"
print(message) yellow = "\033[33m"
blue = "\033[34m"
magenta = "\033[35m"
cyan = "\033[36m"
white = "\033[37m"
italic = "\x1B[3m"
reset = "\x1B[0m"
def console(message: str, required: bool=False):
'''
Helper function for terminal output, with a couple variables for the silent flag. Also clears terminal color each time.
'''
if required:
print(message + text_color.reset)
elif not args.silent:
print(message + text_color.reset)
async def load_config(segment: str = 'meshcentral-account') -> dict:
'''
Function that loads the segment from the meshcentral.conf (by default) file and returns the it in a dict.
'''
async def load_config(segment: str = 'meshcentral-account') -> ConfigParser:
conf_file = args.conf conf_file = args.conf
if not os.path.exists(conf_file): if not os.path.exists(conf_file):
raise ScriptEndTrigger(f'Missing config file {conf_file}. Provide an alternative path.') raise ScriptEndTrigger(f'Missing config file {conf_file}. Provide an alternative path.')
@@ -40,6 +60,10 @@ async def load_config(segment: str = 'meshcentral-account') -> ConfigParser:
return config[segment] return config[segment]
async def init_connection(credentials: dict) -> meshctrl.Session: async def init_connection(credentials: dict) -> meshctrl.Session:
'''
Use the libmeshctrl library to initiate a Secure Websocket (wss) connection to the MeshCentral instance.
'''
session = meshctrl.Session( session = meshctrl.Session(
credentials['websocket_url'], credentials['websocket_url'],
user=credentials['username'], user=credentials['username'],
@@ -48,37 +72,61 @@ async def init_connection(credentials: dict) -> meshctrl.Session:
await session.initialized.wait() await session.initialized.wait()
return session return session
async def translate_id_to_name(target_id: str, group_list: dict) -> str: async def translate_nodeid_to_name(target_id: str, group_list: dict) -> str:
'''
Simple function that looks up nodeid to the human-readable name if existent - otherwise return None.
'''
for group in group_list: for group in group_list:
for device in group_list[group]: for device in group_list[group]:
if device["device_id"] == target_id: if device["device_id"] == target_id:
return device["device_name"] return device["device_name"]
return None
''' '''
Creation and compilation happends in the following section, where the yaml gets read in, and edited accordingly. 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: async def compile_book(meshbook_file: dict) -> dict:
playbook = open(playbook_file, 'r') '''
playbook = await replace_placeholders(yaml.safe_load(playbook)) Simple function that opens the file and replaces placeholders through the next function. After that just return it.
return playbook '''
async def replace_placeholders(playbook: dict) -> dict: meshbook = open(meshbook_file, 'r')
variables = {var["name"]: var["value"] for var in playbook.get("variables", [])} meshbook = await replace_placeholders(yaml.safe_load(meshbook))
return meshbook
for task in playbook.get("tasks", []): async def replace_placeholders(meshbook: dict) -> dict:
command = task.get("command", "") '''
Replace the placeholders in both name and command fields of the tasks. According to the variables defined in the variables list.
'''
variables = {var["name"]: var["value"] for var in meshbook.get("variables", [])}
for task in meshbook.get("tasks", []):
task_name = task.get("name")
for var_name, var_value in variables.items():
placeholder = f"{{{{ {var_name} }}}}"
task_name = task_name.replace(placeholder, var_value)
task["name"] = task_name
command = task.get("command")
for var_name, var_value in variables.items(): for var_name, var_value in variables.items():
placeholder = f"{{{{ {var_name} }}}}" # Create the placeholder string like "{{ host1 }}" placeholder = f"{{{{ {var_name} }}}}" # Create the placeholder string like "{{ host1 }}"
command = command.replace(placeholder, var_value) command = command.replace(placeholder, var_value)
task["command"] = command task["command"] = command
return playbook
return meshbook
''' '''
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. 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: async def compile_group_list(session: meshctrl.Session) -> dict:
'''
Function that retrieves the devices from MeshCentral and compiles it into a efficient list.
'''
devices_response = await session.list_devices(details=False, timeout=10) devices_response = await session.list_devices(details=False, timeout=10)
local_device_list = {} local_device_list = {}
@@ -96,11 +144,16 @@ async def compile_group_list(session: meshctrl.Session) -> dict:
return local_device_list return local_device_list
async def filter_targets(devices: list[dict], os_categories: dict, target_os: str = None) -> list[str]: async def filter_targets(devices: list[dict], os_categories: dict, target_os: str = None) -> list[str]:
"""Filters devices based on reachability and optional OS criteria, supporting nested OS categories.""" '''
Filters devices based on reachability and optional OS criteria, supporting nested OS categories.
'''
valid_devices = [] valid_devices = []
def get_os_variants(category: str, os_map: dict) -> set: def get_os_variants(category: str, os_map: dict) -> set:
"""Extracts all OS names under a given category if it exists.""" '''
Extracts all OS names under a given category if it exists.
'''
for key, value in os_map.items(): for key, value in os_map.items():
if key == category: if key == category:
@@ -135,14 +188,18 @@ async def filter_targets(devices: list[dict], os_categories: dict, target_os: st
return valid_devices return valid_devices
async def gather_targets(playbook: dict, group_list: dict[str, list[dict]], os_categories: dict) -> list[str]: async def gather_targets(meshbook: dict, group_list: dict[str, list[dict]], os_categories: dict) -> list[str]:
"""Finds target devices based on playbook criteria (device or group).""" '''
Finds target devices based on meshbook criteria (device or group).
'''
target_list = [] target_list = []
target_os = playbook.get("target_os") target_os = meshbook.get("target_os")
async def process_device_or_group(pseudo_target, group_list, os_categories, target_os) -> list[str]: async def process_device_or_group(pseudo_target, group_list, os_categories, target_os) -> list[str]:
"""Helper function to process devices or groups.""" '''
Helper function to process devices or groups.
'''
matched_devices = [] matched_devices = []
for group in group_list: for group in group_list:
@@ -154,14 +211,14 @@ async def gather_targets(playbook: dict, group_list: dict[str, list[dict]], os_c
return await filter_targets(matched_devices, os_categories, target_os) return await filter_targets(matched_devices, os_categories, target_os)
return [] return []
match playbook: match meshbook:
case {"device": pseudo_target}: # Single device target case {"device": pseudo_target}: # Single device target
if isinstance(pseudo_target, str): if isinstance(pseudo_target, str):
matched_devices = await process_device_or_group(pseudo_target, group_list, os_categories, target_os) matched_devices = await process_device_or_group(pseudo_target, group_list, os_categories, target_os)
target_list.extend(matched_devices) target_list.extend(matched_devices)
else: else:
console("\033[91mPlease use devices (Notice the 'S') for multiple devices.\x1B[0m", True) console(text_color.yellow + "Please use devices (Notice the 'S') for multiple devices.", True)
case {"devices": pseudo_target}: # List of devices case {"devices": pseudo_target}: # List of devices
if isinstance(pseudo_target, list): if isinstance(pseudo_target, list):
@@ -170,7 +227,7 @@ async def gather_targets(playbook: dict, group_list: dict[str, list[dict]], os_c
target_list.extend(matched_devices) target_list.extend(matched_devices)
else: else:
console("\033[91mThe 'devices' method is being used, but only one string is given. Did you mean 'device'?\x1B[0m", True) console(text_color.yellow + "The 'devices' method is being used, but only one string is given. Did you mean 'device'?", True)
case {"group": pseudo_target}: # Single group target case {"group": pseudo_target}: # Single group target
if isinstance(pseudo_target, str) and pseudo_target in group_list: if isinstance(pseudo_target, str) and pseudo_target in group_list:
@@ -178,7 +235,7 @@ async def gather_targets(playbook: dict, group_list: dict[str, list[dict]], os_c
target_list.extend(matched_devices) target_list.extend(matched_devices)
else: else:
console("\033[91mPlease use groups (Notice the 'S') for multiple groups.\x1B[0m", True) console(text_color.yellow + "Please use groups (Notice the 'S') for multiple groups.", True)
case {"groups": pseudo_target}: # List of groups case {"groups": pseudo_target}: # List of groups
if isinstance(pseudo_target, list): if isinstance(pseudo_target, list):
@@ -187,16 +244,20 @@ async def gather_targets(playbook: dict, group_list: dict[str, list[dict]], os_c
matched_devices = await filter_targets(group_list[sub_pseudo_target], os_categories, target_os) matched_devices = await filter_targets(group_list[sub_pseudo_target], os_categories, target_os)
target_list.extend(matched_devices) target_list.extend(matched_devices)
else: else:
console("\033[91mThe 'groups' method is being used, but only one string is given. Did you mean 'group'?\x1B[0m", True) console(text_color.yellow + "The 'groups' method is being used, but only one string is given. Did you mean 'group'?", True)
return target_list return target_list
async def execute_playbook(session: meshctrl.Session, targets: dict, playbook: dict, group_list: dict) -> None: async def execute_meshbook(session: meshctrl.Session, targets: dict, meshbook: dict, group_list: dict) -> None:
'''
Actual function that handles meshbook execution, also responsible for formatting the resulting JSON.
'''
responses_list = {} responses_list = {}
round = 1 round = 1
for task in playbook["tasks"]: for task in meshbook["tasks"]:
console(("\033[1m\033[92m" + str(round) + ". Running: " + task["name"] + "\033[0m")) console(text_color.green + str(round) + ". Running: " + task["name"])
response = await session.run_command(nodeids=targets, command=task["command"],ignore_output=False,timeout=900) response = await session.run_command(nodeids=targets, command=task["command"],ignore_output=False,timeout=900)
task_batch = [] task_batch = []
@@ -204,7 +265,7 @@ async def execute_playbook(session: meshctrl.Session, targets: dict, playbook: d
device_result = response[device]["result"] device_result = response[device]["result"]
response[device]["result"] = device_result.replace("Run commands completed.", "") response[device]["result"] = device_result.replace("Run commands completed.", "")
response[device]["device_id"] = device response[device]["device_id"] = device
response[device]["device_name"] = await translate_id_to_name(device, group_list) response[device]["device_name"] = await translate_nodeid_to_name(device, group_list)
task_batch.append(response[device]) task_batch.append(response[device])
responses_list["Task " + str(round)] = { responses_list["Task " + str(round)] = {
@@ -221,12 +282,16 @@ async def execute_playbook(session: meshctrl.Session, targets: dict, playbook: d
console(json.dumps(responses_list), True) console(json.dumps(responses_list), True)
async def main(): async def main():
'''
Main function where the program starts. Place from which all comands originate (eventually).
'''
parser = argparse.ArgumentParser(description="Process command-line arguments") parser = argparse.ArgumentParser(description="Process command-line arguments")
parser.add_argument("-pb", "--playbook", type=str, help="Path to the playbook yaml file.", required=True) parser.add_argument("-pb", "--meshbook", type=str, help="Path to the meshbook yaml file.", required=True)
parser.add_argument("-oc", "--oscategories", type=str, help="Path to the Operating System categories JSON file.", required=False, default="./os_categories.json") parser.add_argument("-oc", "--oscategories", type=str, help="Path to the Operating System categories JSON file.", required=False, default="./os_categories.json")
parser.add_argument("--conf", type=str, help="Path for the API configuration file (default: ./meshcentral.conf).", required=False, default="./meshcentral.conf") parser.add_argument("--conf", type=str, help="Path for the API configuration file (default: ./meshcentral.conf).", required=False, default="./meshcentral.conf")
parser.add_argument("--nograce", action="store_true", help="Disable the grace 3 seconds before running the playbook.", required=False) parser.add_argument("--nograce", action="store_true", help="Disable the grace 3 seconds before running the meshbook.", required=False)
parser.add_argument("-i", "--indent", action="store_true", help="Use an JSON indentation of 4 when this flag is passed.", required=False) parser.add_argument("-i", "--indent", action="store_true", help="Use an JSON indentation of 4 when this flag is passed.", required=False)
parser.add_argument("-s", "--silent", action="store_true", help="Suppress terminal output", required=False) parser.add_argument("-s", "--silent", action="store_true", help="Suppress terminal output", required=False)
@@ -238,9 +303,9 @@ async def main():
with open(local_categories_file, "r") as file: with open(local_categories_file, "r") as file:
os_categories = json.load(file) os_categories = json.load(file)
credentials, playbook = await asyncio.gather( credentials, meshbook = await asyncio.gather(
(load_config()), (load_config()),
(compile_book(args.playbook)) (compile_book(args.meshbook))
) )
''' '''
@@ -248,41 +313,42 @@ async def main():
''' '''
console(("-" * 40)) console(("-" * 40))
console("Playbook: " + args.playbook) console("meshbook: " + text_color.yellow + args.meshbook)
console("Operating System Categorisation file: " + args.oscategories) console("Operating System Categorisation file: " + text_color.yellow + args.oscategories)
console("Congiguration file: " + args.conf) console("Configuration file: " + text_color.yellow + args.conf)
if "device" in playbook: console("Target Operating System category given: " + text_color.yellow + meshbook["target_os"])
console("Target device: " + str(playbook["device"])) if "device" in meshbook:
console("Target device: " + text_color.yellow + str(meshbook["device"]))
elif "group" in playbook: elif "group" in meshbook:
console("Target group: " + str(playbook["group"])) console("Target group: " + text_color.yellow + str(meshbook["group"]))
console("Grace: " + str((not args.nograce))) # Negation of bool for correct explanation console("Grace: " + text_color.yellow + str((not args.nograce))) # Negation of bool for correct explanation
console("Silent: False") # Can be pre-defined because if silent flag was passed then none of this would be printed. console("Silent: " + text_color.yellow + "False") # Can be pre-defined because if silent flag was passed then none of this would be printed.
session = await init_connection(credentials) session = await init_connection(credentials)
console(("-" * 40)) console(("-" * 40))
console(("\x1B[3mTrying to load the MeshCentral account credential file...\x1B[0m")) console(text_color.italic + "Trying to load the MeshCentral account credential file...")
console(("\x1B[3mTrying to load the Playbook yaml file and compile it into something workable...\x1B[0m")) console(text_color.italic + "Trying to load the meshbook yaml file and compile it into something workable...")
console(("\x1B[3mTrying to load the Operating System categorisation JSON file...\x1B[0m")) console(text_color.italic + "Trying to load the Operating System categorisation JSON file...")
console(("\x1B[3mConnecting to MeshCentral and establish a session using variables from previous credential file.\x1B[0m")) console(text_color.italic + "Connecting to MeshCentral and establish a session using variables from previous credential file.")
console(text_color.italic + "Generating group list with nodes and reference the targets from that.")
''' '''
End of the main information displaying section. End of the main information displaying section.
''' '''
console(("\x1B[3mGenerating group list with nodes and reference the targets from that.\x1B[0m"))
group_list = await compile_group_list(session) group_list = await compile_group_list(session)
targets_list = await gather_targets(playbook, group_list, os_categories) targets_list = await gather_targets(meshbook, group_list, os_categories)
if len(targets_list) == 0: if len(targets_list) == 0:
console(("\033[91mNo targets found or targets unreachable, quitting.\x1B[0m"), True) console(text_color.red + "No targets found or targets unreachable, quitting.", True)
console(("-" * 40), True) console(("-" * 40), True)
else: else:
console(("-" * 40)) console(("-" * 40))
match playbook: match meshbook:
case {"group": candidate_target_name}: case {"group": candidate_target_name}:
target_name = candidate_target_name target_name = candidate_target_name
@@ -295,22 +361,22 @@ async def main():
case {"devices": candidate_target_name}: case {"devices": candidate_target_name}:
target_name = str(candidate_target_name) target_name = str(candidate_target_name)
console(("\033[91mExecuting playbook on the target(s): " + target_name + ".\x1B[0m")) console(text_color.yellow + "Executing meshbook on the target(s): " + target_name + ".")
if not args.nograce: if not args.nograce:
console(("\033[91mInitiating grace-period...\x1B[0m")) console(text_color.yellow + "Initiating grace-period...")
for x in range(3): for x in range(grace_period):
console(("\033[91m{}...\x1B[0m".format(x+1))) console(text_color.yellow + "{}...".format(x+1)) # Countdown!
await asyncio.sleep(1) await asyncio.sleep(1)
console(("-" * 40)) console(("-" * 40))
await execute_playbook(session, targets_list, playbook, group_list) await execute_meshbook(session, targets_list, meshbook, group_list)
await session.close() await session.close()
except OSError as message: except OSError as message:
console(message, True) console(text_color.red + message, True)
if __name__ == "__main__": if __name__ == "__main__":
asyncio.run(main()) asyncio.run(main())