Automation Recipes¶

from spindlex import SSHClient
from spindlex.hostkeys.policy import RejectPolicy

with SSHClient() as client:
    # Secure default: reject unknown host keys. Make sure the server's key
    # is in ~/.ssh/known_hosts before running (e.g. via `ssh user@host` once).
    client.set_missing_host_key_policy(RejectPolicy())
    client.get_host_keys().load()

    client.connect('server.example.com', username='admin', password='password')

    # exec_command returns (stdin, stdout, stderr)
    stdin, stdout, stderr = client.exec_command('ls -l /tmp')

    # stdout and stderr are iterable and return lines
    for line in stdout:
        print(line.strip())

    # Get the exit status (0 usually means success)
    exit_status = stdout.recv_exit_status()
    print(f"Command exited with status: {exit_status}")

from spindlex import SSHClient

with SSHClient() as client:
    client.connect('server.example.com', username='admin')

    with client.open_sftp() as sftp:
        # Upload a file
        sftp.put('local_file.txt', '/home/admin/remote_file.txt')

        # Download a file
        sftp.get('/var/log/syslog', 'local_syslog.log')

        # List directory
        print("Files in home:")
        for filename in sftp.listdir('/home/admin'):
            print(filename)

from spindlex import SSHClient

def run_sudo(client, command, password):
    """
    Executes a command with sudo, handling the password prompt.
    """
    stdin, stdout, stderr = client.exec_command(f"sudo -S {command}")

    # Provide password when sudo asks
    stdin.write(f"{password}\n")

    # Read the response
    return stdout.read().decode()

with SSHClient() as client:
    client.connect('server01', username='admin')
    result = run_sudo(client, "apt-get update", "my-secret-pass")
    print(result)

import asyncio
from spindlex import AsyncSSHClient

async def run_on_server(hostname, command):
    try:
        async with AsyncSSHClient() as client:
            await client.connect(hostname, username='admin')
            stdin, stdout, stderr = await client.exec_command(command)

            # AsyncChannelFile is also async iterable
            async for line in stdout:
                print(f"[{hostname}] {line.strip()}")

            status = await stdout.recv_exit_status()
            print(f"[{hostname}] Finished with status {status}")
    except Exception as e:
        print(f"[{hostname}] Error: {e}")

async def main():
    servers = ['srv1', 'srv2', 'srv3', 'srv4']
    tasks = [run_on_server(s, 'uptime') for s in servers]
    await asyncio.gather(*tasks)

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

from spindlex import SSHClient

with SSHClient() as bastion:
    bastion.connect('bastion.example.com', username='gatekeeper')

    # Open a direct channel to the internal target through the bastion
    transport = bastion.get_transport()
    dest_addr = ('internal-target.lan', 22)

    # Create a direct-tcpip channel (tunnels TCP to target)
    channel = transport.open_channel("direct-tcpip", dest_addr)

    # Connect to target using the channel as a socket
    with SSHClient() as target:
        target.connect(
            'internal-target.lan', 
            username='admin', 
            sock=channel
        )
        stdin, stdout, stderr = target.exec_command('hostname')
        print(f"Connected to: {stdout.read().decode().strip()}")

from spindlex import SSHClient

with SSHClient() as client:
    client.connect('prod-app-01', username='devops')

    stdin, stdout, stderr = client.exec_command('tail -f /var/log/nginx/access.log')

    print("--- Streaming Remote Logs (Ctrl+C to stop) ---")
    try:
        # ChannelFile is iterable line-by-line
        for line in stdout:
            print(line.strip())
    except KeyboardInterrupt:
        print("Stopping log stream...")

from spindlex import SSHClient

with SSHClient() as client:
    client.connect('secure-host', username='audit-user')

    # Rekey every 100MB or every 15 minutes
    transport = client.get_transport()
    transport.set_rekey_policy(
        bytes_limit=100 * 1024 * 1024, 
        time_limit=900
    )

    # Continue with secure operations
    # ...

import getpass
from spindlex import SSHClient

def interactive_handler(title, instructions, prompts):
    """
    Handles keyboard-interactive challenges.
    'prompts' is a list of (prompt_text, echo_boolean) tuples.
    """
    answers = []
    print(f"\n--- {title} ---")
    if instructions:
        print(instructions)

    for text, echo in prompts:
        if echo:
            ans = input(text)
        else:
            ans = getpass.getpass(text)
        answers.append(ans)
    return answers

with SSHClient() as client:
    # This will trigger the interactive_handler if the server requests it
    client.connect(
        'mfa-enabled-host', 
        username='user', 
        handler=interactive_handler
    )

from spindlex import SSHClient

with SSHClient() as client:
    # Set gss_auth=True to attempt Kerberos authentication
    client.connect(
        'kerberos-host.internal', 
        username='jdoe',
        gss_auth=True,
        gss_deleg_creds=True
    )
    print("Authenticated via Kerberos!")

from spindlex import SSHClient
import os

def rotate_key(client, new_key_path):
    with open(new_key_path, 'r') as f:
        new_key = f.read().strip()

    # Append new key to authorized_keys
    client.exec_command(f'echo "{new_key}" >> ~/.ssh/authorized_keys')
    print("New key added.")

# Example usage
# ...

from spindlex import SSHClient
from datetime import datetime

def backup_config(hostname, remote_path, local_dir):
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    local_path = f"{local_dir}/{hostname}_{timestamp}.conf"

    with SSHClient() as client:
        client.connect(hostname, username='admin')
        with client.open_sftp() as sftp:
            sftp.get(remote_path, local_path)
            print(f"Backup saved to {local_path}")

# backup_config('router01', '/etc/config', './backups')