"""aegis256x4 Python submodule. Simplified API: single functions can return newly allocated buffers or write into user-provided buffers via optional `into=` (and `mac_into=` for detached). Error return codes from the C library raise ValueError. """ import errno from collections.abc import Buffer from ._loader import alloc_aligned, ffi, libc from ._loader import lib as _lib # Constants exposed as functions in C; mirror them as integers at module import time KEYBYTES = _lib.aegis256x4_keybytes() NPUBBYTES = _lib.aegis256x4_npubbytes() ABYTES_MIN = _lib.aegis256x4_abytes_min() ABYTES_MAX = _lib.aegis256x4_abytes_max() TAILBYTES_MAX = _lib.aegis256x4_tailbytes_max() def _ptr(buf): """Return an ffi pointer for a Python buffer, or NULL for None. Args: buf: Any object supporting the Python buffer protocol, or None. Returns: An ffi pointer obtained with ffi.from_buffer, or ffi.NULL when buf is None. """ return ffi.NULL if buf is None else ffi.from_buffer(buf) def encrypt_detached( nonce: Buffer, key: Buffer, message: Buffer, ad: Buffer | None = None, maclen: int = ABYTES_MIN, ct_into: Buffer | None = None, mac_into: Buffer | None = None, ) -> tuple[memoryview, memoryview]: """Encrypt message with associated data, returning ciphertext and MAC separately. Args: nonce: Nonce (32 bytes). key: Key (32 bytes). message: The plaintext message to encrypt. ad: Associated data (optional). maclen: MAC length (16 or 32, default 16). ct_into: Buffer to write ciphertext into (default: bytearray created). mac_into: Buffer to write MAC into (default: bytearray created). Returns: Tuple of (ciphertext, mac) Raises: TypeError: If lengths are invalid. RuntimeError: If encryption fails. """ nonce = memoryview(nonce) key = memoryview(key) message = memoryview(message) ad = memoryview(ad) if ad is not None else None ct_into = memoryview(ct_into) if ct_into is not None else None mac_into = memoryview(mac_into) if mac_into is not None else None if maclen not in (16, 32): raise TypeError("maclen must be 16 or 32") if key.nbytes != KEYBYTES: raise TypeError(f"key length must be {KEYBYTES}") if nonce.nbytes != NPUBBYTES: raise TypeError(f"nonce length must be {NPUBBYTES}") c = ct_into if ct_into is not None else memoryview(bytearray(message.nbytes)) mac = mac_into if mac_into is not None else memoryview(bytearray(maclen)) if c.nbytes != message.nbytes: raise TypeError("into length must equal len(message)") if mac.nbytes != maclen: raise TypeError("mac_into length must equal maclen") rc = _lib.aegis256x4_encrypt_detached( ffi.from_buffer(c), ffi.from_buffer(mac), maclen, _ptr(message), message.nbytes, _ptr(ad), 0 if ad is None else ad.nbytes, _ptr(nonce), _ptr(key), ) if rc != 0: err_num = ffi.errno err_name = errno.errorcode.get(err_num, f"errno_{err_num}") raise RuntimeError(f"encrypt detached failed: {err_name}") return c, mac def decrypt_detached( nonce: Buffer, key: Buffer, ct: Buffer, mac: Buffer, ad: Buffer | None = None, into: Buffer | None = None, ) -> memoryview: """Decrypt ciphertext with detached MAC and associated data. Args: nonce: Nonce (32 bytes). key: Key (32 bytes). ct: The ciphertext to decrypt. mac: The MAC to verify. ad: Associated data (optional). into: Buffer to write plaintext into (default: bytearray created). Returns: Plaintext as bytearray if into not provided. Raises: TypeError: If lengths are invalid. ValueError: If authentication fails. """ nonce = memoryview(nonce) key = memoryview(key) ct = memoryview(ct) mac = memoryview(mac) ad = memoryview(ad) if ad is not None else None into = memoryview(into) if into is not None else None if key.nbytes != KEYBYTES: raise TypeError(f"key length must be {KEYBYTES}") if nonce.nbytes != NPUBBYTES: raise TypeError(f"nonce length must be {NPUBBYTES}") maclen = mac.nbytes if maclen not in (16, 32): raise TypeError("mac length must be 16 or 32") m_out = into if into is not None else memoryview(bytearray(ct.nbytes)) if m_out.nbytes != ct.nbytes: raise TypeError("into length must equal len(ciphertext)") rc = _lib.aegis256x4_decrypt_detached( ffi.from_buffer(m_out), _ptr(ct), ct.nbytes, _ptr(mac), maclen, _ptr(ad), 0 if ad is None else ad.nbytes, _ptr(nonce), _ptr(key), ) if rc != 0: raise ValueError("authentication failed") return memoryview(m_out) def encrypt( nonce: Buffer, key: Buffer, message: Buffer, ad: Buffer | None = None, maclen: int = ABYTES_MIN, into: Buffer | None = None, ) -> memoryview: """Encrypt message with associated data, returning ciphertext with appended MAC. Args: nonce: Nonce (32 bytes). key: Key (32 bytes). message: The plaintext message to encrypt. ad: Associated data (optional). maclen: MAC length (16 or 32, default 16). into: Buffer to write ciphertext+MAC into (default: bytearray created). Returns: Ciphertext with appended MAC as bytearray if into not provided. Raises: TypeError: If lengths are invalid. RuntimeError: If encryption fails. """ nonce = memoryview(nonce) key = memoryview(key) message = memoryview(message) ad = memoryview(ad) if ad is not None else None into = memoryview(into) if into is not None else None if maclen not in (16, 32): raise TypeError("maclen must be 16 or 32") if key.nbytes != KEYBYTES: raise TypeError(f"key length must be {KEYBYTES}") if nonce.nbytes != NPUBBYTES: raise TypeError(f"nonce length must be {NPUBBYTES}") out = into if into is not None else memoryview(bytearray(message.nbytes + maclen)) if out.nbytes != message.nbytes + maclen: raise TypeError("into length must be len(message)+maclen") rc = _lib.aegis256x4_encrypt( ffi.from_buffer(out), maclen, _ptr(message), message.nbytes, _ptr(ad), 0 if ad is None else ad.nbytes, _ptr(nonce), _ptr(key), ) if rc != 0: err_num = ffi.errno err_name = errno.errorcode.get(err_num, f"errno_{err_num}") raise RuntimeError(f"encrypt failed: {err_name}") return out def decrypt( nonce: Buffer, key: Buffer, ct: Buffer, ad: Buffer | None = None, maclen: int = ABYTES_MIN, into: Buffer | None = None, ) -> memoryview: """Decrypt ciphertext with appended MAC and associated data. Args: nonce: Nonce (32 bytes). key: Key (32 bytes). ct: The ciphertext with MAC to decrypt. ad: Associated data (optional). maclen: MAC length (16 or 32, default 16). into: Buffer to write plaintext into (default: bytearray created). Returns: Plaintext as bytearray if into not provided. Raises: TypeError: If lengths are invalid. ValueError: If authentication fails. """ nonce = memoryview(nonce) key = memoryview(key) ct = memoryview(ct) ad = memoryview(ad) if ad is not None else None into = memoryview(into) if into is not None else None if maclen not in (16, 32): raise TypeError("maclen must be 16 or 32") if key.nbytes != KEYBYTES: raise TypeError(f"key length must be {KEYBYTES}") if nonce.nbytes != NPUBBYTES: raise TypeError(f"nonce length must be {NPUBBYTES}") if ct.nbytes < maclen: raise TypeError("ciphertext too short for tag") m_out = into if into is not None else memoryview(bytearray(ct.nbytes - maclen)) if m_out.nbytes != ct.nbytes - maclen: raise TypeError("into length must be len(ciphertext_with_tag)-maclen") rc = _lib.aegis256x4_decrypt( ffi.from_buffer(m_out), _ptr(ct), ct.nbytes, maclen, _ptr(ad), 0 if ad is None else ad.nbytes, _ptr(nonce), _ptr(key), ) if rc != 0: raise ValueError("authentication failed") return m_out def stream( nonce: Buffer | None, key: Buffer, length: int | None = None, into: Buffer | None = None, ) -> memoryview: """Generate a stream of pseudorandom bytes. Args: nonce: Nonce (32 bytes, uses zeroes for nonce if None). key: Key (32 bytes). length: Number of bytes to generate (required if into is None). into: Buffer to write stream into (default: bytearray created). Returns: Pseudorandom bytes as bytearray if into not provided. Raises: TypeError: If lengths are invalid or neither length nor into provided. """ nonce = memoryview(nonce) if nonce is not None else None key = memoryview(key) into = memoryview(into) if into is not None else None if key.nbytes != KEYBYTES: raise TypeError(f"key length must be {KEYBYTES}") if nonce is not None and nonce.nbytes != NPUBBYTES: raise TypeError(f"nonce length must be {NPUBBYTES}") if into is None and length is None: raise TypeError("provide either into or length") out = into if into is not None else memoryview(bytearray(int(length or 0))) _lib.aegis256x4_stream( ffi.from_buffer(out), out.nbytes, _ptr(nonce), _ptr(key), ) return out def encrypt_unauthenticated( message: Buffer, nonce: Buffer, key: Buffer, into: Buffer | None = None, ) -> memoryview: """Encrypt message without authentication (for testing/debugging). Args: message: The plaintext message to encrypt. nonce: Nonce (32 bytes). key: Key (32 bytes). into: Buffer to write ciphertext into (default: bytearray created). Returns: Ciphertext as bytearray if into not provided. Raises: TypeError: If lengths are invalid. """ message = memoryview(message) nonce = memoryview(nonce) key = memoryview(key) into = memoryview(into) if into is not None else None if key.nbytes != KEYBYTES: raise TypeError(f"key length must be {KEYBYTES}") if nonce.nbytes != NPUBBYTES: raise TypeError(f"nonce length must be {NPUBBYTES}") out = into if into is not None else memoryview(bytearray(message.nbytes)) if out.nbytes != message.nbytes: raise TypeError("into length must equal len(message)") _lib.aegis256x4_encrypt_unauthenticated( ffi.from_buffer(out), _ptr(message), message.nbytes, _ptr(nonce), _ptr(key), ) return out def decrypt_unauthenticated( ct: Buffer, nonce: Buffer, key: Buffer, into: Buffer | None = None, ) -> memoryview: """Decrypt ciphertext without authentication (for testing/debugging). Args: ct: The ciphertext to decrypt. nonce: Nonce (32 bytes). key: Key (32 bytes). into: Buffer to write plaintext into (default: bytearray created). Returns: Plaintext as bytearray if into not provided. Raises: TypeError: If lengths are invalid. """ ct = memoryview(ct) nonce = memoryview(nonce) key = memoryview(key) into = memoryview(into) if into is not None else None if key.nbytes != KEYBYTES: raise TypeError(f"key length must be {KEYBYTES}") if nonce.nbytes != NPUBBYTES: raise TypeError(f"nonce length must be {NPUBBYTES}") out = into if into is not None else memoryview(bytearray(ct.nbytes)) if out.nbytes != ct.nbytes: raise TypeError("into length must equal len(ciphertext)") _lib.aegis256x4_decrypt_unauthenticated( ffi.from_buffer(out), _ptr(ct), ct.nbytes, _ptr(nonce), _ptr(key), ) return out # This is missing from C API but convenient to have here def mac( data: Buffer, nonce: Buffer, key: Buffer, maclen: int = ABYTES_MIN, ) -> memoryview: """Compute a MAC for the given data in one shot. Args: data: Data to MAC nonce: Nonce (32 bytes) key: Key (32 bytes) maclen: MAC length (16 or 32, default 16) Returns: MAC bytes """ mac_state = Mac(nonce, key) mac_state.update(data) return mac_state.final(maclen) class Mac: """AEGIS-256X4 MAC state wrapper. Usage: mac = Mac(nonce, key) mac.update(data) tag = mac.final() # defaults to 16-byte MAC # or verify: mac2 = Mac(nonce, key); mac2.update(data); mac2.verify(tag) """ __slots__ = ("_st", "_nonce", "_key") def __init__( self, nonce: Buffer, key: Buffer, _other=None, ) -> None: """Initialize a MAC state with a nonce and key. Args: nonce: Nonce (32 bytes). key: Key (32 bytes). Raises: TypeError: If key or nonce lengths are invalid. """ raw = alloc_aligned(ffi.sizeof("aegis256x4_mac_state"), 64) st = ffi.cast("aegis256x4_mac_state *", raw) self._st = ffi.gc(st, libc.free) if _other is not None: _lib.aegis256x4_mac_state_clone(self._st, _other._st) return # Normal init nonce = memoryview(nonce) key = memoryview(key) if key.nbytes != KEYBYTES: raise TypeError(f"key length must be {KEYBYTES=}") if nonce.nbytes != NPUBBYTES: raise TypeError(f"nonce length must be {NPUBBYTES=}") _lib.aegis256x4_mac_init(self._st, _ptr(key), _ptr(nonce)) def __deepcopy__(self) -> "Mac": """Return a clone of current MAC state.""" return Mac(b"", b"", _other=self) clone = __deepcopy__ def reset(self) -> None: """Reset the MAC state so it can be reused with the same nonce and key.""" _lib.aegis256x4_mac_reset(self._st) def update(self, data: Buffer) -> None: """Absorb data into the MAC state. Args: data: Bytes-like object to authenticate. Raises: RuntimeError: If the underlying C function reports an error. """ data = memoryview(data) rc = _lib.aegis256x4_mac_update(self._st, _ptr(data), data.nbytes) if rc != 0: err_num = ffi.errno err_name = errno.errorcode.get(err_num, f"errno_{err_num}") raise RuntimeError(f"mac update failed: {err_name}") def final( self, maclen: int = ABYTES_MIN, into: Buffer | None = None, ) -> memoryview: """Finalize and return the MAC tag. Args: maclen: Tag length in bytes (16 or 32). Defaults to 16. into: Optional buffer to write the tag into (default: bytearray created). Returns: The tag as a memoryview; if ``into`` is provided, it views that buffer. Raises: TypeError: If lengths are invalid. RuntimeError: If finalization fails in the C library. """ if maclen not in (16, 32): raise TypeError("maclen must be 16 or 32") out = into if into is not None else bytearray(maclen) out = memoryview(out) if out.nbytes != maclen: raise TypeError("into length must equal maclen") rc = _lib.aegis256x4_mac_final(self._st, ffi.from_buffer(out), maclen) if rc != 0: err_num = ffi.errno err_name = errno.errorcode.get(err_num, f"errno_{err_num}") raise RuntimeError(f"mac final failed: {err_name}") return out def verify(self, mac: Buffer) -> bool: """Verify a tag for the current MAC state. Args: mac: The tag to verify (16 or 32 bytes). Returns: True if verification succeeds. Raises: TypeError: If tag length is invalid. ValueError: If verification fails. """ mac = memoryview(mac) maclen = mac.nbytes if maclen not in (16, 32): raise TypeError("mac length must be 16 or 32") rc = _lib.aegis256x4_mac_verify(self._st, _ptr(mac), maclen) if rc != 0: raise ValueError("mac verification failed") return True class Encryptor: """Incremental encryptor. - update(message[, into]) -> returns produced ciphertext bytes - final([into], maclen=16) -> returns tail+tag bytes - final_detached([ct_into], [mac_into], maclen=16) -> returns (tail_bytes, mac) """ __slots__ = ("_st",) def __init__(self, nonce: Buffer, key: Buffer, ad: Buffer | None = None): """Create an incremental encryptor. Args: nonce: Nonce (32 bytes). key: Key (32 bytes). ad: Associated data to bind to the encryption (optional). Raises: TypeError: If key or nonce lengths are invalid. """ key = memoryview(key) nonce = memoryview(nonce) if key.nbytes != KEYBYTES: raise TypeError(f"key length must be {KEYBYTES}") if nonce.nbytes != NPUBBYTES: raise TypeError(f"nonce length must be {NPUBBYTES}") raw = alloc_aligned(ffi.sizeof("aegis256x4_state"), 64) st = ffi.cast("aegis256x4_state *", raw) st = ffi.gc(st, libc.free) _lib.aegis256x4_state_init( st, _ptr(ad) if ad is not None else ffi.NULL, 0 if ad is None else memoryview(ad).nbytes, _ptr(nonce), _ptr(key), ) self._st = st def update(self, message: Buffer, into: Buffer | None = None) -> memoryview: """Encrypt a chunk of the message. Args: message: Plaintext bytes to encrypt. into: Optional destination buffer; must be >= len(message). Returns: The ciphertext for this chunk as a memoryview; when ``into`` is provided, a view of that buffer up to the number of bytes written. Raises: TypeError: If destination buffer is too small. RuntimeError: If the C update call fails. """ message = memoryview(message) out = memoryview(into if into is not None else bytearray(message.nbytes)) if out.nbytes < message.nbytes: raise TypeError("into length must be >= len(message)") written = ffi.new("size_t *") rc = _lib.aegis256x4_state_encrypt_update( self._st, ffi.from_buffer(out), out.nbytes, written, _ptr(message), message.nbytes, ) if rc != 0: err_num = ffi.errno err_name = errno.errorcode.get(err_num, f"errno_{err_num}") raise RuntimeError(f"state encrypt update failed: {err_name}") return out[: int(written[0])] def final(self, into: Buffer | None = None, maclen: int = ABYTES_MIN) -> memoryview: """Finalize encryption, writing any remaining bytes and the tag. Args: into: Optional destination buffer for the tail and tag. maclen: Tag length (16 or 32). Defaults to 16. Returns: A memoryview of the produced bytes (tail + tag). When ``into`` is provided, the returned view references that buffer up to the number of bytes written. Raises: TypeError: If maclen is invalid. RuntimeError: If the C final call fails. """ if maclen not in (16, 32): raise TypeError("maclen must be 16 or 32") # Worst-case final length is leftover tail (<= TAILBYTES_MAX) plus tag out = into if into is not None else bytearray(TAILBYTES_MAX + maclen) out = memoryview(out) written = ffi.new("size_t *") rc = _lib.aegis256x4_state_encrypt_final( self._st, ffi.from_buffer(out), out.nbytes, written, maclen, ) if rc != 0: err_num = ffi.errno err_name = errno.errorcode.get(err_num, f"errno_{err_num}") raise RuntimeError(f"state encrypt final failed: {err_name}") return out[: int(written[0])] def final_detached( self, ct_into: bytearray | None = None, mac_into: bytearray | None = None, maclen: int = ABYTES_MIN, ) -> tuple[bytearray, bytearray]: """Finalize encryption, producing detached tail bytes and tag. Args: ct_into: Optional destination for the remaining ciphertext tail. mac_into: Optional destination for the tag. maclen: Tag length (16 or 32). Defaults to 16. Returns: A tuple of (tail_bytes, mac). When destination buffers are provided, the first element is a slice of ``ct_into`` up to the number of bytes written, and the second is ``mac_into``. Raises: TypeError: If maclen is invalid or mac_into has the wrong length. RuntimeError: If the C final call fails. """ if maclen not in (16, 32): raise TypeError("maclen must be 16 or 32") out = ct_into if ct_into is not None else bytearray(TAILBYTES_MAX) mac = mac_into if mac_into is not None else bytearray(maclen) if len(mac) != maclen: raise TypeError("mac_into length must equal maclen") written = ffi.new("size_t *") rc = _lib.aegis256x4_state_encrypt_detached_final( self._st, ffi.from_buffer(out), len(out), written, ffi.from_buffer(mac), maclen, ) if rc != 0: err_num = ffi.errno err_name = errno.errorcode.get(err_num, f"errno_{err_num}") raise RuntimeError(f"state encrypt detached final failed: {err_name}") return out[: int(written[0])], mac class Decryptor: """Incremental decryptor. - update(ciphertext[, into]) -> returns plaintext bytes - final(mac[, into]) -> returns any remaining plaintext bytes """ __slots__ = ("_st",) def __init__(self, nonce: Buffer, key: Buffer, ad: Buffer | None = None): """Create an incremental decryptor for detached tags. Args: nonce: Nonce (32 bytes). key: Key (32 bytes). ad: Associated data used during encryption (optional). Raises: TypeError: If key or nonce lengths are invalid. """ key = memoryview(key) nonce = memoryview(nonce) if key.nbytes != KEYBYTES: raise TypeError(f"key length must be {KEYBYTES}") if nonce.nbytes != NPUBBYTES: raise TypeError(f"nonce length must be {NPUBBYTES}") raw = alloc_aligned(ffi.sizeof("aegis256x4_state"), 64) st = ffi.cast("aegis256x4_state *", raw) st = ffi.gc(st, libc.free) _lib.aegis256x4_state_init( st, _ptr(ad) if ad is not None else ffi.NULL, 0 if ad is None else memoryview(ad).nbytes, _ptr(nonce), _ptr(key), ) self._st = st def update(self, ct: Buffer, into: Buffer | None = None) -> memoryview: """Process a chunk of ciphertext. Args: ct: Ciphertext bytes (without MAC). into: Optional destination buffer; must be >= len(ciphertext). Returns: A memoryview of the decrypted bytes for this chunk. When ``into`` is provided, the returned view references that buffer up to the number of bytes written. Raises: TypeError: If destination buffer is too small. RuntimeError: If the C update call fails. """ ct = memoryview(ct) out = into if into is not None else bytearray(ct.nbytes) out = memoryview(out) if out.nbytes < ct.nbytes: raise TypeError("into length must be >= len(ciphertext)") written = ffi.new("size_t *") rc = _lib.aegis256x4_state_decrypt_detached_update( self._st, ffi.from_buffer(out), out.nbytes, written, _ptr(ct), ct.nbytes, ) if rc != 0: err_num = ffi.errno err_name = errno.errorcode.get(err_num, f"errno_{err_num}") raise RuntimeError(f"state decrypt update failed: {err_name}") w = int(written[0]) return out[:w] def final(self, mac: Buffer, into: Buffer | None = None) -> memoryview: """Finalize decryption by verifying tag and flushing remaining bytes. Args: mac: Tag to verify (16 or 32 bytes). into: Optional destination buffer for the remaining plaintext bytes. Returns: A memoryview of the remaining plaintext bytes. When ``into`` is provided, the returned view references that buffer up to the number of bytes written. Raises: TypeError: If tag length is invalid. ValueError: If authentication fails. """ mac = memoryview(mac) maclen = mac.nbytes if maclen not in (16, 32): raise TypeError("mac length must be 16 or 32") out = into if into is not None else bytearray(TAILBYTES_MAX) out = memoryview(out) written = ffi.new("size_t *") rc = _lib.aegis256x4_state_decrypt_detached_final( self._st, ffi.from_buffer(out), out.nbytes, written, _ptr(mac), maclen, ) if rc != 0: raise ValueError("authentication failed") w = int(written[0]) return out[:w] def new_state(): """Allocate and return a new aegis256x4_state* with proper alignment. The returned object is an ffi cdata pointer with automatic finalizer. """ # Allocate with 64-byte alignment using libc.posix_memalign raw = alloc_aligned(ffi.sizeof("aegis256x4_state"), 64) ptr = ffi.cast("aegis256x4_state *", raw) return ffi.gc(ptr, libc.free) def new_mac_state(): """Allocate and return a new aegis256x4_mac_state* with proper alignment.""" raw = alloc_aligned(ffi.sizeof("aegis256x4_mac_state"), 64) ptr = ffi.cast("aegis256x4_mac_state *", raw) return ffi.gc(ptr, libc.free) __all__ = [ # constants "KEYBYTES", "NPUBBYTES", "ABYTES_MIN", "ABYTES_MAX", "TAILBYTES_MAX", # one-shot functions "encrypt_detached", "decrypt_detached", "encrypt", "decrypt", "stream", "encrypt_unauthenticated", "decrypt_unauthenticated", "mac", # incremental classes "Encryptor", "Decryptor", "Mac", ]