"""SFTP Transfer Module. This module provides functionality to upload non-zipped feed files to a remote SFTP server. It is designed to be invoked after feed file generation and before local cleanup/zipping (so the original files still exist). Guiding principles (see repository AI instructions): - Minimal invasive changes: standalone module, optional invocation. - Robust logging of milestones and errors; never log secrets. - Retry with backoff for transient failures. - Path traversal prevention: only basename of local files is used remotely. - Host key verification optionally enforced (default: strict=True). - Continue-on-error mode optionally available via env var. - Ready for future async/concurrency although current implementation performs synchronous uploads to keep complexity low. Unit tests will mock paramiko classes; this module MUST NOT attempt any real network connections in test or CI environments. Code paths that would connect are isolated and can be monkeypatched/mocked. """ from __future__ import annotations from dataclasses import dataclass, field from pathlib import Path import logging import os import sys import time from typing import Iterable, List, Optional, Sequence, Any import paramiko # type: ignore __all__ = [ "SFTPConfig", "SFTPTransferManager", "SFTPUploadResult", "SFTPError", "SFTPConfigurationError", "SFTPUploadError", "transfer_non_zipped_files_if_enabled", ] # New: helper detection of interactive TTY for prompting logic. def _is_interactive() -> bool: """Return True if running in an interactive terminal session. We avoid prompting when stdin is not a TTY (CI/CD).""" return sys.stdin.isatty() and sys.stdout.isatty() logger = logging.getLogger( os.environ.get("LOGGER_NAME", "sme-feed-file-exporter") ) class SFTPError(Exception): """Base SFTP exception.""" class SFTPConfigurationError(SFTPError): """Raised when configuration is invalid.""" class SFTPUploadError(SFTPError): """Raised when a file upload ultimately fails after retries.""" @dataclass(frozen=True) class SFTPConfig: """Immutable configuration for SFTP transfers. Attributes: host: SFTP hostname. port: SFTP port (default 22). username: Username for authentication. password: Password (MUST NOT be logged). remote_dir: Remote directory path relative to the user's root. retries: Number of retry attempts after the initial try. backoff_sec: Base seconds for exponential backoff (attempt * backoff_sec). timeout: Socket timeout in seconds. strict_host_key: Enforce host key verification when True. continue_on_error: If True, failures are collected and logged; otherwise first fatal failure raises and stops further uploads. auth_mode: Authentication mode - 'key' (default), 'password', 'auto'. private_key_path: Path to private key file (defaults to ~/.ssh/id_rsa). """ host: str port: int = 22 username: str = "" password: Optional[str] = None # TODO: DEPRECATE - Remove when password auth removed remote_dir: str = "monthly" retries: int = 3 backoff_sec: float = 2.0 timeout: float = 30.0 strict_host_key: bool = True continue_on_error: bool = False auth_mode: str = "key" # Default: key-based auth private_key_path: str = "~/.ssh/id_rsa" # internal field to store sanitized remote dir _sanitized_remote_dir: str = field(init=False, repr=False) def __post_init__(self) -> None: # Basic validation if not self.host: raise SFTPConfigurationError("SFTP host is required") if not self.username: raise SFTPConfigurationError("SFTP username is required") if self.port <= 0: raise SFTPConfigurationError("SFTP port must be positive") # Validate auth_mode valid_modes = {"password", "key", "auto"} if self.auth_mode not in valid_modes: raise SFTPConfigurationError( f"Invalid auth_mode '{self.auth_mode}'. " f"Must be one of: {', '.join(valid_modes)}" ) # sanitize remote directory: remove whitespace and any '..' rd = self.remote_dir.strip().replace("\\", "/") if ".." in rd: raise SFTPConfigurationError( "Remote directory contains illegal traversal component '..'" ) while rd.startswith("/"): rd = rd[1:] object.__setattr__(self, "_sanitized_remote_dir", rd or "") @property def sanitized_remote_dir(self) -> str: """Return sanitized remote directory path safe for concatenation.""" return self._sanitized_remote_dir @dataclass class SFTPUploadResult: """Summary information about an upload batch.""" uploaded: int failed: List[Path] total: int def success(self) -> bool: return self.uploaded == self.total and not self.failed class SFTPTransferManager: """Manager responsible for uploading files via SFTP. Usage: cfg = SFTPConfig(...) mgr = SFTPTransferManager(cfg) mgr.connect() result = mgr.upload_files(list_of_paths) mgr.close() """ def __init__(self, config: SFTPConfig) -> None: # noqa: D401 """Initialize transfer manager with provided configuration.""" self.config = config # Using Any to avoid dependency on paramiko stubs. # Attributes set in connect(). self._ssh_client: Optional[Any] = None self._sftp_client: Optional[Any] = None # NOTE: Actual network connection code exists but will only run when # paramiko is available and this method is invoked. Tests will mock it. def connect(self) -> None: if self._ssh_client is not None: logger.debug( "SFTPTransferManager.connect() called but already connected." ) return # Mock mode: environment variable SFTP_MOCK truthy will bypass real # network operations and create a dummy client exposing put(). This # supports local dry runs and integration tests without hitting a # live SFTP server. (Never connect to live server without explicit # permission per repository instructions.) mock_flag = str(os.environ.get("SFTP_MOCK", "False")).lower() in { "1", "true", "yes", } if mock_flag: logger.info("SFTP mock mode enabled; simulating transfers.") class _DummyClient: def __init__(self) -> None: self.uploads: list[tuple[str, str]] = [] def put(self, local: str, remote: str) -> None: # noqa: D401 # Simulate success immediately. self.uploads.append((local, remote)) logger.debug("(mock) put %s -> %s", local, remote) def close(self) -> None: # pragma: no cover - trivial logger.debug("(mock) sftp client closed") class _DummySSH: def open_sftp(self) -> _DummyClient: # type: ignore return _DummyClient() def close(self) -> None: # pragma: no cover - trivial logger.debug("(mock) ssh client closed") self._ssh_client = _DummySSH() self._sftp_client = self._ssh_client.open_sftp() return logger.info( "Establishing SFTP connection to host '%s':%s", self.config.host, self.config.port, ) ssh = paramiko.SSHClient() # Load system-wide host keys (read-only, /etc/ssh/known_hosts) ssh.load_system_host_keys() # Load user's personal known_hosts file where host keys are # typically stored after first connection (~/.ssh/known_hosts). # Using load_host_keys() (not load_system_host_keys()) ensures # that AutoAddPolicy can save new host keys back to this file. # See: https://docs.paramiko.org/en/stable/api/client.html user_known_hosts = os.path.expanduser("~/.ssh/known_hosts") if os.path.isfile(user_known_hosts): try: ssh.load_host_keys(user_known_hosts) logger.debug( "Loaded user known_hosts from '%s'", user_known_hosts ) except Exception as exc: logger.warning( "Failed to load user known_hosts '%s': %s", user_known_hosts, exc, ) else: # Create ~/.ssh directory and known_hosts if they don't exist # so AutoAddPolicy can save new host keys. ssh_dir = os.path.dirname(user_known_hosts) try: os.makedirs(ssh_dir, mode=0o700, exist_ok=True) Path(user_known_hosts).touch(mode=0o600, exist_ok=True) ssh.load_host_keys(user_known_hosts) logger.debug( "Created and loaded empty known_hosts at '%s'", user_known_hosts, ) except Exception as exc: logger.warning( "Failed to create known_hosts '%s': %s", user_known_hosts, exc, ) if self.config.strict_host_key: # Reject unknown hosts - user must add them manually first ssh.set_missing_host_key_policy(paramiko.RejectPolicy()) else: # Auto-add unknown hosts and save them to known_hosts file ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # Determine authentication parameters based on auth_mode auth_mode = self.config.auth_mode connect_kwargs = { "hostname": self.config.host, "port": self.config.port, "username": self.config.username, "timeout": self.config.timeout, } # TODO: DEPRECATE - Remove 'password' mode block when transitioning # to key-only authentication. Keep only 'key' mode. if auth_mode == "password": # Password-only authentication connect_kwargs["password"] = self.config.password connect_kwargs["look_for_keys"] = False connect_kwargs["allow_agent"] = False logger.debug("Using password-only authentication") elif auth_mode == "key": # Private key-only authentication key_path = os.path.expanduser(self.config.private_key_path) connect_kwargs["key_filename"] = key_path connect_kwargs["look_for_keys"] = False connect_kwargs["allow_agent"] = False logger.debug( "Using private key authentication from '%s'", key_path ) # TODO: DEPRECATE - Remove 'auto' mode when password auth removed. elif auth_mode == "auto": # Try key first, then password - let paramiko handle the logic key_path = os.path.expanduser(self.config.private_key_path) connect_kwargs["key_filename"] = key_path connect_kwargs["password"] = self.config.password # Allow paramiko to try other keys and agent connect_kwargs["look_for_keys"] = True connect_kwargs["allow_agent"] = True logger.debug( "Using auto authentication (key from '%s' + password)", key_path ) # We intentionally do not log password. ssh.connect(**connect_kwargs) self._ssh_client = ssh self._sftp_client = ssh.open_sftp() logger.info("SFTP session established.") def close(self) -> None: if self._sftp_client: try: self._sftp_client.close() except Exception: # pragma: no cover - best effort logger.debug("Failed closing sftp client", exc_info=True) if self._ssh_client: try: self._ssh_client.close() except Exception: # pragma: no cover - best effort logger.debug("Failed closing ssh client", exc_info=True) self._sftp_client = None self._ssh_client = None logger.info("SFTP connection closed.") # Internal helper ----------------------------------------------------- def _remote_target_for(self, local_path: Path) -> str: # Use basename only to prevent traversal. fname = local_path.name if self.config.sanitized_remote_dir: return f"{self.config.sanitized_remote_dir}/{fname}" return fname def upload_file(self, local_path: Path) -> None: if not self._sftp_client: raise SFTPError("SFTP client not connected") remote_path = self._remote_target_for(local_path) attempts = self.config.retries + 1 for attempt in range(1, attempts + 1): try: logger.debug( "Uploading '%s' -> '%s' (attempt %d/%d)", local_path, remote_path, attempt, attempts, ) self._sftp_client.put(str(local_path), remote_path) logger.info("Uploaded file '%s'", local_path.name) return except Exception as exc: # paramiko specific + IO errors if attempt == attempts: logger.error( "Failed to upload '%s': %s", local_path.name, exc ) raise SFTPUploadError( f"Failed to upload {local_path}: {exc}" ) from exc backoff = self.config.backoff_sec * attempt logger.warning( "Attempt %d for '%s' failed (%s). Retry in %.1fs", attempt, local_path.name, exc, backoff, ) time.sleep(backoff) def upload_files(self, files: Sequence[Path]) -> SFTPUploadResult: failed: List[Path] = [] uploaded = 0 total = len(files) logger.info("Beginning SFTP upload batch: %d files", total) for p in files: try: self.upload_file(p) uploaded += 1 except SFTPUploadError: failed.append(p) if not self.config.continue_on_error: logger.error( "Aborting batch after failure of '%s'", p.name ) break logger.info( "SFTP upload batch complete: uploaded=%d failed=%d total=%d", uploaded, len(failed), total, ) return SFTPUploadResult(uploaded=uploaded, failed=failed, total=total) def transfer_non_zipped_files_if_enabled( file_paths: Iterable[str], config: SFTPConfig, ) -> Optional[SFTPUploadResult]: """Convenience function to perform conditional transfer. Args: file_paths: Iterable of local file path strings. config: SFTPConfig instance. Returns: SFTPUploadResult if transfer executed else None. """ if not file_paths: logger.info("No files provided for SFTP transfer; skipping.") return None mgr = SFTPTransferManager(config) mgr.connect() # Real connection only if invoked outside tests. try: path_objs = [Path(p) for p in file_paths] return mgr.upload_files(path_objs) finally: mgr.close() # ---------------------- Dev helper functions ----------------------------- # TODO: DEPRECATE - Remove this entire function when password auth removed. def resolve_password_interactive(existing: Optional[str]) -> Optional[str]: """Return a password using existing value or interactive prompt. Rules: - If existing is provided (non-empty), return as-is. - If SFTP_PROMPT_PASSWORD is truthy AND interactive AND mock or dev environment, prompt user securely for password. - Otherwise return existing (may be None) without prompting. """ if existing: # already have a value return existing # Evaluate environment intent env_pw = os.environ.get("SFTP_PASSWORD") prompt_flag = str( os.environ.get("SFTP_PROMPT_PASSWORD", "False") ).lower() in {"1", "true", "yes"} env = os.environ.get("ENVIRONMENT", "").upper() is_dev_like = env in {"DEV", "LOCAL", ""} if prompt_flag and is_dev_like and _is_interactive(): # Environment-provided password wins over prompt when present if env_pw: return env_pw try: import getpass # local import to avoid overhead elsewhere pwd = getpass.getpass("Enter SFTP password: ") if not pwd: logger.warning("Empty password entered; proceeding with None") return None logger.info("SFTP password captured interactively.") return pwd except Exception as exc: # pragma: no cover - defensive logger.error("Interactive password prompt failed: %s", exc) return None else: if prompt_flag and not _is_interactive(): logger.info( "Password prompt suppressed (non-interactive session)." ) return None # Fall back to environment-provided value when not prompting if env_pw: return env_pw return existing __all__.append("resolve_password_interactive") __all__ = [ "SFTPConfig", "SFTPTransferManager", "SFTPUploadResult", "SFTPError", "SFTPConfigurationError", "SFTPUploadError", "transfer_non_zipped_files_if_enabled", ]