|
|
|
@ -13,10 +13,10 @@ from .. import util
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def connect_to_agent(sock_path='~/.gnupg/S.gpg-agent'):
|
|
|
|
|
def connect_to_agent(sock_path='~/.gnupg/S.gpg-agent', sp=subprocess):
|
|
|
|
|
"""Connect to GPG agent's UNIX socket."""
|
|
|
|
|
sock_path = os.path.expanduser(sock_path)
|
|
|
|
|
subprocess.check_call(['gpg-connect-agent', '/bye'])
|
|
|
|
|
sp.check_call(['gpg-connect-agent', '/bye'])
|
|
|
|
|
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
|
|
|
sock.connect(sock_path)
|
|
|
|
|
return sock
|
|
|
|
@ -105,14 +105,14 @@ def parse_sig(sig):
|
|
|
|
|
return parser(args=sig[1:])
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def sign_digest(sock, keygrip, digest):
|
|
|
|
|
def sign_digest(sock, keygrip, digest, sp=subprocess):
|
|
|
|
|
"""Sign a digest using specified key using GPG agent."""
|
|
|
|
|
hash_algo = 8 # SHA256
|
|
|
|
|
assert len(digest) == 32
|
|
|
|
|
|
|
|
|
|
assert _communicate(sock, 'RESET').startswith(b'OK')
|
|
|
|
|
|
|
|
|
|
ttyname = subprocess.check_output('tty').strip()
|
|
|
|
|
ttyname = sp.check_output(['tty']).strip()
|
|
|
|
|
options = ['ttyname={}'.format(ttyname)] # set TTY for passphrase entry
|
|
|
|
|
|
|
|
|
|
display = os.environ.get('DISPLAY')
|
|
|
|
@ -143,17 +143,17 @@ def sign_digest(sock, keygrip, digest):
|
|
|
|
|
return parse_sig(sig)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def get_keygrip(user_id):
|
|
|
|
|
def get_keygrip(user_id, sp=subprocess):
|
|
|
|
|
"""Get a keygrip of the primary GPG key of the specified user."""
|
|
|
|
|
args = ['gpg2', '--list-keys', '--with-keygrip', user_id]
|
|
|
|
|
output = subprocess.check_output(args).decode('ascii')
|
|
|
|
|
output = sp.check_output(args).decode('ascii')
|
|
|
|
|
return re.findall(r'Keygrip = (\w+)', output)[0]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def export_public_key(user_id):
|
|
|
|
|
def export_public_key(user_id, sp=subprocess):
|
|
|
|
|
"""Export GPG public key for specified `user_id`."""
|
|
|
|
|
args = ['gpg2', '--export'] + ([user_id] if user_id else [])
|
|
|
|
|
result = subprocess.check_output(args=args)
|
|
|
|
|
result = sp.check_output(args=args)
|
|
|
|
|
if not result:
|
|
|
|
|
log.error('could not find public key %r in local GPG keyring', user_id)
|
|
|
|
|
raise KeyError(user_id)
|
|
|
|
|