Example of symmetric GPG of data streams in Python with minimal password exposure (USE AT YOUR OWN RISK)

GPG_from_python.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "683bbda0-b84d-4810-9941-f927f199a88f",
   "metadata": {},
   "source": [
    "# GPG encryption within Python\n",
    "\n",
    "This is just a demo of how it can be done. The approach is to write the password to a named pipe (FIFO) and then send the data over STDIN/STDOUT. The named pipe keeps it in memory (it's own thread).\n",
    "\n",
    "## WARNING\n",
    "\n",
    "While I belive this to be **more secure than writing plaintext files**, it is still (likely) vulnerable to interception.\n",
    "\n",
    "**I AM NOT A SECURITY EXPERT**. Use at your own risk."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "3adfcd05-0d0b-4a82-a35b-90a9963fbaad",
   "metadata": {},
   "outputs": [],
   "source": [
    "from threading import Thread\n",
    "import subprocess\n",
    "import os\n",
    "import tempfile"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "ef804791-efd9-4de9-8179-edf07ae35579",
   "metadata": {},
   "outputs": [],
   "source": [
    "unencrypted = b'Plaintext bytes unencrypted'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "6507905f-e49e-4178-89cc-21b32a9acfcc",
   "metadata": {},
   "outputs": [],
   "source": [
    "# from getpass import getpass\n",
    "# password = getpass('Password: ')\n",
    "password = 'mysecurepassword'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "54fff8b3-d5d3-447d-8b0e-2b7888e9a0f2",
   "metadata": {},
   "outputs": [],
   "source": [
    "class PasswordFIFO:\n",
    "    def __init__(self,password,*,fifo=None):\n",
    "        self.password = password\n",
    "        self.fifo = fifo\n",
    "        if not self.fifo:\n",
    "            self.fifo = tempfile.NamedTemporaryFile().name\n",
    "        \n",
    "        self.thread = Thread(target=self._target)\n",
    "        self.thread.start()\n",
    "    def _target(self):\n",
    "        try:\n",
    "            # This will fail if it exists. For security, we WANT THAT!\n",
    "            os.mkfifo(self.fifo)\n",
    "            with open(self.fifo,mode='wt') as fobj:\n",
    "                fobj.write(self.password)\n",
    "            \n",
    "        except: raise # allow for else\n",
    "        else: # Don't use finally since we don't want to delete existing\n",
    "            os.unlink(self.fifo)\n",
    "        \n",
    "    def join(self):\n",
    "        self.thread.join()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "0d20dc2c-c69c-4abd-b846-0e76db9924a3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-----BEGIN PGP MESSAGE-----\n",
      "\n",
      "jA0ECQMCvWaXf3D682D20lABIbx+Z5KJAmkY1U+jZWv2FtmsFu8pNrJy8kRnuIQw\n",
      "AOg946uxS+OiwIpsw7oXVRJa4EomHaqKN2hsDrwT8G8xP7tep5/OABYK7XehXMdu\n",
      "+Q==\n",
      "=1E6r\n",
      "-----END PGP MESSAGE-----\n",
      "\n"
     ]
    }
   ],
   "source": [
    "def encrypt(data,password,armor=False):\n",
    "    fifo = PasswordFIFO(password)\n",
    "    \n",
    "    cmd = ['gpg','--symmetric',\n",
    "           '--cipher-algo','AES256',\n",
    "           '--batch','--yes', # allow STDIN\n",
    "           '--passphrase-file',fifo.fifo,\n",
    "           '--no-symkey-cache', # May not be needed with stdin but just in case\n",
    "          ]\n",
    "    if armor:\n",
    "        cmd.append('--armor')\n",
    "    \n",
    "    proc = subprocess.run(cmd,\n",
    "                          input=data,\n",
    "                          capture_output=True)\n",
    "    \n",
    "    fifo.join() # Close the thread. Won't happen until gpg reads it\n",
    "    if armor:\n",
    "        return proc.stdout.decode('ascii')\n",
    "    return proc.stdout\n",
    "\n",
    "encrypted = encrypt(unencrypted,password,armor=True)\n",
    "print(encrypted)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "e3393c46-bf0a-44e7-a782-03fa04d5b86b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "b'Plaintext bytes unencrypted'\n"
     ]
    }
   ],
   "source": [
    "def decrypt(data,password):\n",
    "    fifo = PasswordFIFO(password)\n",
    "    \n",
    "    cmd = ['gpg','--decrypt',\n",
    "           '--batch','--yes', # allow STDIN\n",
    "           '--passphrase-file',fifo.fifo,\n",
    "           '--no-symkey-cache', # May not be needed with stdin but just in case\n",
    "          ]\n",
    "    \n",
    "    # convert data if str. Probably armored\n",
    "    if isinstance(data,str):\n",
    "        data = data.encode('ascii')\n",
    "        \n",
    "    proc = subprocess.run(cmd,\n",
    "                          input=data,\n",
    "                          capture_output=True)\n",
    "    \n",
    "    fifo.join() # Close the thread. Won't happen until gpg reads it\n",
    "    return proc.stdout\n",
    "decrypted = decrypt(encrypted,password)\n",
    "assert decrypted == unencrypted # print(f'{decrypted == unencrypted = }')\n",
    "print(decrypted)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9cd85d09-34c6-45d3-83ab-2cbdb521c48b",
   "metadata": {},
   "source": [
    "## License\n",
    "\n",
    "```\n",
    "Copyright 2022 Justin Winokur\n",
    "\n",
    "Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the \"Software\"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:\n",
    "\n",
    "The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.\n",
    "\n",
    "THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0be0ec93-66b1-4c8c-8dea-55b2858f4475",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}