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

gpg: move protocol utils to proto.py

Browse Source
This commit is contained in:
Roman Zeyde
2016-04-30 16:50:01 +03:00
parent f982d785bd
commit ab192619f4
1 changed files with 30 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

150
trezor_agent/gpg/encode.py
View File

@@ -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