From ad3922d385e6929838be69833fc44724143a5390 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Thu, 30 Jul 2026 14:05:05 +0000 Subject: [PATCH] feat: Implement pure python AES-IGE mode Co-authored-by: SWORDIntel <117012829+SWORDIntel@users.noreply.github.com> --- README.md | 16 +++- docs/IMPLEMENTATION.md | 3 +- selftest.py | 94 +++++++++---------- src/crypto_standalone/__init__.py | 2 +- src/crypto_standalone/asymmetric/ed25519.py | 24 ++--- .../asymmetric/nist_curves.py | 68 +++++++------- src/crypto_standalone/asymmetric/rsa.py | 18 ++-- src/crypto_standalone/asymmetric/x25519.py | 18 ++-- src/crypto_standalone/kdf/hkdf.py | 10 +- src/crypto_standalone/kdf/pbkdf2.py | 10 +- src/crypto_standalone/symmetric/__init__.py | 5 +- src/crypto_standalone/symmetric/aes.py | 58 +++++++----- src/crypto_standalone/symmetric/aes_ige.py | 92 ++++++++++++++++++ src/crypto_standalone/utils/memory.py | 18 ++-- tests/adversarial/test_attack_simulations.py | 46 ++++----- tests/adversarial/test_ecdsa_wycheproof.py | 4 +- tests/adversarial/test_fuzz_hypothesis.py | 4 +- tests/unit/test_aes_ige.py | 79 ++++++++++++++++ tests/unit/test_ave_maria_cipher.py | 2 +- 19 files changed, 384 insertions(+), 187 deletions(-) create mode 100644 src/crypto_standalone/symmetric/aes_ige.py create mode 100644 tests/unit/test_aes_ige.py diff --git a/README.md b/README.md index 5410cee..7d449de 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ ## Features -- **AES-256-GCM** (NIST SP 800-38D), **ChaCha20-Poly1305** (RFC 8439), AES-256-CBC/CTR +- **AES-256-GCM** (NIST SP 800-38D), **ChaCha20-Poly1305** (RFC 8439), AES-256-CBC/CTR, **AES-IGE** - **SHA-256/384/512** (FIPS 180-4), **SHA-3/SHAKE** (FIPS 202), HMAC, tagged hashing - **HKDF** (RFC 5869), **PBKDF2** (RFC 2898, 600k+ iterations) - **RSA** (OAEP, PSS, Baillie-PSW primality), **Ed25519** (RFC 8032), **X25519** (RFC 7748), **P-256/P-384** (ECDSA + ECDH) @@ -89,6 +89,20 @@ token = rng.urandom(32) +### AES-IGE Encryption + +```python +from crypto_standalone import aes_ige_encrypt, aes_ige_decrypt + +key = b"\x00" * 32 +iv = b"\x00" * 32 # 32-byte IV (C_0 || P_0) +plaintext = b"blockaligneddata" # Must be exactly a multiple of 16 bytes + +# AES-IGE provides confidentiality only. Must be paired with an authentication mechanism! +ciphertext = aes_ige_encrypt(plaintext, key, iv) +assert aes_ige_decrypt(ciphertext, key, iv) == plaintext +``` + ### Legacy Ciphers (Interoperability Only) ```python diff --git a/docs/IMPLEMENTATION.md b/docs/IMPLEMENTATION.md index 178dd46..a968adb 100644 --- a/docs/IMPLEMENTATION.md +++ b/docs/IMPLEMENTATION.md @@ -4,7 +4,8 @@ - **SHA-2**: Bit-level FIPS 180-4 (32-bit and 64-bit word versions) - **SHA-3/Keccak**: Keccak-f[1600] sponge permutation, 24 rounds -- **AES-256**: Rijndael with 14 rounds, GF(2^8) arithmetic +- **AES**: Rijndael supporting 128/192/256-bit keys, GF(2^8) arithmetic +- **AES-IGE**: Infinite Garble Extension mode with 32-byte IV - **GCM**: GHASH in GF(2^128), CTR mode encryption - **ChaCha20**: 20-round quarter-function, Poly1305 over GF(2^130-5) - **RSA**: Baillie-PSW primality, CRT-based decryption, small-prime sieve diff --git a/selftest.py b/selftest.py index 7e6143d..507540b 100644 --- a/selftest.py +++ b/selftest.py @@ -11,43 +11,43 @@ def test_hashes(): from crypto_standalone import sha256_hex, sha384_hex, sha512_hex, hmac_sha256, tagged_hash - + tests = [ ("SHA-256", sha256_hex(b"abc"), "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"), ("SHA-384", sha384_hex(b"abc"), "cb00753f45a35e8bb5a03d699ac65007272c32ab0eded1631a8b605a43ff5bed8086072ba1e7cc2358baeca134c825a7"), ("SHA-512", sha512_hex(b"abc"), "ddaf35a193617abacc417349ae20413112e6fa4e89a97ea20a9eeee64b55d39a2192992a274fc1a836ba3c23a3feebbd454d4423643ce80e2a9ac94fa54ca49f"), ] - + for name, got, exp in tests: if got != exp: print(f" {name}: FAIL") return False - + mac = hmac_sha256(b"key", b"message") if len(mac) != 32: print(" HMAC-SHA256: FAIL") return False - + tagged = tagged_hash("test", b"data") if len(tagged) != 32: print(" Tagged hash: FAIL") return False - + print(" Hashes: PASS") return True def test_sha3(): from crypto_standalone import sha3_256_hex, sha3_512_hex, shake_128 - + if sha3_256_hex(b"abc") != "3a985da74fe225b2045c172d6bd390bd855f086e3e9d525b46bfe24511431532": print(" SHA3-256: FAIL") return False - + if len(shake_128(b"test", 32)) != 32: print(" SHAKE-128: FAIL") return False - + print(" SHA-3: PASS") return True @@ -55,151 +55,151 @@ def test_sha3(): def test_aead(): from crypto_standalone import AESGCM from crypto_standalone import ChaCha20Poly1305 - + key = b"\x00" * 32 msg = b"secret message" aad = b"metadata" - + gcm = AESGCM(key) blob_gcm = gcm.encrypt_blob(msg, aad=aad) if gcm.decrypt_blob(blob_gcm, aad=aad) != msg: print(" AES-GCM: FAIL") return False - + try: gcm.decrypt_blob(blob_gcm, aad=b"wrong") print(" AES-GCM tamper detection: FAIL") return False except ValueError: pass - + cp = ChaCha20Poly1305(key) blob_cp = cp.encrypt_blob(msg, aad=aad) if cp.decrypt_blob(blob_cp, aad=aad) != msg: print(" ChaCha20-Poly1305: FAIL") return False - + try: cp.decrypt_blob(blob_cp, aad=b"wrong") print(" ChaCha20-Poly1305 tamper detection: FAIL") return False except ValueError: pass - + print(" AEAD: PASS") return True def test_kdf(): from crypto_standalone import hkdf_sha256, pbkdf2_sha256 - + okm = hkdf_sha256(b"salt", b"ikm", b"info", 32) if len(okm) != 32: print(" HKDF: FAIL") return False - + dk = pbkdf2_sha256(b"password", b"salt", 1000, 32) if len(dk) != 32: print(" PBKDF2: FAIL") return False - + print(" KDF: PASS") return True def test_rsa(): from crypto_standalone import generate_rsa_keypair - + kp = generate_rsa_keypair(2048) msg = b"test" - + ct = kp.public.encrypt(msg) if kp.private.decrypt(ct) != msg: print(" RSA PKCS#1 v1.5: FAIL") return False - + ct_oaep = kp.public.encrypt_oaep(msg) if kp.private.decrypt_oaep(ct_oaep) != msg: print(" RSA-OAEP: FAIL") return False - + sig = kp.private.sign(msg) if not kp.public.verify(msg, sig): print(" RSA sign/verify: FAIL") return False - + sig_pss = kp.private.sign_pss(msg) if not kp.public.verify_pss(msg, sig_pss): print(" RSA-PSS: FAIL") return False - + print(" RSA: PASS") return True def test_ed25519(): from crypto_standalone import ed25519_keygen, ed25519_sign, ed25519_verify - + sk, pk = ed25519_keygen() msg = b"test message" sig = ed25519_sign(msg, sk) - + if not ed25519_verify(msg, sig, pk): print(" Ed25519 verify: FAIL") return False - + if ed25519_verify(b"wrong", sig, pk): print(" Ed25519 tamper detection: FAIL") return False - + print(" Ed25519: PASS") return True def test_x25519(): from crypto_standalone import x25519_keygen, x25519 - + sk1, pk1 = x25519_keygen() sk2, pk2 = x25519_keygen() - + s1 = x25519(sk1, pk2) s2 = x25519(sk2, pk1) - + if s1 != s2: print(" X25519 agreement: FAIL") return False - + if len(s1) != 32: print(" X25519 length: FAIL") return False - + print(" X25519: PASS") return True def test_p_curves(): from crypto_standalone import p256_keygen, p384_keygen - + sk256 = p256_keygen() msg = b"test" sig256 = sk256.sign(msg) - + if not sk256.public_key.verify(msg, sig256): print(" P-256 ECDSA: FAIL") return False - + sk256_2 = p256_keygen() shared = sk256.ecdh(sk256_2.public_key) if len(shared) != 32: print(" P-256 ECDH: FAIL") return False - + sk384 = p384_keygen() sig384 = sk384.sign(msg) if not sk384.public_key.verify(msg, sig384): print(" P-384 ECDSA: FAIL") return False - + print(" P-curves: PASS") return True @@ -207,27 +207,27 @@ def test_p_curves(): def test_utilities(): from crypto_standalone import random_bytes, random_below from crypto_standalone import secure_zero, SecureBytes - + if len(random_bytes(32)) != 32: print(" random_bytes: FAIL") return False - + r = random_below(100) if not (0 <= r < 100): print(" random_below: FAIL") return False - + data = bytearray(b"secret") secure_zero(data) if data != bytearray(6): print(" secure_zero: FAIL") return False - + with SecureBytes(32) as sb: if len(sb) != 32: print(" SecureBytes: FAIL") return False - + print(" Utilities: PASS") return True @@ -237,7 +237,7 @@ def main(): print("Military-Grade Crypto Toolkit Self-Test v2.0") print("=" * 60) print() - + tests = [ ("Extended Hashes", test_hashes), ("SHA-3 Family", test_sha3), @@ -249,10 +249,10 @@ def main(): ("NIST P-curves", test_p_curves), ("Utilities", test_utilities), ] - + passed = 0 failed = 0 - + for name, test_fn in tests: print(f"[{name}]") try: @@ -264,7 +264,7 @@ def main(): print(f" EXCEPTION: {e}") failed += 1 print() - + print("=" * 60) print(f"Results: {passed}/{len(tests)} passed") if failed == 0: @@ -272,7 +272,7 @@ def main(): else: print(f"✗ {failed} FAILED") print("=" * 60) - + return 0 if failed == 0 else 1 diff --git a/src/crypto_standalone/__init__.py b/src/crypto_standalone/__init__.py index 4dc7ea8..4b261fd 100644 --- a/src/crypto_standalone/__init__.py +++ b/src/crypto_standalone/__init__.py @@ -2,7 +2,7 @@ __version__ = "2.0.0" -from .symmetric import AES256, AESGCM, ChaCha20Poly1305, chacha20_encrypt, TEA, RedPike, AveMariaCipher +from .symmetric import AES, AES256, AESIGE, aes_ige_encrypt, aes_ige_decrypt, AESGCM, ChaCha20Poly1305, chacha20_encrypt, TEA, RedPike, AveMariaCipher from .hashing import * from .asymmetric import * from .asymmetric import _encode_signature, _decode_signature diff --git a/src/crypto_standalone/asymmetric/ed25519.py b/src/crypto_standalone/asymmetric/ed25519.py index f2e71bb..00b269a 100644 --- a/src/crypto_standalone/asymmetric/ed25519.py +++ b/src/crypto_standalone/asymmetric/ed25519.py @@ -95,12 +95,12 @@ def ed25519_keygen(seed: bytes | None = None) -> tuple[bytes, bytes]: seed = random_bytes(32) if len(seed) != 32: raise ValueError("seed must be 32 bytes") - + h = sha512(seed) a = int.from_bytes(h[:32], "little") a &= (1 << 254) - 8 a |= (1 << 254) - + A = _edwards_scalarmult(_B, a) public_key = _point_compress(A) return seed, public_key @@ -113,22 +113,22 @@ def ed25519_sign(message: bytes, private_key: bytes) -> bytes: """ if len(private_key) != 32: raise ValueError("private key must be 32 bytes") - + h = sha512(private_key) a = int.from_bytes(h[:32], "little") a &= (1 << 254) - 8 a |= (1 << 254) - + A = _edwards_scalarmult(_B, a) public_key = _point_compress(A) - + r = _hint(h[32:] + message) R = _edwards_scalarmult(_B, r) R_bytes = _point_compress(R) - + k = _hint(R_bytes + public_key + message) s = (r + k * a) % _L - + return R_bytes + s.to_bytes(32, "little") @@ -141,21 +141,21 @@ def ed25519_verify(message: bytes, signature: bytes, public_key: bytes) -> bool: return False if len(public_key) != 32: return False - + try: R_bytes = signature[:32] s = int.from_bytes(signature[32:], "little") if s >= _L: return False - + R = _point_decompress(R_bytes) A = _point_decompress(public_key) - + k = _hint(R_bytes + public_key + message) - + lhs = _edwards_scalarmult(_B, s) rhs = _edwards_add(R, _edwards_scalarmult(A, k)) - + return lhs == rhs except Exception: return False diff --git a/src/crypto_standalone/asymmetric/nist_curves.py b/src/crypto_standalone/asymmetric/nist_curves.py index 3c53afe..40a80f7 100644 --- a/src/crypto_standalone/asymmetric/nist_curves.py +++ b/src/crypto_standalone/asymmetric/nist_curves.py @@ -65,16 +65,16 @@ def _point_add(curve: _Curve, P: tuple[int, int] | None, Q: tuple[int, int] | No return Q if Q is None: return P - + x1, y1 = P x2, y2 = Q - + if x1 == x2: if y1 == y2: return _point_double(curve, P) else: return None - + s = ((y2 - y1) * _modinv((x2 - x1) % curve.p, curve.p)) % curve.p x3 = (s * s - x1 - x2) % curve.p y3 = (s * (x1 - x3) - y1) % curve.p @@ -86,7 +86,7 @@ def _point_double(curve: _Curve, P: tuple[int, int]) -> tuple[int, int] | None: x, y = P if y == 0: return None - + s = ((3 * x * x + curve.a) * _modinv((2 * y) % curve.p, curve.p)) % curve.p x3 = (s * s - 2 * x) % curve.p y3 = (s * (x - x3) - y) % curve.p @@ -99,16 +99,16 @@ def _point_mul(curve: _Curve, k: int, P: tuple[int, int]) -> tuple[int, int] | N return None if k < 0: raise ValueError("scalar must be non-negative") - + result = None addend = P - + while k: if k & 1: result = _point_add(curve, result, addend) addend = _point_double(curve, addend) if addend else None k >>= 1 - + return result @@ -121,22 +121,22 @@ def _deterministic_k(curve: _Curve, private_key: int, message_hash: bytes) -> in from hashes import hmac_sha256 except ImportError: from sha2 import hmac_sha256 - + h = int.from_bytes(message_hash, "big") x = private_key q = curve.n - + h1 = h.to_bytes((curve.n.bit_length() + 7) // 8, "big") x_bytes = x.to_bytes((curve.n.bit_length() + 7) // 8, "big") - + v = b"\x01" * 32 k_val = b"\x00" * 32 - + k_val = hmac_sha256(k_val, v + b"\x00" + x_bytes + h1) v = hmac_sha256(k_val, v) k_val = hmac_sha256(k_val, v + b"\x01" + x_bytes + h1) v = hmac_sha256(k_val, v) - + while True: v = hmac_sha256(k_val, v) k_candidate = int.from_bytes(v, "big") @@ -148,85 +148,85 @@ def _deterministic_k(curve: _Curve, private_key: int, message_hash: bytes) -> in class ECDSAPrivateKey: """ECDSA private key.""" - + def __init__(self, curve: _Curve, d: int): if not (1 <= d < curve.n): raise ValueError("invalid private key") self.curve = curve self.d = d self._public_key = None - + @property def public_key(self) -> 'ECDSAPublicKey': if self._public_key is None: Q = _point_mul(self.curve, self.d, self.curve.g) self._public_key = ECDSAPublicKey(self.curve, Q) return self._public_key - + def sign(self, message: bytes, hash_fn=None) -> bytes: """Sign a message. Returns DER-encoded signature.""" if hash_fn is None: hash_fn = sha256 if self.curve.name == "P-256" else sha384 - + z = int.from_bytes(hash_fn(message), "big") z = z % self.curve.n - + k = _deterministic_k(self.curve, self.d, hash_fn(message)) R = _point_mul(self.curve, k, self.curve.g) if R is None: raise RuntimeError("signature generation failed") - + r = R[0] % self.curve.n if r == 0: raise RuntimeError("signature generation failed") - + s = (_modinv(k, self.curve.n) * (z + r * self.d)) % self.curve.n if s == 0: raise RuntimeError("signature generation failed") - + return _encode_signature(r, s) - + def ecdh(self, public_key: 'ECDSAPublicKey') -> bytes: """Perform ECDH key agreement.""" if self.curve.name != public_key.curve.name: raise ValueError("curve mismatch") - + shared_point = _point_mul(self.curve, self.d, public_key.Q) if shared_point is None: raise ValueError("ECDH failed") - + coord_bytes = (self.curve.p.bit_length() + 7) // 8 return shared_point[0].to_bytes(coord_bytes, "big") class ECDSAPublicKey: """ECDSA public key.""" - + def __init__(self, curve: _Curve, Q: tuple[int, int]): self.curve = curve self.Q = Q - + def verify(self, message: bytes, signature: bytes, hash_fn=None) -> bool: """Verify an ECDSA signature.""" if hash_fn is None: hash_fn = sha256 if self.curve.name == "P-256" else sha384 - + try: r, s = _decode_signature(signature) if not (1 <= r < self.curve.n and 1 <= s < self.curve.n): return False - + z = int.from_bytes(hash_fn(message), "big") z = z % self.curve.n - + w = _modinv(s, self.curve.n) u1 = (z * w) % self.curve.n u2 = (r * w) % self.curve.n - + point = _point_add(self.curve, _point_mul(self.curve, u1, self.curve.g), _point_mul(self.curve, u2, self.Q)) if point is None: return False - + return point[0] % self.curve.n == r except Exception: return False @@ -242,7 +242,7 @@ def _encode_int(x: int) -> bytes: if b[0] & 0x80: b = b"\x00" + b return b"\x02" + bytes([len(b)]) + b - + r_enc = _encode_int(r) s_enc = _encode_int(s) seq = r_enc + s_enc @@ -253,7 +253,7 @@ def _decode_signature(sig: bytes) -> tuple[int, int]: """Decode DER signature to (r, s).""" if sig[0] != 0x30: raise ValueError("invalid signature") - + idx = 2 if sig[idx] != 0x02: raise ValueError("invalid signature") @@ -262,14 +262,14 @@ def _decode_signature(sig: bytes) -> tuple[int, int]: idx += 1 r = int.from_bytes(sig[idx : idx + r_len], "big") idx += r_len - + if sig[idx] != 0x02: raise ValueError("invalid signature") idx += 1 s_len = sig[idx] idx += 1 s = int.from_bytes(sig[idx : idx + s_len], "big") - + return r, s diff --git a/src/crypto_standalone/asymmetric/rsa.py b/src/crypto_standalone/asymmetric/rsa.py index 1d70081..f213c84 100644 --- a/src/crypto_standalone/asymmetric/rsa.py +++ b/src/crypto_standalone/asymmetric/rsa.py @@ -172,13 +172,13 @@ def _miller_rabin_base2(n: int) -> bool: return True if n % 2 == 0: return False - + d = n - 1 s = 0 while d % 2 == 0: d //= 2 s += 1 - + x = pow(2, d, n) if x == 1 or x == n - 1: return True @@ -215,7 +215,7 @@ def _strong_lucas_test(n: int) -> bool: return True if n % 2 == 0: return False - + D = 5 while True: if _jacobi(D, n) == -1: @@ -223,14 +223,14 @@ def _strong_lucas_test(n: int) -> bool: D = -D - 2 if D > 0 else -D + 2 if abs(D) > 10000: return False - + Q = (1 - D) // 4 d = n + 1 s = 0 while d % 2 == 0: d //= 2 s += 1 - + def _lucas_chain(k: int) -> tuple[int, int]: U, V, Q_k = 0, 2, 1 for bit in bin(k)[2:]: @@ -250,17 +250,17 @@ def _lucas_chain(k: int) -> tuple[int, int]: Q_k = (Q_k * Q) % n U, V = U_new, V_new return U, V - + U, V = _lucas_chain(d) if U == 0 or V == 0: return True - + for _ in range(s): V = (V * V - 2 * pow(Q, d * (2 ** _), n)) % n if V == 0: return True d *= 2 - + return False @@ -273,7 +273,7 @@ def _is_probable_prime(n: int) -> bool: return True if n % p == 0: return False - + if not _miller_rabin_base2(n): return False if not _strong_lucas_test(n): diff --git a/src/crypto_standalone/asymmetric/x25519.py b/src/crypto_standalone/asymmetric/x25519.py index 2ab18a7..e20d9b3 100644 --- a/src/crypto_standalone/asymmetric/x25519.py +++ b/src/crypto_standalone/asymmetric/x25519.py @@ -29,28 +29,28 @@ def _x25519_scalarmult(k: bytes, u: bytes) -> bytes: """ if len(k) != 32 or len(u) != 32: raise ValueError("k and u must be 32 bytes") - + k_scalar = int.from_bytes(k, "little") k_scalar &= (1 << 255) - 1 k_scalar &= ~7 k_scalar |= (1 << 254) - + u_coord = int.from_bytes(u, "little") % _P - + x_1 = u_coord x_2 = 1 z_2 = 0 x_3 = u_coord z_3 = 1 swap = 0 - + for t in range(254, -1, -1): k_t = (k_scalar >> t) & 1 swap ^= k_t x_2, x_3 = _cswap(swap, x_2, x_3) z_2, z_3 = _cswap(swap, z_2, z_3) swap = k_t - + A = (x_2 + z_2) % _P AA = (A * A) % _P B = (x_2 - z_2) % _P @@ -64,10 +64,10 @@ def _x25519_scalarmult(k: bytes, u: bytes) -> bytes: z_3 = (x_1 * ((DA - CB) ** 2)) % _P x_2 = (AA * BB) % _P z_2 = (E * (AA + _A24 * E)) % _P - + x_2, x_3 = _cswap(swap, x_2, x_3) z_2, z_3 = _cswap(swap, z_2, z_3) - + result = (x_2 * pow(z_2, _P - 2, _P)) % _P return result.to_bytes(32, "little") @@ -81,7 +81,7 @@ def x25519_keygen(private_key: bytes | None = None) -> tuple[bytes, bytes]: private_key = random_bytes(32) if len(private_key) != 32: raise ValueError("private key must be 32 bytes") - + basepoint = bytes([9]) + bytes(31) public_key = _x25519_scalarmult(private_key, basepoint) return private_key, public_key @@ -96,7 +96,7 @@ def x25519(private_key: bytes, public_key: bytes) -> bytes: raise ValueError("private key must be 32 bytes") if len(public_key) != 32: raise ValueError("public key must be 32 bytes") - + return _x25519_scalarmult(private_key, public_key) diff --git a/src/crypto_standalone/kdf/hkdf.py b/src/crypto_standalone/kdf/hkdf.py index 7860caa..f016c9d 100644 --- a/src/crypto_standalone/kdf/hkdf.py +++ b/src/crypto_standalone/kdf/hkdf.py @@ -30,7 +30,7 @@ def hkdf_expand(prk: bytes, info: bytes | None, length: int, hash_fn=None) -> by hash_len = len(hash_fn(b"", b"")) if length > 255 * hash_len: raise ValueError("output length too large") - + n = (length + hash_len - 1) // hash_len okm = b"" t = b"" @@ -64,14 +64,14 @@ def pbkdf2_hmac(password: bytes, salt: bytes, iterations: int, dklen: int, hash_ hash_fn = hmac_sha256 if iterations < 1: raise ValueError("iterations must be >= 1") - + hash_len = len(hash_fn(b"", b"")) if dklen > (2**32 - 1) * hash_len: raise ValueError("derived key too long") - + num_blocks = (dklen + hash_len - 1) // hash_len dk = b"" - + for block_num in range(1, num_blocks + 1): u = hash_fn(password, salt + block_num.to_bytes(4, "big")) result = int.from_bytes(u, "big") @@ -79,7 +79,7 @@ def pbkdf2_hmac(password: bytes, salt: bytes, iterations: int, dklen: int, hash_ u = hash_fn(password, u) result ^= int.from_bytes(u, "big") dk += result.to_bytes(hash_len, "big") - + return dk[:dklen] diff --git a/src/crypto_standalone/kdf/pbkdf2.py b/src/crypto_standalone/kdf/pbkdf2.py index 7860caa..f016c9d 100644 --- a/src/crypto_standalone/kdf/pbkdf2.py +++ b/src/crypto_standalone/kdf/pbkdf2.py @@ -30,7 +30,7 @@ def hkdf_expand(prk: bytes, info: bytes | None, length: int, hash_fn=None) -> by hash_len = len(hash_fn(b"", b"")) if length > 255 * hash_len: raise ValueError("output length too large") - + n = (length + hash_len - 1) // hash_len okm = b"" t = b"" @@ -64,14 +64,14 @@ def pbkdf2_hmac(password: bytes, salt: bytes, iterations: int, dklen: int, hash_ hash_fn = hmac_sha256 if iterations < 1: raise ValueError("iterations must be >= 1") - + hash_len = len(hash_fn(b"", b"")) if dklen > (2**32 - 1) * hash_len: raise ValueError("derived key too long") - + num_blocks = (dklen + hash_len - 1) // hash_len dk = b"" - + for block_num in range(1, num_blocks + 1): u = hash_fn(password, salt + block_num.to_bytes(4, "big")) result = int.from_bytes(u, "big") @@ -79,7 +79,7 @@ def pbkdf2_hmac(password: bytes, salt: bytes, iterations: int, dklen: int, hash_ u = hash_fn(password, u) result ^= int.from_bytes(u, "big") dk += result.to_bytes(hash_len, "big") - + return dk[:dklen] diff --git a/src/crypto_standalone/symmetric/__init__.py b/src/crypto_standalone/symmetric/__init__.py index da9ad35..e059ecd 100644 --- a/src/crypto_standalone/symmetric/__init__.py +++ b/src/crypto_standalone/symmetric/__init__.py @@ -1,8 +1,9 @@ """Symmetric encryption: AES-256 (CBC, CTR, GCM) and ChaCha20-Poly1305.""" -from .aes import AES256 +from .aes import AES, AES256 +from .aes_ige import AESIGE, aes_ige_encrypt, aes_ige_decrypt from .aes_gcm import AESGCM from .chacha20 import ChaCha20Poly1305, chacha20_encrypt from .legacy_ciphers import TEA, RedPike, AveMariaCipher -__all__ = ["AES256", "AESGCM", "ChaCha20Poly1305", "chacha20_encrypt", "TEA", "RedPike", "AveMariaCipher"] +__all__ = ["AES", "AES256", "AESIGE", "aes_ige_encrypt", "aes_ige_decrypt", "AESGCM", "ChaCha20Poly1305", "chacha20_encrypt", "TEA", "RedPike", "AveMariaCipher"] diff --git a/src/crypto_standalone/symmetric/aes.py b/src/crypto_standalone/symmetric/aes.py index 2dc3cd0..daf165f 100644 --- a/src/crypto_standalone/symmetric/aes.py +++ b/src/crypto_standalone/symmetric/aes.py @@ -2,7 +2,6 @@ from __future__ import annotations -from dataclasses import dataclass import os @@ -369,7 +368,7 @@ def _inv_mix_columns(state: list[list[int]]) -> None: def _add_round_key(state: list[list[int]], rk: bytes) -> None: for c in range(4): - w = int.from_bytes(rk[4 * c: 4 * (c + 1)], "big") + w = int.from_bytes(rk[4 * c : 4 * (c + 1)], "big") for r in range(4): state[r][c] ^= (w >> (24 - 8 * r)) & 0xFF @@ -380,19 +379,23 @@ def _rot_word(w: int) -> int: def _sub_word(w: int) -> int: return ( - (SBOX[(w >> 24) & 0xFF] << 24) - | (SBOX[(w >> 16) & 0xFF] << 16) - | (SBOX[(w >> 8) & 0xFF] << 8) - | SBOX[w & 0xFF] + (SBOX[(w >> 24) & 0xFF] << 24) | (SBOX[(w >> 16) & 0xFF] << 16) | (SBOX[(w >> 8) & 0xFF] << 8) | SBOX[w & 0xFF] ) -def _key_expand(key: bytes) -> list[bytes]: - if len(key) != 32: - raise ValueError("AES-256 requires 32-byte key") +def _key_expand(key: bytes) -> tuple[int, list[bytes]]: + if len(key) not in (16, 24, 32): + raise ValueError("AES key must be 16, 24, or 32 bytes") + nb = 4 - nk = 8 - nr = 14 + nk = len(key) // 4 + if nk == 4: + nr = 10 + elif nk == 6: + nr = 12 + else: + nr = 14 + w = [0] * (nb * (nr + 1)) for i in range(nk): @@ -402,7 +405,7 @@ def _key_expand(key: bytes) -> list[bytes]: temp = w[i - 1] if i % nk == 0: temp = _sub_word(_rot_word(temp)) ^ (RCON[i // nk - 1] << 24) - elif i % nk == 4: + elif nk > 6 and i % nk == 4: temp = _sub_word(temp) w[i] = w[i - nk] ^ temp @@ -412,7 +415,7 @@ def _key_expand(key: bytes) -> list[bytes]: for c in range(4): rk.extend(w[4 * r + c].to_bytes(4, "big")) round_keys.append(bytes(rk)) - return round_keys + return nr, round_keys def _xor_block(a: bytes, b: bytes) -> bytes: @@ -437,14 +440,14 @@ def _unpad_pkcs7(data: bytes) -> bytes: return data[:-n] -@dataclass(frozen=True) -class AES256: - key: bytes +class AES: + block_size = 16 - def __post_init__(self) -> None: - if len(self.key) != 32: - raise ValueError("key must be 32 bytes") - object.__setattr__(self, "_rk", _key_expand(self.key)) + def __init__(self, key: bytes) -> None: + if len(key) not in (16, 24, 32): + raise ValueError("AES key must be 16, 24, or 32 bytes") + self.key = key + self._nr, self._rk = _key_expand(self.key) @property def _round_keys(self) -> list[bytes]: @@ -457,14 +460,14 @@ def encrypt_block(self, block: bytes) -> bytes: rks = self._round_keys _add_round_key(state, rks[0]) - for round_idx in range(1, 14): + for round_idx in range(1, self._nr): _sub_bytes(state) _shift_rows(state) _mix_columns(state) _add_round_key(state, rks[round_idx]) _sub_bytes(state) _shift_rows(state) - _add_round_key(state, rks[14]) + _add_round_key(state, rks[self._nr]) return _state_to_bytes(state) def decrypt_block(self, block: bytes) -> bytes: @@ -473,8 +476,8 @@ def decrypt_block(self, block: bytes) -> bytes: state = _bytes_to_state(block) rks = self._round_keys - _add_round_key(state, rks[14]) - for round_idx in range(13, 0, -1): + _add_round_key(state, rks[self._nr]) + for round_idx in range(self._nr - 1, 0, -1): _inv_shift_rows(state) _inv_sub_bytes(state) _add_round_key(state, rks[round_idx]) @@ -542,3 +545,10 @@ def decrypt_ctr(self, data: bytes) -> bytes: out.extend(_xor_block(block, keystream[: len(block)])) counter += 1 return bytes(out) + + +class AES256(AES): + def __init__(self, key: bytes) -> None: + if len(key) != 32: + raise ValueError("key must be 32 bytes") + super().__init__(key) diff --git a/src/crypto_standalone/symmetric/aes_ige.py b/src/crypto_standalone/symmetric/aes_ige.py new file mode 100644 index 0000000..70a0288 --- /dev/null +++ b/src/crypto_standalone/symmetric/aes_ige.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +from .aes import AES, _xor_block + + +class AESIGE: + def __init__(self, cipher: AES) -> None: + self._cipher = cipher + + def encrypt(self, plaintext: bytes, iv: bytes) -> bytes: + if len(iv) != 32: + raise ValueError("AES-IGE IV must be exactly 32 bytes") + if len(plaintext) % 16 != 0: + raise ValueError("Plaintext length must be a multiple of the 16-byte block size") + + c_prev = iv[:16] # C_0 + p_prev = iv[16:] # P_0 + + out = bytearray() + for i in range(0, len(plaintext), 16): + p_i = plaintext[i : i + 16] + xored_in = _xor_block(p_i, c_prev) + e_k = self._cipher.encrypt_block(xored_in) + c_i = _xor_block(e_k, p_prev) + + out.extend(c_i) + c_prev = c_i + p_prev = p_i + + return bytes(out) + + def decrypt(self, ciphertext: bytes, iv: bytes) -> bytes: + if len(iv) != 32: + raise ValueError("AES-IGE IV must be exactly 32 bytes") + if len(ciphertext) % 16 != 0: + raise ValueError("Ciphertext length must be a multiple of the 16-byte block size") + + c_prev = iv[:16] # C_0 + p_prev = iv[16:] # P_0 + + out = bytearray() + for i in range(0, len(ciphertext), 16): + c_i = ciphertext[i : i + 16] + xored_in = _xor_block(c_i, p_prev) + d_k = self._cipher.decrypt_block(xored_in) + p_i = _xor_block(d_k, c_prev) + + out.extend(p_i) + c_prev = c_i + p_prev = p_i + + return bytes(out) + + +def aes_ige_encrypt(plaintext: bytes, key: bytes, iv: bytes) -> bytes: + """ + Encrypts data using AES in Infinite Garble Extension (IGE) mode. + + AES-IGE provides confidentiality only. It does not provide authentication + or integrity protection. It must be paired with a separate authentication + mechanism when used in a protocol that does not already authenticate the + encrypted data. + + Args: + plaintext: The data to encrypt. Must be a multiple of 16 bytes. + key: The AES key (16, 24, or 32 bytes). + iv: The 32-byte IV (C_0 || P_0). + + Returns: + The ciphertext bytes. + """ + return AESIGE(AES(key)).encrypt(plaintext, iv) + + +def aes_ige_decrypt(ciphertext: bytes, key: bytes, iv: bytes) -> bytes: + """ + Decrypts data using AES in Infinite Garble Extension (IGE) mode. + + AES-IGE provides confidentiality only. It does not provide authentication + or integrity protection. It must be paired with a separate authentication + mechanism when used in a protocol that does not already authenticate the + encrypted data. + + Args: + ciphertext: The data to decrypt. Must be a multiple of 16 bytes. + key: The AES key (16, 24, or 32 bytes). + iv: The 32-byte IV (C_0 || P_0). + + Returns: + The plaintext bytes. + """ + return AESIGE(AES(key)).decrypt(ciphertext, iv) diff --git a/src/crypto_standalone/utils/memory.py b/src/crypto_standalone/utils/memory.py index 4f93ee5..b894ab0 100644 --- a/src/crypto_standalone/utils/memory.py +++ b/src/crypto_standalone/utils/memory.py @@ -8,14 +8,14 @@ def secure_zero(data: bytearray) -> None: """ Overwrite bytearray with zeros. Best-effort only. - + Limitations: - Python interns short strings/bytes - CPython caches small integers (-5 to 256) - Immutable bytes objects cannot be zeroed - GC may leave copies in memory - No guarantee against swap/hibernation - + Use this for bytearrays holding sensitive data like keys. """ if not isinstance(data, bytearray): @@ -27,23 +27,23 @@ def secure_zero(data: bytearray) -> None: class SecureBytes: """ Context manager for sensitive byte data. Auto-zeros on exit. - + Example: with SecureBytes(32) as key: key[:] = os.urandom(32) # use key # key is zeroed here """ - + def __init__(self, size: int) -> None: self._data = bytearray(size) - + def __enter__(self) -> bytearray: return self._data - + def __exit__(self, *args: Any) -> None: secure_zero(self._data) - + def __len__(self) -> int: return len(self._data) @@ -51,12 +51,12 @@ def __len__(self) -> int: def constant_time_compare(a: bytes, b: bytes) -> bool: """ Constant-time comparison (re-export from hashes for convenience). - + Note: Pure Python is NOT truly constant-time due to: - Variable-time integer operations - Branch prediction - Cache timing - + This is best-effort. For adversarial timing scenarios, use native crypto. """ try: diff --git a/tests/adversarial/test_attack_simulations.py b/tests/adversarial/test_attack_simulations.py index fbff2a8..4c432d3 100644 --- a/tests/adversarial/test_attack_simulations.py +++ b/tests/adversarial/test_attack_simulations.py @@ -20,7 +20,7 @@ class TestGCMForbiddenAttack: """ GCM nonce reuse (Forbidden Attack). - + When the same nonce is used with the same key for two different messages, an attacker can XOR the ciphertexts to get the XOR of the plaintexts. This also reveals the GHASH key H. @@ -31,27 +31,27 @@ def test_nonce_reuse_reveals_plaintext_xor(self): key = os.urandom(32) gcm = AESGCM(key) nonce = os.urandom(12) - + pt1 = b"message one here" pt2 = b"message two there" - + ct1, tag1 = gcm.encrypt(pt1, nonce) ct2, tag2 = gcm.encrypt(pt2, nonce) - + # XOR of ciphertexts = XOR of plaintexts (keystream cancels) ct_xor = bytes(a ^ b for a, b in zip(ct1, ct2)) pt_xor = bytes(a ^ b for a, b in zip(pt1, pt2)) assert ct_xor == pt_xor # This IS the attack - + def test_nonce_reuse_both_decrypt(self): """Both messages still decrypt correctly (the attack is passive).""" key = os.urandom(32) gcm = AESGCM(key) nonce = os.urandom(12) - + ct1, tag1 = gcm.encrypt(b"secret1", nonce) ct2, tag2 = gcm.encrypt(b"secret2", nonce) - + assert gcm.decrypt(ct1, tag1, nonce) == b"secret1" assert gcm.decrypt(ct2, tag2, nonce) == b"secret2" @@ -59,11 +59,11 @@ def test_nonce_reuse_both_decrypt(self): class TestECDSANonceReuse: """ ECDSA nonce reuse attack. - + If the same nonce k is used for two different messages, the private key can be recovered: d = (z1 - z2) * inv(k) mod n where k = (s1 - s2) * inv(r) mod n. - + Our implementation uses RFC 6979 deterministic k, which guarantees different k for different messages. """ @@ -73,19 +73,19 @@ def test_deterministic_k_prevents_nonce_reuse(self): sk = p256_keygen() msg1 = b"message one" msg2 = b"message two" - + sig1 = sk.sign(msg1) sig2 = sk.sign(msg2) - + r1, s1 = _decode_signature(sig1) r2, s2 = _decode_signature(sig2) - + # Same message should produce same k (and same signature) sig1b = sk.sign(msg1) r1b, s1b = _decode_signature(sig1b) assert r1 == r1b # Same k → same r assert s1 == s1b # Same message → same s - + # Different messages should produce different k # (r values differ with overwhelming probability) # Note: there's a tiny chance r1 == r2 with different k, @@ -93,7 +93,7 @@ def test_deterministic_k_prevents_nonce_reuse(self): if r1 == r2: # If r is the same, s must be different (different z) assert s1 != s2, "Same r and s with different messages = nonce reuse!" - + def test_same_message_same_signature(self): """Deterministic signing: same input = same output.""" sk = p256_keygen() @@ -105,7 +105,7 @@ def test_same_message_same_signature(self): class TestChaCha20NonceReuse: """ ChaCha20 nonce reuse. - + Reusing the same nonce with the same key produces the same keystream, allowing XOR of plaintexts to be recovered. """ @@ -114,13 +114,13 @@ def test_nonce_reuse_reveals_keystream_xor(self): key = os.urandom(32) cp = ChaCha20Poly1305(key) nonce = os.urandom(12) - + pt1 = b"alpha bravo charlie" pt2 = b"delta echo foxtrot" - + ct1, tag1 = cp.encrypt(pt1, nonce) ct2, tag2 = cp.encrypt(pt2, nonce) - + # XOR of ciphertexts = XOR of plaintexts ct_xor = bytes(a ^ b for a, b in zip(ct1, ct2)) pt_xor = bytes(a ^ b for a, b in zip(pt1, pt2)) @@ -130,7 +130,7 @@ def test_nonce_reuse_reveals_keystream_xor(self): class TestSignatureMalleability: """ Signature malleability attacks. - + For ECDSA, (r, s) and (r, -s mod n) are both valid signatures. This can be exploited in blockchain/cryptocurrency contexts where transaction uniqueness depends on signature uniqueness. @@ -142,11 +142,11 @@ def test_ecdsa_malleability_both_valid(self): msg = b"transaction data" sig = sk.sign(msg) r, s = _decode_signature(sig) - + sig_malleable = _encode_signature(r, (-s) % P256.n) assert sk.public_key.verify(msg, sig) assert sk.public_key.verify(msg, sig_malleable) - + def test_ed25519_not_malleable_by_design(self): """Ed25519 signatures are deterministic and tied to the exact message.""" sk, pk = ed25519_keygen() @@ -164,7 +164,7 @@ def test_ed25519_sig_not_valid_as_ecdsa(self): """Ed25519 signature bytes are not valid ECDSA DER.""" sk_ed, pk_ed = ed25519_keygen() sig_ed = ed25519_sign(b"test", sk_ed) - + # Ed25519 signature is 64 raw bytes, not DER # ECDSA verify expects DER, so it should fail gracefully sk_ec = p256_keygen() @@ -180,7 +180,7 @@ def test_x25519_shared_secret_not_valid_aes_key_directly(self): sk1, pk1 = x25519_keygen() sk2, pk2 = x25519_keygen() shared = x25519(sk1, pk2) - + # While the shared secret IS 32 bytes (valid AES-256 key size), # best practice is to derive the actual key via HKDF from crypto_standalone import hkdf_sha256 diff --git a/tests/adversarial/test_ecdsa_wycheproof.py b/tests/adversarial/test_ecdsa_wycheproof.py index fd86f19..a8be9ce 100644 --- a/tests/adversarial/test_ecdsa_wycheproof.py +++ b/tests/adversarial/test_ecdsa_wycheproof.py @@ -27,11 +27,11 @@ def test_p256_malleable_signature(self): msg = b"test" sig = sk.sign(msg) r, s = _decode_signature(sig) - + # Create malleable signature: (r, -s mod n) s_neg = (-s) % P256.n sig_malleable = _encode_signature(r, s_neg) - + # Both should verify (standard ECDSA without low-S constraint) assert sk.public_key.verify(msg, sig) assert sk.public_key.verify(msg, sig_malleable) diff --git a/tests/adversarial/test_fuzz_hypothesis.py b/tests/adversarial/test_fuzz_hypothesis.py index 071c182..5409641 100644 --- a/tests/adversarial/test_fuzz_hypothesis.py +++ b/tests/adversarial/test_fuzz_hypothesis.py @@ -178,10 +178,10 @@ def test_aes_gcm_tamper_detection(plaintext, aad, flip_pos): key = os.urandom(32) gcm = AESGCM(key) blob = gcm.encrypt_blob(plaintext, aad=aad) - + if flip_pos >= len(blob): return # skip if flip position is beyond blob - + tampered = bytearray(blob) tampered[flip_pos] ^= 0x01 with pytest.raises(ValueError): diff --git a/tests/unit/test_aes_ige.py b/tests/unit/test_aes_ige.py new file mode 100644 index 0000000..6b826c3 --- /dev/null +++ b/tests/unit/test_aes_ige.py @@ -0,0 +1,79 @@ +import pytest +import binascii +from src.crypto_standalone.symmetric.aes import AES, AES256 +from src.crypto_standalone.symmetric.aes_ige import aes_ige_encrypt, aes_ige_decrypt + +# Test Vectors Generated using OpenSSL equivalent functionality +AES_128_IGE = { + "key": "9c5deda82f20fd389c3158f8948741d8", + "iv": "5f5093204e42d4c8f61ba70ac0b044edd01515f67299b102a538d0baddab46c3", + "pt": "99844ed5c600021004f9fea56b978ca9b314b45d885a484c1fdc9ad1cc6e8f239aeb160d3a44afbf76624761686eded43b95645c451265c7825711c97bc808d2", + "ct": "626e8fab4bd8c0d807674de8ac3f57a2e096dcc22eb5f21721d3611ad2ce7cbff09325393639e5b0586d87dd529629b4c5f205420a0788a51ce197e17d4df474", +} + +AES_192_IGE = { + "key": "eb662821f13222939a7dcb798066465523f64574bee4ba92", + "iv": "7e73adfb7a244e2c531d97c3963ae530d66ab8c480a08a4bf1420924b1f8b7e0", + "pt": "2826dfa50a9e7bd8ef9dd7dae9a63814230509f529d5dd8261a74c5c68e4583e6e07bc364a72a4a3c99da5fd8c91b199f2eaad6811e61995f59656faf1dde6ea", + "ct": "a4cb85365267b110ec87ad00e5f9128537fe40e2c801c02c8345486af40da3a25f6448b18cd311ce9d8d028855da9a60bf6d43a9c944c986229abe25fddf2eaf", +} + +AES_256_IGE = { + "key": "df9533a306cab216fe8ea0e85894a576d6ddf201a5623542b4438bbcfce107b7", + "iv": "3ad730c66484bb5287ddc74b2130b8f19883e5bddec9484d0ec5b899ae28ca1a", + "pt": "ab2810ff8f4d39ddb54e4238f7ad59d189cb056b1bf5c6f56d250a02bc8a970f0f7869391d9dc7ae68a782f24f8d248030ed6015c7fe03627c861099311c78f1", + "ct": "9577723b7397ce6f969e44c520ee92363130a44b8ae1be984256442a7722ceacf950044f8fdd6bff8ad7fb9d658a9129f04b964e476ec2a25bbd7020eb799fea", +} + + +class TestAESIGE: + @pytest.mark.parametrize("vector", [AES_128_IGE, AES_192_IGE, AES_256_IGE]) + def test_kat(self, vector): + key = binascii.unhexlify(vector["key"]) + iv = binascii.unhexlify(vector["iv"]) + pt = binascii.unhexlify(vector["pt"]) + expected_ct = binascii.unhexlify(vector["ct"]) + + ct = aes_ige_encrypt(pt, key, iv) + assert ct == expected_ct + + dec_pt = aes_ige_decrypt(expected_ct, key, iv) + assert dec_pt == pt + + def test_invalid_iv_length(self): + key = b"\x00" * 32 + pt = b"\x00" * 16 + with pytest.raises(ValueError, match="must be exactly 32 bytes"): + aes_ige_encrypt(pt, key, b"\x00" * 16) + with pytest.raises(ValueError, match="must be exactly 32 bytes"): + aes_ige_decrypt(pt, key, b"\x00" * 16) + + def test_invalid_plaintext_length(self): + key = b"\x00" * 32 + iv = b"\x00" * 32 + pt = b"\x00" * 15 + with pytest.raises(ValueError, match="must be a multiple"): + aes_ige_encrypt(pt, key, iv) + with pytest.raises(ValueError, match="must be a multiple"): + aes_ige_decrypt(pt, key, iv) + + def test_empty_input(self): + key = b"\x00" * 32 + iv = b"\x00" * 32 + assert aes_ige_encrypt(b"", key, iv) == b"" + assert aes_ige_decrypt(b"", key, iv) == b"" + + +class TestAES: + def test_key_lengths(self): + AES(b"\x00" * 16) + AES(b"\x00" * 24) + AES(b"\x00" * 32) + + with pytest.raises(ValueError): + AES(b"\x00" * 15) + + def test_aes256_compat(self): + AES256(b"\x00" * 32) + with pytest.raises(ValueError): + AES256(b"\x00" * 16) diff --git a/tests/unit/test_ave_maria_cipher.py b/tests/unit/test_ave_maria_cipher.py index 3742260..1cb80a6 100644 --- a/tests/unit/test_ave_maria_cipher.py +++ b/tests/unit/test_ave_maria_cipher.py @@ -38,4 +38,4 @@ def test_type_errors(self): with pytest.raises(TypeError): c.encrypt(b"abc") # type: ignore[arg-type] with pytest.raises(TypeError): - c.decrypt(123) # type: ignore[arg-type] \ No newline at end of file + c.decrypt(123) # type: ignore[arg-type]