Add type annotations to crypto_utils and config

mypy.ini

@@ -12,5 +12,11 @@ disallow_untyped_defs = True
 [mypy-journalist_app.api]
 disallow_untyped_defs = True
 
+[mypy-crypto_util]
+disallow_untyped_defs = True
+
+[mypy-sdconfig]
+disallow_untyped_defs = True
+
 [mypy-config]
 ignore_errors = True

securedrop/crypto_util.py

@@ -1,6 +1,8 @@
 # -*- coding: utf-8 -*-
 
 from distutils.version import StrictVersion
+from typing import Optional
+
 import pretty_bad_protocol as gnupg
 import os
 import io
@@ -13,20 +15,14 @@
 from flask import current_app
 from pretty_bad_protocol._util import _is_stream, _make_binary_stream
 from redis import Redis
+from typing import List
+
+from typing import Dict
 
 import rm
 
 from models import Source
 
-import typing
-# https://www.python.org/dev/peps/pep-0484/#runtime-or-type-checking
-if typing.TYPE_CHECKING:
-    # flake8 can not understand type annotation yet.
-    # That is why all type annotation relative import
-    # statements has to be marked as noqa.
-    # http://flake8.pycqa.org/en/latest/user/error-codes.html?highlight=f401stream
-    from typing import Dict, List, Text  # noqa: F401
-
 
 # to fix GPG error #78 on production
 os.environ['USERNAME'] = 'www-data'
@@ -41,7 +37,9 @@
                        'BCDEFGHIJKLMNOPQRSTUVWXYZ')
 
 
-def monkey_patch_delete_handle_status(self, key, value):
+def monkey_patch_delete_handle_status(
+    self: gnupg._parsers.DeleteResult, key: str, value: str
+) -> None:
     """
     Parse a status code from the attached GnuPG process.
     :raises: :exc:`~exceptions.ValueError` if the status message is unknown.
@@ -82,14 +80,14 @@ class CryptoUtil:
     SOURCE_KEY_UID_RE = re.compile(r"(Source|Autogenerated) Key <[-A-Za-z0-9+/=_]+>")
 
     def __init__(self,
-                 scrypt_params,
-                 scrypt_id_pepper,
-                 scrypt_gpg_pepper,
-                 securedrop_root,
-                 word_list,
-                 nouns_file,
-                 adjectives_file,
-                 gpg_key_dir):
+                 scrypt_params: Dict[str, int],
+                 scrypt_id_pepper: str,
+                 scrypt_gpg_pepper: str,
+                 securedrop_root: str,
+                 word_list: str,
+                 nouns_file: str,
+                 adjectives_file: str,
+                 gpg_key_dir: str) -> None:
         self.__securedrop_root = securedrop_root
         self.__word_list = word_list
 
@@ -119,7 +117,7 @@ def __init__(self,
             self.gpg = gpg_binary
 
         # map code for a given language to a localized wordlist
-        self.__language2words = {}  # type: Dict[Text, List[str]]
+        self.__language2words = {}  # type: Dict[str, List[str]]
 
         with io.open(nouns_file) as f:
             self.nouns = f.read().splitlines()
@@ -130,14 +128,14 @@ def __init__(self,
         self.redis = Redis(decode_responses=True)
 
     # Make sure these pass before the app can run
-    def do_runtime_tests(self):
+    def do_runtime_tests(self) -> None:
         if self.scrypt_id_pepper == self.scrypt_gpg_pepper:
             raise AssertionError('scrypt_id_pepper == scrypt_gpg_pepper')
         # crash if we don't have a way to securely remove files
         if not rm.check_secure_delete_capability():
             raise AssertionError("Secure file deletion is not possible.")
 
-    def get_wordlist(self, locale: 'Text') -> 'List[str]':
+    def get_wordlist(self, locale: str) -> List[str]:
         """" Ensure the wordlist for the desired locale is read and available
         in the words global variable. If there is no wordlist for the
         desired local, fallback to the default english wordlist.
