Migrate public keys from GPG to database-backed storage

securedrop/alembic/versions/17c559a7a685_pgp_public_keys.py

@@ -0,0 +1,96 @@
+"""PGP public keys
+
+Revision ID: 17c559a7a685
+Revises: 811334d7105f
+Create Date: 2023-09-21 12:33:56.550634
+
+"""
+
+import traceback
+
+import pretty_bad_protocol as gnupg
+import sqlalchemy as sa
+from alembic import op
+from encryption import EncryptionManager
+from sdconfig import SecureDropConfig
+
+import redwood
+
+# revision identifiers, used by Alembic.
+revision = "17c559a7a685"
+down_revision = "811334d7105f"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """
+    Migrate public keys from the GPG keyring into the SQLite database
+
+    We iterate over all the secret keys in the keyring and see if we
+    can identify the corresponding Source record. If we can, and it
+    doesn't already have key material migrated, export the key and
+    save it in the database.
+    """
+    config = SecureDropConfig.get_current()
+    gpg = gnupg.GPG(
+        binary="gpg2",
+        homedir=str(config.GPG_KEY_DIR),
+        options=["--pinentry-mode loopback", "--trust-model direct"],
+    )
+    # Source keys all have a secret key, so we can filter on that
+    for keyinfo in gpg.list_keys(secret=True):
+        if len(keyinfo["uids"]) > 1:
+            # Source keys should only have one UID
+            continue
+        uid = keyinfo["uids"][0]
+        search = EncryptionManager.SOURCE_KEY_UID_RE.search(uid)
+        if not search:
+            # Didn't match at all
+            continue
+        filesystem_id = search.group(2)
+        # Check that it's a valid ID
+        conn = op.get_bind()
+        result = conn.execute(
+            sa.text(
+                """
+                SELECT pgp_public_key, pgp_fingerprint
+                FROM sources
+                WHERE filesystem_id=:filesystem_id
+                """
+            ).bindparams(filesystem_id=filesystem_id)
+        ).first()
+        if result != (None, None):
+            # Either not in the database or there's already some data in the DB.
+            # In any case, skip.
+            continue
+        fingerprint = keyinfo["fingerprint"]
+        try:
+            public_key = gpg.export_keys(fingerprint)
+            redwood.is_valid_public_key(public_key)
+        except:  # noqa: E722
+            # Exporting the key failed in some manner
+            traceback.print_exc()
+            continue
+
+        # Save to database
+        op.execute(
+            sa.text(
+                """
+                UPDATE sources
+                SET pgp_public_key=:pgp_public_key, pgp_fingerprint=:pgp_fingerprint
+                WHERE filesystem_id=:filesystem_id
+                """
+            ).bindparams(
+                pgp_public_key=public_key,
+                pgp_fingerprint=fingerprint,
+                filesystem_id=filesystem_id,
+            )
+        )
+
+
+def downgrade() -> None:
+    """
+    This is a non-destructive operation, so it's not worth implementing a
+    migration from database storage to GPG.
+    """

securedrop/encryption.py

@@ -40,7 +40,7 @@ class EncryptionManager:
     REDIS_FINGERPRINT_HASH = "sd/crypto-util/fingerprints"
     REDIS_KEY_HASH = "sd/crypto-util/keys"
 
-    SOURCE_KEY_UID_RE = re.compile(r"(Source|Autogenerated) Key <[-A-Za-z0-9+/=_]+>")
+    SOURCE_KEY_UID_RE = re.compile(r"(Source|Autogenerated) Key <([-A-Za-z0-9+/=_]+)>")
 
     def __init__(self, gpg_key_dir: Path, journalist_pub_key: Path) -> None:
         self._gpg_key_dir = gpg_key_dir

securedrop/loaddata.py

@@ -239,7 +239,7 @@ def add_reply(
     db.session.commit()
 
 
-def add_source() -> Tuple[Source, str]:
+def add_source(use_gpg: bool = False) -> Tuple[Source, str]:
     """
     Adds a single source.
     """
@@ -250,6 +250,26 @@ def add_source() -> Tuple[Source, str]:
         source_app_storage=Storage.get_default(),
     )
     source = source_user.get_db_record()
+    if use_gpg:
+        manager = EncryptionManager.get_default()
+        gen_key_input = manager._gpg.gen_key_input(
+            passphrase=source_user.gpg_secret,
+            name_email=source_user.filesystem_id,
+            key_type="RSA",
+            key_length=4096,
+            name_real="Source Key",
+            creation_date="2013-05-14",
+            # '0' is the magic value that tells GPG's batch key generation not
+            # to set an expiration date.
+            expire_date="0",
+        )
+        manager._gpg.gen_key(gen_key_input)
+
+        # Delete the Sequoia-generated keys
+        source.pgp_public_key = None
+        source.pgp_fingerprint = None
+        source.pgp_secret_key = None
+        db.session.add(source)
     db.session.commit()
 
     return source, codename
