netpulse-ssh 0.1.6

High-speed concurrent SSH command runner

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.
• Legacy Equipment Healing: Automatically detects strict OpenSSH cipher drops and seamlessly falls back to legacy algorithms (e.g., diffie-hellman-group1-sha1, 3des-cbc) which are frequently required for older Cisco or Juniper gear.
• Smart Privilege Escalation: Natively supports Cisco enable mode privilege escalation without breaking automation.
• Pagination Suppression: Automatically injects terminal length 0 before command execution to bypass interactive --More-- prompts.
• 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

Concurrent Execution Execute a command across multiple devices concurrently and get a beautiful, structured table summary:

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

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.
• 🏗️ 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.