"""This operator will transfer a file from an s3 bucket to an SFTP server using paramiko."""
from airflow.models import BaseOperator
from airflow.hooks.base import BaseHook
[docs]
class S3ToSFTPOperator(BaseOperator):
"""
Operator to transfer a file from an S3 bucket to an SFTP server.
:param s3_file_uri: the S3 URI of the file to transfer (e.g., s3://bucket-name/path/to/file).
:type s3_file_uri: str
:param sftp_secret_name: the name of the secret containing SFTP credentials.
:type sftp_secret_name: str
:param sftp_file_path: the path on the SFTP server where the file will be uploaded.
:type sftp_file_path: str (default will be root of the SFTP server)
:param create_remote_dirs: whether to create directories on SFTP if they don't exist
:type create_remote_dirs: bool (default True)
"""
template_fields = (
"s3_file_uri",
"sftp_secret_name",
"sftp_file_path",
"secret_host_key",
"secret_username_key",
"secret_password_key",
)
ui_color = "#5ecede"
def __init__(
self,
s3_file_uri: str,
sftp_secret_name: str,
sftp_file_path: str = "/",
secret_host_key: str = "host",
secret_username_key: str = "username",
secret_password_key: str = "password",
port: int = 22,
timeout: int = 30,
create_intermediate_dirs: bool = True, # Deprecated
create_remote_dirs: bool = True,
*args,
**kwargs,
):
super(S3ToSFTPOperator, self).__init__(*args, **kwargs)
self.s3_file_uri = s3_file_uri
self.sftp_secret_name = sftp_secret_name
self.secret_host_key = secret_host_key
self.secret_username_key = secret_username_key
self.secret_password_key = secret_password_key
self.port = port
self.timeout = timeout
self.create_intermediate_dirs = create_intermediate_dirs
self.create_remote_dirs = create_remote_dirs or create_intermediate_dirs
self.sftp_file_path = sftp_file_path
def _parse_s3_uri(self, uri) -> tuple[str, str]:
if not uri.startswith("s3://"):
raise ValueError("Invalid S3 URI (must start with s3://)")
bucket, key = uri[5:].split("/", 1)
if not bucket or not key:
raise ValueError("S3 URI must include bucket and key")
return bucket, key
def _get_secret(self) -> dict:
"""Get SFTP credentials from Airflow Connection.
Supports both password and private key authentication.
Fields are resolved from standard Airflow connection properties first,
then fall back to the connection's extra JSON (for secrets backends
that store non-standard field names like ``username`` instead of ``login``).
"""
try:
conn = BaseHook.get_connection(self.sftp_secret_name)
extra = conn.extra_dejson or {}
# Debug: log available fields (keys only, no values)
self.log.info(
f"Connection '{self.sftp_secret_name}' fields: "
f"host={'set' if conn.host else 'empty'}, "
f"login={'set' if conn.login else 'empty'}, "
f"password={'set' if conn.password else 'empty'}, "
f"port={conn.port}, "
f"extra_keys={list(extra.keys())}"
)
host = conn.host or extra.get("host")
username = conn.login or extra.get("username")
password = conn.password or extra.get("password")
private_key = extra.get("private_key")
port = conn.port or extra.get("port", self.port)
# Fallback: read raw secret from Secrets Manager when
# BaseHook drops non-standard fields (e.g., private_key)
if not password and not private_key:
import boto3
import json
self.log.info(
f"No password or private_key from BaseHook, "
f"falling back to raw Secrets Manager lookup"
)
sm = boto3.client("secretsmanager")
raw = None
for secret_id in [
f"airflow/connections/{self.sftp_secret_name}",
self.sftp_secret_name,
]:
try:
raw = json.loads(
sm.get_secret_value(SecretId=secret_id)["SecretString"]
)
self.log.info(
f"Found secret at '{secret_id}', "
f"keys={list(raw.keys())}"
)
break
except sm.exceptions.ResourceNotFoundException:
self.log.info(f"Secret '{secret_id}' not found, trying next")
continue
if raw:
private_key = raw.get("private_key")
password = raw.get("password")
host = host or raw.get("host")
username = username or raw.get("username")
port = port or raw.get("port", self.port)
if not host or not username:
raise ValueError(
f"Connection '{self.sftp_secret_name}' must contain host and username/login fields."
)
if not password and not private_key:
raise ValueError(
f"Connection '{self.sftp_secret_name}' must contain either password or private_key."
)
return {
"host": host,
"username": username,
"password": password,
"private_key": private_key,
"port": int(port) if port else self.port,
}
except Exception as e:
self.log.error(f"Failed to get connection: {e}")
raise
def _download_from_s3_to_memory(self, bucket: str, key: str):
"""Download file from S3 into memory."""
import boto3
import io
s3 = boto3.client("s3")
buf = io.BytesIO()
s3.download_fileobj(Bucket=bucket, Key=key, Fileobj=buf)
buf.seek(0)
return buf
def _resolve_remote_path(self, key: str) -> str:
"""
Resolve the final remote path for the uploaded file.
Handles both file paths and directory paths intelligently.
"""
import os
# Extract just the filename from the S3 key
filename = os.path.basename(key.rstrip("/"))
# If sftp_file_path already ends with the same filename, use as-is
if self.sftp_file_path.endswith(filename):
return self.sftp_file_path
# If sftp_file_path ends with '/', it's clearly a directory
if self.sftp_file_path.endswith("/"):
return f"{self.sftp_file_path}{filename}"
# Check if sftp_file_path looks like a file (has an extension)
path_basename = os.path.basename(self.sftp_file_path)
if "." in path_basename and not path_basename.startswith("."):
# It looks like a file path, use as-is
self.log.debug(f"Using {self.sftp_file_path} as full file path")
return self.sftp_file_path
# Otherwise, treat it as a directory and append filename
self.log.debug(
f"Treating {self.sftp_file_path} as directory, appending {filename}"
)
return f"{self.sftp_file_path}/{filename}"
def _ensure_remote_directory(self, sftp, remote_dir: str) -> None:
"""Ensure the remote directory exists, creating nested directories if needed."""
# Normalize the path
import os
remote_dir = remote_dir.rstrip("/")
if not remote_dir or remote_dir == ".":
return
# Try to stat the directory first
try:
sftp.stat(remote_dir)
self.log.debug(f"Directory already exists: {remote_dir}")
return
except IOError:
# Directory doesn't exist, we need to create it
self.log.info(f"Directory doesn't exist, will create: {remote_dir}")
# Split the path into parts and create each level
parts = []
head = remote_dir
while head and head != "/":
head, tail = os.path.split(head)
if tail:
parts.append(tail)
parts.reverse()
# Start from root or current directory
if remote_dir.startswith("/"):
current_path = "/"
else:
current_path = "."
# Create each directory level
for part in parts:
if current_path == "/":
current_path = f"/{part}"
else:
current_path = f"{current_path}/{part}"
try:
# Check if this level exists
sftp.stat(current_path)
self.log.debug(f"Directory level exists: {current_path}")
except IOError:
# This level doesn't exist, create it
try:
sftp.mkdir(current_path)
self.log.info(f"Created directory: {current_path}")
except IOError as e:
# Handle specific errors
if "permission denied" in str(e).lower():
raise PermissionError(
f"Permission denied creating directory '{current_path}'. "
f"Please ensure the SFTP user has write permissions."
)
elif (
"file exists" in str(e).lower()
or "folder with the same name" in str(e).lower()
):
# Directory exists (race condition or Egnyte-specific error)
self.log.debug(f"Directory already exists: {current_path}")
continue
else:
# Re-raise other errors
raise IOError(
f"Failed to create directory '{current_path}': {e}"
)
def _upload_to_sftp_from_memory(
self, secret: dict, file_stream, remote_path: str
) -> None:
"""Upload file from memory to SFTP server, creating directories if needed."""
import paramiko
import os
port = secret.get("port", self.port)
transport = paramiko.Transport((secret["host"], port))
try:
if secret.get("private_key"):
from io import StringIO
import re
# Normalize private key to valid PEM format
raw_key = secret["private_key"].replace("\\n", "\n")
# Handle keys stored as single line with spaces instead of newlines
pem_match = re.match(
r"(-----BEGIN [^-]+-----)\s*(.*?)\s*(-----END [^-]+-----)",
raw_key,
re.DOTALL,
)
if pem_match and "\n" not in raw_key.strip():
header, body, footer = pem_match.groups()
body = body.replace(" ", "")
raw_key = f"{header}\n{body}\n{footer}\n"
self.log.info("Reconstructed PEM format from single-line key")
key_file = StringIO(raw_key)
pkey = None
for key_class in [
paramiko.RSAKey,
paramiko.Ed25519Key,
paramiko.ECDSAKey,
paramiko.DSSKey,
]:
try:
key_file.seek(0)
pkey = key_class.from_private_key(key_file)
self.log.info(f"Parsed private key as {key_class.__name__}")
break
except (paramiko.ssh_exception.SSHException, ValueError):
continue
if pkey is None:
raise ValueError(
"Could not parse private key with any supported type "
"(RSA, Ed25519, ECDSA, DSS)"
)
transport.connect(username=secret["username"], pkey=pkey)
else:
transport.connect(
username=secret["username"],
password=secret["password"],
hostkey=None,
)
sftp = paramiko.SFTPClient.from_transport(transport)
try:
# Diagnostic: inspect the SFTP environment
try:
cwd = sftp.normalize(".")
self.log.info(f"SFTP cwd: {cwd}")
dir_listing = sftp.listdir(".")
self.log.info(f"SFTP listing ({len(dir_listing)} items): {dir_listing[:20]}")
except Exception as diag_e:
self.log.warning(f"SFTP diagnostic failed: {diag_e}")
# Extract directory path from the full file path
remote_dir = os.path.dirname(remote_path)
if remote_dir and remote_dir != "/" and self.create_remote_dirs:
# Check if the directory exists, create if it doesn't
self._ensure_remote_directory(sftp, remote_dir)
elif remote_dir and remote_dir != "/" and not self.create_remote_dirs:
# Just check if directory exists, don't create
try:
sftp.stat(remote_dir)
except IOError:
raise FileNotFoundError(
f"Remote directory '{remote_dir}' does not exist and create_remote_dirs=False"
)
# Navigate to target directory
upload_dir = os.path.dirname(remote_path)
upload_filename = os.path.basename(remote_path)
if upload_dir:
self.log.info(f"Changing to remote directory: {upload_dir}")
sftp.chdir(upload_dir)
self.log.info(f"Now in: {sftp.normalize('.')}")
# Write stream to temp file, then use sftp.put()
import tempfile
file_stream.seek(0)
with tempfile.NamedTemporaryFile(delete=False) as tmp:
tmp.write(file_stream.read())
tmp_path = tmp.name
try:
self.log.info(f"Uploading {upload_filename} via put()")
try:
sftp.put(tmp_path, upload_filename)
except (IOError, FileNotFoundError) as put_err:
# Some SFTP servers reject paramiko's CMD_OPEN because it
# sends empty file attributes (no permissions). The system
# sftp command includes permissions and works on these servers.
self.log.warning(
f"paramiko put() failed ({put_err}), "
f"falling back to system sftp command"
)
abs_remote_dir = sftp.normalize(".")
self._upload_via_system_sftp(
secret, tmp_path, abs_remote_dir, upload_filename
)
self.log.info(f"Successfully uploaded file to {remote_path}")
finally:
os.unlink(tmp_path)
finally:
sftp.close()
finally:
transport.close()
def _upload_via_system_sftp(
self,
secret: dict,
local_path: str,
remote_dir: str,
remote_filename: str,
) -> None:
"""Upload a file using the system sftp command.
Fallback for SFTP servers that reject paramiko's CMD_OPEN
(e.g., servers that require explicit file permissions in the
open request, which paramiko does not send).
"""
import subprocess
import tempfile
import re
import os
if not secret.get("private_key"):
raise ValueError(
"System sftp fallback requires private_key authentication "
"(password-only connections are not supported)"
)
raw_key = secret["private_key"].replace("\\n", "\n")
# Normalize to proper PEM format with 64-char base64 line wrapping.
# The key may be stored as a single line (no newlines in body),
# which paramiko tolerates but OpenSSH's libcrypto rejects.
pem_match = re.match(
r"(-----BEGIN [^-]+-----)\s*(.*?)\s*(-----END [^-]+-----)",
raw_key,
re.DOTALL,
)
if pem_match:
header, body, footer = pem_match.groups()
body = re.sub(r"\s+", "", body) # strip all whitespace
wrapped = "\n".join(
body[i : i + 64] for i in range(0, len(body), 64)
)
raw_key = f"{header}\n{wrapped}\n{footer}\n"
# Write private key to temp file with secure permissions
key_fd, key_path = tempfile.mkstemp(suffix=".pem")
try:
os.write(key_fd, raw_key.encode())
os.close(key_fd)
os.chmod(key_path, 0o600)
# Write sftp batch commands
batch_fd, batch_path = tempfile.mkstemp(suffix=".sftp")
try:
commands = f"cd {remote_dir}\nput {local_path} {remote_filename}\n"
os.write(batch_fd, commands.encode())
os.close(batch_fd)
port = str(secret.get("port", self.port))
cmd = [
"sftp",
"-i", key_path,
"-P", port,
"-oBatchMode=yes",
"-oStrictHostKeyChecking=no",
"-b", batch_path,
f"{secret['username']}@{secret['host']}",
]
self.log.info(
f"Running system sftp to {secret['host']}:"
f"{remote_dir}/{remote_filename}"
)
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=120,
)
if result.returncode != 0:
self.log.error(f"sftp stderr: {result.stderr}")
raise IOError(
f"System sftp failed (rc={result.returncode}): "
f"{result.stderr.strip()}"
)
self.log.info("System sftp upload successful")
finally:
os.unlink(batch_path)
finally:
os.unlink(key_path)
[docs]
def execute(self, context):
self.log.info("Starting in-memory S3 -> SFTP transfer")
bucket, key = self._parse_s3_uri(self.s3_file_uri)
secret = self._get_secret()
self.log.info("Downloading from S3: s3://%s/%s", bucket, key)
buf = self._download_from_s3_to_memory(bucket, key)
remote_path = self._resolve_remote_path(key)
self.log.info(
"Uploading to SFTP: %s@%s:%s",
secret["username"],
secret["host"],
remote_path,
)
self._upload_to_sftp_from_memory(secret, buf, remote_path)
self.log.info("Transfer completed successfully")
return remote_path