Split the package into a shared library and separate per-device packages

libagent/gpg/tests/__init__.py

@@ -0,0 +1 @@
+"""Tests for GPG module."""

libagent/gpg/tests/test_decode.py

@@ -0,0 +1,62 @@
+import glob
+import io
+import os
+
+import pytest
+
+from .. import decode, protocol
+from ... import util
+
+
+def test_subpackets():
+    s = io.BytesIO(b'\x00\x05\x02\xAB\xCD\x01\xEF')
+    assert decode.parse_subpackets(util.Reader(s)) == [b'\xAB\xCD', b'\xEF']
+
+
+def test_subpackets_prefix():
+    for n in [0, 1, 2, 4, 5, 10, 191, 192, 193,
+              255, 256, 257, 8383, 8384, 65530]:
+        item = b'?' * n  # create dummy subpacket
+        prefixed = protocol.subpackets(item)
+        result = decode.parse_subpackets(util.Reader(io.BytesIO(prefixed)))
+        assert [item] == result
+
+
+def test_mpi():
+    s = io.BytesIO(b'\x00\x09\x01\x23')
+    assert decode.parse_mpi(util.Reader(s)) == 0x123
+
+    s = io.BytesIO(b'\x00\x09\x01\x23\x00\x03\x05')
+    assert decode.parse_mpis(util.Reader(s), n=2) == [0x123, 5]
+
+
+cwd = os.path.join(os.path.dirname(__file__))
+input_files = glob.glob(os.path.join(cwd, '*.gpg'))
+
+
+@pytest.fixture(params=input_files)
+def public_key_path(request):
+    return request.param
+
+
+def test_gpg_files(public_key_path):  # pylint: disable=redefined-outer-name
+    with open(public_key_path, 'rb') as f:
+        assert list(decode.parse_packets(f))
+
+
+def test_has_custom_subpacket():
+    sig = {'unhashed_subpackets': []}
+    assert not decode.has_custom_subpacket(sig)
+
+    custom_markers = [
+        protocol.CUSTOM_SUBPACKET,
+        protocol.subpacket(10, protocol.CUSTOM_KEY_LABEL),
+    ]
+    for marker in custom_markers:
+        sig = {'unhashed_subpackets': [marker]}
+        assert decode.has_custom_subpacket(sig)
+
+
+def test_load_by_keygrip_missing():
+    with pytest.raises(KeyError):
+        decode.load_by_keygrip(pubkey_bytes=b'', keygrip=b'')

libagent/gpg/tests/test_keyring.py

