NetPulse SSH

High-performance, standalone Python library & CLI for concurrent SSH command execution across network devices.

netpulse-ssh is an ultra-fast, resilient SSH execution engine extracted from the NetPulse discovery suite. It allows network engineers and automation pipelines to execute commands across hundreds of network devices simultaneously, without blocking I/O or crashing due to a single offline host.

✨ Features

• Massive Concurrency: Scales horizontally using asyncio to execute commands across hundreds of targets simultaneously, dramatically reducing execution time.
• Bastion Host Routing (ProxyJump): Natively route concurrent execution traffic through an enterprise DMZ Bastion Host. Multiplexes hundreds of internal connections over a single encrypted tunnel to conserve Bastion CPU and port resources.
• Concurrent File Management: Natively utilizes SCP/SFTP protocols to push large firmware payloads or pull configuration backups from 100+ devices simultaneously into IP-segregated folders.
• Pure Python PTY Emulator: Ships with a fully self-contained, cross-platform Pseudo-Terminal (PTY) emulator. Features native raw keystroke capturing (including Arrow keys and Tab-completion) on Windows, entirely independent of the underlying OS SSH binaries.
• Legacy Equipment Healing: Automatically detects strict OpenSSH cipher drops and seamlessly falls back to legacy algorithms (e.g., diffie-hellman-group1-sha1, 3des-cbc) frequently required for older Cisco or Juniper gear.
• Smart Privilege Escalation: Natively supports Cisco enable mode privilege escalation without breaking automation flows.
• Pagination Suppression: Automatically injects terminal length 0 before command execution to bypass interactive --More-- prompts.
• Automated Deployment Engine (Playbooks): Decouple scripts from targets using YAML Inventory and Playbook definitions. Execute sequential, multi-step configurations effortlessly.
• Interactive "Expect" Engine: Automate interactive prompts (e.g. Do you want to continue? [Y/n]) by programming standard output regex triggers directly in your YAML Playbooks.
• REST API Enabled: Run the built-in FastAPI uvicorn wrapper to serve concurrent execution logic dynamically to web dashboards or automation scripts.

🚀 Quickstart

Installation

Install globally or locally via pip:

pip install netpulse-ssh

CLI Usage

Automated Deployment (Playbooks) Execute a sequential YAML playbook across a grouped YAML inventory:

netpulse-ssh deploy --inventory hosts.yaml --script deploy.yaml

Concurrent Execution Execute a single command across multiple devices concurrently:

netpulse-ssh execute 192.168.1.5 10.0.0.1 -c "show ip interface brief" -u admin -p password

Concurrent File Management (SCP) Push firmware payloads or pull configs across your entire network simultaneously:

# Mass Push Firmware
netpulse-ssh scp push 192.168.1.5 10.0.0.1 --src ./firmware.bin --dest /flash/ -u admin -p password

# Mass Pull Configurations (Auto-segregated into ./backups/<IP>/)
netpulse-ssh scp pull 192.168.1.5 10.0.0.1 --src /etc/nginx/nginx.conf --dest ./backups/ -u admin -p password

Secure Bastion Routing (ProxyJump) Execute commands on internal servers by multiplexing through a single public-facing Bastion host:

netpulse-ssh execute 10.0.0.5 10.0.0.6 -c "uptime" -u admin -p secret -J proxy@bastion.corp.com --bastion-pass proxy_secret

Interactive Shell Drop directly into a live interactive SSH shell to manually configure a specific target:

netpulse-ssh shell 192.168.1.5 -u admin -p password

(If you omit -u or -p, the CLI will securely prompt you for them!)

Python API Integration

Use our cleanly typed Pydantic models and logic natively inside your own tools. The runner is completely agnostic and returns an audit object containing stdout/stderr per host:

import asyncio
from netpulse.ssh.models import SshHostConfig
from netpulse.ssh.runner import SshRunnerService

async def main():
    hosts = [
        SshHostConfig(ip="192.168.1.1", username="admin", password="password"),
        SshHostConfig(ip="10.0.0.1", username="admin", password="password"),
    ]
    
    runner = SshRunnerService()
    audit = await runner.execute_concurrently(hosts, "show version")
    
    for result in audit.results:
        print(f"Host {result.ip} status: {result.status}")
        print(result.stdout)

if __name__ == "__main__":
    asyncio.run(main())

📖 Documentation

Detailed documentation is available in the docs/ directory:

• 🗺️ Usage Guide - Deep dive into CLI and REST API examples.
• 📜 Playbooks & Inventory - Learn how to build Automated Deployment sequences.
• 🏗️ Architecture - Understand the asynchronous execution engine and legacy fallbacks.
• 🧠 Design Decisions - Read about the decoupled persistence approach.
• 🧪 Contributing - The development onboarding guide.

🤝 Contributing

We welcome contributions from the community! netpulse-ssh is open-source and maintained actively. Please read our Contributing Guidelines to understand how to clone the repository and submit your Pull Requests.

📜 License

Distributed under the MIT License. See LICENSE for more information.