distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""
2AES-256-CBC engine for I2P.
3
4Ported from:
5 net.i2p.crypto.CryptixAESEngine (CBC encrypt/decrypt)
6 net.i2p.crypto.AESEngine (base class)
7
8Wraps Python's ``cryptography`` library using AES-CBC with no padding.
9All data lengths must be multiples of 16 bytes.
10"""
11
12from __future__ import annotations
13
14from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
15
16
17# AES block size in bytes
18_BLOCK_SIZE = 16
19
20
21class AESEngine:
22 """AES-256-CBC engine with no automatic padding.
23
24 This mirrors I2P's ``CryptixAESEngine``: CBC mode, 16-byte IV,
25 data length must be a multiple of 16. Single-block ECB helpers
26 (``encrypt_block`` / ``decrypt_block``) are also provided for
27 callers that implement their own chaining.
28
29 All keys must be 16, 24, or 32 bytes (128/192/256-bit AES).
30 The I2P network uses 32-byte (256-bit) session keys.
31 """
32
33 # ------------------------------------------------------------------
34 # CBC bulk operations
35 # ------------------------------------------------------------------
36
37 @staticmethod
38 def encrypt(payload: bytes, key: bytes, iv: bytes) -> bytes:
39 """Encrypt *payload* with AES-CBC using *key* and *iv*.
40
41 Parameters
42 ----------
43 payload:
44 Plaintext whose length **must** be a positive multiple of 16.
45 key:
46 AES key (16, 24, or 32 bytes).
47 iv:
48 Initialisation vector, exactly 16 bytes.
49
50 Returns
51 -------
52 bytes
53 Ciphertext of the same length as *payload*.
54
55 Raises
56 ------
57 ValueError
58 If *payload* length is not a positive multiple of 16, or
59 *iv* is not 16 bytes.
60 """
61 AESEngine._validate(payload, key, iv)
62 cipher = Cipher(algorithms.AES(key), modes.CBC(iv))
63 encryptor = cipher.encryptor()
64 return encryptor.update(payload) + encryptor.finalize()
65
66 @staticmethod
67 def decrypt(payload: bytes, key: bytes, iv: bytes) -> bytes:
68 """Decrypt *payload* with AES-CBC using *key* and *iv*.
69
70 Parameters
71 ----------
72 payload:
73 Ciphertext whose length **must** be a positive multiple of 16.
74 key:
75 AES key (16, 24, or 32 bytes).
76 iv:
77 Initialisation vector, exactly 16 bytes.
78
79 Returns
80 -------
81 bytes
82 Plaintext of the same length as *payload*.
83
84 Raises
85 ------
86 ValueError
87 If *payload* length is not a positive multiple of 16, or
88 *iv* is not 16 bytes.
89 """
90 AESEngine._validate(payload, key, iv)
91 cipher = Cipher(algorithms.AES(key), modes.CBC(iv))
92 decryptor = cipher.decryptor()
93 return decryptor.update(payload) + decryptor.finalize()
94
95 # ------------------------------------------------------------------
96 # Single-block (ECB) operations — 16 bytes exactly
97 # ------------------------------------------------------------------
98
99 @staticmethod
100 def encrypt_block(block: bytes, key: bytes) -> bytes:
101 """Encrypt a single 16-byte block (ECB mode, no IV).
102
103 Parameters
104 ----------
105 block:
106 Exactly 16 bytes of plaintext.
107 key:
108 AES key (16, 24, or 32 bytes).
109
110 Returns
111 -------
112 bytes
113 16 bytes of ciphertext.
114 """
115 if len(block) != _BLOCK_SIZE:
116 raise ValueError(
117 f"Block must be exactly {_BLOCK_SIZE} bytes, got {len(block)}"
118 )
119 cipher = Cipher(algorithms.AES(key), modes.ECB())
120 encryptor = cipher.encryptor()
121 return encryptor.update(block) + encryptor.finalize()
122
123 @staticmethod
124 def decrypt_block(block: bytes, key: bytes) -> bytes:
125 """Decrypt a single 16-byte block (ECB mode, no IV).
126
127 Parameters
128 ----------
129 block:
130 Exactly 16 bytes of ciphertext.
131 key:
132 AES key (16, 24, or 32 bytes).
133
134 Returns
135 -------
136 bytes
137 16 bytes of plaintext.
138 """
139 if len(block) != _BLOCK_SIZE:
140 raise ValueError(
141 f"Block must be exactly {_BLOCK_SIZE} bytes, got {len(block)}"
142 )
143 cipher = Cipher(algorithms.AES(key), modes.ECB())
144 decryptor = cipher.decryptor()
145 return decryptor.update(block) + decryptor.finalize()
146
147 # ------------------------------------------------------------------
148 # Internal helpers
149 # ------------------------------------------------------------------
150
151 @staticmethod
152 def _validate(payload: bytes, key: bytes, iv: bytes) -> None:
153 """Common pre-condition checks for CBC operations."""
154 if len(iv) != _BLOCK_SIZE:
155 raise ValueError(
156 f"IV must be exactly {_BLOCK_SIZE} bytes, got {len(iv)}"
157 )
158 if len(payload) == 0:
159 raise ValueError("Payload must not be empty")
160 if len(payload) % _BLOCK_SIZE != 0:
161 raise ValueError(
162 f"Payload length must be a multiple of {_BLOCK_SIZE}, "
163 f"got {len(payload)}"
164 )