Move passphrase logic from CryptoUtil to PassphrasesGenerator

Move passphrase logic from CryptoUtil to PassphrasesGenerator

Move passphrase logic from CryptoUtil to PassphrasesGenerator

Re-add generate test

Ensure passphrases are random

Fix pylint false positive

securedrop/create-dev-data.py

@@ -10,6 +10,8 @@
 
 from flask import current_app
 
+from passphrases import PassphrasesGenerator
+
 os.environ["SECUREDROP_ENV"] = "dev"  # noqa
 import journalist_app
 
@@ -104,7 +106,7 @@ def create_source_data(
     num_replies: int = 2,
 ) -> None:
     # Store source in database
-    codename = current_app.crypto_util.genrandomid()
+    codename = PassphrasesGenerator.get_default().generate_passphrase()
     filesystem_id = current_app.crypto_util.hash_codename(codename)
     journalist_designation = current_app.crypto_util.display_id()
     source = Source(filesystem_id, journalist_designation)

securedrop/crypto_util.py

@@ -25,6 +25,8 @@
 
 
 # to fix GPG error #78 on production
+from passphrases import DicewarePassphrase
+
 os.environ['USERNAME'] = 'www-data'
 
 # SystemRandom sources from the system rand (e.g. urandom, CryptGenRandom, etc)
