70 Commits
Author SHA1 Message Date
LeoVasanko a1bfeb2434 Build updated for upstream libaegis 0.9.0 simplified API with matching changes on Python side. Set lower MacOS build target (11.0). Build abi3 version for all supported Python versions, reducing the number of wheels. Other build cleanup. 2026-01-10 20:23:37 +00:00
LeoVasanko a26ce6a9fb Considered stable from here onwards. 2026-01-10 20:22:44 +00:00
LeoVasanko 7e6a5bf52a Allow the cipher() function also work without prefix aegis 2026-01-04 14:25:39 +00:00
LeoVasanko 8018e68274 Add nonce_increment and wipe functions on typing. 2026-01-04 14:19:01 +00:00
LeoVasanko ccccfd7918 README: Add the new aeg.cipher() function for fetching variants. 2025-12-31 22:51:19 +00:00
LeoVasanko 4de66b6bc6 [packaging] Release script: Fix further problems with freethreaded wheels and a variable name overlap. 2025-12-31 22:30:28 +00:00
LeoVasanko 66338ef416 [packaging] Fix wheel curing for pypy-linux. 2025-12-31 21:59:37 +00:00
LeoVasanko 509898ac73 Make tool scripts executable. 2025-12-31 21:43:25 +00:00
LeoVasanko 67ec091bc5 Add aeg.cipher(alg) for loading modules by name in str. Added aeg.CIPHERS mapping of the nominal names to module names and Cipher for typing literals. 2025-12-31 21:38:30 +00:00
LeoVasanko 4977a7e79d Added free-threaded non-GIL Python and PyPy wheels, up to 3.15 now. 2025-12-31 21:24:56 +00:00
LeoVasanko d7acf7a315 [release script] Auditwheel only on Linux. 2025-12-23 19:38:59 +00:00
LeoVasanko 7f01de8327 Add README for PyPI. 2025-12-23 19:33:00 +00:00
LeoVasanko 9294fba86d Make manylinux wheels. 2025-12-23 19:20:17 +00:00
LeoVasanko 348812e750 Clean up for release. 2025-12-23 19:09:59 +00:00
LeoVasanko 22dff32b22 Clean up for release. 2025-12-23 19:08:10 +00:00
LeoVasanko 178b689a35 Clean up for release. 2025-12-23 18:54:03 +00:00
LeoVasanko bd8b183e32 Project renamed to aeg, preparing for PyPI release. 2025-12-23 18:25:03 +00:00
LeoVasanko daba717ebb Add tools/release.py script for automatic release process. 2025-11-12 23:47:26 +00:00
LeoVasanko 5b74bcb683 Always generate LF line endings (even on broken OS). 2025-11-12 23:43:25 +00:00
LeoVasanko d9a18f43d5 Ruff formatting. 2025-11-12 23:43:04 +00:00
LeoVasanko 98ee4ab7b3 Fixed Windows build, cleaned up messages during build, using SCM version numbering, added ruff. 2025-11-12 22:59:21 +00:00
LeoVasanko c6148806dd Update pyproject, bump version. 2025-11-10 08:41:40 +00:00
LeoVasanko 6e6bc49466 README tuning. New benchmark results (a bit slower than initial versions were). 2025-11-10 08:40:04 +00:00
LeoVasanko c2ac2e2790 Use a much faster method to wipe buffers. 2025-11-10 08:38:28 +00:00
LeoVasanko c6e77cac66 Cleanup. 2025-11-10 08:00:12 +00:00
LeoVasanko df1acadb1f Bump version 2025-11-09 21:47:08 +00:00
LeoVasanko cb0f9956c3 Convert all input buffers to memoryview before use and use .nbytes, because len() doesn't work correctly with some buffers. Update docs with a Numpy example. 2025-11-09 21:19:50 +00:00
LeoVasanko a29405455a Cleanup benchmark 2025-11-09 20:59:53 +00:00
LeoVasanko 0d63844201 Make Mac class prevent further updates or final after finalisation. Keep cached values for hashlib API. 2025-11-09 20:53:19 +00:00
LeoVasanko 7682d5bc8b Cleanup 2025-11-09 08:39:41 +00:00
LeoVasanko 26b883ecbc Simplify implementation: remove bytes_in and bytes_out counters from all classes. 2025-11-09 08:09:26 +00:00
LeoVasanko 87d9a07868 Wipe state structs automatically after use. Simplified aligned allocator and its use via a single handle. 2025-11-09 08:04:00 +00:00
LeoVasanko eed36e2463 API updates:
- 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.
2025-11-09 06:53:27 +00:00
LeoVasanko 46bc05f547 Add human-readable algorithm name as NAME constant. 2025-11-09 04:01:05 +00:00
LeoVasanko 56d304a82e Constants renamed and values extracted from C code rather than function call at runtime. Documentation update. 2025-11-09 03:43:01 +00:00
LeoVasanko 28ee0e1314 Combine the two generator scripts into one that also reads ALIGNMENT and RATE from C sources. 2025-11-09 01:20:15 +00:00
LeoVasanko 1f506806d4 Bump version 2025-11-08 00:51:40 +00:00
LeoVasanko 17fba3c2f2 README 2025-11-07 22:30:23 +00:00
LeoVasanko d04c593766 Move the build_backend script to tools folder. 2025-11-07 22:26:03 +00:00
LeoVasanko fce6324ae5 Documentation updates, added BUILD.md 2025-11-07 22:20:27 +00:00
LeoVasanko b35dc75513 Use uv in the commands. 2025-11-07 22:00:36 +00:00
LeoVasanko 1da3000a22 README formatting 2025-11-07 21:59:20 +00:00
LeoVasanko f983cbe4c5 util.nonce_increment and util.wipe functions, also exported by each module for convenience. 2025-11-07 21:53:39 +00:00
LeoVasanko f0b4373c0e README updated 2025-11-07 21:52:28 +00:00
LeoVasanko 0efc2595d8 Make random_key and random_nonce return bytearray, and add a nonce_increment utility function. 2025-11-07 19:43:33 +00:00
LeoVasanko a78f5c4ea3 README 2025-11-07 09:11:38 +00:00
LeoVasanko fc1025f56d Do not return True from MAC verify function on success. 2025-11-07 09:11:24 +00:00
LeoVasanko 3d75cc7e55 Add random_key and random_nonce functions to each submodule. Update docstrings to show the correct lengths for keys and nonces. 2025-11-07 08:03:56 +00:00
LeoVasanko 5b49422cb7 Remove examples folder, keep benchmark as pyaegis.benchmark that becomes part of the install. 2025-11-07 07:55:51 +00:00
LeoVasanko cca78f4224 API cleanup for simplified update/final. Returns bytearrays when into is not used. Allows into buffers larger than needed. Misc other changes. 2025-11-07 07:38:18 +00:00
LeoVasanko 2e460b5796 Cleanup. 2025-11-07 07:36:43 +00:00
LeoVasanko 407cedc7a9 More informative build error message when Zig is not installed. 2025-11-07 05:02:01 +00:00
LeoVasanko b5b939cffe Another approach to use zig build with setuptools. 2025-11-07 04:56:39 +00:00
LeoVasanko 5c04940525 Include zig build in the build process. 2025-11-07 04:43:54 +00:00
LeoVasanko e644f1b2ca Switch from hatch to setuptools/CFFI build to produce wheels correctly. 2025-11-07 04:34:32 +00:00
LeoVasanko 56b451ac0c Test cleanup, remove debug. 2025-11-07 04:33:51 +00:00
LeoVasanko 93bdbc26a2 Remove Encryptor.final_detached(), as final() handles both cases identically (only returns MAC). 2025-11-07 04:33:04 +00:00
LeoVasanko 2612dc8187 The update functions of libaegis now return the same number of bytes that went in. Update the binding accordingly. 2025-11-07 04:27:48 +00:00
LeoVasanko 302429b7a4 Add comments to generated files saying how they are generated. 2025-11-07 03:31:29 +00:00
LeoVasanko b83120187c Build cleanup, pathname reorganisation, cdef to repository (although generated with gen_cdef.py). 2025-11-07 03:23:30 +00:00
LeoVasanko e3508025f5 Add .gitignore 2025-11-06 22:52:23 +00:00
LeoVasanko 3767b34e2c Add libaegis submodule 2025-11-06 22:51:32 +00:00
LeoVasanko 0ae8752666 Rename module to pyaegis, implement build with zig. 2025-11-06 21:34:15 +00:00
LeoVasanko d51736f5ba Add pytest dev dependency. 2025-11-06 20:20:23 +00:00
LeoVasanko df6b1b2832 Add tests for all functions of each algorithm against AEGIS test vectors (JSON). Incremental updates are tested with randomized splits to catch a variety of errors. 2025-11-06 19:56:12 +00:00
LeoVasanko c16e785203 Precise update output length calculation to allocate exactly sized buffers. 2025-11-06 19:45:15 +00:00
LeoVasanko 2ffe7b6c0e Add helper function for calculating the incremental update output buffer size. Add bytes_in and bytes_out counters on the incremental classes. Avoid ERANGE errors caused by too small output buffer. 2025-11-06 03:13:46 +00:00
LeoVasanko 618d9e0967 Make key, nonce the first arguments (in this order), require kwargs past the initial few positional arguments. Add ALIGNMENT constant to each module. Add a script to generate all other modules from aegis256x4.py. 2025-11-05 08:26:47 +00:00
LeoVasanko 299586e6ee Cleanup 2025-11-05 06:40:22 +00:00
LeoVasanko 22299a1676 Initial commit 2025-11-05 06:14:07 +00:00
28 changed files with 951 additions and 652 deletions
+3 -3
View File
@@ -3,9 +3,9 @@
*.egg-info *.egg-info
/dist /dist
/build /build
/pyaegis/build /src/aeg/build
/pyaegis/_aegis.*.so /src/aeg/_aegis*.so
/pyaegis/_aegis.*.pyd /src/aeg/_aegis*.pyd
__pycache__ __pycache__
!.gitignore !.gitignore
!.gitmodules !.gitmodules
+7 -7
View File
@@ -1,6 +1,6 @@
# Building pyaegis # Building aeg
This document contains instructions for developers who want to build pyaegis from source. This document contains instructions for developers who want to build aeg from source.
## Prerequisites ## Prerequisites
@@ -11,7 +11,7 @@ This document contains instructions for developers who want to build pyaegis fro
### Installing Zig ### Installing Zig
pyaegis uses Zig to build the underlying libaegis C library. Install Zig from [ziglang.org/download](https://ziglang.org/download/) or using your package manager: aeg uses Zig to build the underlying libaegis C library. Install Zig from [ziglang.org/download](https://ziglang.org/download/) or using your package manager:
- **macOS**: `brew install zig` - **macOS**: `brew install zig`
- **Linux**: See [Zig installation guide](https://github.com/ziglang/zig/wiki/Install-Zig-from-a-Package-Manager) - **Linux**: See [Zig installation guide](https://github.com/ziglang/zig/wiki/Install-Zig-from-a-Package-Manager)
@@ -22,8 +22,8 @@ pyaegis uses Zig to build the underlying libaegis C library. Install Zig from [z
Clone the repository with submodules: Clone the repository with submodules:
```fish ```fish
git clone --recursive https://github.com/LeoVasanko/pyaegis.git git clone --recursive https://github.com/LeoVasanko/aeg.git
cd pyaegis cd aeg
``` ```
If you already cloned without `--recursive`, initialize submodules: If you already cloned without `--recursive`, initialize submodules:
@@ -74,7 +74,7 @@ This creates files in the `dist/` directory.
## Code Generation ## Code Generation
The Python modules and CFFI definitions are generated from C sources and templates. If you modify the core implementation in `pyaegis/aegis256x4.py` or update libaegis headers, regenerate all files: The Python modules and CFFI definitions are generated from C sources and templates. If you modify the core implementation in `src/aeg/aegis256x4.py` or update libaegis headers, regenerate all files:
```fish ```fish
python tools/generate.py python tools/generate.py
@@ -100,7 +100,7 @@ If you cannot install Zig, you may manually compile in the libaegis folder (Zig,
## Project Structure ## Project Structure
- `pyaegis/` - Python package source - `src/aeg/` - Python package source
- `libaegis/` - C library source (submodule) - `libaegis/` - C library source (submodule)
- `tests/` - Test suite - `tests/` - Test suite
- `tools/` - Code generation scripts and `build_backend.py` used to build libaegis - `tools/` - Code generation scripts and `build_backend.py` used to build libaegis
+1 -3
View File
@@ -1,13 +1,11 @@
include pyaegis/aegis_cdef.h include src/aeg/aegis_cdef.h
include setup.py include setup.py
include tools/build_backend.py include tools/build_backend.py
include BUILD.md include BUILD.md
include README.md include README.md
recursive-include libaegis *.c *.h *.zig *.zon recursive-include libaegis *.c *.h *.zig *.zon
include libaegis/CMakeLists.txt
include libaegis/LICENSE include libaegis/LICENSE
include libaegis/README.md include libaegis/README.md
recursive-include libaegis/cmake *.cmake *.cmake.in
graft libaegis/src graft libaegis/src
include libaegis/build.zig include libaegis/build.zig
include libaegis/build.zig.zon include libaegis/build.zig.zon
+74 -60
View File
@@ -1,33 +1,28 @@
# pyaegis # AEGIS Cipher Python Binding
[![PyPI version](https://badge.fury.io/py/pyaegis.svg)](https://badge.fury.io/py/pyaegis) [![PyPI version](https://badge.fury.io/py/aeg.svg)](https://badge.fury.io/py/aeg)
Safe Python bindings for the AEGIS family of very fast authenticated encryption algorithms (via libaegis). Safe Python bindings for the AEGIS family of very fast authenticated encryption algorithms via libaegis. The module runs without compilation required on Windows, Mac and Linux (has precompiled wheels). For other platforms compilation is performed at install time.
AEGIS enables extremely fast Encryption, MAC and CSPRNG - many times faster than AES, ChaCha20 or traditional random number generators. Authenticated Encryption with Additional Data is supported with the MAC derived from the cipher state at the end, making it different from other AEADs like AES-GCM and ChaCha20-Poly1305. The whole internal state thus depends on the prior data, and it is neither Encrypt-Then-Mac nor Mac-The-Encrypt scheme when both features are used together.
## Install ## Install
Using [uv](https://docs.astral.sh/uv/getting-started/installation/): ```sh
```fish pip install aeg
uv pip install git+https://github.com/LeoVasanko/pyaegis.git
``` ```
For development builds, see BUILD.md. Or add to your project using [UV](https://docs.astral.sh/uv/getting-started/installation/):
```sh
## Variants uv add aeg
```
All submodules expose the same API; pick one for your key/nonce size and platform:
- aegis128l (16-byte key, 16-byte nonce)
- aegis256 (32-byte key, 32-byte nonce)
- aegis128x2 / aegis128x4 (multi-lane 128-bit; best throughput on SIMD-capable CPUs)
- aegis256x2 / aegis256x4 (multi-lane 256-bit)
## Quick start ## Quick start
Normal authenticated encryption using the AEGIS-128X4 algorithm: Normal authenticated encryption using the AEGIS-128X4 algorithm:
```python ```python
from pyaegis import aegis128x4 as ciph from aeg import aegis128x4 as ciph
key = ciph.random_key() # Secret key (stored securely) key = ciph.random_key() # Secret key (stored securely)
nonce = ciph.random_nonce() # Public nonce (recreated for each message) nonce = ciph.random_nonce() # Public nonce (recreated for each message)
@@ -38,6 +33,27 @@ pt = ciph.decrypt(key, nonce, ct) # Raises ValueError if anything was tampered
assert pt == msg assert pt == msg
``` ```
## Variants
All submodules expose the same API; pick one for your needs. The 256 bit variants offer maximal security and use larger key and nonce, while the 128 bit variants run slightly faster and use smaller key and nonce while still providing strong security. The MAC length does not depend on the variant. Note that the x2 and x4 variants are typically the fastest (depending on CPU) by utilizing SIMD multi-lane processing for the highest throughput.
| Variant | Key/Nonce Bytes | Notes |
|----------------|----------------:|-------------------------|
| **aegis128l** | 16 | |
| **aegis128x2** | 16 | Fastest on Intel Core |
| **aegis128x4** | 16 | Fastest on AMD and Xeon |
| **aegis256** | 32 | |
| **aegis256x2** | 32 | Fast on Intel Core |
| **aegis256x4** | 32 | Fast on AMD and Xeon |
Instead of importing the submodules, you can obtain one by its name string:
```python
import aeg
ciph = aeg.cipher("AEGIS-128X2") # Also accepts "aegis128x2" and other forms
```
## API overview ## API overview
Common parameters and returns (applies to all items below): Common parameters and returns (applies to all items below):
@@ -49,7 +65,7 @@ Common parameters and returns (applies to all items below):
- into: optional output buffer (see below) - into: optional output buffer (see below)
- maclen: MAC tag length 16 or 32 bytes (default 16) - maclen: MAC tag length 16 or 32 bytes (default 16)
Only the first few can be positional arguments that are always provided in this order. All arguments can be passed as kwargs. The inputs can be any Buffer supporting len() (e.g. `bytes`, `bytearray`, `memoryview`). Only the first few can be positional arguments that are always provided in this order. All arguments can be passed as kwargs. The inputs can be any Buffer (e.g. `bytes`, `bytearray`, `memoryview`).
Most functions return a buffer of bytes. By default a `bytearray` of the correct size is returned. An existing buffer can be provided by `into` argument, in which case the bytes of it that were written to are returned as a memoryview. Most functions return a buffer of bytes. By default a `bytearray` of the correct size is returned. An existing buffer can be provided by `into` argument, in which case the bytes of it that were written to are returned as a memoryview.
@@ -83,13 +99,13 @@ The object releases its state and becomes unusable after final has been called.
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 bytes
- Mac(key, nonce, maclen=16) - Mac(key, nonce, maclen=16)
- update(data) - update(data)
- final([into]) -> mac - final([into]) -> mac bytes
- verify(mac) -> raises ValueError on failure - verify(mac) -> raises ValueError on failure
- digest() -> bytes - digest() -> mac bytes
- hexdigest() -> str - hexdigest() -> mac str
- reset() - reset()
- clone() -> Mac - clone() -> Mac
@@ -123,7 +139,7 @@ Constants (per module): NAME, KEYBYTES, NONCEBYTES, MACBYTES, MACBYTES_LONG, RAT
A cryptographically secure keyed hash is produced. The example uses all zeroes for the nonce to always produce the same hash for the same key: A cryptographically secure keyed hash is produced. The example uses all zeroes for the nonce to always produce the same hash for the same key:
```python ```python
from pyaegis import aegis256x4 as ciph from aeg 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)
@@ -146,7 +162,7 @@ b.verify(mac) # Raises ValueError
Keeping the ciphertext, mac and ad separate. The ad represents a file header that needs to be tamper proofed. Keeping the ciphertext, mac and ad separate. The ad represents a file header that needs to be tamper proofed.
```python ```python
from pyaegis import aegis256x4 as ciph from aeg import aegis256x4 as ciph
key, nonce = ciph.random_key(), ciph.random_nonce() key, nonce = ciph.random_key(), ciph.random_nonce()
ct, mac = ciph.encrypt_detached(key, nonce, b"secret", ad=b"header") ct, mac = ciph.encrypt_detached(key, nonce, b"secret", ad=b"header")
@@ -162,7 +178,7 @@ ciph.wipe(pt)
Class-based interface for incremental updates is an alternative to the one-shot functions. Not to be confused with separately verified ciphertext frames (see the next example). Class-based interface for incremental updates is an alternative to the one-shot functions. Not to be confused with separately verified ciphertext frames (see the next example).
```python ```python
from pyaegis import aegis256x4 as ciph from aeg 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", maclen=16) enc = ciph.Encryptor(key, nonce, ad=b"header", maclen=16)
@@ -181,16 +197,17 @@ dec.final(mac) # raises ValueError on failure
It is often practical to split larger messages into frames that can be individually decrypted and verified. Because every frame needs a different key, we employ the `nonce_increment` utility function to produce sequential nonces for each frame. As for the AEGIS algorithm, each frame is a completely independent invocation. The program will each time produce a completely different random-looking encrypted.bin file. It is often practical to split larger messages into frames that can be individually decrypted and verified. Because every frame needs a different key, we employ the `nonce_increment` utility function to produce sequential nonces for each frame. As for the AEGIS algorithm, each frame is a completely independent invocation. The program will each time produce a completely different random-looking encrypted.bin file.
```python ```python
from pyaegis import aegis128x4 as ciph # Encryption settings
from aeg import aegis128x4 as ciph
message = bytearray(30 * b"Attack at dawn! ")
key = b"sixteenbyte key!" # 16 bytes secret key for aegis128* algorithms key = b"sixteenbyte key!" # 16 bytes secret key for aegis128* algorithms
nonce = ciph.random_nonce()
framebytes = 80 # In real applications 1 MiB or more is practical framebytes = 80 # In real applications 1 MiB or more is practical
maclen = ciph.MACBYTES # 16 maclen = ciph.MACBYTES # 16
message = bytearray(30 * b"Attack at dawn! ")
with open("encrypted.bin", "wb") as f: with open("encrypted.bin", "wb") as f:
f.write(nonce) # Public initial nonce sent with the ciphertext # Public initial nonce sent with the ciphertext
nonce = ciph.random_nonce()
f.write(nonce)
while message: while message:
chunk = message[:framebytes - maclen] chunk = message[:framebytes - maclen]
del message[:len(chunk)] del message[:len(chunk)]
@@ -200,9 +217,8 @@ with open("encrypted.bin", "wb") as f:
``` ```
```python ```python
from pyaegis import aegis128x4 as ciph
# Decryption needs same values as encryption # Decryption needs same values as encryption
from aeg import aegis128x4 as ciph
key = b"sixteenbyte key!" key = b"sixteenbyte key!"
framebytes = 80 framebytes = 80
maclen = ciph.MACBYTES maclen = ciph.MACBYTES
@@ -223,7 +239,7 @@ with open("encrypted.bin", "rb") as f:
The stream generator is much faster than any traditional random number generator, cryptographically secure and seekable. Use `random_key()` for unpredictable output. The stream generator is much faster than any traditional random number generator, cryptographically secure and seekable. Use `random_key()` for unpredictable output.
```python ```python
from pyaegis import aegis128x4 as ciph from aeg import aegis128x4 as ciph
key = b"SeedForReplay001" # A non-random deterministic seed (16 bytes) key = b"SeedForReplay001" # A non-random deterministic seed (16 bytes)
nonce = bytearray(ciph.NONCEBYTES) # All-zeroes nonce nonce = bytearray(ciph.NONCEBYTES) # All-zeroes nonce
@@ -239,7 +255,7 @@ Note: this is seekable by converting the block number to nonce with `idx.to_byte
### 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 a sufficient number of bytes can be used. This includes bytearrays, memoryviews, mmap files, numpy arrays etc.
A `TypeError` is raised if the buffer is too small. For convenience, the functions return a memoryview showing only the bytes actually written. A `TypeError` is raised if the buffer is too small. For convenience, the functions return a memoryview showing only the bytes actually written.
@@ -247,18 +263,18 @@ Foreign arrays can be used. This example fills a Numpy array with random integer
```python ```python
import numpy as np import numpy as np
from pyaegis import aegis128x4 as ciph from aeg import aegis128x4 as ciph
key, nonce = ciph.random_key(), ciph.random_nonce() key, nonce = ciph.random_key(), ciph.random_nonce()
arr = np.empty(10, dtype=np.uint64) # Uninitialised integer array arr = np.empty(10, dtype=np.uint64) # Uninitialised integer array
ciph.stream(key, nonce, into=arr) # Fill with random bytes ciph.stream(key, nonce, into=arr) # Fill with random bytes
print(arr) print(arr)
``` ```
In-place operations are supported when the input and the output point to the same location in memory. When using attached MAC tag, the input buffer needs to be sliced to correct length: In-place operations are supported when the input and the output point to the same location in memory. When using attached MAC tag, the input buffer needs to be sliced to correct length:
```python ```python
from pyaegis import aegis256x4 as ciph from aeg import aegis256x4 as ciph
key, nonce = ciph.random_key(), ciph.random_nonce() key, nonce = ciph.random_key(), ciph.random_nonce()
buf = memoryview(bytearray(1000)) # memoryview[:len] is still in the same buffer (no copy) buf = memoryview(bytearray(1000)) # memoryview[:len] is still in the same buffer (no copy)
buf[:7] = b"message" buf[:7] = b"message"
@@ -276,33 +292,27 @@ Detached and unauthenticated modes can use same size input and output (no MAC ad
Runtime CPU feature detection selects optimized code paths (AES-NI, ARM Crypto, AVX2/AVX-512). Multi-lane variants (x2/x4) offer higher throughput on suitable CPUs. Runtime CPU feature detection selects optimized code paths (AES-NI, ARM Crypto, AVX2/AVX-512). Multi-lane variants (x2/x4) offer higher throughput on suitable CPUs.
Run the built-in benchmark to see which variant is fastest on your machine: Benchmarks using the included benchmark module, run on Intel i7-14700, linux, single core (the software is not multithreaded). Note that the results are in megabits per second, not bytes. The CPU lacks AVX-512 that makes the X4 variants faster on processors supporting it (most AMD, Xeon).
```fish ```sh
uv run -m pyaegis.benchmark uv run -m aeg.benchmark
``` AEGIS-256 103166.24 Mb/s
AEGIS-256X2 184225.50 Mb/s
Benchmarks of the Python module and the C library run on Intel i7-14700, linux, single core (the software is not multithreaded). Note that the results are in megabits per second, not bytes. The CPU lacks AVX-512 that makes the X4 variants faster on AMD hardware. AEGIS-256X4 194018.26 Mb/s
AEGIS-128L 161551.73 Mb/s
```fish AEGIS-128X2 281987.80 Mb/s
$ python -m pyaegis.benchmark AEGIS-128X4 217997.37 Mb/s
AEGIS-256 107666.56 Mb/s AEGIS-128L MAC 188886.40 Mb/s
AEGIS-256X2 191314.53 Mb/s AEGIS-128X2 MAC 306457.97 Mb/s
AEGIS-256X4 211537.44 Mb/s AEGIS-128X4 MAC 299576.59 Mb/s
AEGIS-128L 159074.08 Mb/s AEGIS-256 MAC 100914.04 Mb/s
AEGIS-128X2 307332.53 Mb/s AEGIS-256X2 MAC 190208.20 Mb/s
AEGIS-128X4 230106.70 Mb/s AEGIS-256X4 MAC 315919.87 Mb/s
AEGIS-128L MAC 206082.24 Mb/s
AEGIS-128X2 MAC 366401.20 Mb/s
AEGIS-128X4 MAC 375011.51 Mb/s
AEGIS-256 MAC 110187.03 Mb/s
AEGIS-256X2 MAC 210063.51 Mb/s
AEGIS-256X4 MAC 347406.96 Mb/s
``` ```
The Python library performance is similar to that of the C library: The Python library performance is similar to that of the C library:
```fish ```sh
$ ./libaegis/zig-out/bin/benchmark ./libaegis/zig-out/bin/benchmark
AEGIS-256 107820.86 Mb/s AEGIS-256 107820.86 Mb/s
AEGIS-256X2 205025.57 Mb/s AEGIS-256X2 205025.57 Mb/s
AEGIS-256X4 223361.81 Mb/s AEGIS-256X4 223361.81 Mb/s
@@ -316,3 +326,7 @@ AEGIS-256 MAC 116776.62 Mb/s
AEGIS-256X2 MAC 224150.04 Mb/s AEGIS-256X2 MAC 224150.04 Mb/s
AEGIS-256X4 MAC 392088.05 Mb/s AEGIS-256X4 MAC 392088.05 Mb/s
``` ```
## Alternatives
There is also a package named [pyaegis](https://github.com/jedisct1/pyaegis) on PyPI that is unrelated to this module, but that also binds to the libaegis C library. There are also a number of modules named aegis from different packages not at all related to the encryption algorithm.
View File
+16 -9
View File
@@ -1,16 +1,16 @@
[build-system] [build-system]
requires = ["setuptools>=61.0", "cffi>=2.0.0"] requires = ["setuptools>=61.0", "cffi>=2.0.0", "setuptools-scm>=8.0"]
build-backend = "build_backend" build-backend = "build_backend"
backend-path = ["tools"] backend-path = ["tools"]
[project] [project]
name = "pyaegis" name = "aeg"
version = "0.3.0" dynamic = ["version"]
description = "Python bindings for libaegis" description = "AEGIS encryption easy to use Python binding. Wheels for major platforms."
readme = {file = "README.md", content-type = "text/markdown"}
requires-python = ">=3.10" requires-python = ">=3.10"
classifiers = [ classifiers = [
"Programming Language :: Python :: 3", "Development Status :: 5 - Production/Stable",
"Programming Language :: Python :: 3 :: Only",
"Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: CPython",
"Operating System :: OS Independent", "Operating System :: OS Independent",
"Topic :: Security :: Cryptography", "Topic :: Security :: Cryptography",
@@ -20,16 +20,23 @@ dependencies = [
] ]
[project.urls] [project.urls]
Homepage = "https://github.com/aegis-aead/libaegis" Homepage = "https://git.zi.fi/LeoVasanko/aegis-python"
Repository = "https://github.com/LeoVasanko/aegis-python"
[dependency-groups] [dependency-groups]
dev = [ dev = [
"auditwheel>=6.5.0",
"pytest>=8.4.2", "pytest>=8.4.2",
"ruff>=0.14.4",
"setuptools>=80.9.0", "setuptools>=80.9.0",
"setuptools-scm>=9.2.2",
] ]
[tool.setuptools] [tool.setuptools]
packages = ["pyaegis"] package-dir = {"" = "src"}
packages = ["aeg"]
[tool.setuptools.package-data] [tool.setuptools.package-data]
pyaegis = ["*.h", "*.so", "*.pyd"] aeg = ["*.h"]
[tool.setuptools_scm]
+16 -37
View File
@@ -1,49 +1,24 @@
"""Setup script for pyaegis - builds CFFI extension linking to libaegis.a""" """Setup script for aeg - builds CFFI extension with libaegis C library."""
import sys
import sysconfig
from pathlib import Path from pathlib import Path
from cffi import FFI from cffi import FFI
from setuptools import setup from setuptools import setup
libaegis_static = Path("libaegis/zig-out/lib") / (
"aegis.lib" if sys.platform == "win32" else "libaegis.a"
)
def find_libaegis():
"""Locate libaegis.a - check common locations."""
libaegis_paths = [
Path("libaegis/zig-out/lib/libaegis.a"), # Zig build output (repo build)
Path("libaegis/build/libaegis.a"), # CMake build output (repo build)
Path("/usr/local/lib/libaegis.a"), # System install
Path("/usr/lib/libaegis.a"), # System install
]
for path in libaegis_paths:
if path.exists():
print(f"Found libaegis.a at: {path.resolve()}")
return str(path.resolve())
# Return None instead of raising - will be caught during build
return None
# Read the CDEF header
cdef_path = Path(__file__).parent / "pyaegis" / "aegis_cdef.h"
cdef_content = cdef_path.read_text(encoding="utf-8")
# Create CFFI builder
ffibuilder = FFI() ffibuilder = FFI()
ffibuilder.cdef(cdef_content) ffibuilder.cdef((Path(__file__).parent / "src/aeg/aegis_cdef.h").read_text())
# Include directory for headers # Free-threaded Python does not support Limited API (abi3)
include_dirs = [] is_free_threaded = sysconfig.get_config_var("Py_GIL_DISABLED")
libaegis_include = Path("libaegis/src/include")
if libaegis_include.exists():
include_dirs.append(str(libaegis_include.resolve()))
# Try to find libaegis.a, but don't fail if not found (build backend will build it)
libaegis_static = find_libaegis()
# Set the source
ffibuilder.set_source( ffibuilder.set_source(
"pyaegis._aegis", # module name "aeg._aegis",
""" """
#include "aegis.h" #include "aegis.h"
#include "aegis128l.h" #include "aegis128l.h"
@@ -53,11 +28,15 @@ ffibuilder.set_source(
#include "aegis256x2.h" #include "aegis256x2.h"
#include "aegis256x4.h" #include "aegis256x4.h"
""", """,
include_dirs=include_dirs, include_dirs=["libaegis/src/include"],
extra_objects=[libaegis_static] if libaegis_static else [], extra_objects=[str(libaegis_static.resolve())],
py_limited_api=not is_free_threaded,
) )
if __name__ == "__main__": if __name__ == "__main__":
setup( setup(
cffi_modules=["setup.py:ffibuilder"], cffi_modules=["setup.py:ffibuilder"],
options=(
{"bdist_wheel": {"py_limited_api": "cp310"}} if not is_free_threaded else {}
),
) )
+18
View File
@@ -0,0 +1,18 @@
import importlib
from ._ciphers import CIPHERS, CipherName
from ._typing import Cipher
__all__ = ["cipher", "CIPHERS", "Cipher", "CipherName"]
def cipher(alg: CipherName) -> Cipher:
"""Acquire a cipher module by name."""
name = alg.lower().replace("-", "")
if name == "aegis128":
name = "aegis128l" # AEGIS-128 is dead, the user meant AEGIS-128L
if not name.startswith("aegis"):
name = "aegis" + name
if name in CIPHERS.values():
return importlib.import_module(f".{name}", __package__) # type: ignore[return-value]
raise ValueError(f"Unknown algorithm {alg!r}. Valid options: {', '.join(CIPHERS)}")
+20
View File
@@ -0,0 +1,20 @@
# This file is generated by tools/generate.py. Do not edit.
from typing import Literal
CipherName = Literal[
"AEGIS-128L",
"AEGIS-128X2",
"AEGIS-128X4",
"AEGIS-256",
"AEGIS-256X2",
"AEGIS-256X4",
]
CIPHERS: dict[CipherName, str] = {
"AEGIS-128L": "aegis128l",
"AEGIS-128X2": "aegis128x2",
"AEGIS-128X4": "aegis128x4",
"AEGIS-256": "aegis256",
"AEGIS-256X2": "aegis256x2",
"AEGIS-256X4": "aegis256x4",
}
+1 -1
View File
@@ -1,6 +1,6 @@
"""Loader for libaegis CFFI extension module.""" """Loader for libaegis CFFI extension module."""
from pyaegis._aegis import ffi, lib from aeg._aegis import ffi, lib
__all__ = ["ffi", "lib"] __all__ = ["ffi", "lib"]
+126
View File
@@ -0,0 +1,126 @@
from typing import TYPE_CHECKING, Protocol
if TYPE_CHECKING:
from .util import Buffer
__all__ = ["Cipher"]
class _Mac(Protocol):
def reset(self) -> None: ...
def clone(self) -> "_Mac": ...
def update(self, data: "Buffer") -> None: ...
def final(self, into: "Buffer | None" = None) -> bytearray | memoryview: ...
def digest(self) -> bytes: ...
def hexdigest(self) -> str: ...
def verify(self, mac: "Buffer") -> None: ...
class _Encryptor(Protocol):
def update(
self, message: "Buffer", into: "Buffer | None" = None
) -> bytearray | memoryview: ...
def final(self, into: "Buffer | None" = None) -> bytearray | memoryview: ...
class _Decryptor(Protocol):
def update(
self, ct: "Buffer", into: "Buffer | None" = None
) -> bytearray | memoryview: ...
def final(self, mac: "Buffer") -> None: ...
class Cipher(Protocol):
NAME: str
KEYBYTES: int
NONCEBYTES: int
MACBYTES: int
MACBYTES_LONG: int
ALIGNMENT: int
RATE: int
Mac: type[_Mac]
Encryptor: type[_Encryptor]
Decryptor: type[_Decryptor]
@staticmethod
def random_key() -> bytearray: ...
@staticmethod
def random_nonce() -> bytearray: ...
@staticmethod
def encrypt_detached(
key: "Buffer",
nonce: "Buffer",
message: "Buffer",
ad: "Buffer | None" = None,
*,
maclen: int = ...,
ct_into: "Buffer | None" = None,
mac_into: "Buffer | None" = None,
) -> tuple[bytearray | memoryview, bytearray | memoryview]: ...
@staticmethod
def decrypt_detached(
key: "Buffer",
nonce: "Buffer",
ct: "Buffer",
mac: "Buffer",
ad: "Buffer | None" = None,
*,
into: "Buffer | None" = None,
) -> bytearray | memoryview: ...
@staticmethod
def encrypt(
key: "Buffer",
nonce: "Buffer",
message: "Buffer",
ad: "Buffer | None" = None,
*,
maclen: int = ...,
into: "Buffer | None" = None,
) -> bytearray | memoryview: ...
@staticmethod
def decrypt(
key: "Buffer",
nonce: "Buffer",
ct: "Buffer",
ad: "Buffer | None" = None,
*,
maclen: int = ...,
into: "Buffer | None" = None,
) -> bytearray | memoryview: ...
@staticmethod
def stream(
key: "Buffer",
nonce: "Buffer | None",
length: int | None = None,
*,
into: "Buffer | None" = None,
) -> "bytearray | Buffer": ...
@staticmethod
def encrypt_unauthenticated(
key: "Buffer",
nonce: "Buffer",
message: "Buffer",
*,
into: "Buffer | None" = None,
) -> bytearray | memoryview: ...
@staticmethod
def decrypt_unauthenticated(
key: "Buffer",
nonce: "Buffer",
ct: "Buffer",
*,
into: "Buffer | None" = None,
) -> bytearray | memoryview: ...
@staticmethod
def mac(
key: "Buffer",
nonce: "Buffer",
data: "Buffer",
maclen: int = ...,
into: "Buffer | None" = None,
) -> bytearray | memoryview: ...
@staticmethod
def nonce_increment(nonce: "Buffer") -> None: ...
@staticmethod
def wipe(buffer: "Buffer") -> None: ...
+10 -31
View File
@@ -295,9 +295,7 @@ def decrypt(
out = bytearray(expected_out) out = bytearray(expected_out)
else: else:
if into.nbytes < expected_out: if into.nbytes < expected_out:
raise TypeError( raise TypeError("into length must be at least ct.nbytes - maclen")
"into length must be at least ct.nbytes - maclen"
)
out = into out = into
rc = _lib.aegis128l_decrypt( rc = _lib.aegis128l_decrypt(
@@ -582,7 +580,9 @@ class Mac:
out = into out = into
clone = self.clone() clone = self.clone()
rc = _lib.aegis128l_mac_final(clone._proxy.ptr, ffi.from_buffer(out), memoryview(out).nbytes) rc = _lib.aegis128l_mac_final(
clone._proxy.ptr, ffi.from_buffer(out), memoryview(out).nbytes
)
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}")
@@ -708,24 +708,17 @@ class Encryptor:
raise TypeError( raise TypeError(
"into length must be >= expected output size for this update" "into length must be >= expected output size for this update"
) )
written = ffi.new("size_t *")
rc = _lib.aegis128l_state_encrypt_update( rc = _lib.aegis128l_state_encrypt_update(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out_mv), ffi.from_buffer(out_mv),
out_mv.nbytes,
written,
_ptr(message), _ptr(message),
message.nbytes, message.nbytes,
) )
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( raise RuntimeError(f"state encrypt update failed: {err_name}")
f"state encrypt update failed: {err_name} written {written[0]}" return out if into is None else memoryview(out)[:expected_out] # type: ignore
)
w = int(written[0])
assert w == expected_out
return out if into is None else memoryview(out)[:w] # type: ignore
def final(self, into: Buffer | None = None) -> bytearray | memoryview: def final(self, into: Buffer | None = None) -> bytearray | memoryview:
"""Finalize encryption and return the authentication tag. """Finalize encryption and return the authentication tag.
@@ -746,24 +739,17 @@ class Encryptor:
if into is not None: if into is not None:
into = memoryview(into) into = memoryview(into)
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 *")
rc = _lib.aegis128l_state_encrypt_final( rc = _lib.aegis128l_state_encrypt_final(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out), ffi.from_buffer(out),
memoryview(out).nbytes,
written,
maclen, 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"state encrypt final failed: {err_name}") raise RuntimeError(f"state encrypt final failed: {err_name}")
w = int(written[0])
if into is None:
# Only the tag bytes are returned when we allocate the buffer
assert w == maclen
self._state = None self._state = None
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:maclen] # type: ignore
class Decryptor: class Decryptor:
@@ -837,12 +823,9 @@ class Decryptor:
out_mv = memoryview(out) out_mv = memoryview(out)
if out_mv.nbytes < expected_out: if out_mv.nbytes < expected_out:
raise TypeError("into length must be >= required capacity for this update") raise TypeError("into length must be >= required capacity for this update")
written = ffi.new("size_t *") rc = _lib.aegis128l_state_decrypt_update(
rc = _lib.aegis128l_state_decrypt_detached_update(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out_mv), ffi.from_buffer(out_mv),
out_mv.nbytes,
written,
_ptr(ct), _ptr(ct),
ct.nbytes, ct.nbytes,
) )
@@ -850,9 +833,7 @@ class Decryptor:
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"state decrypt update failed: {err_name}") raise RuntimeError(f"state decrypt update failed: {err_name}")
w = int(written[0]) return out if into is None else memoryview(out)[:expected_out] # type: ignore
assert w == expected_out, f"got {w}, expected {expected_out}, ct.nbytes={ct.nbytes}"
return out if into is None else memoryview(out)[:w] # type: ignore
def final(self, mac: Buffer) -> None: def final(self, mac: Buffer) -> None:
"""Finalize decryption by verifying the MAC tag. """Finalize decryption by verifying the MAC tag.
@@ -871,9 +852,7 @@ class Decryptor:
mac = memoryview(mac) mac = memoryview(mac)
if mac.nbytes != maclen: if mac.nbytes != maclen:
raise TypeError(f"mac length must be {maclen}") raise TypeError(f"mac length must be {maclen}")
rc = _lib.aegis128l_state_decrypt_detached_final( rc = _lib.aegis128l_state_decrypt_final(self._state.ptr, _ptr(mac), maclen)
self._state.ptr, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen
)
if rc != 0: if rc != 0:
raise ValueError("authentication failed") raise ValueError("authentication failed")
self._state = None self._state = None
+10 -31
View File
@@ -295,9 +295,7 @@ def decrypt(
out = bytearray(expected_out) out = bytearray(expected_out)
else: else:
if into.nbytes < expected_out: if into.nbytes < expected_out:
raise TypeError( raise TypeError("into length must be at least ct.nbytes - maclen")
"into length must be at least ct.nbytes - maclen"
)
out = into out = into
rc = _lib.aegis128x2_decrypt( rc = _lib.aegis128x2_decrypt(
@@ -582,7 +580,9 @@ class Mac:
out = into out = into
clone = self.clone() clone = self.clone()
rc = _lib.aegis128x2_mac_final(clone._proxy.ptr, ffi.from_buffer(out), memoryview(out).nbytes) rc = _lib.aegis128x2_mac_final(
clone._proxy.ptr, ffi.from_buffer(out), memoryview(out).nbytes
)
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}")
@@ -708,24 +708,17 @@ class Encryptor:
raise TypeError( raise TypeError(
"into length must be >= expected output size for this update" "into length must be >= expected output size for this update"
) )
written = ffi.new("size_t *")
rc = _lib.aegis128x2_state_encrypt_update( rc = _lib.aegis128x2_state_encrypt_update(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out_mv), ffi.from_buffer(out_mv),
out_mv.nbytes,
written,
_ptr(message), _ptr(message),
message.nbytes, message.nbytes,
) )
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( raise RuntimeError(f"state encrypt update failed: {err_name}")
f"state encrypt update failed: {err_name} written {written[0]}" return out if into is None else memoryview(out)[:expected_out] # type: ignore
)
w = int(written[0])
assert w == expected_out
return out if into is None else memoryview(out)[:w] # type: ignore
def final(self, into: Buffer | None = None) -> bytearray | memoryview: def final(self, into: Buffer | None = None) -> bytearray | memoryview:
"""Finalize encryption and return the authentication tag. """Finalize encryption and return the authentication tag.
@@ -746,24 +739,17 @@ class Encryptor:
if into is not None: if into is not None:
into = memoryview(into) into = memoryview(into)
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 *")
rc = _lib.aegis128x2_state_encrypt_final( rc = _lib.aegis128x2_state_encrypt_final(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out), ffi.from_buffer(out),
memoryview(out).nbytes,
written,
maclen, 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"state encrypt final failed: {err_name}") raise RuntimeError(f"state encrypt final failed: {err_name}")
w = int(written[0])
if into is None:
# Only the tag bytes are returned when we allocate the buffer
assert w == maclen
self._state = None self._state = None
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:maclen] # type: ignore
class Decryptor: class Decryptor:
@@ -837,12 +823,9 @@ class Decryptor:
out_mv = memoryview(out) out_mv = memoryview(out)
if out_mv.nbytes < expected_out: if out_mv.nbytes < expected_out:
raise TypeError("into length must be >= required capacity for this update") raise TypeError("into length must be >= required capacity for this update")
written = ffi.new("size_t *") rc = _lib.aegis128x2_state_decrypt_update(
rc = _lib.aegis128x2_state_decrypt_detached_update(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out_mv), ffi.from_buffer(out_mv),
out_mv.nbytes,
written,
_ptr(ct), _ptr(ct),
ct.nbytes, ct.nbytes,
) )
@@ -850,9 +833,7 @@ class Decryptor:
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"state decrypt update failed: {err_name}") raise RuntimeError(f"state decrypt update failed: {err_name}")
w = int(written[0]) return out if into is None else memoryview(out)[:expected_out] # type: ignore
assert w == expected_out, f"got {w}, expected {expected_out}, ct.nbytes={ct.nbytes}"
return out if into is None else memoryview(out)[:w] # type: ignore
def final(self, mac: Buffer) -> None: def final(self, mac: Buffer) -> None:
"""Finalize decryption by verifying the MAC tag. """Finalize decryption by verifying the MAC tag.
@@ -871,9 +852,7 @@ class Decryptor:
mac = memoryview(mac) mac = memoryview(mac)
if mac.nbytes != maclen: if mac.nbytes != maclen:
raise TypeError(f"mac length must be {maclen}") raise TypeError(f"mac length must be {maclen}")
rc = _lib.aegis128x2_state_decrypt_detached_final( rc = _lib.aegis128x2_state_decrypt_final(self._state.ptr, _ptr(mac), maclen)
self._state.ptr, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen
)
if rc != 0: if rc != 0:
raise ValueError("authentication failed") raise ValueError("authentication failed")
self._state = None self._state = None
+10 -31
View File
@@ -295,9 +295,7 @@ def decrypt(
out = bytearray(expected_out) out = bytearray(expected_out)
else: else:
if into.nbytes < expected_out: if into.nbytes < expected_out:
raise TypeError( raise TypeError("into length must be at least ct.nbytes - maclen")
"into length must be at least ct.nbytes - maclen"
)
out = into out = into
rc = _lib.aegis128x4_decrypt( rc = _lib.aegis128x4_decrypt(
@@ -582,7 +580,9 @@ class Mac:
out = into out = into
clone = self.clone() clone = self.clone()
rc = _lib.aegis128x4_mac_final(clone._proxy.ptr, ffi.from_buffer(out), memoryview(out).nbytes) rc = _lib.aegis128x4_mac_final(
clone._proxy.ptr, ffi.from_buffer(out), memoryview(out).nbytes
)
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}")
@@ -708,24 +708,17 @@ class Encryptor:
raise TypeError( raise TypeError(
"into length must be >= expected output size for this update" "into length must be >= expected output size for this update"
) )
written = ffi.new("size_t *")
rc = _lib.aegis128x4_state_encrypt_update( rc = _lib.aegis128x4_state_encrypt_update(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out_mv), ffi.from_buffer(out_mv),
out_mv.nbytes,
written,
_ptr(message), _ptr(message),
message.nbytes, message.nbytes,
) )
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( raise RuntimeError(f"state encrypt update failed: {err_name}")
f"state encrypt update failed: {err_name} written {written[0]}" return out if into is None else memoryview(out)[:expected_out] # type: ignore
)
w = int(written[0])
assert w == expected_out
return out if into is None else memoryview(out)[:w] # type: ignore
def final(self, into: Buffer | None = None) -> bytearray | memoryview: def final(self, into: Buffer | None = None) -> bytearray | memoryview:
"""Finalize encryption and return the authentication tag. """Finalize encryption and return the authentication tag.
@@ -746,24 +739,17 @@ class Encryptor:
if into is not None: if into is not None:
into = memoryview(into) into = memoryview(into)
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 *")
rc = _lib.aegis128x4_state_encrypt_final( rc = _lib.aegis128x4_state_encrypt_final(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out), ffi.from_buffer(out),
memoryview(out).nbytes,
written,
maclen, 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"state encrypt final failed: {err_name}") raise RuntimeError(f"state encrypt final failed: {err_name}")
w = int(written[0])
if into is None:
# Only the tag bytes are returned when we allocate the buffer
assert w == maclen
self._state = None self._state = None
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:maclen] # type: ignore
class Decryptor: class Decryptor:
@@ -837,12 +823,9 @@ class Decryptor:
out_mv = memoryview(out) out_mv = memoryview(out)
if out_mv.nbytes < expected_out: if out_mv.nbytes < expected_out:
raise TypeError("into length must be >= required capacity for this update") raise TypeError("into length must be >= required capacity for this update")
written = ffi.new("size_t *") rc = _lib.aegis128x4_state_decrypt_update(
rc = _lib.aegis128x4_state_decrypt_detached_update(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out_mv), ffi.from_buffer(out_mv),
out_mv.nbytes,
written,
_ptr(ct), _ptr(ct),
ct.nbytes, ct.nbytes,
) )
@@ -850,9 +833,7 @@ class Decryptor:
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"state decrypt update failed: {err_name}") raise RuntimeError(f"state decrypt update failed: {err_name}")
w = int(written[0]) return out if into is None else memoryview(out)[:expected_out] # type: ignore
assert w == expected_out, f"got {w}, expected {expected_out}, ct.nbytes={ct.nbytes}"
return out if into is None else memoryview(out)[:w] # type: ignore
def final(self, mac: Buffer) -> None: def final(self, mac: Buffer) -> None:
"""Finalize decryption by verifying the MAC tag. """Finalize decryption by verifying the MAC tag.
@@ -871,9 +852,7 @@ class Decryptor:
mac = memoryview(mac) mac = memoryview(mac)
if mac.nbytes != maclen: if mac.nbytes != maclen:
raise TypeError(f"mac length must be {maclen}") raise TypeError(f"mac length must be {maclen}")
rc = _lib.aegis128x4_state_decrypt_detached_final( rc = _lib.aegis128x4_state_decrypt_final(self._state.ptr, _ptr(mac), maclen)
self._state.ptr, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen
)
if rc != 0: if rc != 0:
raise ValueError("authentication failed") raise ValueError("authentication failed")
self._state = None self._state = None
+10 -31
View File
@@ -295,9 +295,7 @@ def decrypt(
out = bytearray(expected_out) out = bytearray(expected_out)
else: else:
if into.nbytes < expected_out: if into.nbytes < expected_out:
raise TypeError( raise TypeError("into length must be at least ct.nbytes - maclen")
"into length must be at least ct.nbytes - maclen"
)
out = into out = into
rc = _lib.aegis256_decrypt( rc = _lib.aegis256_decrypt(
@@ -582,7 +580,9 @@ class Mac:
out = into out = into
clone = self.clone() clone = self.clone()
rc = _lib.aegis256_mac_final(clone._proxy.ptr, ffi.from_buffer(out), memoryview(out).nbytes) rc = _lib.aegis256_mac_final(
clone._proxy.ptr, ffi.from_buffer(out), memoryview(out).nbytes
)
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}")
@@ -708,24 +708,17 @@ class Encryptor:
raise TypeError( raise TypeError(
"into length must be >= expected output size for this update" "into length must be >= expected output size for this update"
) )
written = ffi.new("size_t *")
rc = _lib.aegis256_state_encrypt_update( rc = _lib.aegis256_state_encrypt_update(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out_mv), ffi.from_buffer(out_mv),
out_mv.nbytes,
written,
_ptr(message), _ptr(message),
message.nbytes, message.nbytes,
) )
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( raise RuntimeError(f"state encrypt update failed: {err_name}")
f"state encrypt update failed: {err_name} written {written[0]}" return out if into is None else memoryview(out)[:expected_out] # type: ignore
)
w = int(written[0])
assert w == expected_out
return out if into is None else memoryview(out)[:w] # type: ignore
def final(self, into: Buffer | None = None) -> bytearray | memoryview: def final(self, into: Buffer | None = None) -> bytearray | memoryview:
"""Finalize encryption and return the authentication tag. """Finalize encryption and return the authentication tag.
@@ -746,24 +739,17 @@ class Encryptor:
if into is not None: if into is not None:
into = memoryview(into) into = memoryview(into)
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 *")
rc = _lib.aegis256_state_encrypt_final( rc = _lib.aegis256_state_encrypt_final(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out), ffi.from_buffer(out),
memoryview(out).nbytes,
written,
maclen, 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"state encrypt final failed: {err_name}") raise RuntimeError(f"state encrypt final failed: {err_name}")
w = int(written[0])
if into is None:
# Only the tag bytes are returned when we allocate the buffer
assert w == maclen
self._state = None self._state = None
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:maclen] # type: ignore
class Decryptor: class Decryptor:
@@ -837,12 +823,9 @@ class Decryptor:
out_mv = memoryview(out) out_mv = memoryview(out)
if out_mv.nbytes < expected_out: if out_mv.nbytes < expected_out:
raise TypeError("into length must be >= required capacity for this update") raise TypeError("into length must be >= required capacity for this update")
written = ffi.new("size_t *") rc = _lib.aegis256_state_decrypt_update(
rc = _lib.aegis256_state_decrypt_detached_update(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out_mv), ffi.from_buffer(out_mv),
out_mv.nbytes,
written,
_ptr(ct), _ptr(ct),
ct.nbytes, ct.nbytes,
) )
@@ -850,9 +833,7 @@ class Decryptor:
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"state decrypt update failed: {err_name}") raise RuntimeError(f"state decrypt update failed: {err_name}")
w = int(written[0]) return out if into is None else memoryview(out)[:expected_out] # type: ignore
assert w == expected_out, f"got {w}, expected {expected_out}, ct.nbytes={ct.nbytes}"
return out if into is None else memoryview(out)[:w] # type: ignore
def final(self, mac: Buffer) -> None: def final(self, mac: Buffer) -> None:
"""Finalize decryption by verifying the MAC tag. """Finalize decryption by verifying the MAC tag.
@@ -871,9 +852,7 @@ class Decryptor:
mac = memoryview(mac) mac = memoryview(mac)
if mac.nbytes != maclen: if mac.nbytes != maclen:
raise TypeError(f"mac length must be {maclen}") raise TypeError(f"mac length must be {maclen}")
rc = _lib.aegis256_state_decrypt_detached_final( rc = _lib.aegis256_state_decrypt_final(self._state.ptr, _ptr(mac), maclen)
self._state.ptr, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen
)
if rc != 0: if rc != 0:
raise ValueError("authentication failed") raise ValueError("authentication failed")
self._state = None self._state = None
+10 -31
View File
@@ -295,9 +295,7 @@ def decrypt(
out = bytearray(expected_out) out = bytearray(expected_out)
else: else:
if into.nbytes < expected_out: if into.nbytes < expected_out:
raise TypeError( raise TypeError("into length must be at least ct.nbytes - maclen")
"into length must be at least ct.nbytes - maclen"
)
out = into out = into
rc = _lib.aegis256x2_decrypt( rc = _lib.aegis256x2_decrypt(
@@ -582,7 +580,9 @@ class Mac:
out = into out = into
clone = self.clone() clone = self.clone()
rc = _lib.aegis256x2_mac_final(clone._proxy.ptr, ffi.from_buffer(out), memoryview(out).nbytes) rc = _lib.aegis256x2_mac_final(
clone._proxy.ptr, ffi.from_buffer(out), memoryview(out).nbytes
)
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}")
@@ -708,24 +708,17 @@ class Encryptor:
raise TypeError( raise TypeError(
"into length must be >= expected output size for this update" "into length must be >= expected output size for this update"
) )
written = ffi.new("size_t *")
rc = _lib.aegis256x2_state_encrypt_update( rc = _lib.aegis256x2_state_encrypt_update(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out_mv), ffi.from_buffer(out_mv),
out_mv.nbytes,
written,
_ptr(message), _ptr(message),
message.nbytes, message.nbytes,
) )
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( raise RuntimeError(f"state encrypt update failed: {err_name}")
f"state encrypt update failed: {err_name} written {written[0]}" return out if into is None else memoryview(out)[:expected_out] # type: ignore
)
w = int(written[0])
assert w == expected_out
return out if into is None else memoryview(out)[:w] # type: ignore
def final(self, into: Buffer | None = None) -> bytearray | memoryview: def final(self, into: Buffer | None = None) -> bytearray | memoryview:
"""Finalize encryption and return the authentication tag. """Finalize encryption and return the authentication tag.
@@ -746,24 +739,17 @@ class Encryptor:
if into is not None: if into is not None:
into = memoryview(into) into = memoryview(into)
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 *")
rc = _lib.aegis256x2_state_encrypt_final( rc = _lib.aegis256x2_state_encrypt_final(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out), ffi.from_buffer(out),
memoryview(out).nbytes,
written,
maclen, 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"state encrypt final failed: {err_name}") raise RuntimeError(f"state encrypt final failed: {err_name}")
w = int(written[0])
if into is None:
# Only the tag bytes are returned when we allocate the buffer
assert w == maclen
self._state = None self._state = None
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:maclen] # type: ignore
class Decryptor: class Decryptor:
@@ -837,12 +823,9 @@ class Decryptor:
out_mv = memoryview(out) out_mv = memoryview(out)
if out_mv.nbytes < expected_out: if out_mv.nbytes < expected_out:
raise TypeError("into length must be >= required capacity for this update") raise TypeError("into length must be >= required capacity for this update")
written = ffi.new("size_t *") rc = _lib.aegis256x2_state_decrypt_update(
rc = _lib.aegis256x2_state_decrypt_detached_update(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out_mv), ffi.from_buffer(out_mv),
out_mv.nbytes,
written,
_ptr(ct), _ptr(ct),
ct.nbytes, ct.nbytes,
) )
@@ -850,9 +833,7 @@ class Decryptor:
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"state decrypt update failed: {err_name}") raise RuntimeError(f"state decrypt update failed: {err_name}")
w = int(written[0]) return out if into is None else memoryview(out)[:expected_out] # type: ignore
assert w == expected_out, f"got {w}, expected {expected_out}, ct.nbytes={ct.nbytes}"
return out if into is None else memoryview(out)[:w] # type: ignore
def final(self, mac: Buffer) -> None: def final(self, mac: Buffer) -> None:
"""Finalize decryption by verifying the MAC tag. """Finalize decryption by verifying the MAC tag.
@@ -871,9 +852,7 @@ class Decryptor:
mac = memoryview(mac) mac = memoryview(mac)
if mac.nbytes != maclen: if mac.nbytes != maclen:
raise TypeError(f"mac length must be {maclen}") raise TypeError(f"mac length must be {maclen}")
rc = _lib.aegis256x2_state_decrypt_detached_final( rc = _lib.aegis256x2_state_decrypt_final(self._state.ptr, _ptr(mac), maclen)
self._state.ptr, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen
)
if rc != 0: if rc != 0:
raise ValueError("authentication failed") raise ValueError("authentication failed")
self._state = None self._state = None
+10 -31
View File
@@ -295,9 +295,7 @@ def decrypt(
out = bytearray(expected_out) out = bytearray(expected_out)
else: else:
if into.nbytes < expected_out: if into.nbytes < expected_out:
raise TypeError( raise TypeError("into length must be at least ct.nbytes - maclen")
"into length must be at least ct.nbytes - maclen"
)
out = into out = into
rc = _lib.aegis256x4_decrypt( rc = _lib.aegis256x4_decrypt(
@@ -582,7 +580,9 @@ class Mac:
out = into out = into
clone = self.clone() clone = self.clone()
rc = _lib.aegis256x4_mac_final(clone._proxy.ptr, ffi.from_buffer(out), memoryview(out).nbytes) rc = _lib.aegis256x4_mac_final(
clone._proxy.ptr, ffi.from_buffer(out), memoryview(out).nbytes
)
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}")
@@ -708,24 +708,17 @@ class Encryptor:
raise TypeError( raise TypeError(
"into length must be >= expected output size for this update" "into length must be >= expected output size for this update"
) )
written = ffi.new("size_t *")
rc = _lib.aegis256x4_state_encrypt_update( rc = _lib.aegis256x4_state_encrypt_update(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out_mv), ffi.from_buffer(out_mv),
out_mv.nbytes,
written,
_ptr(message), _ptr(message),
message.nbytes, message.nbytes,
) )
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( raise RuntimeError(f"state encrypt update failed: {err_name}")
f"state encrypt update failed: {err_name} written {written[0]}" return out if into is None else memoryview(out)[:expected_out] # type: ignore
)
w = int(written[0])
assert w == expected_out
return out if into is None else memoryview(out)[:w] # type: ignore
def final(self, into: Buffer | None = None) -> bytearray | memoryview: def final(self, into: Buffer | None = None) -> bytearray | memoryview:
"""Finalize encryption and return the authentication tag. """Finalize encryption and return the authentication tag.
@@ -746,24 +739,17 @@ class Encryptor:
if into is not None: if into is not None:
into = memoryview(into) into = memoryview(into)
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 *")
rc = _lib.aegis256x4_state_encrypt_final( rc = _lib.aegis256x4_state_encrypt_final(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out), ffi.from_buffer(out),
memoryview(out).nbytes,
written,
maclen, 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"state encrypt final failed: {err_name}") raise RuntimeError(f"state encrypt final failed: {err_name}")
w = int(written[0])
if into is None:
# Only the tag bytes are returned when we allocate the buffer
assert w == maclen
self._state = None self._state = None
return out if into is None else memoryview(out)[:w] # type: ignore return out if into is None else memoryview(out)[:maclen] # type: ignore
class Decryptor: class Decryptor:
@@ -837,12 +823,9 @@ class Decryptor:
out_mv = memoryview(out) out_mv = memoryview(out)
if out_mv.nbytes < expected_out: if out_mv.nbytes < expected_out:
raise TypeError("into length must be >= required capacity for this update") raise TypeError("into length must be >= required capacity for this update")
written = ffi.new("size_t *") rc = _lib.aegis256x4_state_decrypt_update(
rc = _lib.aegis256x4_state_decrypt_detached_update(
self._state.ptr, self._state.ptr,
ffi.from_buffer(out_mv), ffi.from_buffer(out_mv),
out_mv.nbytes,
written,
_ptr(ct), _ptr(ct),
ct.nbytes, ct.nbytes,
) )
@@ -850,9 +833,7 @@ class Decryptor:
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"state decrypt update failed: {err_name}") raise RuntimeError(f"state decrypt update failed: {err_name}")
w = int(written[0]) return out if into is None else memoryview(out)[:expected_out] # type: ignore
assert w == expected_out, f"got {w}, expected {expected_out}, ct.nbytes={ct.nbytes}"
return out if into is None else memoryview(out)[:w] # type: ignore
def final(self, mac: Buffer) -> None: def final(self, mac: Buffer) -> None:
"""Finalize decryption by verifying the MAC tag. """Finalize decryption by verifying the MAC tag.
@@ -871,9 +852,7 @@ class Decryptor:
mac = memoryview(mac) mac = memoryview(mac)
if mac.nbytes != maclen: if mac.nbytes != maclen:
raise TypeError(f"mac length must be {maclen}") raise TypeError(f"mac length must be {maclen}")
rc = _lib.aegis256x4_state_decrypt_detached_final( rc = _lib.aegis256x4_state_decrypt_final(self._state.ptr, _ptr(mac), maclen)
self._state.ptr, ffi.NULL, 0, ffi.NULL, _ptr(mac), maclen
)
if rc != 0: if rc != 0:
raise ValueError("authentication failed") raise ValueError("authentication failed")
self._state = None self._state = None
+20 -146
View File
@@ -55,35 +55,10 @@ void aegis128l_state_init(aegis128l_state *st_,
size_t adlen, size_t adlen,
const uint8_t *npub, const uint8_t *npub,
const uint8_t *k); const uint8_t *k);
int aegis128l_state_encrypt_update(aegis128l_state *st_, int aegis128l_state_encrypt_update(aegis128l_state *st_, uint8_t *c, const uint8_t *m, size_t mlen);
uint8_t *c, int aegis128l_state_encrypt_final(aegis128l_state *st_, uint8_t *mac, size_t maclen);
size_t clen_max, int aegis128l_state_decrypt_update(aegis128l_state *st_, uint8_t *m, const uint8_t *c, size_t clen) ;
size_t *written, int aegis128l_state_decrypt_final(aegis128l_state *st_, const uint8_t *mac, size_t maclen) ;
const uint8_t *m,
size_t mlen);
int aegis128l_state_encrypt_detached_final(aegis128l_state *st_,
uint8_t *c,
size_t clen_max,
size_t *written,
uint8_t *mac,
size_t maclen);
int aegis128l_state_encrypt_final(aegis128l_state *st_,
uint8_t *c,
size_t clen_max,
size_t *written,
size_t maclen);
int aegis128l_state_decrypt_detached_update(aegis128l_state *st_,
uint8_t *m,
size_t mlen_max,
size_t *written,
const uint8_t *c,
size_t clen) ;
int aegis128l_state_decrypt_detached_final(aegis128l_state *st_,
uint8_t *m,
size_t mlen_max,
size_t *written,
const uint8_t *mac,
size_t maclen) ;
void aegis128l_stream(uint8_t *out, size_t len, const uint8_t *npub, const uint8_t *k); void aegis128l_stream(uint8_t *out, size_t len, const uint8_t *npub, const uint8_t *k);
void aegis128l_encrypt_unauthenticated(uint8_t *c, void aegis128l_encrypt_unauthenticated(uint8_t *c,
const uint8_t *m, const uint8_t *m,
@@ -151,33 +126,14 @@ void aegis128x2_state_init(aegis128x2_state *st_,
const uint8_t *k); const uint8_t *k);
int aegis128x2_state_encrypt_update(aegis128x2_state *st_, int aegis128x2_state_encrypt_update(aegis128x2_state *st_,
uint8_t *c, uint8_t *c,
size_t clen_max,
size_t *written,
const uint8_t *m, const uint8_t *m,
size_t mlen); size_t mlen);
int aegis128x2_state_encrypt_detached_final(aegis128x2_state *st_, int aegis128x2_state_encrypt_final(aegis128x2_state *st_, uint8_t *mac, size_t maclen);
uint8_t *c, int aegis128x2_state_decrypt_update(aegis128x2_state *st_,
size_t clen_max,
size_t *written,
uint8_t *mac,
size_t maclen);
int aegis128x2_state_encrypt_final(aegis128x2_state *st_,
uint8_t *c,
size_t clen_max,
size_t *written,
size_t maclen);
int aegis128x2_state_decrypt_detached_update(aegis128x2_state *st_,
uint8_t *m, uint8_t *m,
size_t mlen_max,
size_t *written,
const uint8_t *c, const uint8_t *c,
size_t clen) ; size_t clen) ;
int aegis128x2_state_decrypt_detached_final(aegis128x2_state *st_, int aegis128x2_state_decrypt_final(aegis128x2_state *st_, const uint8_t *mac, size_t maclen) ;
uint8_t *m,
size_t mlen_max,
size_t *written,
const uint8_t *mac,
size_t maclen) ;
void aegis128x2_stream(uint8_t *out, size_t len, const uint8_t *npub, const uint8_t *k); void aegis128x2_stream(uint8_t *out, size_t len, const uint8_t *npub, const uint8_t *k);
void aegis128x2_encrypt_unauthenticated(uint8_t *c, void aegis128x2_encrypt_unauthenticated(uint8_t *c,
const uint8_t *m, const uint8_t *m,
@@ -245,33 +201,14 @@ void aegis128x4_state_init(aegis128x4_state *st_,
const uint8_t *k); const uint8_t *k);
int aegis128x4_state_encrypt_update(aegis128x4_state *st_, int aegis128x4_state_encrypt_update(aegis128x4_state *st_,
uint8_t *c, uint8_t *c,
size_t clen_max,
size_t *written,
const uint8_t *m, const uint8_t *m,
size_t mlen); size_t mlen);
int aegis128x4_state_encrypt_detached_final(aegis128x4_state *st_, int aegis128x4_state_encrypt_final(aegis128x4_state *st_, uint8_t *mac, size_t maclen);
uint8_t *c, int aegis128x4_state_decrypt_update(aegis128x4_state *st_,
size_t clen_max,
size_t *written,
uint8_t *mac,
size_t maclen);
int aegis128x4_state_encrypt_final(aegis128x4_state *st_,
uint8_t *c,
size_t clen_max,
size_t *written,
size_t maclen);
int aegis128x4_state_decrypt_detached_update(aegis128x4_state *st_,
uint8_t *m, uint8_t *m,
size_t mlen_max,
size_t *written,
const uint8_t *c, const uint8_t *c,
size_t clen) ; size_t clen) ;
int aegis128x4_state_decrypt_detached_final(aegis128x4_state *st_, int aegis128x4_state_decrypt_final(aegis128x4_state *st_, const uint8_t *mac, size_t maclen) ;
uint8_t *m,
size_t mlen_max,
size_t *written,
const uint8_t *mac,
size_t maclen) ;
void aegis128x4_stream(uint8_t *out, size_t len, const uint8_t *npub, const uint8_t *k); void aegis128x4_stream(uint8_t *out, size_t len, const uint8_t *npub, const uint8_t *k);
void aegis128x4_encrypt_unauthenticated(uint8_t *c, void aegis128x4_encrypt_unauthenticated(uint8_t *c,
const uint8_t *m, const uint8_t *m,
@@ -337,35 +274,10 @@ void aegis256_state_init(aegis256_state *st_,
size_t adlen, size_t adlen,
const uint8_t *npub, const uint8_t *npub,
const uint8_t *k); const uint8_t *k);
int aegis256_state_encrypt_update(aegis256_state *st_, int aegis256_state_encrypt_update(aegis256_state *st_, uint8_t *c, const uint8_t *m, size_t mlen);
uint8_t *c, int aegis256_state_encrypt_final(aegis256_state *st_, uint8_t *mac, size_t maclen);
size_t clen_max, int aegis256_state_decrypt_update(aegis256_state *st_, uint8_t *m, const uint8_t *c, size_t clen) ;
size_t *written, int aegis256_state_decrypt_final(aegis256_state *st_, const uint8_t *mac, size_t maclen) ;
const uint8_t *m,
size_t mlen);
int aegis256_state_encrypt_detached_final(aegis256_state *st_,
uint8_t *c,
size_t clen_max,
size_t *written,
uint8_t *mac,
size_t maclen);
int aegis256_state_encrypt_final(aegis256_state *st_,
uint8_t *c,
size_t clen_max,
size_t *written,
size_t maclen);
int aegis256_state_decrypt_detached_update(aegis256_state *st_,
uint8_t *m,
size_t mlen_max,
size_t *written,
const uint8_t *c,
size_t clen) ;
int aegis256_state_decrypt_detached_final(aegis256_state *st_,
uint8_t *m,
size_t mlen_max,
size_t *written,
const uint8_t *mac,
size_t maclen) ;
void aegis256_stream(uint8_t *out, size_t len, const uint8_t *npub, const uint8_t *k); void aegis256_stream(uint8_t *out, size_t len, const uint8_t *npub, const uint8_t *k);
void aegis256_encrypt_unauthenticated(uint8_t *c, void aegis256_encrypt_unauthenticated(uint8_t *c,
const uint8_t *m, const uint8_t *m,
@@ -433,33 +345,14 @@ void aegis256x2_state_init(aegis256x2_state *st_,
const uint8_t *k); const uint8_t *k);
int aegis256x2_state_encrypt_update(aegis256x2_state *st_, int aegis256x2_state_encrypt_update(aegis256x2_state *st_,
uint8_t *c, uint8_t *c,
size_t clen_max,
size_t *written,
const uint8_t *m, const uint8_t *m,
size_t mlen); size_t mlen);
int aegis256x2_state_encrypt_detached_final(aegis256x2_state *st_, int aegis256x2_state_encrypt_final(aegis256x2_state *st_, uint8_t *mac, size_t maclen);
uint8_t *c, int aegis256x2_state_decrypt_update(aegis256x2_state *st_,
size_t clen_max,
size_t *written,
uint8_t *mac,
size_t maclen);
int aegis256x2_state_encrypt_final(aegis256x2_state *st_,
uint8_t *c,
size_t clen_max,
size_t *written,
size_t maclen);
int aegis256x2_state_decrypt_detached_update(aegis256x2_state *st_,
uint8_t *m, uint8_t *m,
size_t mlen_max,
size_t *written,
const uint8_t *c, const uint8_t *c,
size_t clen) ; size_t clen) ;
int aegis256x2_state_decrypt_detached_final(aegis256x2_state *st_, int aegis256x2_state_decrypt_final(aegis256x2_state *st_, const uint8_t *mac, size_t maclen) ;
uint8_t *m,
size_t mlen_max,
size_t *written,
const uint8_t *mac,
size_t maclen) ;
void aegis256x2_stream(uint8_t *out, size_t len, const uint8_t *npub, const uint8_t *k); void aegis256x2_stream(uint8_t *out, size_t len, const uint8_t *npub, const uint8_t *k);
void aegis256x2_encrypt_unauthenticated(uint8_t *c, void aegis256x2_encrypt_unauthenticated(uint8_t *c,
const uint8_t *m, const uint8_t *m,
@@ -527,33 +420,14 @@ void aegis256x4_state_init(aegis256x4_state *st_,
const uint8_t *k); const uint8_t *k);
int aegis256x4_state_encrypt_update(aegis256x4_state *st_, int aegis256x4_state_encrypt_update(aegis256x4_state *st_,
uint8_t *c, uint8_t *c,
size_t clen_max,
size_t *written,
const uint8_t *m, const uint8_t *m,
size_t mlen); size_t mlen);
int aegis256x4_state_encrypt_detached_final(aegis256x4_state *st_, int aegis256x4_state_encrypt_final(aegis256x4_state *st_, uint8_t *mac, size_t maclen);
uint8_t *c, int aegis256x4_state_decrypt_update(aegis256x4_state *st_,
size_t clen_max,
size_t *written,
uint8_t *mac,
size_t maclen);
int aegis256x4_state_encrypt_final(aegis256x4_state *st_,
uint8_t *c,
size_t clen_max,
size_t *written,
size_t maclen);
int aegis256x4_state_decrypt_detached_update(aegis256x4_state *st_,
uint8_t *m, uint8_t *m,
size_t mlen_max,
size_t *written,
const uint8_t *c, const uint8_t *c,
size_t clen) ; size_t clen) ;
int aegis256x4_state_decrypt_detached_final(aegis256x4_state *st_, int aegis256x4_state_decrypt_final(aegis256x4_state *st_, const uint8_t *mac, size_t maclen) ;
uint8_t *m,
size_t mlen_max,
size_t *written,
const uint8_t *mac,
size_t maclen) ;
void aegis256x4_stream(uint8_t *out, size_t len, const uint8_t *npub, const uint8_t *k); void aegis256x4_stream(uint8_t *out, size_t len, const uint8_t *npub, const uint8_t *k);
void aegis256x4_encrypt_unauthenticated(uint8_t *c, void aegis256x4_encrypt_unauthenticated(uint8_t *c,
const uint8_t *m, const uint8_t *m,
@@ -12,7 +12,7 @@ Output format and throughput units mirror the Zig benchmark (Mb/s).
import secrets import secrets
import time import time
from pyaegis import aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4 from aeg import aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4
MSG_LEN = 16384000 # 16 000 KiB MSG_LEN = 16384000 # 16 000 KiB
ITERATIONS = 100 ITERATIONS = 100
+7 -12
View File
@@ -1,4 +1,4 @@
"""Utility helpers for pyaegis. """Utility helpers for aeg.
Currently provides Python-side aligned allocation helpers that avoid relying Currently provides Python-side aligned allocation helpers that avoid relying
on libc/posix_memalign. Memory is owned by Python; C code only borrows it. on libc/posix_memalign. Memory is owned by Python; C code only borrows it.
@@ -11,12 +11,9 @@ from ._loader import ffi
__all__ = ["new_aligned_struct", "aligned_address", "Buffer", "nonce_increment", "wipe"] __all__ = ["new_aligned_struct", "aligned_address", "Buffer", "nonce_increment", "wipe"]
try: try:
from collections.abc import Buffer as _Buffer # type: ignore[misc] from collections.abc import Buffer # type: ignore
class Buffer(_Buffer, Protocol): # type: ignore[misc]
pass
except ImportError: except ImportError:
# Fallback for Python < 3.12
class Buffer(Protocol): class Buffer(Protocol):
def __buffer__(self, flags: int) -> memoryview: ... def __buffer__(self, flags: int) -> memoryview: ...
@@ -75,13 +72,11 @@ def nonce_increment(nonce: Buffer) -> None:
def wipe(buffer: Buffer) -> None: def wipe(buffer: Buffer) -> None:
"""Set all bytes of the input buffer to zero. """Securely clearing sensitive data from memory. Sets all bytes of the buffer to 0xFF.
Useful for securely clearing sensitive data from memory.
Args: Args:
buffer: The buffer to wipe (modified in place). buffer: The buffer to wipe (modified in place).
""" """
n = memoryview(buffer) # This is the fastest method I have found in Python
for i in range(len(n)): n = memoryview(buffer).cast("B")
n[i] = 0 n[:] = b"\xff" * len(n)
-89
View File
@@ -1,89 +0,0 @@
#! /usr/bin/env python3
import json
import re
def tvdump(topic, tvs):
with open(filename(topic), "w") as f:
f.write(json.dumps(tvs, indent=2))
print(json.dumps(tvs, indent=2))
def filename(topic):
return re.sub(r"[^a-z0-9]+", "-", topic.lower()) + ".json"
header = True
in_tv = False
tv = {}
tvs = []
must_fail = False
with open("../draft-irtf-cfrg-aegis-aead.md") as f:
for line in f:
line = line.strip()
if line == "":
continue
if line.startswith("# Test Vectors"):
header = False
continue
if header:
continue
if line.startswith("## "):
if len(tvs) > 0:
tvdump(topic, tvs)
topic = line[3:]
tv_name = topic
tvs = []
continue
if line.startswith("### "):
tv_name = line[4:]
tv = {"test": tv_name}
in_tv = False
continue
if line == "~~~ test-vectors":
in_tv = True
tv = {"name": tv_name}
if must_fail:
tv["error"] = "verification failed"
must_fail = False
continue
if line == "~~~":
tvs.append(tv)
in_tv = False
current_key = None
continue
if line.find("verification failed") != -1:
must_fail = True
continue
if line == "After initialization:":
tv_name = tv_name + " (after initialization)"
if not in_tv:
continue
parts = line.split(":")
if len(parts) == 2:
key = parts[0].strip()
value = parts[1].strip()
if key == "After Update":
continue
if key in tv:
key = key + "_2"
tv[key] = value
current_key = key
continue
if not current_key:
continue
tv[key] += line.strip()
tvdump(topic, tvs)
+1 -1
View File
@@ -3,7 +3,7 @@ from pathlib import Path
import pytest import pytest
from pyaegis import aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4 from aeg import aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4
from .util import random_split_bytes from .util import random_split_bytes
+1 -1
View File
@@ -3,7 +3,7 @@ from pathlib import Path
import pytest import pytest
from pyaegis import aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4 from aeg import aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4
from .util import random_split_bytes from .util import random_split_bytes
+1 -1
View File
@@ -6,7 +6,7 @@ after calling final(), preventing accidental misuse.
import pytest import pytest
from pyaegis import aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4 from aeg import aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4
# All AEGIS algorithm modules # All AEGIS algorithm modules
ALL_ALGORITHMS = [aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4] ALL_ALGORITHMS = [aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4]
+46 -78
View File
@@ -1,100 +1,68 @@
"""Custom build backend that builds libaegis with Zig before building the Python package.""" """Custom build backend that builds libaegis with Zig before building the Python package."""
import os
import platform
import shutil import shutil
import subprocess import subprocess
import sys import sys
from pathlib import Path from pathlib import Path
from setuptools import build_meta as _orig from setuptools import build_meta
__all__ = [
"build_sdist",
"build_wheel",
"build_editable",
"get_requires_for_build_sdist",
"get_requires_for_build_wheel",
"prepare_metadata_for_build_wheel",
]
_MACOS_TARGET = "11.0"
_prepared = False
def _check_zig_available(): def _prepare():
"""Check if Zig is installed and available.""" """Prepare the build environment and build libaegis."""
global _prepared
if _prepared:
return
_prepared = True
# Set macOS deployment target
if sys.platform == "darwin" and "MACOSX_DEPLOYMENT_TARGET" not in os.environ:
os.environ["MACOSX_DEPLOYMENT_TARGET"] = _MACOS_TARGET
# Check Zig is available
if shutil.which("zig") is None: if shutil.which("zig") is None:
raise RuntimeError( raise RuntimeError(
"\n" + "=" * 70 + "\n" "Zig compiler not found. Install from https://ziglang.org/download/"
"ERROR: Zig compiler not found!\n"
"\n"
"Building pyaegis requires the Zig compiler to build the libaegis\n"
"static library. Please install Zig before building this package.\n"
"\n"
"Installation instructions:\n"
" - Visit: https://ziglang.org/download/\n"
" - Or use a package manager:\n"
" * macOS: brew install zig\n"
" * Linux: See https://github.com/ziglang/zig/wiki/Install-Zig-from-a-Package-Manager\n"
" * Windows: choco install zig or scoop install zig\n"
"\n"
"After installing Zig, please try building again.\n" + "=" * 70 + "\n"
) )
# Build libaegis
def _build_libaegis():
"""Build libaegis static library with Zig."""
# Check Zig availability first
_check_zig_available()
libaegis_dir = Path(__file__).parent.parent / "libaegis" libaegis_dir = Path(__file__).parent.parent / "libaegis"
if not libaegis_dir.exists(): cmd = ["zig", "build", "-Drelease"]
raise FileNotFoundError( if sys.platform == "darwin":
f"libaegis directory not found at {libaegis_dir}. " arch = {"arm64": "aarch64", "x86_64": "x86_64"}.get(platform.machine())
"Cannot build static library." if arch:
) cmd.append(f"-Dtarget={arch}-macos.{_MACOS_TARGET}")
subprocess.run(cmd, cwd=libaegis_dir, check=True)
print("Building libaegis static library with Zig...")
try:
subprocess.run(
["zig", "build", "-Drelease"],
cwd=libaegis_dir,
check=True,
capture_output=False,
)
print("Successfully built libaegis static library")
except subprocess.CalledProcessError as e:
print(
f"\nError: Zig build failed with exit code {e.returncode}\n"
f"Command: {' '.join(e.cmd)}\n",
file=sys.stderr,
)
raise
# Expose all the standard build backend hooks build_sdist = build_meta.build_sdist
def get_requires_for_build_wheel(config_settings=None): get_requires_for_build_sdist = build_meta.get_requires_for_build_sdist
"""Return build requirements and ensure libaegis is built first.""" get_requires_for_build_wheel = build_meta.get_requires_for_build_wheel
_build_libaegis() prepare_metadata_for_build_wheel = build_meta.prepare_metadata_for_build_wheel
return _orig.get_requires_for_build_wheel(config_settings)
def get_requires_for_build_sdist(config_settings=None):
"""Return build requirements for sdist and ensure libaegis is built first."""
_build_libaegis()
return _orig.get_requires_for_build_sdist(config_settings)
_orig_prepare_metadata_for_build_wheel = _orig.prepare_metadata_for_build_wheel
_orig_build_sdist = _orig.build_sdist
def prepare_metadata_for_build_wheel(metadata_directory, config_settings=None):
"""Prepare metadata and ensure libaegis is built (some frontends call this early)."""
_build_libaegis()
return _orig_prepare_metadata_for_build_wheel(metadata_directory, config_settings)
def build_sdist(sdist_directory, config_settings=None):
"""Build sdist, building libaegis first so the sdist can include built artifacts if needed."""
_build_libaegis()
return _orig_build_sdist(sdist_directory, config_settings)
# Wheel build hooks - need libaegis built first
def build_wheel(wheel_directory, config_settings=None, metadata_directory=None): def build_wheel(wheel_directory, config_settings=None, metadata_directory=None):
"""Build wheel with libaegis built first.""" _prepare()
_build_libaegis() return build_meta.build_wheel(wheel_directory, config_settings, metadata_directory)
return _orig.build_wheel(wheel_directory, config_settings, metadata_directory)
def build_editable(wheel_directory, config_settings=None, metadata_directory=None): def build_editable(wheel_directory, config_settings=None, metadata_directory=None):
"""Build editable install with libaegis built first.""" _prepare()
_build_libaegis() return build_meta.build_editable(
return _orig.build_editable(wheel_directory, config_settings, metadata_directory) wheel_directory, config_settings, metadata_directory
)
Regular → Executable
+35 -4
View File
@@ -1,3 +1,4 @@
#!/usr/bin/env -S uv run
"""Generate CFFI cdef and Python modules from libaegis C sources.""" """Generate CFFI cdef and Python modules from libaegis C sources."""
import pathlib import pathlib
@@ -267,17 +268,34 @@ def generate_python_modules(
if dst.exists() and dst.read_text(encoding="utf-8") == new_content: if dst.exists() and dst.read_text(encoding="utf-8") == new_content:
unchanged.append(dst) unchanged.append(dst)
else: else:
dst.write_text(new_content, encoding="utf-8") dst.write_bytes(new_content.encode())
updated.append(dst) updated.append(dst)
return updated, unchanged return updated, unchanged
def generate_ciphers_module(constants: Dict[str, Dict[str, int]]) -> str:
labels = [algo_label(variant) for variant in constants]
literal_items = ", ".join(f'"{label}"' for label in labels)
lines = [
"# This file is generated by tools/generate.py. Do not edit.",
"from typing import Literal",
"",
f"CipherName = Literal[{literal_items}]",
"",
"CIPHERS: dict[CipherName, str] = {",
]
for variant in constants:
lines.append(f' "{algo_label(variant)}": "{variant}",')
lines.append("}")
return "\n".join(lines) + "\n"
def main() -> int: def main() -> int:
root = pathlib.Path(__file__).parent.parent root = pathlib.Path(__file__).parent.parent
libaegis_src_dir = root / "libaegis" / "src" libaegis_src_dir = root / "libaegis" / "src"
include_dir = libaegis_src_dir / "include" include_dir = libaegis_src_dir / "include"
pyaegis_dir = root / "pyaegis" pyaegis_dir = root / "src" / "aeg"
if not include_dir.exists(): if not include_dir.exists():
print(f"Include directory not found: {include_dir}", file=sys.stderr) print(f"Include directory not found: {include_dir}", file=sys.stderr)
@@ -301,10 +319,23 @@ def main() -> int:
if cdef_path.exists() and cdef_path.read_text(encoding="utf-8") == cdef_content: if cdef_path.exists() and cdef_path.read_text(encoding="utf-8") == cdef_content:
print(f" - No changes to {cdef_path}", file=sys.stderr) print(f" - No changes to {cdef_path}", file=sys.stderr)
else: else:
cdef_path.write_text(cdef_content, encoding="utf-8") cdef_path.write_bytes(cdef_content.encode())
print(f" - Updated {cdef_path}", file=sys.stderr) print(f" - Updated {cdef_path}", file=sys.stderr)
print("Step 3: Generating Python modules...", file=sys.stderr) print("Step 3: Generating _ciphers.py...", file=sys.stderr)
ciphers_path = pyaegis_dir / "_ciphers.py"
ciphers_content = generate_ciphers_module(constants)
if (
ciphers_path.exists()
and ciphers_path.read_text(encoding="utf-8") == ciphers_content
):
print(f" - No changes to {ciphers_path.name}", file=sys.stderr)
else:
ciphers_path.write_bytes(ciphers_content.encode())
print(f" - Updated {ciphers_path.name}", file=sys.stderr)
print("Step 4: Generating Python modules...", file=sys.stderr)
try: try:
updated, unchanged = generate_python_modules( updated, unchanged = generate_python_modules(
pyaegis_dir / "aegis256x4.py", pyaegis_dir, constants pyaegis_dir / "aegis256x4.py", pyaegis_dir, constants
+484
View File
@@ -0,0 +1,484 @@
#!/usr/bin/env -S uv run
"""Build wheels for all supported Python versions using uv."""
import os
import platform
import shutil
import subprocess
import sys
from pathlib import Path
from packaging.version import Version
# Import generate module from same directory
sys.path.insert(0, str(Path(__file__).parent))
import generate
# Minimum macOS deployment target for compatibility
MACOS_DEPLOYMENT_TARGET = "11.0"
# ABI3 wheel: built once, works for all GIL-enabled Python versions
# We use a recent Python to build since it doesn't affect the wheel compatibility
ABI3_BUILD_VERSION = "3.14+gil"
# All GIL-enabled Python versions covered by the ABI3 wheel
ABI3_COVERED_VERSIONS = [
"3.10",
"3.11",
"3.12",
"3.13+gil",
"3.14+gil",
"3.15+gil",
]
# Non-ABI3 wheels: each needs its own build (free-threaded and PyPy)
NON_ABI3_VERSIONS = [
"3.14t",
"3.15t",
"pypy3.10",
"pypy3.11",
]
# All versions for testing and benchmarking
ALL_PYTHON_VERSIONS = ABI3_COVERED_VERSIONS + NON_ABI3_VERSIONS
def get_version_from_scm():
"""Get version from setuptools-scm (git tags)."""
try:
result = subprocess.run(
["uv", "run", "-m", "setuptools_scm"],
capture_output=True,
text=True,
check=True,
cwd=Path(__file__).parent.parent,
)
return result.stdout.strip()
except subprocess.CalledProcessError as e:
print(f"✗ Error getting version from setuptools-scm: {e}", file=sys.stderr)
return None
def is_release_version(version):
"""Check if version is a clean release (no dev/post/local identifiers)."""
# A release version is just x.y.z with optional alpha/beta/rc suffixes
# No +local or .devN or .postN
if not version:
return False
return not any(marker in version for marker in ["+", ".dev", ".post"])
def get_next_version(current_version):
"""Get the next release version from a dev version."""
# Parse base version (strips dev/local parts)
try:
v = Version(current_version)
return f"{v.major}.{v.minor}.{v.micro}"
except Exception:
return current_version
def is_working_copy_clean():
"""Check if git working copy is clean."""
result = subprocess.run(
["git", "status", "--porcelain"], capture_output=True, text=True
)
return result.returncode == 0 and not result.stdout.strip()
def make_release_message(version):
"""Generate message for making a release."""
next_version = get_next_version(version)
is_clean = is_working_copy_clean()
msg = "\n⚠️ This is not a clean release version; upload to PyPI skipped.\n\n"
msg += f"To create a release (e.g. {next_version}) and upload to PyPI:\n"
if not is_clean:
msg += " 1. Add and commit changes on the working copy\n"
msg += f" 2. Tag the commit: git tag v{next_version}\n"
msg += " 3. Run this script again\n"
msg += f" 4. Push the tag: git push origin v{next_version}\n"
else:
msg += f" 1. Tag the current commit: git tag v{next_version}\n"
msg += " 2. Run this script again\n"
msg += f" 3. Push the tag: git push origin v{next_version}\n"
msg += (
f"\nIf the build didn't work, delete the tag with git tag -d v{next_version}\n"
)
return msg
def run_command(cmd, description=None, env=None):
"""Run a command and handle errors. If description is None, only print the command."""
if description:
print(f"\n{'=' * 70}")
print(f"{description}")
print(f"{'=' * 70}")
print(f">>> {' '.join(cmd)}")
try:
subprocess.run(cmd, check=True, env=env)
return True
except subprocess.CalledProcessError as e:
print(f"✗ Command failed with exit code {e.returncode}", file=sys.stderr)
return False
def get_build_env():
"""Get environment variables for building wheels."""
env = os.environ.copy()
if platform.system() == "Darwin":
env["MACOSX_DEPLOYMENT_TARGET"] = MACOS_DEPLOYMENT_TARGET
return env
def normalize_line_endings(repo_root: Path):
"""Normalize all text files to LF line endings."""
# Patterns for files to normalize
patterns = [
"src/aeg/**/*.py",
"src/aeg/**/*.h",
"tests/**/*.py",
"tools/**/*.py",
"*.py",
"*.md",
"*.txt",
"*.toml",
"*.in",
]
for pattern in patterns:
for file_path in repo_root.glob(pattern):
if file_path.is_file():
content = file_path.read_bytes()
if b"\r\n" in content:
content = content.replace(b"\r\n", b"\n")
file_path.write_bytes(content)
def get_wheel_pattern(py_version: str, abi3: bool = False) -> str:
"""Get the glob pattern for finding a wheel file."""
if abi3:
# ABI3 wheels always use cp310-abi3 tag (minimum supported version)
# regardless of which Python version was used to build
return "aeg-*-cp310-abi3-*.whl"
elif py_version.startswith("pypy"):
# PyPy wheels use pp3XX format
return f"aeg-*-pp{py_version.replace('pypy', '').replace('.', '')}-*.whl"
elif py_version.endswith("t"):
# Free-threaded Python wheels use cpXXX-cpXXXt format (e.g., cp314-cp314t)
base_version = py_version.replace(".", "").replace("t", "")
return f"aeg-*-cp{base_version}-cp{base_version}t-*.whl"
else:
# Regular CPython wheels use cpXXX-cpXXX format
# Strip +gil suffix used to force non-free-threaded build
base_version = py_version.replace(".", "").replace("+gil", "")
return f"aeg-*-cp{base_version}-cp{base_version}-*.whl"
def build_abi3_wheel(dist_dir: Path, py_version: str) -> Path | None:
"""Build the ABI3 wheel using the specified Python version."""
cmd = ["uv", "build", "--python", py_version, "--wheel", "--quiet"]
if not run_command(cmd, env=get_build_env()):
return None
# Find the ABI3 wheel (always tagged cp310-abi3 regardless of build Python version)
wheel_pattern = get_wheel_pattern(py_version, abi3=True)
wheels = list(dist_dir.glob(wheel_pattern))
if not wheels:
print(f"✗ Could not find ABI3 wheel matching {wheel_pattern}", file=sys.stderr)
return None
wheel = wheels[0]
# Repair wheel with auditwheel for manylinux compatibility (Linux only)
if platform.system() == "Linux":
wheel = repair_wheel_linux(dist_dir, wheel, py_version, abi3=True)
if not wheel:
return None
return wheel
def build_wheel_for_version(dist_dir: Path, py_version: str) -> Path | None:
"""Build a wheel for a specific Python version (non-ABI3)."""
cmd = ["uv", "build", "--python", py_version, "--wheel", "--quiet"]
if not run_command(cmd, env=get_build_env()):
return None
# Find the wheel for this version
wheel_pattern = get_wheel_pattern(py_version, abi3=False)
wheels = list(dist_dir.glob(wheel_pattern))
if not wheels:
print(f"✗ Could not find wheel for Python {py_version}", file=sys.stderr)
return None
wheel = wheels[0]
# Repair wheel with auditwheel for manylinux compatibility (Linux only)
if platform.system() == "Linux":
wheel = repair_wheel_linux(dist_dir, wheel, py_version, abi3=False)
if not wheel:
return None
return wheel
def repair_wheel_linux(
dist_dir: Path, wheel: Path, py_version: str, abi3: bool
) -> Path | None:
"""Repair a wheel with auditwheel for manylinux compatibility (Linux only)."""
repair_cmd = [
"uv",
"run",
"auditwheel",
"repair",
str(wheel),
"-w",
str(dist_dir),
]
if not run_command(repair_cmd):
return None
# Find the repaired wheel (it will have a different name)
wheel_pattern = get_wheel_pattern(py_version, abi3=abi3)
all_wheels = list(dist_dir.glob(wheel_pattern))
repaired_wheels = [w for w in all_wheels if "linux_x86_64" not in str(w)]
if not repaired_wheels:
print(
f"✗ Could not find repaired (manylinux) wheel for Python {py_version}",
file=sys.stderr,
)
return None
repaired_wheel = repaired_wheels[0]
# Remove the unrepaired linux_x86_64 wheels
for w in all_wheels:
if "linux_x86_64" in str(w):
w.unlink()
return repaired_wheel
def test_wheel(wheel: Path, py_version: str) -> bool:
"""Test a wheel with pytest."""
# --isolated: avoid .venv conflicts
# --no-project: don't build from source in current directory, use the wheel
# --refresh-package: force uv to not use cached old versions
test_cmd = [
"uv",
"run",
"--isolated",
"--no-project",
"--refresh-package",
"aeg",
"--python",
py_version,
"--with",
str(wheel),
"--with",
"pytest",
"pytest",
"tests/",
]
return run_command(test_cmd)
def run_benchmark(wheel: Path, py_version: str) -> bool:
"""Run benchmark for a wheel."""
# --isolated: avoid .venv conflicts
# --no-project: don't build from source in current directory, use the wheel
# --refresh-package: force uv to not use cached old versions
bench_cmd = [
"uv",
"run",
"--isolated",
"--no-project",
"--refresh-package",
"aeg",
"--python",
py_version,
"--with",
str(wheel),
"-m",
"aeg.benchmark",
]
return run_command(bench_cmd)
return True
def main():
"""Build wheels for all supported Python versions."""
repo_root = Path(__file__).parent.parent
dist_dir = repo_root / "dist"
# Generate CFFI definitions and Python modules
print(f"\n{'=' * 70}")
print("Code generation from C headers (tools/generate.py)")
print(f"{'=' * 70}")
if generate.main() != 0:
print("✗ Code generation failed", file=sys.stderr)
return 1
# Run ruff to check and fix any issues
print(f"\n{'=' * 70}")
print("Linting and formatting")
print(f"{'=' * 70}")
if not run_command(["uv", "run", "ruff", "check", "--fix", "."]):
print("✗ Ruff check failed", file=sys.stderr)
return 1
# Run ruff format
if not run_command(["uv", "run", "ruff", "format", "."]):
print("✗ Ruff format failed", file=sys.stderr)
return 1
# Normalize all line endings to LF (important for consistent builds)
normalize_line_endings(repo_root)
# Get version from git repo
version = get_version_from_scm()
if not version:
return 1
is_release = is_release_version(version)
# Main header for the packaging process
print(f"\n{'=' * 70}")
print(
f"Packaging aeg-{version}"
+ (" for release" if is_release else " (not release)")
)
print(f"Building: 1 ABI3 wheel (for Python {', '.join(ABI3_COVERED_VERSIONS)})")
print(
f" + {len(NON_ABI3_VERSIONS)} non-ABI3 wheels ({', '.join(NON_ABI3_VERSIONS)})"
)
print(f"Testing/benchmarking: {len(ALL_PYTHON_VERSIONS)} Python versions")
print(f"Output directory: {dist_dir}", end=" ")
# Clean dist directory
if dist_dir.exists():
print("(wiped)")
shutil.rmtree(dist_dir)
else:
print("(created)")
# Clean build directory to remove stale CFFI-generated C code and .so files
build_dir = repo_root / "build"
if build_dir.exists():
print(f"Cleaning build directory: {build_dir}")
shutil.rmtree(build_dir)
# Build distributions
print(f"\n{'=' * 70}")
print("Building distributions")
print(f"{'=' * 70}")
# Build source distribution first
if not run_command(["uv", "build", "--sdist", "--quiet"], env=get_build_env()):
print("✗ Source distribution build failed", file=sys.stderr)
return 1
failed_builds = []
failed_tests = []
successful_wheels = []
wheel_for_version = {} # Map Python version to wheel path
# Build ABI3 wheel (once, works for all GIL-enabled versions)
abi3_wheel = build_abi3_wheel(dist_dir, ABI3_BUILD_VERSION)
if abi3_wheel:
successful_wheels.append(abi3_wheel)
# This wheel works for all ABI3-covered versions
for py_version in ABI3_COVERED_VERSIONS:
wheel_for_version[py_version] = abi3_wheel
else:
failed_builds.append(f"abi3 (built with {ABI3_BUILD_VERSION})")
# Build non-ABI3 wheels (free-threaded and PyPy)
for py_version in NON_ABI3_VERSIONS:
wheel = build_wheel_for_version(dist_dir, py_version)
if wheel:
successful_wheels.append(wheel)
wheel_for_version[py_version] = wheel
else:
failed_builds.append(py_version)
# Test and benchmark each Python version with its appropriate wheel
print(f"\n{'=' * 70}")
print("Testing and benchmarking")
print(f"{'=' * 70}")
for py_version in ALL_PYTHON_VERSIONS:
wheel = wheel_for_version.get(py_version)
if not wheel:
# No wheel available for this version (build failed)
continue
# Test the wheel with pytest
if not test_wheel(wheel, py_version):
failed_tests.append(py_version)
continue
# Run benchmark
if not run_benchmark(wheel, py_version):
failed_tests.append(py_version)
continue
# Summary
print(f"\n{'=' * 70}")
print("BUILD SUMMARY")
print(f"{'=' * 70}")
print(
f"Successful builds: sdist and {len(successful_wheels)} wheels "
f"(1 abi3 + {len(NON_ABI3_VERSIONS)} non-abi3)"
)
print(
f"Tests/benchmarks passed: {len(ALL_PYTHON_VERSIONS) - len(failed_tests) - len(failed_builds)}/{len(ALL_PYTHON_VERSIONS)} Python versions"
)
if failed_builds:
print(f"\nFailed builds: {len(failed_builds)}")
for failed_version in failed_builds:
print(f"{failed_version}")
if failed_tests:
print(f"\nFailed tests/benchmarks: {len(failed_tests)}")
for failed_version in failed_tests:
print(f" ✗ Python {failed_version}")
if not successful_wheels:
print("\n✗ No successful wheels to upload")
return 1
# List files to upload
sdist = list(dist_dir.glob("*.tar.gz"))
upload_files = sdist + successful_wheels
for file in upload_files:
print(f" - {file.name}")
# Only upload if this is a clean release version
if not is_release:
print(make_release_message(version))
return 0
# Upload with twine
upload_cmd = ["uvx", "twine", "upload"] + [str(f) for f in upload_files]
if not run_command(upload_cmd, "Uploading to PyPI with twine"):
print("\n✗ Upload failed")
return 1
print(f"\n{'=' * 70}")
print("All builds and upload completed successfully!")
print(f"{'=' * 70}")
print()
return 0
if __name__ == "__main__":
try:
sys.exit(main())
except KeyboardInterrupt:
sys.exit(1)