From e94bb3a341b1643f9dbc2484d9d8645196e6a190 Mon Sep 17 00:00:00 2001
From: Fede Luzzi
Connpy
+Connpy (v6.0.3)
+
+
+
+
+
+
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 the AI Copilot, an interactive terminal assistant that understands your network context and helps you manage your infrastructure more intelligently.
-The AI Copilot is deeply integrated into your terminal workflow:
-- Terminal Context Awareness: The Copilot can "see" your screen output, helping you diagnose errors or analyze command results in real-time.
-- Dynamic Context Selection: Flexibly select single, range, or line-based terminal blocks to feed the Copilot, filtering out interactive scrolling garbage automatically (e.g., Cisco IOS/XR scrolling, paginators).
-- Hybrid Multi-Agent System: Automatically escalates complex tasks between the Network Engineer (execution) and the Network Architect (strategy).
-- MCP Integration: Dynamically load tools from external providers (6WIND, AWS, etc.) via the Model Context Protocol.
-- Flexible Auth & Keyless AI: Support for advanced LiteLLM credentials (--engineer-auth / --architect-auth) allowing keyless local models (Ollama), cloud engines (Vertex AI), or custom endpoints.
-- Enhanced Session Management: Uniquely generated sessions, robust pagination, and interactive styling translating prompt themes directly to terminal escapes.
-- Semantic Prompt Integration: Emit standard OSC prompt sequences (]133;B) for real-time remote/web front-end command tracking.
-- Interactive Chat: Launch with conn ai for a collaborative troubleshooting session.
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.
+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.
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?"
+
+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
+
+Run commands in parallel directly on target nodes or folder structures:
+conn run router1 "show interface"
+
+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
+
+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
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).
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.
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}"
+
+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).
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" }
+
+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.
Support for single or chained intermediate gateway nodes (SSH, SSM, kubectl, or docker jumphosts) to tunnel traffic safely into target environments.
+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'}.
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"
+
@folder) and subfolders (@subfolder@folder).@profilename) to manage shared credentials easily.tshark is required for live piping (-w).--ns <name>), custom filters (-f <filter>), or configure the Wireshark local path (--set-wireshark-path).prompt-toolkit, including:fzf.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
+
+conn context --ls), show context details (conn context production -s), or delete contexts (conn context production -r).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
+
+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).
Shift the local CLI to communicate with a remote server instance:
+conn config --service-mode remote
+conn config --remote localhost:50051
+
+Manage server-side user credentials for distributed setups:
+conn user --add username
+conn user --list
+conn user --regen-password username
+
+Use --path to specify custom configuration folders in server Mode B.
Configure identity providers (e.g. Authelia, Keycloak) for SSO gRPC authentication using the interactive wizard:
+conn sso --add provider_name
+
+Authenticate client sessions (conn login [username]), check connection status (conn login --status), or close sessions (conn logout).
pip install connpy
-git clone https://github.com/fluzzi/connpy
-cd connpy
-docker compose build
-
-# Run it like a native app (completely silent)
-docker compose run --rm --remove-orphans connpy-app [command]
-
-# Pro Tip: Add this alias for a 100% native experience from any folder
-alias conn='docker compose -f /path/to/connpy/docker-compose.yml run --rm --remove-orphans connpy-app'
+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
-🔒 Privacy & Integration
-Privacy Policy
-Connpy is committed to protecting your privacy:
-- Local Storage: All server addresses, usernames, and passwords are encrypted and stored only on your machine. No data is transmitted to our servers.
-- Data Access: Data is used solely for managing and automating your connections.
-Google Integration
-Used strictly for backup:
-- Backup: Sync your encrypted configuration with your Google Drive account.
-- Scoped Access: Connpy only accesses its own backup files.
+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.
-Usage
-usage: conn [-h] [--add | --del | --mod | --show | --debug] [node|folder] [--sftp]
- conn {profile,move,copy,list,bulk,export,import,ai,run,api,plugin,config,sync,context} ...
-
-Basic Examples:
-# Add a folder and subfolder
-conn --add @office
-conn --add @datacenter@office
-
-# Add a node with a profile
-conn --add server1@datacenter@office --profile @myuser
-
-# Connect to a node (fuzzy match)
-conn server1
-
-# Start the AI Copilot
-conn ai
-
-# Run a command on all nodes in a folder
-conn run @office "uptime"
-
-🔑 SSO / OIDC Provider Management
-In remote mode, connpy supports Single Sign-On (SSO) login. You can manage the configured identity providers (IdPs) directly from the local CLI using the conn sso command suite:
-
-- List configured providers:
-
bash
-conn sso --list
-- Show provider details (sensitive credentials like secrets are masked):
-
bash
-conn sso --show <provider_name>
-- Add or update a provider (opens an interactive configuration wizard):
-
bash
-conn sso --add <provider_name>
-- Delete a provider:
-
bash
-conn sso --del <provider_name>
-
-Security Recommendation (Secret Reference Env Vars)
-To keep sensitive client secrets or shared secrets out of git-tracked configuration files, you can input a variable name prefixed with a $ instead of the literal secret during the conn sso --add prompts (e.g., $CONN_SSO_MYPROVIDER_SECRET). The backend gRPC server will dynamically resolve the value from its environment variables at runtime.
-
-Plugin Requirements for Connpy
-Remote Plugin Execution
-When Connpy operates in remote mode, plugins are executed transparently on the server:
-- The client automatically downloads the plugin source code (Parser class context) to generate the local argparse structure and provide autocompletion.
-- The execution phase (Entrypoint class) is redirected via gRPC streams to execute in the server's memory.
-- You can manage remote plugins using the --remote flag.
-General Structure
-
-- The plugin script must define specific classes:
-- Class
Parser: Handles argparse.ArgumentParser initialization.
-- Class
Entrypoint: Main execution logic (receives args, parser, and connapp).
-- Class
Preload: (Optional) For modifying core app behavior or registering hooks.
-
-Preload Modifications and Hooks
-You can customize the behavior of core classes using hooks:
-- modify(method): Alter class instances (e.g., connapp.config, connapp.ai).
-- register_pre_hook(method): Logic to run before a method execution.
-- register_post_hook(method): Logic to run after a method execution.
-Command Completion Support
-Plugins can provide intelligent tab completion:
-1. Tree-based Completion (Recommended): Define _connpy_tree(info) returning a navigation dictionary.
-2. Legacy Completion: Define _connpy_completion(wordsnumber, words, info).
-
-⚙️ gRPC Service Architecture
-Connpy can operate in a decoupled mode:
-1. Start the API (Server): conn api -s 50051
-2. Configure the Client:
-bash
-conn config --service-mode remote
-conn config --remote-host localhost:50051
-All inventory management and execution will now happen on the server.
-
-🐍 Automation Module (API)
-You can use connpy as a Python library for your own scripts.
-Basic Execution
+11. 🐍 Python API
+Embed connection and automation routines programmatically in Python:
import connpy
-router = connpy.node("uniqueName", "1.1.1.1", user="admin")
+
+# 1. Direct single node interaction
+router = connpy.node("router1", "1.1.1.1", user="admin")
router.run(["show ip int brief"])
print(router.output)
-
-Parallel Tasks with Variables
-import connpy
-config = connpy.configfile()
-nodes = config.getitem("@office", ["router1", "router2"])
-routers = connpy.nodes(nodes, config=config)
+# 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)
-
-AI Programmatic Use
-import connpy
+
+# 3. AI Copilot prompts
myai = connpy.ai(connpy.configfile())
-response = myai.ask("What is the status of the BGP neighbors in the office?")
+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.
-For detailed developer notes and plugin hooks documentation, see the Documentation.
-📜 License
+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
@@ -4241,80 +4268,12 @@ class node:
else:
self.password = [password]
if self.jumphost != "" and config != '':
- self.jumphost = config.getitem(self.jumphost)
- for key in self.jumphost:
- profile = re.search("^@(.*)", str(self.jumphost[key]))
- if profile:
- try:
- self.jumphost[key] = config.profiles[profile.group(1)][key]
- except KeyError:
- self.jumphost[key] = ""
- elif self.jumphost[key] == '' and key == "protocol":
- try:
- self.jumphost[key] = config.profiles["default"][key]
- except KeyError:
- self.jumphost[key] = "ssh"
- if isinstance(self.jumphost["password"],list):
- jumphost_password = []
- for i, s in enumerate(self.jumphost["password"]):
- profile = re.search("^@(.*)", self.jumphost["password"][i])
- if profile:
- jumphost_password.append(config.profiles[profile.group(1)]["password"])
- else:
- jumphost_password.append(self.jumphost["password"][i])
- self.jumphost["password"] = jumphost_password
- else:
- self.jumphost["password"] = [self.jumphost["password"]]
- if self.jumphost["password"] != [""]:
- self.password = self.jumphost["password"] + self.password
-
- if self.jumphost["protocol"] == "ssh":
- jumphost_cmd = self.jumphost["protocol"] + " -W %h:%p"
- if self.jumphost["port"] != '':
- jumphost_cmd = jumphost_cmd + " -p " + self.jumphost["port"]
- if self.jumphost["options"] != '':
- jumphost_cmd = jumphost_cmd + " " + self.jumphost["options"]
- if self.jumphost["user"] == '':
- jumphost_cmd = jumphost_cmd + " {}".format(self.jumphost["host"])
- else:
- jumphost_cmd = jumphost_cmd + " {}".format("@".join([self.jumphost["user"],self.jumphost["host"]]))
- self.jumphost = f"-o ProxyCommand=\"{jumphost_cmd}\""
- elif self.jumphost["protocol"] == "ssm":
- ssm_target = self.jumphost["host"]
- ssm_cmd = f"aws ssm start-session --target {ssm_target} --document-name AWS-StartSSHSession --parameters 'portNumber=22'"
- if isinstance(self.jumphost.get("tags"), dict):
- if "profile" in self.jumphost["tags"]:
- ssm_cmd += f" --profile {self.jumphost['tags']['profile']}"
- if "region" in self.jumphost["tags"]:
- ssm_cmd += f" --region {self.jumphost['tags']['region']}"
- if self.jumphost["options"] != '':
- ssm_cmd += f" {self.jumphost['options']}"
-
- bastion_user_part = f"{self.jumphost['user']}@{ssm_target}" if self.jumphost['user'] else ssm_target
-
- ssh_opts = ""
- if isinstance(self.jumphost.get("tags"), dict) and "ssh_options" in self.jumphost["tags"]:
- ssh_opts = f" {self.jumphost['tags']['ssh_options']}"
-
- inner_ssh = f"ssh{ssh_opts} -o ProxyCommand='{ssm_cmd}' -W %h:%p {bastion_user_part}"
- self.jumphost = f"-o ProxyCommand=\"{inner_ssh}\""
- elif self.jumphost["protocol"] in ["kubectl", "docker"]:
- nc_cmd = "nc"
- if isinstance(self.jumphost.get("tags"), dict) and "nc_command" in self.jumphost["tags"]:
- nc_cmd = self.jumphost["tags"]["nc_command"]
-
- if self.jumphost["protocol"] == "kubectl":
- proxy_cmd = f"kubectl exec "
- if self.jumphost["options"] != '':
- proxy_cmd += f"{self.jumphost['options']} "
- proxy_cmd += f"{self.jumphost['host']} -i -- {nc_cmd} %h %p"
- else:
- proxy_cmd = f"docker "
- if self.jumphost["options"] != '':
- proxy_cmd += f"{self.jumphost['options']} "
- proxy_cmd += f"exec -i {self.jumphost['host']} {nc_cmd} %h %p"
-
- self.jumphost = f"-o ProxyCommand=\"{proxy_cmd}\""
+ 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 = ""
@@ -4323,6 +4282,127 @@ class node:
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.
@@ -5214,7 +5294,8 @@ class node:
attempts = 1
while attempts <= max_attempts:
- child = pexpect.spawn(cmd)
+ 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:
@@ -6437,41 +6518,55 @@ def test(self, commands, expected, vars = None,*, folder = None, prompt = None,