Ir al contenido

AegisCipher

AegisCipher es la API recomendada para la mayoría de los usuarios. Maneja todo el flujo KEM-DEM híbrido — encapsulación ML-KEM seguida de cifrado AES-256-GCM — detrás de una interfaz simple y ergonómica.

Por defecto: SecurityLevel.ML_KEM_768. Use el mismo conjunto de parámetros para cifrar y descifrar.

Instale el paquete con Python 3.11 o posterior:

Ventana de terminal
python -m pip install aegisq-pqc
from aegisq import AegisCipher
cipher = AegisCipher() # ML-KEM-768 by default
keypair = cipher.generate_keypair()
package = cipher.encrypt(b"Hello, AegisQ", keypair.public_key)
plaintext = cipher.decrypt(package, keypair.secret_key)
assert plaintext == b"Hello, AegisQ"

Obtenga las claves públicas mediante un canal confiable. AES-GCM autentica el payload, no la identidad del emisor. AegisQ es software beta; usar algoritmos estandarizados no demuestra certificación del producto ni una auditoría independiente. Consulte el modelo de seguridad antes de usarlo con datos sensibles.

La clase también proporciona tres interfaces adicionales:

  • Context Manager (__enter__ / __exit__) — limpieza de buffers mutables registrados internamente (v1.4.0), no claves del llamador.
  • Cifrado en Streaming (encrypt_stream / decrypt_stream) — procesamiento de archivos grandes en bloques acotados (v1.5.0).
  • Métodos Asíncronos (encrypt_async / decrypt_async) — wrappers de thread pool para asyncio.
class AegisCipher:
# API one-shot principal
def __init__(self, level: SecurityLevel = SecurityLevel.ML_KEM_768) -> None
def generate_keypair(self) -> KeyPair
def encrypt(self, plaintext: bytes, recipient_public_key: bytes) -> bytes
def decrypt(self, encrypted_package: bytes, secret_key: bytes) -> bytes
# Streaming (v1.5.0)
def encrypt_stream(
self,
recipient_public_key: bytes,
plaintext_chunks: Iterable[bytes],
chunk_size: int = 65536,
) -> Iterator[bytes]
def decrypt_stream(
self,
secret_key: bytes,
ciphertext_chunks: Iterable[bytes],
) -> Iterator[bytes]
# Asíncrono
async def encrypt_async(self, plaintext: bytes, recipient_public_key: bytes) -> bytes
async def decrypt_async(self, encrypted_package: bytes, secret_key: bytes) -> bytes
# Context manager (v1.4.0)
def __enter__(self) -> Self
def __exit__(self, exc_type, exc_val, exc_tb) -> bool
# Propiedad
@property
def level(self) -> SecurityLevel
AegisCipher(level: SecurityLevel = SecurityLevel.ML_KEM_768)

Crea una nueva instancia de cipher con el nivel de seguridad especificado.

ParámetroTipoDefaultDescripción
levelSecurityLevelML_KEM_768El nivel de seguridad ML-KEM a usar

Genera un nuevo keypair ML-KEM para el nivel de seguridad configurado.

Retorna: KeyPair — Un objeto con los atributos public_key (bytes) y secret_key (bytes).

keypair = cipher.generate_keypair()
# keypair.public_key → compartir abiertamente
# keypair.secret_key → mantener privado; eliminar bytes inmutables no garantiza borrado
# keypair.level → el SecurityLevel usado

Cifra plaintext usando la clave pública del receptor. Internamente realiza la encapsulación ML-KEM para derivar un shared secret, luego cifra el plaintext con AES-256-GCM.

ParámetroTipoDescripción
plaintextbytesLos datos a cifrar
recipient_public_keybytesLa clave pública ML-KEM del receptor

Retorna: bytes — El Transit Package cifrado: [Capsule | Nonce (12 B) | Auth Tag (16 B) | Ciphertext]

Lanza:

  • InvalidParameterError — Si el tamaño de la clave pública no coincide con el nivel de seguridad
  • RngError — Si el CSPRNG del OS no está disponible

Descifra un Transit Package cifrado usando la clave secreta del receptor. Internamente realiza la desencapsulación ML-KEM para recuperar el shared secret, luego descifra y verifica el ciphertext con AES-256-GCM.

ParámetroTipoDescripción
encrypted_packagebytesEl Transit Package cifrado proveniente de encrypt()
secret_keybytesLa clave secreta ML-KEM del receptor

Retorna: bytes — El plaintext original

Lanza:

  • DecryptionError — Si la verificación del Auth Tag de AES-GCM falla (payload manipulado o clave incorrecta)
  • InvalidParameterError — Si los tamaños del paquete o de la clave son incorrectos

El objeto KeyPair retornado por generate_keypair(). Para documentación completa ver referencia KeyPair.

class KeyPair:
public_key: bytes # Clave de cifrado (compartir abiertamente)
secret_key: bytes # Clave de desencapsulación (mantener privada)
level: SecurityLevel

