distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for KeysAndCert and Destination."""
2
3import base64
4import hashlib
5import os
6
7
8class TestKeysAndCert:
9 def _make_null_kac(self):
10 """Create a KeysAndCert with NULL certificate and ElGamal/DSA keys."""
11 from i2p_data.key_types import PublicKey, SigningPublicKey, EncType
12 from i2p_data.certificate import Certificate
13 from i2p_crypto.dsa import SigType
14
15 pub = PublicKey(os.urandom(256), EncType.ELGAMAL)
16 sig = SigningPublicKey(os.urandom(128), SigType.DSA_SHA1)
17 cert = Certificate.NULL
18 return pub, sig, cert
19
20 def _make_key_cert_kac(self):
21 """Create a KeysAndCert with KEY certificate and EdDSA/X25519."""
22 import struct
23 from i2p_data.key_types import PublicKey, SigningPublicKey, EncType
24 from i2p_data.certificate import KeyCertificate
25 from i2p_crypto.dsa import SigType
26
27 pub = PublicKey(os.urandom(32), EncType.ECIES_X25519)
28 sig = SigningPublicKey(os.urandom(32), SigType.EdDSA_SHA512_Ed25519)
29 # KEY cert: SigType(7) + EncType(4) = 4 bytes payload
30 payload = struct.pack("!HH", 7, 4)
31 cert = KeyCertificate(payload)
32 return pub, sig, cert
33
34 def test_construct_null_cert(self):
35 from i2p_data.keys_and_cert import KeysAndCert
36 pub, sig, cert = self._make_null_kac()
37 kac = KeysAndCert(pub, sig, cert)
38 assert kac.public_key is pub
39 assert kac.signing_public_key is sig
40 assert kac.certificate is cert
41
42 def test_serialize_null_cert_size(self):
43 from i2p_data.keys_and_cert import KeysAndCert
44 pub, sig, cert = self._make_null_kac()
45 kac = KeysAndCert(pub, sig, cert)
46 data = kac.to_bytes()
47 assert len(data) == 256 + 128 + 3 # 387
48
49 def test_roundtrip_null_cert(self):
50 from i2p_data.keys_and_cert import KeysAndCert
51 pub, sig, cert = self._make_null_kac()
52 kac = KeysAndCert(pub, sig, cert)
53 data = kac.to_bytes()
54 kac2 = KeysAndCert.from_bytes(data)
55 assert kac2.public_key == pub
56 assert kac2.signing_public_key == sig
57 assert kac2.to_bytes() == data
58
59 def test_roundtrip_key_cert(self):
60 from i2p_data.keys_and_cert import KeysAndCert
61 pub, sig, cert = self._make_key_cert_kac()
62 kac = KeysAndCert(pub, sig, cert)
63 data = kac.to_bytes()
64 # With KEY cert: 256 + 128 + 3(header) + 4(payload) = 391
65 assert len(data) == 391
66 kac2 = KeysAndCert.from_bytes(data)
67 assert kac2.public_key == pub
68 assert kac2.signing_public_key == sig
69
70 def test_hash_is_sha256(self):
71 from i2p_data.keys_and_cert import KeysAndCert
72 pub, sig, cert = self._make_null_kac()
73 kac = KeysAndCert(pub, sig, cert)
74 expected = hashlib.sha256(kac.to_bytes()).digest()
75 assert kac.hash() == expected
76 assert len(kac.hash()) == 32
77
78 def test_hash_cached(self):
79 from i2p_data.keys_and_cert import KeysAndCert
80 pub, sig, cert = self._make_null_kac()
81 kac = KeysAndCert(pub, sig, cert)
82 h1 = kac.hash()
83 h2 = kac.hash()
84 assert h1 is h2 # Same object, cached
85
86 def test_equality(self):
87 from i2p_data.keys_and_cert import KeysAndCert
88 from i2p_data.key_types import PublicKey, SigningPublicKey, EncType
89 from i2p_data.certificate import Certificate
90 from i2p_crypto.dsa import SigType
91
92 data_pub = os.urandom(256)
93 data_sig = os.urandom(128)
94 kac1 = KeysAndCert(
95 PublicKey(data_pub, EncType.ELGAMAL),
96 SigningPublicKey(data_sig, SigType.DSA_SHA1),
97 Certificate.NULL
98 )
99 kac2 = KeysAndCert(
100 PublicKey(data_pub, EncType.ELGAMAL),
101 SigningPublicKey(data_sig, SigType.DSA_SHA1),
102 Certificate.NULL
103 )
104 assert kac1 == kac2
105 assert hash(kac1) == hash(kac2)
106
107 def test_inequality(self):
108 from i2p_data.keys_and_cert import KeysAndCert
109 pub1, sig1, cert1 = self._make_null_kac()
110 pub2, sig2, cert2 = self._make_null_kac()
111 kac1 = KeysAndCert(pub1, sig1, cert1)
112 kac2 = KeysAndCert(pub2, sig2, cert2)
113 assert kac1 != kac2
114
115 def test_from_bytes_too_short(self):
116 from i2p_data.keys_and_cert import KeysAndCert
117 import pytest
118 with pytest.raises(ValueError):
119 KeysAndCert.from_bytes(b"\x00" * 100)
120
121 def test_key_cert_right_aligned(self):
122 """With X25519 (32 bytes), the key should be right-aligned in 256-byte area."""
123 from i2p_data.keys_and_cert import KeysAndCert
124 pub, sig, cert = self._make_key_cert_kac()
125 kac = KeysAndCert(pub, sig, cert)
126 data = kac.to_bytes()
127 # Public key area: 224 zero bytes + 32 bytes of key
128 assert data[:224] == b"\x00" * 224
129 assert data[224:256] == pub.to_bytes()
130
131
132class TestDestination:
133 def _make_destination(self):
134 from i2p_data.destination import Destination
135 from i2p_data.key_types import PublicKey, SigningPublicKey, EncType
136 from i2p_data.certificate import Certificate
137 from i2p_crypto.dsa import SigType
138
139 pub = PublicKey(os.urandom(256), EncType.ELGAMAL)
140 sig = SigningPublicKey(os.urandom(128), SigType.DSA_SHA1)
141 cert = Certificate.NULL
142 return Destination(pub, sig, cert)
143
144 def test_to_base64_roundtrip(self):
145 from i2p_data.destination import Destination
146 dest = self._make_destination()
147 b64 = dest.to_base64()
148 assert isinstance(b64, str)
149 dest2 = Destination.from_base64(b64)
150 assert dest2.to_bytes() == dest.to_bytes()
151
152 def test_base64_no_padding(self):
153 dest = self._make_destination()
154 b64 = dest.to_base64()
155 assert "=" not in b64
156
157 def test_to_base32_format(self):
158 dest = self._make_destination()
159 b32 = dest.to_base32()
160 assert b32.endswith(".b32.i2p")
161 # SHA-256 = 32 bytes = 52 base32 chars
162 addr_part = b32.replace(".b32.i2p", "")
163 assert len(addr_part) == 52
164
165 def test_base32_is_sha256(self):
166 dest = self._make_destination()
167 b32 = dest.to_base32()
168 addr_part = b32.replace(".b32.i2p", "")
169 expected_hash = hashlib.sha256(dest.to_bytes()).digest()
170 expected_b32 = base64.b32encode(expected_hash).rstrip(b"=").decode("ascii").lower()
171 assert addr_part == expected_b32
172
173 def test_base32_cached(self):
174 dest = self._make_destination()
175 b32a = dest.to_base32()
176 b32b = dest.to_base32()
177 assert b32a is b32b # Cached
178
179 def test_from_bytes_roundtrip(self):
180 from i2p_data.destination import Destination
181 dest = self._make_destination()
182 data = dest.to_bytes()
183 dest2 = Destination.from_bytes(data)
184 assert dest2.to_bytes() == data
185 assert dest2.to_base32() == dest.to_base32()
186
187 def test_different_destinations_different_b32(self):
188 dest1 = self._make_destination()
189 dest2 = self._make_destination()
190 assert dest1.to_base32() != dest2.to_base32()
191
192 def test_inherits_keys_and_cert(self):
193 from i2p_data.keys_and_cert import KeysAndCert
194 dest = self._make_destination()
195 assert isinstance(dest, KeysAndCert)
196 assert len(dest.hash()) == 32