🧦 Sox2 Swiss Army Knife - PGP + SFTP

pgp_sftp.py

# # NOTE - BEFORE RUNNING SCRIPT YOU SHOULD PROBABLY run
# the following in your shell so that the host can be added properly to ssh:
# sftp user@host

# # REQUIREMENTS # #
# pip install pandas
# pip install pgpy
# pip install psycopg2
# pip install pycrypto
# pip install pysftp

# # INPUTS # #
ssn_decrypt_key = "GET THIS FROM LASTPASS"
vendor_sftp_host = "GET THIS FROM LASTPASS"
vendor_sftp_user = "GET THIS FROM LASTPASS"
vendor_sftp_pwd = "GET THIS FROM LASTPASS"
vendor_public_key = "GET THIS FROM LASTPASS"
vendor_sftp_upload_dir = "GET THIS FROM LASTPASS"

# # IMPORTS # #

from dataclasses import dataclass
from dotenv import load_dotenv
import hashlib
from io import StringIO
import pandas as pd
from pgpy import PGPUID, PGPKey, PGPMessage
import pysftp


# # SecureUploader # #


@dataclass
class Secrets:
    vendor_public_key: str = None
    vendor_sftp_host: str = None
    vendor_sftp_user: str = None
    vendor_sftp_pwd: str = None
    vendor_sftp_upload_dir: str = None


class SecureUploader:
    def __init__(self, secrets: Secrets):
        self.secrets = secrets
        if secrets.vendor_public_key:
            self.pgp_pubkey, _ = PGPKey.from_blob(secrets.vendor_public_key)

    def encrypt_for_upload(self, plaintext: str) -> str:
        return str(self.pgp_pubkey.encrypt(PGPMessage.new(plaintext)))

    def upload(self, plaintext: str, output_file_name: str, encrypt_first: bool = True) -> str:
        message = self.encrypt_for_upload(plaintext) if encrypt_first else plaintext

        with pysftp.Connection(
            host=self.secrets.vendor_sftp_host,
            username=self.secrets.vendor_sftp_user,
            password=self.secrets.vendor_sftp_pwd,
            log="pysftp.log",
        ) as sftp:
            with sftp.cd(self.secrets.vendor_sftp_upload_dir):
                sftp.putfo(StringIO(message), output_file_name)


# # Prepare Secrets Uploader # #

secrets = Secrets(
    vendor_sftp_host=vendor_sftp_host,
    vendor_public_key=vendor_public_key,
    vendor_sftp_user=vendor_sftp_user,
    vendor_sftp_pwd=vendor_sftp_pwd,
    vendor_sftp_upload_dir=vendor_sftp_upload_dir,
)

uploader = SecureUploader(secrets)

# # Run the Query # #

sql = "SELECT * FROM yaknow..."
print(f"sql:\n{sql}")

# assuming you've defined a function query(sql) get a Pandas DataFrame:
df = query(sql)
print(f"finished query: df.shape = {df.shape}")

# # Encrypt the DataFrame # # 

csv_encrypted = uploader.encrypt_for_upload(df.to_csv())
print(f"len(csv_encrypted)={len(csv_encrypted):,}")

print("gonna try and upload")
uploader.upload(csv_encrypted, "data.csv", encrypt_first=False)