Encrypt confidential emails

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()

#!/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()