netpulse-ssh 0.1.3

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

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

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.