"""
S3 to FTPS Operator
Transfers files from S3 to FTPS servers (e.g., MN SWIFT mainframe).
Credentials are retrieved via Airflow Connections (BaseHook.get_connection),
which resolves through the configured secrets backend (e.g., AWS Secrets Manager).
yuan@ready.net | 2026-01-09
"""
from airflow.hooks.base import BaseHook
from airflow.models import BaseOperator
from datetime import datetime
from typing import Optional
import logging
import re
log = logging.getLogger(__name__)
[docs]
class S3ToFTPSOperator(BaseOperator):
"""
Transfer file from S3 to FTPS server.
Supports:
- FTPS (FTP over TLS) - required for MN SWIFT mainframe
- SFTP (SSH FTP) - may be used by other states
Credentials retrieved via Airflow Connections (BaseHook.get_connection).
FTPS-specific config is stored in the connection's ``extra`` JSON field.
Airflow Connection fields used:
- host: FTPS server hostname
- login: Username
- password: Password
- port: Server port (default: 21)
- extra (JSON): {
"protocol": "ftps-explicit",
"remote_path": "/base/path",
"tls": {
"verify_server_cert": false,
"ca_bundle_pem": "-----BEGIN CERTIFICATE-----..."
},
"ftp": {
"post_login_commands": ["SITE LRECL=644", "SITE RECFM=FB"],
"transfer_mode": "binary"
},
"mainframe": {
"remote_dataset": "'ADM.MB.UAT1.IBPV.B220701.IF0407.T001'"
}
}
:param s3_bucket: S3 bucket name
:param s3_key: S3 object key (path within bucket)
:param secret_name: Airflow connection ID for FTPS credentials
:param remote_path: Remote path on FTPS server (overrides mainframe.remote_dataset if provided)
:param use_mainframe_dataset: If True, use mainframe.remote_dataset from secret (default: True)
:param dataset_sequence: Zero-padded transaction sequence ("001", "002", ...) used to
vary the FINAL qualifier of the mainframe dataset name, so a run that produces
several files writes each to its own dataset instead of overwriting one.
``mainframe.remote_dataset`` in the connection supplies the prefix and the
environment qualifier; only the trailing ``.Tnnn`` is substituted. Omit to keep
the configured dataset name verbatim (previous behaviour).
**XCom outputs:**
- transfer_status: SUCCESS/FAILURE
- transfer_response: Response details from server
- transfer_timestamp: ISO timestamp when transfer completed
"""
template_fields = ["s3_bucket", "s3_key", "remote_path", "secret_name"]
ui_color = "#f0ad4e"
def __init__(
self,
s3_bucket: str,
s3_key: str,
secret_name: str,
remote_path: Optional[str] = None,
use_mainframe_dataset: bool = True,
dataset_sequence: Optional[str] = None,
**kwargs,
):
super().__init__(**kwargs)
self.s3_bucket = s3_bucket
self.s3_key = s3_key
self.secret_name = secret_name
self.remote_path = remote_path
self.use_mainframe_dataset = use_mainframe_dataset
self.dataset_sequence = dataset_sequence
def _resolve_mainframe_dataset(self, mainframe_config: dict) -> Optional[str]:
"""Resolve the target dataset, substituting the trailing sequence qualifier.
``mainframe.remote_dataset`` is a fully-qualified, single-quoted z/OS name
such as ``'ADM.MB.UAT1.IBPV.B220701.IF0407.T001'``. It is a single fixed
value, so every file in a multi-file run previously STOR'd over the same
dataset and only the last one survived.
Only the final ``.Tnnn`` qualifier is rewritten — the high-level qualifiers
and the environment (UAT1 vs PROD) still come from the connection, so this
cannot silently retarget a different environment. If the configured name
does not end in ``.Tnnn`` the sequence is appended instead.
"""
base = mainframe_config.get("remote_dataset")
if not base or not self.dataset_sequence:
return base
unquoted = base.strip().strip("'")
if re.search(r"\.T\d+$", unquoted):
resolved = re.sub(r"\.T\d+$", f".T{self.dataset_sequence}", unquoted)
else:
resolved = f"{unquoted}.T{self.dataset_sequence}"
return f"'{resolved}'"
def _get_secret(self) -> dict:
"""Retrieve FTPS credentials from Airflow Connection."""
conn = BaseHook.get_connection(self.secret_name)
extra = conn.extra_dejson or {}
return {
"host": conn.host,
"port": conn.port or 21,
"username": conn.login,
"password": conn.password,
"protocol": extra.get("protocol", "ftps-explicit"),
"tls": extra.get("tls", {}),
"ftp": extra.get("ftp", {}),
"mainframe": extra.get("mainframe", {}),
"remote_path": extra.get("remote_path", ""),
}
def _download_from_s3(self, local_path: str) -> None:
"""Download file from S3 to local path."""
import boto3
log.info(f"Downloading s3://{self.s3_bucket}/{self.s3_key} to {local_path}")
s3 = boto3.client("s3")
s3.download_file(self.s3_bucket, self.s3_key, local_path)
log.info(f"Downloaded successfully")
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 _upload_ftps(self, local_path: str, secret: dict) -> dict:
"""Upload file via FTPS (FTP over TLS)."""
from ftplib import FTP_TLS
host = secret["host"]
port = secret.get("port", 21)
username = secret["username"]
password = secret["password"]
tls_config = secret.get("tls", {})
ftp_config = secret.get("ftp", {})
mainframe_config = secret.get("mainframe", {})
remote_dest = self.remote_path
if not remote_dest and self.use_mainframe_dataset:
remote_dest = self._resolve_mainframe_dataset(mainframe_config)
if not remote_dest:
raise ValueError(
"No remote_path provided and mainframe.remote_dataset not in secret"
)
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()
post_login_cmds = ftp_config.get("post_login_commands", [])
for cmd in post_login_cmds:
log.info(f"Executing post-login command: {cmd}")
ftp.sendcmd(cmd)
transfer_mode = ftp_config.get("transfer_mode", "binary")
log.info(f"Uploading to {remote_dest} (mode: {transfer_mode})")
with open(local_path, "rb") as f:
if transfer_mode == "binary":
response = ftp.storbinary(f"STOR {remote_dest}", f)
else:
response = ftp.storlines(f"STOR {remote_dest}", f)
log.info(f"Upload response: {response}")
ftp.quit()
return {
"response": response,
"remote_path": remote_dest,
"host": host,
}
def _upload_sftp(self, local_path: str, secret: dict) -> dict:
"""Upload file via SFTP (SSH FTP)."""
import paramiko
host = secret["host"]
port = secret.get("port", 22)
username = secret["username"]
password = secret.get("password")
private_key = secret.get("private_key")
remote_dest = self.remote_path
if not remote_dest:
raise ValueError("remote_path is required for SFTP")
log.info(f"Connecting to SFTP server {host}:{port}")
transport = paramiko.Transport((host, port))
if private_key:
from io import StringIO
pkey = paramiko.RSAKey.from_private_key(StringIO(private_key))
transport.connect(username=username, pkey=pkey)
else:
transport.connect(username=username, password=password)
sftp = paramiko.SFTPClient.from_transport(transport)
log.info(f"Uploading to {remote_dest}")
sftp.put(local_path, remote_dest)
sftp.close()
transport.close()
return {
"response": "Transfer completed",
"remote_path": remote_dest,
"host": host,
}
[docs]
def execute(self, context):
import tempfile
import os
secret = self._get_secret()
protocol = secret.get("protocol", "ftps-explicit")
with tempfile.TemporaryDirectory() as tmpdir:
filename = os.path.basename(self.s3_key)
local_path = os.path.join(tmpdir, filename)
self._download_from_s3(local_path)
try:
if protocol in ("ftps-explicit", "ftps", "ftp-tls"):
result = self._upload_ftps(local_path, secret)
elif protocol in ("sftp", "ssh"):
result = self._upload_sftp(local_path, secret)
else:
raise ValueError(f"Unsupported protocol: {protocol}")
transfer_status = "SUCCESS"
transfer_response = result
except Exception as e:
log.error(f"Transfer failed: {e}")
transfer_status = "FAILURE"
transfer_response = {"error": str(e)}
raise
finally:
transfer_timestamp = datetime.utcnow().isoformat()
context["ti"].xcom_push(key="transfer_status", value=transfer_status)
context["ti"].xcom_push(
key="transfer_response", value=transfer_response
)
context["ti"].xcom_push(
key="transfer_timestamp", value=transfer_timestamp
)
log.info(f"Transfer {transfer_status}: {transfer_response}")
return {
"transfer_status": transfer_status,
"transfer_response": transfer_response,
"transfer_timestamp": transfer_timestamp,
}