fluzzi32/connpy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add new plugin feature beta1

Browse Source
This commit is contained in:
Federico Luzzi 2023-12-14 16:56:59 -03:00
parent 9975d60a91
commit 4f8497ff26
6 changed files with 422 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

198
README.md
View File

@ -17,6 +17,128 @@ docker compose -f path/to/folder/docker-compose.yml build
docker compose -f path/to/folder/docker-compose.yml run -it connpy-app
```
## Connection manager
### 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
- Create in bulk, copy, move, export and import nodes for easy management.
- Run automation scripts in network devices.
- use GPT AI to help you manage your devices.
- Add plugins with your own scripts.
- Much more!
### Usage:
```
usage: conn [-h] [--add | --del | --mod | --show | --debug] [node|folder] [--sftp]
conn {profile,move,mv,copy,cp,list,ls,bulk,export,import,ai,run,api,plugin,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
-t, --sftp Connects using sftp instead of ssh
```
### Commands:
```
profile Manage profiles
move(mv) Move node
copy(cp) Copy node
list(ls) List profiles, nodes or folders
bulk Add nodes in bulk
export Export connection folder to Yaml file
import Import connection folder to config from Yaml file
ai Make request to an AI
run Run scripts or commands on nodes
api Start and stop connpy api
plugin Manage plugins
config Manage app config
```
### 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
```
## Plugin Requirements for Connpy
### General Structure
- The plugin script must be a Python file.
- Only the following top-level elements are allowed in the plugin script:
- Class definitions
- Function definitions
- Import statements
- The `if __name__ == "__main__":` block for standalone execution
- Pass statements
### Specific Class Requirements
- The plugin script must define at least two specific classes:
1. **Class `Parser`**:
- Must contain only one method: `__init__`.
- The `__init__` method must initialize at least two attributes:
- `self.parser`: An instance of `argparse.ArgumentParser`.
- `self.description`: A string containing the description of the parser.
2. **Class `Entrypoint`**:
- Must have an `__init__` method that accepts exactly three parameters besides `self`:
- `args`: Arguments passed to the plugin.
- The parser instance (typically `self.parser` from the `Parser` class).
- The Connapp instance to interact with the Connpy app.
### Executable Block
- The plugin script can include an executable block:
- `if __name__ == "__main__":`
- This block allows the plugin to be run as a standalone script for testing or independent use.
### Script Verification
- The `verify_script` method in `plugins.py` is used to check the plugin script's compliance with these standards.
- Non-compliant scripts will be rejected to ensure consistency and proper functionality within the plugin system.
-
### Example Script
For a practical example of how to write a compatible plugin script, please refer to the following example:
[Example Plugin Script](https://github.com/fluzzi/awspy)
This script demonstrates the required structure and implementation details according to the plugin system's standards.
## Automation module usage
### Standalone module
```
@ -100,82 +222,6 @@ input = "go to router 1 and get me the full configuration"
result = myia.ask(input, dryrun = False)
print(result)
```
## Connection manager
### 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] [--sftp]
conn {profile,move,copy,list,bulk,export,import,run,config,api,ai} ...
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
-t, --sftp Connects using sftp instead of ssh
```
### Commands:
```
profile Manage profiles
move (mv) Move node
copy (cp) Copy node
list (ls) List profiles, nodes or folders
bulk Add nodes in bulk
export Export connection folder to Yaml file
import Import connection folder to config from Yaml file
run Run scripts or commands on nodes
config Manage app config
api Start and stop connpy api
ai Make request to an AI
```
### 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

64
connpy/__init__.py
View File

@ -17,7 +17,7 @@ Connpy is a connection manager that allows you to store nodes to connect them fa
### Usage
```
usage: conn [-h] [--add | --del | --mod | --show | --debug] [node|folder] [--sftp]
conn {profile,move,copy,list,bulk,export,import,run,config,api,ai} ...
conn {profile,move,mv,copy,cp,list,ls,bulk,export,import,ai,run,api,plugin,config} ...
positional arguments:
node|folder node[@subfolder][@folder]
@ -35,17 +35,18 @@ Options:
-t, --sftp Connects using sftp instead of ssh
Commands:
profile Manage profiles
move (mv) Move node
copy (cp) Copy node
list (ls) List profiles, nodes or folders
bulk Add nodes in bulk
export Export connection folder to Yaml file
import Import connection folder to config from Yaml file
run Run scripts or commands on nodes
config Manage app config
api Start and stop connpy api
ai Make request to an AI
profile Manage profiles
move(mv) Move node
copy(cp) Copy node
list(ls) List profiles, nodes or folders
bulk Add nodes in bulk
export Export connection folder to Yaml file
import Import connection folder to config from Yaml file
ai Make request to an AI
run Run scripts or commands on nodes
api Start and stop connpy api
plugin Manage plugins
config Manage app config
```
### Manage profiles
@ -75,6 +76,45 @@ options:
conn pc@office
conn server
```
### General Structure
- The plugin script must be a Python file.
- Only the following top-level elements are allowed in the plugin script:
- Class definitions
- Function definitions
- Import statements
- The `if __name__ == "__main__":` block for standalone execution
- Pass statements
### Specific Class Requirements
- The plugin script must define at least two specific classes:
1. **Class `Parser`**:
- Must contain only one method: `__init__`.
- The `__init__` method must initialize at least two attributes:
- `self.parser`: An instance of `argparse.ArgumentParser`.
- `self.description`: A string containing the description of the parser.
2. **Class `Entrypoint`**:
- Must have an `__init__` method that accepts exactly three parameters besides `self`:
- `args`: Arguments passed to the plugin.
- The parser instance (typically `self.parser` from the `Parser` class).
- The Connapp instance to interact with the Connpy app.
### Executable Block
- The plugin script can include an executable block:
- `if __name__ == "__main__":`
- This block allows the plugin to be run as a standalone script for testing or independent use.
### Script Verification
- The `verify_script` method in `plugins.py` is used to check the plugin script's compliance with these standards.
- Non-compliant scripts will be rejected to ensure consistency and proper functionality within the plugin system.
-
### Example Script
For a practical example of how to write a compatible plugin script, please refer to the following example:
[Example Plugin Script](https://github.com/fluzzi/awspy)
This script demonstrates the required structure and implementation details according to the plugin system's standards.
## http API
With the Connpy API you can run commands on devices using http requests

2
connpy/_version.py
View File

@ -1,2 +1,2 @@
__version__ = "3.7.1"
__version__ = "3.8.0b1"

2
connpy/configfile.py
View File

@ -49,7 +49,9 @@ class configfile:
'''
home = os.path.expanduser("~")
defaultdir = home + '/.config/conn'
self.defaultdir = defaultdir
Path(defaultdir).mkdir(parents=True, exist_ok=True)
Path(f"{defaultdir}/plugins").mkdir(parents=True, exist_ok=True)
pathfile = defaultdir + '/.folder'
try:
with open(pathfile, "r") as f:

183
connpy/connapp.py
View File

@ -12,7 +12,9 @@ from .core import node,nodes
from ._version import __version__
from .api import start_api,stop_api,debug_api
from .ai import ai
from .plugins import Plugins
import yaml
import shutil
class NoAliasDumper(yaml.SafeDumper):
def ignore_aliases(self, data):
return True
@ -23,8 +25,6 @@ try:
from pyfzf.pyfzf import FzfPrompt
except:
FzfPrompt = None
home = os.path.expanduser("~")
defaultdir = home + '/.config/conn'
@ -68,9 +68,9 @@ class connapp:
'''
#DEFAULTPARSER
defaultparser = argparse.ArgumentParser(prog = "conn", description = "SSH and Telnet connection manager", formatter_class=argparse.RawTextHelpFormatter)
subparsers = defaultparser.add_subparsers(title="Commands")
subparsers = defaultparser.add_subparsers(title="Commands", dest="subcommand")
#NODEPARSER
nodeparser = subparsers.add_parser("node",usage=self._help("usage"), help=self._help("node"),epilog=self._help("end"), formatter_class=argparse.RawTextHelpFormatter)
nodeparser = subparsers.add_parser("node", formatter_class=argparse.RawTextHelpFormatter)
nodecrud = nodeparser.add_mutually_exclusive_group()
nodeparser.add_argument("node", metavar="node|folder", nargs='?', default=None, action=self._store_type, help=self._help("node"))
nodecrud.add_argument("-v","--version", dest="action", action="store_const", help="Show version", const="version", default="connect")
@ -82,7 +82,7 @@ class connapp:
nodeparser.add_argument("-t","--sftp", dest="sftp", action="store_true", help="Connects using sftp instead of ssh")
nodeparser.set_defaults(func=self._func_node)
#PROFILEPARSER
profileparser = subparsers.add_parser("profile", help="Manage profiles")
profileparser = subparsers.add_parser("profile", description="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")
@ -91,53 +91,62 @@ class connapp:
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 = subparsers.add_parser("move", aliases=["mv"], description="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 = subparsers.add_parser("copy", aliases=["cp"], description="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 = subparsers.add_parser("list", aliases=["ls"], description="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.add_argument("--filter", nargs=1, help="Filter results")
lsparser.add_argument("--format", nargs=1, help="Format of the output of nodes using {name}, {NAME}, {location}, {LOCATION}, {host} and {HOST}")
lsparser.set_defaults(func=self._func_others)
#BULKPARSER
bulkparser = subparsers.add_parser("bulk", help="Add nodes in bulk")
bulkparser = subparsers.add_parser("bulk", description="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)
# EXPORTPARSER
exportparser = subparsers.add_parser("export", help="Export connection folder to Yaml file")
exportparser = subparsers.add_parser("export", description="Export connection folder to Yaml file")
exportparser.add_argument("export", nargs="+", action=self._store_type, help="Export /path/to/file.yml [@subfolder1][@folder1] [@subfolderN][@folderN]")
exportparser.set_defaults(func=self._func_export)
# IMPORTPARSER
importparser = subparsers.add_parser("import", help="Import connection folder to config from Yaml file")
importparser = subparsers.add_parser("import", description="Import connection folder to config from Yaml file")
importparser.add_argument("file", nargs=1, action=self._store_type, help="Import /path/to/file.yml")
importparser.set_defaults(func=self._func_import)
# AIPARSER
aiparser = subparsers.add_parser("ai", help="Make request to an AI")
aiparser = subparsers.add_parser("ai", description="Make request to an AI")
aiparser.add_argument("ask", nargs='*', help="Ask connpy AI something")
aiparser.add_argument("--model", nargs=1, help="Set the OPENAI model id")
aiparser.add_argument("--org", nargs=1, help="Set the OPENAI organization id")
aiparser.add_argument("--api_key", nargs=1, help="Set the OPENAI API key")
aiparser.set_defaults(func=self._func_ai)
#RUNPARSER
runparser = subparsers.add_parser("run", help="Run scripts or commands on nodes", formatter_class=argparse.RawTextHelpFormatter)
runparser = subparsers.add_parser("run", description="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")
apiparser = subparsers.add_parser("api", description="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)
#PLUGINSPARSER
pluginparser = subparsers.add_parser("plugin", description="Manage plugins")
plugincrud = pluginparser.add_mutually_exclusive_group(required=True)
plugincrud.add_argument("--add", metavar=("PLUGIN", "FILE"), nargs=2, help="Add new plugin")
plugincrud.add_argument("--del", dest="delete", metavar="PLUGIN", nargs=1, help="Delete plugin")
plugincrud.add_argument("--enable", metavar="PLUGIN", nargs=1, help="Enable plugin")
plugincrud.add_argument("--disable", metavar="PLUGIN", nargs=1, help="Disable plugin")
plugincrud.add_argument("--list", dest="list", action="store_true", help="Disable plugin")
pluginparser.set_defaults(func=self._func_plugin)
#CONFIGPARSER
configparser = subparsers.add_parser("config", help="Manage app config")
configparser = subparsers.add_parser("config", description="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"])
@ -148,16 +157,29 @@ class connapp:
configcrud.add_argument("--openai-api-key", dest="api_key", nargs=1, action=self._store_type, help="Set openai api_key", metavar="API_KEY")
configcrud.add_argument("--openai-model", dest="model", nargs=1, action=self._store_type, help="Set openai model", metavar="MODEL")
configparser.set_defaults(func=self._func_others)
#Add plugins
file_path = self.config.defaultdir + "/plugins"
self.plugins = Plugins()
self.plugins.import_plugins_to_argparse(file_path, subparsers)
#Generate helps
nodeparser.usage = self._help("usage", subparsers)
nodeparser.epilog = self._help("end", subparsers)
nodeparser.help = self._help("node")
#Manage sys arguments
commands = ["node", "profile", "mv", "move","copy", "cp", "bulk", "ls", "list", "run", "config", "api", "ai", "export", "import"]
profilecmds = ["--add", "-a", "--del", "--rm", "-r", "--mod", "--edit", "-e", "--show", "-s"]
self.commands = list(subparsers.choices.keys())
profilecmds = []
for action in profileparser._actions:
profilecmds.extend(action.option_strings)
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:
if len(argv) < 1 or argv[0] not in self.commands:
argv.insert(0,"node")
args = defaultparser.parse_args(argv)
return args.func(args)
if args.subcommand in self.plugins.plugins:
self.plugins.plugins[args.subcommand].Entrypoint(args, self.plugins.plugin_parsers[args.subcommand], self)
else:
return args.func(args)
class _store_type(argparse.Action):
#Custom store type for cli app.
@ -580,7 +602,7 @@ class connapp:
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"
pathfile = self.config.defaultdir + "/.folder"
folder = os.path.abspath(args.data[0]).rstrip('/')
with open(pathfile, "w") as f:
f.write(str(folder))
@ -600,9 +622,106 @@ class connapp:
self.config._saveconfig(self.config.file)
print("Config saved")
def _func_plugin(self, args):
if args.add:
if not os.path.exists(args.add[1]):
print("File {} dosn't exists.".format(args.add[1]))
exit(14)
if args.add[0].isalpha() and args.add[0].islower() and len(args.add[0]) <= 15:
disabled_dest_file = os.path.join(self.config.defaultdir + "/plugins", args.add[0] + ".py.bkp")
if args.add[0] in self.commands or os.path.exists(disabled_dest_file):
print("Plugin name can't be the same as other commands.")
exit(15)
else:
check_bad_script = self.plugins.verify_script(args.add[1])
if check_bad_script:
print(check_bad_script)
exit(16)
else:
try:
dest_file = os.path.join(self.config.defaultdir + "/plugins", args.add[0] + ".py")
shutil.copy2(args.add[1], dest_file)
print(f"Plugin {args.add[0]} added succesfully.")
except:
print("Failed importing plugin file.")
exit(17)
else:
print("Plugin name should be lowercase letters up to 15 characters.")
exit(15)
elif args.delete:
plugin_file = os.path.join(self.config.defaultdir + "/plugins", args.delete[0] + ".py")
disabled_plugin_file = os.path.join(self.config.defaultdir + "/plugins", args.delete[0] + ".py.bkp")
plugin_exist = os.path.exists(plugin_file)
disabled_plugin_exist = os.path.exists(disabled_plugin_file)
if not plugin_exist and not disabled_plugin_exist:
print("Plugin {} dosn't exist.".format(args.delete[0]))
exit(14)
question = [inquirer.Confirm("delete", message="Are you sure you want to delete {} plugin?".format(args.delete[0]))]
confirm = inquirer.prompt(question)
if confirm == None:
exit(7)
if confirm["delete"]:
try:
if plugin_exist:
os.remove(plugin_file)
elif disabled_plugin_exist:
os.remove(disabled_plugin_file)
print(f"plugin {args.delete[0]} deleted succesfully.")
except:
print("Failed deleting plugin file.")
exit(17)
elif args.disable:
plugin_file = os.path.join(self.config.defaultdir + "/plugins", args.disable[0] + ".py")
disabled_plugin_file = os.path.join(self.config.defaultdir + "/plugins", args.disable[0] + ".py.bkp")
if not os.path.exists(plugin_file) or os.path.exists(disabled_plugin_file):
print("Plugin {} dosn't exist or it's disabled.".format(args.disable[0]))
exit(14)
try:
os.rename(plugin_file, disabled_plugin_file)
print(f"plugin {args.disable[0]} disabled succesfully.")
except:
print("Failed disabling plugin file.")
exit(17)
elif args.enable:
plugin_file = os.path.join(self.config.defaultdir + "/plugins", args.enable[0] + ".py")
disabled_plugin_file = os.path.join(self.config.defaultdir + "/plugins", args.enable[0] + ".py.bkp")
if os.path.exists(plugin_file) or not os.path.exists(disabled_plugin_file):
print("Plugin {} dosn't exist or it's enabled.".format(args.enable[0]))
exit(14)
try:
os.rename(disabled_plugin_file, plugin_file)
print(f"plugin {args.enable[0]} enabled succesfully.")
except:
print("Failed enabling plugin file.")
exit(17)
elif args.list:
enabled_files = []
disabled_files = []
plugins = {}
# Iterate over all files in the specified folder
for file in os.listdir(self.config.defaultdir + "/plugins"):
# Check if the file is a Python file
if file.endswith('.py'):
enabled_files.append(os.path.splitext(file)[0])
# Check if the file is a Python backup file
elif file.endswith('.py.bkp'):
disabled_files.append(os.path.splitext(os.path.splitext(file)[0])[0])
if enabled_files:
plugins["Enabled"] = enabled_files
if disabled_files:
plugins["Disabled"] = disabled_files
if plugins:
print(yaml.dump(plugins, sort_keys=False))
else:
print("There are no plugins added.")
def _func_import(self, args):
if not os.path.exists(args.data[0]):
print("File {} dosn't exists".format(args.data[0]))
print("File {} dosn't exist".format(args.data[0]))
exit(14)
print("This could overwrite your current configuration!")
question = [inquirer.Confirm("import", message="Are you sure you want to import {} file?".format(args.data[0]))]
@ -1244,14 +1363,30 @@ class connapp:
raise ValueError
return arg_value
def _help(self, type):
def _help(self, type, parsers = None):
#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 globally or in specified path"
if type == "usage":
return "conn [-h] [--add | --del | --mod | --show | --debug] [node|folder] [--sftp]\n conn {profile,move,copy,list,bulk,export,import,run,config,api,ai} ..."
commands = []
for subcommand, subparser in parsers.choices.items():
if subparser.description != None:
commands.append(subcommand)
commands = ",".join(commands)
usage_help = f"conn [-h] [--add | --del | --mod | --show | --debug] [node|folder] [--sftp]\n conn {{{commands}}} ..."
return usage_help
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 export Export connection folder to Yaml file\n import Import connection folder to config from Yaml file\n run Run scripts or commands on nodes\n config Manage app config\n api Start and stop connpy api\n ai Make request to an AI"
help_dict = {}
for subcommand, subparser in parsers.choices.items():
if subparser.description == None and help_dict:
previous_key = next(reversed(help_dict.keys()))
help_dict[f"{previous_key}({subcommand})"] = help_dict.pop(previous_key)
else:
help_dict[subcommand] = subparser.description
subparser.description = None
commands_help = "Commands:\n"
commands_help += "\n".join([f" {cmd:<15} {help_text}" for cmd, help_text in help_dict.items() if help_text != None])
return commands_help
if type == "bashcompletion":
return '''
#Here starts bash completion for conn

86
connpy/plugins.py Executable file
View File

@ -0,0 +1,86 @@
#!/usr/bin/python3
import ast
import importlib.util
import sys
import argparse
import os
class Plugins:
def __init__(self):
self.plugins = {}
self.plugin_parsers = {}
def verify_script(self, file_path):
with open(file_path, 'r') as file:
source_code = file.read()
try:
tree = ast.parse(source_code)
except SyntaxError as e:
return f"Syntax error in file: {e}"
required_classes = {'Parser', 'Entrypoint'}
found_classes = set()
for node in tree.body:
# Allow only function definitions, class definitions, and pass statements at top-level
if isinstance(node, ast.If):
# Check for the 'if __name__ == "__main__":' block
if not (isinstance(node.test, ast.Compare) and
isinstance(node.test.left, ast.Name) and
node.test.left.id == '__name__' and
isinstance(node.test.comparators[0], ast.Str) and
node.test.comparators[0].s == '__main__'):
return "Only __name__ == __main__ If is allowed"
elif not isinstance(node, (ast.FunctionDef, ast.ClassDef, ast.Import, ast.ImportFrom, ast.Pass)):
return f"Plugin can only have pass, functions, classes and imports. {node} is not allowed" # Reject any other AST types
if isinstance(node, ast.ClassDef) and node.name in required_classes:
found_classes.add(node.name)
if node.name == 'Parser':
# Ensure Parser class has only the __init__ method and assigns self.parser
if not all(isinstance(method, ast.FunctionDef) and method.name == '__init__' for method in node.body):
return "Parser class should only have __init__ method"
# Check if 'self.parser' and 'self.description' are assigned in __init__ method
init_method = node.body[0]
assigned_attrs = [target.attr for expr in init_method.body if isinstance(expr, ast.Assign) for target in expr.targets if isinstance(target, ast.Attribute) and isinstance(target.value, ast.Name) and target.value.id == 'self']
if 'parser' not in assigned_attrs or 'description' not in assigned_attrs:
return "Parser class should set self.parser and self.description" # 'self.parser' or 'self.description' not assigned in __init__
elif node.name == 'Entrypoint':
init_method = next((item for item in node.body if isinstance(item, ast.FunctionDef) and item.name == '__init__'), None)
if not init_method or len(init_method.args.args) != 4: # self, args, parser, conapp
return "Entrypoint class should accept only arguments: args, parser and connapp" # 'Entrypoint' __init__ does not have correct signature
if required_classes == found_classes:
return False
else:
return "Classes Entrypoint and Parser are mandatory"
def import_from_path(self, path):
spec = importlib.util.spec_from_file_location("module.name", path)
module = importlib.util.module_from_spec(spec)
sys.modules["module.name"] = module
spec.loader.exec_module(module)
return module
def import_plugins_to_argparse(self, directory, subparsers):
for filename in os.listdir(directory):
commands = subparsers.choices.keys()
if filename.endswith(".py"):
root_filename = os.path.splitext(filename)[0]
if root_filename in commands:
continue
# Construct the full path
filepath = os.path.join(directory, filename)
check_file = self.verify_script(filepath)
if check_file:
continue
else:
self.plugins[root_filename] = self.import_from_path(filepath)
self.plugin_parsers[root_filename] = self.plugins[root_filename].Parser()
subparsers.add_parser(root_filename, parents=[self.plugin_parsers[root_filename].parser], add_help=False, description=self.plugin_parsers[root_filename].description)