"""Create GPG ECDSA signatures and public keys using TREZOR device.""" import logging import time from . import decode, keyring, proto from .. import client, factory, formats, util log = logging.getLogger(__name__) class HardwareSigner(object): """Sign messages and get public keys from a hardware device.""" def __init__(self, user_id, curve_name): """Connect to the device and retrieve required public key.""" self.client_wrapper = factory.load() self.identity = self.client_wrapper.identity_type() self.identity.proto = 'gpg' self.identity.host = user_id self.curve_name = curve_name def pubkey(self): """Return public key as VerifyingKey object.""" addr = client.get_address(self.identity) public_node = self.client_wrapper.connection.get_public_node( n=addr, ecdsa_curve_name=self.curve_name) return formats.decompress_pubkey( pubkey=public_node.node.public_key, curve_name=self.curve_name) def sign(self, digest): """Sign the digest and return a serialized signature.""" result = self.client_wrapper.connection.sign_identity( identity=self.identity, challenge_hidden=digest, challenge_visual='', ecdsa_curve_name=self.curve_name) assert result.signature[:1] == b'\x00' sig = result.signature[1:] return (util.bytes2num(sig[:32]), util.bytes2num(sig[32:])) def ecdh(self, pubkey): """Derive shared secret using ECDH from remote public key.""" result = self.client_wrapper.connection.sign_identity( identity=self.identity, challenge_hidden=pubkey, challenge_visual=b'', ecdsa_curve_name=self.curve_name) assert len(result.signature) == 65 assert result.signature[:1] == b'\x04' return result.signature def close(self): """Close the connection to the device.""" self.client_wrapper.connection.clear_session() self.client_wrapper.connection.close() class AgentSigner(object): """Sign messages and get public keys using gpg-agent tool.""" def __init__(self, user_id): """Connect to the agent and retrieve required public key.""" self.sock = keyring.connect_to_agent() self.keygrip = keyring.get_keygrip(user_id) def sign(self, digest): """Sign the digest and return an ECDSA/RSA/DSA signature.""" return keyring.sign_digest(sock=self.sock, keygrip=self.keygrip, digest=digest) def close(self): """Close the connection to gpg-agent.""" self.sock.close() def _time_format(t): return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(t)) def create_primary(user_id, pubkey, signer_func): """Export new primary GPG public key, ready for "gpg2 --import".""" pubkey_packet = proto.packet(tag=6, blob=pubkey.data()) user_id_packet = proto.packet(tag=13, blob=user_id.encode('ascii')) data_to_sign = (pubkey.data_to_hash() + user_id_packet[:1] + util.prefix_len('>L', user_id.encode('ascii'))) log.info('signing public key "%s"', user_id) hashed_subpackets = [ proto.subpacket_time(pubkey.created), # signature time # https://tools.ietf.org/html/rfc4880#section-5.2.3.4 proto.subpacket_byte(0x1B, 1 | 2), # key flags (certify & sign) # https://tools.ietf.org/html/rfc4880#section-5.2.3.21 proto.subpacket_byte(0x15, 8), # preferred hash (SHA256) # https://tools.ietf.org/html/rfc4880#section-5.2.3.8 proto.subpacket_byte(0x16, 0), # preferred compression (none) # https://tools.ietf.org/html/rfc4880#section-5.2.3.9 proto.subpacket_byte(0x17, 0x80) # key server prefs (no-modify) # https://tools.ietf.org/html/rfc4880#section-5.2.3.17 ] unhashed_subpackets = [ proto.subpacket(16, pubkey.key_id()), # issuer key id proto.CUSTOM_SUBPACKET] signature = proto.make_signature( signer_func=signer_func, public_algo=pubkey.algo_id, data_to_sign=data_to_sign, sig_type=0x13, # user id & public key hashed_subpackets=hashed_subpackets, unhashed_subpackets=unhashed_subpackets) sign_packet = proto.packet(tag=2, blob=signature) return pubkey_packet + user_id_packet + sign_packet def create_subkey(primary_bytes, pubkey, signer_func, ecdh=False): """Export new subkey to GPG primary key.""" subkey_packet = proto.packet(tag=14, blob=pubkey.data()) primary = decode.load_public_key(primary_bytes) log.info('adding subkey to primary GPG key "%s" (%s)', primary['user_id'], util.hexlify(primary['key_id'])) data_to_sign = primary['_to_hash'] + pubkey.data_to_hash() if ecdh: embedded_sig = None else: # Primary Key Binding Signature hashed_subpackets = [ proto.subpacket_time(pubkey.created)] # signature time unhashed_subpackets = [ proto.subpacket(16, pubkey.key_id())] # issuer key id log.info('confirm signing subkey with hardware device') embedded_sig = proto.make_signature( signer_func=signer_func, data_to_sign=data_to_sign, public_algo=pubkey.algo_id, sig_type=0x19, hashed_subpackets=hashed_subpackets, unhashed_subpackets=unhashed_subpackets) # Subkey Binding Signature # Key flags: https://tools.ietf.org/html/rfc4880#section-5.2.3.21 # (certify & sign) (encrypt) flags = (2) if (not ecdh) else (4 | 8) hashed_subpackets = [ proto.subpacket_time(pubkey.created), # signature time proto.subpacket_byte(0x1B, flags)] unhashed_subpackets = [] unhashed_subpackets.append(proto.subpacket(16, primary['key_id'])) if embedded_sig is not None: unhashed_subpackets.append(proto.subpacket(32, embedded_sig)) unhashed_subpackets.append(proto.CUSTOM_SUBPACKET) log.info('confirm signing subkey with gpg-agent') # TODO: support TREZOR-based primary key gpg_agent = AgentSigner(primary['user_id']) signature = proto.make_signature( signer_func=gpg_agent.sign, data_to_sign=data_to_sign, public_algo=primary['algo'], sig_type=0x18, hashed_subpackets=hashed_subpackets, unhashed_subpackets=unhashed_subpackets) sign_packet = proto.packet(tag=2, blob=signature) return primary_bytes + subkey_packet + sign_packet class Factory(object): """Performs GPG signing operations.""" def __init__(self, user_id, created, curve_name, ecdh=False): """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 = proto.PublicKey( curve_name=curve_name, created=created, verifying_key=self.conn.pubkey(), ecdh=ecdh) self.ecdh = ecdh log.info('%s created at %s for "%s"', self.pubkey, _time_format(self.pubkey.created), user_id) @classmethod def from_public_key(cls, pubkey): """Create from an existing GPG public key.""" f = cls(user_id=pubkey['user_id'], created=pubkey['created'], curve_name=proto.find_curve_by_algo_id(pubkey['algo']), ecdh=(pubkey['algo'] == proto.ECDH_ALGO_ID)) assert f.pubkey.key_id() == pubkey['key_id'] return f def close(self): """Close connection and turn off the screen of the device.""" self.conn.close() def create_primary(self): """Export new subkey to GPG primary key.""" return create_primary(user_id=self.user_id, pubkey=self.pubkey, signer_func=self.conn.sign) def create_subkey(self, primary_bytes): """Export new subkey to GPG primary key.""" return create_subkey(primary_bytes=primary_bytes, pubkey=self.pubkey, signer_func=self.conn.sign, ecdh=self.ecdh) def sign_message(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), _time_format(sign_time)) hashed_subpackets = [proto.subpacket_time(sign_time)] unhashed_subpackets = [ proto.subpacket(16, self.pubkey.key_id())] # issuer key id blob = proto.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 proto.packet(tag=2, blob=blob) def get_shared_secret(self, pubkey): """Derive shared secret using ECDH from remote public key.""" return self.conn.ecdh(pubkey)