@@ -323,7 +343,7 @@ def add_sources(args: argparse.Namespace, journalists: Tuple[Journalist, ...]) -
     )
 
     for i in range(1, args.source_count + 1):
-        source, codename = add_source()
+        source, codename = add_source(use_gpg=args.gpg)
 
         for _ in range(args.messages_per_source):
             submit_message(source, secrets.choice(journalists) if seen_message_count > 0 else None)
@@ -450,6 +470,12 @@ def parse_arguments() -> argparse.Namespace:
         "--seed",
         help=("Random number seed (for reproducible datasets)"),
     )
+    parser.add_argument(
+        "--gpg",
+        help="Create sources with a key pair stored in GPG",
+        action="store_true",
+        default=False,
+    )
     return parser.parse_args()
 
 

securedrop/tests/migrations/migration_17c559a7a685.py

@@ -0,0 +1,116 @@
+import uuid
+
+import pretty_bad_protocol as gnupg
+from db import db
+from journalist_app import create_app
+from sqlalchemy import text
+
+import redwood
+
+
+class UpgradeTester:
+    def __init__(self, config):
+        """This function MUST accept an argument named `config`.
+        You will likely want to save a reference to the config in your
+        class so you can access the database later.
+        """
+        self.config = config
+        self.app = create_app(self.config)
+        self.gpg = gnupg.GPG(
+            binary="gpg2",
+            homedir=str(config.GPG_KEY_DIR),
+            options=["--pinentry-mode loopback", "--trust-model direct"],
+        )
+        self.fingerprint = None
+        # random, chosen by fair dice roll
+        self.filesystem_id = (
+            "HAR5WIY3C4K3MMIVLYXER7DMTYCL5PWZEPNOCR2AIBCVWXDZQDMDFUHEFJM"
+            "Z3JW5D6SLED3YKCBDAKNMSIYOKWEJK3ZRJT3WEFT3S5Q="
+        )
+
+    def load_data(self):
+        """Create a source and GPG key pair"""
+        with self.app.app_context():
+            source = {
+                "uuid": str(uuid.uuid4()),
+                "filesystem_id": self.filesystem_id,
+                "journalist_designation": "psychic webcam",
+                "interaction_count": 0,
+            }
+            sql = """\
+                INSERT INTO sources (uuid, filesystem_id, journalist_designation,
+                    interaction_count)
+                VALUES (:uuid, :filesystem_id, :journalist_designation,
+                    :interaction_count)"""
+            db.engine.execute(text(sql), **source)
+            # Generate the GPG key pair
+            gen_key_input = self.gpg.gen_key_input(
+                passphrase="correct horse battery staple",
+                name_email=self.filesystem_id,
+                key_type="RSA",
+                key_length=4096,
+                name_real="Source Key",
+                creation_date="2013-05-14",
+                expire_date="0",
+            )
+            key = self.gpg.gen_key(gen_key_input)
+            self.fingerprint = str(key.fingerprint)
+
+    def check_upgrade(self):
+        """Verify PGP fields have been populated"""
+        with self.app.app_context():
+            query_sql = """\
+            SELECT pgp_fingerprint, pgp_public_key, pgp_secret_key
+            FROM sources
+            WHERE filesystem_id = :filesystem_id"""
+            source = db.engine.execute(
+                text(query_sql),
+                filesystem_id=self.filesystem_id,
+            ).fetchone()
+            assert source[0] == self.fingerprint
+            assert redwood.is_valid_public_key(source[1]) == self.fingerprint
+            assert source[2] is None
+
+
+class DowngradeTester:
+    def __init__(self, config):
+        self.config = config
+        self.app = create_app(self.config)
+        self.uuid = str(uuid.uuid4())
+
+    def load_data(self):
+        """Create a source with a PGP key pair already migrated"""
+        with self.app.app_context():
+            source = {
+                "uuid": self.uuid,
+                "filesystem_id": "1234",
+                "journalist_designation": "mucky pine",
+                "interaction_count": 0,
+                "pgp_fingerprint": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
+                "pgp_public_key": "very public",
+                "pgp_secret_key": None,
+            }
+            sql = """\
+                INSERT INTO sources (uuid, filesystem_id, journalist_designation,
+                    interaction_count, pgp_fingerprint, pgp_public_key, pgp_secret_key)
+                VALUES (:uuid, :filesystem_id, :journalist_designation,
+                    :interaction_count, :pgp_fingerprint, :pgp_public_key, :pgp_secret_key)"""
+            db.engine.execute(text(sql), **source)
+
+    def check_downgrade(self):
+        """Verify the downgrade does nothing, i.e. the two PGP fields are still populated"""
+        with self.app.app_context():
+            sql = """\
+            SELECT pgp_fingerprint, pgp_public_key, pgp_secret_key
+            FROM sources
+            WHERE uuid = :uuid"""
+            source = db.engine.execute(
+                text(sql),
+                uuid=self.uuid,
+            ).fetchone()
+            print(source)
+            assert source == (
+                "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
+                "very public",
+                None,
+            )