Package connpy
Connection manager
Connpy is a connection manager that allows you to store nodes to connect them fast and password free.
Features
- You can generate profiles and reference them from nodes using @profilename
so you dont need to edit multiple nodes when changing password or other
information.
- Nodes can be stored on @folder or @subfolder@folder to organize your
devices. Then can be referenced using node@subfolder@folder or node@folder
- If you have too many nodes. Get completion script using: conn config
--completion, or use fzf installing pyfzf and running conn config --fzf true
- Much more!
Usage
usage: conn [-h] [--add | --del | --mod | --show | --debug] [node|folder]
conn {profile,move,mv,copy,cp,list,ls,bulk,config} ...
positional arguments:
node|folder node[@subfolder][@folder]
Connect to specific node or show all matching nodes
[@subfolder][@folder]
Show all available connections globaly or in specified path
Options:
-h, --help show this help message and exit
-v, --version Show version
-a, --add Add new node[@subfolder][@folder] or [@subfolder]@folder
-r, --del, --rm Delete node[@subfolder][@folder] or [@subfolder]@folder
-e, --mod, --edit Modify node[@subfolder][@folder]
-s, --show Show node[@subfolder][@folder]
-d, --debug Display all conections steps
Commands:
profile Manage profiles
move (mv) Move node
copy (cp) Copy node
list (ls) List profiles, nodes or folders
bulk Add nodes in bulk
run Run scripts or commands on nodes
config Manage app config
api Start and stop connpy api
Manage profiles
usage: conn profile [-h] (--add | --del | --mod | --show) profile
positional arguments:
profile Name of profile to manage
options:
-h, --help show this help message and exit
-a, --add Add new profile
-r, --del, --rm Delete profile
-e, --mod, --edit Modify profile
-s, --show Show profile
Examples
conn profile --add office-user
conn --add @office
conn --add @datacenter@office
conn --add server@datacenter@office
conn --add pc@office
conn --show server@datacenter@office
conn pc@office
conn server
http API
With the Connpy API you can run commands on devices using http requests
1. List Nodes
Endpoint: /list_nodes
Method: POST
Description: This route returns a list of nodes. It can also filter the list based on a given keyword.
Request Body:
{
"filter": "<keyword>"
}
filter(optional): A keyword to filter the list of nodes. It returns only the nodes that contain the keyword. If not provided, the route will return the entire list of nodes.
Response:
- A JSON array containing the filtered list of nodes.
2. Get Nodes
Endpoint: /get_nodes
Method: POST
Description: This route returns a dictionary of nodes with all their attributes. It can also filter the nodes based on a given keyword.
Request Body:
{
"filter": "<keyword>"
}
filter(optional): A keyword to filter the nodes. It returns only the nodes that contain the keyword. If not provided, the route will return the entire list of nodes.
Response:
- A JSON array containing the filtered nodes.
3. Run Commands
Endpoint: /run_commands
Method: POST
Description: This route runs commands on selected nodes based on the provided action, nodes, and commands. It also supports executing tests by providing expected results.
Request Body:
{
"action": "<action>",
"nodes": "<nodes>",
"commands": "<commands>",
"expected": "<expected>",
"options": "<options>"
}
action(required): The action to be performed. Possible values:runortest.nodes(required): A list of nodes or a single node on which the commands will be executed. The nodes can be specified as individual node names or a node group with the@prefix. Node groups can also be specified as arrays with a list of nodes inside the group.commands(required): A list of commands to be executed on the specified nodes.expected(optional, only used when the action istest): A single expected result for the test.options(optional): Array to pass options to the run command, options are:prompt,parallel,timeout
Response:
- A JSON object with the results of the executed commands on the nodes.
4. Ask AI
Endpoint: /ask_ai
Method: POST
Description: This route sends to chatgpt IA a request that will parse it into an understandable output for the application and then run the request.
Request Body:
{
"input": "<user input request>",
"dryrun": true or false
}
input(required): The user input requesting the AI to perform an action on some devices or get the devices list.dryrun(optional): If set to true, it will return the parameters to run the request but it won't run it. default is false.
Response:
- A JSON array containing the action to run and the parameters and the result of the action.
Automation module
The automation module
Standalone module
import connpy
router = connpy.node("uniqueName","ip/host", user="user", password="pass")
router.run(["term len 0","show run"])
print(router.output)
hasip = router.test("show ip int brief","1.1.1.1")
if hasip:
print("Router has ip 1.1.1.1")
else:
print("router does not have ip 1.1.1.1")
Using manager configuration
import connpy
conf = connpy.configfile()
device = conf.getitem("server@office")
server = connpy.node("unique name", **device, config=conf)
result = server.run(["cd /", "ls -la"])
print(result)
Running parallel tasks
import connpy
conf = connpy.configfile()
#You can get the nodes from the config from a folder and fitlering in it
nodes = conf.getitem("@office", ["router1", "router2", "router3"])
#You can also get each node individually:
nodes = {}
nodes["router1"] = conf.getitem("router1@office")
nodes["router2"] = conf.getitem("router2@office")
nodes["router10"] = conf.getitem("router10@datacenter")
#Also, you can create the nodes manually:
nodes = {}
nodes["router1"] = {"host": "1.1.1.1", "user": "user", "password": "pass1"}
nodes["router2"] = {"host": "1.1.1.2", "user": "user", "password": "pass2"}
nodes["router3"] = {"host": "1.1.1.2", "user": "user", "password": "pass3"}
#Finally you run some tasks on the nodes
mynodes = connpy.nodes(nodes, config = conf)
result = mynodes.test(["show ip int br"], "1.1.1.2")
for i in result:
print("---" + i + "---")
print(result[i])
print()
# Or for one specific node
mynodes.router1.run(["term len 0". "show run"], folder = "/home/user/logs")
Using variables
import connpy
config = connpy.configfile()
nodes = config.getitem("@office", ["router1", "router2", "router3"])
commands = []
commands.append("config t")
commands.append("interface lo {id}")
commands.append("ip add {ip} {mask}")
commands.append("end")
variables = {}
variables["router1@office"] = {"ip": "10.57.57.1"}
variables["router2@office"] = {"ip": "10.57.57.2"}
variables["router3@office"] = {"ip": "10.57.57.3"}
variables["__global__"] = {"id": "57"}
variables["__global__"]["mask"] = "255.255.255.255"
expected = "!"
routers = connpy.nodes(nodes, config = config)
routers.run(commands, variables)
routers.test("ping {ip}", expected, variables)
for key in routers.result:
print(key, ' ---> ', ("pass" if routers.result[key] else "fail"))
Using AI
import connpy
conf = connpy.configfile()
organization = 'openai-org'
api_key = "openai-key"
myia = ai(conf, organization, api_key)
input = "go to router 1 and get me the full configuration"
result = myia.ask(input, dryrun = False)
print(result)
Expand source code
#!/usr/bin/env python3
'''
## Connection manager
Connpy is a connection manager that allows you to store nodes to connect them fast and password free.
### Features
- You can generate profiles and reference them from nodes using @profilename
so you dont need to edit multiple nodes when changing password or other
information.
- Nodes can be stored on @folder or @subfolder@folder to organize your
devices. Then can be referenced using node@subfolder@folder or node@folder
- If you have too many nodes. Get completion script using: conn config
--completion, or use fzf installing pyfzf and running conn config --fzf true
- Much more!
### Usage
```
usage: conn [-h] [--add | --del | --mod | --show | --debug] [node|folder]
conn {profile,move,mv,copy,cp,list,ls,bulk,config} ...
positional arguments:
node|folder node[@subfolder][@folder]
Connect to specific node or show all matching nodes
[@subfolder][@folder]
Show all available connections globaly or in specified path
Options:
-h, --help show this help message and exit
-v, --version Show version
-a, --add Add new node[@subfolder][@folder] or [@subfolder]@folder
-r, --del, --rm Delete node[@subfolder][@folder] or [@subfolder]@folder
-e, --mod, --edit Modify node[@subfolder][@folder]
-s, --show Show node[@subfolder][@folder]
-d, --debug Display all conections steps
Commands:
profile Manage profiles
move (mv) Move node
copy (cp) Copy node
list (ls) List profiles, nodes or folders
bulk Add nodes in bulk
run Run scripts or commands on nodes
config Manage app config
api Start and stop connpy api
```
### Manage profiles
```
usage: conn profile [-h] (--add | --del | --mod | --show) profile
positional arguments:
profile Name of profile to manage
options:
-h, --help show this help message and exit
-a, --add Add new profile
-r, --del, --rm Delete profile
-e, --mod, --edit Modify profile
-s, --show Show profile
```
### Examples
```
conn profile --add office-user
conn --add @office
conn --add @datacenter@office
conn --add server@datacenter@office
conn --add pc@office
conn --show server@datacenter@office
conn pc@office
conn server
```
## http API
With the Connpy API you can run commands on devices using http requests
### 1. List Nodes
**Endpoint**: `/list_nodes`
**Method**: `POST`
**Description**: This route returns a list of nodes. It can also filter the list based on a given keyword.
#### Request Body:
```json
{
"filter": "<keyword>"
}
```
* `filter` (optional): A keyword to filter the list of nodes. It returns only the nodes that contain the keyword. If not provided, the route will return the entire list of nodes.
#### Response:
- A JSON array containing the filtered list of nodes.
---
### 2. Get Nodes
**Endpoint**: `/get_nodes`
**Method**: `POST`
**Description**: This route returns a dictionary of nodes with all their attributes. It can also filter the nodes based on a given keyword.
#### Request Body:
```json
{
"filter": "<keyword>"
}
```
* `filter` (optional): A keyword to filter the nodes. It returns only the nodes that contain the keyword. If not provided, the route will return the entire list of nodes.
#### Response:
- A JSON array containing the filtered nodes.
---
### 3. Run Commands
**Endpoint**: `/run_commands`
**Method**: `POST`
**Description**: This route runs commands on selected nodes based on the provided action, nodes, and commands. It also supports executing tests by providing expected results.
#### Request Body:
```json
{
"action": "<action>",
"nodes": "<nodes>",
"commands": "<commands>",
"expected": "<expected>",
"options": "<options>"
}
```
* `action` (required): The action to be performed. Possible values: `run` or `test`.
* `nodes` (required): A list of nodes or a single node on which the commands will be executed. The nodes can be specified as individual node names or a node group with the `@` prefix. Node groups can also be specified as arrays with a list of nodes inside the group.
* `commands` (required): A list of commands to be executed on the specified nodes.
* `expected` (optional, only used when the action is `test`): A single expected result for the test.
* `options` (optional): Array to pass options to the run command, options are: `prompt`, `parallel`, `timeout`
#### Response:
- A JSON object with the results of the executed commands on the nodes.
---
### 4. Ask AI
**Endpoint**: `/ask_ai`
**Method**: `POST`
**Description**: This route sends to chatgpt IA a request that will parse it into an understandable output for the application and then run the request.
#### Request Body:
```json
{
"input": "<user input request>",
"dryrun": true or false
}
```
* `input` (required): The user input requesting the AI to perform an action on some devices or get the devices list.
* `dryrun` (optional): If set to true, it will return the parameters to run the request but it won't run it. default is false.
#### Response:
- A JSON array containing the action to run and the parameters and the result of the action.
## Automation module
The automation module
### Standalone module
```
import connpy
router = connpy.node("uniqueName","ip/host", user="user", password="pass")
router.run(["term len 0","show run"])
print(router.output)
hasip = router.test("show ip int brief","1.1.1.1")
if hasip:
print("Router has ip 1.1.1.1")
else:
print("router does not have ip 1.1.1.1")
```
### Using manager configuration
```
import connpy
conf = connpy.configfile()
device = conf.getitem("server@office")
server = connpy.node("unique name", **device, config=conf)
result = server.run(["cd /", "ls -la"])
print(result)
```
### Running parallel tasks
```
import connpy
conf = connpy.configfile()
#You can get the nodes from the config from a folder and fitlering in it
nodes = conf.getitem("@office", ["router1", "router2", "router3"])
#You can also get each node individually:
nodes = {}
nodes["router1"] = conf.getitem("router1@office")
nodes["router2"] = conf.getitem("router2@office")
nodes["router10"] = conf.getitem("router10@datacenter")
#Also, you can create the nodes manually:
nodes = {}
nodes["router1"] = {"host": "1.1.1.1", "user": "user", "password": "pass1"}
nodes["router2"] = {"host": "1.1.1.2", "user": "user", "password": "pass2"}
nodes["router3"] = {"host": "1.1.1.2", "user": "user", "password": "pass3"}
#Finally you run some tasks on the nodes
mynodes = connpy.nodes(nodes, config = conf)
result = mynodes.test(["show ip int br"], "1.1.1.2")
for i in result:
print("---" + i + "---")
print(result[i])
print()
# Or for one specific node
mynodes.router1.run(["term len 0". "show run"], folder = "/home/user/logs")
```
### Using variables
```
import connpy
config = connpy.configfile()
nodes = config.getitem("@office", ["router1", "router2", "router3"])
commands = []
commands.append("config t")
commands.append("interface lo {id}")
commands.append("ip add {ip} {mask}")
commands.append("end")
variables = {}
variables["router1@office"] = {"ip": "10.57.57.1"}
variables["router2@office"] = {"ip": "10.57.57.2"}
variables["router3@office"] = {"ip": "10.57.57.3"}
variables["__global__"] = {"id": "57"}
variables["__global__"]["mask"] = "255.255.255.255"
expected = "!"
routers = connpy.nodes(nodes, config = config)
routers.run(commands, variables)
routers.test("ping {ip}", expected, variables)
for key in routers.result:
print(key, ' ---> ', ("pass" if routers.result[key] else "fail"))
```
### Using AI
```
import connpy
conf = connpy.configfile()
organization = 'openai-org'
api_key = "openai-key"
myia = ai(conf, organization, api_key)
input = "go to router 1 and get me the full configuration"
result = myia.ask(input, dryrun = False)
print(result)
```
'''
from .core import node,nodes
from .configfile import configfile
from .connapp import connapp
from .api import *
from .ai import ai
from ._version import __version__
from pkg_resources import get_distribution
__all__ = ["node", "nodes", "configfile", "connapp", "ai"]
__author__ = "Federico Luzzi"
__pdoc__ = {
'core': False,
'completion': False,
'api': False
}Classes
class ai (config, org=None, api_key=None, model='gpt-3.5-turbo', temp=0.7)-
This class generates a ai object. Containts all the information and methods to make requests to openAI chatGPT to run actions on the application.
Attributes:
- model (str): Model of GPT api to use. Default is gpt-3.5-turbo. - temp (float): Value between 0 and 1 that control the randomness of generated text, with higher values increasing creativity. Default is 0.7.Parameters:
- config (obj): Pass the object created with class configfile with key for decryption and extra configuration if you are using connection manager.Optional Parameters:
- org (str): A unique token identifying the user organization to interact with the API. - api_key (str): A unique authentication token required to access and interact with the API. - model (str): Model of GPT api to use. Default is gpt-3.5-turbo. - temp (float): Value between 0 and 1 that control the randomness of generated text, with higher values increasing creativity. Default is 0.7.Expand source code
class ai: ''' This class generates a ai object. Containts all the information and methods to make requests to openAI chatGPT to run actions on the application. ### Attributes: - model (str): Model of GPT api to use. Default is gpt-3.5-turbo. - temp (float): Value between 0 and 1 that control the randomness of generated text, with higher values increasing creativity. Default is 0.7. ''' def __init__(self, config, org = None, api_key = None, model = "gpt-3.5-turbo", temp = 0.7): ''' ### Parameters: - config (obj): Pass the object created with class configfile with key for decryption and extra configuration if you are using connection manager. ### Optional Parameters: - org (str): A unique token identifying the user organization to interact with the API. - api_key (str): A unique authentication token required to access and interact with the API. - model (str): Model of GPT api to use. Default is gpt-3.5-turbo. - temp (float): Value between 0 and 1 that control the randomness of generated text, with higher values increasing creativity. Default is 0.7. ''' self.config = config if org: openai.organization = org else: try: openai.organization = self.config.config["openai"]["organization"] except: raise ValueError("Missing openai organization") if api_key: openai.api_key = api_key else: try: openai.api_key = self.config.config["openai"]["api_key"] except: raise ValueError("Missing openai api_key") self.__prompt = {} self.__prompt["original_system"] = """ When provided with user input for an SSH connection management application, analyze the input and extract the following information: - app_related: True if the input is related to the application's purpose and the request is understood; False if the input is not related, not understood, or if mandatory information like filter is missing. - type: Given a user input, identify the type of request they want to make. The input will represent one of two options: 1. "command" - The user wants to get information from devices by running commands. 2. "list_nodes" - The user wants to get a list of nodes, devices, servers, or routers. Response wich command or list_nodes type depending on the request. - filter: One or more regex patterns indicating the device or group of devices the command should be run on, returned as a Python list (e.g., ['hostname', 'hostname@folder', '@subfolder@folder']). The filter can have different formats, such as: - hostname - hostname@folder - hostname@subfolder@folder - partofhostname - @folder - @subfolder@folder - regex_pattern The filter should be extracted from the user input exactly as it was provided. Always preserve the exact filter pattern provided by the user, with no modifications. Do not process any regex, the application can do that. If no filter is specified, set it to None. - Expected: A value representing an expected output to search for when running the command. For the user this is a optional value. Set it to 'None' if no value was captured. expected value should ALWAYS come from the user input explicitly. - response: An optional field to be filled when app_related is False or when providing an explanation related to the app. This is where you can engage in small talk, answer questions not related to the app, or provide explanations about the extracted information. Always respond in the following format: app_related: {{app_related}} Type: {{command}} Filter: {{filter}} Expected: {{expected}} Response: {{response}} """ self.__prompt["original_user"] = "Get the IP addresses of loopback0 for all routers from w2az1 and e1.*(prod|dev) and check if they have the ip 192.168.1.1" self.__prompt["original_assistant"] = "app_related: True\nType: Command\nFilter: ['w2az1', 'e1.*(prod|dev)']\nExpected: 192.168.1.1" self.__prompt["command_system"] = """ For each device listed below, provide the command(s) needed to perform the specified action, depending on the device OS (e.g., Cisco IOSXR router, Linux server). Always format your response as a Python list (e.g., ['command1', 'command2']). Note that the application knows how to connect to devices via SSH, so you only need to provide the command(s) to run after connecting. If the commands needed are not for the specific OS type, just send an empty list (e.g., []). It is crucial to always include the device name provided in your response, even when there is only one device. your response has to be always like this: node1: ["command1", "command2"] node2: ["command1", "command2", "command3"] node1@folder: ["command1"] Node4@subfolder@folder: [] """ self.__prompt["command_user"]= """ input: show me the full configuration for all this devices: Devices: router1: cisco ios """ self.__prompt["command_assistant"]= """ router1: ['show running-config'] """ self.__prompt["confirmation_system"] = """ Please analyze the user's input and categorize it as either an affirmation or negation. Based on this analysis, respond with: 'True' if the input is an affirmation like 'do it', 'go ahead', 'sure', etc. 'False' if the input is a negation. If the input does not fit into either of these categories, kindly express that you didn't understand and request the user to rephrase their response. """ self.__prompt["confirmation_user"] = "Yes go ahead!" self.__prompt["confirmation_assistant"] = "True" self.model = model self.temp = temp def _retry_function(self, function, max_retries, backoff_num, *args): #Retry openai requests retries = 0 while retries < max_retries: try: myfunction = function(*args) break except (openai.error.APIConnectionError, openai.error.RateLimitError): wait_time = backoff_num * (2 ** retries) time.sleep(wait_time) retries += 1 continue if retries == max_retries: myfunction = False return myfunction def _clean_original_response(self, raw_response): #Parse response for first request to openAI GPT. info_dict = {} info_dict["app_related"] = False current_key = "response" for line in raw_response.split("\n"): if line.strip() == "": line = "\n" possible_keys = ["app_related", "type", "filter", "expected", "response"] if ':' in line and (key := line.split(':', 1)[0].strip().lower()) in possible_keys: key, value = line.split(":", 1) key = key.strip().lower() value = value.strip() # Convert "true" or "false" (case-insensitive) to Python boolean if value.lower() == "true": value = True elif value.lower() == "false": value = False elif value.lower() == "none": value = None if key == "filter": value = ast.literal_eval(value) #store in dictionary info_dict[key] = value current_key = key else: if current_key == "response": if "response" in info_dict: info_dict[current_key] += "\n" + line else: info_dict[current_key] = line return info_dict def _clean_command_response(self, raw_response): #Parse response for command request to openAI GPT. info_dict = {} info_dict["commands"] = [] info_dict["variables"] = {} info_dict["variables"]["__global__"] = {} for line in raw_response.split("\n"): if ":" in line: key, value = line.split(":", 1) key = key.strip() newvalue = {} pattern = r'\[.*?\]' match = re.search(pattern, value.strip()) try: value = ast.literal_eval(match.group(0)) for i,e in enumerate(value, start=1): newvalue[f"command{i}"] = e if f"{{command{i}}}" not in info_dict["commands"]: info_dict["commands"].append(f"{{command{i}}}") info_dict["variables"]["__global__"][f"command{i}"] = "" info_dict["variables"][key] = newvalue except: pass return info_dict def _clean_confirmation_response(self, raw_response): #Parse response for confirmation request to openAI GPT. value = raw_response.strip() if value.strip(".").lower() == "true": value = True elif value.strip(".").lower() == "false": value = False return value def _get_commands(self, user_input, nodes): #Send the request for commands for each device to openAI GPT. output_list = [] for key, value in nodes.items(): tags = value.get('tags', {}) try: os_value = tags.get('os', '') except: os_value = "" output_list.append(f"{key}: {os_value}") output_str = "\n".join(output_list) command_input = f"input: {user_input}\n\nDevices:\n{output_str}" message = [] message.append({"role": "system", "content": dedent(self.__prompt["command_system"])}) message.append({"role": "user", "content": dedent(self.__prompt["command_user"])}) message.append({"role": "assistant", "content": dedent(self.__prompt["command_assistant"])}) message.append({"role": "user", "content": command_input}) response = openai.ChatCompletion.create( model=self.model, messages=message, temperature=self.temp ) output = {} output["dict_response"] = response output["raw_response"] = response["choices"][0]["message"]["content"] output["response"] = self._clean_command_response(output["raw_response"]) return output def _get_filter(self, user_input, chat_history = None): #Send the request to identify the filter and other attributes from the user input to GPT. message = [] message.append({"role": "system", "content": dedent(self.__prompt["original_system"])}) message.append({"role": "user", "content": dedent(self.__prompt["original_user"])}) message.append({"role": "assistant", "content": dedent(self.__prompt["original_assistant"])}) if not chat_history: chat_history = [] chat_history.append({"role": "user", "content": user_input}) message.extend(chat_history) response = openai.ChatCompletion.create( model=self.model, messages=message, temperature=self.temp, top_p=1 ) output = {} output["dict_response"] = response output["raw_response"] = response["choices"][0]["message"]["content"] chat_history.append({"role": "assistant", "content": output["raw_response"]}) output["chat_history"] = chat_history clear_response = self._clean_original_response(output["raw_response"]) output["response"] = self._clean_original_response(output["raw_response"]) return output def _get_confirmation(self, user_input): #Send the request to identify if user is confirming or denying the task message = [] message.append({"role": "system", "content": dedent(self.__prompt["confirmation_system"])}) message.append({"role": "user", "content": dedent(self.__prompt["confirmation_user"])}) message.append({"role": "assistant", "content": dedent(self.__prompt["confirmation_assistant"])}) message.append({"role": "user", "content": user_input}) response = openai.ChatCompletion.create( model=self.model, messages=message, temperature=self.temp, top_p=1 ) output = {} output["dict_response"] = response output["raw_response"] = response["choices"][0]["message"]["content"] output["response"] = self._clean_confirmation_response(output["raw_response"]) return output def confirm(self, user_input, max_retries=3, backoff_num=1): ''' Send the user input to openAI GPT and verify if response is afirmative or negative. ### Parameters: - user_input (str): User response confirming or denying. ### Optional Parameters: - max_retries (int): Maximum number of retries for gpt api. - backoff_num (int): Backoff factor for exponential wait time between retries. ### Returns: bool or str: True, False or str if AI coudn't understand the response ''' result = self._retry_function(self._get_confirmation, max_retries, backoff_num, user_input) if result: output = result["response"] else: output = f"{self.model} api is not responding right now, please try again later." return output def ask(self, user_input, dryrun = False, chat_history = None, max_retries=3, backoff_num=1): ''' Send the user input to openAI GPT and parse the response to run an action in the application. ### Parameters: - user_input (str): Request to send to openAI that will be parsed and returned to execute on the application. AI understands the following tasks: - Run a command on a group of devices. - List a group of devices. - Test a command on a group of devices and verify if the output contain an expected value. ### Optional Parameters: - dryrun (bool): Set to true to get the arguments to use to run in the app. Default is false and it will run the actions directly. - chat_history (list): List in gpt api format for the chat history. - max_retries (int): Maximum number of retries for gpt api. - backoff_num (int): Backoff factor for exponential wait time between retries. ### Returns: dict: Dictionary formed with the following keys: - input: User input received - app_related: True if GPT detected the request to be related to the application. - dryrun: True/False - response: If the request is not related to the app. this key will contain chatGPT answer. - action: The action detected by the AI to run in the app. - filter: If it was detected by the AI, the filter used to get the list of nodes to work on. - nodes: If it's not a dryrun, the list of nodes matched by the filter. - args: A dictionary of arguments required to run command(s) on the nodes. - result: A dictionary with the output of the commands or the test. - chat_history: The chat history between user and chatbot. It can be used as an attribute for next request. ''' output = {} original = self._retry_function(self._get_filter, max_retries, backoff_num, user_input, chat_history) if not original: output["app_related"] = False output["response"] = f"{self.model} api is not responding right now, please try again later." return output output["input"] = user_input output["app_related"] = original["response"]["app_related"] output["dryrun"] = dryrun output["chat_history"] = original["chat_history"] if not output["app_related"]: output["response"] = original["response"]["response"] else: type = original["response"]["type"].lower() if "filter" in original["response"]: output["filter"] = original["response"]["filter"] if not self.config.config["case"]: if isinstance(output["filter"], list): output["filter"] = [item.lower() for item in output["filter"]] else: output["filter"] = output["filter"].lower() if not dryrun or type == "command": thisnodes = self.config._getallnodesfull(output["filter"]) output["nodes"] = list(thisnodes.keys()) if not type == "command": output["action"] = "list_nodes" else: commands = self._retry_function(self._get_commands, max_retries, backoff_num, user_input, thisnodes) if not commands: output = [] output["app_related"] = False output["response"] = f"{self.model} api is not responding right now, please try again later." return output output["args"] = {} output["args"]["commands"] = commands["response"]["commands"] output["args"]["vars"] = commands["response"]["variables"] if original["response"]["expected"]: output["args"]["expected"] = original["response"]["expected"] output["action"] = "test" else: output["action"] = "run" if dryrun: output["task"] = [] if output["action"] == "test": output["task"].append({"Task": "Verify if expected value is in command(s) output"}) output["task"].append({"Expected value to verify": output["args"]["expected"]}) elif output["action"] == "run": output["task"].append({"Task": "Run command(s) on devices and return output"}) varstocommands = deepcopy(output["args"]["vars"]) del varstocommands["__global__"] output["task"].append({"Devices": varstocommands}) if not dryrun: mynodes = nodes(self.config.getitems(output["nodes"]),config=self.config) if output["action"] == "test": output["result"] = mynodes.test(**output["args"]) output["logs"] = mynodes.output elif output["action"] == "run": output["result"] = mynodes.run(**output["args"]) return outputMethods
def ask(self, user_input, dryrun=False, chat_history=None, max_retries=3, backoff_num=1)-
Send the user input to openAI GPT and parse the response to run an action in the application.
Parameters:
- user_input (str): Request to send to openAI that will be parsed and returned to execute on the application. AI understands the following tasks: - Run a command on a group of devices. - List a group of devices. - Test a command on a group of devices and verify if the output contain an expected value.Optional Parameters:
- dryrun (bool): Set to true to get the arguments to use to run in the app. Default is false and it will run the actions directly. - chat_history (list): List in gpt api format for the chat history. - max_retries (int): Maximum number of retries for gpt api. - backoff_num (int): Backoff factor for exponential wait time between retries.Returns:
dict: Dictionary formed with the following keys: - input: User input received - app_related: True if GPT detected the request to be related to the application. - dryrun: True/False - response: If the request is not related to the app. this key will contain chatGPT answer. - action: The action detected by the AI to run in the app. - filter: If it was detected by the AI, the filter used to get the list of nodes to work on. - nodes: If it's not a dryrun, the list of nodes matched by the filter. - args: A dictionary of arguments required to run command(s) on the nodes. - result: A dictionary with the output of the commands or the test. - chat_history: The chat history between user and chatbot. It can be used as an attribute for next request.Expand source code
def ask(self, user_input, dryrun = False, chat_history = None, max_retries=3, backoff_num=1): ''' Send the user input to openAI GPT and parse the response to run an action in the application. ### Parameters: - user_input (str): Request to send to openAI that will be parsed and returned to execute on the application. AI understands the following tasks: - Run a command on a group of devices. - List a group of devices. - Test a command on a group of devices and verify if the output contain an expected value. ### Optional Parameters: - dryrun (bool): Set to true to get the arguments to use to run in the app. Default is false and it will run the actions directly. - chat_history (list): List in gpt api format for the chat history. - max_retries (int): Maximum number of retries for gpt api. - backoff_num (int): Backoff factor for exponential wait time between retries. ### Returns: dict: Dictionary formed with the following keys: - input: User input received - app_related: True if GPT detected the request to be related to the application. - dryrun: True/False - response: If the request is not related to the app. this key will contain chatGPT answer. - action: The action detected by the AI to run in the app. - filter: If it was detected by the AI, the filter used to get the list of nodes to work on. - nodes: If it's not a dryrun, the list of nodes matched by the filter. - args: A dictionary of arguments required to run command(s) on the nodes. - result: A dictionary with the output of the commands or the test. - chat_history: The chat history between user and chatbot. It can be used as an attribute for next request. ''' output = {} original = self._retry_function(self._get_filter, max_retries, backoff_num, user_input, chat_history) if not original: output["app_related"] = False output["response"] = f"{self.model} api is not responding right now, please try again later." return output output["input"] = user_input output["app_related"] = original["response"]["app_related"] output["dryrun"] = dryrun output["chat_history"] = original["chat_history"] if not output["app_related"]: output["response"] = original["response"]["response"] else: type = original["response"]["type"].lower() if "filter" in original["response"]: output["filter"] = original["response"]["filter"] if not self.config.config["case"]: if isinstance(output["filter"], list): output["filter"] = [item.lower() for item in output["filter"]] else: output["filter"] = output["filter"].lower() if not dryrun or type == "command": thisnodes = self.config._getallnodesfull(output["filter"]) output["nodes"] = list(thisnodes.keys()) if not type == "command": output["action"] = "list_nodes" else: commands = self._retry_function(self._get_commands, max_retries, backoff_num, user_input, thisnodes) if not commands: output = [] output["app_related"] = False output["response"] = f"{self.model} api is not responding right now, please try again later." return output output["args"] = {} output["args"]["commands"] = commands["response"]["commands"] output["args"]["vars"] = commands["response"]["variables"] if original["response"]["expected"]: output["args"]["expected"] = original["response"]["expected"] output["action"] = "test" else: output["action"] = "run" if dryrun: output["task"] = [] if output["action"] == "test": output["task"].append({"Task": "Verify if expected value is in command(s) output"}) output["task"].append({"Expected value to verify": output["args"]["expected"]}) elif output["action"] == "run": output["task"].append({"Task": "Run command(s) on devices and return output"}) varstocommands = deepcopy(output["args"]["vars"]) del varstocommands["__global__"] output["task"].append({"Devices": varstocommands}) if not dryrun: mynodes = nodes(self.config.getitems(output["nodes"]),config=self.config) if output["action"] == "test": output["result"] = mynodes.test(**output["args"]) output["logs"] = mynodes.output elif output["action"] == "run": output["result"] = mynodes.run(**output["args"]) return output def confirm(self, user_input, max_retries=3, backoff_num=1)-
Send the user input to openAI GPT and verify if response is afirmative or negative.
Parameters:
- user_input (str): User response confirming or denying.Optional Parameters:
- max_retries (int): Maximum number of retries for gpt api. - backoff_num (int): Backoff factor for exponential wait time between retries.Returns:
bool or str: True, False or str if AI coudn't understand the responseExpand source code
def confirm(self, user_input, max_retries=3, backoff_num=1): ''' Send the user input to openAI GPT and verify if response is afirmative or negative. ### Parameters: - user_input (str): User response confirming or denying. ### Optional Parameters: - max_retries (int): Maximum number of retries for gpt api. - backoff_num (int): Backoff factor for exponential wait time between retries. ### Returns: bool or str: True, False or str if AI coudn't understand the response ''' result = self._retry_function(self._get_confirmation, max_retries, backoff_num, user_input) if result: output = result["response"] else: output = f"{self.model} api is not responding right now, please try again later." return output
class configfile (conf=None, key=None)-
This class generates a configfile object. Containts a dictionary storing, config, nodes and profiles, normaly used by connection manager.
Attributes:
- file (str): Path/file to config file. - key (str): Path/file to RSA key file. - config (dict): Dictionary containing information of connection manager configuration. - connections (dict): Dictionary containing all the nodes added to connection manager. - profiles (dict): Dictionary containing all the profiles added to connection manager. - privatekey (obj): Object containing the private key to encrypt passwords. - publickey (obj): Object containing the public key to decrypt passwords.Optional Parameters:
- conf (str): Path/file to config file. If left empty default path is ~/.config/conn/config.json - key (str): Path/file to RSA key file. If left empty default path is ~/.config/conn/.oskExpand source code
class configfile: ''' This class generates a configfile object. Containts a dictionary storing, config, nodes and profiles, normaly used by connection manager. ### Attributes: - file (str): Path/file to config file. - key (str): Path/file to RSA key file. - config (dict): Dictionary containing information of connection manager configuration. - connections (dict): Dictionary containing all the nodes added to connection manager. - profiles (dict): Dictionary containing all the profiles added to connection manager. - privatekey (obj): Object containing the private key to encrypt passwords. - publickey (obj): Object containing the public key to decrypt passwords. ''' def __init__(self, conf = None, key = None): ''' ### Optional Parameters: - conf (str): Path/file to config file. If left empty default path is ~/.config/conn/config.json - key (str): Path/file to RSA key file. If left empty default path is ~/.config/conn/.osk ''' home = os.path.expanduser("~") defaultdir = home + '/.config/conn' Path(defaultdir).mkdir(parents=True, exist_ok=True) pathfile = defaultdir + '/.folder' try: with open(pathfile, "r") as f: configdir = f.read().strip() except: with open(pathfile, "w") as f: f.write(str(defaultdir)) configdir = defaultdir defaultfile = configdir + '/config.json' defaultkey = configdir + '/.osk' if conf == None: self.file = defaultfile else: self.file = conf if key == None: self.key = defaultkey else: self.key = key if os.path.exists(self.file): config = self._loadconfig(self.file) else: config = self._createconfig(self.file) self.config = config["config"] self.connections = config["connections"] self.profiles = config["profiles"] if not os.path.exists(self.key): self._createkey(self.key) with open(self.key) as f: self.privatekey = RSA.import_key(f.read()) f.close() self.publickey = self.privatekey.publickey() def _loadconfig(self, conf): #Loads config file jsonconf = open(conf) jsondata = json.load(jsonconf) jsonconf.close() return jsondata def _createconfig(self, conf): #Create config file defaultconfig = {'config': {'case': False, 'idletime': 30, 'fzf': False}, 'connections': {}, 'profiles': { "default": { "host":"", "protocol":"ssh", "port":"", "user":"", "password":"", "options":"", "logs":"", "tags": "" }}} if not os.path.exists(conf): with open(conf, "w") as f: json.dump(defaultconfig, f, indent = 4) f.close() os.chmod(conf, 0o600) jsonconf = open(conf) jsondata = json.load(jsonconf) jsonconf.close() return jsondata def _saveconfig(self, conf): #Save config file newconfig = {"config":{}, "connections": {}, "profiles": {}} newconfig["config"] = self.config newconfig["connections"] = self.connections newconfig["profiles"] = self.profiles with open(conf, "w") as f: json.dump(newconfig, f, indent = 4) f.close() def _createkey(self, keyfile): #Create key file key = RSA.generate(2048) with open(keyfile,'wb') as f: f.write(key.export_key('PEM')) f.close() os.chmod(keyfile, 0o600) return key def _explode_unique(self, unique): #Divide unique name into folder, subfolder and id uniques = unique.split("@") if not unique.startswith("@"): result = {"id": uniques[0]} else: result = {} if len(uniques) == 2: result["folder"] = uniques[1] if result["folder"] == "": return False elif len(uniques) == 3: result["folder"] = uniques[2] result["subfolder"] = uniques[1] if result["folder"] == "" or result["subfolder"] == "": return False elif len(uniques) > 3: return False return result def getitem(self, unique, keys = None): ''' Get an node or a group of nodes from configfile which can be passed to node/nodes class ### Parameters: - unique (str): Unique name of the node or folder in config using connection manager style: node[@subfolder][@folder] or [@subfolder]@folder ### Optional Parameters: - keys (list): In case you pass a folder as unique, you can filter nodes inside the folder passing a list. ### Returns: dict: Dictionary containing information of node or multiple dictionaries of multiple nodes. ''' uniques = self._explode_unique(unique) if unique.startswith("@"): if uniques.keys() >= {"folder", "subfolder"}: folder = self.connections[uniques["folder"]][uniques["subfolder"]] else: folder = self.connections[uniques["folder"]] newfolder = deepcopy(folder) newfolder.pop("type") for node in folder.keys(): if node == "type": continue if "type" in newfolder[node].keys(): if newfolder[node]["type"] == "subfolder": newfolder.pop(node) else: newfolder[node].pop("type") if keys == None: newfolder = {"{}{}".format(k,unique):v for k,v in newfolder.items()} return newfolder else: f_newfolder = dict((k, newfolder[k]) for k in keys) f_newfolder = {"{}{}".format(k,unique):v for k,v in f_newfolder.items()} return f_newfolder else: if uniques.keys() >= {"folder", "subfolder"}: node = self.connections[uniques["folder"]][uniques["subfolder"]][uniques["id"]] elif "folder" in uniques.keys(): node = self.connections[uniques["folder"]][uniques["id"]] else: node = self.connections[uniques["id"]] newnode = deepcopy(node) newnode.pop("type") return newnode def getitems(self, uniques): ''' Get a group of nodes from configfile which can be passed to node/nodes class ### Parameters: - uniques (str/list): Regex string name that will match hostnames from the connection manager. It can be a list of strings. ### Returns: dict: Dictionary containing information of node or multiple dictionaries of multiple nodes. ''' nodes = {} for i in uniques: if isinstance(i, dict): name = list(i.keys())[0] mylist = i[name] if not self.config["case"]: name = name.lower() mylist = [item.lower() for item in mylist] this = self.getitem(name, mylist) nodes.update(this) elif i.startswith("@"): if not self.config["case"]: i = i.lower() this = self.getitem(i) nodes.update(this) else: if not self.config["case"]: i = i.lower() this = self.getitem(i) nodes[i] = this return nodes def _connections_add(self,*, id, host, folder='', subfolder='', options='', logs='', password='', port='', protocol='', user='', tags='', type = "connection" ): #Add connection from config if folder == '': self.connections[id] = {"host": host, "options": options, "logs": logs, "password": password, "port": port, "protocol": protocol, "user": user, "tags": tags,"type": type} elif folder != '' and subfolder == '': self.connections[folder][id] = {"host": host, "options": options, "logs": logs, "password": password, "port": port, "protocol": protocol, "user": user, "tags": tags, "type": type} elif folder != '' and subfolder != '': self.connections[folder][subfolder][id] = {"host": host, "options": options, "logs": logs, "password": password, "port": port, "protocol": protocol, "user": user, "tags": tags, "type": type} def _connections_del(self,*, id, folder='', subfolder=''): #Delete connection from config if folder == '': del self.connections[id] elif folder != '' and subfolder == '': del self.connections[folder][id] elif folder != '' and subfolder != '': del self.connections[folder][subfolder][id] def _folder_add(self,*, folder, subfolder = ''): #Add Folder from config if subfolder == '': if folder not in self.connections: self.connections[folder] = {"type": "folder"} else: if subfolder not in self.connections[folder]: self.connections[folder][subfolder] = {"type": "subfolder"} def _folder_del(self,*, folder, subfolder=''): #Delete folder from config if subfolder == '': del self.connections[folder] else: del self.connections[folder][subfolder] def _profiles_add(self,*, id, host = '', options='', logs='', password='', port='', protocol='', user='', tags='' ): #Add profile from config self.profiles[id] = {"host": host, "options": options, "logs": logs, "password": password, "port": port, "protocol": protocol, "user": user, "tags": tags} def _profiles_del(self,*, id ): #Delete profile from config del self.profiles[id] def _getallnodes(self, filter = None): #get all nodes on configfile nodes = [] layer1 = [k for k,v in self.connections.items() if isinstance(v, dict) and v["type"] == "connection"] folders = [k for k,v in self.connections.items() if isinstance(v, dict) and v["type"] == "folder"] nodes.extend(layer1) for f in folders: layer2 = [k + "@" + f for k,v in self.connections[f].items() if isinstance(v, dict) and v["type"] == "connection"] nodes.extend(layer2) subfolders = [k for k,v in self.connections[f].items() if isinstance(v, dict) and v["type"] == "subfolder"] for s in subfolders: layer3 = [k + "@" + s + "@" + f for k,v in self.connections[f][s].items() if isinstance(v, dict) and v["type"] == "connection"] nodes.extend(layer3) if filter: if isinstance(filter, str): nodes = [item for item in nodes if re.search(filter, item)] elif isinstance(filter, list): nodes = [item for item in nodes if any(re.search(pattern, item) for pattern in filter)] else: raise ValueError("filter must be a string or a list of strings") return nodes def _getallnodesfull(self, filter = None): #get all nodes on configfile with all their attributes. nodes = {} layer1 = {k:v for k,v in self.connections.items() if isinstance(v, dict) and v["type"] == "connection"} folders = [k for k,v in self.connections.items() if isinstance(v, dict) and v["type"] == "folder"] nodes.update(layer1) for f in folders: layer2 = {k + "@" + f:v for k,v in self.connections[f].items() if isinstance(v, dict) and v["type"] == "connection"} nodes.update(layer2) subfolders = [k for k,v in self.connections[f].items() if isinstance(v, dict) and v["type"] == "subfolder"] for s in subfolders: layer3 = {k + "@" + s + "@" + f:v for k,v in self.connections[f][s].items() if isinstance(v, dict) and v["type"] == "connection"} nodes.update(layer3) if filter: if isinstance(filter, str): nodes = {k: v for k, v in nodes.items() if re.search(filter, k)} elif isinstance(filter, list): nodes = {k: v for k, v in nodes.items() if any(re.search(pattern, k) for pattern in filter)} else: raise ValueError("filter must be a string or a list of strings") for node, keys in nodes.items(): for key, value in keys.items(): profile = re.search("^@(.*)", str(value)) if profile: try: nodes[node][key] = self.profiles[profile.group(1)][key] except: nodes[node][key] = "" elif value == '' and key == "protocol": try: nodes[node][key] = config.profiles["default"][key] except: nodes[node][key] = "ssh" return nodes def _getallfolders(self): #get all folders on configfile folders = ["@" + k for k,v in self.connections.items() if isinstance(v, dict) and v["type"] == "folder"] subfolders = [] for f in folders: s = ["@" + k + f for k,v in self.connections[f[1:]].items() if isinstance(v, dict) and v["type"] == "subfolder"] subfolders.extend(s) folders.extend(subfolders) return folders def _profileused(self, profile): #Check if profile is used before deleting it nodes = [] layer1 = [k for k,v in self.connections.items() if isinstance(v, dict) and v["type"] == "connection" and ("@" + profile in v.values() or ( isinstance(v["password"],list) and "@" + profile in v["password"]))] folders = [k for k,v in self.connections.items() if isinstance(v, dict) and v["type"] == "folder"] nodes.extend(layer1) for f in folders: layer2 = [k + "@" + f for k,v in self.connections[f].items() if isinstance(v, dict) and v["type"] == "connection" and ("@" + profile in v.values() or ( isinstance(v["password"],list) and "@" + profile in v["password"]))] nodes.extend(layer2) subfolders = [k for k,v in self.connections[f].items() if isinstance(v, dict) and v["type"] == "subfolder"] for s in subfolders: layer3 = [k + "@" + s + "@" + f for k,v in self.connections[f][s].items() if isinstance(v, dict) and v["type"] == "connection" and ("@" + profile in v.values() or ( isinstance(v["password"],list) and "@" + profile in v["password"]))] nodes.extend(layer3) return nodesMethods
def getitem(self, unique, keys=None)-
Get an node or a group of nodes from configfile which can be passed to node/nodes class
Parameters:
- unique (str): Unique name of the node or folder in config using connection manager style: node[@subfolder][@folder] or [@subfolder]@folderOptional Parameters:
- keys (list): In case you pass a folder as unique, you can filter nodes inside the folder passing a list.Returns:
dict: Dictionary containing information of node or multiple dictionaries of multiple nodes.Expand source code
def getitem(self, unique, keys = None): ''' Get an node or a group of nodes from configfile which can be passed to node/nodes class ### Parameters: - unique (str): Unique name of the node or folder in config using connection manager style: node[@subfolder][@folder] or [@subfolder]@folder ### Optional Parameters: - keys (list): In case you pass a folder as unique, you can filter nodes inside the folder passing a list. ### Returns: dict: Dictionary containing information of node or multiple dictionaries of multiple nodes. ''' uniques = self._explode_unique(unique) if unique.startswith("@"): if uniques.keys() >= {"folder", "subfolder"}: folder = self.connections[uniques["folder"]][uniques["subfolder"]] else: folder = self.connections[uniques["folder"]] newfolder = deepcopy(folder) newfolder.pop("type") for node in folder.keys(): if node == "type": continue if "type" in newfolder[node].keys(): if newfolder[node]["type"] == "subfolder": newfolder.pop(node) else: newfolder[node].pop("type") if keys == None: newfolder = {"{}{}".format(k,unique):v for k,v in newfolder.items()} return newfolder else: f_newfolder = dict((k, newfolder[k]) for k in keys) f_newfolder = {"{}{}".format(k,unique):v for k,v in f_newfolder.items()} return f_newfolder else: if uniques.keys() >= {"folder", "subfolder"}: node = self.connections[uniques["folder"]][uniques["subfolder"]][uniques["id"]] elif "folder" in uniques.keys(): node = self.connections[uniques["folder"]][uniques["id"]] else: node = self.connections[uniques["id"]] newnode = deepcopy(node) newnode.pop("type") return newnode def getitems(self, uniques)-
Get a group of nodes from configfile which can be passed to node/nodes class
Parameters:
- uniques (str/list): Regex string name that will match hostnames from the connection manager. It can be a list of strings.Returns:
dict: Dictionary containing information of node or multiple dictionaries of multiple nodes.Expand source code
def getitems(self, uniques): ''' Get a group of nodes from configfile which can be passed to node/nodes class ### Parameters: - uniques (str/list): Regex string name that will match hostnames from the connection manager. It can be a list of strings. ### Returns: dict: Dictionary containing information of node or multiple dictionaries of multiple nodes. ''' nodes = {} for i in uniques: if isinstance(i, dict): name = list(i.keys())[0] mylist = i[name] if not self.config["case"]: name = name.lower() mylist = [item.lower() for item in mylist] this = self.getitem(name, mylist) nodes.update(this) elif i.startswith("@"): if not self.config["case"]: i = i.lower() this = self.getitem(i) nodes.update(this) else: if not self.config["case"]: i = i.lower() this = self.getitem(i) nodes[i] = this return nodes
class connapp (config)-
This class starts the connection manager app. It's normally used by connection manager but you can use it on a script to run the connection manager your way and use a different configfile and key.
Parameters:
- config (obj): Object generated with configfile class, it contains the nodes configuration and the methods to manage the config file.Expand source code
class connapp: ''' This class starts the connection manager app. It's normally used by connection manager but you can use it on a script to run the connection manager your way and use a different configfile and key. ''' def __init__(self, config): ''' ### Parameters: - config (obj): Object generated with configfile class, it contains the nodes configuration and the methods to manage the config file. ''' self.node = node self.connnodes = nodes self.config = config self.nodes = self.config._getallnodes() self.folders = self.config._getallfolders() self.profiles = list(self.config.profiles.keys()) self.case = self.config.config["case"] try: self.fzf = self.config.config["fzf"] except: self.fzf = False def start(self,argv = sys.argv[1:]): ''' ### Parameters: - argv (list): List of arguments to pass to the app. Default: sys.argv[1:] ''' #DEFAULTPARSER defaultparser = argparse.ArgumentParser(prog = "conn", description = "SSH and Telnet connection manager", formatter_class=argparse.RawTextHelpFormatter) subparsers = defaultparser.add_subparsers(title="Commands") #NODEPARSER nodeparser = subparsers.add_parser("node",usage=self._help("usage"), help=self._help("node"),epilog=self._help("end"), formatter_class=argparse.RawTextHelpFormatter) nodecrud = nodeparser.add_mutually_exclusive_group() nodeparser.add_argument("node", metavar="node|folder", nargs='?', default=None, action=self._store_type, type=self._type_node, help=self._help("node")) nodecrud.add_argument("-v","--version", dest="action", action="store_const", help="Show version", const="version", default="connect") nodecrud.add_argument("-a","--add", dest="action", action="store_const", help="Add new node[@subfolder][@folder] or [@subfolder]@folder", const="add", default="connect") nodecrud.add_argument("-r","--del", "--rm", dest="action", action="store_const", help="Delete node[@subfolder][@folder] or [@subfolder]@folder", const="del", default="connect") nodecrud.add_argument("-e","--mod", "--edit", dest="action", action="store_const", help="Modify node[@subfolder][@folder]", const="mod", default="connect") nodecrud.add_argument("-s","--show", dest="action", action="store_const", help="Show node[@subfolder][@folder]", const="show", default="connect") nodecrud.add_argument("-d","--debug", dest="action", action="store_const", help="Display all conections steps", const="debug", default="connect") nodeparser.set_defaults(func=self._func_node) #PROFILEPARSER profileparser = subparsers.add_parser("profile", help="Manage profiles") profileparser.add_argument("profile", nargs=1, action=self._store_type, type=self._type_profile, help="Name of profile to manage") profilecrud = profileparser.add_mutually_exclusive_group(required=True) profilecrud.add_argument("-a", "--add", dest="action", action="store_const", help="Add new profile", const="add") profilecrud.add_argument("-r", "--del", "--rm", dest="action", action="store_const", help="Delete profile", const="del") profilecrud.add_argument("-e", "--mod", "--edit", dest="action", action="store_const", help="Modify profile", const="mod") profilecrud.add_argument("-s", "--show", dest="action", action="store_const", help="Show profile", const="show") profileparser.set_defaults(func=self._func_profile) #MOVEPARSER moveparser = subparsers.add_parser("move", aliases=["mv"], help="Move node") moveparser.add_argument("move", nargs=2, action=self._store_type, help="Move node[@subfolder][@folder] dest_node[@subfolder][@folder]", default="move", type=self._type_node) moveparser.set_defaults(func=self._func_others) #COPYPARSER copyparser = subparsers.add_parser("copy", aliases=["cp"], help="Copy node") copyparser.add_argument("cp", nargs=2, action=self._store_type, help="Copy node[@subfolder][@folder] new_node[@subfolder][@folder]", default="cp", type=self._type_node) copyparser.set_defaults(func=self._func_others) #LISTPARSER lsparser = subparsers.add_parser("list", aliases=["ls"], help="List profiles, nodes or folders") lsparser.add_argument("ls", action=self._store_type, choices=["profiles","nodes","folders"], help="List profiles, nodes or folders", default=False) lsparser.set_defaults(func=self._func_others) #BULKPARSER bulkparser = subparsers.add_parser("bulk", help="Add nodes in bulk") bulkparser.add_argument("bulk", const="bulk", nargs=0, action=self._store_type, help="Add nodes in bulk") bulkparser.set_defaults(func=self._func_others) #RUNPARSER runparser = subparsers.add_parser("run", help="Run scripts or commands on nodes", formatter_class=argparse.RawTextHelpFormatter) runparser.add_argument("run", nargs='+', action=self._store_type, help=self._help("run"), default="run") runparser.add_argument("-g","--generate", dest="action", action="store_const", help="Generate yaml file template", const="generate", default="run") runparser.set_defaults(func=self._func_run) #APIPARSER apiparser = subparsers.add_parser("api", help="Start and stop connpy api") apicrud = apiparser.add_mutually_exclusive_group(required=True) apicrud.add_argument("-s","--start", dest="start", nargs="?", action=self._store_type, help="Start conppy api", type=int, default=8048, metavar="PORT") apicrud.add_argument("-r","--restart", dest="restart", nargs=0, action=self._store_type, help="Restart conppy api") apicrud.add_argument("-x","--stop", dest="stop", nargs=0, action=self._store_type, help="Stop conppy api") apicrud.add_argument("-d", "--debug", dest="debug", nargs="?", action=self._store_type, help="Run connpy server on debug mode", type=int, default=8048, metavar="PORT") apiparser.set_defaults(func=self._func_api) #CONFIGPARSER configparser = subparsers.add_parser("config", help="Manage app config") configcrud = configparser.add_mutually_exclusive_group(required=True) configcrud.add_argument("--allow-uppercase", dest="case", nargs=1, action=self._store_type, help="Allow case sensitive names", choices=["true","false"]) configcrud.add_argument("--fzf", dest="fzf", nargs=1, action=self._store_type, help="Use fzf for lists", choices=["true","false"]) configcrud.add_argument("--keepalive", dest="idletime", nargs=1, action=self._store_type, help="Set keepalive time in seconds, 0 to disable", type=int, metavar="INT") configcrud.add_argument("--completion", dest="completion", nargs=1, choices=["bash","zsh"], action=self._store_type, help="Get terminal completion configuration for conn") configcrud.add_argument("--configfolder", dest="configfolder", nargs=1, action=self._store_type, help="Set the default location for config file", metavar="FOLDER") configcrud.add_argument("--openai", dest="openai", nargs=2, action=self._store_type, help="Set openai organization and api_key", metavar=("ORGANIZATION", "API_KEY")) configparser.set_defaults(func=self._func_others) #Manage sys arguments commands = ["node", "profile", "mv", "move","copy", "cp", "bulk", "ls", "list", "run", "config", "api"] profilecmds = ["--add", "-a", "--del", "--rm", "-r", "--mod", "--edit", "-e", "--show", "-s"] if len(argv) >= 2 and argv[1] == "profile" and argv[0] in profilecmds: argv[1] = argv[0] argv[0] = "profile" if len(argv) < 1 or argv[0] not in commands: argv.insert(0,"node") args = defaultparser.parse_args(argv) return args.func(args) class _store_type(argparse.Action): #Custom store type for cli app. def __call__(self, parser, args, values, option_string=None): setattr(args, "data", values) delattr(args,self.dest) setattr(args, "command", self.dest) def _func_node(self, args): #Function called when connecting or managing nodes. if not self.case and args.data != None: args.data = args.data.lower() actions = {"version": self._version, "connect": self._connect, "debug": self._connect, "add": self._add, "del": self._del, "mod": self._mod, "show": self._show} return actions.get(args.action)(args) def _version(self, args): print(__version__) def _connect(self, args): if args.data == None: matches = self.nodes if len(matches) == 0: print("There are no nodes created") print("try: conn --help") exit(9) else: if args.data.startswith("@"): matches = list(filter(lambda k: args.data in k, self.nodes)) else: matches = list(filter(lambda k: k.startswith(args.data), self.nodes)) if len(matches) == 0: print("{} not found".format(args.data)) exit(2) elif len(matches) > 1: matches[0] = self._choose(matches,"node", "connect") if matches[0] == None: exit(7) node = self.config.getitem(matches[0]) node = self.node(matches[0],**node, config = self.config) if args.action == "debug": node.interact(debug = True) else: node.interact() def _del(self, args): if args.data == None: print("Missing argument node") exit(3) elif args.data.startswith("@"): matches = list(filter(lambda k: k == args.data, self.folders)) else: matches = list(filter(lambda k: k == args.data, self.nodes)) if len(matches) == 0: print("{} not found".format(args.data)) exit(2) question = [inquirer.Confirm("delete", message="Are you sure you want to delete {}?".format(matches[0]))] confirm = inquirer.prompt(question) if confirm == None: exit(7) if confirm["delete"]: uniques = self.config._explode_unique(matches[0]) if args.data.startswith("@"): self.config._folder_del(**uniques) else: self.config._connections_del(**uniques) self.config._saveconfig(self.config.file) print("{} deleted succesfully".format(matches[0])) def _add(self, args): if args.data == None: print("Missing argument node") exit(3) elif args.data.startswith("@"): type = "folder" matches = list(filter(lambda k: k == args.data, self.folders)) reversematches = list(filter(lambda k: "@" + k == args.data, self.nodes)) else: type = "node" matches = list(filter(lambda k: k == args.data, self.nodes)) reversematches = list(filter(lambda k: k == "@" + args.data, self.folders)) if len(matches) > 0: print("{} already exist".format(matches[0])) exit(4) if len(reversematches) > 0: print("{} already exist".format(reversematches[0])) exit(4) else: if type == "folder": uniques = self.config._explode_unique(args.data) if uniques == False: print("Invalid folder {}".format(args.data)) exit(5) if "subfolder" in uniques.keys(): parent = "@" + uniques["folder"] if parent not in self.folders: print("Folder {} not found".format(uniques["folder"])) exit(2) self.config._folder_add(**uniques) self.config._saveconfig(self.config.file) print("{} added succesfully".format(args.data)) if type == "node": nodefolder = args.data.partition("@") nodefolder = "@" + nodefolder[2] if nodefolder not in self.folders and nodefolder != "@": print(nodefolder + " not found") exit(2) uniques = self.config._explode_unique(args.data) if uniques == False: print("Invalid node {}".format(args.data)) exit(5) print("You can use the configured setting in a profile using @profilename.") print("You can also leave empty any value except hostname/IP.") print("You can pass 1 or more passwords using comma separated @profiles") print("You can use this variables on logging file name: ${id} ${unique} ${host} ${port} ${user} ${protocol}") print("Some useful tags to set for automation are 'os', 'screen_length_command', and 'prompt'.") newnode = self._questions_nodes(args.data, uniques) if newnode == False: exit(7) self.config._connections_add(**newnode) self.config._saveconfig(self.config.file) print("{} added succesfully".format(args.data)) def _show(self, args): if args.data == None: print("Missing argument node") exit(3) matches = list(filter(lambda k: k == args.data, self.nodes)) if len(matches) == 0: print("{} not found".format(args.data)) exit(2) node = self.config.getitem(matches[0]) for k, v in node.items(): if isinstance(v, str): print(k + ": " + v) else: print(k + ":") for i in v: print(" - " + i) def _mod(self, args): if args.data == None: print("Missing argument node") exit(3) matches = list(filter(lambda k: k == args.data, self.nodes)) if len(matches) == 0: print("{} not found".format(args.data)) exit(2) node = self.config.getitem(matches[0]) edits = self._questions_edit() if edits == None: exit(7) uniques = self.config._explode_unique(args.data) updatenode = self._questions_nodes(args.data, uniques, edit=edits) if not updatenode: exit(7) uniques.update(node) uniques["type"] = "connection" if sorted(updatenode.items()) == sorted(uniques.items()): print("Nothing to do here") return else: self.config._connections_add(**updatenode) self.config._saveconfig(self.config.file) print("{} edited succesfully".format(args.data)) def _func_profile(self, args): #Function called when managing profiles if not self.case: args.data[0] = args.data[0].lower() actions = {"add": self._profile_add, "del": self._profile_del, "mod": self._profile_mod, "show": self._profile_show} return actions.get(args.action)(args) def _profile_del(self, args): matches = list(filter(lambda k: k == args.data[0], self.profiles)) if len(matches) == 0: print("{} not found".format(args.data[0])) exit(2) if matches[0] == "default": print("Can't delete default profile") exit(6) usedprofile = self.config._profileused(matches[0]) if len(usedprofile) > 0: print("Profile {} used in the following nodes:".format(matches[0])) print(", ".join(usedprofile)) exit(8) question = [inquirer.Confirm("delete", message="Are you sure you want to delete {}?".format(matches[0]))] confirm = inquirer.prompt(question) if confirm["delete"]: self.config._profiles_del(id = matches[0]) self.config._saveconfig(self.config.file) print("{} deleted succesfully".format(matches[0])) def _profile_show(self, args): matches = list(filter(lambda k: k == args.data[0], self.profiles)) if len(matches) == 0: print("{} not found".format(args.data[0])) exit(2) profile = self.config.profiles[matches[0]] for k, v in profile.items(): if isinstance(v, str): print(k + ": " + v) else: print(k + ":") for i in v: print(" - " + i) def _profile_add(self, args): matches = list(filter(lambda k: k == args.data[0], self.profiles)) if len(matches) > 0: print("Profile {} Already exist".format(matches[0])) exit(4) newprofile = self._questions_profiles(args.data[0]) if newprofile == False: exit(7) self.config._profiles_add(**newprofile) self.config._saveconfig(self.config.file) print("{} added succesfully".format(args.data[0])) def _profile_mod(self, args): matches = list(filter(lambda k: k == args.data[0], self.profiles)) if len(matches) == 0: print("{} not found".format(args.data[0])) exit(2) profile = self.config.profiles[matches[0]] oldprofile = {"id": matches[0]} oldprofile.update(profile) edits = self._questions_edit() if edits == None: exit(7) updateprofile = self._questions_profiles(matches[0], edit=edits) if not updateprofile: exit(7) if sorted(updateprofile.items()) == sorted(oldprofile.items()): print("Nothing to do here") return else: self.config._profiles_add(**updateprofile) self.config._saveconfig(self.config.file) print("{} edited succesfully".format(args.data[0])) def _func_others(self, args): #Function called when using other commands actions = {"ls": self._ls, "move": self._mvcp, "cp": self._mvcp, "bulk": self._bulk, "completion": self._completion, "case": self._case, "fzf": self._fzf, "idletime": self._idletime, "configfolder": self._configfolder, "openai": self._openai} return actions.get(args.command)(args) def _ls(self, args): print(*getattr(self, args.data), sep="\n") def _mvcp(self, args): if not self.case: args.data[0] = args.data[0].lower() args.data[1] = args.data[1].lower() source = list(filter(lambda k: k == args.data[0], self.nodes)) dest = list(filter(lambda k: k == args.data[1], self.nodes)) if len(source) != 1: print("{} not found".format(args.data[0])) exit(2) if len(dest) > 0: print("Node {} Already exist".format(args.data[1])) exit(4) nodefolder = args.data[1].partition("@") nodefolder = "@" + nodefolder[2] if nodefolder not in self.folders and nodefolder != "@": print("{} not found".format(nodefolder)) exit(2) olduniques = self.config._explode_unique(args.data[0]) newuniques = self.config._explode_unique(args.data[1]) if newuniques == False: print("Invalid node {}".format(args.data[1])) exit(5) node = self.config.getitem(source[0]) newnode = {**newuniques, **node} self.config._connections_add(**newnode) if args.command == "move": self.config._connections_del(**olduniques) self.config._saveconfig(self.config.file) action = "moved" if args.command == "move" else "copied" print("{} {} succesfully to {}".format(args.data[0],action, args.data[1])) def _bulk(self, args): newnodes = self._questions_bulk() if newnodes == False: exit(7) if not self.case: newnodes["location"] = newnodes["location"].lower() newnodes["ids"] = newnodes["ids"].lower() ids = newnodes["ids"].split(",") hosts = newnodes["host"].split(",") count = 0 for n in ids: unique = n + newnodes["location"] matches = list(filter(lambda k: k == unique, self.nodes)) reversematches = list(filter(lambda k: k == "@" + unique, self.folders)) if len(matches) > 0: print("Node {} already exist, ignoring it".format(unique)) continue if len(reversematches) > 0: print("Folder with name {} already exist, ignoring it".format(unique)) continue newnode = {"id": n} if newnodes["location"] != "": location = self.config._explode_unique(newnodes["location"]) newnode.update(location) if len(hosts) > 1: index = ids.index(n) newnode["host"] = hosts[index] else: newnode["host"] = hosts[0] newnode["protocol"] = newnodes["protocol"] newnode["port"] = newnodes["port"] newnode["options"] = newnodes["options"] newnode["logs"] = newnodes["logs"] newnode["tags"] = newnodes["tags"] newnode["user"] = newnodes["user"] newnode["password"] = newnodes["password"] count +=1 self.config._connections_add(**newnode) self.nodes = self.config._getallnodes() if count > 0: self.config._saveconfig(self.config.file) print("Succesfully added {} nodes".format(count)) else: print("0 nodes added") def _completion(self, args): if args.data[0] == "bash": print(self._help("bashcompletion")) elif args.data[0] == "zsh": print(self._help("zshcompletion")) def _case(self, args): if args.data[0] == "true": args.data[0] = True elif args.data[0] == "false": args.data[0] = False self._change_settings(args.command, args.data[0]) def _fzf(self, args): if args.data[0] == "true": args.data[0] = True elif args.data[0] == "false": args.data[0] = False self._change_settings(args.command, args.data[0]) def _idletime(self, args): if args.data[0] < 0: args.data[0] = 0 self._change_settings(args.command, args.data[0]) def _configfolder(self, args): if not os.path.isdir(args.data[0]): raise argparse.ArgumentTypeError(f"readable_dir:{args.data[0]} is not a valid path") else: pathfile = defaultdir + "/.folder" folder = os.path.abspath(args.data[0]).rstrip('/') with open(pathfile, "w") as f: f.write(str(folder)) print("Config saved") def _openai(self, args): openaikeys = {} openaikeys["organization"] = args.data[0] openaikeys["api_key"] = args.data[1] self._change_settings(args.command, openaikeys) def _change_settings(self, name, value): self.config.config[name] = value self.config._saveconfig(self.config.file) print("Config saved") def _func_run(self, args): if len(args.data) > 1: args.action = "noderun" actions = {"noderun": self._node_run, "generate": self._yaml_generate, "run": self._yaml_run} return actions.get(args.action)(args) def _func_api(self, args): if args.command == "stop" or args.command == "restart": args.data = stop_api() if args.command == "start" or args.command == "restart": if args.data: start_api(args.data) else: start_api() if args.command == "debug": if args.data: debug_api(args.data) else: debug_api() return def _node_run(self, args): command = " ".join(args.data[1:]) matches = list(filter(lambda k: k == args.data[0], self.nodes)) if len(matches) == 0: print("{} not found".format(args.data[0])) exit(2) node = self.config.getitem(matches[0]) node = self.node(matches[0],**node, config = self.config) node.run(command) print(node.output) def _yaml_generate(self, args): if os.path.exists(args.data[0]): print("File {} already exists".format(args.data[0])) exit(14) else: with open(args.data[0], "w") as file: file.write(self._help("generate")) file.close() print("File {} generated succesfully".format(args.data[0])) exit() def _yaml_run(self, args): try: with open(args.data[0]) as file: scripts = yaml.load(file, Loader=yaml.FullLoader) except: print("failed reading file {}".format(args.data[0])) exit(10) for script in scripts["tasks"]: args = {} try: action = script["action"] nodelist = script["nodes"] args["commands"] = script["commands"] output = script["output"] if action == "test": args["expected"] = script["expected"] except KeyError as e: print("'{}' is mandatory".format(e.args[0])) exit(11) nodes = self.connnodes(self.config.getitems(nodelist), config = self.config) stdout = False if output is None: pass elif output == "stdout": stdout = True elif isinstance(output, str) and action == "run": args["folder"] = output try: args["vars"] = script["variables"] except: pass try: options = script["options"] thisoptions = {k: v for k, v in options.items() if k in ["prompt", "parallel", "timeout"]} args.update(thisoptions) except: options = None size = str(os.get_terminal_size()) p = re.search(r'.*columns=([0-9]+)', size) columns = int(p.group(1)) if action == "run": nodes.run(**args) print(script["name"].upper() + "-" * (columns - len(script["name"]))) for i in nodes.status.keys(): print(" " + i + " " + "-" * (columns - len(i) - 13) + (" PASS(0)" if nodes.status[i] == 0 else " FAIL({})".format(nodes.status[i]))) if stdout: for line in nodes.output[i].splitlines(): print(" " + line) elif action == "test": nodes.test(**args) print(script["name"].upper() + "-" * (columns - len(script["name"]))) for i in nodes.status.keys(): print(" " + i + " " + "-" * (columns - len(i) - 13) + (" PASS(0)" if nodes.status[i] == 0 else " FAIL({})".format(nodes.status[i]))) if nodes.status[i] == 0: try: myexpected = args["expected"].format(**args["vars"][i]) except: try: myexpected = args["expected"].format(**args["vars"]["__global__"]) except: myexpected = args["expected"] print(" TEST for '{}' --> ".format(myexpected) + str(nodes.result[i]).upper()) if stdout: if nodes.status[i] == 0: print(" " + "-" * (len(myexpected) + 16 + len(str(nodes.result[i])))) for line in nodes.output[i].splitlines(): print(" " + line) else: print("Wrong action '{}'".format(action)) exit(13) def _choose(self, list, name, action): #Generates an inquirer list to pick if FzfPrompt and self.fzf: fzf = FzfPrompt(executable_path="fzf-tmux") if not self.case: fzf = FzfPrompt(executable_path="fzf-tmux -i") answer = fzf.prompt(list, fzf_options="-d 25%") if len(answer) == 0: return else: return answer[0] else: questions = [inquirer.List(name, message="Pick {} to {}:".format(name,action), choices=list, carousel=True)] answer = inquirer.prompt(questions) if answer == None: return else: return answer[name] def _host_validation(self, answers, current, regex = "^.+$"): #Validate hostname in inquirer when managing nodes if not re.match(regex, current): raise inquirer.errors.ValidationError("", reason="Host cannot be empty") if current.startswith("@"): if current[1:] not in self.profiles: raise inquirer.errors.ValidationError("", reason="Profile {} don't exist".format(current)) return True def _profile_protocol_validation(self, answers, current, regex = "(^ssh$|^telnet$|^$)"): #Validate protocol in inquirer when managing profiles if not re.match(regex, current): raise inquirer.errors.ValidationError("", reason="Pick between ssh, telnet or leave empty") return True def _protocol_validation(self, answers, current, regex = "(^ssh$|^telnet$|^$|^@.+$)"): #Validate protocol in inquirer when managing nodes if not re.match(regex, current): raise inquirer.errors.ValidationError("", reason="Pick between ssh, telnet, leave empty or @profile") if current.startswith("@"): if current[1:] not in self.profiles: raise inquirer.errors.ValidationError("", reason="Profile {} don't exist".format(current)) return True def _profile_port_validation(self, answers, current, regex = "(^[0-9]*$)"): #Validate port in inquirer when managing profiles if not re.match(regex, current): raise inquirer.errors.ValidationError("", reason="Pick a port between 1-65535, @profile o leave empty") try: port = int(current) except: port = 0 if current != "" and not 1 <= int(port) <= 65535: raise inquirer.errors.ValidationError("", reason="Pick a port between 1-65535 or leave empty") return True def _port_validation(self, answers, current, regex = "(^[0-9]*$|^@.+$)"): #Validate port in inquirer when managing nodes if not re.match(regex, current): raise inquirer.errors.ValidationError("", reason="Pick a port between 1-65535, @profile or leave empty") try: port = int(current) except: port = 0 if current.startswith("@"): if current[1:] not in self.profiles: raise inquirer.errors.ValidationError("", reason="Profile {} don't exist".format(current)) elif current != "" and not 1 <= int(port) <= 65535: raise inquirer.errors.ValidationError("", reason="Pick a port between 1-65535, @profile o leave empty") return True def _pass_validation(self, answers, current, regex = "(^@.+$)"): #Validate password in inquirer profiles = current.split(",") for i in profiles: if not re.match(regex, i) or i[1:] not in self.profiles: raise inquirer.errors.ValidationError("", reason="Profile {} don't exist".format(i)) return True def _tags_validation(self, answers, current): #Validation for Tags in inquirer when managing nodes if current.startswith("@"): if current[1:] not in self.profiles: raise inquirer.errors.ValidationError("", reason="Profile {} don't exist".format(current)) elif current != "": isdict = False try: isdict = ast.literal_eval(current) except: pass if not isinstance (isdict, dict): raise inquirer.errors.ValidationError("", reason="Tags should be a python dictionary.".format(current)) return True def _profile_tags_validation(self, answers, current): #Validation for Tags in inquirer when managing profiles if current != "": isdict = False try: isdict = ast.literal_eval(current) except: pass if not isinstance (isdict, dict): raise inquirer.errors.ValidationError("", reason="Tags should be a python dictionary.".format(current)) return True def _default_validation(self, answers, current): #Default validation type used in multiples questions in inquirer if current.startswith("@"): if current[1:] not in self.profiles: raise inquirer.errors.ValidationError("", reason="Profile {} don't exist".format(current)) return True def _bulk_node_validation(self, answers, current, regex = "^[0-9a-zA-Z_.,$#-]+$"): #Validation of nodes when running bulk command if not re.match(regex, current): raise inquirer.errors.ValidationError("", reason="Host cannot be empty") if current.startswith("@"): if current[1:] not in self.profiles: raise inquirer.errors.ValidationError("", reason="Profile {} don't exist".format(current)) return True def _bulk_folder_validation(self, answers, current): #Validation of folders when running bulk command if not self.case: current = current.lower() matches = list(filter(lambda k: k == current, self.folders)) if current != "" and len(matches) == 0: raise inquirer.errors.ValidationError("", reason="Location {} don't exist".format(current)) return True def _bulk_host_validation(self, answers, current, regex = "^.+$"): #Validate hostname when running bulk command if not re.match(regex, current): raise inquirer.errors.ValidationError("", reason="Host cannot be empty") if current.startswith("@"): if current[1:] not in self.profiles: raise inquirer.errors.ValidationError("", reason="Profile {} don't exist".format(current)) hosts = current.split(",") nodes = answers["ids"].split(",") if len(hosts) > 1 and len(hosts) != len(nodes): raise inquirer.errors.ValidationError("", reason="Hosts list should be the same length of nodes list") return True def _questions_edit(self): #Inquirer questions when editing nodes or profiles questions = [] questions.append(inquirer.Confirm("host", message="Edit Hostname/IP?")) questions.append(inquirer.Confirm("protocol", message="Edit Protocol?")) questions.append(inquirer.Confirm("port", message="Edit Port?")) questions.append(inquirer.Confirm("options", message="Edit Options?")) questions.append(inquirer.Confirm("logs", message="Edit logging path/file?")) questions.append(inquirer.Confirm("tags", message="Edit tags?")) questions.append(inquirer.Confirm("user", message="Edit User?")) questions.append(inquirer.Confirm("password", message="Edit password?")) answers = inquirer.prompt(questions) return answers def _questions_nodes(self, unique, uniques = None, edit = None): #Questions when adding or editing nodes try: defaults = self.config.getitem(unique) if "tags" not in defaults: defaults["tags"] = "" except: defaults = { "host":"", "protocol":"", "port":"", "user":"", "options":"", "logs":"" , "tags":""} node = {} if edit == None: edit = { "host":True, "protocol":True, "port":True, "user":True, "password": True,"options":True, "logs":True, "tags":True } questions = [] if edit["host"]: questions.append(inquirer.Text("host", message="Add Hostname or IP", validate=self._host_validation, default=defaults["host"])) else: node["host"] = defaults["host"] if edit["protocol"]: questions.append(inquirer.Text("protocol", message="Select Protocol", validate=self._protocol_validation, default=defaults["protocol"])) else: node["protocol"] = defaults["protocol"] if edit["port"]: questions.append(inquirer.Text("port", message="Select Port Number", validate=self._port_validation, default=defaults["port"])) else: node["port"] = defaults["port"] if edit["options"]: questions.append(inquirer.Text("options", message="Pass extra options to protocol", validate=self._default_validation, default=defaults["options"])) else: node["options"] = defaults["options"] if edit["logs"]: questions.append(inquirer.Text("logs", message="Pick logging path/file ", validate=self._default_validation, default=defaults["logs"].replace("{","{{").replace("}","}}"))) else: node["logs"] = defaults["logs"] if edit["tags"]: questions.append(inquirer.Text("tags", message="Add tags dictionary", validate=self._tags_validation, default=str(defaults["tags"]).replace("{","{{").replace("}","}}"))) else: node["tags"] = defaults["tags"] if edit["user"]: questions.append(inquirer.Text("user", message="Pick username", validate=self._default_validation, default=defaults["user"])) else: node["user"] = defaults["user"] if edit["password"]: questions.append(inquirer.List("password", message="Password: Use a local password, no password or a list of profiles to reference?", choices=["Local Password", "Profiles", "No Password"])) else: node["password"] = defaults["password"] answer = inquirer.prompt(questions) if answer == None: return False if "password" in answer.keys(): if answer["password"] == "Local Password": passq = [inquirer.Password("password", message="Set Password")] passa = inquirer.prompt(passq) if passa == None: return False answer["password"] = self.encrypt(passa["password"]) elif answer["password"] == "Profiles": passq = [(inquirer.Text("password", message="Set a @profile or a comma separated list of @profiles", validate=self._pass_validation))] passa = inquirer.prompt(passq) if passa == None: return False answer["password"] = passa["password"].split(",") elif answer["password"] == "No Password": answer["password"] = "" if "tags" in answer.keys() and not answer["tags"].startswith("@") and answer["tags"]: answer["tags"] = ast.literal_eval(answer["tags"]) result = {**uniques, **answer, **node} result["type"] = "connection" return result def _questions_profiles(self, unique, edit = None): #Questions when adding or editing profiles try: defaults = self.config.profiles[unique] if "tags" not in defaults: defaults["tags"] = "" except: defaults = { "host":"", "protocol":"", "port":"", "user":"", "options":"", "logs":"", "tags": "" } profile = {} if edit == None: edit = { "host":True, "protocol":True, "port":True, "user":True, "password": True,"options":True, "logs":True, "tags":True } questions = [] if edit["host"]: questions.append(inquirer.Text("host", message="Add Hostname or IP", default=defaults["host"])) else: profile["host"] = defaults["host"] if edit["protocol"]: questions.append(inquirer.Text("protocol", message="Select Protocol", validate=self._profile_protocol_validation, default=defaults["protocol"])) else: profile["protocol"] = defaults["protocol"] if edit["port"]: questions.append(inquirer.Text("port", message="Select Port Number", validate=self._profile_port_validation, default=defaults["port"])) else: profile["port"] = defaults["port"] if edit["options"]: questions.append(inquirer.Text("options", message="Pass extra options to protocol", default=defaults["options"])) else: profile["options"] = defaults["options"] if edit["logs"]: questions.append(inquirer.Text("logs", message="Pick logging path/file ", default=defaults["logs"].replace("{","{{").replace("}","}}"))) else: profile["logs"] = defaults["logs"] if edit["tags"]: questions.append(inquirer.Text("tags", message="Add tags dictionary", validate=self._profile_tags_validation, default=str(defaults["tags"]).replace("{","{{").replace("}","}}"))) else: profile["tags"] = defaults["tags"] if edit["user"]: questions.append(inquirer.Text("user", message="Pick username", default=defaults["user"])) else: profile["user"] = defaults["user"] if edit["password"]: questions.append(inquirer.Password("password", message="Set Password")) else: profile["password"] = defaults["password"] answer = inquirer.prompt(questions) if answer == None: return False if "password" in answer.keys(): if answer["password"] != "": answer["password"] = self.encrypt(answer["password"]) if "tags" in answer.keys() and answer["tags"]: answer["tags"] = ast.literal_eval(answer["tags"]) result = {**answer, **profile} result["id"] = unique return result def _questions_bulk(self): #Questions when using bulk command questions = [] questions.append(inquirer.Text("ids", message="add a comma separated list of nodes to add", validate=self._bulk_node_validation)) questions.append(inquirer.Text("location", message="Add a @folder, @subfolder@folder or leave empty", validate=self._bulk_folder_validation)) questions.append(inquirer.Text("host", message="Add comma separated list of Hostnames or IPs", validate=self._bulk_host_validation)) questions.append(inquirer.Text("protocol", message="Select Protocol", validate=self._protocol_validation)) questions.append(inquirer.Text("port", message="Select Port Number", validate=self._port_validation)) questions.append(inquirer.Text("options", message="Pass extra options to protocol", validate=self._default_validation)) questions.append(inquirer.Text("logs", message="Pick logging path/file ", validate=self._default_validation)) questions.append(inquirer.Text("tags", message="Add tags dictionary", validate=self._tags_validation)) questions.append(inquirer.Text("user", message="Pick username", validate=self._default_validation)) questions.append(inquirer.List("password", message="Password: Use a local password, no password or a list of profiles to reference?", choices=["Local Password", "Profiles", "No Password"])) answer = inquirer.prompt(questions) if answer == None: return False if "password" in answer.keys(): if answer["password"] == "Local Password": passq = [inquirer.Password("password", message="Set Password")] passa = inquirer.prompt(passq) answer["password"] = self.encrypt(passa["password"]) elif answer["password"] == "Profiles": passq = [(inquirer.Text("password", message="Set a @profile or a comma separated list of @profiles", validate=self._pass_validation))] passa = inquirer.prompt(passq) answer["password"] = passa["password"].split(",") elif answer["password"] == "No Password": answer["password"] = "" answer["type"] = "connection" if "tags" in answer.keys() and not answer["tags"].startswith("@") and answer["tags"]: answer["tags"] = ast.literal_eval(answer["tags"]) return answer def _type_node(self, arg_value, pat=re.compile(r"^[0-9a-zA-Z_.$@#-]+$")): if not pat.match(arg_value): raise argparse.ArgumentTypeError return arg_value def _type_profile(self, arg_value, pat=re.compile(r"^[0-9a-zA-Z_.$#-]+$")): if not pat.match(arg_value): raise argparse.ArgumentTypeError return arg_value def _help(self, type): #Store text for help and other commands if type == "node": return "node[@subfolder][@folder]\nConnect to specific node or show all matching nodes\n[@subfolder][@folder]\nShow all available connections globaly or in specified path" if type == "usage": return "conn [-h] [--add | --del | --mod | --show | --debug] [node|folder]\n conn {profile,move,mv,copy,cp,list,ls,bulk,config} ..." if type == "end": return "Commands:\n profile Manage profiles\n move (mv) Move node\n copy (cp) Copy node\n list (ls) List profiles, nodes or folders\n bulk Add nodes in bulk\n run Run scripts or commands on nodes\n config Manage app config\n api Start and stop connpy api" if type == "bashcompletion": return ''' #Here starts bash completion for conn _conn() { strings="$(connpy-completion-helper ${#COMP_WORDS[@]} ${COMP_WORDS[@]})" COMPREPLY=($(compgen -W "$strings" -- "${COMP_WORDS[-1]}")) } complete -o nosort -F _conn conn complete -o nosort -F _conn connpy #Here ends bash completion for conn ''' if type == "zshcompletion": return ''' #Here starts zsh completion for conn autoload -U compinit && compinit _conn() { strings=($(connpy-completion-helper ${#words} $words)) compadd "$@" -- `echo $strings` } compdef _conn conn compdef _conn connpy #Here ends zsh completion for conn ''' if type == "run": return "node[@subfolder][@folder] commmand to run\nRun the specific command on the node and print output\n/path/to/file.yaml\nUse a yaml file to run an automation script" if type == "generate": return '''--- tasks: - name: "Config" action: 'run' #Action can be test or run. Mandatory nodes: #List of nodes to work on. Mandatory - 'router1@office' #You can add specific nodes - '@aws' #entire folders or subfolders - '@office': #or filter inside a folder or subfolder - 'router2' - 'router7' commands: #List of commands to send, use {name} to pass variables - 'term len 0' - 'conf t' - 'interface {if}' - 'ip address 10.100.100.{id} 255.255.255.255' - '{commit}' - 'end' variables: #Variables to use on commands and expected. Optional __global__: #Global variables to use on all nodes, fallback if missing in the node. commit: '' if: 'loopback100' router1@office: id: 1 router2@office: id: 2 commit: 'commit' router3@office: id: 3 vrouter1@aws: id: 4 vrouterN@aws: id: 5 output: /home/user/logs #Type of output, if null you only get Connection and test result. Choices are: null,stdout,/path/to/folder. Folder path only works on 'run' action. options: prompt: r'>$|#$|\$$|>.$|#.$|\$.$' #Optional prompt to check on your devices, default should work on most devices. parallel: 10 #Optional number of nodes to run commands on parallel. Default 10. timeout: 20 #Optional time to wait in seconds for prompt, expected or EOF. Default 20. - name: "TestConfig" action: 'test' nodes: - 'router1@office' - '@aws' - '@office': - 'router2' - 'router7' commands: - 'ping 10.100.100.{id}' expected: '!' #Expected text to find when running test action. Mandatory for 'test' variables: router1@office: id: 1 router2@office: id: 2 commit: 'commit' router3@office: id: 3 vrouter1@aws: id: 4 vrouterN@aws: id: 5 output: null ...''' def encrypt(self, password, keyfile=None): ''' Encrypts password using RSA keyfile ### Parameters: - password (str): Plaintext password to encrypt. ### Optional Parameters: - keyfile (str): Path/file to keyfile. Default is config keyfile. ### Returns: str: Encrypted password. ''' if keyfile is None: keyfile = self.config.key with open(keyfile) as f: key = RSA.import_key(f.read()) f.close() publickey = key.publickey() encryptor = PKCS1_OAEP.new(publickey) password = encryptor.encrypt(password.encode("utf-8")) return str(password)Methods
def encrypt(self, password, keyfile=None)-
Encrypts password using RSA keyfile
Parameters:
- password (str): Plaintext password to encrypt.Optional Parameters:
- keyfile (str): Path/file to keyfile. Default is config keyfile.Returns:
str: Encrypted password.Expand source code
def encrypt(self, password, keyfile=None): ''' Encrypts password using RSA keyfile ### Parameters: - password (str): Plaintext password to encrypt. ### Optional Parameters: - keyfile (str): Path/file to keyfile. Default is config keyfile. ### Returns: str: Encrypted password. ''' if keyfile is None: keyfile = self.config.key with open(keyfile) as f: key = RSA.import_key(f.read()) f.close() publickey = key.publickey() encryptor = PKCS1_OAEP.new(publickey) password = encryptor.encrypt(password.encode("utf-8")) return str(password) def start(self, argv=['connpy', '--html', '-o', 'docs/', '--force'])-
Parameters:
- argv (list): List of arguments to pass to the app. Default: sys.argv[1:]Expand source code
def start(self,argv = sys.argv[1:]): ''' ### Parameters: - argv (list): List of arguments to pass to the app. Default: sys.argv[1:] ''' #DEFAULTPARSER defaultparser = argparse.ArgumentParser(prog = "conn", description = "SSH and Telnet connection manager", formatter_class=argparse.RawTextHelpFormatter) subparsers = defaultparser.add_subparsers(title="Commands") #NODEPARSER nodeparser = subparsers.add_parser("node",usage=self._help("usage"), help=self._help("node"),epilog=self._help("end"), formatter_class=argparse.RawTextHelpFormatter) nodecrud = nodeparser.add_mutually_exclusive_group() nodeparser.add_argument("node", metavar="node|folder", nargs='?', default=None, action=self._store_type, type=self._type_node, help=self._help("node")) nodecrud.add_argument("-v","--version", dest="action", action="store_const", help="Show version", const="version", default="connect") nodecrud.add_argument("-a","--add", dest="action", action="store_const", help="Add new node[@subfolder][@folder] or [@subfolder]@folder", const="add", default="connect") nodecrud.add_argument("-r","--del", "--rm", dest="action", action="store_const", help="Delete node[@subfolder][@folder] or [@subfolder]@folder", const="del", default="connect") nodecrud.add_argument("-e","--mod", "--edit", dest="action", action="store_const", help="Modify node[@subfolder][@folder]", const="mod", default="connect") nodecrud.add_argument("-s","--show", dest="action", action="store_const", help="Show node[@subfolder][@folder]", const="show", default="connect") nodecrud.add_argument("-d","--debug", dest="action", action="store_const", help="Display all conections steps", const="debug", default="connect") nodeparser.set_defaults(func=self._func_node) #PROFILEPARSER profileparser = subparsers.add_parser("profile", help="Manage profiles") profileparser.add_argument("profile", nargs=1, action=self._store_type, type=self._type_profile, help="Name of profile to manage") profilecrud = profileparser.add_mutually_exclusive_group(required=True) profilecrud.add_argument("-a", "--add", dest="action", action="store_const", help="Add new profile", const="add") profilecrud.add_argument("-r", "--del", "--rm", dest="action", action="store_const", help="Delete profile", const="del") profilecrud.add_argument("-e", "--mod", "--edit", dest="action", action="store_const", help="Modify profile", const="mod") profilecrud.add_argument("-s", "--show", dest="action", action="store_const", help="Show profile", const="show") profileparser.set_defaults(func=self._func_profile) #MOVEPARSER moveparser = subparsers.add_parser("move", aliases=["mv"], help="Move node") moveparser.add_argument("move", nargs=2, action=self._store_type, help="Move node[@subfolder][@folder] dest_node[@subfolder][@folder]", default="move", type=self._type_node) moveparser.set_defaults(func=self._func_others) #COPYPARSER copyparser = subparsers.add_parser("copy", aliases=["cp"], help="Copy node") copyparser.add_argument("cp", nargs=2, action=self._store_type, help="Copy node[@subfolder][@folder] new_node[@subfolder][@folder]", default="cp", type=self._type_node) copyparser.set_defaults(func=self._func_others) #LISTPARSER lsparser = subparsers.add_parser("list", aliases=["ls"], help="List profiles, nodes or folders") lsparser.add_argument("ls", action=self._store_type, choices=["profiles","nodes","folders"], help="List profiles, nodes or folders", default=False) lsparser.set_defaults(func=self._func_others) #BULKPARSER bulkparser = subparsers.add_parser("bulk", help="Add nodes in bulk") bulkparser.add_argument("bulk", const="bulk", nargs=0, action=self._store_type, help="Add nodes in bulk") bulkparser.set_defaults(func=self._func_others) #RUNPARSER runparser = subparsers.add_parser("run", help="Run scripts or commands on nodes", formatter_class=argparse.RawTextHelpFormatter) runparser.add_argument("run", nargs='+', action=self._store_type, help=self._help("run"), default="run") runparser.add_argument("-g","--generate", dest="action", action="store_const", help="Generate yaml file template", const="generate", default="run") runparser.set_defaults(func=self._func_run) #APIPARSER apiparser = subparsers.add_parser("api", help="Start and stop connpy api") apicrud = apiparser.add_mutually_exclusive_group(required=True) apicrud.add_argument("-s","--start", dest="start", nargs="?", action=self._store_type, help="Start conppy api", type=int, default=8048, metavar="PORT") apicrud.add_argument("-r","--restart", dest="restart", nargs=0, action=self._store_type, help="Restart conppy api") apicrud.add_argument("-x","--stop", dest="stop", nargs=0, action=self._store_type, help="Stop conppy api") apicrud.add_argument("-d", "--debug", dest="debug", nargs="?", action=self._store_type, help="Run connpy server on debug mode", type=int, default=8048, metavar="PORT") apiparser.set_defaults(func=self._func_api) #CONFIGPARSER configparser = subparsers.add_parser("config", help="Manage app config") configcrud = configparser.add_mutually_exclusive_group(required=True) configcrud.add_argument("--allow-uppercase", dest="case", nargs=1, action=self._store_type, help="Allow case sensitive names", choices=["true","false"]) configcrud.add_argument("--fzf", dest="fzf", nargs=1, action=self._store_type, help="Use fzf for lists", choices=["true","false"]) configcrud.add_argument("--keepalive", dest="idletime", nargs=1, action=self._store_type, help="Set keepalive time in seconds, 0 to disable", type=int, metavar="INT") configcrud.add_argument("--completion", dest="completion", nargs=1, choices=["bash","zsh"], action=self._store_type, help="Get terminal completion configuration for conn") configcrud.add_argument("--configfolder", dest="configfolder", nargs=1, action=self._store_type, help="Set the default location for config file", metavar="FOLDER") configcrud.add_argument("--openai", dest="openai", nargs=2, action=self._store_type, help="Set openai organization and api_key", metavar=("ORGANIZATION", "API_KEY")) configparser.set_defaults(func=self._func_others) #Manage sys arguments commands = ["node", "profile", "mv", "move","copy", "cp", "bulk", "ls", "list", "run", "config", "api"] profilecmds = ["--add", "-a", "--del", "--rm", "-r", "--mod", "--edit", "-e", "--show", "-s"] if len(argv) >= 2 and argv[1] == "profile" and argv[0] in profilecmds: argv[1] = argv[0] argv[0] = "profile" if len(argv) < 1 or argv[0] not in commands: argv.insert(0,"node") args = defaultparser.parse_args(argv) return args.func(args)
class node (unique, host, options='', logs='', password='', port='', protocol='', user='', config='', tags='')-
This class generates a node object. Containts all the information and methods to connect and interact with a device using ssh or telnet.
Attributes:
- output (str): Output of the commands you ran with run or test method. - result(bool): True if expected value is found after running the commands using test method. - status (int): 0 if the method run or test run succesfully. 1 if connection failed. 2 if expect timeouts without prompt or EOF.Parameters:
- unique (str): Unique name to assign to the node. - host (str): IP address or hostname of the node.Optional Parameters:
- options (str): Additional options to pass the ssh/telnet for connection. - logs (str): Path/file for storing the logs. You can use ${unique},${host}, ${port}, ${user}, ${protocol} as variables. - password (str): Encrypted or plaintext password. - port (str): Port to connect to node, default 22 for ssh and 23 for telnet. - protocol (str): Select ssh or telnet. Default is ssh. - user (str): Username to of the node. - config (obj): Pass the object created with class configfile with key for decryption and extra configuration if you are using connection manager. - tags (dict) : Tags useful for automation and personal porpuse like "os", "prompt" and "screenleght_command"Expand source code
class node: ''' This class generates a node object. Containts all the information and methods to connect and interact with a device using ssh or telnet. ### Attributes: - output (str): Output of the commands you ran with run or test method. - result(bool): True if expected value is found after running the commands using test method. - status (int): 0 if the method run or test run succesfully. 1 if connection failed. 2 if expect timeouts without prompt or EOF. ''' def __init__(self, unique, host, options='', logs='', password='', port='', protocol='', user='', config='', tags=''): ''' ### Parameters: - unique (str): Unique name to assign to the node. - host (str): IP address or hostname of the node. ### Optional Parameters: - options (str): Additional options to pass the ssh/telnet for connection. - logs (str): Path/file for storing the logs. You can use ${unique},${host}, ${port}, ${user}, ${protocol} as variables. - password (str): Encrypted or plaintext password. - port (str): Port to connect to node, default 22 for ssh and 23 for telnet. - protocol (str): Select ssh or telnet. Default is ssh. - user (str): Username to of the node. - config (obj): Pass the object created with class configfile with key for decryption and extra configuration if you are using connection manager. - tags (dict) : Tags useful for automation and personal porpuse like "os", "prompt" and "screenleght_command" ''' if config == '': self.idletime = 0 self.key = None else: self.idletime = config.config["idletime"] self.key = config.key self.unique = unique attr = {"host": host, "logs": logs, "options":options, "port": port, "protocol": protocol, "user": user, "tags": tags} for key in attr: profile = re.search("^@(.*)", str(attr[key])) if profile and config != '': try: setattr(self,key,config.profiles[profile.group(1)][key]) except: setattr(self,key,"") elif attr[key] == '' and key == "protocol": try: setattr(self,key,config.profiles["default"][key]) except: setattr(self,key,"ssh") else: setattr(self,key,attr[key]) if isinstance(password,list): self.password = [] for i, s in enumerate(password): profile = re.search("^@(.*)", password[i]) if profile and config != '': self.password.append(config.profiles[profile.group(1)]["password"]) else: self.password = [password] def _passtx(self, passwords, *, keyfile=None): # decrypts passwords, used by other methdos. dpass = [] if keyfile is None: keyfile = self.key if keyfile is not None: with open(keyfile) as f: key = RSA.import_key(f.read()) decryptor = PKCS1_OAEP.new(key) for passwd in passwords: if not re.match('^b[\"\'].+[\"\']$', passwd): dpass.append(passwd) else: try: decrypted = decryptor.decrypt(ast.literal_eval(passwd)).decode("utf-8") dpass.append(decrypted) except: raise ValueError("Missing or corrupted key") return dpass def _logfile(self, logfile = None): # translate logs variables and generate logs path. if logfile == None: logfile = self.logs logfile = logfile.replace("${unique}", self.unique) logfile = logfile.replace("${host}", self.host) logfile = logfile.replace("${port}", self.port) logfile = logfile.replace("${user}", self.user) logfile = logfile.replace("${protocol}", self.protocol) now = datetime.datetime.now() dateconf = re.search(r'\$\{date \'(.*)\'}', logfile) if dateconf: logfile = re.sub(r'\$\{date (.*)}',now.strftime(dateconf.group(1)), logfile) return logfile def _logclean(self, logfile, var = False): #Remove special ascii characters and other stuff from logfile. if var == False: t = open(logfile, "r").read() else: t = logfile t = t.replace("\n","",1) t = t.replace("\a","") t = t.replace('\n\n', '\n') t = re.sub(r'.\[K', '', t) while t.find("\b") != -1: t = re.sub('[^\b]\b', '', t) ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/ ]*[@-~])') t = ansi_escape.sub('', t) t = t.lstrip(" \n\r") t = t.replace("\r","") t = t.replace("\x0E","") t = t.replace("\x0F","") if var == False: d = open(logfile, "w") d.write(t) d.close() return else: return t def _filter(self, a): #Set time for last input when using interact self.lastinput = time() return a def _keepalive(self): #Send keepalive ctrl+e when idletime passed without new inputs on interact self.lastinput = time() t = threading.current_thread() while True: if time() - self.lastinput >= self.idletime: self.child.sendcontrol("e") self.lastinput = time() sleep(1) def interact(self, debug = False): ''' Allow user to interact with the node directly, mostly used by connection manager. ### Optional Parameters: - debug (bool): If True, display all the connecting information before interact. Default False. ''' connect = self._connect(debug = debug) if connect == True: size = re.search('columns=([0-9]+).*lines=([0-9]+)',str(os.get_terminal_size())) self.child.setwinsize(int(size.group(2)),int(size.group(1))) print("Connected to " + self.unique + " at " + self.host + (":" if self.port != '' else '') + self.port + " via: " + self.protocol) if 'logfile' in dir(self): if not 'mylog' in dir(self): self.mylog = io.BytesIO() self.child.logfile_read = self.mylog if 'missingtext' in dir(self): print(self.child.after.decode(), end='') if self.idletime > 0: x = threading.Thread(target=self._keepalive) x.daemon = True x.start() if debug: print(self.mylog.getvalue().decode()) self.child.interact(input_filter=self._filter) if "logfile" in dir(self): output = self._logclean(self.mylog.getvalue().decode(), True) with open(self.logfile, "w") as f: f.write(output) f.close() else: print(connect) exit(1) def run(self, commands, vars = None,*, folder = '', prompt = r'>$|#$|\$$|>.$|#.$|\$.$', stdout = False, timeout = 10): ''' Run a command or list of commands on the node and return the output. ### Parameters: - commands (str/list): Commands to run on the node. Should be str or a list of str. You can use variables as {varname} and defining them in optional parameter vars. ### Optional Parameters: - vars (dict): Dictionary containing the definition of variables used in commands parameter. Keys: Variable names. Values: strings. ### Optional Named Parameters: - folder (str): Path where output log should be stored, leave empty to disable logging. - prompt (str): Prompt to be expected after a command is finished running. Usually linux uses ">" or EOF while routers use ">" or "#". The default value should work for most nodes. Change it if your connection need some special symbol. - stdout (bool):Set True to send the command output to stdout. default False. - timeout (int):Time in seconds for expect to wait for prompt/EOF. default 10. ### Returns: str: Output of the commands you ran on the node. ''' connect = self._connect(timeout = timeout) now = datetime.datetime.now().strftime('%Y-%m-%d_%H%M%S') if connect == True: if "prompt" in self.tags: prompt = self.tags["prompt"] expects = [prompt, pexpect.EOF, pexpect.TIMEOUT] output = '' status = '' if not isinstance(commands, list): commands = [commands] if "screen_length_command" in self.tags: commands.insert(0, self.tags["screen_length_command"]) self.mylog = io.BytesIO() self.child.logfile_read = self.mylog for c in commands: if vars is not None: c = c.format(**vars) result = self.child.expect(expects, timeout = timeout) self.child.sendline(c) if result == 2: break if not result == 2: result = self.child.expect(expects, timeout = timeout) self.child.close() output = self._logclean(self.mylog.getvalue().decode(), True) if stdout == True: print(output) if folder != '': with open(folder + "/" + self.unique + "_" + now + ".txt", "w") as f: f.write(output) f.close() self.output = output if result == 2: self.status = 2 else: self.status = 0 return output else: self.output = connect self.status = 1 if stdout == True: print(connect) if folder != '': with open(folder + "/" + self.unique + "_" + now + ".txt", "w") as f: f.write(connect) f.close() return connect def test(self, commands, expected, vars = None,*, prompt = r'>$|#$|\$$|>.$|#.$|\$.$', timeout = 10): ''' Run a command or list of commands on the node, then check if expected value appears on the output after the last command. ### Parameters: - commands (str/list): Commands to run on the node. Should be str or a list of str. You can use variables as {varname} and defining them in optional parameter vars. - expected (str) : Expected text to appear after running all the commands on the node.You can use variables as {varname} and defining them in optional parameter vars. ### Optional Parameters: - vars (dict): Dictionary containing the definition of variables used in commands and expected parameters. Keys: Variable names. Values: strings. ### Optional Named Parameters: - prompt (str): Prompt to be expected after a command is finished running. Usually linux uses ">" or EOF while routers use ">" or "#". The default value should work for most nodes. Change it if your connection need some special symbol. - timeout (int):Time in seconds for expect to wait for prompt/EOF. default 10. ### Returns: bool: true if expected value is found after running the commands false if prompt is found before. ''' connect = self._connect(timeout = timeout) if connect == True: if "prompt" in self.tags: prompt = self.tags["prompt"] expects = [prompt, pexpect.EOF, pexpect.TIMEOUT] output = '' if not isinstance(commands, list): commands = [commands] if "screen_length_command" in self.tags: commands.insert(0, self.tags["screen_length_command"]) self.mylog = io.BytesIO() self.child.logfile_read = self.mylog for c in commands: if vars is not None: c = c.format(**vars) result = self.child.expect(expects, timeout = timeout) self.child.sendline(c) if result == 2: break if not result == 2: result = self.child.expect(expects, timeout = timeout) self.child.close() output = self._logclean(self.mylog.getvalue().decode(), True) self.output = output if result in [0, 1]: lastcommand = commands[-1] if vars is not None: expected = expected.format(**vars) lastcommand = lastcommand.format(**vars) last_command_index = output.rfind(lastcommand) cleaned_output = output[last_command_index + len(lastcommand):].strip() if expected in cleaned_output: self.result = True else: self.result = False self.status = 0 return False if result == 2: self.result = None self.status = 2 return output else: self.result = None self.output = connect self.status = 1 return connect def _connect(self, debug = False, timeout = 10, max_attempts = 3): # Method to connect to the node, it parse all the information, create the ssh/telnet command and login to the node. if self.protocol == "ssh": cmd = "ssh" if self.idletime > 0: cmd = cmd + " -o ServerAliveInterval=" + str(self.idletime) if self.user == '': cmd = cmd + " {}".format(self.host) else: cmd = cmd + " {}".format("@".join([self.user,self.host])) if self.port != '': cmd = cmd + " -p " + self.port if self.options != '': cmd = cmd + " " + self.options if self.logs != '': self.logfile = self._logfile() if self.password[0] != '': passwords = self._passtx(self.password) else: passwords = [] expects = ['yes/no', 'refused', 'supported', 'cipher', 'ssh-keygen.*\"', 'timeout', 'unavailable', 'closed', '[p|P]assword:|[u|U]sername:', r'>$|#$|\$$|>.$|#.$|\$.$', 'suspend', pexpect.EOF, pexpect.TIMEOUT, "No route to host", "resolve hostname", "no matching"] elif self.protocol == "telnet": cmd = "telnet " + self.host if self.port != '': cmd = cmd + " " + self.port if self.options != '': cmd = cmd + " " + self.options if self.logs != '': self.logfile = self._logfile() if self.password[0] != '': passwords = self._passtx(self.password) else: passwords = [] expects = ['[u|U]sername:', 'refused', 'supported', 'cipher', 'ssh-keygen.*\"', 'timeout', 'unavailable', 'closed', '[p|P]assword:', r'>$|#$|\$$|>.$|#.$|\$.$', 'suspend', pexpect.EOF, pexpect.TIMEOUT, "No route to host", "resolve hostname", "no matching"] else: raise ValueError("Invalid protocol: " + self.protocol) attempts = 1 while attempts <= max_attempts: child = pexpect.spawn(cmd) if debug: print(cmd) self.mylog = io.BytesIO() child.logfile_read = self.mylog if len(passwords) > 0: loops = len(passwords) else: loops = 1 endloop = False for i in range(0, loops): while True: results = child.expect(expects, timeout=timeout) if results == 0: if self.protocol == "ssh": child.sendline('yes') elif self.protocol == "telnet": if self.user != '': child.sendline(self.user) else: self.missingtext = True break if results == 4: child.terminate() return "Connection failed code:" + str(results) + "\n" + child.after.decode() if results in [1, 2, 3, 4, 5, 6, 7, 12, 13, 14, 15]: child.terminate() if results == 12 and attempts != max_attempts: attempts += 1 endloop = True break else: return "Connection failed code:" + str(results) if results == 8: if len(passwords) > 0: child.sendline(passwords[i]) else: self.missingtext = True break if results in [9, 11]: endloop = True child.sendline() break if results == 10: child.sendline("\r") sleep(2) if endloop: break if results == 12: continue else: break child.readline(0) self.child = child return TrueMethods
def interact(self, debug=False)-
Allow user to interact with the node directly, mostly used by connection manager.
Optional Parameters:
- debug (bool): If True, display all the connecting information before interact. Default False.Expand source code
def interact(self, debug = False): ''' Allow user to interact with the node directly, mostly used by connection manager. ### Optional Parameters: - debug (bool): If True, display all the connecting information before interact. Default False. ''' connect = self._connect(debug = debug) if connect == True: size = re.search('columns=([0-9]+).*lines=([0-9]+)',str(os.get_terminal_size())) self.child.setwinsize(int(size.group(2)),int(size.group(1))) print("Connected to " + self.unique + " at " + self.host + (":" if self.port != '' else '') + self.port + " via: " + self.protocol) if 'logfile' in dir(self): if not 'mylog' in dir(self): self.mylog = io.BytesIO() self.child.logfile_read = self.mylog if 'missingtext' in dir(self): print(self.child.after.decode(), end='') if self.idletime > 0: x = threading.Thread(target=self._keepalive) x.daemon = True x.start() if debug: print(self.mylog.getvalue().decode()) self.child.interact(input_filter=self._filter) if "logfile" in dir(self): output = self._logclean(self.mylog.getvalue().decode(), True) with open(self.logfile, "w") as f: f.write(output) f.close() else: print(connect) exit(1) def run(self, commands, vars=None, *, folder='', prompt='>$|#$|\\$$|>.$|#.$|\\$.$', stdout=False, timeout=10)-
Run a command or list of commands on the node and return the output.
Parameters:
- commands (str/list): Commands to run on the node. Should be str or a list of str. You can use variables as {varname} and defining them in optional parameter vars.Optional Parameters:
- vars (dict): Dictionary containing the definition of variables used in commands parameter. Keys: Variable names. Values: strings.Optional Named Parameters:
- folder (str): Path where output log should be stored, leave empty to disable logging. - prompt (str): Prompt to be expected after a command is finished running. Usually linux uses ">" or EOF while routers use ">" or "#". The default value should work for most nodes. Change it if your connection need some special symbol. - stdout (bool):Set True to send the command output to stdout. default False. - timeout (int):Time in seconds for expect to wait for prompt/EOF. default 10.Returns:
str: Output of the commands you ran on the node.Expand source code
def run(self, commands, vars = None,*, folder = '', prompt = r'>$|#$|\$$|>.$|#.$|\$.$', stdout = False, timeout = 10): ''' Run a command or list of commands on the node and return the output. ### Parameters: - commands (str/list): Commands to run on the node. Should be str or a list of str. You can use variables as {varname} and defining them in optional parameter vars. ### Optional Parameters: - vars (dict): Dictionary containing the definition of variables used in commands parameter. Keys: Variable names. Values: strings. ### Optional Named Parameters: - folder (str): Path where output log should be stored, leave empty to disable logging. - prompt (str): Prompt to be expected after a command is finished running. Usually linux uses ">" or EOF while routers use ">" or "#". The default value should work for most nodes. Change it if your connection need some special symbol. - stdout (bool):Set True to send the command output to stdout. default False. - timeout (int):Time in seconds for expect to wait for prompt/EOF. default 10. ### Returns: str: Output of the commands you ran on the node. ''' connect = self._connect(timeout = timeout) now = datetime.datetime.now().strftime('%Y-%m-%d_%H%M%S') if connect == True: if "prompt" in self.tags: prompt = self.tags["prompt"] expects = [prompt, pexpect.EOF, pexpect.TIMEOUT] output = '' status = '' if not isinstance(commands, list): commands = [commands] if "screen_length_command" in self.tags: commands.insert(0, self.tags["screen_length_command"]) self.mylog = io.BytesIO() self.child.logfile_read = self.mylog for c in commands: if vars is not None: c = c.format(**vars) result = self.child.expect(expects, timeout = timeout) self.child.sendline(c) if result == 2: break if not result == 2: result = self.child.expect(expects, timeout = timeout) self.child.close() output = self._logclean(self.mylog.getvalue().decode(), True) if stdout == True: print(output) if folder != '': with open(folder + "/" + self.unique + "_" + now + ".txt", "w") as f: f.write(output) f.close() self.output = output if result == 2: self.status = 2 else: self.status = 0 return output else: self.output = connect self.status = 1 if stdout == True: print(connect) if folder != '': with open(folder + "/" + self.unique + "_" + now + ".txt", "w") as f: f.write(connect) f.close() return connect def test(self, commands, expected, vars=None, *, prompt='>$|#$|\\$$|>.$|#.$|\\$.$', timeout=10)-
Run a command or list of commands on the node, then check if expected value appears on the output after the last command.
Parameters:
- commands (str/list): Commands to run on the node. Should be str or a list of str. You can use variables as {varname} and defining them in optional parameter vars. - expected (str) : Expected text to appear after running all the commands on the node.You can use variables as {varname} and defining them in optional parameter vars.Optional Parameters:
- vars (dict): Dictionary containing the definition of variables used in commands and expected parameters. Keys: Variable names. Values: strings.Optional Named Parameters:
- prompt (str): Prompt to be expected after a command is finished running. Usually linux uses ">" or EOF while routers use ">" or "#". The default value should work for most nodes. Change it if your connection need some special symbol. - timeout (int):Time in seconds for expect to wait for prompt/EOF. default 10.Returns:
bool: true if expected value is found after running the commands false if prompt is found before.Expand source code
def test(self, commands, expected, vars = None,*, prompt = r'>$|#$|\$$|>.$|#.$|\$.$', timeout = 10): ''' Run a command or list of commands on the node, then check if expected value appears on the output after the last command. ### Parameters: - commands (str/list): Commands to run on the node. Should be str or a list of str. You can use variables as {varname} and defining them in optional parameter vars. - expected (str) : Expected text to appear after running all the commands on the node.You can use variables as {varname} and defining them in optional parameter vars. ### Optional Parameters: - vars (dict): Dictionary containing the definition of variables used in commands and expected parameters. Keys: Variable names. Values: strings. ### Optional Named Parameters: - prompt (str): Prompt to be expected after a command is finished running. Usually linux uses ">" or EOF while routers use ">" or "#". The default value should work for most nodes. Change it if your connection need some special symbol. - timeout (int):Time in seconds for expect to wait for prompt/EOF. default 10. ### Returns: bool: true if expected value is found after running the commands false if prompt is found before. ''' connect = self._connect(timeout = timeout) if connect == True: if "prompt" in self.tags: prompt = self.tags["prompt"] expects = [prompt, pexpect.EOF, pexpect.TIMEOUT] output = '' if not isinstance(commands, list): commands = [commands] if "screen_length_command" in self.tags: commands.insert(0, self.tags["screen_length_command"]) self.mylog = io.BytesIO() self.child.logfile_read = self.mylog for c in commands: if vars is not None: c = c.format(**vars) result = self.child.expect(expects, timeout = timeout) self.child.sendline(c) if result == 2: break if not result == 2: result = self.child.expect(expects, timeout = timeout) self.child.close() output = self._logclean(self.mylog.getvalue().decode(), True) self.output = output if result in [0, 1]: lastcommand = commands[-1] if vars is not None: expected = expected.format(**vars) lastcommand = lastcommand.format(**vars) last_command_index = output.rfind(lastcommand) cleaned_output = output[last_command_index + len(lastcommand):].strip() if expected in cleaned_output: self.result = True else: self.result = False self.status = 0 return False if result == 2: self.result = None self.status = 2 return output else: self.result = None self.output = connect self.status = 1 return connect
class nodes (nodes: dict, config='')-
This class generates a nodes object. Contains a list of node class objects and methods to run multiple tasks on nodes simultaneously.
Attributes:
- nodelist (list): List of node class objects passed to the init function. - output (dict): Dictionary formed by nodes unique as keys, output of the commands you ran on the node as value. Created after running methods run or test. - result (dict): Dictionary formed by nodes unique as keys, value is True if expected value is found after running the commands, False if prompt is found before. Created after running method test. - status (dict): Dictionary formed by nodes unique as keys, value: 0 if method run or test ended succesfully. 1 if connection failed. 2 if expect timeouts without prompt or EOF. - <unique> (obj): For each item in nodelist, there is an attribute generated with the node unique.Parameters:
- nodes (dict): Dictionary formed by node information: Keys: Unique name for each node. Mandatory Subkeys: host(str). Optional Subkeys: options(str), logs(str), password(str), port(str), protocol(str), user(str). For reference on subkeys check node class.Optional Parameters:
- config (obj): Pass the object created with class configfile with key for decryption and extra configuration if you are using connection manager.Expand source code
class nodes: ''' This class generates a nodes object. Contains a list of node class objects and methods to run multiple tasks on nodes simultaneously. ### Attributes: - nodelist (list): List of node class objects passed to the init function. - output (dict): Dictionary formed by nodes unique as keys, output of the commands you ran on the node as value. Created after running methods run or test. - result (dict): Dictionary formed by nodes unique as keys, value is True if expected value is found after running the commands, False if prompt is found before. Created after running method test. - status (dict): Dictionary formed by nodes unique as keys, value: 0 if method run or test ended succesfully. 1 if connection failed. 2 if expect timeouts without prompt or EOF. - <unique> (obj): For each item in nodelist, there is an attribute generated with the node unique. ''' def __init__(self, nodes: dict, config = ''): ''' ### Parameters: - nodes (dict): Dictionary formed by node information: Keys: Unique name for each node. Mandatory Subkeys: host(str). Optional Subkeys: options(str), logs(str), password(str), port(str), protocol(str), user(str). For reference on subkeys check node class. ### Optional Parameters: - config (obj): Pass the object created with class configfile with key for decryption and extra configuration if you are using connection manager. ''' self.nodelist = [] self.config = config for n in nodes: this = node(n, **nodes[n], config = config) self.nodelist.append(this) setattr(self,n,this) def _splitlist(self, lst, n): #split a list in lists of n members. for i in range(0, len(lst), n): yield lst[i:i + n] def run(self, commands, vars = None,*, folder = None, prompt = None, stdout = None, parallel = 10, timeout = None): ''' Run a command or list of commands on all the nodes in nodelist. ### Parameters: - commands (str/list): Commands to run on the nodes. Should be str or list of str. You can use variables as {varname} and defining them in optional parameter vars. ### Optional Parameters: - vars (dict): Dictionary containing the definition of variables for each node, used in commands parameter. Keys should be formed by nodes unique names. Use special key name __global__ for global variables. Subkeys: Variable names. Values: strings. ### Optional Named Parameters: - folder (str): Path where output log should be stored, leave empty to disable logging. - prompt (str): Prompt to be expected after a command is finished running. Usually linux uses ">" or EOF while routers use ">" or "#". The default value should work for most nodes. Change it if your connection need some special symbol. - stdout (bool): Set True to send the command output to stdout. Default False. - parallel (int): Number of nodes to run the commands simultaneously. Default is 10, if there are more nodes that this value, nodes are groups in groups with max this number of members. - timeout (int): Time in seconds for expect to wait for prompt/EOF. default 10. ###Returns: dict: Dictionary formed by nodes unique as keys, Output of the commands you ran on the node as value. ''' args = {} nodesargs = {} args["commands"] = commands if folder != None: args["folder"] = folder Path(folder).mkdir(parents=True, exist_ok=True) if prompt != None: args["prompt"] = prompt if stdout != None: args["stdout"] = stdout if timeout != None: args["timeout"] = timeout output = {} status = {} tasks = [] for n in self.nodelist: nodesargs[n.unique] = deepcopy(args) if vars != None: nodesargs[n.unique]["vars"] = {} if "__global__" in vars.keys(): nodesargs[n.unique]["vars"].update(vars["__global__"]) if n.unique in vars.keys(): nodesargs[n.unique]["vars"].update(vars[n.unique]) tasks.append(threading.Thread(target=n.run, kwargs=nodesargs[n.unique])) taskslist = list(self._splitlist(tasks, parallel)) for t in taskslist: for i in t: i.start() for i in t: i.join() for i in self.nodelist: output[i.unique] = i.output status[i.unique] = i.status self.output = output self.status = status return output def test(self, commands, expected, vars = None,*, prompt = None, parallel = 10, timeout = None): ''' Run a command or list of commands on all the nodes in nodelist, then check if expected value appears on the output after the last command. ### Parameters: - commands (str/list): Commands to run on the node. Should be str or list of str. - expected (str) : Expected text to appear after running all the commands on the node. ### Optional Parameters: - vars (dict): Dictionary containing the definition of variables for each node, used in commands and expected parameters. Keys should be formed by nodes unique names. Use special key name __global__ for global variables. Subkeys: Variable names. Values: strings. ### Optional Named Parameters: - prompt (str): Prompt to be expected after a command is finished running. Usually linux uses ">" or EOF while routers use ">" or "#". The default value should work for most nodes. Change it if your connection need some special symbol. - parallel (int): Number of nodes to run the commands simultaneously. Default is 10, if there are more nodes that this value, nodes are groups in groups with max this number of members. - timeout (int): Time in seconds for expect to wait for prompt/EOF. default 10. ### Returns: dict: Dictionary formed by nodes unique as keys, value is True if expected value is found after running the commands, False if prompt is found before. ''' args = {} nodesargs = {} args["commands"] = commands args["expected"] = expected if prompt != None: args["prompt"] = prompt if timeout != None: args["timeout"] = timeout output = {} result = {} status = {} tasks = [] for n in self.nodelist: nodesargs[n.unique] = deepcopy(args) if vars != None: nodesargs[n.unique]["vars"] = {} if "__global__" in vars.keys(): nodesargs[n.unique]["vars"].update(vars["__global__"]) if n.unique in vars.keys(): nodesargs[n.unique]["vars"].update(vars[n.unique]) tasks.append(threading.Thread(target=n.test, kwargs=nodesargs[n.unique])) taskslist = list(self._splitlist(tasks, parallel)) for t in taskslist: for i in t: i.start() for i in t: i.join() for i in self.nodelist: result[i.unique] = i.result output[i.unique] = i.output status[i.unique] = i.status self.output = output self.result = result self.status = status return resultMethods
def run(self, commands, vars=None, *, folder=None, prompt=None, stdout=None, parallel=10, timeout=None)-
Run a command or list of commands on all the nodes in nodelist.
Parameters:
- commands (str/list): Commands to run on the nodes. Should be str or list of str. You can use variables as {varname} and defining them in optional parameter vars.Optional Parameters:
- vars (dict): Dictionary containing the definition of variables for each node, used in commands parameter. Keys should be formed by nodes unique names. Use special key name __global__ for global variables. Subkeys: Variable names. Values: strings.Optional Named Parameters:
- folder (str): Path where output log should be stored, leave empty to disable logging. - prompt (str): Prompt to be expected after a command is finished running. Usually linux uses ">" or EOF while routers use ">" or "#". The default value should work for most nodes. Change it if your connection need some special symbol. - stdout (bool): Set True to send the command output to stdout. Default False. - parallel (int): Number of nodes to run the commands simultaneously. Default is 10, if there are more nodes that this value, nodes are groups in groups with max this number of members. - timeout (int): Time in seconds for expect to wait for prompt/EOF. default 10.Returns:
dict: Dictionary formed by nodes unique as keys, Output of the commands you ran on the node as value.Expand source code
def run(self, commands, vars = None,*, folder = None, prompt = None, stdout = None, parallel = 10, timeout = None): ''' Run a command or list of commands on all the nodes in nodelist. ### Parameters: - commands (str/list): Commands to run on the nodes. Should be str or list of str. You can use variables as {varname} and defining them in optional parameter vars. ### Optional Parameters: - vars (dict): Dictionary containing the definition of variables for each node, used in commands parameter. Keys should be formed by nodes unique names. Use special key name __global__ for global variables. Subkeys: Variable names. Values: strings. ### Optional Named Parameters: - folder (str): Path where output log should be stored, leave empty to disable logging. - prompt (str): Prompt to be expected after a command is finished running. Usually linux uses ">" or EOF while routers use ">" or "#". The default value should work for most nodes. Change it if your connection need some special symbol. - stdout (bool): Set True to send the command output to stdout. Default False. - parallel (int): Number of nodes to run the commands simultaneously. Default is 10, if there are more nodes that this value, nodes are groups in groups with max this number of members. - timeout (int): Time in seconds for expect to wait for prompt/EOF. default 10. ###Returns: dict: Dictionary formed by nodes unique as keys, Output of the commands you ran on the node as value. ''' args = {} nodesargs = {} args["commands"] = commands if folder != None: args["folder"] = folder Path(folder).mkdir(parents=True, exist_ok=True) if prompt != None: args["prompt"] = prompt if stdout != None: args["stdout"] = stdout if timeout != None: args["timeout"] = timeout output = {} status = {} tasks = [] for n in self.nodelist: nodesargs[n.unique] = deepcopy(args) if vars != None: nodesargs[n.unique]["vars"] = {} if "__global__" in vars.keys(): nodesargs[n.unique]["vars"].update(vars["__global__"]) if n.unique in vars.keys(): nodesargs[n.unique]["vars"].update(vars[n.unique]) tasks.append(threading.Thread(target=n.run, kwargs=nodesargs[n.unique])) taskslist = list(self._splitlist(tasks, parallel)) for t in taskslist: for i in t: i.start() for i in t: i.join() for i in self.nodelist: output[i.unique] = i.output status[i.unique] = i.status self.output = output self.status = status return output def test(self, commands, expected, vars=None, *, prompt=None, parallel=10, timeout=None)-
Run a command or list of commands on all the nodes in nodelist, then check if expected value appears on the output after the last command.
Parameters:
- commands (str/list): Commands to run on the node. Should be str or list of str. - expected (str) : Expected text to appear after running all the commands on the node.Optional Parameters:
- vars (dict): Dictionary containing the definition of variables for each node, used in commands and expected parameters. Keys should be formed by nodes unique names. Use special key name __global__ for global variables. Subkeys: Variable names. Values: strings.Optional Named Parameters:
- prompt (str): Prompt to be expected after a command is finished running. Usually linux uses ">" or EOF while routers use ">" or "#". The default value should work for most nodes. Change it if your connection need some special symbol. - parallel (int): Number of nodes to run the commands simultaneously. Default is 10, if there are more nodes that this value, nodes are groups in groups with max this number of members. - timeout (int): Time in seconds for expect to wait for prompt/EOF. default 10.Returns:
dict: Dictionary formed by nodes unique as keys, value is True if expected value is found after running the commands, False if prompt is found before.Expand source code
def test(self, commands, expected, vars = None,*, prompt = None, parallel = 10, timeout = None): ''' Run a command or list of commands on all the nodes in nodelist, then check if expected value appears on the output after the last command. ### Parameters: - commands (str/list): Commands to run on the node. Should be str or list of str. - expected (str) : Expected text to appear after running all the commands on the node. ### Optional Parameters: - vars (dict): Dictionary containing the definition of variables for each node, used in commands and expected parameters. Keys should be formed by nodes unique names. Use special key name __global__ for global variables. Subkeys: Variable names. Values: strings. ### Optional Named Parameters: - prompt (str): Prompt to be expected after a command is finished running. Usually linux uses ">" or EOF while routers use ">" or "#". The default value should work for most nodes. Change it if your connection need some special symbol. - parallel (int): Number of nodes to run the commands simultaneously. Default is 10, if there are more nodes that this value, nodes are groups in groups with max this number of members. - timeout (int): Time in seconds for expect to wait for prompt/EOF. default 10. ### Returns: dict: Dictionary formed by nodes unique as keys, value is True if expected value is found after running the commands, False if prompt is found before. ''' args = {} nodesargs = {} args["commands"] = commands args["expected"] = expected if prompt != None: args["prompt"] = prompt if timeout != None: args["timeout"] = timeout output = {} result = {} status = {} tasks = [] for n in self.nodelist: nodesargs[n.unique] = deepcopy(args) if vars != None: nodesargs[n.unique]["vars"] = {} if "__global__" in vars.keys(): nodesargs[n.unique]["vars"].update(vars["__global__"]) if n.unique in vars.keys(): nodesargs[n.unique]["vars"].update(vars[n.unique]) tasks.append(threading.Thread(target=n.test, kwargs=nodesargs[n.unique])) taskslist = list(self._splitlist(tasks, parallel)) for t in taskslist: for i in t: i.start() for i in t: i.join() for i in self.nodelist: result[i.unique] = i.result output[i.unique] = i.output status[i.unique] = i.status self.output = output self.result = result self.status = status return result