jacobpascual/trezor-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' into neopg-wip

Browse Source
This commit is contained in:
Roman Zeyde
2018-07-01 13:52:37 +03:00
parent 6b6d9f5d20 b7a8c42893
commit 4968ca7ff3
19 changed files with 202 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
libagent/device/trezor.py
View File

@@ -7,6 +7,7 @@ import mnemonic
import semver
from . import interface
from .. import util
log = logging.getLogger(__name__)
@@ -46,7 +47,7 @@ class Trezor(interface.Device):
conn.callback_PinMatrixRequest = new_handler
cached_passphrase_ack = None
cached_passphrase_ack = util.ExpiringCache(seconds=float('inf'))
cached_state = None
def _override_passphrase_handler(self, conn):
@@ -57,9 +58,10 @@ class Trezor(interface.Device):
try:
if msg.on_device is True:
return self._defs.PassphraseAck()
if self.__class__.cached_passphrase_ack:
ack = self.__class__.cached_passphrase_ack.get()
if ack:
log.debug('re-using cached %s passphrase', self)
return self.__class__.cached_passphrase_ack
return ack
passphrase = self.ui.get_passphrase()
passphrase = mnemonic.Mnemonic.normalize_string(passphrase)
@@ -70,7 +72,7 @@ class Trezor(interface.Device):
msg = 'Too long passphrase ({} chars)'.format(length)
raise ValueError(msg)
self.__class__.cached_passphrase_ack = ack
self.__class__.cached_passphrase_ack.set(ack)
return ack
except: # noqa
conn.init_device()

8
libagent/device/ui.py
View File

