distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for AES padding and SessionTagManager."""
2
3import os
4from i2p_crypto.garlic_crypto import (
5 apply_aes_padding, verify_aes_padding, SessionTagManager,
6)
7
8
9def test_aes_padding_roundtrip():
10 payload = b"Hello, I2P garlic!"
11 padded = apply_aes_padding(payload)
12 assert len(padded) % 16 == 0
13 recovered = verify_aes_padding(padded)
14 assert recovered == payload
15
16
17def test_aes_padding_empty():
18 padded = apply_aes_padding(b"")
19 assert len(padded) % 16 == 0
20 assert verify_aes_padding(padded) == b""
21
22
23def test_aes_padding_checksum_mismatch():
24 padded = bytearray(apply_aes_padding(b"data"))
25 # Corrupt last byte (part of checksum)
26 padded[-1] ^= 0xFF
27 assert verify_aes_padding(bytes(padded)) is None
28
29
30def test_aes_padding_large_payload():
31 payload = os.urandom(1000)
32 padded = apply_aes_padding(payload)
33 assert len(padded) % 16 == 0
34 assert verify_aes_padding(padded) == payload
35
36
37def test_tag_bundle_generation():
38 stm = SessionTagManager()
39 key = os.urandom(32)
40 tags = stm.generate_tag_bundle(key, count=5)
41 assert len(tags) == 5
42 assert stm.remaining_tags(key) == 5
43
44
45def test_tag_consumption():
46 stm = SessionTagManager()
47 key = os.urandom(32)
48 stm.generate_tag_bundle(key, count=3)
49 tag1 = stm.consume_tag(key)
50 assert tag1 is not None
51 assert len(tag1) == 32
52 assert stm.remaining_tags(key) == 2
53
54
55def test_tag_consume_empty():
56 stm = SessionTagManager()
57 key = os.urandom(32)
58 assert stm.consume_tag(key) is None
59
60
61def test_tag_receive_and_consume():
62 stm = SessionTagManager()
63 key = os.urandom(32)
64 tags = [os.urandom(32) for _ in range(3)]
65 stm.receive_tags(key, tags)
66 assert stm.remaining_tags(key) == 3
67 consumed = stm.consume_tag(key)
68 assert consumed == tags[0]
69
70
71def test_tag_replenishment_threshold():
72 stm = SessionTagManager()
73 key = os.urandom(32)
74 assert stm.should_replenish(key) is True # 0 tags
75 stm.generate_tag_bundle(key, count=15)
76 assert stm.should_replenish(key) is False # 15 tags > 10
77
78
79def test_max_tags_limit():
80 stm = SessionTagManager(max_tags_per_session=10)
81 key = os.urandom(32)
82 stm.generate_tag_bundle(key, count=15)
83 assert stm.remaining_tags(key) == 10 # capped at max