distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for i2p_data.certificate — Certificate type system."""
2import io
3import struct
4
5import pytest
6
7from i2p_data.certificate import (
8 Certificate,
9 CertificateType,
10 KeyCertificate,
11)
12
13
14class TestCertificateType:
15 """CertificateType enum values match the I2P spec."""
16
17 def test_null_is_zero(self):
18 assert CertificateType.NULL == 0
19
20 def test_hashcash_is_one(self):
21 assert CertificateType.HASHCASH == 1
22
23 def test_hidden_is_two(self):
24 assert CertificateType.HIDDEN == 2
25
26 def test_signed_is_three(self):
27 assert CertificateType.SIGNED == 3
28
29 def test_multiple_is_four(self):
30 assert CertificateType.MULTIPLE == 4
31
32 def test_key_is_five(self):
33 assert CertificateType.KEY == 5
34
35
36class TestCertificateNull:
37 """NULL certificate serialization."""
38
39 def test_null_to_bytes(self):
40 assert Certificate.NULL.to_bytes() == b"\x00\x00\x00"
41
42 def test_null_from_bytes_roundtrip(self):
43 cert = Certificate.from_bytes(b"\x00\x00\x00")
44 assert cert.cert_type == CertificateType.NULL
45 assert cert.payload == b""
46 assert cert.to_bytes() == b"\x00\x00\x00"
47
48 def test_null_len(self):
49 assert len(Certificate.NULL) == 3
50
51
52class TestCertificateWithPayload:
53 """Certificate with a payload roundtrips correctly."""
54
55 def test_hashcash_roundtrip(self):
56 payload = b"hashcash-stamp-data"
57 cert = Certificate(CertificateType.HASHCASH, payload)
58 serialized = cert.to_bytes()
59 restored = Certificate.from_bytes(serialized)
60 assert restored.cert_type == CertificateType.HASHCASH
61 assert restored.payload == payload
62
63 def test_serialized_length(self):
64 payload = b"abcdef"
65 cert = Certificate(CertificateType.HASHCASH, payload)
66 assert len(cert) == 3 + len(payload)
67 assert len(cert.to_bytes()) == len(cert)
68
69
70class TestCertificateFromStream:
71 """from_stream reads correctly and positions the stream after the cert."""
72
73 def test_stream_position_after_read(self):
74 cert_bytes = b"\x00\x00\x00"
75 trailing = b"\xDE\xAD"
76 stream = io.BytesIO(cert_bytes + trailing)
77 cert = Certificate.from_stream(stream)
78 assert cert.cert_type == CertificateType.NULL
79 assert stream.read() == trailing
80
81 def test_stream_with_payload(self):
82 payload = b"\x01\x02\x03\x04"
83 data = struct.pack("!BH", CertificateType.HASHCASH, len(payload)) + payload
84 trailing = b"\xFF"
85 stream = io.BytesIO(data + trailing)
86 cert = Certificate.from_stream(stream)
87 assert cert.cert_type == CertificateType.HASHCASH
88 assert cert.payload == payload
89 assert stream.read() == trailing
90
91 def test_stream_short_header_raises(self):
92 with pytest.raises(ValueError, match="3 bytes"):
93 Certificate.from_stream(io.BytesIO(b"\x00\x00"))
94
95 def test_stream_short_payload_raises(self):
96 # Header says 5 bytes payload but only 2 available
97 data = struct.pack("!BH", CertificateType.HASHCASH, 5) + b"\x01\x02"
98 with pytest.raises(ValueError, match="payload bytes"):
99 Certificate.from_stream(io.BytesIO(data))
100
101
102class TestKeyCertificate:
103 """KeyCertificate structured access."""
104
105 def _make_key_cert_payload(self, sig_code: int, enc_code: int,
106 extra: bytes = b"") -> bytes:
107 return struct.pack("!HH", sig_code, enc_code) + extra
108
109 def test_sig_and_enc_type_codes(self):
110 # EdDSA_SHA512_Ed25519 = code 7, ECIES_X25519 = code 4
111 payload = self._make_key_cert_payload(7, 4)
112 kc = KeyCertificate(payload)
113 assert kc.get_sig_type_code() == 7
114 assert kc.get_enc_type_code() == 4
115
116 def test_get_sig_type_returns_enum(self):
117 payload = self._make_key_cert_payload(7, 4)
118 kc = KeyCertificate(payload)
119 from i2p_crypto.dsa import SigType
120 assert kc.get_sig_type() == SigType.EdDSA_SHA512_Ed25519
121
122 def test_extra_key_data(self):
123 extra = b"\xAA\xBB\xCC"
124 payload = self._make_key_cert_payload(7, 4, extra)
125 kc = KeyCertificate(payload)
126 assert kc.get_extra_key_data() == extra
127
128 def test_no_extra_key_data(self):
129 payload = self._make_key_cert_payload(1, 0)
130 kc = KeyCertificate(payload)
131 assert kc.get_extra_key_data() == b""
132
133 def test_payload_too_short_raises(self):
134 with pytest.raises(ValueError, match=">= 4 bytes"):
135 KeyCertificate(b"\x00\x01")
136
137 def test_payload_exactly_four_bytes(self):
138 payload = self._make_key_cert_payload(0, 0)
139 kc = KeyCertificate(payload)
140 assert kc.get_sig_type_code() == 0
141 assert kc.get_enc_type_code() == 0
142
143 def test_cert_type_is_key(self):
144 payload = self._make_key_cert_payload(7, 4)
145 kc = KeyCertificate(payload)
146 assert kc.cert_type == CertificateType.KEY
147
148
149class TestKeyCertificateAutoDetection:
150 """Certificate.from_bytes auto-creates KeyCertificate for type KEY."""
151
152 def test_from_bytes_returns_key_certificate(self):
153 payload = struct.pack("!HH", 7, 4)
154 data = struct.pack("!BH", CertificateType.KEY, len(payload)) + payload
155 cert = Certificate.from_bytes(data)
156 assert isinstance(cert, KeyCertificate)
157 assert cert.get_sig_type_code() == 7
158
159 def test_from_bytes_non_key_returns_certificate(self):
160 data = struct.pack("!BH", CertificateType.HASHCASH, 0)
161 cert = Certificate.from_bytes(data)
162 assert type(cert) is Certificate
163 assert not isinstance(cert, KeyCertificate)
164
165
166class TestCertificateEqualityAndHash:
167 """Equality and hashing."""
168
169 def test_equal_certificates(self):
170 a = Certificate(CertificateType.HASHCASH, b"\x01\x02")
171 b = Certificate(CertificateType.HASHCASH, b"\x01\x02")
172 assert a == b
173
174 def test_unequal_type(self):
175 a = Certificate(CertificateType.HASHCASH, b"")
176 b = Certificate(CertificateType.HIDDEN, b"")
177 assert a != b
178
179 def test_unequal_payload(self):
180 a = Certificate(CertificateType.HASHCASH, b"\x01")
181 b = Certificate(CertificateType.HASHCASH, b"\x02")
182 assert a != b
183
184 def test_hash_equal_for_equal_certs(self):
185 a = Certificate(CertificateType.NULL)
186 b = Certificate(CertificateType.NULL)
187 assert hash(a) == hash(b)
188
189 def test_usable_in_set(self):
190 a = Certificate(CertificateType.NULL)
191 b = Certificate(CertificateType.NULL)
192 c = Certificate(CertificateType.HASHCASH, b"\x01")
193 s = {a, b, c}
194 assert len(s) == 2
195
196 def test_not_equal_to_non_certificate(self):
197 assert Certificate(CertificateType.NULL) != "not a cert"
198
199
200class TestKeyCertificateRoundtrip:
201 """KeyCertificate serialization roundtrip."""
202
203 def test_roundtrip_with_extra_data(self):
204 extra = b"\x01\x02\x03\x04\x05"
205 payload = struct.pack("!HH", 7, 4) + extra
206 kc = KeyCertificate(payload)
207 serialized = kc.to_bytes()
208 restored = Certificate.from_bytes(serialized)
209 assert isinstance(restored, KeyCertificate)
210 assert restored.get_sig_type_code() == 7
211 assert restored.get_enc_type_code() == 4
212 assert restored.get_extra_key_data() == extra
213
214 def test_len_includes_header_and_payload(self):
215 payload = struct.pack("!HH", 7, 4) + b"\xAA"
216 kc = KeyCertificate(payload)
217 assert len(kc) == 3 + len(payload)