tessl/pypi-twisted
An asynchronous networking framework written in Python
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
ssh.mddocs/
SSH and Terminal Protocols
SSH version 2 protocol implementation with client, server, SFTP support, and terminal handling capabilities. Twisted Conch provides comprehensive SSH functionality including key management, authentication, channel forwarding, and terminal protocols.
Capabilities
SSH Transport
Core SSH transport layer handling encryption, key exchange, and connection establishment.
class transport.SSHTransportBase:
"""
Base class for SSH transport implementations.
Attributes:
- supportedCiphers: List of supported encryption ciphers
- supportedMACs: List of supported MAC algorithms
- supportedKeyExchanges: List of supported key exchange methods
"""
supportedCiphers = None
supportedMACs = None
supportedKeyExchanges = None
class transport.SSHClientTransport(transport.SSHTransportBase):
"""
SSH client transport protocol.
"""
def verifyHostKey(self, pubKey, fingerprint):
"""
Verify the server's host key.
Args:
pubKey (bytes): Server's public key
fingerprint (str): Key fingerprint
Returns:
Deferred: Fires when verification complete
"""
def connectionSecure(self):
"""Called when SSH connection is secure and ready for authentication."""
class transport.SSHServerTransport(transport.SSHTransportBase):
"""
SSH server transport protocol.
"""
def getPublicKeys(self):
"""
Get server's public keys.
Returns:
dict: Mapping of key types to public keys
"""
def getPrivateKeys(self):
"""
Get server's private keys.
Returns:
dict: Mapping of key types to private keys
"""SSH Connection
SSH connection layer managing channels and requests.
class connection.SSHConnection:
"""
SSH connection protocol managing multiple channels.
Attributes:
- transport: Underlying SSH transport
- channels: Dict of active channels
"""
transport = None
channels = None
def openChannel(self, channel, extraData=b''):
"""
Open a new SSH channel.
Args:
channel: Channel instance
extraData (bytes): Extra channel data
Returns:
Deferred: Fires when channel is open
"""
def sendRequest(self, request, data, wantReply=False):
"""
Send a global request.
Args:
request (bytes): Request type
data (bytes): Request data
wantReply (bool): Whether to wait for reply
Returns:
Deferred or None: Reply data if wantReply=True
"""SSH Authentication
Client and server authentication handling.
class userauth.SSHUserAuthClient:
"""
SSH client user authentication.
"""
def getPassword(self, prompt=None):
"""
Get password for authentication.
Args:
prompt (str): Password prompt
Returns:
Deferred[str]: Password
"""
def getPrivateKey(self):
"""
Get private key for public key authentication.
Returns:
Deferred[Key]: Private key object
"""
def getPublicKey(self):
"""
Get public key for authentication.
Returns:
bytes: Public key data
"""
class userauth.SSHUserAuthServer:
"""
SSH server user authentication.
"""
def auth_password(self, packet):
"""
Handle password authentication.
Args:
packet (bytes): Authentication packet
Returns:
Deferred[bool]: Success/failure
"""
def auth_publickey(self, packet):
"""
Handle public key authentication.
Args:
packet (bytes): Authentication packet
Returns:
Deferred[bool]: Success/failure
"""SSH Channels
Base classes and implementations for SSH channels.
class channel.SSHChannel:
"""
Base class for SSH channels.
Attributes:
- conn: SSH connection object
- data: Channel-specific data
- localWindowSize: Local window size
- remoteWindowSize: Remote window size
"""
conn = None
data = None
localWindowSize = None
remoteWindowSize = None
def channelOpen(self, specificData):
"""
Called when channel is opened.
Args:
specificData (bytes): Channel-specific data
"""
def dataReceived(self, data):
"""
Called when data is received on channel.
Args:
data (bytes): Received data
"""
def extReceived(self, dataType, data):
"""
Called when extended data is received.
Args:
dataType (int): Extended data type
data (bytes): Extended data
"""
def eofReceived(self):
"""Called when EOF is received on channel."""
def closeReceived(self):
"""Called when channel close is received."""
def write(self, data):
"""
Write data to channel.
Args:
data (bytes): Data to write
Returns:
int: Number of bytes written
"""
def writeExtended(self, dataType, data):
"""
Write extended data to channel.
Args:
dataType (int): Extended data type
data (bytes): Data to write
"""
def loseConnection(self):
"""Close the channel."""SSH Key Management
SSH key handling, generation, and format conversion.
class keys.Key:
"""
SSH key wrapper supporting multiple key formats.
Attributes:
- keyObject: Underlying cryptographic key object
- type: Key type string (e.g., 'rsa', 'ed25519')
"""
keyObject = None
type = None
@classmethod
def fromString(cls, data, type=None, passphrase=None):
"""
Load key from string data.
Args:
data (bytes): Key data
type (str): Key type hint
passphrase (bytes): Passphrase for encrypted keys
Returns:
Key: Key object
"""
@classmethod
def fromFile(cls, filename, type=None, passphrase=None):
"""
Load key from file.
Args:
filename (str): Key file path
type (str): Key type hint
passphrase (bytes): Passphrase for encrypted keys
Returns:
Key: Key object
"""
def toString(self, type, extra=None):
"""
Export key to string format.
Args:
type (str): Output format ('openssh', 'lsh', 'agentv3')
extra: Extra parameters for format
Returns:
bytes: Key data in specified format
"""
def public(self):
"""
Get public key.
Returns:
Key: Public key object
"""
def isPublic(self):
"""
Check if this is a public key.
Returns:
bool: True if public key
"""
def fingerprint(self):
"""
Get key fingerprint.
Returns:
bytes: Key fingerprint
"""
def sign(self, data):
"""
Sign data with private key.
Args:
data (bytes): Data to sign
Returns:
bytes: Signature
"""
def verify(self, signature, data):
"""
Verify signature with public key.
Args:
signature (bytes): Signature to verify
data (bytes): Original data
Returns:
bool: True if signature is valid
"""
def keys.getPublicKeyString(data, type=None):
"""
Extract public key string from key data.
Args:
data (bytes): Key data
type (str): Key type hint
Returns:
bytes: Public key string
"""
def keys.getPrivateKeyObject(data, passphrase=None):
"""
Get private key object from key data.
Args:
data (bytes): Private key data
passphrase (bytes): Key passphrase
Returns:
Key: Private key object
"""
def keys.encryptKey(keyObject, passphrase):
"""
Encrypt a private key with passphrase.
Args:
keyObject: Private key object
passphrase (bytes): Encryption passphrase
Returns:
bytes: Encrypted key data
"""SSH Key Usage Example:
from twisted.conch.ssh import keys
# Load private key from file
private_key = keys.Key.fromFile('/path/to/private_key')
# Get public key
public_key = private_key.public()
# Get fingerprint
fingerprint = public_key.fingerprint()
print(f"Key fingerprint: {fingerprint}")
# Sign data
data = b"Hello, world!"
signature = private_key.sign(data)
# Verify signature
is_valid = public_key.verify(signature, data)
print(f"Signature valid: {is_valid}")SFTP File Transfer
SSH File Transfer Protocol implementation for file operations over SSH.
class filetransfer.FileTransferServer:
"""
SFTP server implementation.
Attributes:
- avatar: User avatar object
"""
avatar = None
def packet_OPEN(self, data):
"""Handle file open request."""
def packet_CLOSE(self, data):
"""Handle file close request."""
def packet_READ(self, data):
"""Handle file read request."""
def packet_WRITE(self, data):
"""Handle file write request."""
def packet_STAT(self, data):
"""Handle file stat request."""
class filetransfer.FileTransferClient:
"""
SFTP client implementation.
"""
def openFile(self, filename, flags, attrs):
"""
Open a remote file.
Args:
filename (bytes): File path
flags (int): Open flags
attrs: File attributes
Returns:
Deferred[SFTPFile]: File object
"""
def removeFile(self, filename):
"""
Remove a remote file.
Args:
filename (bytes): File path
Returns:
Deferred: Completion indicator
"""
def renameFile(self, oldPath, newPath):
"""
Rename a remote file.
Args:
oldPath (bytes): Current file path
newPath (bytes): New file path
Returns:
Deferred: Completion indicator
"""
def makeDirectory(self, path, attrs):
"""
Create a remote directory.
Args:
path (bytes): Directory path
attrs: Directory attributes
Returns:
Deferred: Completion indicator
"""
def getAttrs(self, path):
"""
Get file attributes.
Args:
path (bytes): File path
Returns:
Deferred[dict]: File attributes
"""
class filetransfer.SFTPFile:
"""
SFTP file object for reading/writing remote files.
"""
def readChunk(self, offset, length):
"""
Read chunk of file data.
Args:
offset (int): Read offset
length (int): Bytes to read
Returns:
Deferred[bytes]: File data
"""
def writeChunk(self, offset, data):
"""
Write chunk of file data.
Args:
offset (int): Write offset
data (bytes): Data to write
Returns:
Deferred: Completion indicator
"""
def getAttrs(self):
"""
Get file attributes.
Returns:
Deferred[dict]: File attributes
"""
def close(self):
"""
Close the file.
Returns:
Deferred: Completion indicator
"""SSH Agent
SSH agent protocol for managing SSH keys.
class agent.SSHAgentClient:
"""
SSH agent client for communicating with ssh-agent.
"""
def requestIdentities(self):
"""
Request list of identities from agent.
Returns:
Deferred[list]: List of (key_blob, comment) tuples
"""
def signData(self, blob, data):
"""
Request agent to sign data with key.
Args:
blob (bytes): Key blob
data (bytes): Data to sign
Returns:
Deferred[bytes]: Signature
"""
def addIdentity(self, key, comment=b''):
"""
Add identity to agent.
Args:
key: Private key object
comment (bytes): Key comment
Returns:
Deferred: Completion indicator
"""
def removeIdentity(self, blob):
"""
Remove identity from agent.
Args:
blob (bytes): Key blob to remove
Returns:
Deferred: Completion indicator
"""
class agent.SSHAgentServer:
"""
SSH agent server implementation.
"""
def __init__(self):
self.keys = {} # Storage for keys
def getPublicKeys(self):
"""
Get list of public keys.
Returns:
list: List of public key objects
"""
def getPrivateKey(self, publicKey):
"""
Get private key for given public key.
Args:
publicKey: Public key object
Returns:
Key or None: Private key if available
"""Terminal Protocols
Terminal emulation and line-oriented protocols.
class insults.ServerProtocol:
"""
Terminal server protocol with VT100 emulation.
Attributes:
- width: Terminal width in characters
- height: Terminal height in characters
"""
width = 80
height = 24
def terminalSize(self, width, height):
"""
Handle terminal size change.
Args:
width (int): New terminal width
height (int): New terminal height
"""
def keystroke(self, keyID, modifier):
"""
Handle keystroke input.
Args:
keyID: Key identifier
modifier: Key modifiers
"""
def write(self, data):
"""
Write data to terminal.
Args:
data (bytes): Data to write
"""
class manhole.ColoredManhole:
"""
Interactive Python shell with syntax highlighting.
"""
def __init__(self, namespace=None):
"""
Args:
namespace (dict): Python namespace for shell
"""
class recvline.HistoricRecvLine:
"""
Line receiver with command history.
Attributes:
- historyLines: List of command history
- historyPosition: Current position in history
"""
historyLines = None
historyPosition = None
def lineReceived(self, line):
"""
Handle received line.
Args:
line (str): Received line
"""SSH Factory
Factory classes for creating SSH protocol instances.
class factory.SSHFactory:
"""
Factory for SSH protocol instances.
Attributes:
- publicKeys: Dict of server public keys
- privateKeys: Dict of server private keys
- services: Dict of available SSH services
"""
publicKeys = None
privateKeys = None
services = None
def getPublicKeys(self):
"""
Get server's public keys.
Returns:
dict: Public keys by algorithm name
"""
def getPrivateKeys(self):
"""
Get server's private keys.
Returns:
dict: Private keys by algorithm name
"""
def getPrimes(self):
"""
Get Diffie-Hellman primes.
Returns:
dict: DH primes by size
"""
def getService(self, transport, name):
"""
Get SSH service by name.
Args:
transport: SSH transport
name (bytes): Service name
Returns:
Service instance or None
"""SSH Server Usage Example:
from twisted.conch import manhole_ssh
from twisted.conch.ssh import factory, keys
from twisted.internet import reactor, endpoints
from twisted.cred import portal, checkers
# Create SSH factory
ssh_factory = factory.SSHFactory()
# Load server keys
with open('/etc/ssh/ssh_host_rsa_key') as key_file:
ssh_factory.privateKeys['ssh-rsa'] = keys.Key.fromString(key_file.read())
ssh_factory.publicKeys['ssh-rsa'] = ssh_factory.privateKeys['ssh-rsa'].public()
# Setup authentication
checker = checkers.InMemoryUsernamePasswordDatabaseDontUse()
checker.addUser(b"admin", b"password")
realm = manhole_ssh.TerminalRealm()
portal_obj = portal.Portal(realm, [checker])
ssh_factory.portal = portal_obj
# Start SSH server
endpoint = endpoints.TCP4ServerEndpoint(reactor, 2222)
endpoint.listen(ssh_factory)
print("SSH server listening on port 2222")
reactor.run()Install with Tessl CLI
npx tessl i tessl/pypi-twisted