We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
GPG PINENTRY_LAUNCHED error when sources are deleted
## Description

Deleting a source raises this error.

```pytb
Exception in thread Thread-39:
Traceback (most recent call last):
  File "/usr/lib/python2.7/threading.py", line 801, in __bootstrap_inner
    self.run()
  File "/usr/lib/python2.7/threading.py", line 754, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/usr/local/lib/python2.7/dist-packages/pretty_bad_protocol/_meta.py", line 670, in _read_response
    result._handle_status(keyword, value)
  File "/usr/local/lib/python2.7/dist-packages/pretty_bad_protocol/_parsers.py", line 1024, in _handle_status
    raise ValueError("Unknown status message: %r" % key)
ValueError: Unknown status message: u'PINENTRY_LAUNCHED'
```

## Steps to Reproduce

- Start container
- Start client
- Delete source

## Expected Behavior

No errors.

## Actual Behavior

There are errors but the HTTP code is still 200, and the key is still deleted on the server side.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `securedrop/crypto_util.py`
Content:
```
1 #!/usr/bin/env python
2 # -*- coding: utf-8 -*-
3 
4 from distutils.version import StrictVersion
5 import pretty_bad_protocol as gnupg
6 import os
7 import io
8 import six
9 import scrypt
10 import subprocess
11 from random import SystemRandom
12 
13 from base64 import b32encode
14 from datetime import date
15 from flask import current_app
16 from pretty_bad_protocol._util import _is_stream, _make_binary_stream
17 
18 import typing
19 # https://www.python.org/dev/peps/pep-0484/#runtime-or-type-checking
20 if typing.TYPE_CHECKING:
21     # flake8 can not understand type annotation yet.
22     # That is why all type annotation relative import
23     # statements has to be marked as noqa.
24     # http://flake8.pycqa.org/en/latest/user/error-codes.html?highlight=f401stream
25     from typing import Dict, List, Text  # noqa: F401
26 
27 # to fix gpg error #78 on production
28 os.environ['USERNAME'] = 'www-data'
29 
30 # SystemRandom sources from the system rand (e.g. urandom, CryptGenRandom, etc)
31 # It supplies a CSPRNG but with an interface that supports methods like choice
32 random = SystemRandom()
33 
34 # safe characters for every possible word in the wordlist includes capital
35 # letters because codename hashes are base32-encoded with capital letters
36 DICEWARE_SAFE_CHARS = (' !#%$&)(+*-1032547698;:=?@acbedgfihkjmlonqpsrutwvyxzA'
37                        'BCDEFGHIJKLMNOPQRSTUVWXYZ')
38 
39 
40 class CryptoException(Exception):
41     pass
42 
43 
44 class CryptoUtil:
45 
46     GPG_KEY_TYPE = "RSA"
47     DEFAULT_WORDS_IN_RANDOM_ID = 8
48 
49     # All reply keypairs will be "created" on the same day SecureDrop (then
50     # Strongbox) was publicly released for the first time.
51     # https://www.newyorker.com/news/news-desk/strongbox-and-aaron-swartz
52     DEFAULT_KEY_CREATION_DATE = date(2013, 5, 14)
53 
54     # '0' is the magic value that tells GPG's batch key generation not
55     # to set an expiration date.
56     DEFAULT_KEY_EXPIRATION_DATE = '0'
57 
58     def __init__(self,
59                  scrypt_params,
60                  scrypt_id_pepper,
61                  scrypt_gpg_pepper,
62                  securedrop_root,
63                  word_list,
64                  nouns_file,
65                  adjectives_file,
66                  gpg_key_dir):
67         self.__securedrop_root = securedrop_root
68         self.__word_list = word_list
69 
70         if os.environ.get('SECUREDROP_ENV') == 'test':
71             # Optimize crypto to speed up tests (at the expense of security
72             # DO NOT use these settings in production)
73             self.__gpg_key_length = 1024
74             self.scrypt_params = dict(N=2**1, r=1, p=1)
75         else:  # pragma: no cover
76             self.__gpg_key_length = 4096
77             self.scrypt_params = scrypt_params
78 
79         self.scrypt_id_pepper = scrypt_id_pepper
80         self.scrypt_gpg_pepper = scrypt_gpg_pepper
81 
82         self.do_runtime_tests()
83 
84         # --pinentry-mode, required for SecureDrop on gpg 2.1.x+, was
85         # added in gpg 2.1.
86         self.gpg_key_dir = gpg_key_dir
87         gpg_binary = gnupg.GPG(binary='gpg2', homedir=self.gpg_key_dir)
88         if StrictVersion(gpg_binary.binary_version) >= StrictVersion('2.1'):
89             self.gpg = gnupg.GPG(binary='gpg2',
90                                  homedir=gpg_key_dir,
91                                  options=['--pinentry-mode loopback'])
92         else:
93             self.gpg = gpg_binary
94 
95         # map code for a given language to a localized wordlist
96         self.__language2words = {}  # type: Dict[Text, List[str]]
97 
98         with io.open(nouns_file) as f:
99             self.nouns = f.read().splitlines()
100 
101         with io.open(adjectives_file) as f:
102             self.adjectives = f.read().splitlines()
103 
104     # Make sure these pass before the app can run
105     def do_runtime_tests(self):
106         if self.scrypt_id_pepper == self.scrypt_gpg_pepper:
107             raise AssertionError('scrypt_id_pepper == scrypt_gpg_pepper')
108         # crash if we don't have srm:
109         try:
110             subprocess.check_call(['srm'], stdout=subprocess.PIPE)
111         except subprocess.CalledProcessError:
112             pass
113 
114     def get_wordlist(self, locale):
115         # type: (Text) -> List[str]
116         """" Ensure the wordlist for the desired locale is read and available
117         in the words global variable. If there is no wordlist for the
118         desired local, fallback to the default english wordlist.
119 
120         The localized wordlist are read from wordlists/{locale}.txt but
121         for backward compatibility purposes the english wordlist is read
122         from the config.WORD_LIST file.
123         """
124 
125         if locale not in self.__language2words:
126             if locale != 'en':
127                 path = os.path.join(self.__securedrop_root,
128                                     'wordlists',
129                                     locale + '.txt')
130                 if os.path.exists(path):
131                     wordlist_path = path
132                 else:
133                     wordlist_path = self.__word_list
134             else:
135                 wordlist_path = self.__word_list
136 
137             with io.open(wordlist_path) as f:
138                 content = f.read().splitlines()
139                 self.__language2words[locale] = content
140 
141         return self.__language2words[locale]
142 
143     def genrandomid(self,
144                     words_in_random_id=None,
145                     locale='en'):
146         if words_in_random_id is None:
147             words_in_random_id = self.DEFAULT_WORDS_IN_RANDOM_ID
148         return ' '.join(random.choice(self.get_wordlist(locale))
149                         for x in range(words_in_random_id))
150 
151     def display_id(self):
152         return ' '.join([random.choice(self.adjectives),
153                          random.choice(self.nouns)])
154 
155     def hash_codename(self, codename, salt=None):
156         """Salts and hashes a codename using scrypt.
157 
158         :param str codename: A source's codename.
159         :param str salt: The salt to mix with the codename when hashing.
160         :returns: A base32 encoded string; the salted codename hash.
161         """
162         if salt is None:
163             salt = self.scrypt_id_pepper
164         return b32encode(scrypt.hash(clean(codename),
165                          salt,
166                          **self.scrypt_params)).decode('utf-8')
167 
168     def genkeypair(self, name, secret):
169         """Generate a GPG key through batch file key generation. A source's
170         codename is salted with SCRYPT_GPG_PEPPER and hashed with scrypt to
171         provide the passphrase used to encrypt their private key. Their name
172         should be their filesystem id.
173 
174         >>> if not gpg.list_keys(hash_codename('randomid')):
175         ...     genkeypair(hash_codename('randomid'), 'randomid').type
176         ... else:
177         ...     u'P'
178         u'P'
179 
180         :param str name: The source's filesystem id (their codename, salted
181                          with SCRYPT_ID_PEPPER, and hashed with scrypt).
182         :param str secret: The source's codename.
183         :returns: a :class:`GenKey <gnupg._parser.GenKey>` object, on which
184                   the ``__str__()`` method may be called to return the
185                   generated key's fingeprint.
186 
187         """
188         name = clean(name)
189         secret = self.hash_codename(secret, salt=self.scrypt_gpg_pepper)
190         genkey_obj = self.gpg.gen_key(self.gpg.gen_key_input(
191             key_type=self.GPG_KEY_TYPE,
192             key_length=self.__gpg_key_length,
193             passphrase=secret,
194             name_email=name,
195             creation_date=self.DEFAULT_KEY_CREATION_DATE.isoformat(),
196             expire_date=self.DEFAULT_KEY_EXPIRATION_DATE
197         ))
198         return genkey_obj
199 
200     def delete_reply_keypair(self, source_filesystem_id):
201         key = self.getkey(source_filesystem_id)
202         # If this source was never flagged for review, they won't have a reply
203         # keypair
204         if not key:
205             return
206 
207         # Always delete keys without invoking pinentry-mode = loopback
208         # see: https://lists.gnupg.org/pipermail/gnupg-users/2016-May/055965.html
209         temp_gpg = gnupg.GPG(binary='gpg2', homedir=self.gpg_key_dir)
210         # The subkeys keyword argument deletes both secret and public keys.
211         temp_gpg.delete_keys(key, secret=True, subkeys=True)
212 
213     def getkey(self, name):
214         for key in self.gpg.list_keys():
215             for uid in key['uids']:
216                 if name in uid:
217                     return key['fingerprint']
218         return None
219 
220     def export_pubkey(self, name):
221         fingerprint = self.getkey(name)
222         if fingerprint:
223             return self.gpg.export_keys(fingerprint)
224         else:
225             return None
226 
227     def encrypt(self, plaintext, fingerprints, output=None):
228         # Verify the output path
229         if output:
230             current_app.storage.verify(output)
231 
232         if not isinstance(fingerprints, (list, tuple)):
233             fingerprints = [fingerprints, ]
234         # Remove any spaces from provided fingerprints GPG outputs fingerprints
235         # with spaces for readability, but requires the spaces to be removed
236         # when using fingerprints to specify recipients.
237         fingerprints = [fpr.replace(' ', '') for fpr in fingerprints]
238 
239         if not _is_stream(plaintext):
240             plaintext = _make_binary_stream(plaintext, "utf_8")
241 
242         out = self.gpg.encrypt(plaintext,
243                                *fingerprints,
244                                output=output,
245                                always_trust=True,
246                                armor=False)
247         if out.ok:
248             return out.data
249         else:
250             raise CryptoException(out.stderr)
251 
252     def decrypt(self, secret, ciphertext):
253         """
254         >>> crypto = current_app.crypto_util
255         >>> key = crypto.genkeypair('randomid', 'randomid')
256         >>> message = u'Buenos días, mundo hermoso!'
257         >>> ciphertext = crypto.encrypt(message, str(key))
258         >>> crypto.decrypt('randomid', ciphertext) == message.encode('utf-8')
259         True
260         """
261         hashed_codename = self.hash_codename(secret,
262                                              salt=self.scrypt_gpg_pepper)
263         data = self.gpg.decrypt(ciphertext, passphrase=hashed_codename).data
264 
265         if not six.PY2:  # Python3
266             return data.decode('utf-8')
267         return data
268 
269 
270 def clean(s, also=''):
271     """
272     >>> clean("[]")
273     Traceback (most recent call last):
274       ...
275     CryptoException: invalid input: []
276     >>> clean("Helloworld")
277     'Helloworld'
278     """
279     for c in s:
280         if c not in DICEWARE_SAFE_CHARS and c not in also:
281             raise CryptoException("invalid input: {0}".format(s))
282     # scrypt.hash requires input of type str. Since the wordlist is all ASCII
283     # characters, this conversion is not problematic
284     return str(s)
285
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/securedrop/crypto_util.py b/securedrop/crypto_util.py
--- a/securedrop/crypto_util.py
+++ b/securedrop/crypto_util.py
@@ -37,6 +37,23 @@
                        'BCDEFGHIJKLMNOPQRSTUVWXYZ')
 
 
+def monkey_patch_delete_handle_status(self, key, value):
+        """Parse a status code from the attached GnuPG process.
+        :raises: :exc:`~exceptions.ValueError` if the status message is unknown.
+        """
+        if key in ("DELETE_PROBLEM", "KEY_CONSIDERED"):
+            self.status = self.problem_reason.get(value, "Unknown error: %r"
+                                                  % value)
+        elif key in ("PINENTRY_LAUNCHED"):
+            self.status = key.replace("_", " ").lower()
+        else:
+            raise ValueError("Unknown status message: %r" % key)
+
+
+# Monkey patching to resolve https://github.com/freedomofpress/securedrop/issues/4294
+gnupg._parsers.DeleteResult._handle_status = monkey_patch_delete_handle_status
+
+
 class CryptoException(Exception):
     pass

{"golden_diff": "diff --git a/securedrop/crypto_util.py b/securedrop/crypto_util.py\n--- a/securedrop/crypto_util.py\n+++ b/securedrop/crypto_util.py\n@@ -37,6 +37,23 @@\n                        'BCDEFGHIJKLMNOPQRSTUVWXYZ')\n \n \n+def monkey_patch_delete_handle_status(self, key, value):\n+        \"\"\"Parse a status code from the attached GnuPG process.\n+        :raises: :exc:`~exceptions.ValueError` if the status message is unknown.\n+        \"\"\"\n+        if key in (\"DELETE_PROBLEM\", \"KEY_CONSIDERED\"):\n+            self.status = self.problem_reason.get(value, \"Unknown error: %r\"\n+                                                  % value)\n+        elif key in (\"PINENTRY_LAUNCHED\"):\n+            self.status = key.replace(\"_\", \" \").lower()\n+        else:\n+            raise ValueError(\"Unknown status message: %r\" % key)\n+\n+\n+# Monkey patching to resolve https://github.com/freedomofpress/securedrop/issues/4294\n+gnupg._parsers.DeleteResult._handle_status = monkey_patch_delete_handle_status\n+\n+\n class CryptoException(Exception):\n     pass\n", "issue": "GPG PINENTRY_LAUNCHED error when sources are deleted\n## Description\r\n\r\nDeleting a source raises this error.\r\n\r\n```pytb\r\nException in thread Thread-39:\r\nTraceback (most recent call last):\r\n  File \"/usr/lib/python2.7/threading.py\", line 801, in __bootstrap_inner\r\n    self.run()\r\n  File \"/usr/lib/python2.7/threading.py\", line 754, in run\r\n    self.__target(*self.__args, **self.__kwargs)\r\n  File \"/usr/local/lib/python2.7/dist-packages/pretty_bad_protocol/_meta.py\", line 670, in _read_response\r\n    result._handle_status(keyword, value)\r\n  File \"/usr/local/lib/python2.7/dist-packages/pretty_bad_protocol/_parsers.py\", line 1024, in _handle_status\r\n    raise ValueError(\"Unknown status message: %r\" % key)\r\nValueError: Unknown status message: u'PINENTRY_LAUNCHED'\r\n```\r\n\r\n## Steps to Reproduce\r\n\r\n- Start container\r\n- Start client\r\n- Delete source\r\n\r\n## Expected Behavior\r\n\r\nNo errors.\r\n\r\n## Actual Behavior\r\n\r\nThere are errors but the HTTP code is still 200, and the key is still deleted on the server side.\n", "before_files": [{"content": "#!/usr/bin/env python\n# -*- coding: utf-8 -*-\n\nfrom distutils.version import StrictVersion\nimport pretty_bad_protocol as gnupg\nimport os\nimport io\nimport six\nimport scrypt\nimport subprocess\nfrom random import SystemRandom\n\nfrom base64 import b32encode\nfrom datetime import date\nfrom flask import current_app\nfrom pretty_bad_protocol._util import _is_stream, _make_binary_stream\n\nimport typing\n# https://www.python.org/dev/peps/pep-0484/#runtime-or-type-checking\nif typing.TYPE_CHECKING:\n    # flake8 can not understand type annotation yet.\n    # That is why all type annotation relative import\n    # statements has to be marked as noqa.\n    # http://flake8.pycqa.org/en/latest/user/error-codes.html?highlight=f401stream\n    from typing import Dict, List, Text  # noqa: F401\n\n# to fix gpg error #78 on production\nos.environ['USERNAME'] = 'www-data'\n\n# SystemRandom sources from the system rand (e.g. urandom, CryptGenRandom, etc)\n# It supplies a CSPRNG but with an interface that supports methods like choice\nrandom = SystemRandom()\n\n# safe characters for every possible word in the wordlist includes capital\n# letters because codename hashes are base32-encoded with capital letters\nDICEWARE_SAFE_CHARS = (' !#%$&)(+*-1032547698;:=?@acbedgfihkjmlonqpsrutwvyxzA'\n                       'BCDEFGHIJKLMNOPQRSTUVWXYZ')\n\n\nclass CryptoException(Exception):\n    pass\n\n\nclass CryptoUtil:\n\n    GPG_KEY_TYPE = \"RSA\"\n    DEFAULT_WORDS_IN_RANDOM_ID = 8\n\n    # All reply keypairs will be \"created\" on the same day SecureDrop (then\n    # Strongbox) was publicly released for the first time.\n    # https://www.newyorker.com/news/news-desk/strongbox-and-aaron-swartz\n    DEFAULT_KEY_CREATION_DATE = date(2013, 5, 14)\n\n    # '0' is the magic value that tells GPG's batch key generation not\n    # to set an expiration date.\n    DEFAULT_KEY_EXPIRATION_DATE = '0'\n\n    def __init__(self,\n                 scrypt_params,\n                 scrypt_id_pepper,\n                 scrypt_gpg_pepper,\n                 securedrop_root,\n                 word_list,\n                 nouns_file,\n                 adjectives_file,\n                 gpg_key_dir):\n        self.__securedrop_root = securedrop_root\n        self.__word_list = word_list\n\n        if os.environ.get('SECUREDROP_ENV') == 'test':\n            # Optimize crypto to speed up tests (at the expense of security\n            # DO NOT use these settings in production)\n            self.__gpg_key_length = 1024\n            self.scrypt_params = dict(N=2**1, r=1, p=1)\n        else:  # pragma: no cover\n            self.__gpg_key_length = 4096\n            self.scrypt_params = scrypt_params\n\n        self.scrypt_id_pepper = scrypt_id_pepper\n        self.scrypt_gpg_pepper = scrypt_gpg_pepper\n\n        self.do_runtime_tests()\n\n        # --pinentry-mode, required for SecureDrop on gpg 2.1.x+, was\n        # added in gpg 2.1.\n        self.gpg_key_dir = gpg_key_dir\n        gpg_binary = gnupg.GPG(binary='gpg2', homedir=self.gpg_key_dir)\n        if StrictVersion(gpg_binary.binary_version) >= StrictVersion('2.1'):\n            self.gpg = gnupg.GPG(binary='gpg2',\n                                 homedir=gpg_key_dir,\n                                 options=['--pinentry-mode loopback'])\n        else:\n            self.gpg = gpg_binary\n\n        # map code for a given language to a localized wordlist\n        self.__language2words = {}  # type: Dict[Text, List[str]]\n\n        with io.open(nouns_file) as f:\n            self.nouns = f.read().splitlines()\n\n        with io.open(adjectives_file) as f:\n            self.adjectives = f.read().splitlines()\n\n    # Make sure these pass before the app can run\n    def do_runtime_tests(self):\n        if self.scrypt_id_pepper == self.scrypt_gpg_pepper:\n            raise AssertionError('scrypt_id_pepper == scrypt_gpg_pepper')\n        # crash if we don't have srm:\n        try:\n            subprocess.check_call(['srm'], stdout=subprocess.PIPE)\n        except subprocess.CalledProcessError:\n            pass\n\n    def get_wordlist(self, locale):\n        # type: (Text) -> List[str]\n        \"\"\"\" Ensure the wordlist for the desired locale is read and available\n        in the words global variable. If there is no wordlist for the\n        desired local, fallback to the default english wordlist.\n\n        The localized wordlist are read from wordlists/{locale}.txt but\n        for backward compatibility purposes the english wordlist is read\n        from the config.WORD_LIST file.\n        \"\"\"\n\n        if locale not in self.__language2words:\n            if locale != 'en':\n                path = os.path.join(self.__securedrop_root,\n                                    'wordlists',\n                                    locale + '.txt')\n                if os.path.exists(path):\n                    wordlist_path = path\n                else:\n                    wordlist_path = self.__word_list\n            else:\n                wordlist_path = self.__word_list\n\n            with io.open(wordlist_path) as f:\n                content = f.read().splitlines()\n                self.__language2words[locale] = content\n\n        return self.__language2words[locale]\n\n    def genrandomid(self,\n                    words_in_random_id=None,\n                    locale='en'):\n        if words_in_random_id is None:\n            words_in_random_id = self.DEFAULT_WORDS_IN_RANDOM_ID\n        return ' '.join(random.choice(self.get_wordlist(locale))\n                        for x in range(words_in_random_id))\n\n    def display_id(self):\n        return ' '.join([random.choice(self.adjectives),\n                         random.choice(self.nouns)])\n\n    def hash_codename(self, codename, salt=None):\n        \"\"\"Salts and hashes a codename using scrypt.\n\n        :param str codename: A source's codename.\n        :param str salt: The salt to mix with the codename when hashing.\n        :returns: A base32 encoded string; the salted codename hash.\n        \"\"\"\n        if salt is None:\n            salt = self.scrypt_id_pepper\n        return b32encode(scrypt.hash(clean(codename),\n                         salt,\n                         **self.scrypt_params)).decode('utf-8')\n\n    def genkeypair(self, name, secret):\n        \"\"\"Generate a GPG key through batch file key generation. A source's\n        codename is salted with SCRYPT_GPG_PEPPER and hashed with scrypt to\n        provide the passphrase used to encrypt their private key. Their name\n        should be their filesystem id.\n\n        >>> if not gpg.list_keys(hash_codename('randomid')):\n        ...     genkeypair(hash_codename('randomid'), 'randomid').type\n        ... else:\n        ...     u'P'\n        u'P'\n\n        :param str name: The source's filesystem id (their codename, salted\n                         with SCRYPT_ID_PEPPER, and hashed with scrypt).\n        :param str secret: The source's codename.\n        :returns: a :class:`GenKey <gnupg._parser.GenKey>` object, on which\n                  the ``__str__()`` method may be called to return the\n                  generated key's fingeprint.\n\n        \"\"\"\n        name = clean(name)\n        secret = self.hash_codename(secret, salt=self.scrypt_gpg_pepper)\n        genkey_obj = self.gpg.gen_key(self.gpg.gen_key_input(\n            key_type=self.GPG_KEY_TYPE,\n            key_length=self.__gpg_key_length,\n            passphrase=secret,\n            name_email=name,\n            creation_date=self.DEFAULT_KEY_CREATION_DATE.isoformat(),\n            expire_date=self.DEFAULT_KEY_EXPIRATION_DATE\n        ))\n        return genkey_obj\n\n    def delete_reply_keypair(self, source_filesystem_id):\n        key = self.getkey(source_filesystem_id)\n        # If this source was never flagged for review, they won't have a reply\n        # keypair\n        if not key:\n            return\n\n        # Always delete keys without invoking pinentry-mode = loopback\n        # see: https://lists.gnupg.org/pipermail/gnupg-users/2016-May/055965.html\n        temp_gpg = gnupg.GPG(binary='gpg2', homedir=self.gpg_key_dir)\n        # The subkeys keyword argument deletes both secret and public keys.\n        temp_gpg.delete_keys(key, secret=True, subkeys=True)\n\n    def getkey(self, name):\n        for key in self.gpg.list_keys():\n            for uid in key['uids']:\n                if name in uid:\n                    return key['fingerprint']\n        return None\n\n    def export_pubkey(self, name):\n        fingerprint = self.getkey(name)\n        if fingerprint:\n            return self.gpg.export_keys(fingerprint)\n        else:\n            return None\n\n    def encrypt(self, plaintext, fingerprints, output=None):\n        # Verify the output path\n        if output:\n            current_app.storage.verify(output)\n\n        if not isinstance(fingerprints, (list, tuple)):\n            fingerprints = [fingerprints, ]\n        # Remove any spaces from provided fingerprints GPG outputs fingerprints\n        # with spaces for readability, but requires the spaces to be removed\n        # when using fingerprints to specify recipients.\n        fingerprints = [fpr.replace(' ', '') for fpr in fingerprints]\n\n        if not _is_stream(plaintext):\n            plaintext = _make_binary_stream(plaintext, \"utf_8\")\n\n        out = self.gpg.encrypt(plaintext,\n                               *fingerprints,\n                               output=output,\n                               always_trust=True,\n                               armor=False)\n        if out.ok:\n            return out.data\n        else:\n            raise CryptoException(out.stderr)\n\n    def decrypt(self, secret, ciphertext):\n        \"\"\"\n        >>> crypto = current_app.crypto_util\n        >>> key = crypto.genkeypair('randomid', 'randomid')\n        >>> message = u'Buenos d\u00edas, mundo hermoso!'\n        >>> ciphertext = crypto.encrypt(message, str(key))\n        >>> crypto.decrypt('randomid', ciphertext) == message.encode('utf-8')\n        True\n        \"\"\"\n        hashed_codename = self.hash_codename(secret,\n                                             salt=self.scrypt_gpg_pepper)\n        data = self.gpg.decrypt(ciphertext, passphrase=hashed_codename).data\n\n        if not six.PY2:  # Python3\n            return data.decode('utf-8')\n        return data\n\n\ndef clean(s, also=''):\n    \"\"\"\n    >>> clean(\"[]\")\n    Traceback (most recent call last):\n      ...\n    CryptoException: invalid input: []\n    >>> clean(\"Helloworld\")\n    'Helloworld'\n    \"\"\"\n    for c in s:\n        if c not in DICEWARE_SAFE_CHARS and c not in also:\n            raise CryptoException(\"invalid input: {0}\".format(s))\n    # scrypt.hash requires input of type str. Since the wordlist is all ASCII\n    # characters, this conversion is not problematic\n    return str(s)\n", "path": "securedrop/crypto_util.py"}], "after_files": [{"content": "#!/usr/bin/env python\n# -*- coding: utf-8 -*-\n\nfrom distutils.version import StrictVersion\nimport pretty_bad_protocol as gnupg\nimport os\nimport io\nimport six\nimport scrypt\nimport subprocess\nfrom random import SystemRandom\n\nfrom base64 import b32encode\nfrom datetime import date\nfrom flask import current_app\nfrom pretty_bad_protocol._util import _is_stream, _make_binary_stream\n\nimport typing\n# https://www.python.org/dev/peps/pep-0484/#runtime-or-type-checking\nif typing.TYPE_CHECKING:\n    # flake8 can not understand type annotation yet.\n    # That is why all type annotation relative import\n    # statements has to be marked as noqa.\n    # http://flake8.pycqa.org/en/latest/user/error-codes.html?highlight=f401stream\n    from typing import Dict, List, Text  # noqa: F401\n\n# to fix gpg error #78 on production\nos.environ['USERNAME'] = 'www-data'\n\n# SystemRandom sources from the system rand (e.g. urandom, CryptGenRandom, etc)\n# It supplies a CSPRNG but with an interface that supports methods like choice\nrandom = SystemRandom()\n\n# safe characters for every possible word in the wordlist includes capital\n# letters because codename hashes are base32-encoded with capital letters\nDICEWARE_SAFE_CHARS = (' !#%$&)(+*-1032547698;:=?@acbedgfihkjmlonqpsrutwvyxzA'\n                       'BCDEFGHIJKLMNOPQRSTUVWXYZ')\n\n\ndef monkey_patch_delete_handle_status(self, key, value):\n        \"\"\"Parse a status code from the attached GnuPG process.\n        :raises: :exc:`~exceptions.ValueError` if the status message is unknown.\n        \"\"\"\n        if key in (\"DELETE_PROBLEM\", \"KEY_CONSIDERED\"):\n            self.status = self.problem_reason.get(value, \"Unknown error: %r\"\n                                                  % value)\n        elif key in (\"PINENTRY_LAUNCHED\"):\n            self.status = key.replace(\"_\", \" \").lower()\n        else:\n            raise ValueError(\"Unknown status message: %r\" % key)\n\n\n# Monkey patching to resolve https://github.com/freedomofpress/securedrop/issues/4294\ngnupg._parsers.DeleteResult._handle_status = monkey_patch_delete_handle_status\n\n\nclass CryptoException(Exception):\n    pass\n\n\nclass CryptoUtil:\n\n    GPG_KEY_TYPE = \"RSA\"\n    DEFAULT_WORDS_IN_RANDOM_ID = 8\n\n    # All reply keypairs will be \"created\" on the same day SecureDrop (then\n    # Strongbox) was publicly released for the first time.\n    # https://www.newyorker.com/news/news-desk/strongbox-and-aaron-swartz\n    DEFAULT_KEY_CREATION_DATE = date(2013, 5, 14)\n\n    # '0' is the magic value that tells GPG's batch key generation not\n    # to set an expiration date.\n    DEFAULT_KEY_EXPIRATION_DATE = '0'\n\n    def __init__(self,\n                 scrypt_params,\n                 scrypt_id_pepper,\n                 scrypt_gpg_pepper,\n                 securedrop_root,\n                 word_list,\n                 nouns_file,\n                 adjectives_file,\n                 gpg_key_dir):\n        self.__securedrop_root = securedrop_root\n        self.__word_list = word_list\n\n        if os.environ.get('SECUREDROP_ENV') == 'test':\n            # Optimize crypto to speed up tests (at the expense of security\n            # DO NOT use these settings in production)\n            self.__gpg_key_length = 1024\n            self.scrypt_params = dict(N=2**1, r=1, p=1)\n        else:  # pragma: no cover\n            self.__gpg_key_length = 4096\n            self.scrypt_params = scrypt_params\n\n        self.scrypt_id_pepper = scrypt_id_pepper\n        self.scrypt_gpg_pepper = scrypt_gpg_pepper\n\n        self.do_runtime_tests()\n\n        # --pinentry-mode, required for SecureDrop on gpg 2.1.x+, was\n        # added in gpg 2.1.\n        self.gpg_key_dir = gpg_key_dir\n        gpg_binary = gnupg.GPG(binary='gpg2', homedir=self.gpg_key_dir)\n        if StrictVersion(gpg_binary.binary_version) >= StrictVersion('2.1'):\n            self.gpg = gnupg.GPG(binary='gpg2',\n                                 homedir=gpg_key_dir,\n                                 options=['--pinentry-mode loopback'])\n        else:\n            self.gpg = gpg_binary\n\n        # map code for a given language to a localized wordlist\n        self.__language2words = {}  # type: Dict[Text, List[str]]\n\n        with io.open(nouns_file) as f:\n            self.nouns = f.read().splitlines()\n\n        with io.open(adjectives_file) as f:\n            self.adjectives = f.read().splitlines()\n\n    # Make sure these pass before the app can run\n    def do_runtime_tests(self):\n        if self.scrypt_id_pepper == self.scrypt_gpg_pepper:\n            raise AssertionError('scrypt_id_pepper == scrypt_gpg_pepper')\n        # crash if we don't have srm:\n        try:\n            subprocess.check_call(['srm'], stdout=subprocess.PIPE)\n        except subprocess.CalledProcessError:\n            pass\n\n    def get_wordlist(self, locale):\n        # type: (Text) -> List[str]\n        \"\"\"\" Ensure the wordlist for the desired locale is read and available\n        in the words global variable. If there is no wordlist for the\n        desired local, fallback to the default english wordlist.\n\n        The localized wordlist are read from wordlists/{locale}.txt but\n        for backward compatibility purposes the english wordlist is read\n        from the config.WORD_LIST file.\n        \"\"\"\n\n        if locale not in self.__language2words:\n            if locale != 'en':\n                path = os.path.join(self.__securedrop_root,\n                                    'wordlists',\n                                    locale + '.txt')\n                if os.path.exists(path):\n                    wordlist_path = path\n                else:\n                    wordlist_path = self.__word_list\n            else:\n                wordlist_path = self.__word_list\n\n            with io.open(wordlist_path) as f:\n                content = f.read().splitlines()\n                self.__language2words[locale] = content\n\n        return self.__language2words[locale]\n\n    def genrandomid(self,\n                    words_in_random_id=None,\n                    locale='en'):\n        if words_in_random_id is None:\n            words_in_random_id = self.DEFAULT_WORDS_IN_RANDOM_ID\n        return ' '.join(random.choice(self.get_wordlist(locale))\n                        for x in range(words_in_random_id))\n\n    def display_id(self):\n        return ' '.join([random.choice(self.adjectives),\n                         random.choice(self.nouns)])\n\n    def hash_codename(self, codename, salt=None):\n        \"\"\"Salts and hashes a codename using scrypt.\n\n        :param str codename: A source's codename.\n        :param str salt: The salt to mix with the codename when hashing.\n        :returns: A base32 encoded string; the salted codename hash.\n        \"\"\"\n        if salt is None:\n            salt = self.scrypt_id_pepper\n        return b32encode(scrypt.hash(clean(codename),\n                         salt,\n                         **self.scrypt_params)).decode('utf-8')\n\n    def genkeypair(self, name, secret):\n        \"\"\"Generate a GPG key through batch file key generation. A source's\n        codename is salted with SCRYPT_GPG_PEPPER and hashed with scrypt to\n        provide the passphrase used to encrypt their private key. Their name\n        should be their filesystem id.\n\n        >>> if not gpg.list_keys(hash_codename('randomid')):\n        ...     genkeypair(hash_codename('randomid'), 'randomid').type\n        ... else:\n        ...     u'P'\n        u'P'\n\n        :param str name: The source's filesystem id (their codename, salted\n                         with SCRYPT_ID_PEPPER, and hashed with scrypt).\n        :param str secret: The source's codename.\n        :returns: a :class:`GenKey <gnupg._parser.GenKey>` object, on which\n                  the ``__str__()`` method may be called to return the\n                  generated key's fingeprint.\n\n        \"\"\"\n        name = clean(name)\n        secret = self.hash_codename(secret, salt=self.scrypt_gpg_pepper)\n        genkey_obj = self.gpg.gen_key(self.gpg.gen_key_input(\n            key_type=self.GPG_KEY_TYPE,\n            key_length=self.__gpg_key_length,\n            passphrase=secret,\n            name_email=name,\n            creation_date=self.DEFAULT_KEY_CREATION_DATE.isoformat(),\n            expire_date=self.DEFAULT_KEY_EXPIRATION_DATE\n        ))\n        return genkey_obj\n\n    def delete_reply_keypair(self, source_filesystem_id):\n        key = self.getkey(source_filesystem_id)\n        # If this source was never flagged for review, they won't have a reply\n        # keypair\n        if not key:\n            return\n\n        # Always delete keys without invoking pinentry-mode = loopback\n        # see: https://lists.gnupg.org/pipermail/gnupg-users/2016-May/055965.html\n        temp_gpg = gnupg.GPG(binary='gpg2', homedir=self.gpg_key_dir)\n        # The subkeys keyword argument deletes both secret and public keys.\n        temp_gpg.delete_keys(key, secret=True, subkeys=True)\n\n    def getkey(self, name):\n        for key in self.gpg.list_keys():\n            for uid in key['uids']:\n                if name in uid:\n                    return key['fingerprint']\n        return None\n\n    def export_pubkey(self, name):\n        fingerprint = self.getkey(name)\n        if fingerprint:\n            return self.gpg.export_keys(fingerprint)\n        else:\n            return None\n\n    def encrypt(self, plaintext, fingerprints, output=None):\n        # Verify the output path\n        if output:\n            current_app.storage.verify(output)\n\n        if not isinstance(fingerprints, (list, tuple)):\n            fingerprints = [fingerprints, ]\n        # Remove any spaces from provided fingerprints GPG outputs fingerprints\n        # with spaces for readability, but requires the spaces to be removed\n        # when using fingerprints to specify recipients.\n        fingerprints = [fpr.replace(' ', '') for fpr in fingerprints]\n\n        if not _is_stream(plaintext):\n            plaintext = _make_binary_stream(plaintext, \"utf_8\")\n\n        out = self.gpg.encrypt(plaintext,\n                               *fingerprints,\n                               output=output,\n                               always_trust=True,\n                               armor=False)\n        if out.ok:\n            return out.data\n        else:\n            raise CryptoException(out.stderr)\n\n    def decrypt(self, secret, ciphertext):\n        \"\"\"\n        >>> crypto = current_app.crypto_util\n        >>> key = crypto.genkeypair('randomid', 'randomid')\n        >>> message = u'Buenos d\u00edas, mundo hermoso!'\n        >>> ciphertext = crypto.encrypt(message, str(key))\n        >>> crypto.decrypt('randomid', ciphertext) == message.encode('utf-8')\n        True\n        \"\"\"\n        hashed_codename = self.hash_codename(secret,\n                                             salt=self.scrypt_gpg_pepper)\n        data = self.gpg.decrypt(ciphertext, passphrase=hashed_codename).data\n\n        if not six.PY2:  # Python3\n            return data.decode('utf-8')\n        return data\n\n\ndef clean(s, also=''):\n    \"\"\"\n    >>> clean(\"[]\")\n    Traceback (most recent call last):\n      ...\n    CryptoException: invalid input: []\n    >>> clean(\"Helloworld\")\n    'Helloworld'\n    \"\"\"\n    for c in s:\n        if c not in DICEWARE_SAFE_CHARS and c not in also:\n            raise CryptoException(\"invalid input: {0}\".format(s))\n    # scrypt.hash requires input of type str. Since the wordlist is all ASCII\n    # characters, this conversion is not problematic\n    return str(s)\n", "path": "securedrop/crypto_util.py"}]}