gpg: remove verifying logic from decoding
This commit is contained in:
Roman Zeyde
@@ -6,9 +6,6 @@ import io
|
||||
import logging
|
||||
import struct
|
||||
|
||||
import ecdsa
|
||||
import ed25519
|
||||
|
||||
from . import protocol
|
||||
from .. import util
|
||||
|
||||
@@ -52,45 +49,10 @@ def parse_mpis(s, n):
|
||||
return [parse_mpi(s) for _ in range(n)]
|
||||
|
||||
|
||||
def _parse_nist256p1_verifier(mpi):
|
||||
prefix, x, y = util.split_bits(mpi, 4, 256, 256)
|
||||
assert prefix == 4
|
||||
point = ecdsa.ellipticcurve.Point(curve=ecdsa.NIST256p.curve,
|
||||
x=x, y=y)
|
||||
vk = ecdsa.VerifyingKey.from_public_point(
|
||||
point=point, curve=ecdsa.curves.NIST256p,
|
||||
hashfunc=hashlib.sha256)
|
||||
|
||||
def _nist256p1_verify(signature, digest):
|
||||
result = vk.verify_digest(signature=signature,
|
||||
digest=digest,
|
||||
sigdecode=lambda rs, order: rs)
|
||||
log.debug('nist256p1 ECDSA signature is OK (%s)', result)
|
||||
return _nist256p1_verify, vk
|
||||
|
||||
|
||||
def _parse_ed25519_verifier(mpi):
|
||||
prefix, value = util.split_bits(mpi, 8, 256)
|
||||
assert prefix == 0x40
|
||||
vk = ed25519.VerifyingKey(util.num2bytes(value, size=32))
|
||||
|
||||
def _ed25519_verify(signature, digest):
|
||||
sig = b''.join(util.num2bytes(val, size=32)
|
||||
for val in signature)
|
||||
result = vk.verify(sig, digest)
|
||||
log.debug('ed25519 ECDSA signature is OK (%s)', result)
|
||||
return _ed25519_verify, vk
|
||||
|
||||
|
||||
def _parse_curve25519_verifier(_):
|
||||
log.warning('Curve25519 ECDH is not verified')
|
||||
return None, None
|
||||
|
||||
|
||||
SUPPORTED_CURVES = {
|
||||
b'\x2A\x86\x48\xCE\x3D\x03\x01\x07': _parse_nist256p1_verifier,
|
||||
b'\x2B\x06\x01\x04\x01\xDA\x47\x0F\x01': _parse_ed25519_verifier,
|
||||
b'\x2B\x06\x01\x04\x01\x97\x55\x01\x05\x01': _parse_curve25519_verifier,
|
||||
b'\x2A\x86\x48\xCE\x3D\x03\x01\x07',
|
||||
b'\x2B\x06\x01\x04\x01\xDA\x47\x0F\x01',
|
||||
b'\x2B\x06\x01\x04\x01\x97\x55\x01\x05\x01',
|
||||
}
|
||||
|
||||
RSA_ALGO_IDS = {1, 2, 3}
|
||||
@@ -174,11 +136,9 @@ def _parse_pubkey(stream, packet_type='pubkey'):
|
||||
oid = stream.read(oid_size)
|
||||
assert oid in SUPPORTED_CURVES, util.hexlify(oid)
|
||||
p['curve_oid'] = oid
|
||||
parser = SUPPORTED_CURVES[oid]
|
||||
|
||||
mpi = parse_mpi(stream)
|
||||
log.debug('mpi: %x (%d bits)', mpi, mpi.bit_length())
|
||||
p['verifier'], p['verifying_key'] = parser(mpi)
|
||||
leftover = stream.read()
|
||||
if leftover:
|
||||
leftover = io.BytesIO(leftover)
|
||||
@@ -303,20 +263,7 @@ def load_public_key(pubkey_bytes, use_custom=False, ecdh=False):
|
||||
packets = list(parse_packets(stream))
|
||||
pubkey, userid, signature = packets[:3]
|
||||
packets = packets[3:]
|
||||
|
||||
hash_alg = HASH_ALGORITHMS.get(signature['hash_alg'])
|
||||
if hash_alg is not None:
|
||||
digest = digest_packets(packets=[pubkey, userid, signature],
|
||||
hasher=hashlib.new(hash_alg))
|
||||
assert signature['hash_prefix'] == digest[:2]
|
||||
|
||||
log.debug('loaded public key "%s"', userid['value'])
|
||||
if hash_alg is not None and pubkey.get('verifier'):
|
||||
verify_digest(pubkey=pubkey, digest=digest,
|
||||
signature=signature['sig'], label='GPG public key')
|
||||
else:
|
||||
log.warning('public key %s is not verified!',
|
||||
util.hexlify(pubkey['key_id']))
|
||||
|
||||
packet = pubkey
|
||||
while use_custom:
|
||||
@@ -347,17 +294,6 @@ def load_signature(stream, original_data):
|
||||
return signature, digest
|
||||
|
||||
|
||||
def verify_digest(pubkey, digest, signature, label):
|
||||
"""Verify a digest signature from a specified public key."""
|
||||
verifier = pubkey['verifier']
|
||||
try:
|
||||
verifier(signature, digest)
|
||||
log.debug('%s is OK', label)
|
||||
except ecdsa.keys.BadSignatureError:
|
||||
log.error('Bad %s!', label)
|
||||
raise ValueError('Invalid ECDSA signature for {}'.format(label))
|
||||
|
||||
|
||||
def remove_armor(armored_data):
|
||||
"""Decode armored data into its binary form."""
|
||||
stream = io.BytesIO(armored_data)
|
||||
@@ -366,11 +302,3 @@ def remove_armor(armored_data):
|
||||
payload, checksum = data[:-3], data[-3:]
|
||||
assert util.crc24(payload) == checksum
|
||||
return payload
|
||||
|
||||
|
||||
def verify(pubkey, signature, original_data):
|
||||
"""Verify correctness of public key and signature."""
|
||||
stream = io.BytesIO(remove_armor(signature))
|
||||
signature, digest = load_signature(stream, original_data)
|
||||
verify_digest(pubkey=pubkey, digest=digest,
|
||||
signature=signature['sig'], label='GPG signature')
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
import glob
|
||||
import hashlib
|
||||
import io
|
||||
import os
|
||||
|
||||
@@ -31,84 +30,6 @@ def test_mpi():
|
||||
assert decode.parse_mpis(util.Reader(s), n=2) == [0x123, 5]
|
||||
|
||||
|
||||
def assert_subdict(d, s):
|
||||
for k, v in s.items():
|
||||
assert d[k] == v
|
||||
|
||||
|
||||
def test_primary_nist256p1():
|
||||
# pylint: disable=line-too-long
|
||||
data = b'''-----BEGIN PGP PUBLIC KEY BLOCK-----
|
||||
Version: GnuPG v2
|
||||
|
||||
mFIEV0hI1hMIKoZIzj0DAQcCAwTXY7aq01xMPSU7gTHU9B7Z2CFoCk1Y4WYb8Tiy
|
||||
hurvIZ5la6+UEgAKF9HXpQo0yE+HQOgufoLlCpdE7NoEUb+HtAd0ZXN0aW5niHYE
|
||||
ExMIABIFAldISNYCGwMCFQgCFgACF4AAFgkQTcCehfpEIPILZFRSRVpPUi1HUEeV
|
||||
3QEApHKmBkbLVZNpsB8q9mBzKytxnOHNB3QWDuoKJu/ERi4A/1wRGZ/B0BDazHck
|
||||
zpR9luXTKwMEl+mlZmwEFKZXBmir
|
||||
=oyj0
|
||||
-----END PGP PUBLIC KEY BLOCK-----
|
||||
'''
|
||||
stream = io.BytesIO(decode.remove_armor(data))
|
||||
pubkey, user_id, signature = list(decode.parse_packets(stream))
|
||||
expected_pubkey = {
|
||||
'created': 1464355030, 'type': 'pubkey', 'tag': 6,
|
||||
'version': 4, 'algo': 19, 'key_id': b'M\xc0\x9e\x85\xfaD \xf2',
|
||||
'_to_hash': b'\x99\x00R\x04WHH\xd6\x13\x08*\x86H\xce=\x03\x01\x07\x02\x03\x04\xd7c\xb6\xaa\xd3\\L=%;\x811\xd4\xf4\x1e\xd9\xd8!h\nMX\xe1f\x1b\xf18\xb2\x86\xea\xef!\x9eek\xaf\x94\x12\x00\n\x17\xd1\xd7\xa5\n4\xc8O\x87@\xe8.~\x82\xe5\n\x97D\xec\xda\x04Q\xbf\x87' # nopep8
|
||||
}
|
||||
assert_subdict(pubkey, expected_pubkey)
|
||||
point = pubkey['verifying_key'].pubkey.point
|
||||
assert point.x(), point.y() == (
|
||||
97423441028100245505102979561460969898742433559010922791700160771755342491425,
|
||||
71644624850142103522769833619875243486871666152651730678601507641225861250951
|
||||
)
|
||||
assert_subdict(user_id, {
|
||||
'tag': 13, 'type': 'user_id', 'value': b'testing',
|
||||
'_to_hash': b'\xb4\x00\x00\x00\x07testing'
|
||||
})
|
||||
assert_subdict(signature, {
|
||||
'pubkey_alg': 19, '_is_custom': True, 'hash_alg': 8, 'tag': 2,
|
||||
'sig_type': 19, 'version': 4, 'type': 'signature', 'hash_prefix': b'\x95\xdd',
|
||||
'sig': (74381873592149178031432444136130575481350858387410643140628758456112511206958,
|
||||
41642995320462795718437755373080464775445470754419831653624197847615308982443),
|
||||
'hashed_subpackets': [b'\x02WHH\xd6', b'\x1b\x03', b'\x15\x08', b'\x16\x00', b'\x17\x80'],
|
||||
'unhashed_subpackets': [b'\x10M\xc0\x9e\x85\xfaD \xf2', b'dTREZOR-GPG'],
|
||||
'_to_hash': b'\x04\x13\x13\x08\x00\x12\x05\x02WHH\xd6\x02\x1b\x03\x02\x15\x08\x02\x16\x00\x02\x17\x80\x04\xff\x00\x00\x00\x18' # nopep8
|
||||
})
|
||||
|
||||
digest = decode.digest_packets(packets=[pubkey, user_id, signature],
|
||||
hasher=hashlib.sha256())
|
||||
decode.verify_digest(pubkey=pubkey, digest=digest,
|
||||
signature=signature['sig'],
|
||||
label='GPG primary public key')
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
bad_digest = b'\x00' * len(digest)
|
||||
decode.verify_digest(pubkey=pubkey, digest=bad_digest,
|
||||
signature=signature['sig'],
|
||||
label='GPG primary public key')
|
||||
|
||||
message = b'Hello, World!\n'
|
||||
signature = b'''-----BEGIN PGP SIGNATURE-----
|
||||
Version: GnuPG v2
|
||||
|
||||
iF4EABMIAAYFAldIlfQACgkQTcCehfpEIPKOUgD9FjaeWla4wOuDZ7P6fhkT5nZp
|
||||
KDQU0N5KmNwLlt2kwo4A/jQkBII2cI8tTqOVTLNRXXqIOsMf/fG4jKM/VOFc/01c
|
||||
=dC+z
|
||||
-----END PGP SIGNATURE-----
|
||||
'''
|
||||
decode.verify(pubkey=pubkey, signature=signature, original_data=message)
|
||||
|
||||
pubkey = decode.load_public_key(pubkey_bytes=decode.remove_armor(data))
|
||||
assert_subdict(pubkey, expected_pubkey)
|
||||
assert_subdict(pubkey, {'user_id': b'testing'})
|
||||
|
||||
pubkey = decode.load_public_key(pubkey_bytes=decode.remove_armor(data),
|
||||
use_custom=True)
|
||||
assert_subdict(pubkey, expected_pubkey)
|
||||
assert_subdict(pubkey, {'user_id': b'testing'})
|
||||
|
||||
|
||||
cwd = os.path.join(os.path.dirname(__file__))
|
||||
input_files = glob.glob(os.path.join(cwd, '*.gpg'))
|
||||
|
||||
|
||||
Reference in New Issue
Block a user