SSH Client¶

The SSH client is the primary interface for connecting to SSH servers and executing remote operations. SpindleX provides both a synchronous SSHClient and an asynchronous AsyncSSHClient.

Basic Usage¶

Creating and Configuring a Client¶

SyncAsync

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

# Create client
client = SSHClient()

# Configure host key policy. RejectPolicy is the secure default -
# shown explicitly here to make the security posture obvious.
client.set_missing_host_key_policy(RejectPolicy())

# Load ~/.ssh/known_hosts so trusted servers can be verified.
client.get_host_keys().load()

from spindlex import AsyncSSHClient

# AsyncSSHClient uses a similar configuration but is designed for asyncio
async with AsyncSSHClient() as client:
    # Configuration is often handled during connect
    pass

Connection Methods¶

Password Authentication¶

SyncAsync

client.connect(
    hostname='example.com',
    port=22,
    username='myuser',
    password='mypassword',
    timeout=30
)

await client.connect(
    hostname='example.com',
    username='myuser',
    password='mypassword'
)

Public Key Authentication¶

SyncAsync

from spindlex.crypto import PKey

# Load key from file
private_key = PKey.from_private_key_file('/path/to/key')

client.connect(
    hostname='example.com',
    username='myuser',
    pkey=private_key
)

from spindlex.crypto import PKey

private_key = PKey.from_private_key_file('/path/to/key')

await client.connect(
    hostname='example.com',
    username='myuser',
    pkey=private_key
)

Command Execution¶

Simple Commands¶

SyncAsync

# Execute a simple command
stdin, stdout, stderr = client.exec_command('ls -la')

# Read output
output = stdout.read().decode('utf-8')
error = stderr.read().decode('utf-8')

# Get exit status
exit_status = stdout.get_exit_status()

# Execute a simple command
stdin, stdout, stderr = await client.exec_command('ls -la')

# Read output
output = await stdout.read()
error = await stderr.read()

# Get exit status
exit_status = await stdout.recv_exit_status()

Commands with Input¶

SyncAsync

stdin, stdout, stderr = client.exec_command('cat > /tmp/test.txt')

# Send input to the command
stdin.write('Hello, World!\n')
stdin.flush()
stdin.close()

stdin, stdout, stderr = await client.exec_command('cat > /tmp/test.txt')

# Send input
await stdin.write('Hello, World!\n')
await stdin.close()

Host Key Management¶

Host Key Policies¶

from spindlex.hostkeys.policy import (
    AutoAddPolicy, RejectPolicy, WarningPolicy
)

# Reject all unknown host keys (the secure default - recommended).
client.set_missing_host_key_policy(RejectPolicy())

# Log warning but accept unknown host keys. Use this only when you
# understand that an attacker can still MITM the first connection.
client.set_missing_host_key_policy(WarningPolicy())

# WARNING: AutoAddPolicy trusts every first-seen host key and therefore
# disables MITM protection. Only use it in short-lived disposable test
# environments - never in production.
client.set_missing_host_key_policy(AutoAddPolicy())

Port Forwarding¶

SpindleX supports both local and remote port forwarding (SSH tunneling).

Local Port Forwarding¶

Local port forwarding allows you to forward a port on your local machine to a port on a remote server.

SyncAsync

with SSHClient() as client:
    client.connect('jump-host.example.com', username='user')

    # Forward local port 8080 to remote-server.internal:80
    tunnel_id = client.create_local_port_forward(
        local_port=8080,
        remote_host='remote-server.internal',
        remote_port=80
    )

    print(f"Tunnel {tunnel_id} established. Connect to localhost:8080")

    # Keep the connection open while you use the tunnel
    import time
    while True:
        time.sleep(1)

from spindlex import AsyncSSHClient

async def main():
    async with AsyncSSHClient() as client:
        await client.connect('jump-host.example.com', username='user')

        # Forward local port 8080 to remote-server.internal:80
        tunnel_id = await client.create_local_port_forward(
            local_port=8080,
            remote_host='remote-server.internal',
            remote_port=80
        )

        print(f"Tunnel {tunnel_id} established.")

        # Keep the connection open while you use the tunnel
        while True:
            await asyncio.sleep(1)

Remote Port Forwarding¶

Remote port forwarding allows you to forward a port on the remote server to a port on your local machine.

SyncAsync

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

    # Forward remote port 9090 to localhost:3000
    tunnel_id = client.create_remote_port_forward(
        remote_port=9090,
        local_host='127.0.0.1',
        local_port=3000
    )

    print(f"Remote tunnel {tunnel_id} established on server:9090")

```python async def main(): async with AsyncSSHClient() as client: await client.connect('server.example.com', username='user')

    # Forward remote port 9090 to localhost:3000
    tunnel_id = await client.create_remote_port_forward(
        remote_port=9090,
        local_host='127.0.0.1',
        local_port=3000
    )

    print(f"Remote tunnel {tunnel_id} established.")

Managing Tunnels¶

Once a tunnel is created, you can manage it using the tunnel_id returned by the creation methods.

Closing a Specific Tunnel¶

SyncAsync

client.close_port_forward(tunnel_id)

await client.close_port_forward(tunnel_id)

Listing Active Tunnels¶

SyncAsync

tunnels = client.get_port_forwards()
for tunnel_id, info in tunnels.items():
    print(f"Tunnel {tunnel_id}: {info['local_addr']} -> {info['remote_addr']}")

tunnels = client.get_port_forwards()
for tunnel_id, info in tunnels.items():
    print(f"Tunnel {tunnel_id}: {info['local_addr']} -> {info['remote_addr']}")

Error Handling¶

Common Exceptions¶

from spindlex.exceptions import (
    SSHException,
    AuthenticationException,
    BadHostKeyException,
    TransportException
)

try:
    client.connect('example.com', username='user', password='pass')
except AuthenticationException as e:
    print(f"Authentication failed: {e}")
except BadHostKeyException as e:
    print(f"Host key verification failed: {e}")
except TransportException as e:
    print(f"Transport error: {e}")
except SSHException as e:
    print(f"General SSH error: {e}")

Best Practices¶

Always close connections: Use context managers (with or async with).
Use key-based authentication: More secure than passwords.
Implement proper host key verification: Don't use AutoAddPolicy in production.
Handle timeouts appropriately: Set reasonable timeout values.
Monitor connection health: Check transport.active periodically.