add new snippets

Author: hubshashwat

checksum_checking.py

@@ -0,0 +1,17 @@
+import hashlib
+
+sha256_checksum_1 = hashlib.sha256()
+sha256_checksum_2 = hashlib.sha256()
+
+path1 = 'test_checksum.txt'
+path2 = 'ddd.py'
+
+with open(path1, "rb") as f:
+    sha256_checksum_1.update(f.read())
+with open(path2, "rb") as f:
+    sha256_checksum_2.update(f.read())
+
+if sha256_checksum_1.hexdigest() == sha256_checksum_2.hexdigest():
+    print('same')
+else:
+    print('not same')

pgp_file_decryption.py

@@ -0,0 +1,73 @@
+import io
+import time
+from datetime import timedelta
+import psutil
+import addict
+import gnupg
+
+
+def get_memory_usage():
+    process = psutil.Process()
+    return process.memory_info().rss / (1024 ** 2)
+
+
+class GPGHandler:
+    def __init__(self, gpg_private_ascii: str, passphrase: str):
+        self.gpg_key = gnupg.GPG()
+        self.passphrase = passphrase
+        self.gpg_key.import_keys(gpg_private_ascii)
+        try:
+            self.gpg_key_id = self.gpg_key.list_keys()[-1]["keyid"]
+            print(f"Using GPG_KEY_ID={self.gpg_key_id}")
+        except IndexError:
+            raise ValueError("No GPG key found")
+
+    def decrypt_file_object(self, encrypted_file_object):
+        encrypted_file_object.seek(0)
+        result = self.gpg_key.decrypt_file(encrypted_file_object, always_trust=True, passphrase=self.passphrase)
+        if not result.ok:
+            raise ValueError(f"Decryption failed. KEY_ID={result.key_id}, Status - {result.status}")
+        decrypted_file_object = io.BytesIO(result.data)
+        decrypted_file_object.seek(0)
+        return decrypted_file_object
+
+    def encrypt_file_object(self, file_object: io.BytesIO) -> io.BytesIO:
+        """
+        Encrypt a file object.
+        """
+        encrypted_file_object = io.BytesIO()
+        encrypted_file_object.write(
+            self.gpg_key.encrypt(file_object.getvalue(), self.gpg_key_id, always_trust=True).data
+        )
+        encrypted_file_object.seek(0)
+        return encrypted_file_object
+
+
+if __name__ == '__main__':
+    memory_before = get_memory_usage()
+    print("memory before", memory_before)
+
+    # read pgp file
+    with open("13744603840.zip.pgp", "rb") as file:
+        file_content = file.read()
+
+    # add secrets
+    _secrets = addict.Dict(
+        {
+            "gpg_private_ascii": "",
+            "gpg_pubring_passphrase": ""
+        }
+    )
+
+    _gpg_handler = GPGHandler(
+        gpg_private_ascii=_secrets.gpg_private_ascii,
+        passphrase=_secrets.gpg_pubring_passphrase,
+    )
+    _start = time.time()
+    decrypted_content = _gpg_handler.decrypt_file_object(
+        io.BytesIO(file_content),
+    ).read()
+    _time_taken = str(timedelta(seconds=time.time() - _start))
+    print(f"[DECRYPTION-COMPLETED with READING] for file, TimeTaken: {_time_taken}")
+    memory_after = get_memory_usage()
+    print("memory after", memory_after)

vault_updation_and_retrieval.py

@@ -0,0 +1,38 @@
+import hvac
+
+MOUNT_POINT = ''  # mount point - can be fleet/stack level or a single level
+
+client = hvac.Client(
+    url='https://vault.enterprise.company.co/',
+    token=''  # token you copy from UI
+)
+client.secrets.kv.v2.configure(
+    mount_point=MOUNT_POINT
+)
+
+# Check if the client is authenticated
+if not client.is_authenticated():
+    raise Exception("Vault authentication failed")
+
+clients = ['test1', 'test2']
+
+for _client in clients:
+    # Define the path to the secret
+    secret_path = f'se/services/{_client}/sftp/'
+    try:
+        # Read the username from the specified path
+        response = client.secrets.kv.v2.list_secrets(path=secret_path, mount_point=MOUNT_POINT)
+        username = response['data']['keys']
+        for each_username in username:
+            # retrieve from vault
+            response = client.secrets.kv.v2.read_secret_version(
+                path=secret_path + each_username, mount_point=MOUNT_POINT)['data']['data']
+            print(response)
+            # update from vault
+            response['host'] = "new-host"
+            response['port'] = "22"
+            client.secrets.kv.v2.create_or_update_secret(path=secret_path + each_username, secret=response,
+                                                         mount_point=MOUNT_POINT)
+            print(f'update successful for: {each_username}')
+    except Exception as exc:
+        print(exc)

verify_sftp_connection.py

@@ -0,0 +1,58 @@
+"""
+brew install corkscrew
+"""
+
+import io
+import logging
+
+from paramiko import RSAKey
+import fsspec
+import paramiko
+
+logging.basicConfig()
+logging.getLogger("paramiko").setLevel(logging.DEBUG) # for example
+
+
+def with_proxy():
+    # pkey = RSAKey.from_private_key(io.StringIO(credentials['private_key_ascii']))
+    pkey = paramiko.Ed25519Key.from_private_key(io.StringIO(credentials['private_key_ascii']))
+    pcommand = f'corkscrew {credentials["proxy_host"]} {credentials["proxy_port"]} %s %d' % (credentials['host'], credentials['port'])
+    proxy = paramiko.proxy.ProxyCommand(pcommand)
+
+    authentication_kwargs = dict()
+    authentication_kwargs["pkey"] = pkey
+    authentication_kwargs["sock"] = proxy
+
+    fs = fsspec.filesystem(
+        "sftp", host=credentials['host'], port=credentials['port'], username=credentials['username'], **authentication_kwargs
+    )
+    if fs:
+        print("fs found:")
+        files = fs.ls("/", detail=True)
+        print("Files in the remote directory:")
+        k = []
+        for file in files:
+            k.append(file['name'])
+        print(sorted(k))
+
+
+def without_proxy():
+    pkey = RSAKey.from_private_key(io.StringIO(credentials['private_key_ascii']))
+    authentication_kwargs = dict()
+    authentication_kwargs["pkey"] = pkey
+
+    fs = fsspec.filesystem(
+        "sftp", host=credentials['host'], port=credentials['port'], username=credentials['username'], **authentication_kwargs
+    )
+    if fs:
+        print("fs found:")
+        files = fs.ls("/", detail=True)
+        print("Files in the remote directory:")
+        for file in files:
+            print(file['name'])
+
+
+if __name__ == '__main__':
+    credentials = {"host": "sftp.bloomberg.com", "port": 22, "username": "", "proxy_host": "", "proxy_port": 8080}
+
+    with_proxy()