Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1ecf0d306f | ||
|
|
5824df28cb | ||
|
|
938239ef72 | ||
|
|
f3c0b2b85d | ||
|
|
dd0c0bc0a0 | ||
|
|
40560d2deb | ||
|
|
6fc3098650 | ||
|
|
ff97bf42a8 | ||
|
|
ba2179e827 | ||
|
|
640e812908 | ||
|
|
a644d27b56 | ||
|
|
4445c256b1 | ||
|
|
04f1c3067c | ||
|
|
39e5f00f42 | ||
|
|
88efc4cabc | ||
|
|
04b11e9925 | ||
|
|
d8a9a7ee9d | ||
|
|
f5430a6ad4 |
+3
-3
@@ -3,9 +3,9 @@
|
||||
*.egg-info
|
||||
/dist
|
||||
/build
|
||||
/pyaegis/build
|
||||
/pyaegis/_aegis.*.so
|
||||
/pyaegis/_aegis.*.pyd
|
||||
/src/aeg/build
|
||||
/src/aeg/_aegis*.so
|
||||
/src/aeg/_aegis*.pyd
|
||||
__pycache__
|
||||
!.gitignore
|
||||
!.gitmodules
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -11,7 +11,7 @@ This document contains instructions for developers who want to build pyaegis fro
|
||||
|
||||
### 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`
|
||||
- **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:
|
||||
|
||||
```fish
|
||||
git clone --recursive https://github.com/LeoVasanko/pyaegis.git
|
||||
cd pyaegis
|
||||
git clone --recursive https://github.com/LeoVasanko/aeg.git
|
||||
cd aeg
|
||||
```
|
||||
|
||||
If you already cloned without `--recursive`, initialize submodules:
|
||||
@@ -74,7 +74,7 @@ This creates files in the `dist/` directory.
|
||||
|
||||
## 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
|
||||
python tools/generate.py
|
||||
@@ -100,7 +100,7 @@ If you cannot install Zig, you may manually compile in the libaegis folder (Zig,
|
||||
|
||||
## Project Structure
|
||||
|
||||
- `pyaegis/` - Python package source
|
||||
- `src/aeg/` - Python package source
|
||||
- `libaegis/` - C library source (submodule)
|
||||
- `tests/` - Test suite
|
||||
- `tools/` - Code generation scripts and `build_backend.py` used to build libaegis
|
||||
|
||||
+1
-3
@@ -1,13 +1,11 @@
|
||||
include pyaegis/aegis_cdef.h
|
||||
include src/aeg/aegis_cdef.h
|
||||
include setup.py
|
||||
include tools/build_backend.py
|
||||
include BUILD.md
|
||||
include README.md
|
||||
recursive-include libaegis *.c *.h *.zig *.zon
|
||||
include libaegis/CMakeLists.txt
|
||||
include libaegis/LICENSE
|
||||
include libaegis/README.md
|
||||
recursive-include libaegis/cmake *.cmake *.cmake.in
|
||||
graft libaegis/src
|
||||
include libaegis/build.zig
|
||||
include libaegis/build.zig.zon
|
||||
|
||||
@@ -1,33 +1,41 @@
|
||||
# pyaegis
|
||||
# AEGIS Cipher Python Binding
|
||||
|
||||
[](https://badge.fury.io/py/pyaegis)
|
||||
[](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
|
||||
|
||||
Using [uv](https://docs.astral.sh/uv/getting-started/installation/):
|
||||
```fish
|
||||
uv pip install git+https://github.com/LeoVasanko/pyaegis.git
|
||||
```sh
|
||||
pip install aeg
|
||||
```
|
||||
|
||||
For development builds, see BUILD.md.
|
||||
Or add to your project using [UV](https://docs.astral.sh/uv/getting-started/installation/):
|
||||
```sh
|
||||
uv add aeg
|
||||
```
|
||||
|
||||
## Variants
|
||||
|
||||
All submodules expose the same API; pick one for your key/nonce size and platform:
|
||||
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.
|
||||
|
||||
- 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)
|
||||
| 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 |
|
||||
|
||||
## Quick start
|
||||
|
||||
Normal authenticated encryption using the AEGIS-128X4 algorithm:
|
||||
|
||||
```python
|
||||
from pyaegis import aegis128x4 as ciph
|
||||
from aeg import aegis128x4 as ciph
|
||||
|
||||
key = ciph.random_key() # Secret key (stored securely)
|
||||
nonce = ciph.random_nonce() # Public nonce (recreated for each message)
|
||||
@@ -49,7 +57,7 @@ Common parameters and returns (applies to all items below):
|
||||
- into: optional output buffer (see below)
|
||||
- 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.
|
||||
|
||||
@@ -83,13 +91,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.
|
||||
|
||||
- mac(key, nonce, data, maclen=16, into=None) -> mac
|
||||
- mac(key, nonce, data, maclen=16, into=None) -> mac bytes
|
||||
- Mac(key, nonce, maclen=16)
|
||||
- update(data)
|
||||
- final([into]) -> mac
|
||||
- final([into]) -> mac bytes
|
||||
- verify(mac) -> raises ValueError on failure
|
||||
- digest() -> bytes
|
||||
- hexdigest() -> str
|
||||
- digest() -> mac bytes
|
||||
- hexdigest() -> mac str
|
||||
- reset()
|
||||
- clone() -> Mac
|
||||
|
||||
@@ -123,7 +131,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:
|
||||
```python
|
||||
from pyaegis import aegis256x4 as ciph
|
||||
from aeg import aegis256x4 as ciph
|
||||
key, nonce = ciph.random_key(), bytes(ciph.NONCEBYTES)
|
||||
|
||||
mac = ciph.mac(key, nonce, b"message", maclen=32)
|
||||
@@ -146,7 +154,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.
|
||||
|
||||
```python
|
||||
from pyaegis import aegis256x4 as ciph
|
||||
from aeg import aegis256x4 as ciph
|
||||
key, nonce = ciph.random_key(), ciph.random_nonce()
|
||||
|
||||
ct, mac = ciph.encrypt_detached(key, nonce, b"secret", ad=b"header")
|
||||
@@ -162,7 +170,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).
|
||||
|
||||
```python
|
||||
from pyaegis import aegis256x4 as ciph
|
||||
from aeg import aegis256x4 as ciph
|
||||
key, nonce = ciph.random_key(), ciph.random_nonce()
|
||||
|
||||
enc = ciph.Encryptor(key, nonce, ad=b"header", maclen=16)
|
||||
@@ -181,16 +189,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.
|
||||
|
||||
```python
|
||||
from pyaegis import aegis128x4 as ciph
|
||||
|
||||
message = bytearray(30 * b"Attack at dawn! ")
|
||||
# Encryption settings
|
||||
from aeg import aegis128x4 as ciph
|
||||
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
|
||||
maclen = ciph.MACBYTES # 16
|
||||
|
||||
message = bytearray(30 * b"Attack at dawn! ")
|
||||
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:
|
||||
chunk = message[:framebytes - maclen]
|
||||
del message[:len(chunk)]
|
||||
@@ -200,9 +209,8 @@ with open("encrypted.bin", "wb") as f:
|
||||
```
|
||||
|
||||
```python
|
||||
from pyaegis import aegis128x4 as ciph
|
||||
|
||||
# Decryption needs same values as encryption
|
||||
from aeg import aegis128x4 as ciph
|
||||
key = b"sixteenbyte key!"
|
||||
framebytes = 80
|
||||
maclen = ciph.MACBYTES
|
||||
@@ -223,7 +231,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.
|
||||
|
||||
```python
|
||||
from pyaegis import aegis128x4 as ciph
|
||||
from aeg import aegis128x4 as ciph
|
||||
|
||||
key = b"SeedForReplay001" # A non-random deterministic seed (16 bytes)
|
||||
nonce = bytearray(ciph.NONCEBYTES) # All-zeroes nonce
|
||||
@@ -239,7 +247,7 @@ Note: this is seekable by converting the block number to nonce with `idx.to_byte
|
||||
|
||||
### 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.
|
||||
|
||||
@@ -247,18 +255,18 @@ Foreign arrays can be used. This example fills a Numpy array with random integer
|
||||
|
||||
```python
|
||||
import numpy as np
|
||||
from pyaegis import aegis128x4 as ciph
|
||||
from aeg import aegis128x4 as ciph
|
||||
key, nonce = ciph.random_key(), ciph.random_nonce()
|
||||
|
||||
arr = np.empty(10, dtype=np.uint64) # Uninitialised integer array
|
||||
ciph.stream(key, nonce, into=arr) # Fill with random bytes
|
||||
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:
|
||||
|
||||
```python
|
||||
from pyaegis import aegis256x4 as ciph
|
||||
from aeg import aegis256x4 as ciph
|
||||
key, nonce = ciph.random_key(), ciph.random_nonce()
|
||||
buf = memoryview(bytearray(1000)) # memoryview[:len] is still in the same buffer (no copy)
|
||||
buf[:7] = b"message"
|
||||
@@ -276,33 +284,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.
|
||||
|
||||
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
|
||||
uv run -m pyaegis.benchmark
|
||||
```
|
||||
|
||||
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.
|
||||
|
||||
```fish
|
||||
$ python -m pyaegis.benchmark
|
||||
AEGIS-256 107666.56 Mb/s
|
||||
AEGIS-256X2 191314.53 Mb/s
|
||||
AEGIS-256X4 211537.44 Mb/s
|
||||
AEGIS-128L 159074.08 Mb/s
|
||||
AEGIS-128X2 307332.53 Mb/s
|
||||
AEGIS-128X4 230106.70 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
|
||||
```sh
|
||||
uv run -m aeg.benchmark
|
||||
AEGIS-256 103166.24 Mb/s
|
||||
AEGIS-256X2 184225.50 Mb/s
|
||||
AEGIS-256X4 194018.26 Mb/s
|
||||
AEGIS-128L 161551.73 Mb/s
|
||||
AEGIS-128X2 281987.80 Mb/s
|
||||
AEGIS-128X4 217997.37 Mb/s
|
||||
AEGIS-128L MAC 188886.40 Mb/s
|
||||
AEGIS-128X2 MAC 306457.97 Mb/s
|
||||
AEGIS-128X4 MAC 299576.59 Mb/s
|
||||
AEGIS-256 MAC 100914.04 Mb/s
|
||||
AEGIS-256X2 MAC 190208.20 Mb/s
|
||||
AEGIS-256X4 MAC 315919.87 Mb/s
|
||||
```
|
||||
|
||||
The Python library performance is similar to that of the C library:
|
||||
```fish
|
||||
$ ./libaegis/zig-out/bin/benchmark
|
||||
```sh
|
||||
./libaegis/zig-out/bin/benchmark
|
||||
AEGIS-256 107820.86 Mb/s
|
||||
AEGIS-256X2 205025.57 Mb/s
|
||||
AEGIS-256X4 223361.81 Mb/s
|
||||
@@ -316,3 +318,7 @@ AEGIS-256 MAC 116776.62 Mb/s
|
||||
AEGIS-256X2 MAC 224150.04 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.
|
||||
|
||||
+15
-9
@@ -1,16 +1,15 @@
|
||||
[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"
|
||||
backend-path = ["tools"]
|
||||
|
||||
[project]
|
||||
name = "pyaegis"
|
||||
version = "0.3.0"
|
||||
description = "Python bindings for libaegis"
|
||||
name = "aeg"
|
||||
dynamic = ["version"]
|
||||
description = "AEGIS encryption easy to use Python binding. Wheels for major platforms."
|
||||
readme = {file = "README.md", content-type = "text/markdown"}
|
||||
requires-python = ">=3.10"
|
||||
classifiers = [
|
||||
"Programming Language :: Python :: 3",
|
||||
"Programming Language :: Python :: 3 :: Only",
|
||||
"Programming Language :: Python :: Implementation :: CPython",
|
||||
"Operating System :: OS Independent",
|
||||
"Topic :: Security :: Cryptography",
|
||||
@@ -20,16 +19,23 @@ dependencies = [
|
||||
]
|
||||
|
||||
[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]
|
||||
dev = [
|
||||
"auditwheel>=6.5.0",
|
||||
"pytest>=8.4.2",
|
||||
"ruff>=0.14.4",
|
||||
"setuptools>=80.9.0",
|
||||
"setuptools-scm>=9.2.2",
|
||||
]
|
||||
|
||||
[tool.setuptools]
|
||||
packages = ["pyaegis"]
|
||||
package-dir = {"" = "src"}
|
||||
packages = ["aeg"]
|
||||
|
||||
[tool.setuptools.package-data]
|
||||
pyaegis = ["*.h", "*.so", "*.pyd"]
|
||||
aeg = ["*.h", "*.so", "*.pyd"]
|
||||
|
||||
[tool.setuptools_scm]
|
||||
|
||||
@@ -1,49 +1,35 @@
|
||||
"""Setup script for pyaegis - builds CFFI extension linking to libaegis.a"""
|
||||
"""Setup script for aeg - builds CFFI extension with libaegis C library."""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from cffi import FFI
|
||||
from setuptools import setup
|
||||
|
||||
# Locate the static library (built by build_backend.py before this runs)
|
||||
lib_name = "aegis.lib" if sys.platform == "win32" else "libaegis.a"
|
||||
libaegis_static = Path("libaegis/zig-out/lib") / lib_name
|
||||
if not libaegis_static.exists():
|
||||
raise RuntimeError(f"libaegis static library not found at {libaegis_static}")
|
||||
libaegis_static = str(libaegis_static.resolve())
|
||||
|
||||
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
|
||||
|
||||
# Include directory for headers
|
||||
libaegis_include = Path("libaegis/src/include")
|
||||
if not libaegis_include.exists():
|
||||
raise RuntimeError(f"libaegis include directory not found at {libaegis_include}")
|
||||
include_dirs = [str(libaegis_include)]
|
||||
|
||||
# Read the CDEF header
|
||||
cdef_path = Path(__file__).parent / "pyaegis" / "aegis_cdef.h"
|
||||
cdef_path = Path(__file__).parent / "src" / "aeg" / "aegis_cdef.h"
|
||||
cdef_content = cdef_path.read_text(encoding="utf-8")
|
||||
|
||||
# Create CFFI builder
|
||||
ffibuilder = FFI()
|
||||
ffibuilder.cdef(cdef_content)
|
||||
|
||||
# Include directory for headers
|
||||
include_dirs = []
|
||||
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(
|
||||
"pyaegis._aegis", # module name
|
||||
"aeg._aegis", # module name
|
||||
"""
|
||||
#include "aegis.h"
|
||||
#include "aegis128l.h"
|
||||
@@ -54,7 +40,7 @@ ffibuilder.set_source(
|
||||
#include "aegis256x4.h"
|
||||
""",
|
||||
include_dirs=include_dirs,
|
||||
extra_objects=[libaegis_static] if libaegis_static else [],
|
||||
extra_objects=[libaegis_static],
|
||||
)
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
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 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)}")
|
||||
@@ -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,6 +1,6 @@
|
||||
"""Loader for libaegis CFFI extension module."""
|
||||
|
||||
from pyaegis._aegis import ffi, lib
|
||||
from aeg._aegis import ffi, lib
|
||||
|
||||
__all__ = ["ffi", "lib"]
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
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: ...
|
||||
@@ -295,9 +295,7 @@ def decrypt(
|
||||
out = bytearray(expected_out)
|
||||
else:
|
||||
if into.nbytes < expected_out:
|
||||
raise TypeError(
|
||||
"into length must be at least ct.nbytes - maclen"
|
||||
)
|
||||
raise TypeError("into length must be at least ct.nbytes - maclen")
|
||||
out = into
|
||||
|
||||
rc = _lib.aegis128l_decrypt(
|
||||
@@ -582,7 +580,9 @@ class Mac:
|
||||
out = into
|
||||
|
||||
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:
|
||||
err_num = ffi.errno
|
||||
err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
|
||||
@@ -851,7 +851,9 @@ class Decryptor:
|
||||
err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
|
||||
raise RuntimeError(f"state decrypt update failed: {err_name}")
|
||||
w = int(written[0])
|
||||
assert w == expected_out, f"got {w}, expected {expected_out}, ct.nbytes={ct.nbytes}"
|
||||
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:
|
||||
@@ -295,9 +295,7 @@ def decrypt(
|
||||
out = bytearray(expected_out)
|
||||
else:
|
||||
if into.nbytes < expected_out:
|
||||
raise TypeError(
|
||||
"into length must be at least ct.nbytes - maclen"
|
||||
)
|
||||
raise TypeError("into length must be at least ct.nbytes - maclen")
|
||||
out = into
|
||||
|
||||
rc = _lib.aegis128x2_decrypt(
|
||||
@@ -582,7 +580,9 @@ class Mac:
|
||||
out = into
|
||||
|
||||
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:
|
||||
err_num = ffi.errno
|
||||
err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
|
||||
@@ -851,7 +851,9 @@ class Decryptor:
|
||||
err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
|
||||
raise RuntimeError(f"state decrypt update failed: {err_name}")
|
||||
w = int(written[0])
|
||||
assert w == expected_out, f"got {w}, expected {expected_out}, ct.nbytes={ct.nbytes}"
|
||||
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:
|
||||
@@ -295,9 +295,7 @@ def decrypt(
|
||||
out = bytearray(expected_out)
|
||||
else:
|
||||
if into.nbytes < expected_out:
|
||||
raise TypeError(
|
||||
"into length must be at least ct.nbytes - maclen"
|
||||
)
|
||||
raise TypeError("into length must be at least ct.nbytes - maclen")
|
||||
out = into
|
||||
|
||||
rc = _lib.aegis128x4_decrypt(
|
||||
@@ -582,7 +580,9 @@ class Mac:
|
||||
out = into
|
||||
|
||||
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:
|
||||
err_num = ffi.errno
|
||||
err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
|
||||
@@ -851,7 +851,9 @@ class Decryptor:
|
||||
err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
|
||||
raise RuntimeError(f"state decrypt update failed: {err_name}")
|
||||
w = int(written[0])
|
||||
assert w == expected_out, f"got {w}, expected {expected_out}, ct.nbytes={ct.nbytes}"
|
||||
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:
|
||||
@@ -295,9 +295,7 @@ def decrypt(
|
||||
out = bytearray(expected_out)
|
||||
else:
|
||||
if into.nbytes < expected_out:
|
||||
raise TypeError(
|
||||
"into length must be at least ct.nbytes - maclen"
|
||||
)
|
||||
raise TypeError("into length must be at least ct.nbytes - maclen")
|
||||
out = into
|
||||
|
||||
rc = _lib.aegis256_decrypt(
|
||||
@@ -582,7 +580,9 @@ class Mac:
|
||||
out = into
|
||||
|
||||
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:
|
||||
err_num = ffi.errno
|
||||
err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
|
||||
@@ -851,7 +851,9 @@ class Decryptor:
|
||||
err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
|
||||
raise RuntimeError(f"state decrypt update failed: {err_name}")
|
||||
w = int(written[0])
|
||||
assert w == expected_out, f"got {w}, expected {expected_out}, ct.nbytes={ct.nbytes}"
|
||||
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:
|
||||
@@ -295,9 +295,7 @@ def decrypt(
|
||||
out = bytearray(expected_out)
|
||||
else:
|
||||
if into.nbytes < expected_out:
|
||||
raise TypeError(
|
||||
"into length must be at least ct.nbytes - maclen"
|
||||
)
|
||||
raise TypeError("into length must be at least ct.nbytes - maclen")
|
||||
out = into
|
||||
|
||||
rc = _lib.aegis256x2_decrypt(
|
||||
@@ -582,7 +580,9 @@ class Mac:
|
||||
out = into
|
||||
|
||||
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:
|
||||
err_num = ffi.errno
|
||||
err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
|
||||
@@ -851,7 +851,9 @@ class Decryptor:
|
||||
err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
|
||||
raise RuntimeError(f"state decrypt update failed: {err_name}")
|
||||
w = int(written[0])
|
||||
assert w == expected_out, f"got {w}, expected {expected_out}, ct.nbytes={ct.nbytes}"
|
||||
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:
|
||||
@@ -295,9 +295,7 @@ def decrypt(
|
||||
out = bytearray(expected_out)
|
||||
else:
|
||||
if into.nbytes < expected_out:
|
||||
raise TypeError(
|
||||
"into length must be at least ct.nbytes - maclen"
|
||||
)
|
||||
raise TypeError("into length must be at least ct.nbytes - maclen")
|
||||
out = into
|
||||
|
||||
rc = _lib.aegis256x4_decrypt(
|
||||
@@ -582,7 +580,9 @@ class Mac:
|
||||
out = into
|
||||
|
||||
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:
|
||||
err_num = ffi.errno
|
||||
err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
|
||||
@@ -851,7 +851,9 @@ class Decryptor:
|
||||
err_name = errno.errorcode.get(err_num, f"errno_{err_num}")
|
||||
raise RuntimeError(f"state decrypt update failed: {err_name}")
|
||||
w = int(written[0])
|
||||
assert w == expected_out, f"got {w}, expected {expected_out}, ct.nbytes={ct.nbytes}"
|
||||
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:
|
||||
@@ -12,7 +12,7 @@ Output format and throughput units mirror the Zig benchmark (Mb/s).
|
||||
import secrets
|
||||
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
|
||||
ITERATIONS = 100
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Utility helpers for pyaegis.
|
||||
"""Utility helpers for aeg.
|
||||
|
||||
Currently provides Python-side aligned allocation helpers that avoid relying
|
||||
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"]
|
||||
|
||||
try:
|
||||
from collections.abc import Buffer as _Buffer # type: ignore[misc]
|
||||
|
||||
class Buffer(_Buffer, Protocol): # type: ignore[misc]
|
||||
pass
|
||||
from collections.abc import Buffer # type: ignore
|
||||
except ImportError:
|
||||
|
||||
# Fallback for Python < 3.12
|
||||
class Buffer(Protocol):
|
||||
def __buffer__(self, flags: int) -> memoryview: ...
|
||||
|
||||
@@ -75,13 +72,11 @@ def nonce_increment(nonce: Buffer) -> None:
|
||||
|
||||
|
||||
def wipe(buffer: Buffer) -> None:
|
||||
"""Set all bytes of the input buffer to zero.
|
||||
|
||||
Useful for securely clearing sensitive data from memory.
|
||||
"""Securely clearing sensitive data from memory. Sets all bytes of the buffer to 0xFF.
|
||||
|
||||
Args:
|
||||
buffer: The buffer to wipe (modified in place).
|
||||
"""
|
||||
n = memoryview(buffer)
|
||||
for i in range(len(n)):
|
||||
n[i] = 0
|
||||
# This is the fastest method I have found in Python
|
||||
n = memoryview(buffer).cast("B")
|
||||
n[:] = b"\xff" * len(n)
|
||||
@@ -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)
|
||||
@@ -3,7 +3,7 @@ from pathlib import Path
|
||||
|
||||
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
|
||||
|
||||
|
||||
+1
-1
@@ -3,7 +3,7 @@ from pathlib import Path
|
||||
|
||||
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
|
||||
|
||||
|
||||
@@ -6,7 +6,7 @@ after calling final(), preventing accidental misuse.
|
||||
|
||||
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_ALGORITHMS = [aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4]
|
||||
|
||||
@@ -15,7 +15,7 @@ def _check_zig_available():
|
||||
"\n" + "=" * 70 + "\n"
|
||||
"ERROR: Zig compiler not found!\n"
|
||||
"\n"
|
||||
"Building pyaegis requires the Zig compiler to build the libaegis\n"
|
||||
"Building aeg requires the Zig compiler to build the libaegis\n"
|
||||
"static library. Please install Zig before building this package.\n"
|
||||
"\n"
|
||||
"Installation instructions:\n"
|
||||
|
||||
Regular → Executable
+35
-4
@@ -1,3 +1,4 @@
|
||||
#!/usr/bin/env -S uv run
|
||||
"""Generate CFFI cdef and Python modules from libaegis C sources."""
|
||||
|
||||
import pathlib
|
||||
@@ -267,17 +268,34 @@ def generate_python_modules(
|
||||
if dst.exists() and dst.read_text(encoding="utf-8") == new_content:
|
||||
unchanged.append(dst)
|
||||
else:
|
||||
dst.write_text(new_content, encoding="utf-8")
|
||||
dst.write_bytes(new_content.encode())
|
||||
updated.append(dst)
|
||||
|
||||
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:
|
||||
root = pathlib.Path(__file__).parent.parent
|
||||
libaegis_src_dir = root / "libaegis" / "src"
|
||||
include_dir = libaegis_src_dir / "include"
|
||||
pyaegis_dir = root / "pyaegis"
|
||||
pyaegis_dir = root / "src" / "aeg"
|
||||
|
||||
if not include_dir.exists():
|
||||
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:
|
||||
print(f" - No changes to {cdef_path}", file=sys.stderr)
|
||||
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("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:
|
||||
updated, unchanged = generate_python_modules(
|
||||
pyaegis_dir / "aegis256x4.py", pyaegis_dir, constants
|
||||
|
||||
Executable
+346
@@ -0,0 +1,346 @@
|
||||
#!/usr/bin/env -S uv run
|
||||
"""Build wheels for all supported Python versions using uv."""
|
||||
|
||||
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
|
||||
|
||||
PYTHON_VERSIONS = [
|
||||
"3.10",
|
||||
"3.11",
|
||||
"3.12",
|
||||
"3.13",
|
||||
"3.14",
|
||||
"3.14t",
|
||||
"3.15",
|
||||
"3.15t",
|
||||
"pypy3.10",
|
||||
"pypy3.11",
|
||||
]
|
||||
|
||||
|
||||
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):
|
||||
"""Run a command and handle errors."""
|
||||
print(f"\n{'=' * 70}")
|
||||
print(f"{description}")
|
||||
print(f"{'=' * 70}")
|
||||
print(f">>> {' '.join(cmd)}")
|
||||
try:
|
||||
subprocess.run(cmd, check=True)
|
||||
print(f"✓ {description} completed successfully")
|
||||
return True
|
||||
except subprocess.CalledProcessError as e:
|
||||
print(f"✗ {description} failed with exit code {e.returncode}", file=sys.stderr)
|
||||
return False
|
||||
|
||||
|
||||
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 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
|
||||
if not run_command(
|
||||
["uv", "run", "ruff", "check", "--fix", "."], "Running ruff check --fix"
|
||||
):
|
||||
print("✗ Ruff check failed", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
# Run ruff format
|
||||
if not run_command(["uv", "run", "ruff", "format", "."], "Running 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 wheels for Python versions: {', '.join(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)")
|
||||
print(f"{'=' * 70}")
|
||||
|
||||
# Build source distribution first
|
||||
if not run_command(
|
||||
["uv", "build", "--sdist", "--quiet"], "Building source distribution"
|
||||
):
|
||||
print("✗ Source distribution build failed", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
failed_builds = []
|
||||
successful_wheels = []
|
||||
|
||||
for py_version in PYTHON_VERSIONS:
|
||||
# Build wheel
|
||||
description = f"Building wheel for Python {py_version}"
|
||||
cmd = ["uv", "build", "--python", py_version, "--wheel", "--quiet"]
|
||||
|
||||
if not run_command(cmd, description):
|
||||
failed_builds.append(py_version)
|
||||
continue
|
||||
|
||||
# Find the wheel for this version
|
||||
if py_version.startswith("pypy"):
|
||||
# PyPy wheels use pp3XX format
|
||||
wheel_pattern = (
|
||||
f"aeg-*-pp{py_version.replace('pypy', '').replace('.', '')}-*.whl"
|
||||
)
|
||||
else:
|
||||
wheel_pattern = (
|
||||
f"aeg-*-cp{py_version.replace('.', '').replace('t', '')}-*.whl"
|
||||
)
|
||||
wheels = list(dist_dir.glob(wheel_pattern))
|
||||
if not wheels:
|
||||
print(f"✗ Could not find wheel for Python {py_version}", file=sys.stderr)
|
||||
failed_builds.append(py_version)
|
||||
continue
|
||||
|
||||
wheel = wheels[0]
|
||||
|
||||
# Repair wheel with auditwheel for manylinux compatibility (Linux only)
|
||||
if platform.system() == "Linux":
|
||||
repair_cmd = [
|
||||
"uv",
|
||||
"run",
|
||||
"auditwheel",
|
||||
"repair",
|
||||
str(wheel),
|
||||
"-w",
|
||||
str(dist_dir),
|
||||
]
|
||||
if not run_command(
|
||||
repair_cmd, f"Repairing wheel for Python {py_version} with auditwheel"
|
||||
):
|
||||
print(
|
||||
f"✗ Auditwheel repair failed for Python {py_version}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
failed_builds.append(py_version)
|
||||
continue
|
||||
|
||||
# Find the repaired wheel (it will have a different name)
|
||||
all_wheels = list(
|
||||
dist_dir.glob(f"aeg-*-cp{py_version.replace('.', '')}-*.whl")
|
||||
)
|
||||
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,
|
||||
)
|
||||
failed_builds.append(py_version)
|
||||
continue
|
||||
|
||||
wheel = repaired_wheels[0] # Use the repaired wheel for testing
|
||||
|
||||
# Remove the unrepaired linux_x86_64 wheels
|
||||
for w in all_wheels:
|
||||
if "linux_x86_64" in str(w):
|
||||
w.unlink()
|
||||
|
||||
# Test the wheel with pytest (use --isolated to avoid .venv conflicts)
|
||||
test_cmd = [
|
||||
"uv",
|
||||
"run",
|
||||
"--isolated",
|
||||
"--python",
|
||||
py_version,
|
||||
"--with",
|
||||
str(wheel),
|
||||
"--with",
|
||||
"pytest",
|
||||
"pytest",
|
||||
]
|
||||
if not run_command(
|
||||
test_cmd, f"Testing wheel for Python {py_version} with pytest"
|
||||
):
|
||||
print(f"✗ Tests failed for Python {py_version}", file=sys.stderr)
|
||||
failed_builds.append(py_version)
|
||||
continue
|
||||
|
||||
# Run benchmark (use --isolated to avoid .venv conflicts)
|
||||
bench_cmd = [
|
||||
"uv",
|
||||
"run",
|
||||
"--isolated",
|
||||
"--python",
|
||||
py_version,
|
||||
"--with",
|
||||
str(wheel),
|
||||
"-m",
|
||||
"aeg.benchmark",
|
||||
]
|
||||
if not run_command(bench_cmd, f"Running benchmark for Python {py_version}"):
|
||||
print(f"✗ Benchmark failed for Python {py_version}", file=sys.stderr)
|
||||
failed_builds.append(py_version)
|
||||
continue
|
||||
|
||||
successful_wheels.append(wheel)
|
||||
|
||||
# Summary
|
||||
print(f"\n{'=' * 70}")
|
||||
print("BUILD SUMMARY")
|
||||
print(f"{'=' * 70}")
|
||||
print(
|
||||
f"Successful builds: sdist and {len(successful_wheels)}/{len(PYTHON_VERSIONS)} wheels"
|
||||
)
|
||||
|
||||
if failed_builds:
|
||||
print(f"\nFailed builds: {len(failed_builds)}")
|
||||
for version in failed_builds:
|
||||
print(f" ✗ Python {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__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user