mirror of https://github.com/apache/kafka.git
93 lines
2.6 KiB
Python
93 lines
2.6 KiB
Python
#
|
|
# 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 as e:
|
|
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 suplied passphrase may be ignored
|
|
agent_kill()
|
|
try:
|
|
sign(key_id, passphrase, content, signature)
|
|
verify(content, signature)
|
|
except subprocess.CalledProcessError as e:
|
|
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()
|
|
|
|
|