Connpy (v6.1.0)

Connpy is a powerful Connection Manager and Network Automation Platform for Linux, Mac, and Docker. It provides a unified interface for SSH, SFTP, Telnet, kubectl, Docker pods, and AWS SSM.

The v6 release introduces a comprehensive AI Copilot and AI Playbook Engine, transforming your terminal into an interactive network assistant that understands your device outputs, configures parameters safely, and runs simulations.

1. 🤖 AI System

1a. Terminal Copilot (Ctrl+Space)

Invoke the context-aware AI Copilot directly inside any active terminal session by pressing Ctrl + Space. * Context Modes: Cycles through LINES (sends raw scroll buffer), SINGLE (captures exactly one command + output block), and RANGE (logical group of recent commands) using Ctrl+Up/Down. * Slash Commands (/): Control the AI persona and safety settings: * /architect / /engineer: Swaps the agent between high-level strategist and technical executor. * /trust / /untrust: Configures auto-run behavior for suggested non-destructive commands. * /os [system]: Manually overrides target OS parsing rules (e.g. /os cisco_ios). * /prompt [regex]: Overrides command prompt detection bounds. * /clear: Clear context history.

1b. AI Chat (conn ai)

Start a standalone persistent session with the AI Copilot. Manage sessions using --list, --resume, --session <id> (to restore a specific history), --delete <id>, or send a quick single-shot question directly from the terminal prompt:

conn ai "how do i check bgp summary on cisco?"

1c. MCP Integration

Connect to external data sources and tools dynamically via the Model Context Protocol (MCP). Use the interactive wizard or command actions to configure MCP servers:

conn ai --mcp

1d. Local Interactive Shell (conn shell)

Launch a local interactive shell with AI Copilot support enabled directly on your host machine:

conn shell                     # Start local shell (default: $SHELL or /bin/bash)
conn shell -c /bin/zsh         # Override shell executable
conn shell --capture session.log # Log session output to file

• Nested Sessions & Passthrough: Supports running nested conn / connpy connections inside conn shell. Automatically detects foreground conn processes and forwards Ctrl+Space down to the active device connection instead of triggering the local Copilot.
• Shell Configuration: Configure default shell command, prompt regex, or OS type via conn config:

conn config --shell-command /bin/zsh
conn config --shell-prompt "\$\s*$"
conn config --shell-os ubuntu

2. ⚙️ Automation & Playbooks

2a. Quick Run (conn run)

Run commands in parallel directly on target nodes or folder structures:

conn run router1 "show interface"

2b. YAML Playbook Engine

Execute complex structured automation playbooks defined in YAML configuration files. Supports multi-task execution, variables (using global, per-node, or regex matching definitions), timeouts, and variable parallel execution bounds.

# example_playbook.yaml
- name: Verify Network Operations
  hosts: "@office"
  parallel: true
  tasks:
    - name: Get interface brief
      run: "show ip interface brief"
    - name: Check OSPF state
      run: "show ip ospf neighbor"
      test: "FULL"

Execute using the playbooks runner:

conn run example_playbook.yaml

2c. AI-Assisted Automation

Leverage AI to generate playbook templates (--generate-ai), simulate command changes before execution (--preflight-ai), or analyze consolidated execution logs post-run (--analyze). Use --test "expected text1" "expected text2" to specify assert-style output validations. * To generate an empty template: conn run --generate

3. 📂 Inventory Management

3a. Nodes

Manage connections using standard commands: add (conn --add node1), edit (conn --mod node1), delete (conn --del node1), show configuration (conn --show node1), or connect (conn node1).

3b. Profiles

Define credentials and templates globally and reference them inside node fields using the @profile_name placeholder. Manage profiles interactively or via commands:

conn profile -a profile_name
# Or equivalently:
conn -a profile profile_name

During the interactive conn --add prompt, you can input @profile_name in the username or password fields to reference it.

3c. Folders, Move, Copy, List

Organize nodes into logical folder hierarchies (@office, @datacenter@office). Move items (conn move [src] [dst]), copy (conn copy [src] [dst]), or list items with custom filters and formatting:

conn list nodes --filter ".*-prod" --format "{name} ({host}) runs {protocol}"

3d. Bulk, Export, Import

Bulk import connections from formatted text files (conn bulk -f nodes.txt), or export/import connection folders using YAML configurations (conn export @folder > backup.yaml / conn import backup.yaml).

3e. Tags System

Customize connection settings dynamically using tags. Configure per-node settings like custom OS types (os), prompt regex rules (prompt), and page length triggers (screen_length_command).

# Custom tags dictionary (YANG / VSR context)
tags: { "os": "cisco_ios", "prompt": ".*#", "screen_length_command": "terminal length 0" }

4. 🔌 Protocols & Connection Features

4a. SSH / SFTP / Telnet / kubectl / Docker / AWS SSM

Connect to various architectures using native protocols: * SSH / Telnet: Standard CLI protocols. * SFTP: Transfer files securely (conn --sftp node). * Docker: Connect directly to local container names (host set to container name/ID). * Kubernetes (kubectl): Connect to pods (namespace customizable via options). * AWS SSM: Connect to EC2 instances using Instance IDs as hosts.

4b. Jumphosts

Support for single or chained intermediate gateway nodes (SSH, SSM, kubectl, or docker jumphosts) to tunnel traffic safely into target environments.

4c. Debug Mode, Keepalive, Logging

Track connection steps (conn --debug node), set idle keepalive intervals (conn config --keepalive <seconds>), or define dynamic output log files using variables like ${unique}, ${host}, ${port}, ${user}, ${protocol}, or ${date 'format'}.

5. 🖥️ Remote Capture (conn capture - Core Plugin)

Perform remote packet capture (tcpdump) on hosts over secure SSH reverse tunnels and stream packets live into your local Wireshark GUI:

conn capture router1 eth0 -w -f "port 80"

• Requirements: Local installation of Wireshark or tshark is required for live piping (-w).
• Advanced flags: Specify network namespaces (--ns <name>), custom filters (-f <filter>), or configure the Wireshark local path (--set-wireshark-path).

6. 🛡️ Context Filtering

Prevent accidental command execution in production by setting active regex contexts. This hides non-matching inventory items and restricts execution scope:

conn context production -a --regex ".*-prod"
conn context production --set

• Manage Contexts: List defined filters (conn context --ls), show context details (conn context production -s), or delete contexts (conn context production -r).

7. 🔌 Plugin System

Extend connpy features and hook into core execution events (pre/post hooks) by writing Python scripts. Add, update, delete, or list plugins locally, or execute them on remote instances:

conn plugin --add my_plugin script.py
conn plugin --update my_plugin script.py
conn plugin --remote --sync

8. ⚙️ gRPC Client-Server Architecture

8a. Server (start/stop/restart/debug)

Execute tasks on a centralized remote host. Start gRPC server (conn api -s 50051), stop (conn api -x), restart (conn api -r), or debug in the foreground (conn api -d).

8b. Client Config

Shift the local CLI to communicate with a remote server instance:

conn config --service-mode remote
conn config --remote localhost:50051

8c. User Management & API Tokens

Manage server-side user credentials and Personal Access Tokens (PAT) for automated setups:

conn user --add username
conn user --list
conn user --regen-password username

# Personal Access Tokens (PAT) for non-interactive API access
conn user --create-token "CI/CD Token" --expires-in 30
conn user --list-tokens
conn user --revoke-token <token_id>

Use --path to specify custom configuration folders in server Mode B. Pass API tokens via CONNPY_TOKEN environment variable.

8d. SSO / OIDC

Configure identity providers (e.g. Authelia, Keycloak) for SSO gRPC authentication using the interactive wizard:

conn sso --add provider_name

8e. Login / Logout

Authenticate client sessions (conn login [username]), check connection status (conn login --status), or close sessions (conn logout).

9. ⚡ Installation & Configuration

9a. pip install

pip install connpy

9b. Shell Completion + FZF

Install autocompletions and fuzzy-search wrappers into your shell profile:

eval "$(conn config --completion bash)"
eval "$(conn config --fzf-wrapper bash)"

9c. conn config options

View configuration details (conn config) or customize variables like case sensitivity (--allow-uppercase), FZF list picker (--fzf true), configurations directory (--configfolder), or persistent AI API keys and models (--engineer-model).

9d. Theming

Customize CLI panel styles and colors by pointing to built-in presets or external YAML styles:

conn config --theme /path/to/theme.yaml

10. 🔒 Privacy, Security & Synchronization (conn sync)

Encrypts inventory and profiles locally via RSA/OAEP. Backup and sync configurations to Google Drive manually (conn sync --once, --list, --restore) or schedule auto-sync. Segregate restores (--nodes / --config) or sync remote nodes with --sync-remote.

11. 🐍 Python API

Embed connection and automation routines programmatically in Python:

import connpy

# 1. Direct single node interaction
router = connpy.node("router1", "1.1.1.1", user="admin")
router.run(["show ip int brief"])
print(router.output)

# 2. Parallel nodes execution with variables
config = connpy.configfile()
nodes_info = config.getitem("@office", ["router1", "router2"])
routers = connpy.nodes(nodes_info, config=config)
variables = {
    "router1@office": {"id": "1"},
    "__global__": {"mask": "255.255.255.0"}
}
routers.run(["interface lo{id}", "ip address 10.0.0.{id} {mask}"], variables)

# 3. AI Copilot prompts
myai = connpy.ai(connpy.configfile())
response = myai.ask("Show BGP status.")
print(response)

Supports additional programmatic features like node.test(), node.interact(), configfile.encrypt(), connapp embeds, and ClassHook / MethodHook plugin hooks.

12. 🐳 Docker Deployment

Run connpy containerized and silent:

docker compose run --rm connpy-app [command]

Add alias conn='docker compose run --rm connpy-app' for a transparent container experience.

13. 📜 License

PolyForm Noncommercial 1.0.0

Sub-modules

connpy.ai

connpy.cli

connpy.grpc_layer

connpy.mcp_client

connpy.proto

connpy.services

connpy.tunnels

connpy.utils

Classes

class Plugins

Expand source code