@@ -84,12 +86,10 @@ def __init__(self,
                  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
 
         if os.environ.get('SECUREDROP_ENV') in ('dev', 'test'):
             # Optimize crypto to speed up tests (at the expense of security
@@ -116,9 +116,6 @@ def __init__(self,
         else:
             self.gpg = gpg_binary
 
-        # map code for a given language to a localized wordlist
-        self.__language2words = {}  # type: Dict[str, List[str]]
-
         with io.open(nouns_file) as f:
             self.nouns = f.read().splitlines()
 
@@ -135,40 +132,6 @@ def do_runtime_tests(self) -> None:
         if not rm.check_secure_delete_capability():
             raise AssertionError("Secure file deletion is not possible.")
 
-    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.
-
-        The localized wordlist are read from wordlists/{locale}.txt but
-        for backward compatibility purposes the english wordlist is read
-        from the config.WORD_LIST file.
-        """
-
-        if locale not in self.__language2words:
-            if locale != 'en':
-                path = os.path.join(self.__securedrop_root,
-                                    'wordlists',
-                                    locale + '.txt')
-                if os.path.exists(path):
-                    wordlist_path = path
-                else:
-                    wordlist_path = self.__word_list
-            else:
-                wordlist_path = self.__word_list
-
-            with io.open(wordlist_path) as f:
-                content = f.read().splitlines()
-                self.__language2words[locale] = content
-
-        return self.__language2words[locale]
-
-    def genrandomid(self,
-                    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) -> str:
         """Generate random journalist_designation until we get an unused one"""
 
@@ -186,7 +149,7 @@ def display_id(self) -> str:
 
         raise ValueError("Could not generate unique journalist designation for new source")
 
-    def hash_codename(self, codename: str, salt: Optional[str] = None) -> str:
+    def hash_codename(self, codename: DicewarePassphrase, salt: Optional[str] = None) -> str:
         """Salts and hashes a codename using scrypt.
 
         :param codename: A source's codename.
@@ -195,10 +158,9 @@ def hash_codename(self, codename: str, salt: Optional[str] = None) -> str:
         """
         if salt is None:
             salt = self.scrypt_id_pepper
-        _validate_name_for_diceware(codename)
         return b32encode(scrypt.hash(codename, salt, **self.scrypt_params)).decode('utf-8')
 
-    def genkeypair(self, name: str, secret: str) -> gnupg._parsers.GenKey:
+    def genkeypair(self, name: str, secret: DicewarePassphrase) -> 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
@@ -219,11 +181,11 @@ def genkeypair(self, name: str, secret: str) -> gnupg._parsers.GenKey:
 
         """
         _validate_name_for_diceware(name)
-        secret = self.hash_codename(secret, salt=self.scrypt_gpg_pepper)
+        hashed_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,
             key_length=self.__gpg_key_length,
-            passphrase=secret,
+            passphrase=hashed_secret,
             name_email=name,
             name_real="Source Key",
             creation_date=self.DEFAULT_KEY_CREATION_DATE.isoformat(),
@@ -333,7 +295,7 @@ def encrypt(self, plaintext: str, fingerprints: List[str], output: Optional[str]
         else:
             raise CryptoException(out.stderr)
 
-    def decrypt(self, secret: str, ciphertext: bytes) -> str:
+    def decrypt(self, secret: DicewarePassphrase, ciphertext: bytes) -> str:
         """
         >>> crypto = current_app.crypto_util
         >>> key = crypto.genkeypair('randomid', 'randomid')

securedrop/journalist_app/__init__.py

@@ -70,7 +70,6 @@ def create_app(config: 'SDConfig') -> Flask:
         scrypt_id_pepper=config.SCRYPT_ID_PEPPER,
         scrypt_gpg_pepper=config.SCRYPT_GPG_PEPPER,
         securedrop_root=config.SECUREDROP_ROOT,
-        word_list=config.WORD_LIST,
         nouns_file=config.NOUNS,
         adjectives_file=config.ADJECTIVES,
         gpg_key_dir=config.GPG_KEY_DIR,

securedrop/journalist_app/account.py

@@ -6,9 +6,11 @@
                    flash, session)
 from flask_babel import gettext
 
+import i18n
 from db import db
-from journalist_app.utils import (make_password, set_diceware_password, set_name, validate_user,
+from journalist_app.utils import (set_diceware_password, set_name, validate_user,
                                   validate_hotp_secret)
+from passphrases import PassphrasesGenerator
 from sdconfig import SDConfig
 
 
@@ -17,7 +19,9 @@ def make_blueprint(config: SDConfig) -> Blueprint:
 
     @view.route('/account', methods=('GET',))
     def edit() -> str:
-        password = make_password(config)
+        password = PassphrasesGenerator.get_default().generate_passphrase(
+            preferred_language=i18n.get_language(config)
+        )
         return render_template('edit_account.html',
                                password=password)
 

securedrop/journalist_app/admin.py

@@ -11,14 +11,16 @@
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm.exc import NoResultFound
 
+import i18n
 from db import db
 from models import (InstanceConfig, Journalist, InvalidUsernameException,
                     FirstOrLastNameError, PasswordError)
 from journalist_app.decorators import admin_required
-from journalist_app.utils import (make_password, commit_account_changes, set_diceware_password,
+from journalist_app.utils import (commit_account_changes, set_diceware_password,
                                   validate_hotp_secret, revoke_token)
 from journalist_app.forms import LogoForm, NewUserForm, SubmissionPreferencesForm
 from sdconfig import SDConfig
+from passphrases import PassphrasesGenerator
 
 
 def make_blueprint(config: SDConfig) -> Blueprint:
@@ -124,8 +126,11 @@ def add_user() -> Union[str, werkzeug.Response]:
                 return redirect(url_for('admin.new_user_two_factor',
                                         uid=new_user.id))
 
+        password = PassphrasesGenerator.get_default().generate_passphrase(
+            preferred_language=i18n.get_language(config)
+        )
         return render_template("admin_add_user.html",
-                               password=make_password(config),
+                               password=password,
                                form=form)
 
     @view.route('/2fa', methods=('GET', 'POST'))
@@ -221,7 +226,9 @@ def edit_user(user_id: int) -> Union[str, werkzeug.Response]:
 
             commit_account_changes(user)
 
-        password = make_password(config)
+        password = PassphrasesGenerator.get_default().generate_passphrase(
+            preferred_language=i18n.get_language(config)
+        )
         return render_template("edit_account.html", user=user,
                                password=password)
 

securedrop/journalist_app/utils.py

@@ -11,8 +11,6 @@
 from flask_babel import gettext, ngettext
 from sqlalchemy.exc import IntegrityError
 
-import i18n
-
 from db import db
 from models import (
     BadTokenException,
@@ -35,8 +33,6 @@
 )
 from store import add_checksum_for_file
 
-from sdconfig import SDConfig
-
 
 def logged_in() -> bool:
     # When a user is logged in, we push their user ID (database primary key)
@@ -313,18 +309,6 @@ def col_delete(cols_selected: List[str]) -> werkzeug.Response:
     return redirect(url_for('main.index'))
 
 
-def make_password(config: SDConfig) -> str:
-    while True:
-        password = current_app.crypto_util.genrandomid(
-            7,
-            i18n.get_language(config))
-        try:
-            Journalist.check_password_acceptable(password)
-            return password
-        except PasswordError:
-            continue
-
-
 def delete_collection(filesystem_id: str) -> None:
     # Delete the source's collection of submissions
     path = current_app.storage.path(filesystem_id)

securedrop/manage.py

@@ -17,10 +17,11 @@
 from flask.ctx import AppContext
 from typing import List
 
+from passphrases import PassphrasesGenerator
+
 sys.path.insert(0, "/var/www/securedrop")  # noqa: E402
 
 import qrcode
-from flask import current_app
 from sqlalchemy.orm.exc import NoResultFound
 
 os.environ['SECUREDROP_ENV'] = 'dev'  # noqa
@@ -30,7 +31,6 @@
     FirstOrLastNameError,
     InvalidUsernameException,
     Journalist,
-    PasswordError,
 )
 from management import app_context, config
 from management.run import run
@@ -177,24 +177,14 @@ def _get_yubikey_usage() -> bool:
             print('Invalid answer. Please type "y" or "n"')
 
 
-def _make_password() -> str:
-    while True:
-        password = current_app.crypto_util.genrandomid(7)
-        try:
-            Journalist.check_password_acceptable(password)
-            return password
-        except PasswordError:
-            continue
-
-
 def _add_user(is_admin: bool = False, context: Optional[AppContext] = None) -> int:
     with context or app_context():
         username = _get_username()
         first_name = _get_first_name()
         last_name = _get_last_name()
 
         print("Note: Passwords are now autogenerated.")
-        password = _make_password()
+        password = PassphrasesGenerator.get_default().generate_passphrase()
         print("This user's password is: {}".format(password))
 
         is_hotp = _get_yubikey_usage()

securedrop/models.py

@@ -77,10 +77,6 @@ class Source(db.Model):
     # when deletion of the source was requested
     deleted_at = Column(DateTime)
 
-    # Don't create or bother checking excessively long codenames to prevent DoS
-    NUM_WORDS = 7
-    MAX_CODENAME_LEN = 128
-
     def __init__(self,
                  filesystem_id: 'Optional[str]' = None,
                  journalist_designation: 'Optional[str]' = None) -> None: