Encrypt confidential emails

Now that confidential issue emails are not sent in the clear it might be interesting to consider what it would take to modify the current process so that it sent encrypted emails with the actual contents of the issue, if possible.

We could modify the python script that was deployed to redact the message to add a few additional steps: query the gitlab API (requires a personal access token) to look up a user's ID, and then lookup that ID's OpenPGP key. If they have one set, then encrypt the mail to them, and if they do not, then do what we do now (send the "This issue is confidential and the contents of the message have been redacted." message).

Its actually quite easy to get the public key of a user via the API, if you know their email address. To figure out how, I wrote this script that will spit out the OpenPGP key of any email in gitlab as long as the user's email is public:

import argparse
import requests
import sys

GITLAB_API_URL = 'https://gitlab.torproject.org/api/v4'
GITLAB_PRIVATE_TOKEN = '<redacted>'

def get_user_id(email):
    """Fetch the GitLab user ID based on the email address."""
    headers = {'Private-Token': GITLAB_PRIVATE_TOKEN}
    try:
        response = requests.get(f"{GITLAB_API_URL}/users?search={email}", headers=headers)
        response.raise_for_status()  # Raises an HTTPError if the response code was unsuccessful
        users = response.json()
        if not users:
            print(f"API returned no users for email: {email}")
            return None
        return users[0]['id']
    except requests.exceptions.HTTPError as e:
        print(f"HTTPError: {e.response.status_code} {e.response.reason}")
    except requests.exceptions.RequestException as e:
        print(f"RequestException: {e}")
    return None

def get_user_gpg_keys(user_id):
    """Fetch the GPG keys associated with a GitLab user ID."""
    headers = {'Private-Token': GITLAB_PRIVATE_TOKEN}
    try:
        response = requests.get(f"{GITLAB_API_URL}/users/{user_id}/gpg_keys", headers=headers)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(f"HTTPError: {e.response.status_code} {e.response.reason}")
    except requests.exceptions.RequestException as e:
        print(f"RequestException: {e}")
    return []

def main():
    parser = argparse.ArgumentParser(description='Fetch a GitLab user\'s GPG key by email address.')
    parser.add_argument('email', type=str, help='The email address of the user to search for.')
    args = parser.parse_args()

    user_id = get_user_id(args.email)
    if not user_id:
        sys.exit("Exiting: No user found or error occurred.")
    gpg_keys = get_user_gpg_keys(user_id)
    if not gpg_keys:
        print(f"No GPG keys found for user with ID {user_id}")
        return
    for key in gpg_keys:
        print(f"{key['key']}")

if __name__ == "__main__":
    main()

Considering that is how you would pull the key from the API endpoint, then you could imagine modifying the existing deployed script to incorporate this. Here is some untested somewhat pseudo-code modification of the currently deployed script. Because I can't test this, there are some obvious things that are wrong, but the basic idea is there:

#!/usr/bin/python3 -X utf8

import argparse
from email.parser import Parser
import email.policy
from io import StringIO
import logging
import quopri
from typing import TextIO
import subprocess
import sys
import requests
import gnupg

gpg = gnupg.GPG(gnupghome='/path/to/.gnupg')
gitlab_api_url = 'https://gitlab.torproject.org/api/v4'
gitlab_private_token = 'prviate access token' 

def get_user_gpg_key(email):
    """Fetch the user's GPG key from GitLab API."""
    headers = {'Private-Token': gitlab_private_token}
    users = requests.get(f"{gitlab_api_url}/users?search={email}", headers=headers).json()
    if not users:
        return None
    user_id = users[0]['id']
    keys = requests.get(f"{gitlab_api_url}/users/{user_id}/gpg_keys", headers=headers).json()
    if not keys:
        return None
    return keys[0]['key']

def encrypt_message(message, gpg_key):
    """Encrypt the message with the provided GPG key."""
    imported_key = gpg.import_keys(gpg_key)
    encrypted_data = gpg.encrypt(message, imported_key.fingerprints[0], always_trust=True)
    return str(encrypted_data)

def main():
    logging.basicConfig(level="INFO", format="%(levelname)s: %(message)s")
    parser = argparse.ArgumentParser()
    parser.add_argument("--stdout", action="store_true", help="send the email to stdout instead of sendmail(1)")
    parser.add_argument("--from", "-f", dest="sender", help="sender address")
    parser.add_argument("recipients", nargs="+", help="recipient address(es)")
    args = parser.parse_args()

    email_content = transform_email(sys.stdin, args.recipients[0])  # Assuming one recipient for simplicity

    if args.stdout:
        print(email_content)
    else:
        cmd = ["/usr/sbin/sendmail", "-G", "-i", "-f", args.sender, "--"] + args.recipients
        try:
            subprocess.Popen(cmd, stdin=subprocess.PIPE, encoding="utf-8").communicate(email_content)
        except Exception as e:
            import traceback
            print("4.3.0 %s: %s" % (e, traceback.format_exc()))
            sys.exit(75)  # TEMPFAIL

def transform_email(stream: TextIO, recipient_email):
    msg = Parser(policy=email.policy.compat32).parse(stream)
    if msg.get('X-GitLab-ConfidentialIssue', 'false') != 'true':
        return str(msg)

    gpg_key = get_user_gpg_key(recipient_email)
    if gpg_key:
        # Encrypt the message if a GPG key is found
        encrypted_msg = encrypt_message(str(msg), gpg_key)
        return encrypted_msg
    else:
        # Redact if no GPG key is found
        for part in msg.walk():
            if part.is_multipart():
                continue
            if part.get_content_type() != 'text/plain':
                continue
            plain_text = part.as_string()
            signature = quopri.decodestring(plain_text.split("-- ").pop()).decode("ascii")
            redaction_msg = """This issue is confidential and the contents of the message have been redacted.

--
""" + signature

        del msg['X-GitLab-ConfidentialIssue']
        msg['X-GitLab-ConfidentialIssue'] = 'redacted'
        msg.set_type("text/plain")
        msg.del_param("boundary", header="Content-Type")
        msg.set_payload(redaction_msg)
        return str(msg)

if __name__ == "__main__":
    main()

Note that there's an upstream issue about this, of course: https://gitlab.com/gitlab-org/gitlab/-/issues/19056