Refactor CryptoUtil.display_id() to simplify source creation logic

Author: nabla-c0d3

securedrop/crypto_util.py

@@ -5,9 +5,7 @@
 
 import pretty_bad_protocol as gnupg
 import os
-import io
 import re
-from random import SystemRandom
 
 from datetime import date
 from flask import current_app
@@ -19,8 +17,6 @@
 
 import rm
 
-from models import Source
-
 
 # monkey patch to work with Focal gnupg.
 # https://github.com/isislovecruft/python-gnupg/issues/250
@@ -31,15 +27,6 @@
 # to fix GPG error #78 on production
 os.environ['USERNAME'] = 'www-data'
 
-# SystemRandom sources from the system rand (e.g. urandom, CryptGenRandom, etc)
-# It supplies a CSPRNG but with an interface that supports methods like choice
-random = SystemRandom()
-
-# safe characters for every possible word in the wordlist includes capital
-# letters because codename hashes are base32-encoded with capital letters
-DICEWARE_SAFE_CHARS = (' !#%$&)(+*-1032547698;:=?@acbedgfihkjmlonqpsrutwvyxzA'
-                       'BCDEFGHIJKLMNOPQRSTUVWXYZ')
-
 
 def monkey_patch_delete_handle_status(
     self: gnupg._parsers.DeleteResult, key: str, value: str
@@ -84,8 +71,6 @@ class CryptoUtil:
 
     def __init__(self,
                  securedrop_root: str,
-                 nouns_file: str,
-                 adjectives_file: str,
                  gpg_key_dir: str) -> None:
         self.__securedrop_root = securedrop_root
 
@@ -109,12 +94,6 @@ def __init__(self,
         else:
             self.gpg = gpg_binary
 
-        with io.open(nouns_file) as f:
-            self.nouns = f.read().splitlines()
-
-        with io.open(adjectives_file) as f:
-            self.adjectives = f.read().splitlines()
-
         self.redis = Redis(decode_responses=True)
 
     # Make sure these pass before the app can run
@@ -123,23 +102,6 @@ def do_runtime_tests(self) -> None:
         if not rm.check_secure_delete_capability():
             raise AssertionError("Secure file deletion is not possible.")
 
-    def display_id(self) -> str:
-        """Generate random journalist_designation until we get an unused one"""
-
-        tries = 0
-
-        while tries < 50:
-            new_designation = ' '.join([random.choice(self.adjectives),
-                                        random.choice(self.nouns)])
-
-            collisions = Source.query.filter(Source.journalist_designation == new_designation)
-            if collisions.count() == 0:
-                return new_designation
-
-            tries += 1
-
-        raise ValueError("Could not generate unique journalist designation for new source")
-
     def genkeypair(self, source_user: SourceUser) -> gnupg._parsers.GenKey:
         """Generate a GPG key through batch file key generation.
         """

securedrop/journalist_app/__init__.py

@@ -80,8 +80,6 @@ def create_app(config: 'SDConfig') -> Flask:
     # breaks code analysis tools) for code that uses current_app.storage; it should be refactored
     app.crypto_util = CryptoUtil(
         securedrop_root=config.SECUREDROP_ROOT,
-        nouns_file=config.NOUNS,
-        adjectives_file=config.ADJECTIVES,
         gpg_key_dir=config.GPG_KEY_DIR,
     )
 

securedrop/loaddata.py

@@ -116,7 +116,7 @@ def set_source_count(s: str) -> int:
 
 
 def add_journalist(
-    username: str = "",
+    username: str,
     is_admin: bool = False,
     first_name: str = "",
     last_name: str = "",
@@ -128,9 +128,6 @@ def add_journalist(
     test_password = "correct horse battery staple profanity oil chewy"
     test_otp_secret = "JHCOGO7VCER3EJ4L"
 
-    if not username:
-        username = current_app.crypto_util.display_id()
-
     journalist = Journalist(
         username=username,
         password=test_password,
@@ -258,7 +255,6 @@ def add_source() -> Tuple[Source, str]:
     source_user = create_source_user(
         db_session=db.session,
         source_passphrase=codename,
-        source_app_crypto_util=current_app.crypto_util,
         source_app_storage=current_app.storage,
     )
     source = source_user.get_db_record()
@@ -319,7 +315,7 @@ def create_default_journalists() -> Tuple[Journalist, ...]:
 def add_journalists(args: argparse.Namespace) -> None:
     total = args.journalist_count
     for i in range(1, total + 1):
-        add_journalist(progress=(i, total))
+        add_journalist(username=f"journalist{str(i)}", progress=(i, total))
 
 
 def add_sources(args: argparse.Namespace, journalists: Tuple[Journalist, ...]) -> None:

securedrop/source_app/__init__.py

@@ -77,8 +77,6 @@ def setup_i18n() -> None:
     # breaks code analysis tools) for code that uses current_app.storage; it should be refactored
     app.crypto_util = CryptoUtil(
         securedrop_root=config.SECUREDROP_ROOT,
-        nouns_file=config.NOUNS,
-        adjectives_file=config.ADJECTIVES,
         gpg_key_dir=config.GPG_KEY_DIR,
     )
 

securedrop/source_app/main.py

@@ -72,7 +72,6 @@ def create() -> werkzeug.Response:
                 new_source_user = create_source_user(
                     db_session=db.session,
                     source_passphrase=codename,
-                    source_app_crypto_util=current_app.crypto_util,
                     source_app_storage=current_app.storage,
                 )
             except (SourcePassphraseCollisionError, SourceDesignationCollisionError) as e:

securedrop/source_user.py

@@ -1,7 +1,9 @@
 import os
 
 from base64 import b32encode
-from typing import Optional
+from pathlib import Path
+from random import SystemRandom
+from typing import Optional, List
 from typing import TYPE_CHECKING
 
 from cryptography.hazmat.backends import default_backend
@@ -12,7 +14,6 @@
 import models
 
 if TYPE_CHECKING:
-    from crypto_util import CryptoUtil
     from passphrases import DicewarePassphrase
     from store import Storage
 
@@ -66,23 +67,41 @@ class SourceDesignationCollisionError(Exception):
 def create_source_user(
     db_session: Session,
     source_passphrase: "DicewarePassphrase",
-    source_app_crypto_util: "CryptoUtil",
     source_app_storage: "Storage",
 ) -> SourceUser:
     # Derive the source's info from their passphrase
     scrypt_manager = _SourceScryptManager.get_default()
     filesystem_id = scrypt_manager.derive_source_filesystem_id(source_passphrase)
     gpg_secret = scrypt_manager.derive_source_gpg_secret(source_passphrase)
 
-    # Create a designation for the source
-    try:
-        # TODO: Move display_id() out of CryptoUtil
-        journalist_designation = source_app_crypto_util.display_id()
-    except ValueError:
+    # Create a unique journalist designation for the source
+    # TODO: Add unique=True to models.Source.journalist_designation to enforce uniqueness
+    #  as the logic below has a race condition (time we check VS time when we add to the DB)
+    designation_generation_attempts = 0
+    valid_designation = None
+    designation_generator = _DesignationGenerator.get_default()
+    while designation_generation_attempts < 50:
+        # Generate a designation
+        designation_generation_attempts += 1
+        new_designation = designation_generator.generate_journalist_designation()
+
+        # Check to see if it's already used by an existing source
+        existing_source_with_same_designation = db_session.query(
+            models.Source
+        ).filter_by(journalist_designation=new_designation).one_or_none()
+        if not existing_source_with_same_designation:
+            # The designation is not already used - good to go
+            valid_designation = new_designation
+            break
+
+    if not valid_designation:
+        # Could not generate a designation that is not already used
         raise SourceDesignationCollisionError()
 
     # Store the source in the DB
-    source_db_record = models.Source(filesystem_id, journalist_designation)
+    source_db_record = models.Source(
+        filesystem_id=filesystem_id, journalist_designation=valid_designation
+    )
     db_session.add(source_db_record)
     try:
         db_session.commit()
@@ -99,7 +118,7 @@ def create_source_user(
     return SourceUser(source_db_record, filesystem_id, gpg_secret)
 
 
-_default_scrypt_mgr = None  # type: Optional[_SourceScryptManager]
+_default_scrypt_mgr: Optional["_SourceScryptManager"] = None
 
 
 class _SourceScryptManager:
@@ -160,3 +179,51 @@ def get_default(cls) -> "_SourceScryptManager":
                 scrypt_p=config.SCRYPT_PARAMS["p"],
             )
         return _default_scrypt_mgr
+
+
+_default_designation_generator: Optional["_DesignationGenerator"] = None
+
+
+class _DesignationGenerator:
+
+    def __init__(self, nouns: List[str], adjectives: List[str]):
+        self._random_generator = SystemRandom()
+
+        # Ensure that there are no empty lists or empty strings
+        if not nouns:
+            raise ValueError("Nouns word list is empty")
+        shortest_noun = min(nouns, key=len)
+        shortest_noun_length = len(shortest_noun)
+        if shortest_noun_length < 1:
+            raise ValueError("Nouns word list contains an empty string")
+
+        if not adjectives:
+            raise ValueError("Adjectives word list is empty")
+        shortest_adjective = min(adjectives, key=len)
+        shortest_adjective_length = len(shortest_adjective)
+        if shortest_adjective_length < 1:
+            raise ValueError("Adjectives word list contains an empty string")
+
+        self._nouns = nouns
+        self._adjectives = adjectives
+
+    def generate_journalist_designation(self) -> str:
+        random_adjective = self._random_generator.choice(self._adjectives)
+        random_noun = self._random_generator.choice(self._nouns)
+        return f"{random_adjective} {random_noun}"
+
+    @classmethod
+    def get_default(cls) -> "_DesignationGenerator":
+        # Late import so _SourceScryptManager can be used without a config.py in the parent folder
+        from sdconfig import config
+
+        global _default_designation_generator
+        if _default_designation_generator is None:
+            # Parse the nouns and adjectives files from the config
+            nouns = Path(config.NOUNS).read_text().strip().splitlines()
+            adjectives = Path(config.ADJECTIVES).read_text().strip().splitlines()
+
+            # Create the generator
+            _default_designation_generator = cls(nouns=nouns, adjectives=adjectives)
+
+        return _default_designation_generator

securedrop/tests/conftest.py

@@ -239,7 +239,6 @@ def test_source(journalist_app: Flask) -> Dict[str, Any]:
         source_user = create_source_user(
             db_session=db.session,
             source_passphrase=passphrase,
-            source_app_crypto_util=journalist_app.crypto_util,
             source_app_storage=journalist_app.storage,
         )
         journalist_app.crypto_util.genkeypair(source_user)

securedrop/tests/functional/test_source.py

@@ -1,5 +1,6 @@
 import werkzeug
 
+import source_user
 from ..test_journalist import VALID_PASSWORD
 from . import source_navigation_steps, journalist_navigation_steps
 from . import functional_test
@@ -11,14 +12,17 @@ class TestSourceInterfaceDesignationCollision(
         source_navigation_steps.SourceNavigationStepsMixin):
 
     def start_source_server(self, source_port):
-        self.source_app.crypto_util.adjectives = \
-            self.source_app.crypto_util.adjectives[:1]
-        self.source_app.crypto_util.nouns = self.source_app.crypto_util.nouns[:1]
+        # Generator that always returns the same journalist designation
+        source_user._default_designation_generator = source_user._DesignationGenerator(
+            nouns=["accent"],
+            adjectives=["tonic"],
+        )
+
         config.SESSION_EXPIRATION_MINUTES = self.session_expiration / 60.0
 
         self.source_app.run(port=source_port, debug=True, use_reloader=False, threaded=True)
 
-    def test_display_id_designation_collisions(self):
+    def test_journalist_designation_collisions(self):
         self._source_visits_source_homepage()
         self._source_chooses_to_submit_documents()
         self._source_continues_to_submit_page()

securedrop/tests/test_crypto_util.py

@@ -6,25 +6,17 @@
 import os
 import pytest
 
-from flask import url_for, session
-
 from passphrases import PassphraseGenerator
 
 from source_user import create_source_user
 
 
 os.environ['SECUREDROP_ENV'] = 'test'  # noqa
-import crypto_util
 
 from crypto_util import CryptoUtil, CryptoException
 from db import db
 
 
-def test_word_list_does_not_contain_empty_strings(journalist_app):
-    assert '' not in journalist_app.crypto_util.nouns
-    assert '' not in journalist_app.crypto_util.adjectives
-
-
 def test_encrypt_success(source_app, config, test_source):
     message = 'test'
 
@@ -124,36 +116,12 @@ def test_basic_encrypt_then_decrypt_multiple_recipients(source_app,
         assert plaintext == message
 
 
-def test_display_id(source_app):
-    id = source_app.crypto_util.display_id()
-    id_words = id.split()
-
-    assert len(id_words) == 2
-    assert id_words[0] in source_app.crypto_util.adjectives
-    assert id_words[1] in source_app.crypto_util.nouns
-
-
-def test_display_id_designation_collisions(source_app):
-    with source_app.test_client() as app:
-        app.get(url_for('main.generate'))
-        source_app.crypto_util.adjectives = source_app.crypto_util.adjectives[:1]
-        source_app.crypto_util.nouns = source_app.crypto_util.nouns[:1]
-        tab_id = next(iter(session['codenames'].keys()))
-        app.post(url_for('main.create'), data={'tab_id': tab_id}, follow_redirects=True)
-
-        with pytest.raises(ValueError) as err:
-            source_app.crypto_util.display_id()
-
-        assert 'Could not generate unique journalist designation for new source' in str(err)
-
-
 def test_genkeypair(source_app):
     with source_app.app_context():
         source_user = create_source_user(
             db_session=db.session,
             source_passphrase=PassphraseGenerator.get_default().generate_passphrase(),
             source_app_storage=source_app.storage,
-            source_app_crypto_util=source_app.crypto_util,
         )
         source_app.crypto_util.genkeypair(source_user)
 
@@ -186,7 +154,6 @@ def test_reply_keypair_creation_and_expiration_dates(source_app):
             db_session=db.session,
             source_passphrase=PassphraseGenerator.get_default().generate_passphrase(),
             source_app_storage=source_app.storage,
-            source_app_crypto_util=source_app.crypto_util,
         )
         source_app.crypto_util.genkeypair(source_user)
 
@@ -275,16 +242,10 @@ def test_get_pubkey(source_app, test_source):
     assert pubkey is None
 
 
-@given(
-    name=text(alphabet=crypto_util.DICEWARE_SAFE_CHARS),
-    secret=text(alphabet=crypto_util.DICEWARE_SAFE_CHARS),
-    message=text()
-)
+@given(message=text())
 def test_encrypt_then_decrypt_gives_same_result(
         source_app,
         test_source,
-        name,
-        secret,
         message
 ):
     """Test that encrypting, then decrypting a string gives the original string.