@@ -166,14 +164,12 @@ def get_wordlist(self, locale: 'Text') -> 'List[str]':
         return self.__language2words[locale]
 
     def genrandomid(self,
-                    words_in_random_id=None,
-                    locale='en'):
-        if words_in_random_id is None:
-            words_in_random_id = self.DEFAULT_WORDS_IN_RANDOM_ID
+                    words_in_random_id: int = DEFAULT_WORDS_IN_RANDOM_ID,
+                    locale: str = 'en') -> str:
         return ' '.join(random.choice(self.get_wordlist(locale))
                         for x in range(words_in_random_id))
 
-    def display_id(self):
+    def display_id(self) -> str:
         """Generate random journalist_designation until we get an unused one"""
 
         tries = 0
@@ -190,20 +186,19 @@ def display_id(self):
 
         raise ValueError("Could not generate unique journalist designation for new source")
 
-    def hash_codename(self, codename, salt=None):
+    def hash_codename(self, codename: str, salt: Optional[str] = None) -> str:
         """Salts and hashes a codename using scrypt.
 
-        :param str codename: A source's codename.
-        :param str salt: The salt to mix with the codename when hashing.
+        :param codename: A source's codename.
+        :param salt: The salt to mix with the codename when hashing.
         :returns: A base32 encoded string; the salted codename hash.
         """
         if salt is None:
             salt = self.scrypt_id_pepper
-        return b32encode(scrypt.hash(clean(codename),
-                         salt,
-                         **self.scrypt_params)).decode('utf-8')
+        _validate_name_for_diceware(codename)
+        return b32encode(scrypt.hash(codename, salt, **self.scrypt_params)).decode('utf-8')
 
-    def genkeypair(self, name, secret):
+    def genkeypair(self, name: str, secret: str) -> gnupg._parsers.GenKey:
         """Generate a GPG key through batch file key generation. A source's
         codename is salted with SCRYPT_GPG_PEPPER and hashed with scrypt to
         provide the passphrase used to encrypt their private key. Their name
@@ -215,15 +210,15 @@ def genkeypair(self, name, secret):
         ...     u'P'
         u'P'
 
-        :param str name: The source's filesystem id (their codename, salted
+        :param name: The source's filesystem id (their codename, salted
                          with SCRYPT_ID_PEPPER, and hashed with scrypt).
-        :param str secret: The source's codename.
+        :param secret: The source's codename.
         :returns: a :class:`GenKey <gnupg._parser.GenKey>` object, on which
                   the ``__str__()`` method may be called to return the
                   generated key's fingeprint.
 
         """
-        name = clean(name)
+        _validate_name_for_diceware(name)
         secret = self.hash_codename(secret, salt=self.scrypt_gpg_pepper)
         genkey_obj = self.gpg.gen_key(self.gpg.gen_key_input(
             key_type=self.GPG_KEY_TYPE,
@@ -236,7 +231,7 @@ def genkeypair(self, name, secret):
         ))
         return genkey_obj
 
-    def find_source_key(self, fingerprint: str) -> typing.Optional[typing.Dict]:
+    def find_source_key(self, fingerprint: str) -> Optional[Dict]:
         """
         Searches the GPG keyring for a source key.
 
@@ -257,7 +252,7 @@ def find_source_key(self, fingerprint: str) -> typing.Optional[typing.Dict]:
                     return None
         return None
 
-    def delete_reply_keypair(self, source_filesystem_id):
+    def delete_reply_keypair(self, source_filesystem_id: str) -> None:
         fingerprint = self.get_fingerprint(source_filesystem_id)
 
         # If this source was never flagged for review, they won't have a reply
@@ -279,7 +274,7 @@ def delete_reply_keypair(self, source_filesystem_id):
         self.redis.hdel(self.REDIS_KEY_HASH, self.get_fingerprint(source_filesystem_id))
         self.redis.hdel(self.REDIS_FINGERPRINT_HASH, source_filesystem_id)
 
-    def get_fingerprint(self, name):
+    def get_fingerprint(self, name: str) -> Optional[str]:
         """
         Returns the fingerprint of the GPG key for the given name.
 
@@ -297,7 +292,7 @@ def get_fingerprint(self, name):
 
         return None
 
