Implementing SFTP with PGP for ACH Files: Automated Retrieval & Decryption Pipeline #

import logging
import hashlib
import paramiko
import pgpy
from pathlib import Path
from typing import Optional, Tuple, Dict, Any
from datetime import datetime, timezone

logger = logging.getLogger("ach_ingestion_pipeline")
logger.setLevel(logging.INFO)

class ACHSecureIngestor:
    def __init__(
        self,
        sftp_host: str,
        sftp_port: int,
        sftp_user: str,
        sftp_key_path: Path,
        pgp_key_path: Path,
        pgp_passphrase: str,
        host_key_fingerprint: str
    ):
        self.sftp_config = {
            "host": sftp_host,
            "port": sftp_port,
            "username": sftp_user,
            "key_filename": str(sftp_key_path)
        }
        self.pgp_key_path = pgp_key_path
        self.pgp_passphrase = pgp_passphrase
        self.expected_host_key = host_key_fingerprint
        self._load_pgp_key()

    def _load_pgp_key(self) -> None:
        try:
            self.pgp_key, _ = pgpy.PGPKey.from_file(str(self.pgp_key_path))
            if self.pgp_passphrase:
                with self.pgp_key.unlock(self.pgp_passphrase):
                    logger.info("PGP private key loaded and unlocked successfully.")
        except pgpy.errors.PGPError as e:
            logger.critical("Failed to load or unlock PGP key: %s", e)
            raise

    def _compute_sha256(self, data: bytes) -> str:
        return hashlib.sha256(data).hexdigest()

    def _establish_sftp(self) -> paramiko.SFTPClient:
        client = paramiko.SSHClient()
        client.load_system_host_keys()
        client.set_missing_host_key_policy(paramiko.RejectPolicy())
        
        try:
            client.connect(**self.sftp_config, timeout=30)
            transport = client.get_transport()
            if not transport:
                raise ConnectionError("SSH transport failed to initialize.")
            
            # Strict host key verification
            remote_key = transport.get_remote_server_key()
            if remote_key.get_fingerprint().hex() != self.expected_host_key:
                raise SecurityError("Host key mismatch detected. Connection aborted.")
                
            return client.open_sftp()
        except Exception as e:
            logger.error("SFTP connection failed: %s", e)
            client.close()
            raise

    def retrieve_and_decrypt(self, remote_path: str, local_dir: Path) -> Dict[str, Any]:
        local_dir.mkdir(parents=True, exist_ok=True)
        local_encrypted = local_dir / Path(remote_path).name
        local_decrypted = local_dir / f"{local_encrypted.stem}_decrypted.nacha"
        
        audit_log: Dict[str, Any] = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "remote_path": remote_path,
            "status": "pending"
        }

        with self._establish_sftp() as sftp:
            try:
                # 1. Retrieve encrypted payload
                sftp.get(remote_path, str(local_encrypted))
                encrypted_bytes = local_encrypted.read_bytes()
                audit_log["encrypted_sha256"] = self._compute_sha256(encrypted_bytes)
                logger.info("Retrieved encrypted ACH file: %s", remote_path)

                # 2. Decrypt payload (preserving exact byte sequence)
                encrypted_msg = pgpy.PGPMessage.from_blob(encrypted_bytes)
                with self.pgp_key.unlock(self.pgp_passphrase):
                    decrypted_msg = self.pgp_key.decrypt(encrypted_msg)
                
                plaintext_bytes = decrypted_msg.message.encode("utf-8") if isinstance(decrypted_msg.message, str) else decrypted_msg.message
                local_decrypted.write_bytes(plaintext_bytes)
                
                audit_log["decrypted_sha256"] = self._compute_sha256(plaintext_bytes)
                audit_log["status"] = "success"
                audit_log["key_fingerprint"] = self.pgp_key.fingerprint
                logger.info("Decryption successful. Plaintext written to: %s", local_decrypted)
                
                return audit_log

            except pgpy.errors.PGPDecryptionError as e:
                audit_log["status"] = "decryption_failed"
                audit_log["error"] = str(e)
                logger.error("PGP decryption failed: %s", e)
                raise
            except Exception as e:
                audit_log["status"] = "pipeline_failed"
                audit_log["error"] = str(e)
                logger.error("SFTP/Decryption pipeline error: %s", e)
                raise
            finally:
                # Cleanup encrypted intermediate file
                if local_encrypted.exists():
                    local_encrypted.unlink()