distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for SSU2 transport protocol types."""
2
3import os
4import struct
5
6import pytest
7
8
9class TestSSU2Header:
10 def test_construct(self):
11 from i2p_transport.ssu2 import SSU2Header
12 h = SSU2Header(dest_conn_id=0xDEADBEEF, pkt_num=42,
13 header_type=0, version=2)
14 assert h.dest_conn_id == 0xDEADBEEF
15 assert h.pkt_num == 42
16
17 def test_short_header_roundtrip(self):
18 from i2p_transport.ssu2 import SSU2Header
19 h = SSU2Header(dest_conn_id=12345, pkt_num=100,
20 header_type=0, version=2)
21 data = h.to_bytes()
22 h2 = SSU2Header.from_bytes(data)
23 assert h2.dest_conn_id == 12345
24 assert h2.pkt_num == 100
25
26 def test_header_size(self):
27 from i2p_transport.ssu2 import SSU2Header
28 h = SSU2Header(dest_conn_id=0, pkt_num=0, header_type=0, version=2)
29 data = h.to_bytes()
30 assert len(data) == 16 # dest_id(8) + pkt_num(4) + type(1) + version(1) + net_id(2)
31
32
33class TestSSU2Block:
34 def test_construct(self):
35 from i2p_transport.ssu2 import SSU2Block, BlockType
36 block = SSU2Block(block_type=BlockType.DATA, payload=b"test")
37 assert block.block_type == BlockType.DATA
38 assert block.payload == b"test"
39
40 def test_roundtrip(self):
41 from i2p_transport.ssu2 import SSU2Block, BlockType
42 block = SSU2Block(block_type=BlockType.DATA, payload=os.urandom(64))
43 data = block.to_bytes()
44 block2 = SSU2Block.from_bytes(data)
45 assert block2.block_type == block.block_type
46 assert block2.payload == block.payload
47
48 def test_ack_block(self):
49 from i2p_transport.ssu2 import SSU2Block, BlockType
50 # ACK: ack_through(4) + num_ranges(1) + ranges
51 ack_payload = struct.pack("!IB", 100, 0)
52 block = SSU2Block(BlockType.ACK, ack_payload)
53 data = block.to_bytes()
54 block2 = SSU2Block.from_bytes(data)
55 assert block2.block_type == BlockType.ACK
56
57 def test_address_block(self):
58 from i2p_transport.ssu2 import SSU2Block, BlockType
59 block = SSU2Block(BlockType.ADDRESS, b"\x04\x01\x02\x03\x04\x30\x39")
60 data = block.to_bytes()
61 block2 = SSU2Block.from_bytes(data)
62 assert block2.block_type == BlockType.ADDRESS
63
64 def test_termination_block(self):
65 from i2p_transport.ssu2 import SSU2Block, BlockType
66 block = SSU2Block(BlockType.TERMINATION, struct.pack("!BQ", 0, 50))
67 data = block.to_bytes()
68 block2 = SSU2Block.from_bytes(data)
69 assert block2.block_type == BlockType.TERMINATION
70
71 def test_padding_block(self):
72 from i2p_transport.ssu2 import SSU2Block, BlockType
73 block = SSU2Block(BlockType.PADDING, os.urandom(32))
74 data = block.to_bytes()
75 block2 = SSU2Block.from_bytes(data)
76 assert block2.block_type == BlockType.PADDING
77
78 def test_multiple_blocks(self):
79 import io
80 from i2p_transport.ssu2 import SSU2Block, BlockType
81 b1 = SSU2Block(BlockType.DATA, b"first")
82 b2 = SSU2Block(BlockType.DATA, b"second")
83 buf = b1.to_bytes() + b2.to_bytes()
84 stream = io.BytesIO(buf)
85 r1 = SSU2Block.from_stream(stream)
86 r2 = SSU2Block.from_stream(stream)
87 assert r1.payload == b"first"
88 assert r2.payload == b"second"
89
90
91class TestSSU2SessionState:
92 def test_initial_state(self):
93 from i2p_transport.ssu2 import SSU2SessionState, SSU2State
94 state = SSU2SessionState()
95 assert state.state == SSU2State.UNKNOWN
96
97 def test_token_request_flow(self):
98 from i2p_transport.ssu2 import SSU2SessionState, SSU2State
99 state = SSU2SessionState()
100 state.send_token_request()
101 assert state.state == SSU2State.TOKEN_REQUEST_SENT
102
103 def test_session_request_flow(self):
104 from i2p_transport.ssu2 import SSU2SessionState, SSU2State
105 state = SSU2SessionState()
106 state.send_session_request()
107 assert state.state == SSU2State.SESSION_REQUEST_SENT
108
109 def test_established(self):
110 from i2p_transport.ssu2 import SSU2SessionState, SSU2State
111 state = SSU2SessionState()
112 state.send_session_request()
113 state.receive_session_created()
114 assert state.state == SSU2State.SESSION_CONFIRMED
115 state.confirm()
116 assert state.state == SSU2State.ESTABLISHED
117
118 def test_terminate(self):
119 from i2p_transport.ssu2 import SSU2SessionState, SSU2State
120 state = SSU2SessionState()
121 state.send_session_request()
122 state.receive_session_created()
123 state.confirm()
124 state.terminate()
125 assert state.state == SSU2State.TERMINATED
126
127 def test_peer_test(self):
128 from i2p_transport.ssu2 import SSU2SessionState
129 state = SSU2SessionState()
130 state.send_session_request()
131 state.receive_session_created()
132 state.confirm()
133 state.start_peer_test()
134 assert state.peer_test_active
135
136 def test_path_validation(self):
137 from i2p_transport.ssu2 import SSU2SessionState
138 state = SSU2SessionState()
139 state.send_session_request()
140 state.receive_session_created()
141 state.confirm()
142 nonce = state.start_path_validation()
143 assert nonce is not None
144 assert state.path_challenge_pending