cryptography 41.0.7: Cipher

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: verify Cipher in pkg:pypi/cryptography@41.0.7
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/cryptography@41.0.7
Demonstrate these symbols/APIs:
  - Cipher

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:83ca343a270ce55e02b60ec3c5c35ef879dc455d05117fdb49c04ca952794c15","contract":["Cipher encryptor and decryptor perform symmetric block encryption and decryption with AES and CBC mode","Cipher with AES and GCM mode supports AEAD encryption with authentication tag and additional authenticated data","Cipher with AES-GCM decryptor raises InvalidTag when authentication tag is tampered or corrupted","Cipher encryptor and decryptor perform stream encryption and decryption with AES and CTR mode","Cipher initialization with AES algorithm validates key length and raises ValueError for invalid key sizes"],"goal":"verify Cipher in pkg:pypi/cryptography@41.0.7","kind":"HOW","packages":["pkg:pypi/cryptography@41.0.7"],"schemaVersion":1,"symbols":["Cipher"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/cryptography@41.0.7"],"schemaVersion":1,"subject":"pkg:pypi/cryptography@41.0.7","symbols":["Cipher"],"verifierAdapter":"python@1"}

[project]
name = "verify-cryptography-cipher"
version = "0.1.0"
dependencies = [
    "cryptography==41.0.7",
]

cryptography==41.0.7

"""Demonstration of symmetric encryption and decryption using cryptography Cipher."""

from typing import Tuple
from cryptography.hazmat.primitives import padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes


def aes_cbc_encrypt(key: bytes, iv: bytes, plaintext: bytes) -> bytes:
    """Encrypt plaintext using AES in CBC mode with PKCS7 padding."""
    padder = padding.PKCS7(algorithms.AES.block_size).padder()
    padded_data = padder.update(plaintext) + padder.finalize()

    cipher = Cipher(algorithms.AES(key), modes.CBC(iv))
    encryptor = cipher.encryptor()
    return encryptor.update(padded_data) + encryptor.finalize()


def aes_cbc_decrypt(key: bytes, iv: bytes, ciphertext: bytes) -> bytes:
    """Decrypt ciphertext using AES in CBC mode and remove PKCS7 padding."""
    cipher = Cipher(algorithms.AES(key), modes.CBC(iv))
    decryptor = cipher.decryptor()
    padded_data = decryptor.update(ciphertext) + decryptor.finalize()

    unpadder = padding.PKCS7(algorithms.AES.block_size).unpadder()
    return unpadder.update(padded_data) + unpadder.finalize()


def aes_gcm_encrypt(
    key: bytes, nonce: bytes, plaintext: bytes, associated_data: bytes = b""
) -> Tuple[bytes, bytes]:
    """Encrypt plaintext using AES in GCM mode and return (ciphertext, tag)."""
    cipher = Cipher(algorithms.AES(key), modes.GCM(nonce))
    encryptor = cipher.encryptor()
    if associated_data:
        encryptor.authenticate_additional_data(associated_data)
    ciphertext = encryptor.update(plaintext) + encryptor.finalize()
    return ciphertext, encryptor.tag


def aes_gcm_decrypt(
    key: bytes,
    nonce: bytes,
    tag: bytes,
    ciphertext: bytes,
    associated_data: bytes = b"",
) -> bytes:
    """Decrypt ciphertext using AES in GCM mode verifying the authentication tag."""
    cipher = Cipher(algorithms.AES(key), modes.GCM(nonce, tag))
    decryptor = cipher.decryptor()
    if associated_data:
        decryptor.authenticate_additional_data(associated_data)
    return decryptor.update(ciphertext) + decryptor.finalize()


def aes_ctr_crypt(key: bytes, nonce: bytes, data: bytes) -> bytes:
    """Encrypt or decrypt data using AES in CTR mode (symmetric stream cipher)."""
    cipher = Cipher(algorithms.AES(key), modes.CTR(nonce))
    cryptor = cipher.encryptor()
    return cryptor.update(data) + cryptor.finalize()

{
  "schemaVersion": 1,
  "goal": "verify Cipher in pkg:pypi/cryptography@41.0.7",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/cryptography@41.0.7"
  ],
  "symbols": [
    "Cipher"
  ]
}

"""Contract verification test for cryptography Cipher."""

import os
import sys

# Ensure project root is on sys.path for local module imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from sample import (
    aes_cbc_encrypt,
    aes_cbc_decrypt,
    aes_gcm_encrypt,
    aes_gcm_decrypt,
    aes_ctr_crypt,
)


