2016-04-24 10:04:53 +00:00
|
|
|
"""Create GPG ECDSA signatures and public keys using TREZOR device."""
|
|
|
|
import base64
|
|
|
|
import hashlib
|
|
|
|
import logging
|
|
|
|
import struct
|
|
|
|
import time
|
|
|
|
|
2016-04-28 11:45:45 +00:00
|
|
|
from . import agent, decode
|
2016-04-24 10:04:53 +00:00
|
|
|
from .. import client, factory, formats, util
|
|
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
|
|
|
def packet(tag, blob):
|
|
|
|
"""Create small GPG packet."""
|
2016-04-30 12:34:18 +00:00
|
|
|
assert len(blob) < 2**32
|
|
|
|
|
|
|
|
if len(blob) < 2**8:
|
|
|
|
length_type = 0
|
|
|
|
elif len(blob) < 2**16:
|
|
|
|
length_type = 1
|
|
|
|
else:
|
|
|
|
length_type = 2
|
|
|
|
|
|
|
|
fmt = ['>B', '>H', '>L'][length_type]
|
2016-04-24 10:04:53 +00:00
|
|
|
leading_byte = 0x80 | (tag << 2) | (length_type)
|
2016-04-30 12:34:18 +00:00
|
|
|
return struct.pack('>B', leading_byte) + util.prefix_len(fmt, blob)
|
2016-04-24 10:04:53 +00:00
|
|
|
|
|
|
|
|
|
|
|
def subpacket(subpacket_type, fmt, *values):
|
|
|
|
"""Create GPG subpacket."""
|
|
|
|
blob = struct.pack(fmt, *values) if values else fmt
|
|
|
|
return struct.pack('>B', subpacket_type) + blob
|
|
|
|
|
|
|
|
|
|
|
|
def subpacket_long(subpacket_type, value):
|
|
|
|
"""Create GPG subpacket with 32-bit unsigned integer."""
|
|
|
|
return subpacket(subpacket_type, '>L', value)
|
|
|
|
|
|
|
|
|
|
|
|
def subpacket_time(value):
|
|
|
|
"""Create GPG subpacket with time in seconds (since Epoch)."""
|
|
|
|
return subpacket_long(2, value)
|
|
|
|
|
|
|
|
|
|
|
|
def subpacket_byte(subpacket_type, value):
|
|
|
|
"""Create GPG subpacket with 8-bit unsigned integer."""
|
|
|
|
return subpacket(subpacket_type, '>B', value)
|
|
|
|
|
|
|
|
|
|
|
|
def subpackets(*items):
|
|
|
|
"""Serialize several GPG subpackets."""
|
|
|
|
prefixed = [util.prefix_len('>B', item) for item in items]
|
|
|
|
return util.prefix_len('>H', b''.join(prefixed))
|
|
|
|
|
|
|
|
|
|
|
|
def mpi(value):
|
|
|
|
"""Serialize multipresicion integer using GPG format."""
|
|
|
|
bits = value.bit_length()
|
|
|
|
data_size = (bits + 7) // 8
|
2016-04-30 10:01:14 +00:00
|
|
|
data_bytes = bytearray(data_size)
|
2016-04-24 10:04:53 +00:00
|
|
|
for i in range(data_size):
|
|
|
|
data_bytes[i] = value & 0xFF
|
|
|
|
value = value >> 8
|
|
|
|
|
|
|
|
data_bytes.reverse()
|
2016-04-30 10:01:14 +00:00
|
|
|
return struct.pack('>H', bits) + bytes(data_bytes)
|
2016-04-24 10:04:53 +00:00
|
|
|
|
|
|
|
|
2016-04-28 09:34:00 +00:00
|
|
|
def _serialize_nist256(vk):
|
2016-04-24 10:04:53 +00:00
|
|
|
return mpi((4 << 512) |
|
|
|
|
(vk.pubkey.point.x() << 256) |
|
|
|
|
(vk.pubkey.point.y()))
|
|
|
|
|
|
|
|
|
2016-04-28 09:34:00 +00:00
|
|
|
def _serialize_ed25519(vk):
|
2016-04-24 10:04:53 +00:00
|
|
|
return mpi((0x40 << 256) |
|
|
|
|
util.bytes2num(vk.to_bytes()))
|
|
|
|
|
|
|
|
|
|
|
|
SUPPORTED_CURVES = {
|
|
|
|
formats.CURVE_NIST256: {
|
|
|
|
# https://tools.ietf.org/html/rfc6637#section-11
|
|
|
|
'oid': b'\x2A\x86\x48\xCE\x3D\x03\x01\x07',
|
|
|
|
'algo_id': 19,
|
2016-04-28 09:34:00 +00:00
|
|
|
'serialize': _serialize_nist256
|
2016-04-24 10:04:53 +00:00
|
|
|
},
|
|
|
|
formats.CURVE_ED25519: {
|
|
|
|
'oid': b'\x2B\x06\x01\x04\x01\xDA\x47\x0F\x01',
|
|
|
|
'algo_id': 22,
|
2016-04-28 09:34:00 +00:00
|
|
|
'serialize': _serialize_ed25519
|
2016-04-24 10:04:53 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
2016-04-30 13:27:43 +00:00
|
|
|
MARKER = b'TREZOR-GPG'
|
|
|
|
|
|
|
|
|
2016-04-24 10:04:53 +00:00
|
|
|
def _find_curve_by_algo_id(algo_id):
|
|
|
|
curve_name, = [name for name, info in SUPPORTED_CURVES.items()
|
|
|
|
if info['algo_id'] == algo_id]
|
|
|
|
return curve_name
|
|
|
|
|
|
|
|
|
2016-04-28 11:45:45 +00:00
|
|
|
class HardwareSigner(object):
|
2016-04-30 07:04:44 +00:00
|
|
|
"""Sign messages and get public keys from a hardware device."""
|
|
|
|
|
2016-04-28 11:45:45 +00:00
|
|
|
def __init__(self, user_id, curve_name):
|
2016-04-30 07:04:44 +00:00
|
|
|
"""Connect to the device and retrieve required public key."""
|
2016-04-24 10:04:53 +00:00
|
|
|
self.client_wrapper = factory.load()
|
|
|
|
self.identity = self.client_wrapper.identity_type()
|
|
|
|
self.identity.proto = 'gpg'
|
|
|
|
self.identity.host = user_id
|
2016-04-28 11:45:45 +00:00
|
|
|
self.curve_name = curve_name
|
2016-04-24 10:04:53 +00:00
|
|
|
|
2016-04-28 11:45:45 +00:00
|
|
|
def pubkey(self):
|
2016-04-30 07:04:44 +00:00
|
|
|
"""Return public key as VerifyingKey object."""
|
2016-04-24 10:04:53 +00:00
|
|
|
addr = client.get_address(self.identity)
|
|
|
|
public_node = self.client_wrapper.connection.get_public_node(
|
|
|
|
n=addr, ecdsa_curve_name=self.curve_name)
|
|
|
|
|
2016-04-28 11:45:45 +00:00
|
|
|
return formats.decompress_pubkey(
|
2016-04-24 10:04:53 +00:00
|
|
|
pubkey=public_node.node.public_key,
|
|
|
|
curve_name=self.curve_name)
|
|
|
|
|
2016-04-29 19:23:12 +00:00
|
|
|
def sign(self, digest):
|
2016-04-30 10:01:14 +00:00
|
|
|
"""Sign the digest and return a serialized signature."""
|
2016-04-28 11:45:45 +00:00
|
|
|
result = self.client_wrapper.connection.sign_identity(
|
|
|
|
identity=self.identity,
|
|
|
|
challenge_hidden=digest,
|
2016-04-29 19:23:12 +00:00
|
|
|
challenge_visual=util.hexlify(digest),
|
2016-04-28 11:45:45 +00:00
|
|
|
ecdsa_curve_name=self.curve_name)
|
|
|
|
assert result.signature[:1] == b'\x00'
|
|
|
|
sig = result.signature[1:]
|
|
|
|
return mpi(util.bytes2num(sig[:32])) + mpi(util.bytes2num(sig[32:]))
|
|
|
|
|
|
|
|
def close(self):
|
2016-04-30 07:04:44 +00:00
|
|
|
"""Close the connection to the device."""
|
2016-04-28 11:45:45 +00:00
|
|
|
self.client_wrapper.connection.clear_session()
|
|
|
|
self.client_wrapper.connection.close()
|
|
|
|
|
|
|
|
|
|
|
|
class AgentSigner(object):
|
2016-04-30 07:04:44 +00:00
|
|
|
"""Sign messages and get public keys using gpg-agent tool."""
|
|
|
|
|
2016-04-30 10:20:06 +00:00
|
|
|
def __init__(self, user_id):
|
2016-04-30 07:04:44 +00:00
|
|
|
"""Connect to the agent and retrieve required public key."""
|
2016-04-28 11:45:45 +00:00
|
|
|
self.sock = agent.connect()
|
|
|
|
self.keygrip = agent.get_keygrip(user_id)
|
|
|
|
self.public_key = decode.load_from_gpg(user_id)
|
|
|
|
|
|
|
|
def pubkey(self):
|
2016-04-30 07:04:44 +00:00
|
|
|
"""Return public key as VerifyingKey object."""
|
2016-04-28 11:45:45 +00:00
|
|
|
return self.public_key['verifying_key']
|
|
|
|
|
2016-04-29 19:23:12 +00:00
|
|
|
def sign(self, digest):
|
2016-04-30 07:04:44 +00:00
|
|
|
"""Sign the digest and return an ECDSA signature."""
|
2016-04-30 12:39:32 +00:00
|
|
|
params = agent.sign(sock=self.sock,
|
|
|
|
keygrip=self.keygrip, digest=digest)
|
2016-04-30 10:01:14 +00:00
|
|
|
return b''.join(mpi(p) for p in params)
|
2016-04-28 11:45:45 +00:00
|
|
|
|
|
|
|
def close(self):
|
2016-04-30 07:04:44 +00:00
|
|
|
"""Close the connection to gpg-agent."""
|
2016-04-28 11:45:45 +00:00
|
|
|
self.sock.close()
|
|
|
|
|
|
|
|
|
2016-04-29 14:45:16 +00:00
|
|
|
class PublicKey(object):
|
2016-04-30 06:29:04 +00:00
|
|
|
"""GPG representation for public key packets."""
|
|
|
|
|
2016-04-29 14:45:16 +00:00
|
|
|
def __init__(self, curve_name, created, verifying_key):
|
2016-04-30 06:29:04 +00:00
|
|
|
"""Contruct using a ECDSA VerifyingKey object."""
|
|
|
|
self.curve_info = SUPPORTED_CURVES[curve_name]
|
|
|
|
self.created = int(created) # time since Epoch
|
2016-04-29 14:45:16 +00:00
|
|
|
self.verifying_key = verifying_key
|
2016-04-30 10:20:06 +00:00
|
|
|
self.algo_id = self.curve_info['algo_id']
|
2016-04-24 10:04:53 +00:00
|
|
|
|
2016-04-29 19:28:41 +00:00
|
|
|
def data(self):
|
|
|
|
"""Data for packet creation."""
|
2016-04-24 10:04:53 +00:00
|
|
|
header = struct.pack('>BLB',
|
|
|
|
4, # version
|
|
|
|
self.created, # creation
|
2016-04-30 10:20:06 +00:00
|
|
|
self.algo_id) # public key algorithm ID
|
2016-04-30 06:29:04 +00:00
|
|
|
oid = util.prefix_len('>B', self.curve_info['oid'])
|
|
|
|
blob = self.curve_info['serialize'](self.verifying_key)
|
2016-04-24 10:04:53 +00:00
|
|
|
return header + oid + blob
|
|
|
|
|
2016-04-29 19:28:41 +00:00
|
|
|
def data_to_hash(self):
|
|
|
|
"""Data for digest computation."""
|
|
|
|
return b'\x99' + util.prefix_len('>H', self.data())
|
2016-04-24 10:04:53 +00:00
|
|
|
|
2016-04-29 19:45:52 +00:00
|
|
|
def _fingerprint(self):
|
|
|
|
return hashlib.sha1(self.data_to_hash()).digest()
|
|
|
|
|
2016-04-24 10:04:53 +00:00
|
|
|
def key_id(self):
|
|
|
|
"""Short (8 byte) GPG key ID."""
|
2016-04-29 19:45:52 +00:00
|
|
|
return self._fingerprint()[-8:]
|
2016-04-24 10:04:53 +00:00
|
|
|
|
2016-04-29 19:45:52 +00:00
|
|
|
def __repr__(self):
|
2016-04-24 10:04:53 +00:00
|
|
|
"""Short (8 hexadecimal digits) GPG key ID."""
|
2016-04-29 19:45:52 +00:00
|
|
|
return '<{}>'.format(util.hexlify(self.key_id()))
|
|
|
|
|
|
|
|
__str__ = __repr__
|
2016-04-24 10:04:53 +00:00
|
|
|
|
2016-04-29 14:45:16 +00:00
|
|
|
|
|
|
|
class Signer(object):
|
|
|
|
"""Performs GPG signing operations."""
|
|
|
|
|
|
|
|
def __init__(self, user_id, created, curve_name):
|
|
|
|
"""Construct and loads a public key from the device."""
|
|
|
|
self.user_id = user_id
|
|
|
|
assert curve_name in formats.SUPPORTED_CURVES
|
|
|
|
|
|
|
|
self.conn = HardwareSigner(user_id, curve_name=curve_name)
|
|
|
|
self.pubkey = PublicKey(
|
|
|
|
curve_name=curve_name, created=created,
|
|
|
|
verifying_key=self.conn.pubkey())
|
|
|
|
|
|
|
|
log.info('%s GPG public key %s created at %s', curve_name,
|
2016-04-29 19:45:52 +00:00
|
|
|
self.pubkey, util.time_format(self.pubkey.created))
|
2016-04-29 14:45:16 +00:00
|
|
|
|
|
|
|
@classmethod
|
|
|
|
def from_public_key(cls, pubkey, user_id):
|
|
|
|
"""
|
|
|
|
Create from an existing GPG public key.
|
|
|
|
|
|
|
|
`pubkey` should be loaded via `decode.load_from_gpg(user_id)`
|
|
|
|
from the local GPG keyring.
|
|
|
|
"""
|
|
|
|
s = Signer(user_id=user_id,
|
|
|
|
created=pubkey['created'],
|
|
|
|
curve_name=_find_curve_by_algo_id(pubkey['algo']))
|
|
|
|
assert s.pubkey.key_id() == pubkey['key_id']
|
|
|
|
return s
|
|
|
|
|
2016-04-24 10:04:53 +00:00
|
|
|
def close(self):
|
|
|
|
"""Close connection and turn off the screen of the device."""
|
2016-04-28 11:45:45 +00:00
|
|
|
self.conn.close()
|
2016-04-24 10:04:53 +00:00
|
|
|
|
|
|
|
def export(self):
|
|
|
|
"""Export GPG public key, ready for "gpg2 --import"."""
|
2016-04-29 19:28:41 +00:00
|
|
|
pubkey_packet = packet(tag=6, blob=self.pubkey.data())
|
2016-04-24 10:04:53 +00:00
|
|
|
user_id_packet = packet(tag=13, blob=self.user_id)
|
|
|
|
|
2016-04-29 19:28:41 +00:00
|
|
|
data_to_sign = (self.pubkey.data_to_hash() +
|
2016-04-24 10:04:53 +00:00
|
|
|
user_id_packet[:1] +
|
|
|
|
util.prefix_len('>L', self.user_id))
|
|
|
|
log.info('signing public key "%s"', self.user_id)
|
|
|
|
hashed_subpackets = [
|
2016-04-29 14:45:16 +00:00
|
|
|
subpacket_time(self.pubkey.created), # signature creaion time
|
2016-04-24 10:04:53 +00:00
|
|
|
subpacket_byte(0x1B, 1 | 2), # key flags (certify & sign)
|
|
|
|
subpacket_byte(0x15, 8), # preferred hash (SHA256)
|
|
|
|
subpacket_byte(0x16, 0), # preferred compression (none)
|
|
|
|
subpacket_byte(0x17, 0x80)] # key server prefs (no-modify)
|
2016-04-29 14:45:16 +00:00
|
|
|
unhashed_subpackets = [
|
2016-04-30 13:27:43 +00:00
|
|
|
subpacket(16, self.pubkey.key_id()), # issuer key id
|
|
|
|
subpacket(100, MARKER)]
|
2016-04-29 14:45:16 +00:00
|
|
|
|
|
|
|
signature = _make_signature(
|
2016-04-30 10:20:06 +00:00
|
|
|
signer_func=self.conn.sign,
|
|
|
|
public_algo=self.pubkey.algo_id,
|
2016-04-29 14:45:16 +00:00
|
|
|
data_to_sign=data_to_sign,
|
|
|
|
sig_type=0x13, # user id & public key
|
|
|
|
hashed_subpackets=hashed_subpackets,
|
|
|
|
unhashed_subpackets=unhashed_subpackets)
|
2016-04-24 10:04:53 +00:00
|
|
|
|
|
|
|
sign_packet = packet(tag=2, blob=signature)
|
|
|
|
return pubkey_packet + user_id_packet + sign_packet
|
|
|
|
|
2016-04-29 19:45:52 +00:00
|
|
|
def subkey(self):
|
2016-04-30 07:04:44 +00:00
|
|
|
"""Export a subkey to `self.user_id` GPG primary key."""
|
2016-04-29 19:28:41 +00:00
|
|
|
subkey_packet = packet(tag=14, blob=self.pubkey.data())
|
2016-04-29 19:45:52 +00:00
|
|
|
primary = decode.load_from_gpg(self.user_id)
|
|
|
|
keygrip = agent.get_keygrip(self.user_id)
|
|
|
|
log.info('adding as subkey to %s (%s)', self.user_id, keygrip)
|
2016-04-29 19:28:41 +00:00
|
|
|
data_to_sign = primary['_to_hash'] + self.pubkey.data_to_hash()
|
2016-04-30 10:25:14 +00:00
|
|
|
|
|
|
|
# Primary Key Binding Signature
|
2016-04-28 19:10:40 +00:00
|
|
|
hashed_subpackets = [
|
2016-04-29 14:45:16 +00:00
|
|
|
subpacket_time(self.pubkey.created)] # signature creaion time
|
|
|
|
unhashed_subpackets = [
|
|
|
|
subpacket(16, self.pubkey.key_id())] # issuer key id
|
2016-04-30 12:39:32 +00:00
|
|
|
embedded_sig = _make_signature(signer_func=self.conn.sign,
|
|
|
|
data_to_sign=data_to_sign,
|
|
|
|
public_algo=self.pubkey.algo_id,
|
|
|
|
sig_type=0x19,
|
|
|
|
hashed_subpackets=hashed_subpackets,
|
|
|
|
unhashed_subpackets=unhashed_subpackets)
|
|
|
|
log.info('embedded signature: %r', embedded_sig)
|
2016-04-30 10:25:14 +00:00
|
|
|
|
|
|
|
# Subkey Binding Signature
|
2016-04-28 18:31:01 +00:00
|
|
|
hashed_subpackets = [
|
2016-04-29 14:45:16 +00:00
|
|
|
subpacket_time(self.pubkey.created), # signature creaion time
|
2016-04-28 18:31:01 +00:00
|
|
|
subpacket_byte(0x1B, 2)] # key flags (certify & sign)
|
2016-04-29 14:45:16 +00:00
|
|
|
unhashed_subpackets = [
|
|
|
|
subpacket(16, primary['key_id']), # issuer key id
|
2016-04-30 13:27:43 +00:00
|
|
|
subpacket(32, embedded_sig),
|
|
|
|
subpacket(100, MARKER)]
|
2016-04-30 10:20:06 +00:00
|
|
|
gpg_agent = AgentSigner(self.user_id)
|
|
|
|
signature = _make_signature(signer_func=gpg_agent.sign,
|
2016-04-29 14:45:16 +00:00
|
|
|
data_to_sign=data_to_sign,
|
2016-04-30 10:25:14 +00:00
|
|
|
public_algo=primary['algo'],
|
2016-04-29 14:45:16 +00:00
|
|
|
sig_type=0x18,
|
|
|
|
hashed_subpackets=hashed_subpackets,
|
2016-04-30 10:25:14 +00:00
|
|
|
unhashed_subpackets=unhashed_subpackets)
|
2016-04-28 18:31:01 +00:00
|
|
|
sign_packet = packet(tag=2, blob=signature)
|
|
|
|
return subkey_packet + sign_packet
|
|
|
|
|
2016-04-24 10:04:53 +00:00
|
|
|
def sign(self, msg, sign_time=None):
|
|
|
|
"""Sign GPG message at specified time."""
|
|
|
|
if sign_time is None:
|
|
|
|
sign_time = int(time.time())
|
|
|
|
|
|
|
|
log.info('signing %d byte message at %s',
|
|
|
|
len(msg), util.time_format(sign_time))
|
|
|
|
hashed_subpackets = [subpacket_time(sign_time)]
|
2016-04-29 14:45:16 +00:00
|
|
|
unhashed_subpackets = [
|
|
|
|
subpacket(16, self.pubkey.key_id())] # issuer key id
|
|
|
|
|
2016-04-30 10:25:14 +00:00
|
|
|
blob = _make_signature(signer_func=self.conn.sign,
|
|
|
|
data_to_sign=msg,
|
|
|
|
public_algo=self.pubkey.algo_id,
|
|
|
|
hashed_subpackets=hashed_subpackets,
|
|
|
|
unhashed_subpackets=unhashed_subpackets)
|
2016-04-24 10:04:53 +00:00
|
|
|
return packet(tag=2, blob=blob)
|
|
|
|
|
2016-04-29 14:45:16 +00:00
|
|
|
|
2016-04-30 10:20:06 +00:00
|
|
|
def _make_signature(signer_func, data_to_sign, public_algo,
|
|
|
|
hashed_subpackets, unhashed_subpackets, sig_type=0):
|
2016-04-30 12:39:32 +00:00
|
|
|
# pylint: disable=too-many-arguments
|
2016-04-29 14:45:16 +00:00
|
|
|
header = struct.pack('>BBBB',
|
|
|
|
4, # version
|
|
|
|
sig_type, # rfc4880 (section-5.2.1)
|
2016-04-30 10:20:06 +00:00
|
|
|
public_algo,
|
2016-04-29 14:45:16 +00:00
|
|
|
8) # hash_alg (SHA256)
|
|
|
|
hashed = subpackets(*hashed_subpackets)
|
|
|
|
unhashed = subpackets(*unhashed_subpackets)
|
|
|
|
tail = b'\x04\xff' + struct.pack('>L', len(header) + len(hashed))
|
|
|
|
data_to_hash = data_to_sign + header + hashed + tail
|
|
|
|
|
|
|
|
log.debug('hashing %d bytes', len(data_to_hash))
|
|
|
|
digest = hashlib.sha256(data_to_hash).digest()
|
|
|
|
|
2016-04-30 10:20:06 +00:00
|
|
|
sig = signer_func(digest=digest)
|
2016-04-29 14:45:16 +00:00
|
|
|
|
|
|
|
return bytes(header + hashed + unhashed +
|
|
|
|
digest[:2] + # used for decoder's sanity check
|
|
|
|
sig) # actual ECDSA signature
|
2016-04-24 10:04:53 +00:00
|
|
|
|
|
|
|
|
|
|
|
def _split_lines(body, size):
|
|
|
|
lines = []
|
|
|
|
for i in range(0, len(body), size):
|
|
|
|
lines.append(body[i:i+size] + '\n')
|
|
|
|
return ''.join(lines)
|
|
|
|
|
|
|
|
|
|
|
|
def armor(blob, type_str):
|
|
|
|
"""See https://tools.ietf.org/html/rfc4880#section-6 for details."""
|
|
|
|
head = '-----BEGIN PGP {}-----\nVersion: GnuPG v2\n\n'.format(type_str)
|
|
|
|
body = base64.b64encode(blob)
|
|
|
|
checksum = base64.b64encode(util.crc24(blob))
|
|
|
|
tail = '-----END PGP {}-----\n'.format(type_str)
|
|
|
|
return head + _split_lines(body, 64) + '=' + checksum + '\n' + tail
|