El repr omite bytes crudos y longitudes explícitas. El nivel sigue identificando los tamaños de los parámetros. Ejemplo para ML-KEM-768:

KeyPair(level=MlKem768, fp=<16-hex>)

<16-hex> son los primeros 8 bytes de SHA3-256(public_key): un identificador público truncado estable, no autenticación de identidad. Permite correlacionar el uso de una clave entre logs.

Para transporte por archivo/red, KeyPair expone:

MétodoRetornaFormato
public_key_b64()strBase64 URL-safe (RFC 4648 §5), sin padding
public_key_pem()strPEM-like con -----BEGIN ML-KEM PUBLIC KEY-----
public_key_json()strJSON auto-descriptivo con algorithm, level, public_key
export_secret_key_raw(password)bytesBlob opaco cifrado con AES-256-GCM (HKDF-SHA3-256)
export_secret_key_pem(password)strPEM-like -----BEGIN ENCRYPTED ML-KEM PRIVATE KEY-----

Para helpers de persistencia basada en archivos (save_* / load_*), ver Serialización de Claves.

Para payloads que no caben en memoria (videos, backups, JSON grandes), usá la API de streaming. Ambos métodos están basados en generadores — pasás un iterable de chunks de plaintext y obtenés un iterable de chunks de ciphertext.

CHUNK = 65_536
with open("video.mp4", "rb") as src, open("video.aegisq", "wb") as out:
plaintext_iter = iter(lambda: src.read(CHUNK), b"")
for ct_chunk in cipher.encrypt_stream(keypair.public_key, plaintext_iter):
out.write(ct_chunk)

El Transit Package en modo stream se autodelimita:

[ HEADER: capsule | base_nonce (12 B) | chunk_size (4 B BE) ]
[ FRAME 0: len (4 B BE) | ciphertext | tag (16 B) ]
[ FRAME 1: ... ]
[ EOF MARKER: len=0 | tag (16 B sobre plaintext vacío) ]

El nonce AES-GCM de cada chunk se deriva de su índice (i.to_be_bytes() || base_nonce[4..12]) y su AAD es el índice de chunk de 4 bytes en big-endian — previniendo ataques de reordenamiento de chunks.

Documentación completa: Cifrado en Streaming.

encrypt_async y decrypt_async delegan sus equivalentes síncronos. generate_keypair() sigue siendo síncrono y puede bloquear el event loop si se llama directamente desde una coroutine.

Variantes no bloqueantes de encrypt() / decrypt() para código asyncio. Ejecutan la implementación sincrónica en el ThreadPoolExecutor por defecto, así el event loop queda responsive incluso con payloads grandes.

import asyncio
from aegisq import AegisCipher
async def main():
cipher = AegisCipher()
keypair = await asyncio.to_thread(cipher.generate_keypair)
package = await cipher.encrypt_async(b"secreto", keypair.public_key)
plaintext = await cipher.decrypt_async(package, keypair.secret_key)
print(plaintext) # b"secreto"
asyncio.run(main())

Documentación completa: Métodos Asíncronos.

AegisCipher puede usarse dentro de un bloque with. Al salir, sobrescribe buffers mutables registrados con su hook privado. Los métodos públicos actuales no registran buffers allí. No borra claves/plaintext del llamador ni cierra un stream suspendido; tampoco impide nuevas llamadas al cipher.

from aegisq import AegisCipher
with AegisCipher() as cipher:
keypair = cipher.generate_keypair()
package = cipher.encrypt(b"hola", keypair.public_key)
# __exit__ zeroiza cualquier buffer registrado; las excepciones se propagan igual.

__repr__ refleja el estado de la sesión:

>>> repr(cipher)
'AegisCipher(level=SecurityLevel.ML_KEM_768, inactive)'
>>> with cipher:
... repr(cipher)
...
'AegisCipher(level=SecurityLevel.ML_KEM_768, active)'

Documentación completa: Context Manager.

from aegisq import AegisCipher, SecurityLevel
# 1. Bob (receptor) genera un keypair — la clave pública se comparte abiertamente
cipher_bob = AegisCipher(level=SecurityLevel.ML_KEM_768)
keypair = cipher_bob.generate_keypair()
public_key: bytes = keypair.public_key # 1184 bytes — compartir con cualquiera
secret_key: bytes = keypair.secret_key # 2400 bytes — no compartir ni registrar
# 2. Alice (emisor) cifra usando la clave pública de Bob
cipher_alice = AegisCipher(level=SecurityLevel.ML_KEM_768)
payload = "Registros médicos ultra secretos".encode("utf-8")
encrypted_package: bytes = cipher_alice.encrypt(
plaintext=payload,
recipient_public_key=public_key,
)
# encrypted_package = [ ML-KEM Capsule (1088 B) | Nonce (12 B) | Tag (16 B) | Ciphertext ]
# Esto es lo ÚNICO que Alice envía a Bob por la red.
# 3. Bob descifra el paquete
decrypted_payload: bytes = cipher_bob.decrypt(
encrypted_package=encrypted_package,
secret_key=secret_key,
)
assert decrypted_payload == payload # ✓