@@ -0,0 +1,101 @@
+import io
+
+import mock
+
+from .. import keyring
+
+
+def test_unescape_short():
+    assert keyring.unescape(b'abc%0AX%0D %25;.-+()') == b'abc\nX\r %;.-+()'
+
+
+def test_unescape_long():
+    escaped = (b'D (7:sig-val(3:dsa(1:r32:\x1d\x15.\x12\xe8h\x19\xd9O\xeb\x06'
+               b'yD?a:/\xae\xdb\xac\x93\xa6\x86\xcbs\xb8\x03\xf1\xcb\x89\xc7'
+               b'\x1f)(1:s32:%25\xb5\x04\x94\xc7\xc4X\xc7\xe0%0D\x08\xbb%0DuN'
+               b'\x9c6}[\xc2=t\x8c\xfdD\x81\xe8\xdd\x86=\xe2\xa9)))')
+    unescaped = (b'D (7:sig-val(3:dsa(1:r32:\x1d\x15.\x12\xe8h\x19\xd9O\xeb'
+                 b'\x06yD?a:/\xae\xdb\xac\x93\xa6\x86\xcbs\xb8\x03\xf1\xcb\x89'
+                 b'\xc7\x1f)(1:s32:%\xb5\x04\x94\xc7\xc4X\xc7\xe0\r\x08\xbb\ru'
+                 b'N\x9c6}[\xc2=t\x8c\xfdD\x81\xe8\xdd\x86=\xe2\xa9)))')
+    assert keyring.unescape(escaped) == unescaped
+
+
+def test_parse_term():
+    assert keyring.parse(b'4:abcdXXX') == (b'abcd', b'XXX')
+
+
+def test_parse_ecdsa():
+    sig, rest = keyring.parse(b'(7:sig-val(5:ecdsa'
+                              b'(1:r2:\x01\x02)(1:s2:\x03\x04)))')
+    values = [[b'r', b'\x01\x02'], [b's', b'\x03\x04']]
+    assert sig == [b'sig-val', [b'ecdsa'] + values]
+    assert rest == b''
+    assert keyring.parse_sig(sig) == (0x102, 0x304)
+
+
+def test_parse_rsa():
+    sig, rest = keyring.parse(b'(7:sig-val(3:rsa(1:s4:\x01\x02\x03\x04)))')
+    assert sig == [b'sig-val', [b'rsa', [b's', b'\x01\x02\x03\x04']]]
+    assert rest == b''
+    assert keyring.parse_sig(sig) == (0x1020304,)
+
+
+class FakeSocket(object):
+    def __init__(self):
+        self.rx = io.BytesIO()
+        self.tx = io.BytesIO()
+
+    def recv(self, n):
+        return self.rx.read(n)
+
+    def sendall(self, data):
+        self.tx.write(data)
+
+
+def test_sign_digest():
+    sock = FakeSocket()
+    sock.rx.write(b'OK Pleased to meet you, process XYZ\n')
+    sock.rx.write(b'OK\n' * 6)
+    sock.rx.write(b'D (7:sig-val(3:rsa(1:s16:0123456789ABCDEF)))\n')
+    sock.rx.seek(0)
+    keygrip = '1234'
+    digest = b'A' * 32
+    sp = mock.Mock(spec=['check_output'])
+    sp.check_output.return_value = '/dev/pts/0'
+    sig = keyring.sign_digest(sock=sock, keygrip=keygrip,
+                              digest=digest, sp=sp,
+                              environ={'DISPLAY': ':0'})
+    assert sig == (0x30313233343536373839414243444546,)
+    assert sock.tx.getvalue() == b'''RESET
+OPTION ttyname=/dev/pts/0
+OPTION display=:0
+SIGKEY 1234
+SETHASH 8 4141414141414141414141414141414141414141414141414141414141414141
+SETKEYDESC Sign+a+new+TREZOR-based+subkey
+PKSIGN
+'''
+
+
+def test_iterlines():
+    sock = FakeSocket()
+    sock.rx.write(b'foo\nbar\nxyz')
+    sock.rx.seek(0)
+    assert list(keyring.iterlines(sock)) == [b'foo', b'bar']
+
+
+def test_get_agent_sock_path():
+    sp = mock.Mock(spec=['check_output'])
+    sp.check_output.return_value = b'''sysconfdir:/usr/local/etc/gnupg
+bindir:/usr/local/bin
+libexecdir:/usr/local/libexec
+libdir:/usr/local/lib/gnupg
+datadir:/usr/local/share/gnupg
+localedir:/usr/local/share/locale
+dirmngr-socket:/run/user/1000/gnupg/S.dirmngr
+agent-ssh-socket:/run/user/1000/gnupg/S.gpg-agent.ssh
+agent-socket:/run/user/1000/gnupg/S.gpg-agent
+homedir:/home/roman/.gnupg
+'''
+    expected = b'/run/user/1000/gnupg/S.gpg-agent'
+    assert keyring.get_agent_sock_path(sp=sp) == expected

libagent/gpg/tests/test_protocol.py

