"""File transfer adapters behind the one "file_transfer" Connector type — SFTP (paramiko), FTP and FTPS (explicit TLS, ftplib). Task Designer's "File Transfer" node lists, downloads, uploads, moves or deletes files on the server. Every operation opens one session and closes it again. Paths on the node are relative to the connector's base folder unless they start with "/". SFTP host keys: when the connector has a pinned fingerprint ("SHA256:…"), a server presenting any other key is refused before the password is sent. Without one the key is accepted — the Test button shows the server's fingerprint so it can be pinned. """ import asyncio import base64 import fnmatch import ftplib import hashlib import io import posixpath import stat from dataclasses import dataclass from datetime import datetime, timezone from typing import Any MAX_FILE_BYTES = 50 * 1024 * 1024 MAX_FILES = 20 TIMEOUT = 30 class FileTransferError(Exception): pass @dataclass class RemoteFile: name: str path: str size: int | None modified: str | None is_dir: bool def as_dict(self) -> dict: return {"name": self.name, "path": self.path, "size": self.size, "modified": self.modified, "is_dir": self.is_dir} def _settings(config: dict, default_port: int) -> tuple[str, int, str, str]: host = (config.get("host") or "").strip() user = (config.get("username") or "").strip() if not host or not user: raise FileTransferError("the connector needs the host and username") try: port = int(config.get("port") or default_port) except ValueError as exc: raise FileTransferError("the port must be a number") from exc return host, port, user, config.get("password") or "" def resolve(config: dict, path: str) -> str: """A node's path against the connector's base folder.""" base = (config.get("base_dir") or "/").strip() or "/" path = (path or "").strip() return posixpath.normpath(path if path.startswith("/") else posixpath.join(base, path)) if path else posixpath.normpath(base) def host_key_fingerprint(key_bytes: bytes) -> str: return "SHA256:" + base64.b64encode(hashlib.sha256(key_bytes).digest()).decode().rstrip("=") # ---- SFTP -------------------------------------------------------------------------------------- class _Sftp: def __init__(self, config: dict): import paramiko host, port, user, password = _settings(config, 22) self.transport = paramiko.Transport((host, port)) self.transport.banner_timeout = TIMEOUT try: self.transport.start_client(timeout=TIMEOUT) key = self.transport.get_remote_server_key() self.fingerprint = host_key_fingerprint(key.asbytes()) pinned = (config.get("host_key_fingerprint") or "").strip() if pinned and pinned.rstrip("=") != self.fingerprint: raise FileTransferError(f"the server's host key {self.fingerprint} doesn't match the pinned fingerprint — refusing to connect") private_key = (config.get("private_key") or "").strip() if private_key: self.transport.auth_publickey(user, self._load_key(private_key, config.get("key_passphrase") or None)) else: self.transport.auth_password(user, password) if not self.transport.is_authenticated(): raise FileTransferError("the SFTP server rejected the login") self.sftp = paramiko.SFTPClient.from_transport(self.transport) self.sftp.get_channel().settimeout(TIMEOUT) except paramiko.AuthenticationException as exc: self.transport.close() raise FileTransferError(f"the SFTP server rejected the login: {exc}") from exc except Exception: self.transport.close() raise @staticmethod def _load_key(text: str, passphrase: str | None): import paramiko for cls in (paramiko.Ed25519Key, paramiko.ECDSAKey, paramiko.RSAKey): try: return cls.from_private_key(io.StringIO(text), password=passphrase) except paramiko.PasswordRequiredException as exc: raise FileTransferError("the private key is encrypted — add its passphrase") from exc except paramiko.SSHException: continue raise FileTransferError("couldn't read the private key — paste an OpenSSH or PEM private key (Ed25519, ECDSA or RSA)") def list(self, path: str) -> list[RemoteFile]: out = [] for a in self.sftp.listdir_attr(path): modified = datetime.fromtimestamp(a.st_mtime, timezone.utc).isoformat() if a.st_mtime else None out.append(RemoteFile(a.filename, posixpath.join(path, a.filename), a.st_size, modified, stat.S_ISDIR(a.st_mode or 0))) return out def is_dir(self, path: str) -> bool: try: return stat.S_ISDIR(self.sftp.stat(path).st_mode or 0) except FileNotFoundError: return False def get(self, path: str) -> bytes: size = self.sftp.stat(path).st_size or 0 if size > MAX_FILE_BYTES: raise FileTransferError(f"{path} is larger than {MAX_FILE_BYTES // (1024 * 1024)} MB") buf = io.BytesIO() self.sftp.getfo(path, buf) return buf.getvalue() def put(self, path: str, data: bytes) -> None: self.sftp.putfo(io.BytesIO(data), path) def rename(self, src: str, dst: str) -> None: self.sftp.posix_rename(src, dst) def delete(self, path: str) -> None: self.sftp.remove(path) def close(self) -> None: self.transport.close() # ---- FTP / FTPS ---------------------------------------------------------------------------------- class _Ftp: def __init__(self, config: dict, tls: bool): host, port, user, password = _settings(config, 21) self.ftp = ftplib.FTP_TLS(timeout=TIMEOUT) if tls else ftplib.FTP(timeout=TIMEOUT) self.fingerprint = None try: self.ftp.connect(host, port) self.ftp.login(user, password) if tls: self.ftp.prot_p() # encrypt the data connection too self.ftp.set_pasv((config.get("passive") or "yes") != "no") except ftplib.error_perm as exc: self.ftp.close() raise FileTransferError(f"the FTP server rejected the login: {exc}") from exc except Exception: self.ftp.close() raise def list(self, path: str) -> list[RemoteFile]: try: out = [] for name, facts in self.ftp.mlsd(path, facts=["type", "size", "modify"]): if name in (".", "..") or facts.get("type") in ("cdir", "pdir"): continue modified = None if m := facts.get("modify"): modified = datetime.strptime(m[:14], "%Y%m%d%H%M%S").replace(tzinfo=timezone.utc).isoformat() size = int(facts["size"]) if facts.get("size", "").isdigit() else None out.append(RemoteFile(name, posixpath.join(path, name), size, modified, facts.get("type") == "dir")) return out except ftplib.error_perm: # Older servers without MLSD: names only. return [RemoteFile(posixpath.basename(n), posixpath.join(path, posixpath.basename(n)), None, None, False) for n in self.ftp.nlst(path)] def is_dir(self, path: str) -> bool: current = self.ftp.pwd() try: self.ftp.cwd(path) return True except ftplib.error_perm: return False finally: self.ftp.cwd(current) def get(self, path: str) -> bytes: buf = io.BytesIO() def write(chunk: bytes) -> None: if buf.tell() + len(chunk) > MAX_FILE_BYTES: raise FileTransferError(f"{path} is larger than {MAX_FILE_BYTES // (1024 * 1024)} MB") buf.write(chunk) self.ftp.retrbinary(f"RETR {path}", write) return buf.getvalue() def put(self, path: str, data: bytes) -> None: self.ftp.storbinary(f"STOR {path}", io.BytesIO(data)) def rename(self, src: str, dst: str) -> None: self.ftp.rename(src, dst) def delete(self, path: str) -> None: self.ftp.delete(path) def close(self) -> None: try: self.ftp.quit() except Exception: # noqa: BLE001 - already closed / connection dropped self.ftp.close() def _open(config: dict): provider = config.get("provider") if provider == "sftp": return _Sftp(config) if provider in ("ftp", "ftps"): return _Ftp(config, tls=provider == "ftps") raise FileTransferError("choose SFTP, FTP or FTPS on the connector") # ---- Operations (sync; called through asyncio.to_thread) --------------------------------------- def _targets(session, path: str, pattern: str) -> list[RemoteFile]: """A pattern means "matching files in the folder at path"; no pattern means path itself is one file.""" if pattern: files = [f for f in session.list(path) if not f.is_dir and fnmatch.fnmatch(f.name, pattern)] return sorted(files, key=lambda f: f.name) return [RemoteFile(posixpath.basename(path), path, None, None, False)] def _run_sync(config: dict, operation: str, remote_path: str, pattern: str, target: str, upload: tuple[str, bytes] | None) -> dict: path = resolve(config, remote_path) session = _open(config) try: if operation == "list": files = [f for f in session.list(path) if not pattern or fnmatch.fnmatch(f.name, pattern)] files.sort(key=lambda f: (not f.is_dir, f.name)) return {"files": [f.as_dict() for f in files], "names": [f.name for f in files], "count": len(files), "folder": path} if operation == "download": targets = _targets(session, path, pattern) if len(targets) > MAX_FILES: raise FileTransferError(f"{len(targets)} files match — narrow the pattern (at most {MAX_FILES} per run)") return {"downloads": [(f.name, f.path, session.get(f.path)) for f in targets], "folder": path} if operation == "upload": filename, data = upload or ("", b"") # A "save as" name means remote_path is the folder; otherwise it is # a folder if one exists there, else the full destination path. if target: dest = posixpath.join(path, target) else: dest = posixpath.join(path, filename) if session.is_dir(path) else path session.put(dest, data) return {"remote_path": dest, "size": len(data)} if operation == "move": if not target: raise FileTransferError("set where to move the file(s) to") dest = resolve(config, target) targets = _targets(session, path, pattern) into_folder = target.endswith("/") or len(targets) > 1 or session.is_dir(dest) moved = [] for f in targets: new_path = posixpath.join(dest, f.name) if into_folder else dest session.rename(f.path, new_path) moved.append(new_path) return {"moved": moved, "count": len(moved), "remote_path": moved[0] if moved else ""} if operation == "delete": targets = _targets(session, path, pattern) for f in targets: session.delete(f.path) return {"deleted": [f.path for f in targets], "count": len(targets)} raise FileTransferError(f"unknown operation '{operation}'") finally: session.close() async def run_operation(config: dict, operation: str, remote_path: str, *, pattern: str = "", target: str = "", upload: tuple[str, bytes] | None = None) -> dict: return await asyncio.wait_for(asyncio.to_thread(_run_sync, config, operation, remote_path, pattern.strip(), target.strip(), upload), timeout=TIMEOUT * 4) def _test_sync(config: dict) -> tuple[bool, str]: session = _open(config) try: folder = resolve(config, "") count = len(session.list(folder)) detail = f"Connected — {count} item(s) in {folder}" if session.fingerprint: pinned = bool((config.get("host_key_fingerprint") or "").strip()) detail += f"; host key {session.fingerprint}" + (" (pinned)" if pinned else " — pin it on the connector") return True, detail finally: session.close() async def test_credentials(config: dict) -> tuple[bool, str]: return await asyncio.wait_for(asyncio.to_thread(_test_sync, config), timeout=TIMEOUT * 2) def _f(key: str, label: str, *, secret: bool = False, required: bool = True, placeholder: str = "", help: str = "", options: list | None = None, advanced: bool = False, multiline: bool = False) -> dict: return {"key": key, "label": label, "secret": secret, "required": required, "placeholder": placeholder, "help": help, "options": options, "advanced": advanced, "multiline": multiline} _BASE_DIR = _f("base_dir", "Base folder", required=False, placeholder="/", help='Paths on task nodes are relative to this folder, e.g. "/outbound/invoices".') PROVIDERS: list[dict] = [ { "id": "sftp", "name": "SFTP", "tagline": "File transfer over SSH — password or private key", "environments": [], "docs_url": None, "capabilities": [], "notes": "Save once, then press Test: it shows the server's host key fingerprint to pin.", "fields": [ _f("host", "Host", placeholder="sftp.example.com"), _f("port", "Port", required=False, placeholder="22"), _f("username", "Username"), _f("password", "Password", secret=True, required=False, help="Leave blank when signing in with a private key."), _f("private_key", "Private key", secret=True, required=False, advanced=True, help="OpenSSH or PEM private key (Ed25519, ECDSA or RSA)."), _f("key_passphrase", "Key passphrase", secret=True, required=False, advanced=True), _f("host_key_fingerprint", "Host key fingerprint", required=False, placeholder="SHA256:…", help="Pin the server's key so FlowFabric refuses an impostor. The Test button shows it."), _BASE_DIR, ], }, { "id": "ftps", "name": "FTPS", "tagline": "FTP with explicit TLS — logins and files encrypted", "environments": [], "docs_url": None, "capabilities": [], "notes": "", "fields": [ _f("host", "Host", placeholder="ftp.example.com"), _f("port", "Port", required=False, placeholder="21"), _f("username", "Username"), _f("password", "Password", secret=True, required=False), _BASE_DIR, _f("passive", "Connection mode", options=[["yes", "Passive (works through firewalls)"], ["no", "Active"]], advanced=True), ], }, { "id": "ftp", "name": "FTP", "tagline": "Plain FTP — only on trusted networks, nothing is encrypted", "environments": [], "docs_url": None, "capabilities": [], "notes": "Plain FTP sends the password and files unencrypted. Prefer SFTP or FTPS.", "fields": [ _f("host", "Host", placeholder="ftp.example.com"), _f("port", "Port", required=False, placeholder="21"), _f("username", "Username", placeholder="anonymous"), _f("password", "Password", secret=True, required=False), _BASE_DIR, _f("passive", "Connection mode", options=[["yes", "Passive (works through firewalls)"], ["no", "Active"]], advanced=True), ], }, ]