def test_aes_cbc_roundtrip() -> None:
    """Assertion 1: Cipher encryptor and decryptor perform symmetric block encryption and decryption with AES and CBC mode."""
    key = b"\x01" * 32  # 256-bit key
    iv = b"\x02" * 16   # 128-bit IV
    plaintext = b"Test message for AES-CBC contract verification."

    # Direct Cipher API usage
    cipher_enc = Cipher(algorithms.AES(key), modes.CBC(iv))
    encryptor = cipher_enc.encryptor()
    padded_block = plaintext + b"\x00" * (16 - (len(plaintext) % 16))
    ct = encryptor.update(padded_block) + encryptor.finalize()
    assert len(ct) == len(padded_block)
    assert ct != padded_block

    cipher_dec = Cipher(algorithms.AES(key), modes.CBC(iv))
    decryptor = cipher_dec.decryptor()
    pt_recovered = decryptor.update(ct) + decryptor.finalize()
    assert pt_recovered == padded_block

    # Helper function roundtrip test
    ciphertext = aes_cbc_encrypt(key, iv, plaintext)
    decrypted = aes_cbc_decrypt(key, iv, ciphertext)
    assert decrypted == plaintext


def test_aes_gcm_aead() -> None:
    """Assertion 2: Cipher with AES and GCM mode supports AEAD encryption with authentication tag and additional authenticated data."""
    key = b"\x10" * 32
    nonce = b"\x20" * 12
    aad = b"sample-authenticated-metadata"
    plaintext = b"Sensitive payload needing AEAD protection."

    # Direct Cipher API usage
    cipher_enc = Cipher(algorithms.AES(key), modes.GCM(nonce))
    encryptor = cipher_enc.encryptor()
    encryptor.authenticate_additional_data(aad)
    ciphertext = encryptor.update(plaintext) + encryptor.finalize()
    tag = encryptor.tag
    assert len(tag) == 16
    assert ciphertext != plaintext

    cipher_dec = Cipher(algorithms.AES(key), modes.GCM(nonce, tag))
    decryptor = cipher_dec.decryptor()
    decryptor.authenticate_additional_data(aad)
    recovered = decryptor.update(ciphertext) + decryptor.finalize()
    assert recovered == plaintext

    # Helper function roundtrip test
    ct, helper_tag = aes_gcm_encrypt(key, nonce, plaintext, associated_data=aad)
    assert helper_tag == tag
    decrypted = aes_gcm_decrypt(key, nonce, helper_tag, ct, associated_data=aad)
    assert decrypted == plaintext


def test_aes_gcm_tag_validation() -> None:
    """Assertion 3: Cipher with AES-GCM decryptor raises InvalidTag when authentication tag is tampered or corrupted."""
    key = b"\x30" * 32
    nonce = b"\x40" * 12
    aad = b"auth-data"
    plaintext = b"Payload for tag tampering test."

    ciphertext, tag = aes_gcm_encrypt(key, nonce, plaintext, associated_data=aad)

    # Tamper with tag
    corrupted_tag = bytes([tag[0] ^ 0xFF]) + tag[1:]
    raised = False
    try:
        aes_gcm_decrypt(key, nonce, corrupted_tag, ciphertext, associated_data=aad)
    except InvalidTag:
        raised = True
    assert raised, "Expected InvalidTag on corrupted authentication tag"

    # Tamper with associated data
    raised_aad = False
    try:
        aes_gcm_decrypt(key, nonce, tag, ciphertext, associated_data=b"wrong-metadata")
    except InvalidTag:
        raised_aad = True
    assert raised_aad, "Expected InvalidTag on mismatched associated data"


def test_aes_ctr_stream() -> None:
    """Assertion 4: Cipher encryptor and decryptor perform stream encryption and decryption with AES and CTR mode."""
    key = b"\x50" * 32
    nonce = b"\x60" * 16
    plaintext = b"Arbitrary length stream data without padding requirement."

    # Helper function stream roundtrip
    ciphertext = aes_ctr_crypt(key, nonce, plaintext)
    assert ciphertext != plaintext
    assert len(ciphertext) == len(plaintext)

    decrypted = aes_ctr_crypt(key, nonce, ciphertext)
    assert decrypted == plaintext

    # Direct Cipher API check
    cipher = Cipher(algorithms.AES(key), modes.CTR(nonce))
    direct_ct = cipher.encryptor().update(plaintext) + cipher.encryptor().finalize()
    direct_pt = cipher.decryptor().update(direct_ct) + cipher.decryptor().finalize()
    assert direct_pt == plaintext


def test_invalid_key_validation() -> None:
    """Assertion 5: Cipher initialization with AES algorithm validates key length and raises ValueError for invalid key sizes."""
    invalid_short_key = b"short_key"
    raised = False
    try:
        Cipher(algorithms.AES(invalid_short_key), modes.ECB())
    except ValueError:
        raised = True
    assert raised, "Expected ValueError for invalid AES key length"


def main() -> None:
    test_aes_cbc_roundtrip()
    test_aes_gcm_aead()
    test_aes_gcm_tag_validation()
    test_aes_ctr_stream()
    test_invalid_key_validation()
    print("All contract assertions passed.")


if __name__ == "__main__":
    main()