@@ -0,0 +1,107 @@
+import ecdsa
+import ed25519
+import pytest
+
+from .. import protocol
+from ... import formats
+
+
+def test_packet():
+    assert protocol.packet(1, b'') == b'\x84\x00'
+    assert protocol.packet(2, b'A') == b'\x88\x01A'
+    blob = b'B' * 0xAB
+    assert protocol.packet(3, blob) == b'\x8c\xAB' + blob
+    blob = b'C' * 0x1234
+    assert protocol.packet(3, blob) == b'\x8d\x12\x34' + blob
+    blob = b'D' * 0x12345678
+    assert protocol.packet(4, blob) == b'\x92\x12\x34\x56\x78' + blob
+
+
+def test_subpackets():
+    assert protocol.subpacket(1, b'') == b'\x01'
+    assert protocol.subpacket(2, '>H', 0x0304) == b'\x02\x03\x04'
+    assert protocol.subpacket_long(9, 0x12345678) == b'\x09\x12\x34\x56\x78'
+    assert protocol.subpacket_time(0x12345678) == b'\x02\x12\x34\x56\x78'
+    assert protocol.subpacket_byte(0xAB, 0xCD) == b'\xAB\xCD'
+    assert protocol.subpackets() == b'\x00\x00'
+    assert protocol.subpackets(b'ABC', b'12345') == b'\x00\x0A\x03ABC\x0512345'
+
+
+def test_mpi():
+    assert protocol.mpi(0x123) == b'\x00\x09\x01\x23'
+
+
+def test_armor():
+    data = bytearray(range(256))
+    assert protocol.armor(data, 'TEST') == '''-----BEGIN PGP TEST-----
+Version: GnuPG v2
+
+AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4v
+MDEyMzQ1Njc4OTo7PD0+P0BBQkNERUZHSElKS0xNTk9QUVJTVFVWV1hZWltcXV5f
+YGFiY2RlZmdoaWprbG1ub3BxcnN0dXZ3eHl6e3x9fn+AgYKDhIWGh4iJiouMjY6P
+kJGSk5SVlpeYmZqbnJ2en6ChoqOkpaanqKmqq6ytrq+wsbKztLW2t7i5uru8vb6/
+wMHCw8TFxsfIycrLzM3Oz9DR0tPU1dbX2Nna29zd3t/g4eLj5OXm5+jp6uvs7e7v
+8PHy8/T19vf4+fr7/P3+/w==
+=W700
+-----END PGP TEST-----
+'''
+
+
+def test_make_signature():
+    def signer_func(digest):
+        assert digest == (b'\xd0\xe5]|\x8bP\xe6\x91\xb3\xe8+\xf4A\xf0`(\xb1'
+                          b'\xc7\xf4;\x86\x97s\xdb\x9a\xda\xee< \xcb\x9e\x00')
+        return (7, 8)
+
+    sig = protocol.make_signature(
+        signer_func=signer_func,
+        data_to_sign=b'Hello World!',
+        public_algo=22,
+        hashed_subpackets=[protocol.subpacket_time(1)],
+        unhashed_subpackets=[],
+        sig_type=25)
+    assert sig == (b'\x04\x19\x16\x08\x00\x06\x05\x02'
+                   b'\x00\x00\x00\x01\x00\x00\xd0\xe5\x00\x03\x07\x00\x04\x08')
+
+
+def test_nist256p1():
+    sk = ecdsa.SigningKey.from_secret_exponent(secexp=1, curve=ecdsa.NIST256p)
+    vk = sk.get_verifying_key()
+    pk = protocol.PublicKey(curve_name=formats.CURVE_NIST256,
+                            created=42, verifying_key=vk)
+    assert repr(pk) == 'GPG public key nist256p1/F82361D9'
+    assert pk.keygrip() == b'\x95\x85.\x91\x7f\xe2\xc3\x91R\xba\x99\x81\x92\xb5y\x1d\xb1\\\xdc\xf0'
+
+
+def test_nist256p1_ecdh():
+    sk = ecdsa.SigningKey.from_secret_exponent(secexp=1, curve=ecdsa.NIST256p)
+    vk = sk.get_verifying_key()
+    pk = protocol.PublicKey(curve_name=formats.CURVE_NIST256,
+                            created=42, verifying_key=vk, ecdh=True)
+    assert repr(pk) == 'GPG public key nist256p1/5811DF46'
+    assert pk.keygrip() == b'\x95\x85.\x91\x7f\xe2\xc3\x91R\xba\x99\x81\x92\xb5y\x1d\xb1\\\xdc\xf0'
+
+
+def test_ed25519():
+    sk = ed25519.SigningKey(b'\x00' * 32)
+    vk = sk.get_verifying_key()
+    pk = protocol.PublicKey(curve_name=formats.CURVE_ED25519,
+                            created=42, verifying_key=vk)
+    assert repr(pk) == 'GPG public key ed25519/36B40FE6'
+    assert pk.keygrip() == b'\xbf\x01\x90l\x17\xb64\xa3-\xf4\xc0gr\x99\x18<\xddBQ?'
+
+
+def test_curve25519():
+    sk = ed25519.SigningKey(b'\x00' * 32)
+    vk = sk.get_verifying_key()
+    pk = protocol.PublicKey(curve_name=formats.ECDH_CURVE25519,
+                            created=42, verifying_key=vk)
+    assert repr(pk) == 'GPG public key curve25519/69460384'
+    assert pk.keygrip() == b'x\xd6\x86\xe4\xa6\xfc;\x0fY\xe1}Lw\xc4\x9ed\xf1Q\x8a\x00'
+
+
+def test_get_curve_name_by_oid():
+    for name, info in protocol.SUPPORTED_CURVES.items():
+        assert protocol.get_curve_name_by_oid(info['oid']) == name
+    with pytest.raises(KeyError):
+        protocol.get_curve_name_by_oid('BAD_OID')