Interacting with SFTP Repositories: A Paramiko Integration

Introduction

This article explores a Python library designed to interact with Secure File Transfer Protocol (SFTP) repositories. Specifically, it integrates with Fortanix, a cryptographic operations platform. The library leverages Paramiko, a Python implementation of the SSH protocol, to facilitate secure communication with remote servers. Through this article, we'll delve into the functionalities provided by this library, examining its core components and demonstrating their usage.

Index

• Configuration Dataclass
• SFTP Class Initialisation
• Authentication Process
• File Operations
  • Listing Files
  • Changing Directory
  • Deleting Files
  • Downloading Files
  • Uploading Files
• Cleanup Process

Code Examples

Configuration Dataclass

import base64
from dataclasses import dataclass
import logging
import paramiko


@dataclass
class Configuration:
    """
    Dataclass to store SFTP configuration details.

    Attributes:
        host (str): The hostname or IP address of the SFTP server.
        port (int): The port number of the SFTP server.
        username (str): The username used for authentication.
        password (str): The password used for authentication.
    """
    host: str = None
    port: int = None
    username: str = None
    password: str = None

SFTP Class Initialisation

class SFTP(object):
    """
    Class to interact with SFTP repositories.

    Methods:
        __init__: Initializes variables and performs authentication.
        auth: Performs authentication process.
        list_dir: Lists all files in the current directory.
        change_dir: Changes current directory.
        delete_file: Deletes a file from the current directory.
        download: Downloads a file from the SFTP server.
        upload: Uploads a file to the SFTP server.
        __del__: Cleans up resources upon object deletion.
    """
    def __init__(self, credentials: dict) -> None:
        """
        Initializes variables and performs authentication.

        Args:
            credentials (dict): Credentials.
        """
        # Logging configuration
        logging.basicConfig(level=logging.INFO, format="%(asctime)s %(module)s %(levelname)s %(message)s",
                            datefmt="%Y-%m-%d %H:%M:%S")

        # Credentials/configuration
        self.sftp_client: paramiko.sftp_client.SFTPClient = None
        self.__transport: paramiko.transport.Transport = None

        self.__transport, self.sftp_client = self.auth(credentials=credentials)

Authentication Process

    def auth(self, credentials: dict) -> tuple:
        """
        Authenticate (On-Premise).
        Performs the authentication process.

        Args:
            credentials (dict): A dictionary containing SFTP credentials.

        Returns:
            tuple: A tuple containing the Transport and SFTPClient objects.
        """
        logging.info(msg="Opens session")

        # Initialize local dataclass
        configuration = Configuration()

        # Extract credentials
        configuration.host = credentials["custom_metadata"]["host"]
        configuration.port = int(credentials["custom_metadata"]["port"])
        configuration.username = credentials["custom_metadata"]["username"]
        configuration.password = base64.b64decode(credentials["value"]).decode("utf-8")

        # Establish connection
        transport = paramiko.Transport((configuration.host, configuration.port))
        transport.connect(username=configuration.username, password=configuration.password)
        sftp_client = paramiko.SFTPClient.from_transport(transport)

        return transport, sftp_client

File Operations

    def list_dir(self, path: str = ".") -> list:
        """
        Lists all files in the current directory.

        Args:
            path (str, optional): The path to the target directory. Defaults to ".".

        Returns:
            list: A list of files in the specified directory.
        """
        logging.info(msg="List files")

        return self.sftp_client.listdir(path)

    def change_dir(self, path: str = ".") -> None:
        """
        Changes the current directory.

        Args:
            path (str, optional): The path to the target directory. Defaults to ".".
        """
        logging.info(msg="Change current directory")

        self.sftp_client.chdir(path)

    def delete_file(self, path: str) -> None:
        """
        Deletes a file from the current directory.

        Args:
            path (str): The path of the file to delete.
        """
        logging.info(msg=f"Delete file {path}")

        self.sftp_client.remove(path)

    def download(self, remotepath: str, localpath: str) -> None:
        """
        Downloads a file from the SFTP server.

        Args:
            remotepath (str): The path of the file on the SFTP server.
            localpath (str): The destination path on the local host.
        """
        logging.info(msg=f"Download file {remotepath}")

        self.sftp_client.get(remotepath, localpath)

    def upload(self, localpath: str, remotepath: str) -> None:
        """
        Uploads a file to the SFTP server.

        Args:
            localpath (str): The path of the file on the local host.
            remotepath (str): The destination path on the SFTP server.
        """
        logging.info(msg=f"Upload file {localpath}")

        self.sftp_client.put(localpath, remotepath)

Cleanup Process

    def __del__(self) -> None:
        """
        Cleans up resources upon object deletion.
        """
        logging.info(msg="Closes session")
        self.__transport.close()
        self.sftp_client.close()

Conclusion

This library provides a convenient interface for interacting with SFTP repositories, offering functionalities such as listing files, navigating directories, deleting, downloading, and uploading files. Leveraging the Paramiko library, it ensures secure communication with remote servers, making it suitable for various applications requiring file transfer capabilities over SSH connections.