14 Commits
27 changed files with 665 additions and 207 deletions
+3 -3
View File
@@ -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
+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
@@ -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
View File
@@ -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
+42 -30
View File
@@ -1,33 +1,41 @@
# 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
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)
@@ -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)
@@ -182,7 +190,7 @@ It is often practical to split larger messages into frames that can be individua
```python
# Encryption settings
from pyaegis import aegis128x4 as ciph
from aeg import aegis128x4 as ciph
key = b"sixteenbyte key!" # 16 bytes secret key for aegis128* algorithms
framebytes = 80 # In real applications 1 MiB or more is practical
maclen = ciph.MACBYTES # 16
@@ -202,7 +210,7 @@ with open("encrypted.bin", "wb") as f:
```python
# Decryption needs same values as encryption
from pyaegis import aegis128x4 as ciph
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
@@ -247,7 +255,7 @@ 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
@@ -258,7 +266,7 @@ 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,10 +284,10 @@ 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.
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 AMD hardware.
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
```sh
uv run -m aeg.benchmark
AEGIS-256 103166.24 Mb/s
AEGIS-256X2 184225.50 Mb/s
AEGIS-256X4 194018.26 Mb/s
@@ -295,8 +303,8 @@ 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
@@ -310,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.
View File
+15 -9
View File
@@ -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.1"
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/LeoVasanko/pyaegis"
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]
+16 -30
View File
@@ -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__":
+16
View File
@@ -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)}")
+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."""
from pyaegis._aegis import ffi, lib
from aeg._aegis import ffi, lib
__all__ = ["ffi", "lib"]
+122
View File
@@ -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:
+7 -5
View File
@@ -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:
@@ -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 -1
View File
@@ -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.
-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
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
View File
@@ -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
View File
@@ -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]
+1 -1
View File
@@ -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
View File
@@ -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
+346
View File
@@ -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())