trezor-agent/libagent/gpg/keyring.py

248 lines
7.0 KiB
Python
Raw Normal View History

2016-04-27 18:01:21 +00:00
"""Tools for doing signature using gpg-agent."""
2016-10-18 09:10:28 +00:00
from __future__ import absolute_import, print_function, unicode_literals
2016-04-27 18:01:21 +00:00
import binascii
import io
import logging
import os
import re
import socket
import subprocess
2016-04-27 18:01:21 +00:00
from .. import util
log = logging.getLogger(__name__)
2017-10-21 15:46:04 +00:00
def get_agent_sock_path(env=None, sp=subprocess):
"""Parse gpgconf output to find out GPG agent UNIX socket path."""
2017-10-21 15:46:04 +00:00
output = sp.check_output(['gpgconf', '--list-dirs'], env=env)
lines = output.strip().split(b'\n')
dirs = dict(line.split(b':', 1) for line in lines)
2017-10-21 14:58:22 +00:00
log.debug('gpgconf --list-dirs: %s', dirs)
return dirs[b'agent-socket']
2017-10-21 15:46:04 +00:00
def connect_to_agent(env=None, sp=subprocess):
2016-04-27 18:01:21 +00:00
"""Connect to GPG agent's UNIX socket."""
2017-10-21 15:46:04 +00:00
sock_path = get_agent_sock_path(sp=sp, env=env)
2016-10-18 19:23:40 +00:00
sp.check_call(['gpg-connect-agent', '/bye']) # Make sure it's running
2016-04-27 18:01:21 +00:00
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.connect(sock_path)
return sock
2016-05-28 20:02:45 +00:00
def communicate(sock, msg):
2016-05-30 14:49:21 +00:00
"""Send a message and receive a single line."""
2016-06-01 15:34:22 +00:00
sendline(sock, msg.encode('ascii'))
2016-05-28 20:02:45 +00:00
return recvline(sock)
2016-04-27 18:01:21 +00:00
2016-06-01 15:34:22 +00:00
def sendline(sock, msg):
"""Send a binary message, followed by EOL."""
log.debug('<- %r', msg)
sock.sendall(msg + b'\n')
2016-05-28 20:02:45 +00:00
def recvline(sock):
2016-05-30 14:49:21 +00:00
"""Receive a single line from the socket."""
2016-04-27 18:01:21 +00:00
reply = io.BytesIO()
while True:
c = sock.recv(1)
2016-05-27 10:43:55 +00:00
if not c:
2016-06-02 18:39:14 +00:00
return None # socket is closed
2016-05-23 20:03:02 +00:00
if c == b'\n':
2016-04-27 18:01:21 +00:00
break
reply.write(c)
2016-05-23 20:03:02 +00:00
result = reply.getvalue()
2016-06-01 15:34:22 +00:00
log.debug('-> %r', result)
2016-05-23 20:03:02 +00:00
return result
2016-04-27 18:01:21 +00:00
2016-07-26 14:50:49 +00:00
def iterlines(conn):
"""Iterate over input, split by lines."""
while True:
line = recvline(conn)
if line is None:
break
yield line
2016-05-26 20:16:08 +00:00
def unescape(s):
2016-05-27 08:19:10 +00:00
"""Unescape ASSUAN message (0xAB <-> '%AB')."""
2016-04-27 18:01:21 +00:00
s = bytearray(s)
i = 0
while i < len(s):
if s[i] == ord('%'):
2016-05-23 20:03:02 +00:00
hex_bytes = bytes(s[i+1:i+3])
value = int(hex_bytes.decode('ascii'), 16)
2016-04-27 18:01:21 +00:00
s[i:i+3] = [value]
i += 1
return bytes(s)
2016-05-26 20:16:08 +00:00
def parse_term(s):
"""Parse single s-expr term from bytes."""
2016-05-23 20:03:02 +00:00
size, s = s.split(b':', 1)
2016-04-27 18:01:21 +00:00
size = int(size)
return s[:size], s[size:]
2016-05-26 20:16:08 +00:00
def parse(s):
"""Parse full s-expr from bytes."""
2016-05-23 20:03:02 +00:00
if s.startswith(b'('):
2016-04-27 18:01:21 +00:00
s = s[1:]
2016-05-26 20:16:08 +00:00
name, s = parse_term(s)
2016-04-27 18:01:21 +00:00
values = [name]
2016-05-23 20:03:02 +00:00
while not s.startswith(b')'):
2016-05-26 20:16:08 +00:00
value, s = parse(s)
2016-04-27 18:01:21 +00:00
values.append(value)
return values, s[1:]
else:
2016-05-26 20:16:08 +00:00
return parse_term(s)
2016-04-27 18:01:21 +00:00
def _parse_ecdsa_sig(args):
(r, sig_r), (s, sig_s) = args
2016-05-23 20:03:02 +00:00
assert r == b'r'
assert s == b's'
2016-04-29 14:45:16 +00:00
return (util.bytes2num(sig_r),
util.bytes2num(sig_s))
# DSA and EDDSA happen to have the same structure as ECDSA signatures
_parse_dsa_sig = _parse_ecdsa_sig
_parse_eddsa_sig = _parse_ecdsa_sig
2016-04-29 14:45:16 +00:00
def _parse_rsa_sig(args):
(s, sig_s), = args
2016-05-23 20:03:02 +00:00
assert s == b's'
2016-04-30 07:56:15 +00:00
return (util.bytes2num(sig_s),)
2016-05-26 20:16:08 +00:00
def parse_sig(sig):
"""Parse signature integer values from s-expr."""
label, sig = sig
2016-05-23 20:03:02 +00:00
assert label == b'sig-val'
algo_name = sig[0]
2016-05-23 20:03:02 +00:00
parser = {b'rsa': _parse_rsa_sig,
b'ecdsa': _parse_ecdsa_sig,
b'eddsa': _parse_eddsa_sig,
2016-05-23 20:03:02 +00:00
b'dsa': _parse_dsa_sig}[algo_name]
return parser(args=sig[1:])
2016-05-27 10:43:55 +00:00
def sign_digest(sock, keygrip, digest, sp=subprocess, environ=None):
2016-04-27 18:01:21 +00:00
"""Sign a digest using specified key using GPG agent."""
hash_algo = 8 # SHA256
assert len(digest) == 32
2016-05-28 20:02:45 +00:00
assert communicate(sock, 'RESET').startswith(b'OK')
ttyname = sp.check_output(['tty']).strip()
options = ['ttyname={}'.format(ttyname)] # set TTY for passphrase entry
2016-05-27 10:43:55 +00:00
display = (environ or os.environ).get('DISPLAY')
if display is not None:
options.append('display={}'.format(display))
for opt in options:
2016-05-28 20:02:45 +00:00
assert communicate(sock, 'OPTION {}'.format(opt)) == b'OK'
2016-05-28 20:02:45 +00:00
assert communicate(sock, 'SIGKEY {}'.format(keygrip)) == b'OK'
2016-05-26 20:16:08 +00:00
hex_digest = binascii.hexlify(digest).upper().decode('ascii')
2016-05-28 20:02:45 +00:00
assert communicate(sock, 'SETHASH {} {}'.format(hash_algo,
2016-05-30 14:49:21 +00:00
hex_digest)) == b'OK'
2016-04-29 07:25:46 +00:00
2016-05-28 20:02:45 +00:00
assert communicate(sock, 'SETKEYDESC '
2016-05-30 14:49:21 +00:00
'Sign+a+new+TREZOR-based+subkey') == b'OK'
2016-05-28 20:02:45 +00:00
assert communicate(sock, 'PKSIGN') == b'OK'
2016-10-20 16:22:07 +00:00
while True:
line = recvline(sock).strip()
if line.startswith(b'S PROGRESS'):
continue
else:
break
2016-05-26 20:16:08 +00:00
line = unescape(line)
2016-05-23 20:03:02 +00:00
log.debug('unescaped: %r', line)
prefix, sig = line.split(b' ', 1)
if prefix != b'D':
raise ValueError(prefix)
2016-04-27 18:01:21 +00:00
2016-05-26 20:16:08 +00:00
sig, leftover = parse(sig)
assert not leftover, leftover
2016-05-26 20:16:08 +00:00
return parse_sig(sig)
2016-04-27 18:01:21 +00:00
def get_gnupg_components(sp=subprocess):
"""Parse GnuPG components' paths."""
output = sp.check_output(['gpgconf', '--list-components'])
components = dict(re.findall('(.*):.*:(.*)', output.decode('ascii')))
log.debug('gpgconf --list-components: %s', components)
return components
def get_gnupg_binary(sp=subprocess):
"""Starting GnuPG 2.2.x, the default installation uses `gpg`."""
return get_gnupg_components(sp=sp)['gpg']
def gpg_command(args, env=None, sp=subprocess):
"""Prepare common GPG command line arguments."""
if env is None:
env = os.environ
cmd = [get_gnupg_binary(sp=sp)]
homedir = env.get('GNUPGHOME')
if homedir:
cmd.extend(['--homedir', homedir])
return cmd + args
def get_keygrip(user_id, sp=subprocess):
2016-04-27 18:01:21 +00:00
"""Get a keygrip of the primary GPG key of the specified user."""
args = gpg_command(['--list-keys', '--with-keygrip', user_id], sp=sp)
output = sp.check_output(args).decode('ascii')
2016-04-27 18:01:21 +00:00
return re.findall(r'Keygrip = (\w+)', output)[0]
2016-06-04 16:45:03 +00:00
def gpg_version(sp=subprocess):
"""Get a keygrip of the primary GPG key of the specified user."""
args = gpg_command(['--version'], sp=sp)
output = sp.check_output(args)
2016-06-04 16:45:03 +00:00
line = output.split(b'\n')[0] # b'gpg (GnuPG) 2.1.11'
return line.split(b' ')[-1] # b'2.1.11'
2017-10-21 16:16:48 +00:00
def export_public_key(user_id, env=None, sp=subprocess):
2016-05-27 08:19:10 +00:00
"""Export GPG public key for specified `user_id`."""
args = gpg_command(['--export', user_id], sp=sp)
2017-10-21 16:16:48 +00:00
result = sp.check_output(args=args, env=env)
2016-05-27 08:19:10 +00:00
if not result:
log.error('could not find public key %r in local GPG keyring', user_id)
raise KeyError(user_id)
2016-05-27 08:19:10 +00:00
return result
2017-10-21 16:16:48 +00:00
def export_public_keys(env=None, sp=subprocess):
"""Export all GPG public keys."""
args = gpg_command(['--export'], sp=sp)
2017-10-21 16:16:48 +00:00
result = sp.check_output(args=args, env=env)
if not result:
raise KeyError('No GPG public keys found at env: {!r}'.format(env))
return result
def create_agent_signer(user_id):
"""Sign digest with existing GPG keys using gpg-agent tool."""
2017-10-21 15:46:04 +00:00
sock = connect_to_agent(env=os.environ)
keygrip = get_keygrip(user_id)
def sign(digest):
"""Sign the digest and return an ECDSA/RSA/DSA signature."""
return sign_digest(sock=sock, keygrip=keygrip, digest=digest)
return sign