@@ -24,7 +24,7 @@ class UI(object):
self.options_getter = create_default_options_getter()
self.device_name = device_type.__name__
def get_pin(self):
def get_pin(self, name=None):
"""Ask the user for (scrambled) PIN."""
description = (
'Use the numeric keypad to describe number positions.\n'
@@ -33,16 +33,16 @@ class UI(object):
' 4 5 6\n'
' 1 2 3')
return interact(
title='{} PIN'.format(self.device_name),
title='{} PIN'.format(name or self.device_name),
prompt='PIN:',
description=description,
binary=self.pin_entry_binary,
options=self.options_getter())
def get_passphrase(self):
def get_passphrase(self, name=None):
"""Ask the user for passphrase."""
return interact(
title='{} passphrase'.format(self.device_name),
title='{} passphrase'.format(name or self.device_name),
prompt='Passphrase:',
description=None,
binary=self.passphrase_entry_binary,

16
libagent/gpg/__init__.py
View File

@@ -148,6 +148,7 @@ export PATH={0}
-vv \
--pin-entry-binary={pin_entry_binary} \
--passphrase-entry-binary={passphrase_entry_binary} \
--cache-expiry-seconds={cache_expiry_seconds} \
$*
""".format(os.environ['PATH'], agent_path, **vars(args)))
check_call(['chmod', '700', f.name])
@@ -179,7 +180,8 @@ fi
# Generate new GPG identity and import into GPG keyring
pubkey = write_file(os.path.join(homedir, 'pubkey.asc'),
export_public_key(device_type, args))
check_call(keyring.gpg_command(['--homedir', homedir, '--quiet',
verbosity = ('-' + ('v' * args.verbose)) if args.verbose else '--quiet'
check_call(keyring.gpg_command(['--homedir', homedir, verbosity,
'--import', pubkey.name]))
# Make new GPG identity with "ultimate" trust (via its fingerprint)
@@ -229,6 +231,8 @@ def run_agent(device_type):
help='Path to PIN entry UI helper.')
p.add_argument('--passphrase-entry-binary', type=str, default='pinentry',
help='Path to passphrase entry UI helper.')
p.add_argument('--cache-expiry-seconds', type=float, default=float('inf'),
help='Expire passphrase from cache after this duration.')
args, _ = p.parse_known_args()
@@ -245,6 +249,8 @@ def run_agent(device_type):
pubkey_bytes = keyring.export_public_keys(env=env)
device_type.ui = device.ui.UI(device_type=device_type,
config=vars(args))
device_type.cached_passphrase_ack = util.ExpiringCache(
seconds=float(args.cache_expiry_seconds))
handler = agent.Handler(device=device_type(),
pubkey_bytes=pubkey_bytes)
@@ -272,7 +278,9 @@ def run_agent(device_type):
def main(device_type):
"""Parse command-line arguments."""
parser = argparse.ArgumentParser()
epilog = ('See https://github.com/romanz/trezor-agent/blob/master/'
'doc/README-GPG.md for usage examples.')
parser = argparse.ArgumentParser(epilog=epilog)
agent_package = device_type.package_name()
resources_map = {r.key: r for r in pkg_resources.require(agent_package)}
@@ -299,6 +307,8 @@ def main(device_type):
help='Path to PIN entry UI helper.')
p.add_argument('--passphrase-entry-binary', type=str, default='pinentry',
help='Path to passphrase entry UI helper.')
p.add_argument('--cache-expiry-seconds', type=float, default=float('inf'),
help='Expire passphrase from cache after this duration.')
p.set_defaults(func=run_init)
@@ -308,5 +318,7 @@ def main(device_type):
args = parser.parse_args()
device_type.ui = device.ui.UI(device_type=device_type, config=vars(args))
device_type.cached_passphrase_ack = util.ExpiringCache(
seconds=float(args.cache_expiry_seconds))
return args.func(device_type=device_type, args=args)

29
libagent/gpg/agent.py
View File

@@ -92,7 +92,7 @@ class Handler(object):
b'OPTION': lambda _, args: self.handle_option(*args),
b'SETKEYDESC': None,
b'NOP': None,
b'GETINFO': lambda conn, _: keyring.sendline(conn, b'D ' + self.version),
b'GETINFO': self.handle_getinfo,
b'AGENT_ID': lambda conn, _: keyring.sendline(conn, b'D TREZOR'), # "Fake" agent ID
b'SIGKEY': lambda _, args: self.set_key(*args),
b'SETKEY': lambda _, args: self.set_key(*args),
@@ -102,6 +102,7 @@ class Handler(object):
b'HAVEKEY': lambda _, args: self.have_key(*args),
b'KEYINFO': _key_info,
b'SCD': self.handle_scd,
b'GET_PASSPHRASE': self.handle_get_passphrase,
}
def reset(self):
@@ -115,6 +116,32 @@ class Handler(object):
self.options.append(opt)
log.debug('options: %s', self.options)
def handle_get_passphrase(self, conn, _):
"""Allow simple GPG symmetric encryption (using a passphrase)."""
p1 = self.client.device.ui.get_passphrase('Symmetric encryption')
p2 = self.client.device.ui.get_passphrase('Re-enter encryption')
if p1 == p2:
result = b'D ' + util.assuan_serialize(p1.encode('ascii'))
keyring.sendline(conn, result, confidential=True)
else:
log.warning('Passphrase does not match!')
def handle_getinfo(self, conn, args):
"""Handle some of the GETINFO messages."""
result = None
if args[0] == b'version':
result = self.version
elif args[0] == b's2k_count':
# Use highest number of S2K iterations.
# https://www.gnupg.org/documentation/manuals/gnupg/OpenPGP-Options.html
# https://tools.ietf.org/html/rfc4880#section-3.7.1.3
result = '{}'.format(64 << 20).encode('ascii')
else:
log.warning('Unknown GETINFO command: %s', args)
if result:
keyring.sendline(conn, b'D ' + result)
def handle_scd(self, conn, args):
"""No support for smart-card device protocol."""
reply = {

4
libagent/gpg/keyring.py
View File

@@ -48,9 +48,9 @@ def communicate(sock, msg):
return recvline(sock)
def sendline(sock, msg):
def sendline(sock, msg, confidential=False):
"""Send a binary message, followed by EOL."""
log.debug('<- %r', msg)
log.debug('<- %r', ('<snip>' if confidential else msg))
sock.sendall(msg + b'\n')

54
libagent/ssh/__init__.py
View File

@@ -23,9 +23,11 @@ log = logging.getLogger(__name__)
UNIX_SOCKET_TIMEOUT = 0.1
def ssh_args(label):
def ssh_args(conn):
"""Create SSH command for connecting specified server."""
identity = device.interface.string_to_identity(label)
I, = conn.identities
identity = I.identity_dict
pubkey_tempfile, = conn.public_keys_as_files()
args = []
if 'port' in identity:
@@ -33,12 +35,15 @@ def ssh_args(label):
if 'user' in identity:
args += ['-l', identity['user']]
args += ['-o', 'IdentityFile={}'.format(pubkey_tempfile.name)]
args += ['-o', 'IdentitiesOnly=true']
return args + [identity['host']]
def mosh_args(label):
def mosh_args(conn):
"""Create SSH command for connecting specified server."""
identity = device.interface.string_to_identity(label)
I, = conn.identities
identity = I.identity_dict
args = []
if 'port' in identity:
@@ -60,7 +65,10 @@ def _to_unicode(s):
def create_agent_parser(device_type):
"""Create an ArgumentParser for this tool."""
p = configargparse.ArgParser(default_config_files=['~/.ssh/agent.config'])
epilog = ('See https://github.com/romanz/trezor-agent/blob/master/'
'doc/README-SSH.md for usage examples.')
p = configargparse.ArgParser(default_config_files=['~/.ssh/agent.config'],
epilog=epilog)
p.add_argument('-v', '--verbose', default=0, action='count')
agent_package = device_type.package_name()
@@ -89,6 +97,8 @@ def create_agent_parser(device_type):
help='Path to PIN entry UI helper.')
p.add_argument('--passphrase-entry-binary', type=str, default='pinentry',
help='Path to passphrase entry UI helper.')
p.add_argument('--cache-expiry-seconds', type=float, default=float('inf'),
help='Expire passphrase from cache after this duration.')
g = p.add_mutually_exclusive_group()
g.add_argument('-d', '--daemonize', default=False, action='store_true',
@@ -191,6 +201,7 @@ class JustInTimeConnection(object):
self.conn_factory = conn_factory
self.identities = identities
self.public_keys_cache = public_keys
self.public_keys_tempfiles = []
def public_keys(self):
"""Return a list of SSH public keys (in textual format)."""
@@ -207,6 +218,17 @@ class JustInTimeConnection(object):
pk['identity'] = identity
return public_keys
def public_keys_as_files(self):
"""Store public keys as temporary SSH identity files."""
if not self.public_keys_tempfiles:
for pk in self.public_keys():
f = tempfile.NamedTemporaryFile(prefix='trezor-ssh-pubkey-', mode='w')
f.write(pk)
f.flush()
self.public_keys_tempfiles.append(f)
return self.public_keys_tempfiles
def sign(self, blob, identity):
"""Sign a given blob using the specified identity on the device."""
conn = self.conn_factory()
@@ -236,6 +258,7 @@ def main(device_type):
util.setup_logging(verbosity=args.verbose, filename=args.log_file)
public_keys = None
filename = None
if args.identity.startswith('/'):
filename = args.identity
contents = open(filename, 'rb').read().decode('utf-8')
@@ -250,14 +273,22 @@ def main(device_type):
identity.identity_dict['proto'] = u'ssh'
log.info('identity #%d: %s', index, identity.to_string())
sock_path = _get_sock_path(args)
# override default PIN/passphrase entry tools (relevant for TREZOR/Keepkey):
device_type.ui = device.ui.UI(device_type=device_type, config=vars(args))
device_type.cached_passphrase_ack = util.ExpiringCache(
args.cache_expiry_seconds)
conn = JustInTimeConnection(
conn_factory=lambda: client.Client(device_type()),
identities=identities, public_keys=public_keys)
sock_path = _get_sock_path(args)
command = args.command
context = _dummy_context()
if args.connect:
command = ['ssh'] + ssh_args(args.identity) + args.command
command = ['ssh'] + ssh_args(conn) + args.command
elif args.mosh:
command = ['mosh'] + mosh_args(args.identity) + args.command
command = ['mosh'] + mosh_args(conn) + args.command
elif args.daemonize:
out = 'SSH_AUTH_SOCK={0}; export SSH_AUTH_SOCK;\n'.format(sock_path)
sys.stdout.write(out)
@@ -272,13 +303,6 @@ def main(device_type):
command = os.environ['SHELL']
sys.stdin.close()
# override default PIN/passphrase entry tools (relevant for TREZOR/Keepkey):
device_type.ui = device.ui.UI(device_type=device_type, config=vars(args))
conn = JustInTimeConnection(
conn_factory=lambda: client.Client(device_type()),
identities=identities, public_keys=public_keys)
if command or args.daemonize or args.foreground:
with context:
return run_server(conn=conn, command=command, sock_path=sock_path,

23
libagent/tests/test_util.py
View File

@@ -121,3 +121,26 @@ def test_assuan_serialize():
assert util.assuan_serialize(b'') == b''
assert util.assuan_serialize(b'123\n456') == b'123%0A456'
assert util.assuan_serialize(b'\r\n') == b'%0D%0A'
def test_cache():
timer = mock.Mock(side_effect=range(7))
c = util.ExpiringCache(seconds=2, timer=timer) # t=0
assert c.get() is None # t=1
obj = 'foo'
c.set(obj) # t=2
assert c.get() is obj # t=3
assert c.get() is obj # t=4
assert c.get() is None # t=5
assert c.get() is None # t=6
def test_cache_inf():
timer = mock.Mock(side_effect=range(6))
c = util.ExpiringCache(seconds=float('inf'), timer=timer)
obj = 'foo'
c.set(obj)
assert c.get() is obj
assert c.get() is obj
assert c.get() is obj
assert c.get() is obj

23
libagent/util.py
View File

@@ -5,6 +5,7 @@ import functools
import io
import logging
import struct
import time
log = logging.getLogger(__name__)
@@ -255,3 +256,25 @@ def assuan_serialize(data):
escaped = '%{:02X}'.format(ord(c)).encode('ascii')
data = data.replace(c, escaped)
return data
class ExpiringCache(object):
"""Simple cache with a deadline."""
def __init__(self, seconds, timer=time.time):
"""C-tor."""
self.duration = seconds
self.timer = timer
self.value = None
self.set(None)
def get(self):
"""Returns existing value, or None if deadline has expired."""
if self.timer() > self.deadline:
self.value = None
return self.value
def set(self, value):
"""Set new value and reset the deadline for expiration."""
self.deadline = self.timer() + self.duration
self.value = value