class Plugins:
    def __init__(self):
        self.plugins = {}
        self.plugin_parsers = {}
        self.preloads = {}
        self.remote_plugins = {}
        self.preferences = {}

    def _load_preferences(self, config_dir):
        import json
        path = os.path.join(config_dir, "plugin_preferences.json")
        try:
            with open(path) as f:
                self.preferences = json.load(f)
        except (FileNotFoundError, json.JSONDecodeError):
            self.preferences = {}

    def _save_preferences(self, config_dir):
        import json
        path = os.path.join(config_dir, "plugin_preferences.json")
        try:
            with open(path, "w") as f:
                json.dump(self.preferences, f, indent=4)
        except OSError as e:
            printer.error(f"Failed to save plugin preferences: {e}")


    def verify_script(self, file_path):
        """
        Verifies that a given Python script meets specific structural requirements.

        This function checks a Python script for compliance with predefined structural 
        rules. It ensures that the script contains only allowed top-level elements 
        (functions, classes, imports, pass statements, and a specific if __name__ block) 
        and that it includes mandatory classes with specific attributes and methods.

        ### Arguments:
            - file_path (str): The file path of the Python script to be verified.

        ### Returns:
            - str: A message indicating the type of violation if the script doesn't meet 
                 the requirements, or False if all requirements are met.

        ### Verifications:
            - The presence of only allowed top-level elements.
            - The existence of two specific classes: 'Parser' and 'Entrypoint'. and/or specific class: Preload.
            - 'Parser' class must only have an '__init__' method and must assign 'self.parser'.
            - 'Entrypoint' class must have an '__init__' method accepting specific arguments.

        If any of these checks fail, the function returns an error message indicating 
        the reason. If the script passes all checks, the function returns False, 
        indicating successful verification.

        ### Exceptions:
                - SyntaxError: If the script contains a syntax error, it is caught and 
                               returned as a part of the error message.
        """
        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}"


        has_parser = False
        has_entrypoint = False
        has_preload = False

        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
                        ((hasattr(ast, 'Str') and isinstance(node.test.comparators[0], getattr(ast, 'Str')) and node.test.comparators[0].s == '__main__') or
                         (hasattr(ast, 'Constant') and isinstance(node.test.comparators[0], getattr(ast, 'Constant')) and node.test.comparators[0].value == '__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):

                if node.name == 'Parser':
                    has_parser = True
                    # 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' is 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:
                        return "Parser class should set self.parser"


                elif node.name == 'Entrypoint':
                    has_entrypoint = True
                    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 have method __init__ and accept only arguments: args, parser and connapp"  # 'Entrypoint' __init__ does not have correct signature

                elif node.name == 'Preload':
                    has_preload = True
                    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) != 2:  # self, connapp
                        return "Preload class should have method __init__ and accept only argument: connapp"  # 'Preload' __init__ does not have correct signature

        # Applying the combination logic based on class presence
        if has_parser and not has_entrypoint:
            return "Parser requires Entrypoint class to be present."
        elif has_entrypoint and not has_parser:
            return "Entrypoint requires Parser class to be present."
    
        if not (has_parser or has_entrypoint or has_preload):
            return "No valid class (Parser, Entrypoint, or Preload) found."

        return False  # All requirements met, no error

    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, remote_enabled=False):
        if not os.path.exists(directory):
            return
        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
                
                # Check preferences: if remote is preferred AND remote is enabled, skip local loading
                if remote_enabled and self.preferences.get(root_filename) == "remote":
                    continue

                # Construct the full path
                filepath = os.path.join(directory, filename)
                check_file = self.verify_script(filepath)
                if check_file:
                    printer.error(f"Failed to load plugin: {filename}. Reason: {check_file}")
                    continue
                else:
                    self.plugins[root_filename] = self._import_from_path(filepath)
                    if hasattr(self.plugins[root_filename], "Parser"):
                        self.plugin_parsers[root_filename] = self.plugins[root_filename].Parser()
                        plugin = self.plugin_parsers[root_filename]
                        # Default to RichHelpFormatter if plugin doesn't set one
                        try:
                            from rich_argparse import RichHelpFormatter as _RHF
                            fmt = plugin.parser.formatter_class
                            if fmt is argparse.HelpFormatter or fmt is argparse.RawTextHelpFormatter or fmt is argparse.RawDescriptionHelpFormatter:
                                fmt = _RHF
                        except ImportError:
                            fmt = plugin.parser.formatter_class
                        subparsers.add_parser(root_filename, parents=[self.plugin_parsers[root_filename].parser], add_help=False, help=plugin.parser.description, usage=plugin.parser.usage, description=plugin.parser.description, epilog=plugin.parser.epilog, formatter_class=fmt)
                    if hasattr(self.plugins[root_filename], "Preload"):
                        self.preloads[root_filename] = self.plugins[root_filename]

    def _import_remote_plugins_to_argparse(self, plugin_stub, subparsers, cache_dir, force_sync=False):
        import hashlib
        os.makedirs(cache_dir, exist_ok=True)
        
        try:
            remote_plugins_info = plugin_stub.list_plugins()
        except Exception:
            return

        # Pruning: Remove local cached files that are no longer on the server
        for local_file in os.listdir(cache_dir):
            if local_file.endswith(".py"):
                name = local_file[:-3]
                if name not in remote_plugins_info:
                    try:
                        os.remove(os.path.join(cache_dir, local_file))
                    except Exception:
                        pass

        for name, info in remote_plugins_info.items():
            if not info.get("enabled", True):
                continue
                
            pref = self.preferences.get(name, "local")
            if pref != "remote" and name in self.plugins:
                continue
            if not force_sync and name in subparsers.choices:
                continue

            cache_path = os.path.join(cache_dir, f"{name}.py")
            
            # Hash comparison
            remote_hash = info.get("hash", "")
            local_hash = ""
            if os.path.exists(cache_path):
                try:
                    with open(cache_path, "rb") as f:
                        local_hash = hashlib.md5(f.read()).hexdigest()
                except Exception:
                    pass

            # Update only if hash differs or force_sync is True
            if force_sync or remote_hash != local_hash or not os.path.exists(cache_path):
                try:
                    source = plugin_stub.get_plugin_source(name)
                    with open(cache_path, "w") as f:
                        f.write(source)
                except Exception as e:
                    printer.warning(f"Failed to sync remote plugin {name}: {e}")
                    continue

            # Verify and load
            check_file = self.verify_script(cache_path)
            if check_file:
                printer.warning(f"Remote plugin {name} failed verification: {check_file}")
                continue

            module = self._import_from_path(cache_path)
            if hasattr(module, "Parser"):
                self.plugin_parsers[name] = module.Parser()
                self.remote_plugins[name] = True
                plugin = self.plugin_parsers[name]
                try:
                    from rich_argparse import RichHelpFormatter as _RHF
                    fmt = plugin.parser.formatter_class
                    if fmt is argparse.HelpFormatter or fmt is argparse.RawTextHelpFormatter or fmt is argparse.RawDescriptionHelpFormatter:
                        fmt = _RHF
                except ImportError:
                    fmt = plugin.parser.formatter_class
                
                # If force_sync, we might be re-registering, but argparse subparsers.add_parser 
                # might fail if it exists. We check if it's already there.
                if name not in subparsers.choices:
                    subparsers.add_parser(
                        name, 
                        parents=[plugin.parser], 
                        add_help=False, 
                        help=f"[remote] {plugin.parser.description}", 
                        usage=plugin.parser.usage, 
                        description=plugin.parser.description, 
                        epilog=plugin.parser.epilog, 
                        formatter_class=fmt
                    )

Methods

def verify_script(self, file_path)

Expand source code

