From ab192619f47e7fb212d036c237bf8e2e630843e0 Mon Sep 17 00:00:00 2001 From: Roman Zeyde Date: Sat, 30 Apr 2016 16:50:01 +0300 Subject: [PATCH] gpg: move protocol utils to proto.py --- trezor_agent/gpg/encode.py | 150 ++++++++----------------------------- 1 file changed, 30 insertions(+), 120 deletions(-) diff --git a/trezor_agent/gpg/encode.py b/trezor_agent/gpg/encode.py index a95c139..3a2a4e8 100644 --- a/trezor_agent/gpg/encode.py +++ b/trezor_agent/gpg/encode.py @@ -5,103 +5,12 @@ import logging import struct import time -from . import agent, decode +from . import agent, decode, proto from .. import client, factory, formats, util log = logging.getLogger(__name__) -def packet(tag, blob): - """Create small GPG packet.""" - 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] - leading_byte = 0x80 | (tag << 2) | (length_type) - return struct.pack('>B', leading_byte) + util.prefix_len(fmt, blob) - - -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 - data_bytes = bytearray(data_size) - for i in range(data_size): - data_bytes[i] = value & 0xFF - value = value >> 8 - - data_bytes.reverse() - return struct.pack('>H', bits) + bytes(data_bytes) - - -def _serialize_nist256(vk): - return mpi((4 << 512) | - (vk.pubkey.point.x() << 256) | - (vk.pubkey.point.y())) - - -def _serialize_ed25519(vk): - 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, - 'serialize': _serialize_nist256 - }, - formats.CURVE_ED25519: { - 'oid': b'\x2B\x06\x01\x04\x01\xDA\x47\x0F\x01', - 'algo_id': 22, - 'serialize': _serialize_ed25519 - } -} - - -MARKER = b'TREZOR-GPG' - - -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 - - class HardwareSigner(object): """Sign messages and get public keys from a hardware device.""" @@ -132,7 +41,8 @@ class HardwareSigner(object): 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:])) + return (proto.mpi(util.bytes2num(sig[:32])) + + proto.mpi(util.bytes2num(sig[32:]))) def close(self): """Close the connection to the device.""" @@ -157,7 +67,7 @@ class AgentSigner(object): """Sign the digest and return an ECDSA signature.""" params = agent.sign(sock=self.sock, keygrip=self.keygrip, digest=digest) - return b''.join(mpi(p) for p in params) + return b''.join(proto.mpi(p) for p in params) def close(self): """Close the connection to gpg-agent.""" @@ -169,7 +79,7 @@ class PublicKey(object): def __init__(self, curve_name, created, verifying_key): """Contruct using a ECDSA VerifyingKey object.""" - self.curve_info = SUPPORTED_CURVES[curve_name] + self.curve_info = proto.SUPPORTED_CURVES[curve_name] self.created = int(created) # time since Epoch self.verifying_key = verifying_key self.algo_id = self.curve_info['algo_id'] @@ -228,7 +138,7 @@ class Signer(object): """ s = Signer(user_id=user_id, created=pubkey['created'], - curve_name=_find_curve_by_algo_id(pubkey['algo'])) + curve_name=proto.find_curve_by_algo_id(pubkey['algo'])) assert s.pubkey.key_id() == pubkey['key_id'] return s @@ -238,22 +148,22 @@ class Signer(object): def export(self): """Export GPG public key, ready for "gpg2 --import".""" - pubkey_packet = packet(tag=6, blob=self.pubkey.data()) - user_id_packet = packet(tag=13, blob=self.user_id) + pubkey_packet = proto.packet(tag=6, blob=self.pubkey.data()) + user_id_packet = proto.packet(tag=13, blob=self.user_id) data_to_sign = (self.pubkey.data_to_hash() + user_id_packet[:1] + util.prefix_len('>L', self.user_id)) log.info('signing public key "%s"', self.user_id) hashed_subpackets = [ - subpacket_time(self.pubkey.created), # signature creaion time - 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) + proto.subpacket_time(self.pubkey.created), # signature time + proto.subpacket_byte(0x1B, 1 | 2), # key flags (certify & sign) + proto.subpacket_byte(0x15, 8), # preferred hash (SHA256) + proto.subpacket_byte(0x16, 0), # preferred compression (none) + proto.subpacket_byte(0x17, 0x80)] # key server prefs (no-modify) unhashed_subpackets = [ - subpacket(16, self.pubkey.key_id()), # issuer key id - subpacket(100, MARKER)] + proto.subpacket(16, self.pubkey.key_id()), # issuer key id + proto.MARKER_SUBPACKET] signature = _make_signature( signer_func=self.conn.sign, @@ -263,12 +173,12 @@ class Signer(object): hashed_subpackets=hashed_subpackets, unhashed_subpackets=unhashed_subpackets) - sign_packet = packet(tag=2, blob=signature) + sign_packet = proto.packet(tag=2, blob=signature) return pubkey_packet + user_id_packet + sign_packet def subkey(self): """Export a subkey to `self.user_id` GPG primary key.""" - subkey_packet = packet(tag=14, blob=self.pubkey.data()) + subkey_packet = proto.packet(tag=14, blob=self.pubkey.data()) 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) @@ -276,9 +186,9 @@ class Signer(object): # Primary Key Binding Signature hashed_subpackets = [ - subpacket_time(self.pubkey.created)] # signature creaion time + proto.subpacket_time(self.pubkey.created)] # signature time unhashed_subpackets = [ - subpacket(16, self.pubkey.key_id())] # issuer key id + proto.subpacket(16, self.pubkey.key_id())] # issuer key id embedded_sig = _make_signature(signer_func=self.conn.sign, data_to_sign=data_to_sign, public_algo=self.pubkey.algo_id, @@ -289,12 +199,12 @@ class Signer(object): # Subkey Binding Signature hashed_subpackets = [ - subpacket_time(self.pubkey.created), # signature creaion time - subpacket_byte(0x1B, 2)] # key flags (certify & sign) + proto.subpacket_time(self.pubkey.created), # signature time + proto.subpacket_byte(0x1B, 2)] # key flags (certify & sign) unhashed_subpackets = [ - subpacket(16, primary['key_id']), # issuer key id - subpacket(32, embedded_sig), - subpacket(100, MARKER)] + proto.subpacket(16, primary['key_id']), # issuer key id + proto.subpacket(32, embedded_sig), + proto.MARKER_SUBPACKET] gpg_agent = AgentSigner(self.user_id) signature = _make_signature(signer_func=gpg_agent.sign, data_to_sign=data_to_sign, @@ -302,7 +212,7 @@ class Signer(object): sig_type=0x18, hashed_subpackets=hashed_subpackets, unhashed_subpackets=unhashed_subpackets) - sign_packet = packet(tag=2, blob=signature) + sign_packet = proto.packet(tag=2, blob=signature) return subkey_packet + sign_packet def sign(self, msg, sign_time=None): @@ -312,16 +222,16 @@ class Signer(object): log.info('signing %d byte message at %s', len(msg), util.time_format(sign_time)) - hashed_subpackets = [subpacket_time(sign_time)] + hashed_subpackets = [proto.subpacket_time(sign_time)] unhashed_subpackets = [ - subpacket(16, self.pubkey.key_id())] # issuer key id + proto.subpacket(16, self.pubkey.key_id())] # issuer key id 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) - return packet(tag=2, blob=blob) + return proto.packet(tag=2, blob=blob) def _make_signature(signer_func, data_to_sign, public_algo, @@ -332,8 +242,8 @@ def _make_signature(signer_func, data_to_sign, public_algo, sig_type, # rfc4880 (section-5.2.1) public_algo, 8) # hash_alg (SHA256) - hashed = subpackets(*hashed_subpackets) - unhashed = subpackets(*unhashed_subpackets) + hashed = proto.subpackets(*hashed_subpackets) + unhashed = proto.subpackets(*unhashed_subpackets) tail = b'\x04\xff' + struct.pack('>L', len(header) + len(hashed)) data_to_hash = data_to_sign + header + hashed + tail