def encrypt_to_source()

in client/securedrop_client/crypto.py [0:0]

• 40 lines of code
• 9 McCabe index (conditional complexity)

    def encrypt_to_source(self, source_uuid: str, data: str) -> str:
        """
        :param data: A string of data to encrypt to a source.
        """
        session = self.session_maker()
        source = session.query(Source).filter_by(uuid=source_uuid).one()

        # do not attempt to encrypt if the journalist key is missing
        if not self.journalist_key_fingerprint:
            raise CryptoError("Could not encrypt reply due to missing fingerprint for journalist")

        # do not attempt to encrypt if the source key is missing
        if not (source.fingerprint and source.public_key):
            raise CryptoError(f"Could not encrypt reply: no key for source {source_uuid}")

        try:
            self.import_key(source)
        except CryptoError as e:
            raise CryptoError("Could not import key before encrypting reply: {e}") from e

        cmd = self._gpg_cmd_base()

        with (
            tempfile.NamedTemporaryFile("w+") as content,
            tempfile.NamedTemporaryFile("w+") as stdout,
            tempfile.NamedTemporaryFile("w+") as stderr,
        ):
            content.write(data)
            content.seek(0)

            cmd.extend(
                [
                    "--encrypt",
                    "-r",
                    source.fingerprint,
                    "-r",
                    self.journalist_key_fingerprint,
                    "--armor",
                ]
            )
            if not self.is_qubes:
                # In Qubes, the ciphertext will go to stdout.
                # In addition the option below cannot be passed
                # through the gpg client wrapper.
                cmd.extend(["-o-"])  # write to stdout
            cmd.extend([content.name])

            try:
                subprocess.check_call(cmd, stdout=stdout, stderr=stderr)
            except subprocess.CalledProcessError as e:
                stderr.seek(0)
                err = stderr.read()
                raise CryptoError(f"Could not encrypt to source {source_uuid}: {e}\n{err}")

            stdout.seek(0)
            return stdout.read()