"""
FTPS to S3 Operator
Downloads files from FTPS/SFTP servers to S3 (for inbound file processing).
Credentials are retrieved from AWS Secrets Manager.
yuan@ready.net | 2026-01-16
"""
from airflow.models import BaseOperator
from datetime import datetime
from typing import Optional, List
import logging
import fnmatch
log = logging.getLogger(__name__)
[docs]
class FTPSToS3Operator(BaseOperator):
"""
Download files from FTPS/SFTP server to S3.
Supports:
- FTPS (FTP over TLS) - required for MN SWIFT mainframe
- SFTP (SSH FTP) - may be used by other states
Credentials retrieved from AWS Secrets Manager.
:param secret_name: AWS Secrets Manager secret name containing FTPS credentials
:param remote_path: Remote directory path on FTPS server
:param file_pattern: Glob pattern to match files (e.g., "*.IF0411.*")
:param s3_bucket: Destination S3 bucket
:param s3_prefix: Destination S3 prefix (path within bucket)
:param delete_after_download: If True, delete remote file after successful download
**XCom outputs:**
- downloaded_files: List of dicts with file info (name, s3_key, size)
- download_count: Number of files downloaded
- download_timestamp: ISO timestamp when download completed
"""
template_fields = [
"remote_path",
"file_pattern",
"s3_bucket",
"s3_prefix",
"secret_name",
"site_commands",
"use_mainframe_dataset",
"protocol",
]
ui_color = "#5bc0de"
def __init__(
self,
secret_name: str,
remote_path: str,
file_pattern: str,
s3_bucket: str,
s3_prefix: str,
delete_after_download: bool = False,
site_commands: Optional[List[str]] = None,
use_mainframe_dataset: bool = False,
protocol: Optional[str] = None,
**kwargs,
):
super().__init__(**kwargs)
self.secret_name = secret_name
self.remote_path = remote_path
self.file_pattern = file_pattern
self.s3_bucket = s3_bucket
self.s3_prefix = s3_prefix
self.delete_after_download = delete_after_download
self.site_commands = site_commands
self.use_mainframe_dataset = use_mainframe_dataset
# Optional override for the transport protocol. When set, it takes
# precedence over the secret's `extra.protocol` field. Use this when
# the same connection is shared with other operators that don't read
# `extra.protocol` (e.g., S3ToSFTPOperator).
self.protocol = protocol
def _get_secret(self) -> dict:
"""Retrieve FTPS/SFTP credentials from Airflow Connection.
Mirrors S3ToSFTPOperator's fallback pattern: standard Airflow connection
fields (host/login/password) take precedence, with `extra` filling in
missing values and providing SFTP-only material like `private_key`.
Final fallback: read the raw Secrets Manager JSON. BaseHook silently
drops non-standard top-level fields (e.g., `private_key`, `username`)
when the secret isn't shaped like a standard Airflow Connection — the
raw read gives us those back.
"""
from airflow.hooks.base import BaseHook
conn = BaseHook.get_connection(self.secret_name)
extra = conn.extra_dejson or {}
host = conn.host or extra.get("host")
port = conn.port or extra.get("port") or 21
username = conn.login or extra.get("username")
password = conn.password or extra.get("password")
private_key = extra.get("private_key")
protocol = extra.get("protocol", "ftps-explicit")
pgp_private_key = extra.get("pgp_private_key")
pgp_passphrase = extra.get("pgp_passphrase")
if not password and not private_key:
log.info(
f"BaseHook returned no password or private_key for "
f"'{self.secret_name}'; falling back to raw Secrets Manager lookup"
)
try:
import boto3
import json
sm = boto3.client("secretsmanager")
raw = None
for secret_id in (
f"airflow/connections/{self.secret_name}",
self.secret_name,
):
try:
raw = json.loads(
sm.get_secret_value(SecretId=secret_id)["SecretString"]
)
log.info(
f"Loaded raw secret '{secret_id}' "
f"keys={sorted(raw.keys())}"
)
break
except sm.exceptions.ResourceNotFoundException:
continue
if raw:
host = host or raw.get("host")
port = port or raw.get("port") or 21
username = username or raw.get("username")
password = password or raw.get("password")
private_key = private_key or raw.get("private_key")
protocol = raw.get("protocol", protocol)
pgp_private_key = pgp_private_key or raw.get("pgp_private_key")
pgp_passphrase = pgp_passphrase or raw.get("pgp_passphrase")
except Exception as e: # noqa: BLE001 — fallback is best-effort
log.warning(f"Raw Secrets Manager fallback failed: {e}")
return {
"host": host,
"port": int(port) if port else 21,
"username": username,
"password": password,
"private_key": private_key,
"protocol": protocol,
"tls": extra.get("tls", {}),
"ftp": extra.get("ftp", {}),
"mainframe": extra.get("mainframe", {}),
"pgp_private_key": pgp_private_key,
"pgp_passphrase": pgp_passphrase,
}
def _upload_to_s3(self, local_path: str, s3_key: str) -> int:
"""Upload file to S3, return file size."""
import boto3
import os
file_size = os.path.getsize(local_path)
log.info(f"Uploading {local_path} to s3://{self.s3_bucket}/{s3_key}")
s3 = boto3.client("s3")
s3.upload_file(local_path, self.s3_bucket, s3_key)
log.info(f"Uploaded successfully ({file_size} bytes)")
return file_size
def _create_ssl_context(self, tls_config: dict):
"""Create SSL context from TLS configuration.
When verify_server_cert is false, creates a maximally permissive
context with SECLEVEL=0 to support IBM mainframes and other servers
that use older cipher suites (e.g., RSA key exchange without
forward secrecy). OpenSSL 3.x defaults block these at SECLEVEL=1.
"""
import ssl
import tempfile
import os
if not tls_config.get("verify_server_cert", True):
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
ctx.set_ciphers("DEFAULT:@SECLEVEL=0")
return ctx
ctx = ssl.create_default_context()
ctx.check_hostname = True
ctx.verify_mode = ssl.CERT_REQUIRED
ca_pem = tls_config.get("ca_bundle_pem")
if ca_pem:
with tempfile.NamedTemporaryFile(
mode="w", suffix=".pem", delete=False
) as f:
f.write(ca_pem)
ca_file = f.name
ctx.load_verify_locations(ca_file)
os.unlink(ca_file)
return ctx
def _list_and_download_ftps(
self, secret: dict, local_dir: str
) -> List[dict]:
"""List and download matching files via FTPS."""
from ftplib import FTP_TLS
import os
host = secret["host"]
port = secret.get("port", 21)
username = secret["username"]
password = secret["password"]
tls_config = secret.get("tls", {})
ssl_context = self._create_ssl_context(tls_config)
log.info(f"Connecting to FTPS server {host}:{port}")
ftp = FTP_TLS(context=ssl_context)
ftp.connect(host, port)
ftp.login(username, password)
ftp.prot_p()
# Send SITE commands (e.g. for mainframe record format)
if self.site_commands:
for cmd in self.site_commands:
log.info(f"Sending FTP command: {cmd}")
ftp.sendcmd(cmd)
if self.use_mainframe_dataset:
# Mainframe dataset mode: pass quoted pattern to nlst() for z/OS dataset listing.
# z/OS FTP requires single-quoted fully-qualified dataset names and supports
# wildcards in nlst. Without a pattern, nlst() only lists the user's default prefix.
quoted_pattern = f"'{self.file_pattern}'"
log.info(f"Listing mainframe datasets with pattern: {quoted_pattern}")
files = ftp.nlst(quoted_pattern)
# z/OS may return dataset names with surrounding quotes — strip them
files = [f.strip("'") for f in files]
log.info(f"Found {len(files)} datasets on mainframe: {files}")
# Apply local fnmatch as a safety filter (server may return extras)
matching_files = [f for f in files if fnmatch.fnmatch(f, self.file_pattern)]
log.info(
f"Found {len(matching_files)} datasets matching pattern '{self.file_pattern}'"
)
downloaded = []
for filename in matching_files:
local_path = os.path.join(local_dir, filename)
log.info(f"Downloading dataset '{filename}'")
# Use text mode (retrlines) for mainframe — triggers server-side
# EBCDIC→ASCII conversion. Binary mode (retrbinary) would preserve
# raw EBCDIC encoding which downstream parsers can't read.
lines = []
ftp.retrlines(f"RETR '{filename}'", lines.append)
with open(local_path, "w", encoding="utf-8") as f:
f.write("\n".join(lines) + "\n")
file_size = os.path.getsize(local_path)
s3_key = f"{self.s3_prefix.rstrip('/')}/{filename}"
# Upload to S3
self._upload_to_s3(local_path, s3_key)
downloaded.append(
{
"name": filename,
"s3_key": s3_key,
"s3_bucket": self.s3_bucket,
"size": file_size,
"local_path": local_path,
}
)
# Delete from remote if configured
if self.delete_after_download:
log.info(f"Deleting remote dataset: '{filename}'")
ftp.delete(f"'{filename}'")
else:
# Standard directory-based FTP mode
if self.remote_path:
log.info(f"Changing to directory: {self.remote_path}")
ftp.cwd(self.remote_path)
# List files
files = ftp.nlst()
log.info(f"Found {len(files)} files in remote directory")
# Filter by pattern
matching_files = [f for f in files if fnmatch.fnmatch(f, self.file_pattern)]
log.info(
f"Found {len(matching_files)} files matching pattern '{self.file_pattern}'"
)
downloaded = []
for filename in matching_files:
local_path = os.path.join(local_dir, filename)
log.info(f"Downloading {filename}")
with open(local_path, "wb") as f:
ftp.retrbinary(f"RETR {filename}", f.write)
file_size = os.path.getsize(local_path)
s3_key = f"{self.s3_prefix.rstrip('/')}/{filename}"
# Upload to S3
self._upload_to_s3(local_path, s3_key)
downloaded.append(
{
"name": filename,
"s3_key": s3_key,
"s3_bucket": self.s3_bucket,
"size": file_size,
"local_path": local_path,
}
)
# Delete from remote if configured
if self.delete_after_download:
log.info(f"Deleting remote file: {filename}")
ftp.delete(filename)
ftp.quit()
return downloaded
def _list_and_download_sftp(
self, secret: dict, local_dir: str
) -> List[dict]:
"""List and download matching files via SFTP."""
import paramiko
import os
import stat
host = secret["host"]
port = secret.get("port", 22)
username = secret["username"]
password = secret.get("password")
private_key = secret.get("private_key")
log.info(f"Connecting to SFTP server {host}:{port}")
transport = paramiko.Transport((host, port))
if private_key:
from io import StringIO
import re
# Normalize private key to valid PEM format.
# Mirrors S3ToSFTPOperator._upload_to_sftp_from_memory so both
# operators handle the same single-line / escaped-newline storage
# format used in AWS Secrets Manager.
raw_key = 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"
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)
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=username, pkey=pkey)
else:
transport.connect(username=username, password=password)
sftp = paramiko.SFTPClient.from_transport(transport)
# List files in remote directory
remote_dir = self.remote_path or "."
files = sftp.listdir(remote_dir)
log.info(f"Found {len(files)} files in remote directory")
# Filter by pattern
matching_files = [f for f in files if fnmatch.fnmatch(f, self.file_pattern)]
log.info(
f"Found {len(matching_files)} files matching pattern '{self.file_pattern}'"
)
downloaded = []
for filename in matching_files:
remote_file = f"{remote_dir.rstrip('/')}/{filename}"
local_path = os.path.join(local_dir, filename)
log.info(f"Downloading {filename}")
sftp.get(remote_file, local_path)
file_size = os.path.getsize(local_path)
s3_key = f"{self.s3_prefix.rstrip('/')}/{filename}"
# Upload to S3
self._upload_to_s3(local_path, s3_key)
downloaded.append(
{
"name": filename,
"s3_key": s3_key,
"s3_bucket": self.s3_bucket,
"size": file_size,
"local_path": local_path,
}
)
# Delete from remote if configured
if self.delete_after_download:
log.info(f"Deleting remote file: {filename}")
sftp.remove(remote_file)
sftp.close()
transport.close()
return downloaded
[docs]
def execute(self, context):
import tempfile
secret = self._get_secret()
# Caller-supplied protocol wins so dags can drive transport choice
# from state config rather than relying on the secret's extras.
protocol = self.protocol or secret.get("protocol", "ftps-explicit")
log.info(
f"FTPSToS3Operator: secret={self.secret_name} protocol={protocol} "
f"(override={self.protocol!r}, secret_default={secret.get('protocol')!r})"
)
with tempfile.TemporaryDirectory() as tmpdir:
if protocol in ("ftps-explicit", "ftps", "ftp-tls"):
downloaded = self._list_and_download_ftps(secret, tmpdir)
elif protocol in ("sftp", "ssh"):
downloaded = self._list_and_download_sftp(secret, tmpdir)
else:
raise ValueError(f"Unsupported protocol: {protocol}")
download_timestamp = datetime.utcnow().isoformat()
log.info(f"Downloaded {len(downloaded)} files")
context["ti"].xcom_push(key="downloaded_files", value=downloaded)
context["ti"].xcom_push(key="download_count", value=len(downloaded))
context["ti"].xcom_push(key="download_timestamp", value=download_timestamp)
return {
"downloaded_files": downloaded,
"download_count": len(downloaded),
"download_timestamp": download_timestamp,
}