def verify_script(self, file_path):
    """
    Verifies that a given Python script meets specific structural requirements.

    This function checks a Python script for compliance with predefined structural 
    rules. It ensures that the script contains only allowed top-level elements 
    (functions, classes, imports, pass statements, and a specific if __name__ block) 
    and that it includes mandatory classes with specific attributes and methods.

    ### Arguments:
        - file_path (str): The file path of the Python script to be verified.

    ### Returns:
        - str: A message indicating the type of violation if the script doesn't meet 
             the requirements, or False if all requirements are met.

    ### Verifications:
        - The presence of only allowed top-level elements.
        - The existence of two specific classes: 'Parser' and 'Entrypoint'. and/or specific class: Preload.
        - 'Parser' class must only have an '__init__' method and must assign 'self.parser'.
        - 'Entrypoint' class must have an '__init__' method accepting specific arguments.

    If any of these checks fail, the function returns an error message indicating 
    the reason. If the script passes all checks, the function returns False, 
    indicating successful verification.

    ### Exceptions:
            - SyntaxError: If the script contains a syntax error, it is caught and 
                           returned as a part of the error message.
    """
    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}"


    has_parser = False
    has_entrypoint = False
    has_preload = False

    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
                    ((hasattr(ast, 'Str') and isinstance(node.test.comparators[0], getattr(ast, 'Str')) and node.test.comparators[0].s == '__main__') or
                     (hasattr(ast, 'Constant') and isinstance(node.test.comparators[0], getattr(ast, 'Constant')) and node.test.comparators[0].value == '__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):

            if node.name == 'Parser':
                has_parser = True
                # 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' is 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:
                    return "Parser class should set self.parser"


            elif node.name == 'Entrypoint':
                has_entrypoint = True
                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 have method __init__ and accept only arguments: args, parser and connapp"  # 'Entrypoint' __init__ does not have correct signature

            elif node.name == 'Preload':
                has_preload = True
                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) != 2:  # self, connapp
                    return "Preload class should have method __init__ and accept only argument: connapp"  # 'Preload' __init__ does not have correct signature

    # Applying the combination logic based on class presence
    if has_parser and not has_entrypoint:
        return "Parser requires Entrypoint class to be present."
    elif has_entrypoint and not has_parser:
        return "Entrypoint requires Parser class to be present."

    if not (has_parser or has_entrypoint or has_preload):
        return "No valid class (Parser, Entrypoint, or Preload) found."

    return False  # All requirements met, no error

Verifies that a given Python script meets specific structural requirements.

This function checks a Python script for compliance with predefined structural rules. It ensures that the script contains only allowed top-level elements (functions, classes, imports, pass statements, and a specific if name block) and that it includes mandatory classes with specific attributes and methods.

Arguments:

- file_path (str): The file path of the Python script to be verified.

Returns:

- str: A message indicating the type of violation if the script doesn't meet 
     the requirements, or False if all requirements are met.

Verifications:

- The presence of only allowed top-level elements.
- The existence of two specific classes: 'Parser' and 'Entrypoint'. and/or specific class: Preload.
- 'Parser' class must only have an '__init__' method and must assign 'self.parser'.
- 'Entrypoint' class must have an '__init__' method accepting specific arguments.

If any of these checks fail, the function returns an error message indicating the reason. If the script passes all checks, the function returns False, indicating successful verification.

Exceptions:

    - SyntaxError: If the script contains a syntax error, it is caught and 
                   returned as a part of the error message.

class configfile (conf=None, key=None, shared_config=None)

Expand source code

@ClassHook
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, shared_config = None):
        self._shared_config = shared_config
        ''' 
            
        ### Optional Parameters:  

            - conf (str): Path/file to config file. If left empty default
                          path is ~/.config/conn/config.yaml

            - 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'
        
        if conf is None:
            # Standard path: use ~/.config/conn and respect .folder redirection
            self.anchor_path = defaultdir
            self.defaultdir = defaultdir
            Path(defaultdir).mkdir(parents=True, exist_ok=True)
            
            pathfile = defaultdir + '/.folder'
            try:
                with open(pathfile, "r") as f:
                    configdir = f.read().strip()
            except (FileNotFoundError, IOError):
                with open(pathfile, "w") as f:
                    f.write(str(defaultdir))
                configdir = defaultdir
            
            self.defaultdir = configdir
            self.file = configdir + '/config.yaml'
            self.key = key or (configdir + '/.osk')

            # Ensure redirected directories exist
            Path(configdir).mkdir(parents=True, exist_ok=True)
            Path(f"{configdir}/plugins").mkdir(parents=True, exist_ok=True)
            
            # Backwards compatibility: Migrate from JSON to YAML only for default path
            legacy_json = configdir + '/config.json'
            legacy_noext = configdir + '/config'
            legacy_file = None
            if os.path.exists(legacy_json): legacy_file = legacy_json
            elif os.path.exists(legacy_noext): legacy_file = legacy_noext
            
            if not os.path.exists(self.file) and legacy_file:
                try:
                    with open(legacy_file, 'r') as f:
                        old_data = json.load(f)
                    if not self._validate_config(old_data):
                        printer.warning(f"Legacy config {legacy_file} has invalid structure, skipping migration.")
                    else:
                        with open(self.file, 'w') as f:
                            yaml.dump(old_data, f, Dumper=NoAliasDumper, default_flow_style=False, sort_keys=False)
                        # Verify the written YAML can be read back correctly
                        with open(self.file, 'r') as f:
                            verify = yaml.safe_load(f)
                        if not self._validate_config(verify):
                            os.remove(self.file)
                            printer.warning("YAML verification failed after migration, keeping legacy config.")
                        else:
                            # Note: cachefile is derived later, we use temp one for migration sync
                            temp_cache = configdir + '/.config.cache.json'
                            with open(temp_cache, 'w') as f:
                                json.dump(old_data, f)
                            shutil.move(legacy_file, legacy_file + ".backup")
                            printer.success(f"Migrated legacy config ({len(old_data.get('connections',{}))} folders/nodes) into YAML and Cache successfully!")
                except Exception as e:
                    if os.path.exists(self.file):
                        try: os.remove(self.file)
                        except OSError: pass
                    printer.warning(f"Failed to migrate legacy config: {e}")
        else:
            # Custom path (common in tests): isolate everything to the conf parent directory
            self.file = os.path.abspath(conf)
            configdir = os.path.dirname(self.file)
            self.anchor_path = configdir
            self.defaultdir = configdir
            self.key = os.path.abspath(key) if key else (configdir + '/.osk')

        # Sidecar files always live next to the config file (or in the redirected configdir)
        self.cachefile = configdir + '/.config.cache.json'
        self.fzf_cachefile = configdir + '/.fzf_nodes_cache.txt'
        self.folders_cachefile = configdir + '/.folders_cache.txt'
        self.profiles_cachefile = configdir + '/.profiles_cache.txt'
            
        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())
        self.publickey = self.privatekey.publickey()

        # Self-heal text caches if they are missing
        if not os.path.exists(self.fzf_cachefile) or not os.path.exists(self.folders_cachefile) or not os.path.exists(self.profiles_cachefile):
            self._generate_nodes_cache()


    def get_effective_setting(self, key, default=None):
        """Get config setting with shared fallback for inheritable keys."""
        val = self.config.get(key)
        if key == "ai":
            if val is not None:
                if self._shared_config:
                    import copy
                    # Deep merge: shared as base, user overrides
                    base = copy.deepcopy(self._shared_config.config.get(key, {}))
                    if isinstance(base, dict) and isinstance(val, dict):
                        # Credential isolation:
                        # If user defines engineer credentials, discard shared ones
                        if "engineer_api_key" in val or "engineer_auth" in val:
                            base.pop("engineer_api_key", None)
                            base.pop("engineer_auth", None)
                        # If user defines architect credentials, discard shared ones
                        if "architect_api_key" in val or "architect_auth" in val:
                            base.pop("architect_api_key", None)
                            base.pop("architect_auth", None)
                            
                        # Recursive update for inner dictionaries (like mcp_servers or model details)
                        def deep_merge(d1, d2):
                            for k, v in d2.items():
                                if isinstance(v, dict) and k in d1 and isinstance(d1[k], dict):
                                    deep_merge(d1[k], v)
                                else:
                                    d1[k] = copy.deepcopy(v)
                        deep_merge(base, val)
                        return base
                return val
            elif self._shared_config:
                return self._shared_config.config.get(key, default)
        
        return val if val is not None else default


    def _validate_config(self, data):
        """Verify config data has the required structure."""
        if not isinstance(data, dict):
            return False
        required = {"config", "connections", "profiles"}
        return required.issubset(data.keys())

    def _loadconfig(self, conf):
        #Loads config file using dual cache
        cache_exists = os.path.exists(self.cachefile)
        yaml_time = os.path.getmtime(conf) if os.path.exists(conf) else 0
        cache_time = os.path.getmtime(self.cachefile) if cache_exists else 0

        if not cache_exists or yaml_time > cache_time:
            with open(conf, 'r') as f:
                data = yaml.safe_load(f)
            if not self._validate_config(data):
                # YAML is broken, try to recover from cache
                if cache_exists:
                    printer.warning("Config file appears corrupt, recovering from cache...")
                    with open(self.cachefile, 'r') as f:
                        data = json.load(f)
                    if self._validate_config(data):
                        # Re-write the YAML from good cache
                        with open(conf, 'w') as f:
                            yaml.dump(data, f, Dumper=NoAliasDumper, default_flow_style=False, sort_keys=False)
                        return data
                # Both broken or no cache - create fresh
                printer.error("Config file is corrupt and no valid cache exists. Creating default config.")
                return self._createconfig(conf)
            try:
                with open(self.cachefile, 'w') as f:
                    json.dump(data, f)
            except Exception:
                pass
            return data
        else:
            with open(self.cachefile, 'r') as f:
                data = json.load(f)
            if not self._validate_config(data):
                # Cache broken, try yaml
                with open(conf, 'r') as f:
                    data = yaml.safe_load(f)
                if self._validate_config(data):
                    return data
                # Both broken
                printer.error("Both config and cache are corrupt. Creating default config.")
                return self._createconfig(conf)
            return data

    def _createconfig(self, conf):
        #Create config file (always writes defaults, safe for recovery)
        defaultconfig = {'config': {'case': False, 'idletime': 30, 'fzf': False}, 'connections': {}, 'profiles': { "default": { "host":"", "protocol":"ssh", "port":"", "user":"", "password":"", "options":"", "logs":"", "tags": "", "jumphost":""}}}
        with open(conf, "w") as f:
            yaml.dump(defaultconfig, f, Dumper=NoAliasDumper, default_flow_style=False, sort_keys=False)
        os.chmod(conf, 0o600)
        try:
            with open(self.cachefile, 'w') as f:
                json.dump(defaultconfig, f)
        except Exception:
            pass
        return defaultconfig

    @MethodHook
    def _saveconfig(self, conf):
        #Save config file atomically to prevent corruption
        newconfig = {"config":{}, "connections": {}, "profiles": {}}
        newconfig["config"] = self.config
        newconfig["connections"] = self.connections
        newconfig["profiles"] = self.profiles
        tmpfile = conf + '.tmp'
        try:
            with open(tmpfile, "w") as f:
                yaml.dump(newconfig, f, Dumper=NoAliasDumper, default_flow_style=False, sort_keys=False)
            # Atomic replace: only overwrite original if write succeeded
            shutil.move(tmpfile, conf)
            with open(self.cachefile, "w") as f:
                json.dump(newconfig, f)
            self._generate_nodes_cache()
        except (IOError, OSError) as e:
            printer.error(f"Failed to save config: {e}")
            # Clean up temp file if it exists
            if os.path.exists(tmpfile):
                try:
                    os.remove(tmpfile)
                except OSError:
                    pass
            return 1
        return 0

    def _generate_nodes_cache(self, nodes=None, folders=None, profiles=None):
        try:
            if nodes is None:
                nodes = self._getallnodes()
            if folders is None:
                folders = self._getallfolders()
            if profiles is None:
                profiles = list(self.profiles.keys())
            
            with open(self.fzf_cachefile, "w") as f:
                f.write("\n".join(nodes))
            with open(self.folders_cachefile, "w") as f:
                f.write("\n".join(folders))
            with open(self.profiles_cachefile, "w") as f:
                f.write("\n".join(profiles))
        except Exception:
            pass


    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

    @MethodHook
    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

    @MethodHook
    def getitem(self, unique, keys = None, extract = False):
        '''
        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.
            - extract (bool): If True, extract information from profiles. 
                              Default False.

        ### 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_name in folder.keys():
                if node_name == "type":
                    continue
                if "type" in newfolder[node_name].keys():
                    if newfolder[node_name]["type"] == "subfolder":
                        newfolder.pop(node_name)
                    else:
                        newfolder[node_name].pop("type")
            
            if keys != None:
                newfolder = dict((k, newfolder[k]) for k in keys)
            
            if extract:
                for node_name, node_keys in newfolder.items():
                    for key, value in node_keys.items():
                        profile = re.search("^@(.*)", str(value))
                        if profile:
                            try:
                                newfolder[node_name][key] = self.profiles[profile.group(1)][key]
                            except KeyError:
                                newfolder[node_name][key] = ""
                        elif value == '' and key == "protocol":
                            try:
                                newfolder[node_name][key] = self.profiles["default"][key]
                            except KeyError:
                                newfolder[node_name][key] = "ssh"
            
            newfolder = {"{}{}".format(k,unique):v for k,v in newfolder.items()}
            return 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")
            
            if extract:
                for key, value in newnode.items():
                    profile = re.search("^@(.*)", str(value))
                    if profile:
                        try:
                            newnode[key] = self.profiles[profile.group(1)][key]
                        except KeyError:
                            newnode[key] = ""
                    elif value == '' and key == "protocol":
                        try:
                            newnode[key] = self.profiles["default"][key]
                        except KeyError:
                            newnode[key] = "ssh"
            return newnode

    @MethodHook
    def getitems(self, uniques, extract = False):
        '''
        Get a group of nodes from configfile which can be passed to node/nodes class

        ### Parameters:  

            - uniques (str/list): String name that will match hostnames 
                                  from the connection manager. It can be a 
                                  list of strings.

        ### Optional Parameters:

            - extract (bool): If True, extract information from profiles. 
                              Default False.

        ### Returns:  

            dict: Dictionary containing information of node or multiple 
                  dictionaries of multiple nodes.

        '''
        nodes = {}
        if isinstance(uniques, str):
            uniques = [uniques]
        for i in uniques:
            if i.startswith("@"):
                if not self.config["case"]:
                    i = i.lower()
                this = self.getitem(i, extract = extract)
                nodes.update(this)
            else:
                if not self.config["case"]:
                    i = i.lower()
                this = self.getitem(i, extract = extract)
                nodes[i] = this
        return nodes


    @MethodHook
    def _connections_add(self,*, id, host, folder='', subfolder='', options='', logs='', password='', port='', protocol='', user='', tags='', jumphost='', 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,"jumphost": jumphost,"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, "jumphost": jumphost, "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,  "jumphost": jumphost, "type": type}
            

    @MethodHook
    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]

    @MethodHook
    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"}

    @MethodHook
    def _folder_del(self,*, folder, subfolder=''):
        #Delete folder from config
        if subfolder == '':
            del self.connections[folder]
        else:
            del self.connections[folder][subfolder]


    @MethodHook
    def _profiles_add(self,*, id, host = '', options='', logs='', password='', port='', protocol='', user='', tags='', jumphost='' ):
        #Add profile from config
        self.profiles[id] = {"host": host, "options": options, "logs": logs, "password": password, "port": port, "protocol": protocol, "user": user, "tags": tags, "jumphost": jumphost}
            

    @MethodHook
    def _profiles_del(self,*, id ):
        #Delete profile from config
        del self.profiles[id]
        
    @MethodHook
    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.get("type") == "connection"]
        folders = [k for k,v in self.connections.items() if isinstance(v, dict) and v.get("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.get("type") == "connection"]
            nodes.extend(layer2)
            subfolders = [k for k,v in self.connections[f].items() if isinstance(v, dict) and v.get("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.get("type") == "connection"]
                nodes.extend(layer3)
        if filter:
            flat_filter = []
            if isinstance(filter, str):
                flat_filter = [filter]
            elif isinstance(filter, list):
                for item in filter:
                    if isinstance(item, str):
                        flat_filter.append(item)
            else:
                printer.error("Filter must be a string or a list of strings")
                sys.exit(1)
            flags = re.IGNORECASE if not self.config.get("case", False) else 0
            nodes = [item for item in nodes if any(re.search(pattern, item, flags) for pattern in flat_filter)]
        return nodes

    @MethodHook
    def _getallnodesfull(self, filter = None, extract = True):
        #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.get("type") == "connection"}
        folders = [k for k,v in self.connections.items() if isinstance(v, dict) and v.get("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.get("type") == "connection"}
            nodes.update(layer2)
            subfolders = [k for k,v in self.connections[f].items() if isinstance(v, dict) and v.get("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.get("type") == "connection"}
                nodes.update(layer3)
        if filter:
            flat_filter = []
            if isinstance(filter, str):
                flat_filter = [filter]
            elif isinstance(filter, list):
                for item in filter:
                    if isinstance(item, str):
                        flat_filter.append(item)
            else:
                printer.error("Filter must be a string or a list of strings")
                sys.exit(1)
            flat_filter = ["^(?!.*@).+$" if item == "@" else item for item in flat_filter]
            nodes = {k: v for k, v in nodes.items() if any(re.search(pattern, k) for pattern in flat_filter)}
        if extract:
            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 KeyError:
                            nodes[node][key] = ""
                    elif value == '' and key == "protocol":
                        try:
                            nodes[node][key] = self.profiles["default"][key]
                        except KeyError:
                            nodes[node][key] = "ssh"
        return nodes


    @MethodHook
    def _getallfolders(self):
        #get all folders on configfile
        folders = ["@" + k for k,v in self.connections.items() if isinstance(v, dict) and v.get("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.get("type") == "subfolder"]
            subfolders.extend(s)
        folders.extend(subfolders)
        return folders

    @MethodHook
    def _profileused(self, profile):
        #Return all the nodes that uses this profile.
        nodes = []
        layer1 = [k for k,v in self.connections.items() if isinstance(v, dict) and v.get("type") == "connection" and ("@" + profile in v.values() or ( isinstance(v.get("password"),list) and "@" + profile in v.get("password")))]
        folders = [k for k,v in self.connections.items() if isinstance(v, dict) and v.get("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.get("type") == "connection" and ("@" + profile in v.values() or ( isinstance(v.get("password"),list) and "@" + profile in v.get("password")))]
            nodes.extend(layer2)
            subfolders = [k for k,v in self.connections[f].items() if isinstance(v, dict) and v.get("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.get("type") == "connection" and ("@" + profile in v.values() or ( isinstance(v.get("password"),list) and "@" + profile in v.get("password")))]
                nodes.extend(layer3)
        return nodes

    @MethodHook
    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.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)

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.

Methods

def encrypt(self, password, keyfile=None)

Expand source code

@MethodHook
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.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)

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.

def get_effective_setting(self, key, default=None)

Expand source code

def get_effective_setting(self, key, default=None):
    """Get config setting with shared fallback for inheritable keys."""
    val = self.config.get(key)
    if key == "ai":
        if val is not None:
            if self._shared_config:
                import copy
                # Deep merge: shared as base, user overrides
                base = copy.deepcopy(self._shared_config.config.get(key, {}))
                if isinstance(base, dict) and isinstance(val, dict):
                    # Credential isolation:
                    # If user defines engineer credentials, discard shared ones
                    if "engineer_api_key" in val or "engineer_auth" in val:
                        base.pop("engineer_api_key", None)
                        base.pop("engineer_auth", None)
                    # If user defines architect credentials, discard shared ones
                    if "architect_api_key" in val or "architect_auth" in val:
                        base.pop("architect_api_key", None)
                        base.pop("architect_auth", None)
                        
                    # Recursive update for inner dictionaries (like mcp_servers or model details)
                    def deep_merge(d1, d2):
                        for k, v in d2.items():
                            if isinstance(v, dict) and k in d1 and isinstance(d1[k], dict):
                                deep_merge(d1[k], v)
                            else:
                                d1[k] = copy.deepcopy(v)
                    deep_merge(base, val)
                    return base
            return val
        elif self._shared_config:
            return self._shared_config.config.get(key, default)
    
    return val if val is not None else default

Get config setting with shared fallback for inheritable keys.

def getitem(self, unique, keys=None, extract=False)

Expand source code

@MethodHook
def getitem(self, unique, keys = None, extract = False):
    '''
    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.
        - extract (bool): If True, extract information from profiles. 
                          Default False.

    ### 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_name in folder.keys():
            if node_name == "type":
                continue
            if "type" in newfolder[node_name].keys():
                if newfolder[node_name]["type"] == "subfolder":
                    newfolder.pop(node_name)
                else:
                    newfolder[node_name].pop("type")
        
        if keys != None:
            newfolder = dict((k, newfolder[k]) for k in keys)
        
        if extract:
            for node_name, node_keys in newfolder.items():
                for key, value in node_keys.items():
                    profile = re.search("^@(.*)", str(value))
                    if profile:
                        try:
                            newfolder[node_name][key] = self.profiles[profile.group(1)][key]
                        except KeyError:
                            newfolder[node_name][key] = ""
                    elif value == '' and key == "protocol":
                        try:
                            newfolder[node_name][key] = self.profiles["default"][key]
                        except KeyError:
                            newfolder[node_name][key] = "ssh"
        
        newfolder = {"{}{}".format(k,unique):v for k,v in newfolder.items()}
        return 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")
        
        if extract:
            for key, value in newnode.items():
                profile = re.search("^@(.*)", str(value))
                if profile:
                    try:
                        newnode[key] = self.profiles[profile.group(1)][key]
                    except KeyError:
                        newnode[key] = ""
                elif value == '' and key == "protocol":
                    try:
                        newnode[key] = self.profiles["default"][key]
                    except KeyError:
                        newnode[key] = "ssh"
        return newnode

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.
- extract (bool): If True, extract information from profiles. 
                  Default False.

Returns:

dict: Dictionary containing information of node or multiple 
      dictionaries of multiple nodes.

def getitems(self, uniques, extract=False)

Expand source code

@MethodHook
def getitems(self, uniques, extract = False):
    '''
    Get a group of nodes from configfile which can be passed to node/nodes class

    ### Parameters:  

        - uniques (str/list): String name that will match hostnames 
                              from the connection manager. It can be a 
                              list of strings.

    ### Optional Parameters:

        - extract (bool): If True, extract information from profiles. 
                          Default False.

    ### Returns:  

        dict: Dictionary containing information of node or multiple 
              dictionaries of multiple nodes.

    '''
    nodes = {}
    if isinstance(uniques, str):
        uniques = [uniques]
    for i in uniques:
        if i.startswith("@"):
            if not self.config["case"]:
                i = i.lower()
            this = self.getitem(i, extract = extract)
            nodes.update(this)
        else:
            if not self.config["case"]:
                i = i.lower()
            this = self.getitem(i, extract = extract)
            nodes[i] = this
    return nodes

Get a group of nodes from configfile which can be passed to node/nodes class

Parameters:

- uniques (str/list): String name that will match hostnames 
                      from the connection manager. It can be a 
                      list of strings.

Optional Parameters:

- extract (bool): If True, extract information from profiles. 
                  Default False.

Returns:

dict: Dictionary containing information of node or multiple 
      dictionaries of multiple nodes.

class node (unique,
host,
options='',
logs='',
password='',
port='',
protocol='',
user='',
config='',
tags='',
jumphost='')

Expand source code

@ClassHook
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 successfully.
                        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='', jumphost=''):
        ''' 
            
        ### 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, telnet, kubectl or docker. 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"
                              
            - jumphost (str): Reference another node to be used as a jumphost
        '''
        self.config = config
        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, "jumphost": jumphost}
        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 KeyError:
                    setattr(self,key,"")
            elif attr[key] == '' and key == "protocol":
                try:
                    setattr(self,key,config.profiles["default"][key])
                except (KeyError, AttributeError):
                    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.append(password[i])
        else:
            self.password = [password]
        if self.jumphost != "" and config != '':
            raw_cmd, jh_passwords = self._build_jumphost_chain(self.jumphost, config)
            if jh_passwords:
                self.password = jh_passwords + self.password
            if raw_cmd:
                escaped = raw_cmd.replace('\\', '\\\\').replace('"', '\\"')
                self.jumphost = f'-o ProxyCommand="{escaped}"'
            else:
                self.jumphost = ""
        
        self.output = ""
        self.status = 1
        self.result = {}
        self.cmd_byte_positions = [(0, None)]

    @staticmethod
    def _resolve_jumphost_data(jh_dict, config):
        '''Resolve @profile references and normalize passwords in a jumphost dict.'''
        for key in jh_dict:
            profile = re.search("^@(.*)", str(jh_dict[key]))
            if profile:
                try:
                    jh_dict[key] = config.profiles[profile.group(1)][key]
                except KeyError:
                    jh_dict[key] = ""
            elif jh_dict[key] == '' and key == "protocol":
                try:
                    jh_dict[key] = config.profiles["default"][key]
                except KeyError:
                    jh_dict[key] = "ssh"
        if isinstance(jh_dict["password"], list):
            resolved = []
            for p in jh_dict["password"]:
                profile = re.search("^@(.*)", p)
                if profile:
                    resolved.append(config.profiles[profile.group(1)]["password"])
                else:
                    resolved.append(p)
            jh_dict["password"] = resolved
        else:
            jh_dict["password"] = [jh_dict["password"]]
        return jh_dict

    def _build_jumphost_chain(self, jumphost_name, config, visited=None, depth=0, target_host="%h", target_port="%p"):
        '''Recursively build ProxyCommand for chained jumphosts.

        Returns:
            tuple: (raw_proxy_command, passwords_list)
                - raw_proxy_command: Command string to embed in ProxyCommand
                - passwords_list: Ordered passwords (innermost first)

        Raises:
            ValueError: On circular references or exceeding max depth (5).
        '''
        if depth >= 5:
            raise ValueError("Jumphost chain exceeds maximum depth of 5 hops")
        if visited is None:
            visited = []
        if jumphost_name in visited:
            cycle = " -> ".join(visited + [jumphost_name])
            raise ValueError(f"Circular jumphost reference detected: {cycle}")
        visited = visited + [jumphost_name]

        jh = config.getitem(jumphost_name)
        jh = self._resolve_jumphost_data(jh, config)

        passwords = []
        inner_proxy_opt = ""

        # Recursively resolve inner jumphost
        if jh.get("jumphost", "") != "":
            if jh["protocol"] not in ["ssh"]:
                raise ValueError(
                    f"Jumphost '{jumphost_name}' uses protocol '{jh['protocol']}' "
                    f"which does not support chained jumphosts. "
                    f"Only SSH jumphosts can have their own jumphosts."
                )
            parent_port = jh["port"] if jh["port"] != "" else "22"
            inner_raw_cmd, inner_passwords = self._build_jumphost_chain(
                jh["jumphost"], config, visited, depth + 1, target_host=jh["host"], target_port=parent_port
            )
            passwords = inner_passwords
            escaped = inner_raw_cmd.replace('\\', '\\\\').replace('"', '\\"')
            inner_proxy_opt = f'-o ProxyCommand="{escaped}"'

        # Collect this hop's passwords
        if jh["password"] != [""]:
            passwords = passwords + jh["password"]

        t_port = target_port if target_port != "" else "22"

        # Build raw command based on protocol
        if jh["protocol"] == "ssh":
            cmd = f"ssh -W {target_host}:{t_port}"
            if inner_proxy_opt:
                cmd += f" {inner_proxy_opt}"
            if jh["port"] != '':
                cmd += f" -p {jh['port']}"
            if jh["options"] != '':
                cmd += f" {jh['options']}"
            user_host = f"{jh['user']}@{jh['host']}" if jh['user'] != '' else jh['host']
            cmd += f" {user_host}"
        elif jh["protocol"] == "ssm":
            ssm_target = jh["host"]
            ssm_cmd = f"aws ssm start-session --target {ssm_target} --document-name AWS-StartSSHSession --parameters 'portNumber=22'"
            if isinstance(jh.get("tags"), dict):
                if "profile" in jh["tags"]:
                    ssm_cmd += f" --profile {jh['tags']['profile']}"
                if "region" in jh["tags"]:
                    ssm_cmd += f" --region {jh['tags']['region']}"
            if jh["options"] != '':
                ssm_cmd += f" {jh['options']}"
            bastion_user_part = f"{jh['user']}@{ssm_target}" if jh['user'] else ssm_target
            ssh_opts = ""
            if isinstance(jh.get("tags"), dict) and "ssh_options" in jh["tags"]:
                ssh_opts = f" {jh['tags']['ssh_options']}"
            cmd = f"ssh{ssh_opts} -o ProxyCommand='{ssm_cmd}' -W {target_host}:{t_port} {bastion_user_part}"
        elif jh["protocol"] in ["kubectl", "docker"]:
            nc_cmd = "nc"
            if isinstance(jh.get("tags"), dict) and "nc_command" in jh["tags"]:
                nc_cmd = jh["tags"]["nc_command"]
            if jh["protocol"] == "kubectl":
                cmd = "kubectl exec "
                if jh["options"] != '':
                    cmd += f"{jh['options']} "
                cmd += f"{jh['host']} -i -- {nc_cmd} {target_host} {t_port}"
            else:
                cmd = "docker "
                if jh["options"] != '':
                    cmd += f"{jh['options']} "
                cmd += f"exec -i {jh['host']} {nc_cmd} {target_host} {t_port}"
        else:
            return "", passwords

        return cmd, passwords

    @MethodHook
    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 Exception:
                    printer.error("Decryption failed: Missing or corrupted key.")
                    printer.info("Verify your RSA key and configuration settings.")
                    sys.exit(1)
        return dpass

    

    @MethodHook
    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

    @MethodHook
    def _logclean(self, logfile, var = False):
        """Remove special ascii characters and process terminal cursor movements to clean logs."""
        from .utils import log_cleaner
        
        if var == False:
            try:
                with open(logfile, "r") as f:
                    t = f.read()
            except:
                return
        else:
            t = logfile
            
        result = log_cleaner(t)

        if var == False:
            try:
                with open(logfile, "w") as f:
                    f.write(result)
            except:
                pass
            return
        else:
            return result

    @MethodHook
    def _savelog(self):
        '''Save the log buffer to the file at regular intervals if there are changes.'''
        t = threading.current_thread()
        prev_size = 0  # Store the previous size of the buffer

        while getattr(t, "do_run", True):  # Check if thread is signaled to stop
            current_size = self.mylog.tell()  # Current size of the buffer

            # Only save if the buffer size has changed
            if current_size != prev_size:
                with open(self.logfile, "w") as f:  # Use "w" to overwrite the file
                    f.write(self._logclean(self.mylog.getvalue().decode(), True))
                prev_size = current_size  # Update the previous size
            sleep(5)

    @MethodHook
    def _filter(self, a):
        #Set time for last input when using interact
        self.lastinput = time()
        return a

    @MethodHook
    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 _setup_interact_environment(self, debug=False, logger=None, async_mode=False):
        try:
            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)))
        except OSError:
            pass
        if logger and self.protocol != "local":
            port_str = f":{self.port}" if self.port and self.protocol not in ["ssm", "kubectl", "docker"] else ""
            logger("success", f"Connected to {self.unique} at {self.host}{port_str} via: {self.protocol}")

        # Always initialize self.mylog to capture terminal context for the AI Copilot
        if not hasattr(self, 'mylog'):
            self.mylog = io.BytesIO()
            
        if not async_mode:
            self.child.logfile_read = self.mylog
            
        # Only start disk-logging tasks if logfile is configured
        if 'logfile' in dir(self):
            if not async_mode:
                # Start the _savelog thread (sync mode)
                log_thread = threading.Thread(target=self._savelog)
                log_thread.daemon = True
                log_thread.start()
        if 'missingtext' in dir(self):
            print(self.child.after.decode(), end='')
        if self.idletime > 0 and not async_mode:
            x = threading.Thread(target=self._keepalive)
            x.daemon = True
            x.start()
        if debug:
            if 'mylog' in dir(self):
                if not async_mode:
                    print(self.mylog.getvalue().decode())

    def _teardown_interact_environment(self):
        if 'logfile' in dir(self) and hasattr(self, 'mylog'):
            with open(self.logfile, "w") as f:
                f.write(self._logclean(self.mylog.getvalue().decode(), True))

    def _is_child_connpy_active(self, child_fd: int) -> bool:
        if self.protocol != "local":
            return False
        try:
            fg_pgid = os.tcgetpgrp(child_fd)
            if fg_pgid <= 0:
                return False
            cmdline_path = f"/proc/{fg_pgid}/cmdline"
            if os.path.exists(cmdline_path):
                with open(cmdline_path, "rb") as f:
                    raw_args = f.read().split(b"\x00")
                    cmdline_str = " ".join([arg.decode(errors="ignore") for arg in raw_args if arg])
                    
                    is_active = False
                    for raw_arg in raw_args:
                        arg_str = raw_arg.decode(errors="ignore")
                        if not arg_str:
                            continue
                        base_name = os.path.basename(arg_str)
                        if base_name in ["conn", "connpy", "connapp"] or "connpy" in arg_str or "connapp" in arg_str:
                            is_active = True
                            break

                    return is_active
        except Exception:
            pass
        return False

    async def _async_interact_loop(self, local_stream, resize_callback, copilot_handler=None):
        local_stream.setup(resize_callback=resize_callback)
        self.current_local_stream = local_stream
        try:
            child_fd = self.child.child_fd
            
            # 1. Flush ghost buffer (Clean UX)
            ghost_buffer = b''
            if getattr(self, 'missingtext', False):
                # If we are missing the password, we MUST show the password prompt
                ghost_buffer = (self.child.after or b'') + (self.child.buffer or b'')
            else:
                # We auto-logged in. Hide the messy password negotiation and just keep any pending live stream.
                ghost_buffer = self.child.buffer or b''

            # Fix user's pet peeve: Strip leading newlines to avoid the empty lines 
            # the router echoes after receiving the password or blank line.
            if not getattr(self, 'missingtext', False):
                ghost_buffer = ghost_buffer.lstrip(b'\r\n ')

            if ghost_buffer:
                # Add a single clean newline so it doesn't merge with the Connected message
                await local_stream.write(b'\r\n' + ghost_buffer)
                if hasattr(self, 'mylog'):
                    self.mylog.write(b'\n' + ghost_buffer)
                    
            self.child.buffer = b''
            self.child.before = b''
            
            # 2. Set child fd non-blocking
            flags = fcntl.fcntl(child_fd, fcntl.F_GETFL)
            fcntl.fcntl(child_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
            
            loop = asyncio.get_running_loop()
            child_reader_queue = asyncio.Queue()
            
            # Reset and track command byte positions for copilot context navigation
            # Each entry is (byte_position, command_text_or_None)
            self.cmd_byte_positions = [(self.mylog.tell() if hasattr(self, 'mylog') else 0, None)]
            
            def _child_read_ready():
                try:
                    # Increase buffer to 64KB for better high-speed handling
                    data = os.read(child_fd, 65536)
                    if data:
                        child_reader_queue.put_nowait(data)
                    else:
                        child_reader_queue.put_nowait(b'')
                except BlockingIOError:
                    pass
                except OSError:
                    child_reader_queue.put_nowait(b'')
                    
            loop.add_reader(child_fd, _child_read_ready)
            self.lastinput = time()
            
            async def ingress_task():
                while True:
                    data = await local_stream.read()
                    if not data:
                        break
                    
                    # Copilot interception
                    if copilot_handler and b'\x00' in data:
                        if self._is_child_connpy_active(child_fd):
                            try:
                                os.write(child_fd, data)
                            except OSError:
                                break
                            self.lastinput = time()
                            continue

                        # Build node info from available metadata and ensure values are strings (not bytes)
                        def to_str(val):
                            if isinstance(val, bytes):
                                return val.decode(errors='replace')
                            return str(val) if val is not None else "unknown"
                            
                        node_info = {
                            "name": to_str(getattr(self, 'unique', 'unknown')), 
                            "host": to_str(getattr(self, 'host', 'unknown'))
                        }
                        if isinstance(getattr(self, 'tags', None), dict):
                            node_info["os"] = to_str(self.tags.get("os", "unknown"))
                            node_info["prompt"] = to_str(self.tags.get("prompt", r'>$|#$|\$$|>.$|#.$|\$.$'))
                        
                        # Invoke copilot (async callback handles UI)
                        await copilot_handler(self.mylog.getvalue(), node_info, local_stream, child_fd, self.cmd_byte_positions)
                        continue
                    
                    # Remove any stray \x00 bytes and forward normally
                    clean_data = data.replace(b'\x00', b'')
                    if clean_data:
                        # Track command boundaries when user hits Enter or presses Ctrl+C
                        if hasattr(self, 'mylog') and (b'\r' in clean_data or b'\n' in clean_data or b'\x03' in clean_data):
                            pos = self.mylog.tell()
                            marker_cmd = "CANCELLED" if b'\x03' in clean_data else None
                            self.cmd_byte_positions.append((pos, marker_cmd))
                            if hasattr(self, 'current_local_stream') and self.current_local_stream is not None:
                                try:
                                    await self.current_local_stream.write(f'\x1b]133;B;{pos}\x07'.encode())
                                except Exception:
                                    pass

                        try:
                            os.write(child_fd, clean_data)
                        except OSError:
                            break
                        self.lastinput = time()
                    
            async def egress_task():
                # Continue stripping newlines from the live stream until we hit real text
                skip_newlines = not getattr(self, 'missingtext', False) and not ghost_buffer
                while True:
                    data = await child_reader_queue.get()
                    if not data:
                        break
                    
                    # Batching Optimization: Drain the queue to batch writes during high-volume bursts
                    # Helps the terminal parse ANSI faster and reduces syscalls.
                    chunks = [data]
                    while not child_reader_queue.empty():
                        try:
                            extra = child_reader_queue.get_nowait()
                            if not extra:
                                chunks.append(b'') # Re-put EOF later or handle it
                                break
                            chunks.append(extra)
                        except asyncio.QueueEmpty:
                            break
                    
                    has_eof = chunks[-1] == b''
                    if has_eof:
                        chunks.pop()
                    
                    if chunks:
                        combined_data = b''.join(chunks)
                        if skip_newlines:
                            stripped = combined_data.lstrip(b'\r\n')
                            if stripped:
                                skip_newlines = False
                                combined_data = stripped
                            else:
                                if has_eof: break
                                continue
                                
                        await local_stream.write(combined_data)
                        if hasattr(self, 'mylog'):
                            self.mylog.write(combined_data)
                    
                    if has_eof:
                        break
                        
            async def keepalive_task():
                while True:
                    await asyncio.sleep(1)
                    if time() - self.lastinput >= self.idletime:
                        try:
                            self.child.sendcontrol("e")
                            self.lastinput = time()
                        except Exception:
                            pass
                            
            async def savelog_task():
                prev_size = 0
                while True:
                    await asyncio.sleep(5)
                    current_size = self.mylog.tell()
                    if current_size != prev_size:
                        try:
                            # Move heavy log cleaning to a thread to avoid freezing the interaction loop
                            raw_log = self.mylog.getvalue().decode(errors='replace')
                            cleaned_log = await asyncio.to_thread(self._logclean, raw_log, True)
                            with open(self.logfile, "w") as f:
                                f.write(cleaned_log)
                            prev_size = current_size
                        except Exception:
                            pass

            async def pwd_tracker_task():
                import socket
                hostname = socket.gethostname()
                last_cwd = None
                while True:
                    await asyncio.sleep(0.3)
                    try:
                        child_pid = getattr(self.child, 'pid', None)
                        if child_pid:
                            new_cwd = os.readlink(f"/proc/{child_pid}/cwd")
                            if new_cwd != last_cwd:
                                last_cwd = new_cwd
                                try:
                                    os.chdir(new_cwd)
                                except Exception:
                                    pass
                                try:
                                    await local_stream.write(f"\033]7;file://{hostname}{new_cwd}\007".encode())
                                except Exception:
                                    pass
                                if isinstance(self.tags, dict):
                                    self.tags["cwd"] = new_cwd
                    except Exception:
                        pass

            try:
                # We wait for either the user (ingress) or the child (egress) to finish
                tasks = [
                    asyncio.create_task(ingress_task()),
                    asyncio.create_task(egress_task())
                ]
                if self.protocol == "local":
                    tasks.append(asyncio.create_task(pwd_tracker_task()))
                if self.idletime > 0:
                    tasks.append(asyncio.create_task(keepalive_task()))
                if hasattr(self, 'logfile') and hasattr(self, 'mylog'):
                    tasks.append(asyncio.create_task(savelog_task()))
                
                done, pending = await asyncio.wait(
                    [tasks[0], tasks[1]], 
                    return_when=asyncio.FIRST_COMPLETED
                )
                
                # If ingress finished first (user quit), give egress a small window to catch up 
                # on the remaining output in the queue.
                if tasks[0] in done and tasks[1] not in done:
                    try:
                        await asyncio.wait_for(tasks[1], timeout=0.2)
                    except (asyncio.TimeoutError, asyncio.CancelledError):
                        pass
                
                for t in tasks:
                    if t not in done:
                        t.cancel()
                    
                # Final log sync on thread to avoid losing last lines
                if hasattr(self, 'logfile') and hasattr(self, 'mylog'):
                    try:
                        raw_log = self.mylog.getvalue().decode(errors='replace')
                        cleaned_log = await asyncio.to_thread(self._logclean, raw_log, True)
                        with open(self.logfile, "w") as f:
                            f.write(cleaned_log)
                    except Exception:
                        pass

            finally:
                loop.remove_reader(child_fd)
                try:
                    flags = fcntl.fcntl(child_fd, fcntl.F_GETFL)
                    fcntl.fcntl(child_fd, fcntl.F_SETFL, flags & ~os.O_NONBLOCK)
                except Exception:
                    pass
        finally:
            self.current_local_stream = None
            local_stream.teardown()

    @MethodHook
    async def inject_commands(self, commands, child_fd, on_inject=None):
        """
        Inject a list of commands into the node's PTY.
        Handles screen_length_command, history tracking and delays.
        """
        if not commands:
            return

        # 0. Clear line
        os.write(child_fd, b'\x15')
        await asyncio.sleep(0.1)

        # 1. Prepare list (prepend screen_length if exists)
        slc = self.tags.get("screen_length_command") if hasattr(self, 'tags') and isinstance(self.tags, dict) else None
        
        to_send = list(commands)
        if slc and slc not in to_send: # avoid duplicates if already there
             to_send.insert(0, slc)

        # 2. Inject one by one
        for cmd in to_send:
            # Register in node's official history (SKIP if it's the administrative screen length command)
            if cmd != slc and hasattr(self, 'cmd_byte_positions') and self.cmd_byte_positions is not None:
                log_pos = self.mylog.tell() if hasattr(self, 'mylog') else 0
                self.cmd_byte_positions.append((log_pos, cmd))
                if hasattr(self, 'current_local_stream') and self.current_local_stream is not None:
                    try:
                        await self.current_local_stream.write(f'\x1b]133;B;{log_pos}\x07'.encode())
                    except Exception:
                        pass
            
            # Write physically to PTY
            os.write(child_fd, (cmd + "\n").encode())
            
            # Notify (e.g., for gRPC or logs) - SKIP for administrative SLC
            if on_inject and cmd != slc:
                if asyncio.iscoroutinefunction(on_inject):
                    await on_inject(cmd)
                else:
                    on_inject(cmd)
            
            # Delay to avoid overwhelming the router
            await asyncio.sleep(0.8)

    @MethodHook
    def interact(self, debug=False, logger=None):
        '''
        Asynchronous interactive session using Smart Tunnel architecture.
        Allows multiplexing I/O and handling SIGWINCH events locally without blocking.
        '''
        connect = self._connect(debug=debug, logger=logger)
        if connect == True:
            try:
                self._setup_interact_environment(debug=debug, logger=logger, async_mode=True)
                
                local_stream = LocalStream()
                
                def resize_callback(rows, cols):
                    try:
                        self.child.setwinsize(rows, cols)
                    except Exception:
                        pass
                
                # Build local copilot handler
                copilot_handler = self._build_local_copilot_handler()
                
                asyncio.run(self._async_interact_loop(local_stream, resize_callback, copilot_handler=copilot_handler))
            finally:
                self._teardown_interact_environment()
        else:
            if logger:
                logger("error", str(connect))
            else:
                printer.error(f"Connection failed: {str(connect)}")
            sys.exit(1)

    def _build_local_copilot_handler(self):
        """Build copilot handler for local CLI sessions using rich for rendering."""
        config = getattr(self, 'config', None) if hasattr(self, 'config') else None
        return self._copilot_handler(config)

    def _copilot_handler(self, config):
        """Unified copilot handler for local session."""
        import asyncio
        import os

        async def handler(buffer, node_info, stream, child_fd, cmd_byte_positions=None):
            try:
                from .cli.terminal_ui import CopilotInterface
                from .services.ai_service import AIService

                interface = CopilotInterface(
                    config, 
                    history=getattr(stream, 'copilot_history', None),
                    session_state=getattr(stream, 'copilot_state', None)
                )
                # Save history back to stream for persistence in current session
                stream.copilot_history = interface.history
                stream.copilot_state = interface.session_state
                
                ai_service = AIService(config)
                
                async def on_ai_call(active_buffer, question, chunk_callback, merged_node_info):
                    return await ai_service.aask_copilot(
                        active_buffer,
                        question,
                        node_info=merged_node_info,
                        chunk_callback=chunk_callback
                    )
                # Get raw bytes from BytesIO
                raw_bytes = self.mylog.getvalue()
                
                # Stop terminal reading so prompt_toolkit (in run_session) 
                # has exclusive control of stdin without LocalStream interference.
                if hasattr(stream, 'stop_reading'):
                    stream.stop_reading()
                elif hasattr(stream, '_loop') and hasattr(stream, 'stdin_fd'):
                    # Fallback if the method is missing (in LocalStream)
                    stream._loop.remove_reader(stream.stdin_fd)
                
                try:
                    with copilot_terminal_mode():
                        while True:
                            action, commands, custom_cmd = await interface.run_session(
                                raw_bytes=raw_bytes,
                                cmd_byte_positions=self.cmd_byte_positions,
                                node_info=node_info,
                                on_ai_call=on_ai_call
                            )
                            if action == "continue":
                                continue
                            break
                finally:
                    print("\033[2m Returning to session...\033[0m", flush=True)
                    # Restart terminal reading to return to interactive SSH/Telnet mode
                    if hasattr(stream, 'start_reading'):
                        stream.start_reading()
                    elif hasattr(stream, '_loop') and hasattr(stream, 'stdin_fd'):
                        stream._loop.add_reader(stream.stdin_fd, stream._read_ready)
                
                if action in ("send_all", "custom"):
                    cmds_to_send = commands if action == "send_all" else custom_cmd
                    await self.inject_commands(cmds_to_send, child_fd)
                else:
                    os.write(child_fd, b'\x15\r')
            except Exception as e:
                import traceback
                print(f"\n[ERROR in Copilot Handler] {e}", flush=True)
                traceback.print_exc()
                os.write(child_fd, b'\x15\r')

        return handler

    @MethodHook
    def run(self, commands, vars = None,*, folder = '', prompt = r'>$|#$|\$$|>.$|#.$|\$.$', stdout = False, timeout = 10, logger = None):
        '''
        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, logger = logger)
        now = datetime.datetime.now().strftime('%Y-%m-%d_%H%M%S')
        if connect == True:
            if logger:
                port_str = f":{self.port}" if self.port and self.protocol not in ["ssm", "kubectl", "docker"] else ""
                logger("success", f"Connected to {self.unique} at {self.host}{port_str} via: {self.protocol}")

            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:
                    try:
                        c = c.format(**vars)
                    except KeyError as e:
                        self.output = f"Error: Variable {e} not defined in task or inventory"
                        self.status = 1
                        return self.output
                result = self.child.expect(expects, timeout = timeout)
                # Only set terminal size on devices without a
                # screen_length_command (e.g. Linux/bash servers).
                # Routers already disable pagination via that command.
                # After setwinsize, consume any SIGWINCH re-render
                # prompt (~40ms on bash) with a short timeout.
                if c == commands[0] and "screen_length_command" not in self.tags:
                    try:
                        self.child.setwinsize(65535, 65535)
                    except Exception:
                        try:
                            self.child.setwinsize(10000, 10000)
                        except Exception:
                            pass
                    self.child.expect(expects, timeout = 1)
                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 logger:
                logger("output", 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 logger:
                logger("error", f"Connection failed: {connect}")
            if folder != '':
                with open(folder + "/" + self.unique + "_" + now + ".txt", "w") as f:
                    f.write(connect)

                    f.close()
            return connect

    @MethodHook
    def test(self, commands, expected, vars = None,*, folder = '', prompt = r'>$|#$|\$$|>.$|#.$|\$.$', timeout = 10, logger = None):
        '''
        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: 

            - folder (str): Path where output log should be stored, leave 
                            empty to not store logs.

            - 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.

        '''
        now = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
        connect = self._connect(timeout = timeout, logger = logger)
        if connect == True:
            if logger:
                port_str = f":{self.port}" if self.port and self.protocol not in ["ssm", "kubectl", "docker"] else ""
                logger("success", f"Connected to {self.unique} at {self.host}{port_str} via: {self.protocol}")

            if "prompt" in self.tags:
                prompt = self.tags["prompt"]
            expects = [prompt, pexpect.EOF, pexpect.TIMEOUT]
            output = ''
            if not isinstance(commands, list):
                commands = [commands]
            if not isinstance(expected, list):
                expected = [expected]
            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:
                    try:
                        c = c.format(**vars)
                    except KeyError as e:
                        self.output = f"Error: Variable {e} not defined in task or inventory"
                        self.status = 1
                        return self.output
                result = self.child.expect(expects, timeout = timeout)
                if c == commands[0] and "screen_length_command" not in self.tags:
                    try:
                        self.child.setwinsize(65535, 65535)
                    except Exception:
                        try:
                            self.child.setwinsize(10000, 10000)
                        except Exception:
                            pass
                    self.child.expect(expects, timeout = 1)
                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 logger:
                logger("output", output)
            if folder != '':
                with open(folder + "/" + self.unique + "_" + now + ".txt", "w") as f:
                    f.write(output)
                    f.close()
            self.output = output
            if result in [0, 1]:
                # lastcommand = commands[-1]
                # if vars is not None:
                    # lastcommand = lastcommand.format(**vars)
                # last_command_index = output.rfind(lastcommand)
                # cleaned_output = output[last_command_index + len(lastcommand):].strip()
                self.result = {}
                for e in expected:
                    if vars is not None:
                        e = e.format(**vars)
                    updatedprompt = re.sub(r'(?<!\\)\$', '', prompt)
                    cleaned_output = output
                    try:
                        newpattern = f".*({updatedprompt}).*{e}.*"
                        cleaned_output = re.sub(newpattern, '', cleaned_output)
                    except re.error:
                        try:
                            escaped_e = re.escape(e)
                            newpattern = f".*({updatedprompt}).*{escaped_e}.*"
                            cleaned_output = re.sub(newpattern, '', cleaned_output)
                        except re.error:
                            pass

                    if e in cleaned_output:
                        self.result[e] = True
                    else:
                        try:
                            if re.search(e, cleaned_output):
                                self.result[e] = True
                            else:
                                self.result[e] = False
                        except re.error:
                            self.result[e] = False
                self.status = 0
                return self.result
            if result == 2:
                self.result = None
                self.status = 2
                return output
        else:
            self.result = None
            self.output = connect
            self.status = 1
            return connect

    @MethodHook
    def _generate_ssh_sftp_cmd(self):
        cmd = self.protocol
        if self.port:
            if self.protocol == "ssh":
                cmd += " -p " + self.port
            elif self.protocol == "sftp":
                cmd += " -P " + self.port
        if self.options:
            opts = self.options
            if self.protocol == "sftp":
                # Strip SSH-only flags that sftp doesn't support
                opts = re.sub(r'(?<!\S)-[XxtTAaNf]\b', '', opts).strip()
            if opts:
                cmd += " " + opts
        if self.jumphost:
            cmd += " " + self.jumphost
        user_host = f"{self.user}@{self.host}" if self.user else self.host
        cmd += f" {user_host}"
        return cmd

    @MethodHook
    def _generate_telnet_cmd(self):
        cmd = f"telnet {self.host}"
        if self.port:
            cmd += f" {self.port}"
        if self.options:
            cmd += f" {self.options}"
        return cmd

    @MethodHook
    def _generate_kube_cmd(self):
        cmd = f"kubectl exec {self.options} {self.host} -it --"
        kube_command = self.tags.get("kube_command", "/bin/bash") if isinstance(self.tags, dict) else "/bin/bash"
        cmd += f" {kube_command}"
        return cmd

    @MethodHook
    def _generate_docker_cmd(self):
        cmd = f"docker {self.options} exec -it {self.host}"
        docker_command = self.tags.get("docker_command", "/bin/bash") if isinstance(self.tags, dict) else "/bin/bash"
        cmd += f" {docker_command}"
        return cmd

    @MethodHook
    def _generate_ssm_cmd(self):
        region = self.tags.get("region", "") if isinstance(self.tags, dict) else ""
        profile = self.tags.get("profile", "") if isinstance(self.tags, dict) else ""
        cmd = f"aws ssm start-session --target {self.host}"
        if region:
            cmd += f" --region {region}"
        if profile:
            cmd += f" --profile {profile}"
        if self.options:
            cmd += f" {self.options}"
        return cmd


    @MethodHook
    def _get_cmd(self):
        if self.protocol in ["ssh", "sftp"]:
            return self._generate_ssh_sftp_cmd()
        elif self.protocol == "telnet":
            return self._generate_telnet_cmd()
        elif self.protocol == "kubectl":
            return self._generate_kube_cmd()
        elif self.protocol == "docker":
            return self._generate_docker_cmd()
        elif self.protocol == "ssm":
            return self._generate_ssm_cmd()
        elif self.protocol == "local":
            return self.host
        else:
            printer.error(f"Invalid protocol: {self.protocol}")
            sys.exit(1)

    @MethodHook
    def _connect(self, debug=False, timeout=10, max_attempts=3, logger=None):
        if self.protocol == "local":
            cmd = self._get_cmd()
            args = shlex.split(cmd)
            self.child = pexpect.spawn(args[0], args[1:], env=os.environ.copy())
            from pexpect import fdpexpect
            self.raw_child = fdpexpect.fdspawn(self.child.child_fd)
            if self.logs != '':
                self.logfile = self._logfile()
            return True

        cmd = self._get_cmd()
        passwords = self._passtx(self.password) if self.password and any(self.password) else []
        if self.logs != '':
            self.logfile = self._logfile()
        default_prompt = r'>$|#$|\$$|>.$|#.$|\$.$'
        prompt = self.tags.get("prompt", default_prompt) if isinstance(self.tags, dict) else default_prompt
        password_prompt = '[p|P]assword:|[u|U]sername:' if self.protocol != 'telnet' else '[p|P]assword:'

        expects = {
            "ssh": ['yes/no', 'refused', 'supported', 'Invalid|[u|U]sage: ssh', 'ssh-keygen.*\"', 'timeout|timed.out', 'unavailable', 'closed', password_prompt, prompt, 'suspend', pexpect.EOF, pexpect.TIMEOUT, "No route to host", "resolve hostname", "no matching", "[b|B]ad (owner|permissions)"],
            "sftp": ['yes/no', 'refused', 'supported', 'Invalid|[u|U]sage: sftp', 'ssh-keygen.*\"', 'timeout|timed.out', 'unavailable', 'closed', password_prompt, prompt, 'suspend', pexpect.EOF, pexpect.TIMEOUT, "No route to host", "resolve hostname", "no matching", "[b|B]ad (owner|permissions)"],
            "telnet": ['[u|U]sername:', 'refused', 'supported', 'invalid|unrecognized option', 'ssh-keygen.*\"', 'timeout|timed.out', 'unavailable', 'closed', password_prompt, prompt, 'suspend', pexpect.EOF, pexpect.TIMEOUT, "No route to host", "resolve hostname", "no matching", "[b|B]ad (owner|permissions)"],
            "kubectl": ['[u|U]sername:', '[r|R]efused', '[E|e]rror', 'DEPRECATED', pexpect.TIMEOUT, password_prompt, prompt, pexpect.EOF, "expired|invalid"],
            "docker": ['[u|U]sername:', 'Cannot', '[E|e]rror', 'failed', 'not a docker command', 'unknown', 'unable to resolve', pexpect.TIMEOUT, password_prompt, prompt, pexpect.EOF],
            "ssm": ['[u|U]sername:', 'Cannot', '[E|e]rror', 'failed', 'SessionManagerPlugin', '[u|U]nknown', 'unable to resolve', pexpect.TIMEOUT, password_prompt, prompt, pexpect.EOF]
        }

        error_indices = {
            "ssh": [1, 2, 3, 4, 5, 6, 7, 12, 13, 14, 15, 16],
            "sftp": [1, 2, 3, 4, 5, 6, 7, 12, 13, 14, 15, 16],
            "telnet": [1, 2, 3, 4, 5, 6, 7, 12, 13, 14, 15, 16],
            "kubectl": [1, 2, 3, 4, 8],  # Define error indices for kube
            "docker": [1, 2, 3, 4, 5, 6, 7],  # Define error indices for docker
            "ssm": [1, 2, 3, 4, 5, 6, 7]
        }

        eof_indices = {
            "ssh": [8, 9, 10, 11],
            "sftp": [8, 9, 10, 11],
            "telnet": [8, 9, 10, 11],
            "kubectl": [5, 6, 7],  # Define eof indices for kube
            "docker": [8, 9, 10],  # Define eof indices for docker
            "ssm": [8, 9, 10]
        }

        initial_indices = {
            "ssh": [0],
            "sftp": [0],
            "telnet": [0],
            "kubectl": [0],  # Define special indices for kube
            "docker": [0],  # Define special indices for docker
            "ssm": [0]
        }

        attempts = 1
        while attempts <= max_attempts:
            args = shlex.split(cmd)
            child = pexpect.spawn(args[0], args[1:])
            if isinstance(self.tags, dict) and self.tags.get("console"):
                child.sendline()
            if debug:
                if logger:
                    logger("debug", f"Command:\n{cmd}")
                self.mylog = io.BytesIO()
                self.mylog.write(f"[i] [DEBUG] Command:\r\n    {cmd}\r\n".encode())
                child.logfile_read = self.mylog


            endloop = False
            for i in range(len(passwords) if passwords else 1):
                while True:
                    results = child.expect(expects[self.protocol], timeout=timeout)
                    results_value = expects[self.protocol][results]
                    
                    if results in initial_indices[self.protocol]:
                        if self.protocol in ["ssh", "sftp"]:
                            child.sendline('yes')
                        elif self.protocol in ["telnet", "kubectl", "docker", "ssm"]:
                            if self.user:
                                child.sendline(self.user)
                            else:
                                self.missingtext = True
                                break
                    
                    elif results in error_indices[self.protocol]:
                        child.terminate()
                        if results_value == pexpect.TIMEOUT and attempts != max_attempts:
                            attempts += 1
                            endloop = True
                            break
                        else:
                            after = "Connection timeout" if results_value == pexpect.TIMEOUT else child.after.decode()
                            return f"Connection failed code: {results}\n{child.before.decode().lstrip()}{after}{child.readline().decode()}".rstrip()
                    
                    elif results in eof_indices[self.protocol]:
                        if results_value == password_prompt:
                            if passwords:
                                child.sendline(passwords[i])
                            else:
                                self.missingtext = True
                            break
                        elif results_value == "suspend":
                            child.sendline("\r")
                            sleep(2)
                        else:
                            endloop = True
                            child.sendline()
                            break
                    
                if endloop:
                    break
            if results_value == pexpect.TIMEOUT:
                continue
            else:
                break

        if isinstance(self.tags, dict) and self.tags.get("post_connect_commands"):
            cmds = self.tags.get("post_connect_commands")
            commands = [cmds] if isinstance(cmds, str) else cmds
            for command in commands:
                child.sendline(command)
                sleep(1)
        child.readline(0)
        self.child = child
        from pexpect import fdpexpect
        self.raw_child = fdpexpect.fdspawn(self.child.child_fd)
        return True

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 successfully.
                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, telnet, kubectl or docker. 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"

- jumphost (str): Reference another node to be used as a jumphost

Methods

async def inject_commands(self, commands, child_fd, on_inject=None)

Expand source code

@MethodHook
async def inject_commands(self, commands, child_fd, on_inject=None):
    """
    Inject a list of commands into the node's PTY.
    Handles screen_length_command, history tracking and delays.
    """
    if not commands:
        return

    # 0. Clear line
    os.write(child_fd, b'\x15')
    await asyncio.sleep(0.1)

    # 1. Prepare list (prepend screen_length if exists)
    slc = self.tags.get("screen_length_command") if hasattr(self, 'tags') and isinstance(self.tags, dict) else None
    
    to_send = list(commands)
    if slc and slc not in to_send: # avoid duplicates if already there
         to_send.insert(0, slc)

    # 2. Inject one by one
    for cmd in to_send:
        # Register in node's official history (SKIP if it's the administrative screen length command)
        if cmd != slc and hasattr(self, 'cmd_byte_positions') and self.cmd_byte_positions is not None:
            log_pos = self.mylog.tell() if hasattr(self, 'mylog') else 0
            self.cmd_byte_positions.append((log_pos, cmd))
            if hasattr(self, 'current_local_stream') and self.current_local_stream is not None:
                try:
                    await self.current_local_stream.write(f'\x1b]133;B;{log_pos}\x07'.encode())
                except Exception:
                    pass
        
        # Write physically to PTY
        os.write(child_fd, (cmd + "\n").encode())
        
        # Notify (e.g., for gRPC or logs) - SKIP for administrative SLC
        if on_inject and cmd != slc:
            if asyncio.iscoroutinefunction(on_inject):
                await on_inject(cmd)
            else:
                on_inject(cmd)
        
        # Delay to avoid overwhelming the router
        await asyncio.sleep(0.8)

Inject a list of commands into the node's PTY. Handles screen_length_command, history tracking and delays.

def interact(self, debug=False, logger=None)

Expand source code

@MethodHook
def interact(self, debug=False, logger=None):
    '''
    Asynchronous interactive session using Smart Tunnel architecture.
    Allows multiplexing I/O and handling SIGWINCH events locally without blocking.
    '''
    connect = self._connect(debug=debug, logger=logger)
    if connect == True:
        try:
            self._setup_interact_environment(debug=debug, logger=logger, async_mode=True)
            
            local_stream = LocalStream()
            
            def resize_callback(rows, cols):
                try:
                    self.child.setwinsize(rows, cols)
                except Exception:
                    pass
            
            # Build local copilot handler
            copilot_handler = self._build_local_copilot_handler()
            
            asyncio.run(self._async_interact_loop(local_stream, resize_callback, copilot_handler=copilot_handler))
        finally:
            self._teardown_interact_environment()
    else:
        if logger:
            logger("error", str(connect))
        else:
            printer.error(f"Connection failed: {str(connect)}")
        sys.exit(1)

Asynchronous interactive session using Smart Tunnel architecture. Allows multiplexing I/O and handling SIGWINCH events locally without blocking.

def run(self,
commands,
vars=None,
*,
folder='',
prompt='>$|#$|\\$$|>.$|#.$|\\$.$',
stdout=False,
timeout=10,
logger=None)

Expand source code

@MethodHook
def run(self, commands, vars = None,*, folder = '', prompt = r'>$|#$|\$$|>.$|#.$|\$.$', stdout = False, timeout = 10, logger = None):
    '''
    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, logger = logger)
    now = datetime.datetime.now().strftime('%Y-%m-%d_%H%M%S')
    if connect == True:
        if logger:
            port_str = f":{self.port}" if self.port and self.protocol not in ["ssm", "kubectl", "docker"] else ""
            logger("success", f"Connected to {self.unique} at {self.host}{port_str} via: {self.protocol}")

        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:
                try:
                    c = c.format(**vars)
                except KeyError as e:
                    self.output = f"Error: Variable {e} not defined in task or inventory"
                    self.status = 1
                    return self.output
            result = self.child.expect(expects, timeout = timeout)
            # Only set terminal size on devices without a
            # screen_length_command (e.g. Linux/bash servers).
            # Routers already disable pagination via that command.
            # After setwinsize, consume any SIGWINCH re-render
            # prompt (~40ms on bash) with a short timeout.
            if c == commands[0] and "screen_length_command" not in self.tags:
                try:
                    self.child.setwinsize(65535, 65535)
                except Exception:
                    try:
                        self.child.setwinsize(10000, 10000)
                    except Exception:
                        pass
                self.child.expect(expects, timeout = 1)
            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 logger:
            logger("output", 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 logger:
            logger("error", f"Connection failed: {connect}")
        if folder != '':
            with open(folder + "/" + self.unique + "_" + now + ".txt", "w") as f:
                f.write(connect)

                f.close()
        return connect

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.

def test(self,
commands,
expected,
vars=None,
*,
folder='',
prompt='>$|#$|\\$$|>.$|#.$|\\$.$',
timeout=10,
logger=None)

Expand source code

@MethodHook
def test(self, commands, expected, vars = None,*, folder = '', prompt = r'>$|#$|\$$|>.$|#.$|\$.$', timeout = 10, logger = None):
    '''
    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: 

        - folder (str): Path where output log should be stored, leave 
                        empty to not store logs.

        - 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.

    '''
    now = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
    connect = self._connect(timeout = timeout, logger = logger)
    if connect == True:
        if logger:
            port_str = f":{self.port}" if self.port and self.protocol not in ["ssm", "kubectl", "docker"] else ""
            logger("success", f"Connected to {self.unique} at {self.host}{port_str} via: {self.protocol}")

        if "prompt" in self.tags:
            prompt = self.tags["prompt"]
        expects = [prompt, pexpect.EOF, pexpect.TIMEOUT]
        output = ''
        if not isinstance(commands, list):
            commands = [commands]
        if not isinstance(expected, list):
            expected = [expected]
        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:
                try:
                    c = c.format(**vars)
                except KeyError as e:
                    self.output = f"Error: Variable {e} not defined in task or inventory"
                    self.status = 1
                    return self.output
            result = self.child.expect(expects, timeout = timeout)
            if c == commands[0] and "screen_length_command" not in self.tags:
                try:
                    self.child.setwinsize(65535, 65535)
                except Exception:
                    try:
                        self.child.setwinsize(10000, 10000)
                    except Exception:
                        pass
                self.child.expect(expects, timeout = 1)
            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 logger:
            logger("output", output)
        if folder != '':
            with open(folder + "/" + self.unique + "_" + now + ".txt", "w") as f:
                f.write(output)
                f.close()
        self.output = output
        if result in [0, 1]:
            # lastcommand = commands[-1]
            # if vars is not None:
                # lastcommand = lastcommand.format(**vars)
            # last_command_index = output.rfind(lastcommand)
            # cleaned_output = output[last_command_index + len(lastcommand):].strip()
            self.result = {}
            for e in expected:
                if vars is not None:
                    e = e.format(**vars)
                updatedprompt = re.sub(r'(?<!\\)\$', '', prompt)
                cleaned_output = output
                try:
                    newpattern = f".*({updatedprompt}).*{e}.*"
                    cleaned_output = re.sub(newpattern, '', cleaned_output)
                except re.error:
                    try:
                        escaped_e = re.escape(e)
                        newpattern = f".*({updatedprompt}).*{escaped_e}.*"
                        cleaned_output = re.sub(newpattern, '', cleaned_output)
                    except re.error:
                        pass

                if e in cleaned_output:
                    self.result[e] = True
                else:
                    try:
                        if re.search(e, cleaned_output):
                            self.result[e] = True
                        else:
                            self.result[e] = False
                    except re.error:
                        self.result[e] = False
            self.status = 0
            return self.result
        if result == 2:
            self.result = None
            self.status = 2
            return output
    else:
        self.result = None
        self.output = connect
        self.status = 1
        return connect

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:

- folder (str): Path where output log should be stored, leave 
                empty to not store logs.

- 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.

class nodes (nodes: dict, config='')

Expand source code

@ClassHook
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 successfully.
                           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)

    
    @MethodHook
    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]


    @MethodHook
    def run(self, commands, vars = None,*, folder = None, prompt = None, stdout = None, parallel = 10, timeout = None, on_complete = None, logger = 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.

            - on_complete (callable): Optional callback called when each node 
                                      finishes. Receives (unique, output, status).
                                      Called from the node's thread so it must
                                      be thread-safe.

        ###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 and on_complete is None:
            args["stdout"] = stdout
        if timeout != None:
            args["timeout"] = timeout
        output = {}
        status = {}
        tasks = []

        def _run_node(node_obj, node_args, callback):
            """Wrapper that runs a node and fires the callback on completion."""
            node_obj.run(**node_args)
            if callback:
                callback(node_obj.unique, node_obj.output, node_obj.status)

        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__"])
                for var_key, var_val in vars.items():
                    if var_key == "__global__":
                        continue
                    try:
                        if re.search(var_key, n.unique, re.IGNORECASE):
                            nodesargs[n.unique]["vars"].update(var_val)
                    except re.error:
                        if var_key == n.unique:
                            nodesargs[n.unique]["vars"].update(var_val)
            
            # Pass the logger to the node
            nodesargs[n.unique]["logger"] = logger

            if on_complete:
                tasks.append(threading.Thread(target=_run_node, args=(n, nodesargs[n.unique], on_complete)))
            else:
                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

    @MethodHook
    def test(self, commands, expected, vars = None,*, folder = None, prompt = None, parallel = 10, timeout = None, on_complete = None, logger = 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.

            - on_complete (callable): Optional callback called when each node 
                                      finishes. Receives (unique, output, status).
                                      Called from the node's thread so it must
                                      be thread-safe.

        ### 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 folder != None:
            args["folder"] = folder
            Path(folder).mkdir(parents=True, exist_ok=True)
        if prompt != None:
            args["prompt"] = prompt
        if timeout != None:
            args["timeout"] = timeout
        output = {}
        result = {}
        status = {}
        tasks = []

        def _test_node(node_obj, node_args, callback):
            """Wrapper that runs a node test and fires the callback on completion."""
            node_obj.test(**node_args)
            if callback:
                callback(node_obj.unique, node_obj.output, node_obj.status, node_obj.result)

        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__"])
                for var_key, var_val in vars.items():
                    if var_key == "__global__":
                        continue
                    try:
                        if re.search(var_key, n.unique, re.IGNORECASE):
                            nodesargs[n.unique]["vars"].update(var_val)
                    except re.error:
                        if var_key == n.unique:
                            nodesargs[n.unique]["vars"].update(var_val)
            nodesargs[n.unique]["logger"] = logger
            
            if on_complete:
                tasks.append(threading.Thread(target=_test_node, args=(n, nodesargs[n.unique], on_complete)))
            else:
                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

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 successfully.
                   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.

Methods

def run(self,
commands,
vars=None,
*,
folder=None,
prompt=None,
stdout=None,
parallel=10,
timeout=None,
on_complete=None,
logger=None)

Expand source code

@MethodHook
def run(self, commands, vars = None,*, folder = None, prompt = None, stdout = None, parallel = 10, timeout = None, on_complete = None, logger = 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.

        - on_complete (callable): Optional callback called when each node 
                                  finishes. Receives (unique, output, status).
                                  Called from the node's thread so it must
                                  be thread-safe.

    ###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 and on_complete is None:
        args["stdout"] = stdout
    if timeout != None:
        args["timeout"] = timeout
    output = {}
    status = {}
    tasks = []

    def _run_node(node_obj, node_args, callback):
        """Wrapper that runs a node and fires the callback on completion."""
        node_obj.run(**node_args)
        if callback:
            callback(node_obj.unique, node_obj.output, node_obj.status)

    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__"])
            for var_key, var_val in vars.items():
                if var_key == "__global__":
                    continue
                try:
                    if re.search(var_key, n.unique, re.IGNORECASE):
                        nodesargs[n.unique]["vars"].update(var_val)
                except re.error:
                    if var_key == n.unique:
                        nodesargs[n.unique]["vars"].update(var_val)
        
        # Pass the logger to the node
        nodesargs[n.unique]["logger"] = logger

        if on_complete:
            tasks.append(threading.Thread(target=_run_node, args=(n, nodesargs[n.unique], on_complete)))
        else:
            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

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.

- on_complete (callable): Optional callback called when each node 
                          finishes. Receives (unique, output, status).
                          Called from the node's thread so it must
                          be thread-safe.

Returns:

dict: Dictionary formed by nodes unique as keys, Output of the 
      commands you ran on the node as value.

def test(self,
commands,
expected,
vars=None,
*,
folder=None,
prompt=None,
parallel=10,
timeout=None,
on_complete=None,
logger=None)

Expand source code

@MethodHook
def test(self, commands, expected, vars = None,*, folder = None, prompt = None, parallel = 10, timeout = None, on_complete = None, logger = 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.

        - on_complete (callable): Optional callback called when each node 
                                  finishes. Receives (unique, output, status).
                                  Called from the node's thread so it must
                                  be thread-safe.

    ### 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 folder != None:
        args["folder"] = folder
        Path(folder).mkdir(parents=True, exist_ok=True)
    if prompt != None:
        args["prompt"] = prompt
    if timeout != None:
        args["timeout"] = timeout
    output = {}
    result = {}
    status = {}
    tasks = []

    def _test_node(node_obj, node_args, callback):
        """Wrapper that runs a node test and fires the callback on completion."""
        node_obj.test(**node_args)
        if callback:
            callback(node_obj.unique, node_obj.output, node_obj.status, node_obj.result)

    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__"])
            for var_key, var_val in vars.items():
                if var_key == "__global__":
                    continue
                try:
                    if re.search(var_key, n.unique, re.IGNORECASE):
                        nodesargs[n.unique]["vars"].update(var_val)
                except re.error:
                    if var_key == n.unique:
                        nodesargs[n.unique]["vars"].update(var_val)
        nodesargs[n.unique]["logger"] = logger
        
        if on_complete:
            tasks.append(threading.Thread(target=_test_node, args=(n, nodesargs[n.unique], on_complete)))
        else:
            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

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.

- on_complete (callable): Optional callback called when each node 
                          finishes. Receives (unique, output, status).
                          Called from the node's thread so it must
                          be thread-safe.

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.