release/gpg.py

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

"""
Auxiliary functions to interact with GNU Privacy Guard (GPG).
"""

import hashlib
import subprocess
import tempfile

from runtime import execute


def key_exists(key_id):
    """
    Checks whether the specified GPG key exists locally.
    """
    try:
        execute(f"gpg --list-keys {key_id}")
    except Exception:
        return False
    return True


def agent_kill():
    """
    Tries to kill the GPG agent process.
    """
    try:
        execute("gpgconf --kill gpg-agent")
    except FileNotFoundError as e:
        if e.filename != 'gpgconf':
            raise e


def sign(key_id, passphrase, content, target):
    """
    Generates a GPG signature, using the given key and passphrase,
    of the specified content into the target path.
    """
    execute(f"gpg --passphrase-fd 0 -u {key_id} --armor --output {target} --detach-sig {content}", input=passphrase.encode())


def verify(content, signature):
    """
    Verify the given GPG signature for the specified content.
    """
    execute(f"gpg --verify {signature} {content}")


def valid_passphrase(key_id, passphrase):
    """
    Checks whether the given passphrase is workable for the given key.
    """
    with tempfile.TemporaryDirectory() as tmpdir:
        content = __file__
        signature = tmpdir + '/sig.asc'
        # if the agent is running, the supplied passphrase may be ignored
        agent_kill()
        try:
            sign(key_id, passphrase, content, signature)
            verify(content, signature)
        except subprocess.CalledProcessError:
            return False
    return True


def key_pass_id(key_id, passphrase):
    """
    Generates a deterministic identifier for the key and passphrase combination.
    """
    h = hashlib.sha512()
    h.update(key_id.encode())
    h.update(passphrase.encode())
    return h.hexdigest()