What is SFTP?
SFTP, or SSH File Transfer Protocol, is a network protocol that provides secure file access, file transfer, and file management over a reliable data stream. Despite the similarity in name, SFTP is not FTP run over SSH. It is an entirely different protocol designed from the ground up as an extension of the SSH-2 protocol. SFTP runs inside an SSH tunnel, leveraging SSH's authentication and encryption capabilities to protect both the data and the commands exchanged between client and server.
At its core, SFTP is a remote file system protocol. Unlike traditional FTP, which uses separate control and data connections, SFTP multiplexes everything through a single SSH connection. This means you can perform file operations—upload, download, rename, delete, list directories, change permissions—all over one encrypted channel, without the complexities of passive/active mode negotiations or firewall-unfriendly port ranges.
SFTP vs FTPS: A Critical Distinction
Developers often confuse SFTP with FTPS. The distinction is crucial:
- FTPS (FTP over SSL/TLS) is the legacy FTP protocol with TLS encryption layered on top. It still uses separate control and data connections, requires certificate management, and suffers from the same NAT/firewall traversal issues as plain FTP.
- SFTP (SSH File Transfer Protocol) is a completely independent protocol that operates within an SSH session. It uses a single port (typically 22), benefits from SSH's robust key-based authentication, and provides a richer set of file system operations out of the box.
SFTP's protocol operations are more granular and stateful than FTP's. For instance, SFTP supports atomic file renames, recursive directory listing with file attributes in a single response, and fine-grained permission management—capabilities that require awkward workarounds in FTP.
Why SFTP Matters for Modern Development
🚀 Deploy your AI agent in 10 minutes
Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.
Try it free →In contemporary software development, SFTP is indispensable across several domains:
- Secure deployment pipelines: Shipping code artifacts to production servers without exposing credentials over plaintext channels
- Automated data exchange: ETL processes pulling files from third-party vendors or pushing reports to business partners
- Infrastructure management: Provisioning configuration files to remote machines in a programmatic, auditable manner
- Compliance requirements: Regulations like HIPAA, PCI-DSS, and GDPR mandate encrypted data in transit—SFTP satisfies these with minimal operational overhead
- Cloud and hybrid architectures: Moving data between on-premises systems and cloud storage gateways that expose SFTP endpoints
Beyond security, SFTP offers reliability. The protocol handles network interruptions gracefully: most implementations support resume for interrupted transfers, and the SSH layer provides keep-alive mechanisms to detect stale connections. For long-running batch jobs, this is a significant advantage over HTTP-based file transfers that may require chunked uploads and manual checkpointing.
How the SFTP Protocol Works
The Layered Architecture
Understanding SFTP requires understanding its layered design. The stack looks like this:
- Layer 1 — SSH Transport: Handles TCP connection, key exchange, encryption cipher negotiation, and message integrity (HMAC). This layer produces a secure, authenticated pipe.
- Layer 2 — SSH Authentication: Within the encrypted tunnel, the client authenticates using password, public key, keyboard-interactive, or GSSAPI methods. Once authenticated, the SSH server opens a "session" or "subsystem" channel.
- Layer 3 — SFTP Subsystem: The client requests the
sftp-serversubsystem. From this point onward, all data on the channel conforms to the SFTP binary protocol—a structured request/response format defined in the IETF draft-ietf-secsh-filexfer standards.
SFTP Protocol Operations
The SFTP protocol uses a binary packet format. Each request packet contains a length field, a type code, a request ID (for matching responses), and operation-specific parameters. The core operations include:
- SSH_FXP_INIT — Client sends its version and supported extensions; server responds with its version.
- SSH_FXP_OPEN — Opens a file with flags (read, write, append, create, truncate) and desired attributes.
- SSH_FXP_READ / SSH_FXP_WRITE — Reads or writes a block of data at a specified offset within an open file handle.
- SSH_FXP_CLOSE — Closes an open file handle.
- SSH_FXP_OPENDIR / SSH_FXP_READDIR — Opens a directory for enumeration and reads one or more entries.
- SSH_FXP_STAT / SSH_FXP_LSTAT / SSH_FXP_FSTAT — Retrieves file metadata (size, permissions, timestamps).
- SSH_FXP_MKDIR / SSH_FXP_RMDIR — Creates or removes directories.
- SSH_FXP_REMOVE — Deletes a file.
- SSH_FXP_RENAME — Atomically renames a file (classic rename or POSIX rename with overwrite semantics).
- SSH_FXP_SETSTAT — Modifies file attributes like permissions and timestamps.
Each request elicits a response: either a success confirmation with payload data, or an error status code with a message. The protocol is synchronous at the request level but supports pipelining—a client can issue multiple requests before receiving responses, allowing high-throughput transfers over high-latency links.
The Wire Format
SFTP packets are length-prefixed and type-tagged. A simplified view of a READ request on the wire:
+------+------+--------+----+------+------+
| Len | Type | Req ID | Handle | Offset| Len |
| 4B | 1B | 4B | 4B+str | 8B | 4B |
+------+------+--------+---------+-------+------+This binary efficiency means SFTP outperforms text-based protocols like FTP in parsing overhead. Libraries handle this encoding transparently, but understanding the structure helps when debugging raw packet dumps or implementing custom extensions.
Implementing an SFTP Client in Python
For practical implementation, we'll use the paramiko library, which provides a pure-Python SSH-2 implementation with full SFTP support. Paramiko is widely adopted, well-documented, and suitable for both interactive and automated use cases. We'll also explore asyncssh for high-concurrency scenarios.
Setting Up Your Environment
Install the required libraries:
pip install paramiko
pip install asyncssh # optional, for async implementationsFor testing, you'll need an SSH server with SFTP enabled. Most Linux distributions ship with OpenSSH's sftp-server subsystem enabled by default. You can verify this in /etc/ssh/sshd_config:
Subsystem sftp internal-sftp-serverEstablishing an SFTP Session
The first step is establishing an SSH connection and opening an SFTP channel. Here's a complete, production-ready connection manager that handles both password and key-based authentication:
import paramiko
import os
import logging
from io import StringIO
from typing import Optional, Tuple
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SFTPConnection:
"""Manages an SFTP connection with automatic reconnection support."""
def __init__(
self,
host: str,
port: int = 22,
username: str = None,
password: str = None,
private_key_path: str = None,
private_key_passphrase: str = None,
timeout: float = 10.0,
):
self.host = host
self.port = port
self.username = username
self.password = password
self.private_key_path = private_key_path
self.private_key_passphrase = private_key_passphrase
self.timeout = timeout
self._transport: Optional[paramiko.Transport] = None
self._sftp: Optional[paramiko.SFTPClient] = None
def connect(self) -> paramiko.SFTPClient:
"""Establish the SSH connection and open the SFTP channel."""
self._transport = paramiko.Transport((self.host, self.port))
self._transport.start_client()
# Determine authentication method
if self.private_key_path:
private_key = self._load_private_key()
self._transport.auth_publickey(self.username, private_key)
logger.info(f"Authenticated with key to {self.host}")
elif self.password:
self._transport.auth_password(self.username, self.password)
logger.info(f"Authenticated with password to {self.host}")
else:
# Attempt password-less key agent authentication
agent = paramiko.Agent()
agent_keys = agent.get_keys()
if agent_keys:
for key in agent_keys:
try:
self._transport.auth_publickey(self.username, key)
logger.info(f"Authenticated via SSH agent to {self.host}")
break
except paramiko.AuthenticationException:
continue
else:
raise paramiko.AuthenticationException("No acceptable key found in agent")
else:
raise ValueError("No authentication credentials provided")
# Open SFTP channel
self._sftp = paramiko.SFTPClient.from_transport(self._transport)
logger.info(f"SFTP channel opened to {self.host}")
return self._sftp
def _load_private_key(self) -> paramiko.PKey:
"""Load a private key from file, handling different formats."""
key_path = os.path.expanduser(self.private_key_path)
if not os.path.isfile(key_path):
raise FileNotFoundError(f"Private key not found: {key_path}")
# Try different key formats
key_classes = [
paramiko.RSAKey,
paramiko.ECDSACertainTypesKey,
paramiko.Ed25519Key,
paramiko.DSSKey,
]
passphrase = self.private_key_passphrase
for key_class in key_classes:
try:
return key_class.from_private_key_file(key_path, passphrase)
except paramiko.ssh_exception.PasswordRequiredException:
raise # Re-raise if passphrase is required but not provided
except Exception:
continue
# If none of the specific types worked, try the generic method
try:
return paramiko.PKey.from_private_key_file(key_path, passphrase)
except paramiko.ssh_exception.PasswordRequiredException:
raise
except Exception as e:
raise ValueError(f"Could not load private key from {key_path}: {e}")
@property
def sftp(self) -> paramiko.SFTPClient:
"""Get the active SFTP client, connecting if necessary."""
if self._sftp is None or self._sftp.sock is None:
return self.connect()
return self._sftp
def close(self):
"""Gracefully close the SFTP channel and SSH connection."""
if self._sftp:
try:
self._sftp.close()
except Exception:
pass
self._sftp = None
if self._transport:
try:
self._transport.close()
except Exception:
pass
self._transport = None
logger.info(f"Connection to {self.host} closed")
def __enter__(self):
return self.connect()
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
return FalseFile Upload and Download Operations
With the connection established, here are robust implementations for common transfer operations. Notice the use of SFTPClient.put and SFTPClient.get which handle chunked transfers internally, along with callback-based progress reporting:
import os
import sys
from contextlib import contextmanager
from typing import Callable, Optional
class SFTPTransfer:
"""High-level SFTP transfer operations with progress tracking."""
def __init__(self, sftp_client: paramiko.SFTPClient):
self.sftp = sftp_client
def upload_file(
self,
local_path: str,
remote_path: str,
callback: Optional[Callable[[int, int], None]] = None,
overwrite: bool = True,
) -> None:
"""
Upload a single file to the remote server.
Args:
local_path: Absolute or relative path to the local file
remote_path: Destination path on the remote server
callback: Progress function receiving (bytes_transferred, total_bytes)
overwrite: Whether to overwrite existing remote file
"""
if not os.path.isfile(local_path):
raise FileNotFoundError(f"Local file not found: {local_path}")
file_size = os.path.getsize(local_path)
remote_dir = os.path.dirname(remote_path)
# Ensure remote directory exists
self._ensure_remote_directory(remote_dir)
# Define internal callback to track progress
transferred_bytes = [0]
def internal_callback(chunk_transferred, _):
transferred_bytes[0] += chunk_transferred
if callback:
callback(transferred_bytes[0], file_size)
try:
self.sftp.put(
localpath=local_path,
remotepath=remote_path,
callback=internal_callback,
confirm=True, # Wait for server confirmation of each chunk
)
print(f"\nUpload complete: {local_path} -> {remote_path}")
except Exception as e:
# Attempt cleanup of partial file
try:
self.sftp.remove(remote_path)
except Exception:
pass
raise IOError(f"Upload failed: {e}")
def download_file(
self,
remote_path: str,
local_path: str,
callback: Optional[Callable[[int, int], None]] = None,
resume: bool = False,
) -> None:
"""
Download a file from the remote server with optional resume support.
Args:
remote_path: Path to the remote file
local_path: Local destination path
callback: Progress function receiving (bytes_transferred, total_bytes)
resume: If True and local file exists, attempt to resume download
"""
# Ensure local directory exists
local_dir = os.path.dirname(local_path)
if local_dir:
os.makedirs(local_dir, exist_ok=True)
# Get remote file attributes
try:
remote_stat = self.sftp.stat(remote_path)
remote_size = remote_stat.st_size
except IOError as e:
raise IOError(f"Cannot stat remote file '{remote_path}': {e}")
# Determine resume offset
local_offset = 0
if resume and os.path.exists(local_path):
local_offset = os.path.getsize(local_path)
if local_offset >= remote_size:
print("Local file is already complete.")
return
print(f"Resuming download from byte {local_offset}")
transferred_bytes = [local_offset]
def internal_callback(chunk_transferred, _):
transferred_bytes[0] += chunk_transferred
if callback:
callback(transferred_bytes[0], remote_size)
try:
self.sftp.get(
remotepath=remote_path,
localpath=local_path,
callback=internal_callback,
)
print(f"\nDownload complete: {remote_path} -> {local_path}")
except Exception as e:
# Keep partial file for resume
raise IOError(f"Download failed: {e}")
def upload_directory(
self,
local_dir: str,
remote_dir: str,
recursive: bool = True,
) -> int:
"""
Recursively upload a directory tree, preserving structure.
Returns the count of files transferred.
"""
if not os.path.isdir(local_dir):
raise NotADirectoryError(f"Local path is not a directory: {local_dir}")
file_count = 0
for dirpath, _, filenames in os.walk(local_dir):
# Compute relative path for remote side
rel_path = os.path.relpath(dirpath, local_dir)
if rel_path == ".":
current_remote_dir = remote_dir
else:
# Normalize path separators for the remote system
current_remote_dir = os.path.join(remote_dir, rel_path).replace("\\", "/")
# Ensure remote subdirectory exists
self._ensure_remote_directory(current_remote_dir)
for filename in filenames:
local_file = os.path.join(dirpath, filename)
remote_file = os.path.join(current_remote_dir, filename).replace("\\", "/")
self.upload_file(local_file, remote_file)
file_count += 1
if not recursive:
break
return file_count
def _ensure_remote_directory(self, remote_path: str) -> None:
"""Create a remote directory path if it doesn't exist."""
if remote_path in ("", ".", "/"):
return
try:
self.sftp.stat(remote_path)
# Path exists, verify it's a directory
return
except IOError:
pass # Path doesn't exist
# Recursively create parent directories
parent = os.path.dirname(remote_path)
if parent and parent != remote_path:
self._ensure_remote_directory(parent)
try:
self.sftp.mkdir(remote_path)
print(f"Created remote directory: {remote_path}")
except IOError as e:
# Check if it was created by a concurrent operation
try:
self.sftp.stat(remote_path)
return
except IOError:
raise IOError(f"Cannot create remote directory '{remote_path}': {e}")Remote File Listing and Metadata
Listing remote directories with file attributes is essential for building file browsers, sync tools, or audit scripts. The SFTPClient.listdir_attr method returns a list of SFTPAttributes objects, each containing the full stat information:
import time
import stat as stat_module
from datetime import datetime
from typing import List, Dict
class SFTPBrowser:
"""Remote file system browsing utilities."""
def __init__(self, sftp_client: paramiko.SFTPClient):
self.sftp = sftp_client
def list_directory(self, remote_path: str = ".") -> List[Dict]:
"""
List a remote directory with full metadata.
Returns a list of dictionaries with keys:
name, size, mode, atime, mtime, owner, group, is_dir
"""
entries = []
try:
for attr in self.sftp.listdir_attr(remote_path):
entry = {
"name": attr.filename,
"size": attr.st_size if attr.st_size is not None else 0,
"mode": attr.st_mode,
"is_dir": stat_module.S_ISDIR(attr.st_mode) if attr.st_mode else False,
"is_link": stat_module.S_ISLNK(attr.st_mode) if attr.st_mode else False,
"atime": datetime.fromtimestamp(attr.st_atime) if attr.st_atime else None,
"mtime": datetime.fromtimestamp(attr.st_mtime) if attr.st_mtime else None,
"owner": attr.st_uid,
"group": attr.st_gid,
}
entries.append(entry)
except IOError as e:
raise IOError(f"Failed to list directory '{remote_path}': {e}")
return entries
def list_recursive(self, remote_path: str = ".", max_depth: int = 5) -> Dict[str, List[Dict]]:
"""
Recursively list a remote directory tree up to max_depth levels.
Returns a dictionary mapping remote paths to their directory listings.
"""
tree = {}
self._walk_remote(remote_path, tree, max_depth, current_depth=0)
return tree
def _walk_remote(self, path: str, tree: Dict, max_depth: int, current_depth: int):
if current_depth > max_depth:
return
entries = self.list_directory(path)
tree[path] = entries
for entry in entries:
if entry["is_dir"] and not entry["is_link"]:
child_path = f"{path}/{entry['name']}".replace("//", "/")
self._walk_remote(child_path, tree, max_depth, current_depth + 1)
def search_files(self, remote_dir: str, pattern: str) -> List[str]:
"""
Search for files matching a glob-like pattern in a remote directory.
This is a simple substring-based search; for full globbing,
retrieve the listing and filter with fnmatch locally.
"""
import fnmatch
matches = []
try:
entries = self.sftp.listdir(remote_dir)
for filename in entries:
if fnmatch.fnmatch(filename, pattern):
matches.append(f"{remote_dir}/{filename}".replace("//", "/"))
except IOError as e:
raise IOError(f"Search failed in '{remote_dir}': {e}")
return matches
def get_file_info(self, remote_path: str) -> Dict:
"""
Retrieve detailed stat information for a single remote file or directory.
"""
try:
attr = self.sftp.stat(remote_path)
except IOError as e:
raise IOError(f"Cannot stat '{remote_path}': {e}")
return {
"path": remote_path,
"size": attr.st_size or 0,
"mode": attr.st_mode,
"permissions": stat_module.S_IMODE(attr.st_mode) if attr.st_mode else 0,
"is_dir": stat_module.S_ISDIR(attr.st_mode) if attr.st_mode else False,
"atime": datetime.fromtimestamp(attr.st_atime) if attr.st_atime else None,
"mtime": datetime.fromtimestamp(attr.st_mtime) if attr.st_mtime else None,
"uid": attr.st_uid,
"gid": attr.st_gid,
}Complete Integration Example
Here's how all the components work together in a real-world script that syncs a local directory to a remote SFTP server, preserving timestamps and handling errors gracefully:
#!/usr/bin/env python3
"""
sftp_sync.py - Bidirectional directory sync over SFTP.
Usage:
python sftp_sync.py push --host example.com --user deploy \
--key ~/.ssh/id_rsa --local ./build/ --remote /var/www/app/
python sftp_sync.py pull --host example.com --user deploy \
--key ~/.ssh/id_rsa --remote /var/www/logs/ --local ./logs/
"""
import argparse
import os
import sys
import logging
from datetime import datetime
# Assuming the classes above are in a module called sftp_client
# For this example, we inline the essential logic
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("sftp_sync")
def sync_push(connection, local_dir: str, remote_dir: str):
"""Push local files to remote, uploading only changed files."""
sftp = connection.sftp
transfer = SFTPTransfer(sftp)
for dirpath, _, filenames in os.walk(local_dir):
rel_path = os.path.relpath(dirpath, local_dir)
if rel_path == ".":
current_remote = remote_dir
else:
current_remote = f"{remote_dir}/{rel_path}".replace("\\", "/")
for filename in filenames:
local_file = os.path.join(dirpath, filename)
remote_file = f"{current_remote}/{filename}".replace("//", "/")
local_mtime = os.path.getmtime(local_file)
local_size = os.path.getsize(local_file)
# Check if remote file needs updating
try:
remote_stat = sftp.stat(remote_file)
# Compare modification times and sizes
if (abs(remote_stat.st_mtime - local_mtime) < 2 and
remote_stat.st_size == local_size):
logger.debug(f"Skipping unchanged: {filename}")
continue
except IOError:
pass # Remote file doesn't exist, upload needed
logger.info(f"Uploading: {filename} ({local_size} bytes)")
transfer.upload_file(local_file, remote_file)
# Preserve modification timestamp
try:
remote_stat = sftp.stat(remote_file)
attr = paramiko.SFTPAttributes()
attr.st_mtime = local_mtime
attr.st_atime = local_mtime
attr.st_mode = remote_stat.st_mode
sftp.utime(remote_file, attr)
except Exception as e:
logger.warning(f"Could not preserve timestamp on {filename}: {e}")
def sync_pull(connection, remote_dir: str, local_dir: str):
"""Pull remote files to local, downloading only changed files."""
sftp = connection.sftp
transfer = SFTPTransfer(sftp)
browser = SFTPBrowser(sftp)
remote_tree = browser.list_recursive(remote_dir, max_depth=10)
for remote_path, entries in remote_tree.items():
rel_path = os.path.relpath(remote_path, remote_dir)
if rel_path == ".":
current_local = local_dir
else:
current_local = os.path.join(local_dir, rel_path)
os.makedirs(current_local, exist_ok=True)
for entry in entries:
if entry["is_dir"]:
continue
remote_file = f"{remote_path}/{entry['name']}".replace("//", "/")
local_file = os.path.join(current_local, entry["name"])
# Check if download is needed
if os.path.exists(local_file):
local_mtime = os.path.getmtime(local_file)
remote_mtime = entry["mtime"]
if remote_mtime:
remote_mtime_ts = remote_mtime.timestamp()
if abs(local_mtime - remote_mtime_ts) < 2:
logger.debug(f"Skipping unchanged: {entry['name']}")
continue
logger.info(f"Downloading: {entry['name']} ({entry['size']} bytes)")
transfer.download_file(remote_file, local_file)
# Preserve remote modification timestamp locally
if entry["mtime"]:
mtime_ts = entry["mtime"].timestamp()
os.utime(local_file, (mtime_ts, mtime_ts))
def main():
parser = argparse.ArgumentParser(description="SFTP Directory Sync Tool")
parser.add_argument("direction", choices=["push", "pull"])
parser.add_argument("--host", required=True)
parser.add_argument("--port", type=int, default=22)
parser.add_argument("--user", required=True)
parser.add_argument("--password")
parser.add_argument("--key")
parser.add_argument("--key-passphrase")
parser.add_argument("--local", required=True)
parser.add_argument("--remote", required=True)
args = parser.parse_args()
connection = SFTPConnection(
host=args.host,
port=args.port,
username=args.user,
password=args.password,
private_key_path=args.key,
private_key_passphrase=args.key_passphrase,
)
try:
connection.connect()
start_time = datetime.now()
logger.info(f"Starting {args.direction} sync...")
if args.direction == "push":
sync_push(connection, args.local, args.remote)
else:
sync_pull(connection, args.remote, args.local)
elapsed = (datetime.now() - start_time).total_seconds()
logger.info(f"Sync complete in {elapsed:.2f} seconds")
except Exception as e:
logger.error(f"Sync failed: {e}")
sys.exit(1)
finally:
connection.close()
if __name__ == "__main__":
main()Asynchronous SFTP with AsyncSSH
For high-concurrency applications—such as monitoring hundreds of remote servers or handling many simultaneous transfers—an asynchronous approach is essential. The asyncssh library provides an asyncio-native SFTP implementation that can handle thousands of connections with minimal thread overhead:
import asyncio
import asyncssh
from typing import List, Dict
async def fetch_remote_inventory(hosts: List[Dict]) -> List[Dict]:
"""
Connect to multiple SFTP servers concurrently and fetch file inventories.
Each host dict should contain: host, username, password or client_keys.
"""
results = []
async def process_one(host_info: Dict):
hostname = host_info["host"]
try:
async with asyncssh.connect(
host=hostname,
username=host_info.get("username"),
password=host_info.get("password"),
client_keys=host_info.get("client_keys", ()),
known_hosts=None, # Disable for demo; use proper known_hosts in production
) as conn:
async with conn.start_sftp_client() as sftp:
# List root directory
entries = await sftp.listdir(".")
file_info = []
for entry in entries:
stat = await sftp.stat(entry)
file_info.append({
"name": entry,
"size": stat.st_size,
"mtime": stat.st_mtime,
})
return {"host": hostname, "status": "ok", "files": file_info}
except Exception as e:
return {"host": hostname, "status": "error", "message": str(e)}
tasks = [process_one(host) for host in hosts]
results = await asyncio.gather(*tasks)
return list(results)
async def async_upload_example():
"""Demonstrate concurrent uploads to multiple servers."""
async with asyncssh.connect(
host="sftp.example.com",
username="deploy",
client_keys=["~/.ssh/id_rsa"],
) as conn:
async with conn.start_sftp_client() as sftp:
# Upload a file asynchronously
local_path = "./deploy.tar.gz"
remote_path = "/var/www/releases/deploy.tar.gz"
async with open(local_path, "rb") as local_file:
async with sftp.open(remote_path, "wb") as remote_file:
while True:
chunk = local_file.read(32768)
if not chunk:
break
await remote_file.write(chunk)
print(f"Async upload complete: {remote_path}")
# Run the example
if __name__ == "__main__":
hosts = [
{"host": "server1.example.com", "username": "ops"},
{"host": "server2.example.com", "username": "ops"},
]
results = asyncio.run(fetch_remote_inventory(hosts))
for result in results:
print(f"{result['host']}: {result['status']}")Best Practices for Production SFTP Implementations
1. Connection Management
Always use connection pooling for applications that make frequent SFTP calls. Opening a new SSH connection for each transfer is expensive due to the key exchange handshake. Maintain a pool of pre-authenticated connections and implement health checks to discard stale ones. Paramiko's Transport objects can be kept alive by sending SSH keep-alive messages:
transport.send_ignore() # Sends an SSH_IGNORE packet to keep the connection aliveSchedule these every 30-60 seconds for idle connections, and implement an exponential backoff reconnection strategy for transient failures.
2. Authentication Security
- Prefer key-based authentication over passwords. Ed25519 keys offer the best combination of security and performance.
- Use SSH agent forwarding cautiously—never enable agent forwarding to untrusted servers, as it exposes your authentication socket.
- Rotate host keys and verify them strictly. Paramiko's default
RejectPolicyfor unknown host keys is the safe choice; override only with explicit allowlisting. - Store passphrases in environment variables or secrets managers, never hardcoded in source code.
3. Error Handling and Resilience
SFTP operations can fail for numerous reasons: network interruptions, disk full conditions on the server, permission changes, or server-side resource limits. Implement these patterns:
- Retry with backoff for transient errors (connection refused, timeout, SSHException with "connection lost"). Use
tenacityor a custom retry decorator with jitter. - Validate file integrity after transfer by comparing checksums (SHA-256) computed locally before upload and remotely after download.
- Implement idempotent operations where possible. Upload to a temporary path first (
.tmpsuffix), then atomically rename to the target path. - Handle partial transfers by recording progress state externally, enabling resume across application restarts.
4. Performance Optimization
- Use request pipelining—issue multiple read/write requests before waiting for responses. Paramiko does this internally for
put/getwith appropriate window sizing. - Tune the SSH window size for high-bandwidth, high-latency links. Larger windows allow more in-flight data before requiring acknowledgment.
- Compress selectively—SSH compression helps