-    def get_pubkey(self, name):
+    def get_pubkey(self, name: str) -> Optional[str]:
         """
         Returns the GPG public key for the given name.
 
@@ -315,13 +310,11 @@ def get_pubkey(self, name):
         self.redis.hset(self.REDIS_KEY_HASH, fingerprint, key)
         return key
 
-    def encrypt(self, plaintext, fingerprints, output=None):
+    def encrypt(self, plaintext: str, fingerprints: List[str], output: Optional[str] = None) -> str:
         # Verify the output path
         if output:
             current_app.storage.verify(output)
 
-        if not isinstance(fingerprints, (list, tuple)):
-            fingerprints = [fingerprints, ]
         # Remove any spaces from provided fingerprints GPG outputs fingerprints
         # with spaces for readability, but requires the spaces to be removed
         # when using fingerprints to specify recipients.
@@ -340,7 +333,7 @@ def encrypt(self, plaintext, fingerprints, output=None):
         else:
             raise CryptoException(out.stderr)
 
-    def decrypt(self, secret, ciphertext):
+    def decrypt(self, secret: str, ciphertext: bytes) -> str:
         """
         >>> crypto = current_app.crypto_util
         >>> key = crypto.genkeypair('randomid', 'randomid')
@@ -356,18 +349,7 @@ def decrypt(self, secret, ciphertext):
         return data.decode('utf-8')
 
 
-def clean(s, also=''):
-    """
-    >>> clean("[]")
-    Traceback (most recent call last):
-      ...
-    CryptoException: invalid input: []
-    >>> clean("Helloworld")
-    'Helloworld'
-    """
-    for c in s:
-        if c not in DICEWARE_SAFE_CHARS and c not in also:
-            raise CryptoException("invalid input: {0}".format(s))
-    # scrypt.hash requires input of type str. Since the wordlist is all ASCII
-    # characters, this conversion is not problematic
-    return str(s)
+def _validate_name_for_diceware(name: str) -> None:
+    for char in name:
+        if char not in DICEWARE_SAFE_CHARS:
+            raise CryptoException("invalid input: {0}".format(name))

securedrop/journalist_app/__init__.py

@@ -43,36 +43,28 @@ def create_app(config: 'SDConfig') -> Flask:
                 template_folder=config.JOURNALIST_TEMPLATES_DIR,
                 static_folder=path.join(config.SECUREDROP_ROOT, 'static'))
 
-    app.config.from_object(config.JournalistInterfaceFlaskConfig)  # type: ignore
-    app.sdconfig = config
+    app.config.from_object(config.JOURNALIST_APP_FLASK_CONFIG_CLS)
     app.session_interface = JournalistInterfaceSessionInterface()
 
     csrf = CSRFProtect(app)
     Environment(app)
 
-    if config.DATABASE_ENGINE == "sqlite":
-        db_uri = (config.DATABASE_ENGINE + ":///" +
-                  config.DATABASE_FILE)
-    else:
-        db_uri = (
-            config.DATABASE_ENGINE + '://' +
-            config.DATABASE_USERNAME + ':' +
-            config.DATABASE_PASSWORD + '@' +
-            config.DATABASE_HOST + '/' +
-            config.DATABASE_NAME
-        )
     app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
-    app.config['SQLALCHEMY_DATABASE_URI'] = db_uri
+    app.config['SQLALCHEMY_DATABASE_URI'] = config.DATABASE_URI
     db.init_app(app)
 
     v2_enabled = path.exists(path.join(config.SECUREDROP_DATA_ROOT, 'source_v2_url'))
     v3_enabled = path.exists(path.join(config.SECUREDROP_DATA_ROOT, 'source_v3_url'))
     app.config.update(V2_ONION_ENABLED=v2_enabled, V3_ONION_ENABLED=v3_enabled)
 
+    # TODO: Attaching a Storage dynamically like this disables all type checking (and
+    # breaks code analysis tools) for code that uses current_app.storage; it should be refactored
     app.storage = Storage(config.STORE_DIR,
                           config.TEMP_DIR,
                           config.JOURNALIST_KEY)
 
+    # TODO: Attaching a CryptoUtil dynamically like this disables all type checking (and
+    # breaks code analysis tools) for code that uses current_app.storage; it should be refactored
     app.crypto_util = CryptoUtil(
         scrypt_params=config.SCRYPT_PARAMS,
         scrypt_id_pepper=config.SCRYPT_ID_PEPPER,