LeoVasanko/aegis-python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

API updates:

Browse Source 
- Mac class follows hashlib API: digest functions added and finalization no longer modifies state.
- Encryptor and Decryptor now raise RuntimeError if still used after final.

Documentation updated with the changes and  further examples.

Tests updated with the changes, new test module for error cases (test_raises).

Docstrings improved.
This commit is contained in:
LeoVasanko
2025-11-09 06:53:27 +00:00
parent 46bc05f547
commit eed36e2463
11 changed files with 1191 additions and 711 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+44 -15
View File
@@ -70,22 +70,26 @@ No MAC tag, vulnerable to alterations:
### Incremental AEAD ### Incremental AEAD
Stateful classes that can be used for processing the data in separate chunks: Stateful classes that can be used for processing the data in separate chunks:
- Encryptor(key, nonce, ad=None) - Encryptor(key, nonce, ad=None, maclen=16)
- update(message[, into]) -> ciphertext_chunk - update(message[, into]) -> ciphertext_chunk
- final([into], maclen=16) -> mac_tag - final([into]) -> mac_tag
- Decryptor(key, nonce, ad=None) - Decryptor(key, nonce, ad=None, maclen=16)
- update(ct_chunk[, into]) -> plaintext_chunk - update(ct_chunk[, into]) -> plaintext_chunk
- final(mac) -> None (raises ValueError on failure) - final(mac) -> raises ValueError on failure
### Message Authentication Code ### Message Authentication Code
No encryption, but prevents changes to the data without the correct key. No encryption, but prevents changes to the data without the correct key.
- mac(key, nonce, data, maclen=16, into=None) -> mac - mac(key, nonce, data, maclen=16, into=None) -> mac
- Mac(key, nonce) - Mac(key, nonce, maclen=16)
- update(data) - update(data)
- final(maclen=16[, into]) -> mac - final([into]) -> mac
- verify(mac) -> bool (True on success; raises ValueError on failure) - verify(mac) -> raises ValueError on failure
- digest() -> bytes
- hexdigest() -> str
The `Mac` class follows the Python hashlib API for compatibility with code expecting hash objects. Finalizing does not alter the state, so further updates appending to the already input data can be issued even after calling the other methods that calculate the MAC.
### Keystream generation ### Keystream generation
@@ -119,12 +123,18 @@ from pyaegis import aegis256x4 as ciph
key, nonce = ciph.random_key(), bytes(ciph.NONCEBYTES) key, nonce = ciph.random_key(), bytes(ciph.NONCEBYTES)
mac = ciph.mac(key, nonce, b"message", maclen=32) mac = ciph.mac(key, nonce, b"message", maclen=32)
print(mac) print(mac.hex())
st = ciph.Mac(key, nonce) # Alternative class-based API
st.update(b"message") a = ciph.Mac(key, nonce, maclen=32)
st.update(b"Mallory Says Hello!") a.update(b"message")
st.verify(mac) # Raises ValueError print(a.hexdigest())
# Verification
b = ciph.Mac(key, nonce, maclen=32)
b.update(b"message")
b.update(b"Mallory Says Hello!")
b.verify(mac) # Raises ValueError
``` ```
### Detached mode encryption and decryption ### Detached mode encryption and decryption
@@ -151,12 +161,12 @@ Class-based interface for incremental updates is an alternative to the one-shot
from pyaegis import aegis256x4 as ciph from pyaegis import aegis256x4 as ciph
key, nonce = ciph.random_key(), ciph.random_nonce() key, nonce = ciph.random_key(), ciph.random_nonce()
enc = ciph.Encryptor(key, nonce, ad=b"header") enc = ciph.Encryptor(key, nonce, ad=b"header", maclen=16)
c1 = enc.update(b"chunk1") c1 = enc.update(b"chunk1")
c2 = enc.update(b"chunk2") c2 = enc.update(b"chunk2")
mac = enc.final(maclen=16) mac = enc.final()
dec = ciph.Decryptor(key, nonce, ad=b"header") dec = ciph.Decryptor(key, nonce, ad=b"header", maclen=16)
p1 = dec.update(c1) p1 = dec.update(c1)
p2 = dec.update(c2) p2 = dec.update(c2)
dec.final(mac) # raises ValueError on failure dec.final(mac) # raises ValueError on failure
@@ -204,6 +214,25 @@ with open("encrypted.bin", "rb") as f:
print(pt) print(pt)
``` ```
### Random generator
The stream generator is much faster than any traditional random number generator, cryptographically secure and seekable. Use `random_key()` for unpredictable output.
```python
from pyaegis import aegis128x4 as ciph
key = b"SeedForReplay001" # A non-random deterministic seed (16 bytes)
nonce = bytearray(ciph.NONCEBYTES) # All-zeroes nonce
# Generate multiple blocks of pseudorandom data
for i in range(5):
rand = ciph.stream(key, nonce, 10)
print(f"Block {int.from_bytes(nonce, "little")}: {rand.hex()}")
ciph.nonce_increment(nonce)
```
Note: this is seekable by converting the block number to nonce with `idx.to_bytes(ciph.NONCEBYTES, "little")`, given some fixed block size (e.g. 1 MiB).
### Preallocated output buffers (into=) ### Preallocated output buffers (into=)
For advanced use cases, the output buffer can be supplied with `into` kwarg. Any type of writable buffer with len() >= space required can be used. This includes bytearrays, memoryviews, mmap files, numpy.getbuffer etc. For advanced use cases, the output buffer can be supplied with `into` kwarg. Any type of writable buffer with len() >= space required can be used. This includes bytearrays, memoryviews, mmap files, numpy.getbuffer etc.
pyaegis/aegis128l.py
+142 -113
View File
@@ -9,7 +9,7 @@ from ._loader import ffi
from ._loader import lib as _lib from ._loader import lib as _lib
from .util import Buffer, new_aligned_struct, nonce_increment, wipe from .util import Buffer, new_aligned_struct, nonce_increment, wipe
NAME = "AEGIS-128L" #: Algorithm name NAME = "AEGIS-128L" #: Algorithm display name
KEYBYTES = 16 #: Key size in bytes (varies by algorithm) KEYBYTES = 16 #: Key size in bytes (varies by algorithm)
NONCEBYTES = 16 #: Nonce size in bytes (varies by algorithm) NONCEBYTES = 16 #: Nonce size in bytes (varies by algorithm)
MACBYTES = 16 #: Normal MAC size (always 16) MACBYTES = 16 #: Normal MAC size (always 16)
@@ -20,15 +20,22 @@ RATE = 64 #: Byte chunk size in internal processing
def random_key() -> bytearray: def random_key() -> bytearray:
""" """
Generate a random key using cryptographically secure random bytes. Generate a secret key using cryptographically secure random bytes.
It is recommended to wipe() the key after no longer needed to keep it secret. It is recommended to wipe() the key after no longer needed.
""" """
return bytearray(secrets.token_bytes(KEYBYTES)) return bytearray(secrets.token_bytes(KEYBYTES))
def random_nonce() -> bytearray: def random_nonce() -> bytearray:
"""Generate a random nonce using cryptographically secure random bytes.""" """
Generate a public nonce using cryptographically secure random bytes.
Nonces (a number used once) are public data that may be sent together
with the ciphertext, but they need to be unique for each use.
See also: nonce_increment() can be used to derive sequential nonces.
"""
return bytearray(secrets.token_bytes(NONCEBYTES)) return bytearray(secrets.token_bytes(NONCEBYTES))
@@ -46,11 +53,11 @@ def encrypt_detached(
ct_into: Buffer | None = None, ct_into: Buffer | None = None,
mac_into: Buffer | None = None, mac_into: Buffer | None = None,
) -> tuple[bytearray | memoryview, bytearray | memoryview]: ) -> tuple[bytearray | memoryview, bytearray | memoryview]:
f"""Encrypt message with associated data, returning ciphertext and MAC separately. """Encrypt message with associated data, returning ciphertext and MAC separately.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -114,11 +121,11 @@ def decrypt_detached(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext with detached MAC and associated data. """Decrypt ciphertext with detached MAC and associated data.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext to decrypt. ct: The ciphertext to decrypt.
mac: The MAC to verify. mac: The MAC to verify.
ad: Associated data (optional). ad: Associated data (optional).
@@ -170,11 +177,11 @@ def encrypt(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Encrypt message with associated data, returning ciphertext with appended MAC. """Encrypt message with associated data, returning ciphertext with appended MAC.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -226,11 +233,11 @@ def decrypt(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext with appended MAC and associated data. """Decrypt ciphertext with appended MAC and associated data.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext with MAC to decrypt. ct: The ciphertext with MAC to decrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -283,11 +290,11 @@ def stream(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | Buffer: ) -> bytearray | Buffer:
f"""Generate a stream of pseudorandom bytes. """Generate a stream of pseudorandom bytes.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}, uses zeroes for nonce if None). nonce: Public nonce (generate with random_nonce(), uses zeroes for nonce if None).
length: Number of bytes to generate (required if into is None). length: Number of bytes to generate (required if into is None).
into: Buffer to write stream into (default: bytearray created). into: Buffer to write stream into (default: bytearray created).
@@ -325,11 +332,11 @@ def encrypt_unauthenticated(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Encrypt message without authentication (for testing/debugging). """Encrypt message without authentication (for testing/debugging).
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
into: Buffer to write ciphertext into (default: bytearray created). into: Buffer to write ciphertext into (default: bytearray created).
@@ -366,11 +373,11 @@ def decrypt_unauthenticated(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext without authentication (for testing/debugging). """Decrypt ciphertext without authentication (for testing/debugging).
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext to decrypt. ct: The ciphertext to decrypt.
into: Buffer to write plaintext into (default: bytearray created). into: Buffer to write plaintext into (default: bytearray created).
@@ -408,11 +415,11 @@ def mac(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Compute a MAC for the given data in one shot. """Compute a MAC for the given data in one shot.
Args: Args:
key: Key ({KEYBYTES=}) key: Secret key (generate with random_key())
nonce: Nonce ({NONCEBYTES=}) nonce: Public nonce (generate with random_nonce())
data: Data to MAC data: Data to MAC
maclen: MAC length (16 or 32, default 16) maclen: MAC length (16 or 32, default 16)
into: Buffer to write MAC into (default: bytearray created) into: Buffer to write MAC into (default: bytearray created)
@@ -420,70 +427,59 @@ def mac(
Returns: Returns:
MAC bytes as bytearray if into not provided, memoryview of into otherwise MAC bytes as bytearray if into not provided, memoryview of into otherwise
""" """
mac_state = Mac(key, nonce) mac_state = Mac(key, nonce, maclen)
mac_state.update(data) mac_state.update(data)
return mac_state.final(maclen, into) return mac_state.final(into)
class Mac: class Mac:
"""AEGIS-128L MAC state wrapper. """AEGIS-128L MAC state wrapper.
Usage: Example:
mac = Mac(key, nonce) a = Mac(key, nonce)
mac.update(data) a.update(data)
tag = mac.final() # defaults to 16-byte MAC mac = a.final()
# or verify:
mac2 = Mac(key, nonce); mac2.update(data); mac2.verify(tag)
""" """
__slots__ = ("_st", "_owner") __slots__ = ("_st", "_owner", "_maclen")
def __init__( def __init__(self, key: Buffer, nonce: Buffer, maclen: int = MACBYTES) -> None:
self, """Create a MAC with the given key, nonce, and tag length.
key: Buffer,
nonce: Buffer,
_other=None,
) -> None:
f"""Initialize a MAC state with a nonce and key.
Args:
key: Key ({KEYBYTES=}).
nonce: Nonce ({NONCEBYTES=}).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
st, owner = new_aligned_struct("aegis128l_mac_state", ALIGNMENT) if maclen not in (16, 32):
self._st = st raise TypeError("maclen must be 16 or 32")
self._owner = owner
if _other is not None: # clone path
_lib.aegis128l_mac_state_clone(self._st, _other._st)
return
# Normal init path
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
raise TypeError(f"nonce length must be {NONCEBYTES}") raise TypeError(f"nonce length must be {NONCEBYTES}")
self._maclen = maclen
st, owner = new_aligned_struct("aegis128l_mac_state", ALIGNMENT)
self._st = st
self._owner = owner
_lib.aegis128l_mac_init(self._st, _ptr(key), _ptr(nonce)) _lib.aegis128l_mac_init(self._st, _ptr(key), _ptr(nonce))
def __deepcopy__(self) -> "Mac": def __deepcopy__(self) -> "Mac":
"""Return a clone of current MAC state.""" """Return a clone of current MAC state."""
return Mac(b"", b"", _other=self) clone = object.__new__(Mac)
clone._maclen = self._maclen
clone._st, clone._owner = new_aligned_struct("aegis128l_mac_state", ALIGNMENT)
_lib.aegis128l_mac_state_clone(clone._st, self._st)
return clone
clone = __deepcopy__ clone = __deepcopy__
def reset(self) -> None: def reset(self) -> None:
"""Reset the MAC state so it can be reused with the same nonce and key.""" """Reset back to the original state, prior to any updates."""
_lib.aegis128l_mac_reset(self._st) _lib.aegis128l_mac_reset(self._st)
def update(self, data: Buffer) -> None: def update(self, data: Buffer) -> None:
"""Absorb data into the MAC state. """Update the MAC state with more data.
Args: Repeated calls to update() are equivalent to a single call with the concatenated data.
data: Bytes-like object to authenticate.
Raises:
RuntimeError: If the underlying C function reports an error.
""" """
rc = _lib.aegis128l_mac_update(self._st, _ptr(data), len(data)) rc = _lib.aegis128l_mac_update(self._st, _ptr(data), len(data))
if rc != 0: if rc != 0:
@@ -491,15 +487,13 @@ class Mac:
err_name = errno.errorcode.get(err_num, f"errno_{err_num}") err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
raise RuntimeError(f"mac update failed: {err_name}") raise RuntimeError(f"mac update failed: {err_name}")
def final( def final(self, into: Buffer | None = None) -> bytearray | memoryview:
self, """Calculate and return the MAC tag for the currently input data.
maclen: int = MACBYTES,
into: Buffer | None = None, Unlike the C library, this method does not alter the current state,
) -> bytearray | memoryview: allowing for multiple calls and further updates on the same object.
"""Finalize and return the MAC tag.
Args: Args:
maclen: Tag length in bytes (16 or 32). Defaults to 16.
into: Optional buffer to write the tag into (default: bytearray created). into: Optional buffer to write the tag into (default: bytearray created).
Returns: Returns:
@@ -509,30 +503,36 @@ class Mac:
TypeError: If lengths are invalid. TypeError: If lengths are invalid.
RuntimeError: If finalization fails in the C library. RuntimeError: If finalization fails in the C library.
""" """
if maclen not in (16, 32): maclen = self._maclen
raise TypeError("maclen must be 16 or 32")
if into is None: if into is None:
out = bytearray(maclen) out = bytearray(maclen)
else: else:
if len(into) < maclen: if len(into) < maclen:
raise TypeError("into length must be at least maclen") raise TypeError("into length must be at least maclen")
out = into out = into
out_mv = memoryview(out)
rc = _lib.aegis128l_mac_final(self._st, ffi.from_buffer(out_mv), maclen) rc = _lib.aegis128l_mac_final(self.clone()._st, ffi.from_buffer(out), maclen)
if rc != 0: if rc != 0:
err_num = ffi.errno err_num = ffi.errno
err_name = errno.errorcode.get(err_num, f"errno_{err_num}") err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
raise RuntimeError(f"mac final failed: {err_name}") raise RuntimeError(f"mac final failed: {err_name}")
return out if into is None else memoryview(out)[:maclen] # type: ignore return out if into is None else memoryview(out)[:maclen] # type: ignore
def digest(self) -> bytes:
"""Calculate and return the MAC tag as bytes."""
return bytes(self.final())
def hexdigest(self) -> str:
"""Calculate and return the MAC tag as a hex string."""
return self.digest().hex()
def verify(self, mac: Buffer): def verify(self, mac: Buffer):
"""Verify a tag for the current MAC state. """Verify that the data entered so far matches the given MAC tag.
Unlike the C library, this method does not alter the current state.
Args: Args:
mac: The tag to verify (16 or 32 bytes). mac: The tag to verify against (16 or 32 bytes).
Returns:
Only if verification succeeds.
Raises: Raises:
TypeError: If tag length is invalid. TypeError: If tag length is invalid.
@@ -541,7 +541,9 @@ class Mac:
maclen = len(mac) maclen = len(mac)
if maclen not in (16, 32): if maclen not in (16, 32):
raise TypeError("mac length must be 16 or 32") raise TypeError("mac length must be 16 or 32")
rc = _lib.aegis128l_mac_verify(self._st, _ptr(mac), maclen)
cloned = self.clone()
rc = _lib.aegis128l_mac_verify(cloned._st, _ptr(mac), maclen)
if rc != 0: if rc != 0:
raise ValueError("mac verification failed") raise ValueError("mac verification failed")
@@ -550,23 +552,31 @@ class Encryptor:
"""Incremental encryptor. """Incremental encryptor.
- update(message[, into]) -> returns produced ciphertext bytes - update(message[, into]) -> returns produced ciphertext bytes
- final([into], maclen=16) -> returns tail+tag bytes - final([into]) -> returns MAC tag
- final_detached([ct_into], [mac_into], maclen=16) -> returns (tail_bytes, mac)
""" """
__slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out") __slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out", "_maclen")
def __init__(self, key: Buffer, nonce: Buffer, ad: Buffer | None = None): def __init__(
f"""Create an incremental encryptor. self,
key: Buffer,
nonce: Buffer,
ad: Buffer | None = None,
maclen: int = MACBYTES,
):
"""Create an incremental encryptor.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
ad: Associated data to bind to the encryption (optional). ad: Associated data to bind to the encryption (optional).
maclen: MAC length (16 or 32, default 16).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
if maclen not in (16, 32):
raise TypeError("maclen must be 16 or 32")
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
@@ -583,6 +593,7 @@ class Encryptor:
self._owner = owner self._owner = owner
self._bytes_in = 0 self._bytes_in = 0
self._bytes_out = 0 self._bytes_out = 0
self._maclen = maclen
@property @property
def bytes_in(self) -> int: def bytes_in(self) -> int:
@@ -611,8 +622,10 @@ class Encryptor:
Raises: Raises:
TypeError: If destination buffer is too small. TypeError: If destination buffer is too small.
RuntimeError: If the C update call fails. RuntimeError: If the C update call fails or if called after final().
""" """
if self._st is None:
raise RuntimeError("Cannot call update() after final()")
expected_out = len(message) expected_out = len(message)
out = into if into is not None else bytearray(expected_out) out = into if into is not None else bytearray(expected_out)
out_mv = memoryview(out) out_mv = memoryview(out)
@@ -641,24 +654,21 @@ class Encryptor:
self._bytes_out += w self._bytes_out += w
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:w] # type: ignore
def final( def final(self, into: Buffer | None = None) -> bytearray | memoryview:
self, into: Buffer | None = None, maclen: int = MACBYTES """Finalize encryption and return the authentication tag.
) -> bytearray | memoryview:
"""Finalize encryption, writing any remaining bytes and the tag.
Args: Args:
into: Optional destination buffer for the tail and tag. into: Optional destination buffer for the tag.
maclen: Tag length (16 or 32). Defaults to 16.
Returns: Returns:
A memoryview of the produced bytes (tail + tag) if into provided, bytearray slice otherwise. The authentication tag as bytearray if into not provided, memoryview of into otherwise.
Raises: Raises:
TypeError: If maclen is invalid. RuntimeError: If the C final call fails or if called after final().
RuntimeError: If the C final call fails.
""" """
if maclen not in (16, 32): if self._st is None:
raise TypeError("maclen must be 16 or 32") raise RuntimeError("Cannot call final() after final()")
maclen = self._maclen
# Only the authentication tag is produced here; allocate exactly maclen # Only the authentication tag is produced here; allocate exactly maclen
out = into if into is not None else bytearray(maclen) out = into if into is not None else bytearray(maclen)
written = ffi.new("size_t *") written = ffi.new("size_t *")
@@ -678,6 +688,8 @@ class Encryptor:
# Only the tag bytes are returned when we allocate the buffer # Only the tag bytes are returned when we allocate the buffer
assert w == maclen assert w == maclen
self._bytes_out += w self._bytes_out += w
self._st = None
self._owner = None
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:w] # type: ignore
@@ -688,19 +700,28 @@ class Decryptor:
- final(mac) -> verifies the MAC tag - final(mac) -> verifies the MAC tag
""" """
__slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out") __slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out", "_maclen")
def __init__(self, key: Buffer, nonce: Buffer, ad: Buffer | None = None): def __init__(
f"""Create an incremental decryptor for detached tags. self,
key: Buffer,
nonce: Buffer,
ad: Buffer | None = None,
maclen: int = MACBYTES,
):
"""Create an incremental decryptor for detached tags.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ad: Associated data used during encryption (optional). ad: Associated data used during encryption (optional).
maclen: MAC length (16 or 32, default 16).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
if maclen not in (16, 32):
raise TypeError("maclen must be 16 or 32")
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
@@ -717,6 +738,7 @@ class Decryptor:
self._owner = owner self._owner = owner
self._bytes_in = 0 self._bytes_in = 0
self._bytes_out = 0 self._bytes_out = 0
self._maclen = maclen
@property @property
def bytes_in(self) -> int: def bytes_in(self) -> int:
@@ -740,8 +762,10 @@ class Decryptor:
Raises: Raises:
TypeError: If destination buffer is too small. TypeError: If destination buffer is too small.
RuntimeError: If the C update call fails. RuntimeError: If the C update call fails or if called after final().
""" """
if self._st is None:
raise RuntimeError("Cannot call update() after final()")
expected_out = len(ct) expected_out = len(ct)
out = into if into is not None else bytearray(expected_out) out = into if into is not None else bytearray(expected_out)
out_mv = memoryview(out) out_mv = memoryview(out)
@@ -770,20 +794,25 @@ class Decryptor:
"""Finalize decryption by verifying the MAC tag. """Finalize decryption by verifying the MAC tag.
Args: Args:
mac: Tag to verify (16 or 32 bytes). mac: Tag to verify.
Raises: Raises:
TypeError: If tag length is invalid. TypeError: If tag length doesn't match the expected maclen.
ValueError: If authentication fails. ValueError: If authentication fails.
RuntimeError: If called after final().
""" """
maclen = len(mac) if self._st is None:
if maclen not in (16, 32): raise RuntimeError("Cannot call final() after final()")
raise TypeError("mac length must be 16 or 32") maclen = self._maclen
if len(mac) != maclen:
raise TypeError(f"mac length must be {maclen}")
rc = _lib.aegis128l_state_decrypt_detached_final( rc = _lib.aegis128l_state_decrypt_detached_final(
self._st, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen self._st, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen
) )
if rc != 0: if rc != 0:
raise ValueError("authentication failed") raise ValueError("authentication failed")
self._st = None
self._owner = None
def new_state(): def new_state():
pyaegis/aegis128x2.py
+142 -113
View File
@@ -9,7 +9,7 @@ from ._loader import ffi
from ._loader import lib as _lib from ._loader import lib as _lib
from .util import Buffer, new_aligned_struct, nonce_increment, wipe from .util import Buffer, new_aligned_struct, nonce_increment, wipe
NAME = "AEGIS-128X2" #: Algorithm name NAME = "AEGIS-128X2" #: Algorithm display name
KEYBYTES = 16 #: Key size in bytes (varies by algorithm) KEYBYTES = 16 #: Key size in bytes (varies by algorithm)
NONCEBYTES = 16 #: Nonce size in bytes (varies by algorithm) NONCEBYTES = 16 #: Nonce size in bytes (varies by algorithm)
MACBYTES = 16 #: Normal MAC size (always 16) MACBYTES = 16 #: Normal MAC size (always 16)
@@ -20,15 +20,22 @@ RATE = 64 #: Byte chunk size in internal processing
def random_key() -> bytearray: def random_key() -> bytearray:
""" """
Generate a random key using cryptographically secure random bytes. Generate a secret key using cryptographically secure random bytes.
It is recommended to wipe() the key after no longer needed to keep it secret. It is recommended to wipe() the key after no longer needed.
""" """
return bytearray(secrets.token_bytes(KEYBYTES)) return bytearray(secrets.token_bytes(KEYBYTES))
def random_nonce() -> bytearray: def random_nonce() -> bytearray:
"""Generate a random nonce using cryptographically secure random bytes.""" """
Generate a public nonce using cryptographically secure random bytes.
Nonces (a number used once) are public data that may be sent together
with the ciphertext, but they need to be unique for each use.
See also: nonce_increment() can be used to derive sequential nonces.
"""
return bytearray(secrets.token_bytes(NONCEBYTES)) return bytearray(secrets.token_bytes(NONCEBYTES))
@@ -46,11 +53,11 @@ def encrypt_detached(
ct_into: Buffer | None = None, ct_into: Buffer | None = None,
mac_into: Buffer | None = None, mac_into: Buffer | None = None,
) -> tuple[bytearray | memoryview, bytearray | memoryview]: ) -> tuple[bytearray | memoryview, bytearray | memoryview]:
f"""Encrypt message with associated data, returning ciphertext and MAC separately. """Encrypt message with associated data, returning ciphertext and MAC separately.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -114,11 +121,11 @@ def decrypt_detached(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext with detached MAC and associated data. """Decrypt ciphertext with detached MAC and associated data.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext to decrypt. ct: The ciphertext to decrypt.
mac: The MAC to verify. mac: The MAC to verify.
ad: Associated data (optional). ad: Associated data (optional).
@@ -170,11 +177,11 @@ def encrypt(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Encrypt message with associated data, returning ciphertext with appended MAC. """Encrypt message with associated data, returning ciphertext with appended MAC.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -226,11 +233,11 @@ def decrypt(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext with appended MAC and associated data. """Decrypt ciphertext with appended MAC and associated data.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext with MAC to decrypt. ct: The ciphertext with MAC to decrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -283,11 +290,11 @@ def stream(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | Buffer: ) -> bytearray | Buffer:
f"""Generate a stream of pseudorandom bytes. """Generate a stream of pseudorandom bytes.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}, uses zeroes for nonce if None). nonce: Public nonce (generate with random_nonce(), uses zeroes for nonce if None).
length: Number of bytes to generate (required if into is None). length: Number of bytes to generate (required if into is None).
into: Buffer to write stream into (default: bytearray created). into: Buffer to write stream into (default: bytearray created).
@@ -325,11 +332,11 @@ def encrypt_unauthenticated(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Encrypt message without authentication (for testing/debugging). """Encrypt message without authentication (for testing/debugging).
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
into: Buffer to write ciphertext into (default: bytearray created). into: Buffer to write ciphertext into (default: bytearray created).
@@ -366,11 +373,11 @@ def decrypt_unauthenticated(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext without authentication (for testing/debugging). """Decrypt ciphertext without authentication (for testing/debugging).
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext to decrypt. ct: The ciphertext to decrypt.
into: Buffer to write plaintext into (default: bytearray created). into: Buffer to write plaintext into (default: bytearray created).
@@ -408,11 +415,11 @@ def mac(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Compute a MAC for the given data in one shot. """Compute a MAC for the given data in one shot.
Args: Args:
key: Key ({KEYBYTES=}) key: Secret key (generate with random_key())
nonce: Nonce ({NONCEBYTES=}) nonce: Public nonce (generate with random_nonce())
data: Data to MAC data: Data to MAC
maclen: MAC length (16 or 32, default 16) maclen: MAC length (16 or 32, default 16)
into: Buffer to write MAC into (default: bytearray created) into: Buffer to write MAC into (default: bytearray created)
@@ -420,70 +427,59 @@ def mac(
Returns: Returns:
MAC bytes as bytearray if into not provided, memoryview of into otherwise MAC bytes as bytearray if into not provided, memoryview of into otherwise
""" """
mac_state = Mac(key, nonce) mac_state = Mac(key, nonce, maclen)
mac_state.update(data) mac_state.update(data)
return mac_state.final(maclen, into) return mac_state.final(into)
class Mac: class Mac:
"""AEGIS-128X2 MAC state wrapper. """AEGIS-128X2 MAC state wrapper.
Usage: Example:
mac = Mac(key, nonce) a = Mac(key, nonce)
mac.update(data) a.update(data)
tag = mac.final() # defaults to 16-byte MAC mac = a.final()
# or verify:
mac2 = Mac(key, nonce); mac2.update(data); mac2.verify(tag)
""" """
__slots__ = ("_st", "_owner") __slots__ = ("_st", "_owner", "_maclen")
def __init__( def __init__(self, key: Buffer, nonce: Buffer, maclen: int = MACBYTES) -> None:
self, """Create a MAC with the given key, nonce, and tag length.
key: Buffer,
nonce: Buffer,
_other=None,
) -> None:
f"""Initialize a MAC state with a nonce and key.
Args:
key: Key ({KEYBYTES=}).
nonce: Nonce ({NONCEBYTES=}).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
st, owner = new_aligned_struct("aegis128x2_mac_state", ALIGNMENT) if maclen not in (16, 32):
self._st = st raise TypeError("maclen must be 16 or 32")
self._owner = owner
if _other is not None: # clone path
_lib.aegis128x2_mac_state_clone(self._st, _other._st)
return
# Normal init path
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
raise TypeError(f"nonce length must be {NONCEBYTES}") raise TypeError(f"nonce length must be {NONCEBYTES}")
self._maclen = maclen
st, owner = new_aligned_struct("aegis128x2_mac_state", ALIGNMENT)
self._st = st
self._owner = owner
_lib.aegis128x2_mac_init(self._st, _ptr(key), _ptr(nonce)) _lib.aegis128x2_mac_init(self._st, _ptr(key), _ptr(nonce))
def __deepcopy__(self) -> "Mac": def __deepcopy__(self) -> "Mac":
"""Return a clone of current MAC state.""" """Return a clone of current MAC state."""
return Mac(b"", b"", _other=self) clone = object.__new__(Mac)
clone._maclen = self._maclen
clone._st, clone._owner = new_aligned_struct("aegis128x2_mac_state", ALIGNMENT)
_lib.aegis128x2_mac_state_clone(clone._st, self._st)
return clone
clone = __deepcopy__ clone = __deepcopy__
def reset(self) -> None: def reset(self) -> None:
"""Reset the MAC state so it can be reused with the same nonce and key.""" """Reset back to the original state, prior to any updates."""
_lib.aegis128x2_mac_reset(self._st) _lib.aegis128x2_mac_reset(self._st)
def update(self, data: Buffer) -> None: def update(self, data: Buffer) -> None:
"""Absorb data into the MAC state. """Update the MAC state with more data.
Args: Repeated calls to update() are equivalent to a single call with the concatenated data.
data: Bytes-like object to authenticate.
Raises:
RuntimeError: If the underlying C function reports an error.
""" """
rc = _lib.aegis128x2_mac_update(self._st, _ptr(data), len(data)) rc = _lib.aegis128x2_mac_update(self._st, _ptr(data), len(data))
if rc != 0: if rc != 0:
@@ -491,15 +487,13 @@ class Mac:
err_name = errno.errorcode.get(err_num, f"errno_{err_num}") err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
raise RuntimeError(f"mac update failed: {err_name}") raise RuntimeError(f"mac update failed: {err_name}")
def final( def final(self, into: Buffer | None = None) -> bytearray | memoryview:
self, """Calculate and return the MAC tag for the currently input data.
maclen: int = MACBYTES,
into: Buffer | None = None, Unlike the C library, this method does not alter the current state,
) -> bytearray | memoryview: allowing for multiple calls and further updates on the same object.
"""Finalize and return the MAC tag.
Args: Args:
maclen: Tag length in bytes (16 or 32). Defaults to 16.
into: Optional buffer to write the tag into (default: bytearray created). into: Optional buffer to write the tag into (default: bytearray created).
Returns: Returns:
@@ -509,30 +503,36 @@ class Mac:
TypeError: If lengths are invalid. TypeError: If lengths are invalid.
RuntimeError: If finalization fails in the C library. RuntimeError: If finalization fails in the C library.
""" """
if maclen not in (16, 32): maclen = self._maclen
raise TypeError("maclen must be 16 or 32")
if into is None: if into is None:
out = bytearray(maclen) out = bytearray(maclen)
else: else:
if len(into) < maclen: if len(into) < maclen:
raise TypeError("into length must be at least maclen") raise TypeError("into length must be at least maclen")
out = into out = into
out_mv = memoryview(out)
rc = _lib.aegis128x2_mac_final(self._st, ffi.from_buffer(out_mv), maclen) rc = _lib.aegis128x2_mac_final(self.clone()._st, ffi.from_buffer(out), maclen)
if rc != 0: if rc != 0:
err_num = ffi.errno err_num = ffi.errno
err_name = errno.errorcode.get(err_num, f"errno_{err_num}") err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
raise RuntimeError(f"mac final failed: {err_name}") raise RuntimeError(f"mac final failed: {err_name}")
return out if into is None else memoryview(out)[:maclen] # type: ignore return out if into is None else memoryview(out)[:maclen] # type: ignore
def digest(self) -> bytes:
"""Calculate and return the MAC tag as bytes."""
return bytes(self.final())
def hexdigest(self) -> str:
"""Calculate and return the MAC tag as a hex string."""
return self.digest().hex()
def verify(self, mac: Buffer): def verify(self, mac: Buffer):
"""Verify a tag for the current MAC state. """Verify that the data entered so far matches the given MAC tag.
Unlike the C library, this method does not alter the current state.
Args: Args:
mac: The tag to verify (16 or 32 bytes). mac: The tag to verify against (16 or 32 bytes).
Returns:
Only if verification succeeds.
Raises: Raises:
TypeError: If tag length is invalid. TypeError: If tag length is invalid.
@@ -541,7 +541,9 @@ class Mac:
maclen = len(mac) maclen = len(mac)
if maclen not in (16, 32): if maclen not in (16, 32):
raise TypeError("mac length must be 16 or 32") raise TypeError("mac length must be 16 or 32")
rc = _lib.aegis128x2_mac_verify(self._st, _ptr(mac), maclen)
cloned = self.clone()
rc = _lib.aegis128x2_mac_verify(cloned._st, _ptr(mac), maclen)
if rc != 0: if rc != 0:
raise ValueError("mac verification failed") raise ValueError("mac verification failed")
@@ -550,23 +552,31 @@ class Encryptor:
"""Incremental encryptor. """Incremental encryptor.
- update(message[, into]) -> returns produced ciphertext bytes - update(message[, into]) -> returns produced ciphertext bytes
- final([into], maclen=16) -> returns tail+tag bytes - final([into]) -> returns MAC tag
- final_detached([ct_into], [mac_into], maclen=16) -> returns (tail_bytes, mac)
""" """
__slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out") __slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out", "_maclen")
def __init__(self, key: Buffer, nonce: Buffer, ad: Buffer | None = None): def __init__(
f"""Create an incremental encryptor. self,
key: Buffer,
nonce: Buffer,
ad: Buffer | None = None,
maclen: int = MACBYTES,
):
"""Create an incremental encryptor.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
ad: Associated data to bind to the encryption (optional). ad: Associated data to bind to the encryption (optional).
maclen: MAC length (16 or 32, default 16).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
if maclen not in (16, 32):
raise TypeError("maclen must be 16 or 32")
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
@@ -583,6 +593,7 @@ class Encryptor:
self._owner = owner self._owner = owner
self._bytes_in = 0 self._bytes_in = 0
self._bytes_out = 0 self._bytes_out = 0
self._maclen = maclen
@property @property
def bytes_in(self) -> int: def bytes_in(self) -> int:
@@ -611,8 +622,10 @@ class Encryptor:
Raises: Raises:
TypeError: If destination buffer is too small. TypeError: If destination buffer is too small.
RuntimeError: If the C update call fails. RuntimeError: If the C update call fails or if called after final().
""" """
if self._st is None:
raise RuntimeError("Cannot call update() after final()")
expected_out = len(message) expected_out = len(message)
out = into if into is not None else bytearray(expected_out) out = into if into is not None else bytearray(expected_out)
out_mv = memoryview(out) out_mv = memoryview(out)
@@ -641,24 +654,21 @@ class Encryptor:
self._bytes_out += w self._bytes_out += w
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:w] # type: ignore
def final( def final(self, into: Buffer | None = None) -> bytearray | memoryview:
self, into: Buffer | None = None, maclen: int = MACBYTES """Finalize encryption and return the authentication tag.
) -> bytearray | memoryview:
"""Finalize encryption, writing any remaining bytes and the tag.
Args: Args:
into: Optional destination buffer for the tail and tag. into: Optional destination buffer for the tag.
maclen: Tag length (16 or 32). Defaults to 16.
Returns: Returns:
A memoryview of the produced bytes (tail + tag) if into provided, bytearray slice otherwise. The authentication tag as bytearray if into not provided, memoryview of into otherwise.
Raises: Raises:
TypeError: If maclen is invalid. RuntimeError: If the C final call fails or if called after final().
RuntimeError: If the C final call fails.
""" """
if maclen not in (16, 32): if self._st is None:
raise TypeError("maclen must be 16 or 32") raise RuntimeError("Cannot call final() after final()")
maclen = self._maclen
# Only the authentication tag is produced here; allocate exactly maclen # Only the authentication tag is produced here; allocate exactly maclen
out = into if into is not None else bytearray(maclen) out = into if into is not None else bytearray(maclen)
written = ffi.new("size_t *") written = ffi.new("size_t *")
@@ -678,6 +688,8 @@ class Encryptor:
# Only the tag bytes are returned when we allocate the buffer # Only the tag bytes are returned when we allocate the buffer
assert w == maclen assert w == maclen
self._bytes_out += w self._bytes_out += w
self._st = None
self._owner = None
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:w] # type: ignore
@@ -688,19 +700,28 @@ class Decryptor:
- final(mac) -> verifies the MAC tag - final(mac) -> verifies the MAC tag
""" """
__slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out") __slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out", "_maclen")
def __init__(self, key: Buffer, nonce: Buffer, ad: Buffer | None = None): def __init__(
f"""Create an incremental decryptor for detached tags. self,
key: Buffer,
nonce: Buffer,
ad: Buffer | None = None,
maclen: int = MACBYTES,
):
"""Create an incremental decryptor for detached tags.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ad: Associated data used during encryption (optional). ad: Associated data used during encryption (optional).
maclen: MAC length (16 or 32, default 16).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
if maclen not in (16, 32):
raise TypeError("maclen must be 16 or 32")
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
@@ -717,6 +738,7 @@ class Decryptor:
self._owner = owner self._owner = owner
self._bytes_in = 0 self._bytes_in = 0
self._bytes_out = 0 self._bytes_out = 0
self._maclen = maclen
@property @property
def bytes_in(self) -> int: def bytes_in(self) -> int:
@@ -740,8 +762,10 @@ class Decryptor:
Raises: Raises:
TypeError: If destination buffer is too small. TypeError: If destination buffer is too small.
RuntimeError: If the C update call fails. RuntimeError: If the C update call fails or if called after final().
""" """
if self._st is None:
raise RuntimeError("Cannot call update() after final()")
expected_out = len(ct) expected_out = len(ct)
out = into if into is not None else bytearray(expected_out) out = into if into is not None else bytearray(expected_out)
out_mv = memoryview(out) out_mv = memoryview(out)
@@ -770,20 +794,25 @@ class Decryptor:
"""Finalize decryption by verifying the MAC tag. """Finalize decryption by verifying the MAC tag.
Args: Args:
mac: Tag to verify (16 or 32 bytes). mac: Tag to verify.
Raises: Raises:
TypeError: If tag length is invalid. TypeError: If tag length doesn't match the expected maclen.
ValueError: If authentication fails. ValueError: If authentication fails.
RuntimeError: If called after final().
""" """
maclen = len(mac) if self._st is None:
if maclen not in (16, 32): raise RuntimeError("Cannot call final() after final()")
raise TypeError("mac length must be 16 or 32") maclen = self._maclen
if len(mac) != maclen:
raise TypeError(f"mac length must be {maclen}")
rc = _lib.aegis128x2_state_decrypt_detached_final( rc = _lib.aegis128x2_state_decrypt_detached_final(
self._st, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen self._st, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen
) )
if rc != 0: if rc != 0:
raise ValueError("authentication failed") raise ValueError("authentication failed")
self._st = None
self._owner = None
def new_state(): def new_state():
pyaegis/aegis128x4.py
+142 -113
View File
@@ -9,7 +9,7 @@ from ._loader import ffi
from ._loader import lib as _lib from ._loader import lib as _lib
from .util import Buffer, new_aligned_struct, nonce_increment, wipe from .util import Buffer, new_aligned_struct, nonce_increment, wipe
NAME = "AEGIS-128X4" #: Algorithm name NAME = "AEGIS-128X4" #: Algorithm display name
KEYBYTES = 16 #: Key size in bytes (varies by algorithm) KEYBYTES = 16 #: Key size in bytes (varies by algorithm)
NONCEBYTES = 16 #: Nonce size in bytes (varies by algorithm) NONCEBYTES = 16 #: Nonce size in bytes (varies by algorithm)
MACBYTES = 16 #: Normal MAC size (always 16) MACBYTES = 16 #: Normal MAC size (always 16)
@@ -20,15 +20,22 @@ RATE = 64 #: Byte chunk size in internal processing
def random_key() -> bytearray: def random_key() -> bytearray:
""" """
Generate a random key using cryptographically secure random bytes. Generate a secret key using cryptographically secure random bytes.
It is recommended to wipe() the key after no longer needed to keep it secret. It is recommended to wipe() the key after no longer needed.
""" """
return bytearray(secrets.token_bytes(KEYBYTES)) return bytearray(secrets.token_bytes(KEYBYTES))
def random_nonce() -> bytearray: def random_nonce() -> bytearray:
"""Generate a random nonce using cryptographically secure random bytes.""" """
Generate a public nonce using cryptographically secure random bytes.
Nonces (a number used once) are public data that may be sent together
with the ciphertext, but they need to be unique for each use.
See also: nonce_increment() can be used to derive sequential nonces.
"""
return bytearray(secrets.token_bytes(NONCEBYTES)) return bytearray(secrets.token_bytes(NONCEBYTES))
@@ -46,11 +53,11 @@ def encrypt_detached(
ct_into: Buffer | None = None, ct_into: Buffer | None = None,
mac_into: Buffer | None = None, mac_into: Buffer | None = None,
) -> tuple[bytearray | memoryview, bytearray | memoryview]: ) -> tuple[bytearray | memoryview, bytearray | memoryview]:
f"""Encrypt message with associated data, returning ciphertext and MAC separately. """Encrypt message with associated data, returning ciphertext and MAC separately.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -114,11 +121,11 @@ def decrypt_detached(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext with detached MAC and associated data. """Decrypt ciphertext with detached MAC and associated data.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext to decrypt. ct: The ciphertext to decrypt.
mac: The MAC to verify. mac: The MAC to verify.
ad: Associated data (optional). ad: Associated data (optional).
@@ -170,11 +177,11 @@ def encrypt(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Encrypt message with associated data, returning ciphertext with appended MAC. """Encrypt message with associated data, returning ciphertext with appended MAC.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -226,11 +233,11 @@ def decrypt(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext with appended MAC and associated data. """Decrypt ciphertext with appended MAC and associated data.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext with MAC to decrypt. ct: The ciphertext with MAC to decrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -283,11 +290,11 @@ def stream(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | Buffer: ) -> bytearray | Buffer:
f"""Generate a stream of pseudorandom bytes. """Generate a stream of pseudorandom bytes.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}, uses zeroes for nonce if None). nonce: Public nonce (generate with random_nonce(), uses zeroes for nonce if None).
length: Number of bytes to generate (required if into is None). length: Number of bytes to generate (required if into is None).
into: Buffer to write stream into (default: bytearray created). into: Buffer to write stream into (default: bytearray created).
@@ -325,11 +332,11 @@ def encrypt_unauthenticated(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Encrypt message without authentication (for testing/debugging). """Encrypt message without authentication (for testing/debugging).
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
into: Buffer to write ciphertext into (default: bytearray created). into: Buffer to write ciphertext into (default: bytearray created).
@@ -366,11 +373,11 @@ def decrypt_unauthenticated(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext without authentication (for testing/debugging). """Decrypt ciphertext without authentication (for testing/debugging).
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext to decrypt. ct: The ciphertext to decrypt.
into: Buffer to write plaintext into (default: bytearray created). into: Buffer to write plaintext into (default: bytearray created).
@@ -408,11 +415,11 @@ def mac(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Compute a MAC for the given data in one shot. """Compute a MAC for the given data in one shot.
Args: Args:
key: Key ({KEYBYTES=}) key: Secret key (generate with random_key())
nonce: Nonce ({NONCEBYTES=}) nonce: Public nonce (generate with random_nonce())
data: Data to MAC data: Data to MAC
maclen: MAC length (16 or 32, default 16) maclen: MAC length (16 or 32, default 16)
into: Buffer to write MAC into (default: bytearray created) into: Buffer to write MAC into (default: bytearray created)
@@ -420,70 +427,59 @@ def mac(
Returns: Returns:
MAC bytes as bytearray if into not provided, memoryview of into otherwise MAC bytes as bytearray if into not provided, memoryview of into otherwise
""" """
mac_state = Mac(key, nonce) mac_state = Mac(key, nonce, maclen)
mac_state.update(data) mac_state.update(data)
return mac_state.final(maclen, into) return mac_state.final(into)
class Mac: class Mac:
"""AEGIS-128X4 MAC state wrapper. """AEGIS-128X4 MAC state wrapper.
Usage: Example:
mac = Mac(key, nonce) a = Mac(key, nonce)
mac.update(data) a.update(data)
tag = mac.final() # defaults to 16-byte MAC mac = a.final()
# or verify:
mac2 = Mac(key, nonce); mac2.update(data); mac2.verify(tag)
""" """
__slots__ = ("_st", "_owner") __slots__ = ("_st", "_owner", "_maclen")
def __init__( def __init__(self, key: Buffer, nonce: Buffer, maclen: int = MACBYTES) -> None:
self, """Create a MAC with the given key, nonce, and tag length.
key: Buffer,
nonce: Buffer,
_other=None,
) -> None:
f"""Initialize a MAC state with a nonce and key.
Args:
key: Key ({KEYBYTES=}).
nonce: Nonce ({NONCEBYTES=}).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
st, owner = new_aligned_struct("aegis128x4_mac_state", ALIGNMENT) if maclen not in (16, 32):
self._st = st raise TypeError("maclen must be 16 or 32")
self._owner = owner
if _other is not None: # clone path
_lib.aegis128x4_mac_state_clone(self._st, _other._st)
return
# Normal init path
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
raise TypeError(f"nonce length must be {NONCEBYTES}") raise TypeError(f"nonce length must be {NONCEBYTES}")
self._maclen = maclen
st, owner = new_aligned_struct("aegis128x4_mac_state", ALIGNMENT)
self._st = st
self._owner = owner
_lib.aegis128x4_mac_init(self._st, _ptr(key), _ptr(nonce)) _lib.aegis128x4_mac_init(self._st, _ptr(key), _ptr(nonce))
def __deepcopy__(self) -> "Mac": def __deepcopy__(self) -> "Mac":
"""Return a clone of current MAC state.""" """Return a clone of current MAC state."""
return Mac(b"", b"", _other=self) clone = object.__new__(Mac)
clone._maclen = self._maclen
clone._st, clone._owner = new_aligned_struct("aegis128x4_mac_state", ALIGNMENT)
_lib.aegis128x4_mac_state_clone(clone._st, self._st)
return clone
clone = __deepcopy__ clone = __deepcopy__
def reset(self) -> None: def reset(self) -> None:
"""Reset the MAC state so it can be reused with the same nonce and key.""" """Reset back to the original state, prior to any updates."""
_lib.aegis128x4_mac_reset(self._st) _lib.aegis128x4_mac_reset(self._st)
def update(self, data: Buffer) -> None: def update(self, data: Buffer) -> None:
"""Absorb data into the MAC state. """Update the MAC state with more data.
Args: Repeated calls to update() are equivalent to a single call with the concatenated data.
data: Bytes-like object to authenticate.
Raises:
RuntimeError: If the underlying C function reports an error.
""" """
rc = _lib.aegis128x4_mac_update(self._st, _ptr(data), len(data)) rc = _lib.aegis128x4_mac_update(self._st, _ptr(data), len(data))
if rc != 0: if rc != 0:
@@ -491,15 +487,13 @@ class Mac:
err_name = errno.errorcode.get(err_num, f"errno_{err_num}") err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
raise RuntimeError(f"mac update failed: {err_name}") raise RuntimeError(f"mac update failed: {err_name}")
def final( def final(self, into: Buffer | None = None) -> bytearray | memoryview:
self, """Calculate and return the MAC tag for the currently input data.
maclen: int = MACBYTES,
into: Buffer | None = None, Unlike the C library, this method does not alter the current state,
) -> bytearray | memoryview: allowing for multiple calls and further updates on the same object.
"""Finalize and return the MAC tag.
Args: Args:
maclen: Tag length in bytes (16 or 32). Defaults to 16.
into: Optional buffer to write the tag into (default: bytearray created). into: Optional buffer to write the tag into (default: bytearray created).
Returns: Returns:
@@ -509,30 +503,36 @@ class Mac:
TypeError: If lengths are invalid. TypeError: If lengths are invalid.
RuntimeError: If finalization fails in the C library. RuntimeError: If finalization fails in the C library.
""" """
if maclen not in (16, 32): maclen = self._maclen
raise TypeError("maclen must be 16 or 32")
if into is None: if into is None:
out = bytearray(maclen) out = bytearray(maclen)
else: else:
if len(into) < maclen: if len(into) < maclen:
raise TypeError("into length must be at least maclen") raise TypeError("into length must be at least maclen")
out = into out = into
out_mv = memoryview(out)
rc = _lib.aegis128x4_mac_final(self._st, ffi.from_buffer(out_mv), maclen) rc = _lib.aegis128x4_mac_final(self.clone()._st, ffi.from_buffer(out), maclen)
if rc != 0: if rc != 0:
err_num = ffi.errno err_num = ffi.errno
err_name = errno.errorcode.get(err_num, f"errno_{err_num}") err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
raise RuntimeError(f"mac final failed: {err_name}") raise RuntimeError(f"mac final failed: {err_name}")
return out if into is None else memoryview(out)[:maclen] # type: ignore return out if into is None else memoryview(out)[:maclen] # type: ignore
def digest(self) -> bytes:
"""Calculate and return the MAC tag as bytes."""
return bytes(self.final())
def hexdigest(self) -> str:
"""Calculate and return the MAC tag as a hex string."""
return self.digest().hex()
def verify(self, mac: Buffer): def verify(self, mac: Buffer):
"""Verify a tag for the current MAC state. """Verify that the data entered so far matches the given MAC tag.
Unlike the C library, this method does not alter the current state.
Args: Args:
mac: The tag to verify (16 or 32 bytes). mac: The tag to verify against (16 or 32 bytes).
Returns:
Only if verification succeeds.
Raises: Raises:
TypeError: If tag length is invalid. TypeError: If tag length is invalid.
@@ -541,7 +541,9 @@ class Mac:
maclen = len(mac) maclen = len(mac)
if maclen not in (16, 32): if maclen not in (16, 32):
raise TypeError("mac length must be 16 or 32") raise TypeError("mac length must be 16 or 32")
rc = _lib.aegis128x4_mac_verify(self._st, _ptr(mac), maclen)
cloned = self.clone()
rc = _lib.aegis128x4_mac_verify(cloned._st, _ptr(mac), maclen)
if rc != 0: if rc != 0:
raise ValueError("mac verification failed") raise ValueError("mac verification failed")
@@ -550,23 +552,31 @@ class Encryptor:
"""Incremental encryptor. """Incremental encryptor.
- update(message[, into]) -> returns produced ciphertext bytes - update(message[, into]) -> returns produced ciphertext bytes
- final([into], maclen=16) -> returns tail+tag bytes - final([into]) -> returns MAC tag
- final_detached([ct_into], [mac_into], maclen=16) -> returns (tail_bytes, mac)
""" """
__slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out") __slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out", "_maclen")
def __init__(self, key: Buffer, nonce: Buffer, ad: Buffer | None = None): def __init__(
f"""Create an incremental encryptor. self,
key: Buffer,
nonce: Buffer,
ad: Buffer | None = None,
maclen: int = MACBYTES,
):
"""Create an incremental encryptor.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
ad: Associated data to bind to the encryption (optional). ad: Associated data to bind to the encryption (optional).
maclen: MAC length (16 or 32, default 16).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
if maclen not in (16, 32):
raise TypeError("maclen must be 16 or 32")
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
@@ -583,6 +593,7 @@ class Encryptor:
self._owner = owner self._owner = owner
self._bytes_in = 0 self._bytes_in = 0
self._bytes_out = 0 self._bytes_out = 0
self._maclen = maclen
@property @property
def bytes_in(self) -> int: def bytes_in(self) -> int:
@@ -611,8 +622,10 @@ class Encryptor:
Raises: Raises:
TypeError: If destination buffer is too small. TypeError: If destination buffer is too small.
RuntimeError: If the C update call fails. RuntimeError: If the C update call fails or if called after final().
""" """
if self._st is None:
raise RuntimeError("Cannot call update() after final()")
expected_out = len(message) expected_out = len(message)
out = into if into is not None else bytearray(expected_out) out = into if into is not None else bytearray(expected_out)
out_mv = memoryview(out) out_mv = memoryview(out)
@@ -641,24 +654,21 @@ class Encryptor:
self._bytes_out += w self._bytes_out += w
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:w] # type: ignore
def final( def final(self, into: Buffer | None = None) -> bytearray | memoryview:
self, into: Buffer | None = None, maclen: int = MACBYTES """Finalize encryption and return the authentication tag.
) -> bytearray | memoryview:
"""Finalize encryption, writing any remaining bytes and the tag.
Args: Args:
into: Optional destination buffer for the tail and tag. into: Optional destination buffer for the tag.
maclen: Tag length (16 or 32). Defaults to 16.
Returns: Returns:
A memoryview of the produced bytes (tail + tag) if into provided, bytearray slice otherwise. The authentication tag as bytearray if into not provided, memoryview of into otherwise.
Raises: Raises:
TypeError: If maclen is invalid. RuntimeError: If the C final call fails or if called after final().
RuntimeError: If the C final call fails.
""" """
if maclen not in (16, 32): if self._st is None:
raise TypeError("maclen must be 16 or 32") raise RuntimeError("Cannot call final() after final()")
maclen = self._maclen
# Only the authentication tag is produced here; allocate exactly maclen # Only the authentication tag is produced here; allocate exactly maclen
out = into if into is not None else bytearray(maclen) out = into if into is not None else bytearray(maclen)
written = ffi.new("size_t *") written = ffi.new("size_t *")
@@ -678,6 +688,8 @@ class Encryptor:
# Only the tag bytes are returned when we allocate the buffer # Only the tag bytes are returned when we allocate the buffer
assert w == maclen assert w == maclen
self._bytes_out += w self._bytes_out += w
self._st = None
self._owner = None
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:w] # type: ignore
@@ -688,19 +700,28 @@ class Decryptor:
- final(mac) -> verifies the MAC tag - final(mac) -> verifies the MAC tag
""" """
__slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out") __slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out", "_maclen")
def __init__(self, key: Buffer, nonce: Buffer, ad: Buffer | None = None): def __init__(
f"""Create an incremental decryptor for detached tags. self,
key: Buffer,
nonce: Buffer,
ad: Buffer | None = None,
maclen: int = MACBYTES,
):
"""Create an incremental decryptor for detached tags.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ad: Associated data used during encryption (optional). ad: Associated data used during encryption (optional).
maclen: MAC length (16 or 32, default 16).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
if maclen not in (16, 32):
raise TypeError("maclen must be 16 or 32")
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
@@ -717,6 +738,7 @@ class Decryptor:
self._owner = owner self._owner = owner
self._bytes_in = 0 self._bytes_in = 0
self._bytes_out = 0 self._bytes_out = 0
self._maclen = maclen
@property @property
def bytes_in(self) -> int: def bytes_in(self) -> int:
@@ -740,8 +762,10 @@ class Decryptor:
Raises: Raises:
TypeError: If destination buffer is too small. TypeError: If destination buffer is too small.
RuntimeError: If the C update call fails. RuntimeError: If the C update call fails or if called after final().
""" """
if self._st is None:
raise RuntimeError("Cannot call update() after final()")
expected_out = len(ct) expected_out = len(ct)
out = into if into is not None else bytearray(expected_out) out = into if into is not None else bytearray(expected_out)
out_mv = memoryview(out) out_mv = memoryview(out)
@@ -770,20 +794,25 @@ class Decryptor:
"""Finalize decryption by verifying the MAC tag. """Finalize decryption by verifying the MAC tag.
Args: Args:
mac: Tag to verify (16 or 32 bytes). mac: Tag to verify.
Raises: Raises:
TypeError: If tag length is invalid. TypeError: If tag length doesn't match the expected maclen.
ValueError: If authentication fails. ValueError: If authentication fails.
RuntimeError: If called after final().
""" """
maclen = len(mac) if self._st is None:
if maclen not in (16, 32): raise RuntimeError("Cannot call final() after final()")
raise TypeError("mac length must be 16 or 32") maclen = self._maclen
if len(mac) != maclen:
raise TypeError(f"mac length must be {maclen}")
rc = _lib.aegis128x4_state_decrypt_detached_final( rc = _lib.aegis128x4_state_decrypt_detached_final(
self._st, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen self._st, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen
) )
if rc != 0: if rc != 0:
raise ValueError("authentication failed") raise ValueError("authentication failed")
self._st = None
self._owner = None
def new_state(): def new_state():
pyaegis/aegis256.py
+142 -113
View File
@@ -9,7 +9,7 @@ from ._loader import ffi
from ._loader import lib as _lib from ._loader import lib as _lib
from .util import Buffer, new_aligned_struct, nonce_increment, wipe from .util import Buffer, new_aligned_struct, nonce_increment, wipe
NAME = "AEGIS-256" #: Algorithm name NAME = "AEGIS-256" #: Algorithm display name
KEYBYTES = 32 #: Key size in bytes (varies by algorithm) KEYBYTES = 32 #: Key size in bytes (varies by algorithm)
NONCEBYTES = 32 #: Nonce size in bytes (varies by algorithm) NONCEBYTES = 32 #: Nonce size in bytes (varies by algorithm)
MACBYTES = 16 #: Normal MAC size (always 16) MACBYTES = 16 #: Normal MAC size (always 16)
@@ -20,15 +20,22 @@ RATE = 64 #: Byte chunk size in internal processing
def random_key() -> bytearray: def random_key() -> bytearray:
""" """
Generate a random key using cryptographically secure random bytes. Generate a secret key using cryptographically secure random bytes.
It is recommended to wipe() the key after no longer needed to keep it secret. It is recommended to wipe() the key after no longer needed.
""" """
return bytearray(secrets.token_bytes(KEYBYTES)) return bytearray(secrets.token_bytes(KEYBYTES))
def random_nonce() -> bytearray: def random_nonce() -> bytearray:
"""Generate a random nonce using cryptographically secure random bytes.""" """
Generate a public nonce using cryptographically secure random bytes.
Nonces (a number used once) are public data that may be sent together
with the ciphertext, but they need to be unique for each use.
See also: nonce_increment() can be used to derive sequential nonces.
"""
return bytearray(secrets.token_bytes(NONCEBYTES)) return bytearray(secrets.token_bytes(NONCEBYTES))
@@ -46,11 +53,11 @@ def encrypt_detached(
ct_into: Buffer | None = None, ct_into: Buffer | None = None,
mac_into: Buffer | None = None, mac_into: Buffer | None = None,
) -> tuple[bytearray | memoryview, bytearray | memoryview]: ) -> tuple[bytearray | memoryview, bytearray | memoryview]:
f"""Encrypt message with associated data, returning ciphertext and MAC separately. """Encrypt message with associated data, returning ciphertext and MAC separately.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -114,11 +121,11 @@ def decrypt_detached(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext with detached MAC and associated data. """Decrypt ciphertext with detached MAC and associated data.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext to decrypt. ct: The ciphertext to decrypt.
mac: The MAC to verify. mac: The MAC to verify.
ad: Associated data (optional). ad: Associated data (optional).
@@ -170,11 +177,11 @@ def encrypt(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Encrypt message with associated data, returning ciphertext with appended MAC. """Encrypt message with associated data, returning ciphertext with appended MAC.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -226,11 +233,11 @@ def decrypt(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext with appended MAC and associated data. """Decrypt ciphertext with appended MAC and associated data.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext with MAC to decrypt. ct: The ciphertext with MAC to decrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -283,11 +290,11 @@ def stream(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | Buffer: ) -> bytearray | Buffer:
f"""Generate a stream of pseudorandom bytes. """Generate a stream of pseudorandom bytes.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}, uses zeroes for nonce if None). nonce: Public nonce (generate with random_nonce(), uses zeroes for nonce if None).
length: Number of bytes to generate (required if into is None). length: Number of bytes to generate (required if into is None).
into: Buffer to write stream into (default: bytearray created). into: Buffer to write stream into (default: bytearray created).
@@ -325,11 +332,11 @@ def encrypt_unauthenticated(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Encrypt message without authentication (for testing/debugging). """Encrypt message without authentication (for testing/debugging).
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
into: Buffer to write ciphertext into (default: bytearray created). into: Buffer to write ciphertext into (default: bytearray created).
@@ -366,11 +373,11 @@ def decrypt_unauthenticated(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext without authentication (for testing/debugging). """Decrypt ciphertext without authentication (for testing/debugging).
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext to decrypt. ct: The ciphertext to decrypt.
into: Buffer to write plaintext into (default: bytearray created). into: Buffer to write plaintext into (default: bytearray created).
@@ -408,11 +415,11 @@ def mac(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Compute a MAC for the given data in one shot. """Compute a MAC for the given data in one shot.
Args: Args:
key: Key ({KEYBYTES=}) key: Secret key (generate with random_key())
nonce: Nonce ({NONCEBYTES=}) nonce: Public nonce (generate with random_nonce())
data: Data to MAC data: Data to MAC
maclen: MAC length (16 or 32, default 16) maclen: MAC length (16 or 32, default 16)
into: Buffer to write MAC into (default: bytearray created) into: Buffer to write MAC into (default: bytearray created)
@@ -420,70 +427,59 @@ def mac(
Returns: Returns:
MAC bytes as bytearray if into not provided, memoryview of into otherwise MAC bytes as bytearray if into not provided, memoryview of into otherwise
""" """
mac_state = Mac(key, nonce) mac_state = Mac(key, nonce, maclen)
mac_state.update(data) mac_state.update(data)
return mac_state.final(maclen, into) return mac_state.final(into)
class Mac: class Mac:
"""AEGIS-256 MAC state wrapper. """AEGIS-256 MAC state wrapper.
Usage: Example:
mac = Mac(key, nonce) a = Mac(key, nonce)
mac.update(data) a.update(data)
tag = mac.final() # defaults to 16-byte MAC mac = a.final()
# or verify:
mac2 = Mac(key, nonce); mac2.update(data); mac2.verify(tag)
""" """
__slots__ = ("_st", "_owner") __slots__ = ("_st", "_owner", "_maclen")
def __init__( def __init__(self, key: Buffer, nonce: Buffer, maclen: int = MACBYTES) -> None:
self, """Create a MAC with the given key, nonce, and tag length.
key: Buffer,
nonce: Buffer,
_other=None,
) -> None:
f"""Initialize a MAC state with a nonce and key.
Args:
key: Key ({KEYBYTES=}).
nonce: Nonce ({NONCEBYTES=}).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
st, owner = new_aligned_struct("aegis256_mac_state", ALIGNMENT) if maclen not in (16, 32):
self._st = st raise TypeError("maclen must be 16 or 32")
self._owner = owner
if _other is not None: # clone path
_lib.aegis256_mac_state_clone(self._st, _other._st)
return
# Normal init path
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
raise TypeError(f"nonce length must be {NONCEBYTES}") raise TypeError(f"nonce length must be {NONCEBYTES}")
self._maclen = maclen
st, owner = new_aligned_struct("aegis256_mac_state", ALIGNMENT)
self._st = st
self._owner = owner
_lib.aegis256_mac_init(self._st, _ptr(key), _ptr(nonce)) _lib.aegis256_mac_init(self._st, _ptr(key), _ptr(nonce))
def __deepcopy__(self) -> "Mac": def __deepcopy__(self) -> "Mac":
"""Return a clone of current MAC state.""" """Return a clone of current MAC state."""
return Mac(b"", b"", _other=self) clone = object.__new__(Mac)
clone._maclen = self._maclen
clone._st, clone._owner = new_aligned_struct("aegis256_mac_state", ALIGNMENT)
_lib.aegis256_mac_state_clone(clone._st, self._st)
return clone
clone = __deepcopy__ clone = __deepcopy__
def reset(self) -> None: def reset(self) -> None:
"""Reset the MAC state so it can be reused with the same nonce and key.""" """Reset back to the original state, prior to any updates."""
_lib.aegis256_mac_reset(self._st) _lib.aegis256_mac_reset(self._st)
def update(self, data: Buffer) -> None: def update(self, data: Buffer) -> None:
"""Absorb data into the MAC state. """Update the MAC state with more data.
Args: Repeated calls to update() are equivalent to a single call with the concatenated data.
data: Bytes-like object to authenticate.
Raises:
RuntimeError: If the underlying C function reports an error.
""" """
rc = _lib.aegis256_mac_update(self._st, _ptr(data), len(data)) rc = _lib.aegis256_mac_update(self._st, _ptr(data), len(data))
if rc != 0: if rc != 0:
@@ -491,15 +487,13 @@ class Mac:
err_name = errno.errorcode.get(err_num, f"errno_{err_num}") err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
raise RuntimeError(f"mac update failed: {err_name}") raise RuntimeError(f"mac update failed: {err_name}")
def final( def final(self, into: Buffer | None = None) -> bytearray | memoryview:
self, """Calculate and return the MAC tag for the currently input data.
maclen: int = MACBYTES,
into: Buffer | None = None, Unlike the C library, this method does not alter the current state,
) -> bytearray | memoryview: allowing for multiple calls and further updates on the same object.
"""Finalize and return the MAC tag.
Args: Args:
maclen: Tag length in bytes (16 or 32). Defaults to 16.
into: Optional buffer to write the tag into (default: bytearray created). into: Optional buffer to write the tag into (default: bytearray created).
Returns: Returns:
@@ -509,30 +503,36 @@ class Mac:
TypeError: If lengths are invalid. TypeError: If lengths are invalid.
RuntimeError: If finalization fails in the C library. RuntimeError: If finalization fails in the C library.
""" """
if maclen not in (16, 32): maclen = self._maclen
raise TypeError("maclen must be 16 or 32")
if into is None: if into is None:
out = bytearray(maclen) out = bytearray(maclen)
else: else:
if len(into) < maclen: if len(into) < maclen:
raise TypeError("into length must be at least maclen") raise TypeError("into length must be at least maclen")
out = into out = into
out_mv = memoryview(out)
rc = _lib.aegis256_mac_final(self._st, ffi.from_buffer(out_mv), maclen) rc = _lib.aegis256_mac_final(self.clone()._st, ffi.from_buffer(out), maclen)
if rc != 0: if rc != 0:
err_num = ffi.errno err_num = ffi.errno
err_name = errno.errorcode.get(err_num, f"errno_{err_num}") err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
raise RuntimeError(f"mac final failed: {err_name}") raise RuntimeError(f"mac final failed: {err_name}")
return out if into is None else memoryview(out)[:maclen] # type: ignore return out if into is None else memoryview(out)[:maclen] # type: ignore
def digest(self) -> bytes:
"""Calculate and return the MAC tag as bytes."""
return bytes(self.final())
def hexdigest(self) -> str:
"""Calculate and return the MAC tag as a hex string."""
return self.digest().hex()
def verify(self, mac: Buffer): def verify(self, mac: Buffer):
"""Verify a tag for the current MAC state. """Verify that the data entered so far matches the given MAC tag.
Unlike the C library, this method does not alter the current state.
Args: Args:
mac: The tag to verify (16 or 32 bytes). mac: The tag to verify against (16 or 32 bytes).
Returns:
Only if verification succeeds.
Raises: Raises:
TypeError: If tag length is invalid. TypeError: If tag length is invalid.
@@ -541,7 +541,9 @@ class Mac:
maclen = len(mac) maclen = len(mac)
if maclen not in (16, 32): if maclen not in (16, 32):
raise TypeError("mac length must be 16 or 32") raise TypeError("mac length must be 16 or 32")
rc = _lib.aegis256_mac_verify(self._st, _ptr(mac), maclen)
cloned = self.clone()
rc = _lib.aegis256_mac_verify(cloned._st, _ptr(mac), maclen)
if rc != 0: if rc != 0:
raise ValueError("mac verification failed") raise ValueError("mac verification failed")
@@ -550,23 +552,31 @@ class Encryptor:
"""Incremental encryptor. """Incremental encryptor.
- update(message[, into]) -> returns produced ciphertext bytes - update(message[, into]) -> returns produced ciphertext bytes
- final([into], maclen=16) -> returns tail+tag bytes - final([into]) -> returns MAC tag
- final_detached([ct_into], [mac_into], maclen=16) -> returns (tail_bytes, mac)
""" """
__slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out") __slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out", "_maclen")
def __init__(self, key: Buffer, nonce: Buffer, ad: Buffer | None = None): def __init__(
f"""Create an incremental encryptor. self,
key: Buffer,
nonce: Buffer,
ad: Buffer | None = None,
maclen: int = MACBYTES,
):
"""Create an incremental encryptor.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
ad: Associated data to bind to the encryption (optional). ad: Associated data to bind to the encryption (optional).
maclen: MAC length (16 or 32, default 16).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
if maclen not in (16, 32):
raise TypeError("maclen must be 16 or 32")
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
@@ -583,6 +593,7 @@ class Encryptor:
self._owner = owner self._owner = owner
self._bytes_in = 0 self._bytes_in = 0
self._bytes_out = 0 self._bytes_out = 0
self._maclen = maclen
@property @property
def bytes_in(self) -> int: def bytes_in(self) -> int:
@@ -611,8 +622,10 @@ class Encryptor:
Raises: Raises:
TypeError: If destination buffer is too small. TypeError: If destination buffer is too small.
RuntimeError: If the C update call fails. RuntimeError: If the C update call fails or if called after final().
""" """
if self._st is None:
raise RuntimeError("Cannot call update() after final()")
expected_out = len(message) expected_out = len(message)
out = into if into is not None else bytearray(expected_out) out = into if into is not None else bytearray(expected_out)
out_mv = memoryview(out) out_mv = memoryview(out)
@@ -641,24 +654,21 @@ class Encryptor:
self._bytes_out += w self._bytes_out += w
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:w] # type: ignore
def final( def final(self, into: Buffer | None = None) -> bytearray | memoryview:
self, into: Buffer | None = None, maclen: int = MACBYTES """Finalize encryption and return the authentication tag.
) -> bytearray | memoryview:
"""Finalize encryption, writing any remaining bytes and the tag.
Args: Args:
into: Optional destination buffer for the tail and tag. into: Optional destination buffer for the tag.
maclen: Tag length (16 or 32). Defaults to 16.
Returns: Returns:
A memoryview of the produced bytes (tail + tag) if into provided, bytearray slice otherwise. The authentication tag as bytearray if into not provided, memoryview of into otherwise.
Raises: Raises:
TypeError: If maclen is invalid. RuntimeError: If the C final call fails or if called after final().
RuntimeError: If the C final call fails.
""" """
if maclen not in (16, 32): if self._st is None:
raise TypeError("maclen must be 16 or 32") raise RuntimeError("Cannot call final() after final()")
maclen = self._maclen
# Only the authentication tag is produced here; allocate exactly maclen # Only the authentication tag is produced here; allocate exactly maclen
out = into if into is not None else bytearray(maclen) out = into if into is not None else bytearray(maclen)
written = ffi.new("size_t *") written = ffi.new("size_t *")
@@ -678,6 +688,8 @@ class Encryptor:
# Only the tag bytes are returned when we allocate the buffer # Only the tag bytes are returned when we allocate the buffer
assert w == maclen assert w == maclen
self._bytes_out += w self._bytes_out += w
self._st = None
self._owner = None
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:w] # type: ignore
@@ -688,19 +700,28 @@ class Decryptor:
- final(mac) -> verifies the MAC tag - final(mac) -> verifies the MAC tag
""" """
__slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out") __slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out", "_maclen")
def __init__(self, key: Buffer, nonce: Buffer, ad: Buffer | None = None): def __init__(
f"""Create an incremental decryptor for detached tags. self,
key: Buffer,
nonce: Buffer,
ad: Buffer | None = None,
maclen: int = MACBYTES,
):
"""Create an incremental decryptor for detached tags.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ad: Associated data used during encryption (optional). ad: Associated data used during encryption (optional).
maclen: MAC length (16 or 32, default 16).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
if maclen not in (16, 32):
raise TypeError("maclen must be 16 or 32")
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
@@ -717,6 +738,7 @@ class Decryptor:
self._owner = owner self._owner = owner
self._bytes_in = 0 self._bytes_in = 0
self._bytes_out = 0 self._bytes_out = 0
self._maclen = maclen
@property @property
def bytes_in(self) -> int: def bytes_in(self) -> int:
@@ -740,8 +762,10 @@ class Decryptor:
Raises: Raises:
TypeError: If destination buffer is too small. TypeError: If destination buffer is too small.
RuntimeError: If the C update call fails. RuntimeError: If the C update call fails or if called after final().
""" """
if self._st is None:
raise RuntimeError("Cannot call update() after final()")
expected_out = len(ct) expected_out = len(ct)
out = into if into is not None else bytearray(expected_out) out = into if into is not None else bytearray(expected_out)
out_mv = memoryview(out) out_mv = memoryview(out)
@@ -770,20 +794,25 @@ class Decryptor:
"""Finalize decryption by verifying the MAC tag. """Finalize decryption by verifying the MAC tag.
Args: Args:
mac: Tag to verify (16 or 32 bytes). mac: Tag to verify.
Raises: Raises:
TypeError: If tag length is invalid. TypeError: If tag length doesn't match the expected maclen.
ValueError: If authentication fails. ValueError: If authentication fails.
RuntimeError: If called after final().
""" """
maclen = len(mac) if self._st is None:
if maclen not in (16, 32): raise RuntimeError("Cannot call final() after final()")
raise TypeError("mac length must be 16 or 32") maclen = self._maclen
if len(mac) != maclen:
raise TypeError(f"mac length must be {maclen}")
rc = _lib.aegis256_state_decrypt_detached_final( rc = _lib.aegis256_state_decrypt_detached_final(
self._st, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen self._st, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen
) )
if rc != 0: if rc != 0:
raise ValueError("authentication failed") raise ValueError("authentication failed")
self._st = None
self._owner = None
def new_state(): def new_state():
pyaegis/aegis256x2.py
+142 -113
View File
@@ -9,7 +9,7 @@ from ._loader import ffi
from ._loader import lib as _lib from ._loader import lib as _lib
from .util import Buffer, new_aligned_struct, nonce_increment, wipe from .util import Buffer, new_aligned_struct, nonce_increment, wipe
NAME = "AEGIS-256X2" #: Algorithm name NAME = "AEGIS-256X2" #: Algorithm display name
KEYBYTES = 32 #: Key size in bytes (varies by algorithm) KEYBYTES = 32 #: Key size in bytes (varies by algorithm)
NONCEBYTES = 32 #: Nonce size in bytes (varies by algorithm) NONCEBYTES = 32 #: Nonce size in bytes (varies by algorithm)
MACBYTES = 16 #: Normal MAC size (always 16) MACBYTES = 16 #: Normal MAC size (always 16)
@@ -20,15 +20,22 @@ RATE = 64 #: Byte chunk size in internal processing
def random_key() -> bytearray: def random_key() -> bytearray:
""" """
Generate a random key using cryptographically secure random bytes. Generate a secret key using cryptographically secure random bytes.
It is recommended to wipe() the key after no longer needed to keep it secret. It is recommended to wipe() the key after no longer needed.
""" """
return bytearray(secrets.token_bytes(KEYBYTES)) return bytearray(secrets.token_bytes(KEYBYTES))
def random_nonce() -> bytearray: def random_nonce() -> bytearray:
"""Generate a random nonce using cryptographically secure random bytes.""" """
Generate a public nonce using cryptographically secure random bytes.
Nonces (a number used once) are public data that may be sent together
with the ciphertext, but they need to be unique for each use.
See also: nonce_increment() can be used to derive sequential nonces.
"""
return bytearray(secrets.token_bytes(NONCEBYTES)) return bytearray(secrets.token_bytes(NONCEBYTES))
@@ -46,11 +53,11 @@ def encrypt_detached(
ct_into: Buffer | None = None, ct_into: Buffer | None = None,
mac_into: Buffer | None = None, mac_into: Buffer | None = None,
) -> tuple[bytearray | memoryview, bytearray | memoryview]: ) -> tuple[bytearray | memoryview, bytearray | memoryview]:
f"""Encrypt message with associated data, returning ciphertext and MAC separately. """Encrypt message with associated data, returning ciphertext and MAC separately.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -114,11 +121,11 @@ def decrypt_detached(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext with detached MAC and associated data. """Decrypt ciphertext with detached MAC and associated data.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext to decrypt. ct: The ciphertext to decrypt.
mac: The MAC to verify. mac: The MAC to verify.
ad: Associated data (optional). ad: Associated data (optional).
@@ -170,11 +177,11 @@ def encrypt(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Encrypt message with associated data, returning ciphertext with appended MAC. """Encrypt message with associated data, returning ciphertext with appended MAC.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -226,11 +233,11 @@ def decrypt(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext with appended MAC and associated data. """Decrypt ciphertext with appended MAC and associated data.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext with MAC to decrypt. ct: The ciphertext with MAC to decrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -283,11 +290,11 @@ def stream(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | Buffer: ) -> bytearray | Buffer:
f"""Generate a stream of pseudorandom bytes. """Generate a stream of pseudorandom bytes.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}, uses zeroes for nonce if None). nonce: Public nonce (generate with random_nonce(), uses zeroes for nonce if None).
length: Number of bytes to generate (required if into is None). length: Number of bytes to generate (required if into is None).
into: Buffer to write stream into (default: bytearray created). into: Buffer to write stream into (default: bytearray created).
@@ -325,11 +332,11 @@ def encrypt_unauthenticated(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Encrypt message without authentication (for testing/debugging). """Encrypt message without authentication (for testing/debugging).
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
into: Buffer to write ciphertext into (default: bytearray created). into: Buffer to write ciphertext into (default: bytearray created).
@@ -366,11 +373,11 @@ def decrypt_unauthenticated(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext without authentication (for testing/debugging). """Decrypt ciphertext without authentication (for testing/debugging).
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext to decrypt. ct: The ciphertext to decrypt.
into: Buffer to write plaintext into (default: bytearray created). into: Buffer to write plaintext into (default: bytearray created).
@@ -408,11 +415,11 @@ def mac(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Compute a MAC for the given data in one shot. """Compute a MAC for the given data in one shot.
Args: Args:
key: Key ({KEYBYTES=}) key: Secret key (generate with random_key())
nonce: Nonce ({NONCEBYTES=}) nonce: Public nonce (generate with random_nonce())
data: Data to MAC data: Data to MAC
maclen: MAC length (16 or 32, default 16) maclen: MAC length (16 or 32, default 16)
into: Buffer to write MAC into (default: bytearray created) into: Buffer to write MAC into (default: bytearray created)
@@ -420,70 +427,59 @@ def mac(
Returns: Returns:
MAC bytes as bytearray if into not provided, memoryview of into otherwise MAC bytes as bytearray if into not provided, memoryview of into otherwise
""" """
mac_state = Mac(key, nonce) mac_state = Mac(key, nonce, maclen)
mac_state.update(data) mac_state.update(data)
return mac_state.final(maclen, into) return mac_state.final(into)
class Mac: class Mac:
"""AEGIS-256X2 MAC state wrapper. """AEGIS-256X2 MAC state wrapper.
Usage: Example:
mac = Mac(key, nonce) a = Mac(key, nonce)
mac.update(data) a.update(data)
tag = mac.final() # defaults to 16-byte MAC mac = a.final()
# or verify:
mac2 = Mac(key, nonce); mac2.update(data); mac2.verify(tag)
""" """
__slots__ = ("_st", "_owner") __slots__ = ("_st", "_owner", "_maclen")
def __init__( def __init__(self, key: Buffer, nonce: Buffer, maclen: int = MACBYTES) -> None:
self, """Create a MAC with the given key, nonce, and tag length.
key: Buffer,
nonce: Buffer,
_other=None,
) -> None:
f"""Initialize a MAC state with a nonce and key.
Args:
key: Key ({KEYBYTES=}).
nonce: Nonce ({NONCEBYTES=}).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
st, owner = new_aligned_struct("aegis256x2_mac_state", ALIGNMENT) if maclen not in (16, 32):
self._st = st raise TypeError("maclen must be 16 or 32")
self._owner = owner
if _other is not None: # clone path
_lib.aegis256x2_mac_state_clone(self._st, _other._st)
return
# Normal init path
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
raise TypeError(f"nonce length must be {NONCEBYTES}") raise TypeError(f"nonce length must be {NONCEBYTES}")
self._maclen = maclen
st, owner = new_aligned_struct("aegis256x2_mac_state", ALIGNMENT)
self._st = st
self._owner = owner
_lib.aegis256x2_mac_init(self._st, _ptr(key), _ptr(nonce)) _lib.aegis256x2_mac_init(self._st, _ptr(key), _ptr(nonce))
def __deepcopy__(self) -> "Mac": def __deepcopy__(self) -> "Mac":
"""Return a clone of current MAC state.""" """Return a clone of current MAC state."""
return Mac(b"", b"", _other=self) clone = object.__new__(Mac)
clone._maclen = self._maclen
clone._st, clone._owner = new_aligned_struct("aegis256x2_mac_state", ALIGNMENT)
_lib.aegis256x2_mac_state_clone(clone._st, self._st)
return clone
clone = __deepcopy__ clone = __deepcopy__
def reset(self) -> None: def reset(self) -> None:
"""Reset the MAC state so it can be reused with the same nonce and key.""" """Reset back to the original state, prior to any updates."""
_lib.aegis256x2_mac_reset(self._st) _lib.aegis256x2_mac_reset(self._st)
def update(self, data: Buffer) -> None: def update(self, data: Buffer) -> None:
"""Absorb data into the MAC state. """Update the MAC state with more data.
Args: Repeated calls to update() are equivalent to a single call with the concatenated data.
data: Bytes-like object to authenticate.
Raises:
RuntimeError: If the underlying C function reports an error.
""" """
rc = _lib.aegis256x2_mac_update(self._st, _ptr(data), len(data)) rc = _lib.aegis256x2_mac_update(self._st, _ptr(data), len(data))
if rc != 0: if rc != 0:
@@ -491,15 +487,13 @@ class Mac:
err_name = errno.errorcode.get(err_num, f"errno_{err_num}") err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
raise RuntimeError(f"mac update failed: {err_name}") raise RuntimeError(f"mac update failed: {err_name}")
def final( def final(self, into: Buffer | None = None) -> bytearray | memoryview:
self, """Calculate and return the MAC tag for the currently input data.
maclen: int = MACBYTES,
into: Buffer | None = None, Unlike the C library, this method does not alter the current state,
) -> bytearray | memoryview: allowing for multiple calls and further updates on the same object.
"""Finalize and return the MAC tag.
Args: Args:
maclen: Tag length in bytes (16 or 32). Defaults to 16.
into: Optional buffer to write the tag into (default: bytearray created). into: Optional buffer to write the tag into (default: bytearray created).
Returns: Returns:
@@ -509,30 +503,36 @@ class Mac:
TypeError: If lengths are invalid. TypeError: If lengths are invalid.
RuntimeError: If finalization fails in the C library. RuntimeError: If finalization fails in the C library.
""" """
if maclen not in (16, 32): maclen = self._maclen
raise TypeError("maclen must be 16 or 32")
if into is None: if into is None:
out = bytearray(maclen) out = bytearray(maclen)
else: else:
if len(into) < maclen: if len(into) < maclen:
raise TypeError("into length must be at least maclen") raise TypeError("into length must be at least maclen")
out = into out = into
out_mv = memoryview(out)
rc = _lib.aegis256x2_mac_final(self._st, ffi.from_buffer(out_mv), maclen) rc = _lib.aegis256x2_mac_final(self.clone()._st, ffi.from_buffer(out), maclen)
if rc != 0: if rc != 0:
err_num = ffi.errno err_num = ffi.errno
err_name = errno.errorcode.get(err_num, f"errno_{err_num}") err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
raise RuntimeError(f"mac final failed: {err_name}") raise RuntimeError(f"mac final failed: {err_name}")
return out if into is None else memoryview(out)[:maclen] # type: ignore return out if into is None else memoryview(out)[:maclen] # type: ignore
def digest(self) -> bytes:
"""Calculate and return the MAC tag as bytes."""
return bytes(self.final())
def hexdigest(self) -> str:
"""Calculate and return the MAC tag as a hex string."""
return self.digest().hex()
def verify(self, mac: Buffer): def verify(self, mac: Buffer):
"""Verify a tag for the current MAC state. """Verify that the data entered so far matches the given MAC tag.
Unlike the C library, this method does not alter the current state.
Args: Args:
mac: The tag to verify (16 or 32 bytes). mac: The tag to verify against (16 or 32 bytes).
Returns:
Only if verification succeeds.
Raises: Raises:
TypeError: If tag length is invalid. TypeError: If tag length is invalid.
@@ -541,7 +541,9 @@ class Mac:
maclen = len(mac) maclen = len(mac)
if maclen not in (16, 32): if maclen not in (16, 32):
raise TypeError("mac length must be 16 or 32") raise TypeError("mac length must be 16 or 32")
rc = _lib.aegis256x2_mac_verify(self._st, _ptr(mac), maclen)
cloned = self.clone()
rc = _lib.aegis256x2_mac_verify(cloned._st, _ptr(mac), maclen)
if rc != 0: if rc != 0:
raise ValueError("mac verification failed") raise ValueError("mac verification failed")
@@ -550,23 +552,31 @@ class Encryptor:
"""Incremental encryptor. """Incremental encryptor.
- update(message[, into]) -> returns produced ciphertext bytes - update(message[, into]) -> returns produced ciphertext bytes
- final([into], maclen=16) -> returns tail+tag bytes - final([into]) -> returns MAC tag
- final_detached([ct_into], [mac_into], maclen=16) -> returns (tail_bytes, mac)
""" """
__slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out") __slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out", "_maclen")
def __init__(self, key: Buffer, nonce: Buffer, ad: Buffer | None = None): def __init__(
f"""Create an incremental encryptor. self,
key: Buffer,
nonce: Buffer,
ad: Buffer | None = None,
maclen: int = MACBYTES,
):
"""Create an incremental encryptor.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
ad: Associated data to bind to the encryption (optional). ad: Associated data to bind to the encryption (optional).
maclen: MAC length (16 or 32, default 16).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
if maclen not in (16, 32):
raise TypeError("maclen must be 16 or 32")
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
@@ -583,6 +593,7 @@ class Encryptor:
self._owner = owner self._owner = owner
self._bytes_in = 0 self._bytes_in = 0
self._bytes_out = 0 self._bytes_out = 0
self._maclen = maclen
@property @property
def bytes_in(self) -> int: def bytes_in(self) -> int:
@@ -611,8 +622,10 @@ class Encryptor:
Raises: Raises:
TypeError: If destination buffer is too small. TypeError: If destination buffer is too small.
RuntimeError: If the C update call fails. RuntimeError: If the C update call fails or if called after final().
""" """
if self._st is None:
raise RuntimeError("Cannot call update() after final()")
expected_out = len(message) expected_out = len(message)
out = into if into is not None else bytearray(expected_out) out = into if into is not None else bytearray(expected_out)
out_mv = memoryview(out) out_mv = memoryview(out)
@@ -641,24 +654,21 @@ class Encryptor:
self._bytes_out += w self._bytes_out += w
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:w] # type: ignore
def final( def final(self, into: Buffer | None = None) -> bytearray | memoryview:
self, into: Buffer | None = None, maclen: int = MACBYTES """Finalize encryption and return the authentication tag.
) -> bytearray | memoryview:
"""Finalize encryption, writing any remaining bytes and the tag.
Args: Args:
into: Optional destination buffer for the tail and tag. into: Optional destination buffer for the tag.
maclen: Tag length (16 or 32). Defaults to 16.
Returns: Returns:
A memoryview of the produced bytes (tail + tag) if into provided, bytearray slice otherwise. The authentication tag as bytearray if into not provided, memoryview of into otherwise.
Raises: Raises:
TypeError: If maclen is invalid. RuntimeError: If the C final call fails or if called after final().
RuntimeError: If the C final call fails.
""" """
if maclen not in (16, 32): if self._st is None:
raise TypeError("maclen must be 16 or 32") raise RuntimeError("Cannot call final() after final()")
maclen = self._maclen
# Only the authentication tag is produced here; allocate exactly maclen # Only the authentication tag is produced here; allocate exactly maclen
out = into if into is not None else bytearray(maclen) out = into if into is not None else bytearray(maclen)
written = ffi.new("size_t *") written = ffi.new("size_t *")
@@ -678,6 +688,8 @@ class Encryptor:
# Only the tag bytes are returned when we allocate the buffer # Only the tag bytes are returned when we allocate the buffer
assert w == maclen assert w == maclen
self._bytes_out += w self._bytes_out += w
self._st = None
self._owner = None
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:w] # type: ignore
@@ -688,19 +700,28 @@ class Decryptor:
- final(mac) -> verifies the MAC tag - final(mac) -> verifies the MAC tag
""" """
__slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out") __slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out", "_maclen")
def __init__(self, key: Buffer, nonce: Buffer, ad: Buffer | None = None): def __init__(
f"""Create an incremental decryptor for detached tags. self,
key: Buffer,
nonce: Buffer,
ad: Buffer | None = None,
maclen: int = MACBYTES,
):
"""Create an incremental decryptor for detached tags.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ad: Associated data used during encryption (optional). ad: Associated data used during encryption (optional).
maclen: MAC length (16 or 32, default 16).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
if maclen not in (16, 32):
raise TypeError("maclen must be 16 or 32")
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
@@ -717,6 +738,7 @@ class Decryptor:
self._owner = owner self._owner = owner
self._bytes_in = 0 self._bytes_in = 0
self._bytes_out = 0 self._bytes_out = 0
self._maclen = maclen
@property @property
def bytes_in(self) -> int: def bytes_in(self) -> int:
@@ -740,8 +762,10 @@ class Decryptor:
Raises: Raises:
TypeError: If destination buffer is too small. TypeError: If destination buffer is too small.
RuntimeError: If the C update call fails. RuntimeError: If the C update call fails or if called after final().
""" """
if self._st is None:
raise RuntimeError("Cannot call update() after final()")
expected_out = len(ct) expected_out = len(ct)
out = into if into is not None else bytearray(expected_out) out = into if into is not None else bytearray(expected_out)
out_mv = memoryview(out) out_mv = memoryview(out)
@@ -770,20 +794,25 @@ class Decryptor:
"""Finalize decryption by verifying the MAC tag. """Finalize decryption by verifying the MAC tag.
Args: Args:
mac: Tag to verify (16 or 32 bytes). mac: Tag to verify.
Raises: Raises:
TypeError: If tag length is invalid. TypeError: If tag length doesn't match the expected maclen.
ValueError: If authentication fails. ValueError: If authentication fails.
RuntimeError: If called after final().
""" """
maclen = len(mac) if self._st is None:
if maclen not in (16, 32): raise RuntimeError("Cannot call final() after final()")
raise TypeError("mac length must be 16 or 32") maclen = self._maclen
if len(mac) != maclen:
raise TypeError(f"mac length must be {maclen}")
rc = _lib.aegis256x2_state_decrypt_detached_final( rc = _lib.aegis256x2_state_decrypt_detached_final(
self._st, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen self._st, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen
) )
if rc != 0: if rc != 0:
raise ValueError("authentication failed") raise ValueError("authentication failed")
self._st = None
self._owner = None
def new_state(): def new_state():
pyaegis/aegis256x4.py
+142 -113
View File
@@ -9,7 +9,7 @@ from ._loader import ffi
from ._loader import lib as _lib from ._loader import lib as _lib
from .util import Buffer, new_aligned_struct, nonce_increment, wipe from .util import Buffer, new_aligned_struct, nonce_increment, wipe
NAME = "AEGIS-256X4" #: Algorithm name NAME = "AEGIS-256X4" #: Algorithm display name
KEYBYTES = 32 #: Key size in bytes (varies by algorithm) KEYBYTES = 32 #: Key size in bytes (varies by algorithm)
NONCEBYTES = 32 #: Nonce size in bytes (varies by algorithm) NONCEBYTES = 32 #: Nonce size in bytes (varies by algorithm)
MACBYTES = 16 #: Normal MAC size (always 16) MACBYTES = 16 #: Normal MAC size (always 16)
@@ -20,15 +20,22 @@ RATE = 64 #: Byte chunk size in internal processing
def random_key() -> bytearray: def random_key() -> bytearray:
""" """
Generate a random key using cryptographically secure random bytes. Generate a secret key using cryptographically secure random bytes.
It is recommended to wipe() the key after no longer needed to keep it secret. It is recommended to wipe() the key after no longer needed.
""" """
return bytearray(secrets.token_bytes(KEYBYTES)) return bytearray(secrets.token_bytes(KEYBYTES))
def random_nonce() -> bytearray: def random_nonce() -> bytearray:
"""Generate a random nonce using cryptographically secure random bytes.""" """
Generate a public nonce using cryptographically secure random bytes.
Nonces (a number used once) are public data that may be sent together
with the ciphertext, but they need to be unique for each use.
See also: nonce_increment() can be used to derive sequential nonces.
"""
return bytearray(secrets.token_bytes(NONCEBYTES)) return bytearray(secrets.token_bytes(NONCEBYTES))
@@ -46,11 +53,11 @@ def encrypt_detached(
ct_into: Buffer | None = None, ct_into: Buffer | None = None,
mac_into: Buffer | None = None, mac_into: Buffer | None = None,
) -> tuple[bytearray | memoryview, bytearray | memoryview]: ) -> tuple[bytearray | memoryview, bytearray | memoryview]:
f"""Encrypt message with associated data, returning ciphertext and MAC separately. """Encrypt message with associated data, returning ciphertext and MAC separately.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -114,11 +121,11 @@ def decrypt_detached(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext with detached MAC and associated data. """Decrypt ciphertext with detached MAC and associated data.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext to decrypt. ct: The ciphertext to decrypt.
mac: The MAC to verify. mac: The MAC to verify.
ad: Associated data (optional). ad: Associated data (optional).
@@ -170,11 +177,11 @@ def encrypt(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Encrypt message with associated data, returning ciphertext with appended MAC. """Encrypt message with associated data, returning ciphertext with appended MAC.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -226,11 +233,11 @@ def decrypt(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext with appended MAC and associated data. """Decrypt ciphertext with appended MAC and associated data.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext with MAC to decrypt. ct: The ciphertext with MAC to decrypt.
ad: Associated data (optional). ad: Associated data (optional).
maclen: MAC length (16 or 32, default 16). maclen: MAC length (16 or 32, default 16).
@@ -283,11 +290,11 @@ def stream(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | Buffer: ) -> bytearray | Buffer:
f"""Generate a stream of pseudorandom bytes. """Generate a stream of pseudorandom bytes.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}, uses zeroes for nonce if None). nonce: Public nonce (generate with random_nonce(), uses zeroes for nonce if None).
length: Number of bytes to generate (required if into is None). length: Number of bytes to generate (required if into is None).
into: Buffer to write stream into (default: bytearray created). into: Buffer to write stream into (default: bytearray created).
@@ -325,11 +332,11 @@ def encrypt_unauthenticated(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Encrypt message without authentication (for testing/debugging). """Encrypt message without authentication (for testing/debugging).
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
message: The plaintext message to encrypt. message: The plaintext message to encrypt.
into: Buffer to write ciphertext into (default: bytearray created). into: Buffer to write ciphertext into (default: bytearray created).
@@ -366,11 +373,11 @@ def decrypt_unauthenticated(
*, *,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Decrypt ciphertext without authentication (for testing/debugging). """Decrypt ciphertext without authentication (for testing/debugging).
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ct: The ciphertext to decrypt. ct: The ciphertext to decrypt.
into: Buffer to write plaintext into (default: bytearray created). into: Buffer to write plaintext into (default: bytearray created).
@@ -408,11 +415,11 @@ def mac(
maclen: int = MACBYTES, maclen: int = MACBYTES,
into: Buffer | None = None, into: Buffer | None = None,
) -> bytearray | memoryview: ) -> bytearray | memoryview:
f"""Compute a MAC for the given data in one shot. """Compute a MAC for the given data in one shot.
Args: Args:
key: Key ({KEYBYTES=}) key: Secret key (generate with random_key())
nonce: Nonce ({NONCEBYTES=}) nonce: Public nonce (generate with random_nonce())
data: Data to MAC data: Data to MAC
maclen: MAC length (16 or 32, default 16) maclen: MAC length (16 or 32, default 16)
into: Buffer to write MAC into (default: bytearray created) into: Buffer to write MAC into (default: bytearray created)
@@ -420,70 +427,59 @@ def mac(
Returns: Returns:
MAC bytes as bytearray if into not provided, memoryview of into otherwise MAC bytes as bytearray if into not provided, memoryview of into otherwise
""" """
mac_state = Mac(key, nonce) mac_state = Mac(key, nonce, maclen)
mac_state.update(data) mac_state.update(data)
return mac_state.final(maclen, into) return mac_state.final(into)
class Mac: class Mac:
"""AEGIS-256X4 MAC state wrapper. """AEGIS-256X4 MAC state wrapper.
Usage: Example:
mac = Mac(key, nonce) a = Mac(key, nonce)
mac.update(data) a.update(data)
tag = mac.final() # defaults to 16-byte MAC mac = a.final()
# or verify:
mac2 = Mac(key, nonce); mac2.update(data); mac2.verify(tag)
""" """
__slots__ = ("_st", "_owner") __slots__ = ("_st", "_owner", "_maclen")
def __init__( def __init__(self, key: Buffer, nonce: Buffer, maclen: int = MACBYTES) -> None:
self, """Create a MAC with the given key, nonce, and tag length.
key: Buffer,
nonce: Buffer,
_other=None,
) -> None:
f"""Initialize a MAC state with a nonce and key.
Args:
key: Key ({KEYBYTES=}).
nonce: Nonce ({NONCEBYTES=}).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
st, owner = new_aligned_struct("aegis256x4_mac_state", ALIGNMENT) if maclen not in (16, 32):
self._st = st raise TypeError("maclen must be 16 or 32")
self._owner = owner
if _other is not None: # clone path
_lib.aegis256x4_mac_state_clone(self._st, _other._st)
return
# Normal init path
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
raise TypeError(f"nonce length must be {NONCEBYTES}") raise TypeError(f"nonce length must be {NONCEBYTES}")
self._maclen = maclen
st, owner = new_aligned_struct("aegis256x4_mac_state", ALIGNMENT)
self._st = st
self._owner = owner
_lib.aegis256x4_mac_init(self._st, _ptr(key), _ptr(nonce)) _lib.aegis256x4_mac_init(self._st, _ptr(key), _ptr(nonce))
def __deepcopy__(self) -> "Mac": def __deepcopy__(self) -> "Mac":
"""Return a clone of current MAC state.""" """Return a clone of current MAC state."""
return Mac(b"", b"", _other=self) clone = object.__new__(Mac)
clone._maclen = self._maclen
clone._st, clone._owner = new_aligned_struct("aegis256x4_mac_state", ALIGNMENT)
_lib.aegis256x4_mac_state_clone(clone._st, self._st)
return clone
clone = __deepcopy__ clone = __deepcopy__
def reset(self) -> None: def reset(self) -> None:
"""Reset the MAC state so it can be reused with the same nonce and key.""" """Reset back to the original state, prior to any updates."""
_lib.aegis256x4_mac_reset(self._st) _lib.aegis256x4_mac_reset(self._st)
def update(self, data: Buffer) -> None: def update(self, data: Buffer) -> None:
"""Absorb data into the MAC state. """Update the MAC state with more data.
Args: Repeated calls to update() are equivalent to a single call with the concatenated data.
data: Bytes-like object to authenticate.
Raises:
RuntimeError: If the underlying C function reports an error.
""" """
rc = _lib.aegis256x4_mac_update(self._st, _ptr(data), len(data)) rc = _lib.aegis256x4_mac_update(self._st, _ptr(data), len(data))
if rc != 0: if rc != 0:
@@ -491,15 +487,13 @@ class Mac:
err_name = errno.errorcode.get(err_num, f"errno_{err_num}") err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
raise RuntimeError(f"mac update failed: {err_name}") raise RuntimeError(f"mac update failed: {err_name}")
def final( def final(self, into: Buffer | None = None) -> bytearray | memoryview:
self, """Calculate and return the MAC tag for the currently input data.
maclen: int = MACBYTES,
into: Buffer | None = None, Unlike the C library, this method does not alter the current state,
) -> bytearray | memoryview: allowing for multiple calls and further updates on the same object.
"""Finalize and return the MAC tag.
Args: Args:
maclen: Tag length in bytes (16 or 32). Defaults to 16.
into: Optional buffer to write the tag into (default: bytearray created). into: Optional buffer to write the tag into (default: bytearray created).
Returns: Returns:
@@ -509,30 +503,36 @@ class Mac:
TypeError: If lengths are invalid. TypeError: If lengths are invalid.
RuntimeError: If finalization fails in the C library. RuntimeError: If finalization fails in the C library.
""" """
if maclen not in (16, 32): maclen = self._maclen
raise TypeError("maclen must be 16 or 32")
if into is None: if into is None:
out = bytearray(maclen) out = bytearray(maclen)
else: else:
if len(into) < maclen: if len(into) < maclen:
raise TypeError("into length must be at least maclen") raise TypeError("into length must be at least maclen")
out = into out = into
out_mv = memoryview(out)
rc = _lib.aegis256x4_mac_final(self._st, ffi.from_buffer(out_mv), maclen) rc = _lib.aegis256x4_mac_final(self.clone()._st, ffi.from_buffer(out), maclen)
if rc != 0: if rc != 0:
err_num = ffi.errno err_num = ffi.errno
err_name = errno.errorcode.get(err_num, f"errno_{err_num}") err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
raise RuntimeError(f"mac final failed: {err_name}") raise RuntimeError(f"mac final failed: {err_name}")
return out if into is None else memoryview(out)[:maclen] # type: ignore return out if into is None else memoryview(out)[:maclen] # type: ignore
def digest(self) -> bytes:
"""Calculate and return the MAC tag as bytes."""
return bytes(self.final())
def hexdigest(self) -> str:
"""Calculate and return the MAC tag as a hex string."""
return self.digest().hex()
def verify(self, mac: Buffer): def verify(self, mac: Buffer):
"""Verify a tag for the current MAC state. """Verify that the data entered so far matches the given MAC tag.
Unlike the C library, this method does not alter the current state.
Args: Args:
mac: The tag to verify (16 or 32 bytes). mac: The tag to verify against (16 or 32 bytes).
Returns:
Only if verification succeeds.
Raises: Raises:
TypeError: If tag length is invalid. TypeError: If tag length is invalid.
@@ -541,7 +541,9 @@ class Mac:
maclen = len(mac) maclen = len(mac)
if maclen not in (16, 32): if maclen not in (16, 32):
raise TypeError("mac length must be 16 or 32") raise TypeError("mac length must be 16 or 32")
rc = _lib.aegis256x4_mac_verify(self._st, _ptr(mac), maclen)
cloned = self.clone()
rc = _lib.aegis256x4_mac_verify(cloned._st, _ptr(mac), maclen)
if rc != 0: if rc != 0:
raise ValueError("mac verification failed") raise ValueError("mac verification failed")
@@ -550,23 +552,31 @@ class Encryptor:
"""Incremental encryptor. """Incremental encryptor.
- update(message[, into]) -> returns produced ciphertext bytes - update(message[, into]) -> returns produced ciphertext bytes
- final([into], maclen=16) -> returns tail+tag bytes - final([into]) -> returns MAC tag
- final_detached([ct_into], [mac_into], maclen=16) -> returns (tail_bytes, mac)
""" """
__slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out") __slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out", "_maclen")
def __init__(self, key: Buffer, nonce: Buffer, ad: Buffer | None = None): def __init__(
f"""Create an incremental encryptor. self,
key: Buffer,
nonce: Buffer,
ad: Buffer | None = None,
maclen: int = MACBYTES,
):
"""Create an incremental encryptor.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (generate with random_key()).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (generate with random_nonce()).
ad: Associated data to bind to the encryption (optional). ad: Associated data to bind to the encryption (optional).
maclen: MAC length (16 or 32, default 16).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
if maclen not in (16, 32):
raise TypeError("maclen must be 16 or 32")
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
@@ -583,6 +593,7 @@ class Encryptor:
self._owner = owner self._owner = owner
self._bytes_in = 0 self._bytes_in = 0
self._bytes_out = 0 self._bytes_out = 0
self._maclen = maclen
@property @property
def bytes_in(self) -> int: def bytes_in(self) -> int:
@@ -611,8 +622,10 @@ class Encryptor:
Raises: Raises:
TypeError: If destination buffer is too small. TypeError: If destination buffer is too small.
RuntimeError: If the C update call fails. RuntimeError: If the C update call fails or if called after final().
""" """
if self._st is None:
raise RuntimeError("Cannot call update() after final()")
expected_out = len(message) expected_out = len(message)
out = into if into is not None else bytearray(expected_out) out = into if into is not None else bytearray(expected_out)
out_mv = memoryview(out) out_mv = memoryview(out)
@@ -641,24 +654,21 @@ class Encryptor:
self._bytes_out += w self._bytes_out += w
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:w] # type: ignore
def final( def final(self, into: Buffer | None = None) -> bytearray | memoryview:
self, into: Buffer | None = None, maclen: int = MACBYTES """Finalize encryption and return the authentication tag.
) -> bytearray | memoryview:
"""Finalize encryption, writing any remaining bytes and the tag.
Args: Args:
into: Optional destination buffer for the tail and tag. into: Optional destination buffer for the tag.
maclen: Tag length (16 or 32). Defaults to 16.
Returns: Returns:
A memoryview of the produced bytes (tail + tag) if into provided, bytearray slice otherwise. The authentication tag as bytearray if into not provided, memoryview of into otherwise.
Raises: Raises:
TypeError: If maclen is invalid. RuntimeError: If the C final call fails or if called after final().
RuntimeError: If the C final call fails.
""" """
if maclen not in (16, 32): if self._st is None:
raise TypeError("maclen must be 16 or 32") raise RuntimeError("Cannot call final() after final()")
maclen = self._maclen
# Only the authentication tag is produced here; allocate exactly maclen # Only the authentication tag is produced here; allocate exactly maclen
out = into if into is not None else bytearray(maclen) out = into if into is not None else bytearray(maclen)
written = ffi.new("size_t *") written = ffi.new("size_t *")
@@ -678,6 +688,8 @@ class Encryptor:
# Only the tag bytes are returned when we allocate the buffer # Only the tag bytes are returned when we allocate the buffer
assert w == maclen assert w == maclen
self._bytes_out += w self._bytes_out += w
self._st = None
self._owner = None
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:w] # type: ignore
@@ -688,19 +700,28 @@ class Decryptor:
- final(mac) -> verifies the MAC tag - final(mac) -> verifies the MAC tag
""" """
__slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out") __slots__ = ("_st", "_owner", "_bytes_in", "_bytes_out", "_maclen")
def __init__(self, key: Buffer, nonce: Buffer, ad: Buffer | None = None): def __init__(
f"""Create an incremental decryptor for detached tags. self,
key: Buffer,
nonce: Buffer,
ad: Buffer | None = None,
maclen: int = MACBYTES,
):
"""Create an incremental decryptor for detached tags.
Args: Args:
key: Key ({KEYBYTES=}). key: Secret key (same key used during encryption).
nonce: Nonce ({NONCEBYTES=}). nonce: Public nonce (same nonce used during encryption).
ad: Associated data used during encryption (optional). ad: Associated data used during encryption (optional).
maclen: MAC length (16 or 32, default 16).
Raises: Raises:
TypeError: If key or nonce lengths are invalid. TypeError: If key, nonce, or maclen are invalid.
""" """
if maclen not in (16, 32):
raise TypeError("maclen must be 16 or 32")
if len(key) != KEYBYTES: if len(key) != KEYBYTES:
raise TypeError(f"key length must be {KEYBYTES}") raise TypeError(f"key length must be {KEYBYTES}")
if len(nonce) != NONCEBYTES: if len(nonce) != NONCEBYTES:
@@ -717,6 +738,7 @@ class Decryptor:
self._owner = owner self._owner = owner
self._bytes_in = 0 self._bytes_in = 0
self._bytes_out = 0 self._bytes_out = 0
self._maclen = maclen
@property @property
def bytes_in(self) -> int: def bytes_in(self) -> int:
@@ -740,8 +762,10 @@ class Decryptor:
Raises: Raises:
TypeError: If destination buffer is too small. TypeError: If destination buffer is too small.
RuntimeError: If the C update call fails. RuntimeError: If the C update call fails or if called after final().
""" """
if self._st is None:
raise RuntimeError("Cannot call update() after final()")
expected_out = len(ct) expected_out = len(ct)
out = into if into is not None else bytearray(expected_out) out = into if into is not None else bytearray(expected_out)
out_mv = memoryview(out) out_mv = memoryview(out)
@@ -770,20 +794,25 @@ class Decryptor:
"""Finalize decryption by verifying the MAC tag. """Finalize decryption by verifying the MAC tag.
Args: Args:
mac: Tag to verify (16 or 32 bytes). mac: Tag to verify.
Raises: Raises:
TypeError: If tag length is invalid. TypeError: If tag length doesn't match the expected maclen.
ValueError: If authentication fails. ValueError: If authentication fails.
RuntimeError: If called after final().
""" """
maclen = len(mac) if self._st is None:
if maclen not in (16, 32): raise RuntimeError("Cannot call final() after final()")
raise TypeError("mac length must be 16 or 32") maclen = self._maclen
if len(mac) != maclen:
raise TypeError(f"mac length must be {maclen}")
rc = _lib.aegis256x4_state_decrypt_detached_final( rc = _lib.aegis256x4_state_decrypt_detached_final(
self._st, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen self._st, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen
) )
if rc != 0: if rc != 0:
raise ValueError("authentication failed") raise ValueError("authentication failed")
self._st = None
self._owner = None
def new_state(): def new_state():
pyaegis/benchmark.py
+2 -2
View File
@@ -57,14 +57,14 @@ def bench_mac(ciph) -> None:
buf = bytearray(MSG_LEN) buf = bytearray(MSG_LEN)
buf[:] = _random_bytes(len(buf)) buf[:] = _random_bytes(len(buf))
mac0 = ciph.Mac(key, nonce) mac0 = ciph.Mac(key, nonce, maclen=ciph.MACBYTES_LONG)
mac_out = bytearray(ciph.MACBYTES_LONG) mac_out = bytearray(ciph.MACBYTES_LONG)
t0 = time.perf_counter() t0 = time.perf_counter()
for _ in range(ITERATIONS): for _ in range(ITERATIONS):
mac = mac0.clone() mac = mac0.clone()
mac.update(buf) mac.update(buf)
mac.final(maclen=ciph.MACBYTES_LONG, into=mac_out) mac.final(into=mac_out)
t1 = time.perf_counter() t1 = time.perf_counter()
_ = mac_out[0] _ = mac_out[0]
tests/test_encrypt.py
+8 -12
View File
@@ -147,14 +147,12 @@ def test_encrypt_decrypt_incremental(vector):
expected_tag128 = bytes.fromhex(vector["tag128"]) expected_tag128 = bytes.fromhex(vector["tag128"])
# Incremental encryption with random chunking # Incremental encryption with random chunking
encryptor = alg.Encryptor(key, nonce, ad) encryptor = alg.Encryptor(key, nonce, ad, maclen=16)
ct_chunks = [] ct_chunks = []
for chunk in random_split_bytes(msg): for chunk in random_split_bytes(msg):
ct_result = encryptor.update(chunk) ct_result = encryptor.update(chunk)
ct_chunks.append(bytes(ct_result)) ct_chunks.append(bytes(ct_result))
final_output = encryptor.final(maclen=16) computed_mac = bytes(encryptor.final())
ct_chunks.append(bytes(final_output[:-16])) # ciphertext part
computed_mac = bytes(final_output[-16:]) # MAC part
# Combine ciphertext chunks # Combine ciphertext chunks
computed_ct = b"".join(ct_chunks) computed_ct = b"".join(ct_chunks)
@@ -170,7 +168,7 @@ def test_encrypt_decrypt_incremental(vector):
) )
# Incremental decryption with different random chunking # Incremental decryption with different random chunking
decryptor = alg.Decryptor(key, nonce, ad) decryptor = alg.Decryptor(key, nonce, ad, maclen=16)
pt_chunks = [] pt_chunks = []
for chunk in random_split_bytes(computed_ct): for chunk in random_split_bytes(computed_ct):
pt_chunks.append(bytes(decryptor.update(chunk))) pt_chunks.append(bytes(decryptor.update(chunk)))
@@ -187,14 +185,12 @@ def test_encrypt_decrypt_incremental(vector):
expected_tag256 = bytes.fromhex(vector["tag256"]) expected_tag256 = bytes.fromhex(vector["tag256"])
# Incremental encryption with random chunking # Incremental encryption with random chunking
encryptor = alg.Encryptor(key, nonce, ad) encryptor = alg.Encryptor(key, nonce, ad, maclen=32)
ct_chunks = [] ct_chunks = []
for chunk in random_split_bytes(msg): for chunk in random_split_bytes(msg):
ct_result = encryptor.update(chunk) ct_result = encryptor.update(chunk)
ct_chunks.append(bytes(ct_result)) ct_chunks.append(bytes(ct_result))
final_output = encryptor.final(maclen=32) computed_mac = bytes(encryptor.final())
ct_chunks.append(bytes(final_output[:-32])) # ciphertext part
computed_mac = bytes(final_output[-32:]) # MAC part
# Combine ciphertext chunks # Combine ciphertext chunks
computed_ct = b"".join(ct_chunks) computed_ct = b"".join(ct_chunks)
@@ -210,7 +206,7 @@ def test_encrypt_decrypt_incremental(vector):
) )
# Incremental decryption with different random chunking # Incremental decryption with different random chunking
decryptor = alg.Decryptor(key, nonce, ad) decryptor = alg.Decryptor(key, nonce, ad, maclen=32)
pt_chunks = [] pt_chunks = []
for chunk in random_split_bytes(computed_ct): for chunk in random_split_bytes(computed_ct):
pt_chunks.append(bytes(decryptor.update(chunk))) pt_chunks.append(bytes(decryptor.update(chunk)))
@@ -229,14 +225,14 @@ def test_encrypt_decrypt_incremental(vector):
# Test that incremental decryption fails with the provided (invalid) MACs # Test that incremental decryption fails with the provided (invalid) MACs
if "tag128" in vector: if "tag128" in vector:
invalid_mac = bytes.fromhex(vector["tag128"]) invalid_mac = bytes.fromhex(vector["tag128"])
decryptor = alg.Decryptor(key, nonce, ad) decryptor = alg.Decryptor(key, nonce, ad, maclen=16)
decryptor.update(ct) # This should succeed decryptor.update(ct) # This should succeed
with pytest.raises(ValueError, match="authentication failed"): with pytest.raises(ValueError, match="authentication failed"):
decryptor.final(invalid_mac) decryptor.final(invalid_mac)
if "tag256" in vector: if "tag256" in vector:
invalid_mac = bytes.fromhex(vector["tag256"]) invalid_mac = bytes.fromhex(vector["tag256"])
decryptor = alg.Decryptor(key, nonce, ad) decryptor = alg.Decryptor(key, nonce, ad, maclen=32)
decryptor.update(ct) # This should succeed decryptor.update(ct) # This should succeed
with pytest.raises(ValueError, match="authentication failed"): with pytest.raises(ValueError, match="authentication failed"):
decryptor.final(invalid_mac) decryptor.final(invalid_mac)
tests/test_mac.py
+5 -4
View File
@@ -2,6 +2,7 @@ import json
from pathlib import Path from pathlib import Path
import pytest import pytest
from pyaegis import aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4 from pyaegis import aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4
from .util import random_split_bytes from .util import random_split_bytes
@@ -81,10 +82,10 @@ def test_mac_class(vector):
# Test 128-bit MAC if present # Test 128-bit MAC if present
if "tag128" in vector: if "tag128" in vector:
expected_tag128 = bytes.fromhex(vector["tag128"]) expected_tag128 = bytes.fromhex(vector["tag128"])
mac_state = alg.Mac(key, nonce) mac_state = alg.Mac(key, nonce, maclen=16)
for chunk in random_split_bytes(data): for chunk in random_split_bytes(data):
mac_state.update(chunk) mac_state.update(chunk)
computed_tag128 = mac_state.final(maclen=16) computed_tag128 = mac_state.final()
assert computed_tag128 == expected_tag128, ( assert computed_tag128 == expected_tag128, (
f"128-bit MAC mismatch for {vector['name']}" f"128-bit MAC mismatch for {vector['name']}"
) )
@@ -92,10 +93,10 @@ def test_mac_class(vector):
# Test 256-bit MAC if present # Test 256-bit MAC if present
if "tag256" in vector: if "tag256" in vector:
expected_tag256 = bytes.fromhex(vector["tag256"]) expected_tag256 = bytes.fromhex(vector["tag256"])
mac_state = alg.Mac(key, nonce) mac_state = alg.Mac(key, nonce, maclen=32)
for chunk in random_split_bytes(data): for chunk in random_split_bytes(data):
mac_state.update(chunk) mac_state.update(chunk)
computed_tag256 = mac_state.final(maclen=32) computed_tag256 = mac_state.final()
assert computed_tag256 == expected_tag256, ( assert computed_tag256 == expected_tag256, (
f"256-bit MAC mismatch for {vector['name']}" f"256-bit MAC mismatch for {vector['name']}"
) )
tests/test_raises.py
+280
View File
@@ -0,0 +1,280 @@
"""Tests for Encryptor and Decryptor finalization behavior.
This module verifies that Encryptor and Decryptor objects become unusable
after calling final(), preventing accidental misuse.
"""
import pytest
from pyaegis import aegis256x4
class TestEncryptorFinalization:
"""Test that Encryptor becomes unusable after final()."""
def test_update_after_final_raises(self):
"""Test that calling update() after final() raises RuntimeError."""
key = aegis256x4.random_key()
nonce = aegis256x4.random_nonce()
encryptor = aegis256x4.Encryptor(key, nonce)
# Encrypt some data and finalize
encryptor.update(b"Hello, world!")
tag = encryptor.final()
# Attempting to update after final should raise RuntimeError
with pytest.raises(
RuntimeError, match="Cannot call update\\(\\) after final\\(\\)"
):
encryptor.update(b"More data")
def test_final_after_final_raises(self):
"""Test that calling final() after final() raises RuntimeError."""
key = aegis256x4.random_key()
nonce = aegis256x4.random_nonce()
encryptor = aegis256x4.Encryptor(key, nonce)
# Encrypt some data and finalize
encryptor.update(b"Hello, world!")
tag = encryptor.final()
# Attempting to call final again should raise RuntimeError
with pytest.raises(
RuntimeError, match="Cannot call final\\(\\) after final\\(\\)"
):
encryptor.final()
def test_update_then_final_after_final_raises(self):
"""Test that both update() and final() fail after final()."""
key = aegis256x4.random_key()
nonce = aegis256x4.random_nonce()
encryptor = aegis256x4.Encryptor(key, nonce)
# Encrypt and finalize
encryptor.update(b"Test data")
tag = encryptor.final()
# Both operations should fail
with pytest.raises(
RuntimeError, match="Cannot call update\\(\\) after final\\(\\)"
):
encryptor.update(b"More data")
with pytest.raises(
RuntimeError, match="Cannot call final\\(\\) after final\\(\\)"
):
encryptor.final()
def test_properties_accessible_after_final(self):
"""Test that properties like bytes_in and bytes_out are still accessible after final()."""
key = aegis256x4.random_key()
nonce = aegis256x4.random_nonce()
encryptor = aegis256x4.Encryptor(key, nonce)
message = b"Hello, world!"
encryptor.update(message)
tag = encryptor.final()
# Properties should still be accessible
assert encryptor.bytes_in == len(message)
assert encryptor.bytes_out == len(message) + len(tag)
def test_empty_encryption_finalization(self):
"""Test that finalization works correctly with no update() calls."""
key = aegis256x4.random_key()
nonce = aegis256x4.random_nonce()
encryptor = aegis256x4.Encryptor(key, nonce)
# Finalize without any updates
tag = encryptor.final()
assert len(tag) == aegis256x4.MACBYTES
# Should still be unusable after
with pytest.raises(
RuntimeError, match="Cannot call update\\(\\) after final\\(\\)"
):
encryptor.update(b"Data")
class TestDecryptorFinalization:
"""Test that Decryptor becomes unusable after final()."""
def test_update_after_final_raises(self):
"""Test that calling update() after final() raises RuntimeError."""
key = aegis256x4.random_key()
nonce = aegis256x4.random_nonce()
message = b"Hello, world!"
# Encrypt first to get valid ciphertext and tag
ct, tag = aegis256x4.encrypt_detached(key, nonce, message)
# Now test decryption
decryptor = aegis256x4.Decryptor(key, nonce)
decryptor.update(ct)
decryptor.final(tag)
# Attempting to update after final should raise RuntimeError
with pytest.raises(
RuntimeError, match="Cannot call update\\(\\) after final\\(\\)"
):
decryptor.update(b"More ciphertext")
def test_final_after_final_raises(self):
"""Test that calling final() after final() raises RuntimeError."""
key = aegis256x4.random_key()
nonce = aegis256x4.random_nonce()
message = b"Hello, world!"
# Encrypt first to get valid ciphertext and tag
ct, tag = aegis256x4.encrypt_detached(key, nonce, message)
# Now test decryption
decryptor = aegis256x4.Decryptor(key, nonce)
decryptor.update(ct)
decryptor.final(tag)
# Attempting to call final again should raise RuntimeError
with pytest.raises(
RuntimeError, match="Cannot call final\\(\\) after final\\(\\)"
):
decryptor.final(tag)
def test_update_then_final_after_final_raises(self):
"""Test that both update() and final() fail after final()."""
key = aegis256x4.random_key()
nonce = aegis256x4.random_nonce()
message = b"Test data"
# Encrypt first
ct, tag = aegis256x4.encrypt_detached(key, nonce, message)
# Decrypt and finalize
decryptor = aegis256x4.Decryptor(key, nonce)
decryptor.update(ct)
decryptor.final(tag)
# Both operations should fail
with pytest.raises(
RuntimeError, match="Cannot call update\\(\\) after final\\(\\)"
):
decryptor.update(b"More ciphertext")
with pytest.raises(
RuntimeError, match="Cannot call final\\(\\) after final\\(\\)"
):
decryptor.final(tag)
def test_properties_accessible_after_final(self):
"""Test that properties like bytes_in and bytes_out are still accessible after final()."""
key = aegis256x4.random_key()
nonce = aegis256x4.random_nonce()
message = b"Hello, world!"
# Encrypt first
ct, tag = aegis256x4.encrypt_detached(key, nonce, message)
# Decrypt
decryptor = aegis256x4.Decryptor(key, nonce)
decryptor.update(ct)
decryptor.final(tag)
# Properties should still be accessible
assert decryptor.bytes_in == len(ct)
assert decryptor.bytes_out == len(message)
def test_empty_decryption_finalization(self):
"""Test that finalization works correctly with no update() calls."""
key = aegis256x4.random_key()
nonce = aegis256x4.random_nonce()
# Encrypt empty message
ct, tag = aegis256x4.encrypt_detached(key, nonce, b"")
# Decrypt without any updates
decryptor = aegis256x4.Decryptor(key, nonce)
decryptor.final(tag) # Should work with empty ciphertext
# Should still be unusable after
with pytest.raises(
RuntimeError, match="Cannot call update\\(\\) after final\\(\\)"
):
decryptor.update(b"Data")
def test_failed_verification_still_finalizes(self):
"""Test that even if verification fails, the object becomes unusable."""
key = aegis256x4.random_key()
nonce = aegis256x4.random_nonce()
message = b"Hello, world!"
# Encrypt first
ct, tag = aegis256x4.encrypt_detached(key, nonce, message)
# Decrypt but use wrong tag
decryptor = aegis256x4.Decryptor(key, nonce)
decryptor.update(ct)
# Try to finalize with invalid tag - should raise ValueError
bad_tag = bytes(len(tag)) # All zeros
with pytest.raises(ValueError, match="authentication failed"):
decryptor.final(bad_tag)
# Object should NOT be finalized on failure - should still be usable
# This is a design decision: failed verification shouldn't lock the object
# Let's verify current behavior
try:
decryptor.update(b"test")
# If this doesn't raise, the object is still usable after failed verification
# This might be the desired behavior
except RuntimeError:
# If this raises, failed verification also finalizes the object
pass
class TestMultipleChunksBeforeFinalization:
"""Test that multiple update() calls work before final()."""
def test_encryptor_multiple_updates(self):
"""Test that Encryptor can handle multiple update() calls before final()."""
key = aegis256x4.random_key()
nonce = aegis256x4.random_nonce()
encryptor = aegis256x4.Encryptor(key, nonce)
# Multiple updates
encryptor.update(b"Hello, ")
encryptor.update(b"world!")
encryptor.update(b" More data.")
# Should still work
tag = encryptor.final()
assert len(tag) == aegis256x4.MACBYTES
# Now unusable
with pytest.raises(RuntimeError):
encryptor.update(b"More")
def test_decryptor_multiple_updates(self):
"""Test that Decryptor can handle multiple update() calls before final()."""
key = aegis256x4.random_key()
nonce = aegis256x4.random_nonce()
# Encrypt in chunks
encryptor = aegis256x4.Encryptor(key, nonce)
ct1 = encryptor.update(b"Hello, ")
ct2 = encryptor.update(b"world!")
tag = encryptor.final()
# Decrypt in chunks
decryptor = aegis256x4.Decryptor(key, nonce)
decryptor.update(ct1)
decryptor.update(ct2)
decryptor.final(tag)
# Now unusable
with pytest.raises(RuntimeError):
decryptor.update(b"More")