noiseframework 1.3.0
A professional, secure, and easy-to-use implementation of the Noise Protocol Framework in Python
Requires Python: >=3.8
Keywords: noise, noise-protocol, cryptography, encryption, handshake, security, transport-security, chacha20-poly1305, curve25519

# NoiseFramework

[![Official Website](https://img.shields.io/badge/🌐%20Official%20Website-www.noiseframework.com-4A90E2?style=for-the-badge&logo=globe&logoColor=white)](https://www.noiseframework.com)
<br>

<br>

[![PyPI version](https://img.shields.io/pypi/v/noiseframework.svg)](https://pypi.org/project/noiseframework/)
[![Python Version](https://img.shields.io/badge/python-3.8%2B-blue.svg)](https://www.python.org/downloads/)
[![License](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE)
[![GitHub Issues](https://img.shields.io/github/issues/juliuspleunes4/noiseframework)](https://github.com/juliuspleunes4/noiseframework/issues)
[![GitHub Stars](https://img.shields.io/github/stars/juliuspleunes4/noiseframework)](https://github.com/juliuspleunes4/noiseframework/stargazers)
[![Code Style](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black)

> A professional, secure, and easy-to-use implementation of the [Noise Protocol Framework](https://noiseprotocol.org/) in Python.

**NoiseFramework** provides cryptographically sound, specification-compliant implementations of Noise handshake patterns for building secure communication channels. It is designed to be both simple to integrate into applications and robust enough for production use.

---

## 📋 Table of Contents

- [Features](#-features)
- [Installation](#-installation)
- [Quick Start](#-quick-start)
  - [Python API](#python-api)
  - [Command-Line Interface](#command-line-interface)
- [Python API Documentation](#-python-api-documentation)
  - [Basic Handshake (Noise_XX)](#basic-handshake-noise_xx)
  - [Anonymous Pattern (Noise_NN)](#anonymous-pattern-noise_nn)
  - [Pre-Shared Key Pattern (Noise_IK)](#pre-shared-key-pattern-noise_ik)
  - [Transport Layer Encryption](#transport-layer-encryption)
  - [Error Handling](#error-handling)
  - [Logging](#logging)
  - [Message Framing](#message-framing)
  - [Async/Await Support](#asyncawait-support)
  - [Pre-Shared Key (PSK) Support](#pre-shared-key-psk-support)
  - [Fallback Pattern Support](#fallback-pattern-support)
  - [High-Level Connection API](#high-level-connection-api)
- [CLI Documentation](#-cli-documentation)
  - [Generate Keypair](#generate-keypair)
  - [Validate Pattern](#validate-pattern)
  - [Show Information](#show-information)
- [Supported Patterns](#-supported-patterns)
- [Cryptographic Primitives](#-cryptographic-primitives)
- [Architecture](#-architecture)
- [Testing](#-testing)
- [Performance](#-performance)
- [Contributing](#-contributing)
- [Security](#-security)
- [FAQ](#-faq)
- [License](#-license)
- [Acknowledgments](#-acknowledgments)

---

## ✨ Features

- **📜 Spec-Compliant**: Implements the [Noise Protocol Framework specification](https://noiseprotocol.org/noise.html) faithfully
- **🔒 Secure by Default**: Uses well-vetted cryptographic primitives from trusted libraries
- **🐍 Pythonic API**: Simple, type-hinted interfaces that are easy to use and hard to misuse
- **🛠️ CLI Tool**: Command-line interface for encryption, decryption, and handshake operations
- **✅ Well-Tested**: Comprehensive test suite with 311 tests achieving 100% pass rate
- **📦 Zero Config**: Works out-of-the-box with sensible defaults
- **🔧 Flexible**: Supports multiple DH functions, cipher suites, and hash functions
- **🛡️ PSK Support**: Pre-Shared Key patterns for quantum-resistant authentication (psk0-psk4)
- **📖 Documented**: Extensive documentation with examples and best practices
- **🔍 Helpful Errors**: Custom exceptions with actionable error messages guide you to fix issues quickly
- **📝 Built-in Logging**: Comprehensive logging support for debugging and monitoring
- **📦 Message Framing**: Automatic length-prefixed framing for stream-based transports
- **⚡ Async/Await**: Full async support for modern Python asyncio applications
- **🔗 High-Level Connection API**: `NoiseConnection` and `AsyncNoiseConnection` for easy-to-use connection management

---

## 📦 Installation

### From PyPI (Recommended)

```bash
pip install noiseframework
```

### From Source

```bash
git clone https://github.com/juliuspleunes4/noiseframework.git
cd noiseframework
pip install -e .
```

### Requirements

- Python 3.8 or higher
- Dependencies are automatically installed via pip

---

## 🚀 Quick Start

### Python API

```python
from noiseframework import NoiseHandshake, NoiseTransport

# === INITIATOR SIDE ===
initiator = NoiseHandshake("Noise_XX_25519_ChaChaPoly_SHA256")
initiator.set_as_initiator()
initiator.generate_static_keypair()
initiator.initialize()

# === RESPONDER SIDE ===
responder = NoiseHandshake("Noise_XX_25519_ChaChaPoly_SHA256")
responder.set_as_responder()
responder.generate_static_keypair()
responder.initialize()

# === HANDSHAKE ===
msg1 = initiator.write_message(b"")
responder.read_message(msg1)

msg2 = responder.write_message(b"")
initiator.read_message(msg2)

msg3 = initiator.write_message(b"")
responder.read_message(msg3)

# === TRANSPORT ENCRYPTION ===
init_send, init_recv = initiator.to_transport()
resp_send, resp_recv = responder.to_transport()

init_transport = NoiseTransport(init_send, init_recv)
resp_transport = NoiseTransport(resp_send, resp_recv)

# Send encrypted messages
ciphertext = init_transport.send(b"Hello, secure world!")
plaintext = resp_transport.receive(ciphertext)
print(plaintext)  # b"Hello, secure world!"
```

### Command-Line Interface

```bash
# Generate a keypair
noiseframework generate-keypair --dh 25519 -o mykey
# Creates: mykey_private.key, mykey_public.key

# Validate a pattern string
noiseframework validate-pattern "Noise_XX_25519_ChaChaPoly_SHA256"

# Show supported primitives
noiseframework info

# Use shorter aliases
noiseframework genkey --dh 25519 -o mykey
noiseframework validate "Noise_XX_25519_ChaChaPoly_SHA256"
```

---

## 📖 Python API Documentation

### Basic Handshake (Noise_XX)

The `XX` pattern provides mutual authentication with no prior knowledge required. Both parties exchange static keys during the handshake.

```python
from noiseframework import NoiseHandshake, NoiseTransport

# === INITIATOR SIDE ===
initiator = NoiseHandshake("Noise_XX_25519_ChaChaPoly_SHA256")
initiator.set_as_initiator()
initiator.generate_static_keypair()  # Generate static key
initiator.initialize()

# Send first message (-> e)
msg1 = initiator.write_message(b"")

# === RESPONDER SIDE ===
responder = NoiseHandshake("Noise_XX_25519_ChaChaPoly_SHA256")
responder.set_as_responder()
responder.generate_static_keypair()  # Generate static key
responder.initialize()

# Process first message and send response (-> e, ee, s, es)
responder.read_message(msg1)
msg2 = responder.write_message(b"")

# === INITIATOR SIDE (continued) ===
# Process second message and send final (-> s, se)
initiator.read_message(msg2)
msg3 = initiator.write_message(b"")

# === RESPONDER SIDE (continued) ===
# Process final message
responder.read_message(msg3)

# === BOTH SIDES NOW HAVE SECURE CHANNEL ===
# Get transport cipher pairs
init_send, init_recv = initiator.to_transport()
resp_send, resp_recv = responder.to_transport()

# Create transport wrappers
init_transport = NoiseTransport(init_send, init_recv)
resp_transport = NoiseTransport(resp_send, resp_recv)

# Send encrypted data (initiator -> responder)
ciphertext = init_transport.send(b"Secret payload")
plaintext = resp_transport.receive(ciphertext)
assert plaintext == b"Secret payload"

# Send encrypted data (responder -> initiator)
ciphertext = resp_transport.send(b"Response data")
plaintext = init_transport.receive(ciphertext)
assert plaintext == b"Response data"
```

### Anonymous Pattern (Noise_NN)

The `NN` pattern provides encryption without authentication. No static keys are required.

```python
from noiseframework import NoiseHandshake, NoiseTransport

# === INITIATOR SIDE ===
initiator = NoiseHandshake("Noise_NN_25519_ChaChaPoly_SHA256")
initiator.set_as_initiator()
initiator.initialize()

# Send first message (-> e)
msg1 = initiator.write_message(b"")

# === RESPONDER SIDE ===
responder = NoiseHandshake("Noise_NN_25519_ChaChaPoly_SHA256")
responder.set_as_responder()
responder.initialize()

# Process first message and send response (-> e, ee)
responder.read_message(msg1)
msg2 = responder.write_message(b"")

# === INITIATOR SIDE (continued) ===
# Process second message - handshake complete
initiator.read_message(msg2)

# === CREATE TRANSPORT ===
init_send, init_recv = initiator.to_transport()
resp_send, resp_recv = responder.to_transport()

init_transport = NoiseTransport(init_send, init_recv)
resp_transport = NoiseTransport(resp_send, resp_recv)

# Now both sides can communicate securely (but without authentication)
ciphertext = init_transport.send(b"Anonymous message")
plaintext = resp_transport.receive(ciphertext)
```

### Pre-Shared Key Pattern (Noise_IK)

The `IK` pattern allows the initiator to know the responder's static public key in advance. The initiator's identity is hidden.

```python
from noiseframework import NoiseHandshake, NoiseTransport

# === SETUP: Generate responder's static keypair ===
responder_setup = NoiseHandshake("Noise_IK_25519_ChaChaPoly_SHA256")
responder_setup.set_as_responder()
responder_setup.generate_static_keypair()
responder_private = responder_setup.static_private
responder_public = responder_setup.static_public

# === INITIATOR SIDE ===
initiator = NoiseHandshake("Noise_IK_25519_ChaChaPoly_SHA256")
initiator.set_as_initiator()
initiator.generate_static_keypair()  # Generate own static key
initiator.set_remote_static_public_key(responder_public)  # Know responder's key
initiator.initialize()

# Send first message (-> e, es, s, ss)
msg1 = initiator.write_message(b"")

# === RESPONDER SIDE ===
responder = NoiseHandshake("Noise_IK_25519_ChaChaPoly_SHA256")
responder.set_as_responder()
responder.set_static_keypair(responder_private, responder_public)  # Use existing keypair
responder.initialize()

# Process first message and send response (-> e, ee, se)
responder.read_message(msg1)
msg2 = responder.write_message(b"")

# === INITIATOR SIDE (continued) ===
# Process second message - handshake complete
initiator.read_message(msg2)

# === CREATE TRANSPORT ===
init_send, init_recv = initiator.to_transport()
resp_send, resp_recv = responder.to_transport()

init_transport = NoiseTransport(init_send, init_recv)
resp_transport = NoiseTransport(resp_send, resp_recv)

# Secure authenticated communication
ciphertext = init_transport.send(b"Authenticated message")
plaintext = resp_transport.receive(ciphertext)
```

### Transport Layer Encryption

After handshake completion, use the transport layer for ongoing encrypted communication:

```python
from noiseframework import NoiseTransport

# After successful handshake, get cipher states
send_cipher, recv_cipher = handshake.to_transport()

# Create transport wrapper
transport = NoiseTransport(send_cipher, recv_cipher)

# Encrypt and send data
ciphertext = transport.send(b"Sensitive data")

# Decrypt received data
plaintext = transport.receive(ciphertext)

# Send with associated data (authenticated but not encrypted)
ciphertext = transport.send(b"payload", ad=b"metadata")
plaintext = transport.receive(ciphertext, ad=b"metadata")

# Track nonces
print(f"Messages sent: {transport.get_send_nonce()}")
print(f"Messages received: {transport.get_receive_nonce()}")

# Transport automatically handles:
# - Nonce increment
# - Authentication tags
# - AEAD encryption/decryption
```

### Error Handling

NoiseFramework provides helpful custom exceptions with actionable error messages:

```python
from noiseframework import NoiseHandshake, NoiseTransport
from noiseframework.exceptions import (
    NoiseError,                  # Base class - catches all framework errors
    UnsupportedPatternError,     # Invalid pattern
    RoleNotSetError,             # Role not set
    AuthenticationError,         # Decryption/authentication failure
)

# Catch specific exceptions
try:
    hs = NoiseHandshake("Invalid_Pattern")
except UnsupportedPatternError as e:
    print(f"Pattern error: {e}")
    # Output: Invalid pattern string format: 'Invalid_Pattern'.
    #         Expected format: Noise_PATTERN_DH_CIPHER_HASH

# Handle state errors
try:
    hs = NoiseHandshake("Noise_XX_25519_ChaChaPoly_SHA256")
    hs.write_message()  # Error: role not set
except RoleNotSetError as e:
    print(f"State error: {e}")
    # Output: Cannot write handshake message: role not set.
    #         Call set_as_initiator() or set_as_responder() first.

# Handle authentication failures
try:
    ciphertext_tampered = ciphertext[:-1] + b"\x00"
    transport.receive(ciphertext_tampered)
except AuthenticationError as e:
    print(f"Authentication failed: {e}")
    # DO NOT process the message - discard it

# Catch all framework errors
try:
    # ... noise operations ...
except NoiseError as e:
    print(f"Framework error: {type(e).__name__}: {e}")
```

**Available exceptions**: `NoiseError` (base), `HandshakeError`, `RoleNotSetError`, `RoleAlreadySetError`, `WrongTurnError`, `HandshakeCompleteError`, `MissingKeyError`, `PatternError`, `UnsupportedPatternError`, `UnsupportedPrimitiveError`, `StateError`, `NoKeySetError`, `NonceOverflowError`, `InvalidKeySizeError`, `TransportError`, `AuthenticationError`, `CryptoError`, `ValidationError`, `FramingError`.

See `examples/error_handling_example.py` for comprehensive error handling examples.

---

### Logging

NoiseFramework includes comprehensive logging support for debugging and monitoring. All major operations are logged at appropriate levels.

#### Basic Logging Setup

```python
import logging
from noiseframework import NoiseHandshake, NoiseTransport

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)-8s] %(name)s: %(message)s"
)

# Use NoiseFramework normally - logging happens automatically
handshake = NoiseHandshake("Noise_XX_25519_ChaChaPoly_SHA256")
handshake.set_as_initiator()  # Logs: "Role set as INITIATOR"
handshake.generate_static_keypair()  # Logs: "Generated static keypair"
handshake.initialize()  # Logs: "Handshake initialized"

msg = handshake.write_message(b"")  # Logs: "Sent handshake message 1"
```

#### Log Levels

- **DEBUG**: Detailed operations (message sizes, nonces, token processing, key operations)
- **INFO**: Major events (role changes, handshake completion, message send/receive)
- **WARNING**: Potential issues (nonce approaching limit)
- **ERROR**: Failures (validation errors, authentication failures)

#### Custom Logger

```python
# Create custom logger with specific configuration
custom_logger = logging.getLogger("myapp.noise")
custom_logger.setLevel(logging.DEBUG)

# Add custom handler
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
custom_logger.addHandler(handler)

# Pass custom logger to NoiseHandshake
handshake = NoiseHandshake(
    "Noise_XX_25519_ChaChaPoly_SHA256",
    logger=custom_logger
)

# Pass custom logger to NoiseTransport
transport = NoiseTransport(send_cipher, recv_cipher, logger=custom_logger)
```

#### Filtering Logs by Module

```python
# Only show INFO+ logs from handshake module
logging.getLogger("noiseframework.noise.handshake").setLevel(logging.INFO)

# Show DEBUG logs from transport module
logging.getLogger("noiseframework.transport.transport").setLevel(logging.DEBUG)

# Disable logs from state module
logging.getLogger("noiseframework.noise.state").setLevel(logging.WARNING)
```

#### Example Log Output

```
2025-11-24 15:30:01 [INFO    ] noiseframework.noise.handshake.NoiseHandshake: Role set as INITIATOR
2025-11-24 15:30:01 [DEBUG   ] noiseframework.noise.handshake.NoiseHandshake: Generating static keypair (Curve25519)
2025-11-24 15:30:01 [INFO    ] noiseframework.noise.handshake.NoiseHandshake: Generated static keypair
2025-11-24 15:30:01 [INFO    ] noiseframework.noise.handshake.NoiseHandshake: Handshake initialized
2025-11-24 15:30:01 [DEBUG   ] noiseframework.noise.handshake.NoiseHandshake: Writing handshake message 1 (payload=0 bytes)
2025-11-24 15:30:01 [INFO    ] noiseframework.noise.handshake.NoiseHandshake: Sent handshake message 1 (ciphertext=32 bytes)
2025-11-24 15:30:02 [INFO    ] noiseframework.noise.handshake.NoiseHandshake: Handshake complete - ready for transport mode
2025-11-24 15:30:02 [INFO    ] noiseframework.noise.handshake.NoiseHandshake: Created transport ciphers (initiator: send=c1, receive=c2)
2025-11-24 15:30:03 [INFO    ] noiseframework.transport.transport.NoiseTransport: Sent encrypted message (ciphertext=29 bytes)
2025-11-24 15:30:03 [WARNING ] noiseframework.transport.transport.NoiseTransport: Send cipher nonce high: 9223372036854775808 (approaching 2^64 limit - consider rekeying)
```

See [`examples/logging_example.py`](examples/logging_example.py) for more detailed examples.

### Message Framing

When using Noise over stream-based transports (TCP, pipes, etc.), you need a way to preserve message boundaries. NoiseFramework provides length-prefixed framing utilities for this purpose.

#### Basic Usage

```python
import socket
from noiseframework import NoiseHandshake, FramedWriter, FramedReader

# After handshake completes...
transport = handshake.to_transport()

# Wrap socket for framed communication
writer = FramedWriter(sock.makefile('wb'))
reader = FramedReader(sock.makefile('rb'))

# Send framed encrypted messages
ciphertext = transport.send(b"Hello, World!")
writer.write_message(ciphertext)

# Receive framed encrypted messages
framed_message = reader.read_message()
plaintext = transport.receive(framed_message)
```

#### Frame Format

NoiseFramework uses a simple 4-byte big-endian length prefix:

```
┌──────────────┬────────────────────┐
│ Length (4B)  │  Message Data      │
│ big-endian   │  (0 to 2^32-1 B)   │
└──────────────┴────────────────────┘
```

#### Configuration

```python
# Default maximum message size is 16 MB
reader = FramedReader(stream)  # max_message_size=16*1024*1024

# Set custom maximum
reader = FramedReader(stream, max_message_size=1024*1024)  # 1 MB limit

# Add logging
import logging
logger = logging.getLogger("myapp.framing")
reader = FramedReader(stream, logger=logger)
writer = FramedWriter(stream, logger=logger)
```

#### TCP Example

```python
import socket
from noiseframework import NoiseHandshake, FramedWriter, FramedReader

# Server
def server(port):
    with socket.socket() as sock:
        sock.bind(('localhost', port))
        sock.listen(1)
        conn, _ = sock.accept()
        
        # Noise handshake (responder)
        hs = NoiseHandshake("Noise_XX_25519_ChaChaPoly_SHA256")
        hs.set_as_responder()
        hs.generate_static_keypair()
        hs.initialize()
        
        reader = FramedReader(conn.makefile('rb'))
        writer = FramedWriter(conn.makefile('wb'))
        
        # Handshake messages with framing
        msg1 = reader.read_message()
        msg2 = hs.read_message(msg1)
        msg2_out = hs.write_message(msg2)
        writer.write_message(msg2_out)
        
        msg3 = reader.read_message()
        hs.read_message(msg3)
        
        # Transport mode
        transport = hs.to_transport()
        encrypted = reader.read_message()
        plaintext = transport.receive(encrypted)
        print(f"Server received: {plaintext}")

# Client
def client(port):
    with socket.socket() as sock:
        sock.connect(('localhost', port))
        
        # Noise handshake (initiator)
        hs = NoiseHandshake("Noise_XX_25519_ChaChaPoly_SHA256")
        hs.set_as_initiator()
        hs.generate_static_keypair()
        hs.initialize()
        
        reader = FramedReader(sock.makefile('rb'))
        writer = FramedWriter(sock.makefile('wb'))
        
        # Handshake messages with framing
        msg1 = hs.write_message(b"")
        writer.write_message(msg1)
        
        msg2 = reader.read_message()
        msg3_payload = hs.read_message(msg2)
        msg3 = hs.write_message(msg3_payload)
        writer.write_message(msg3)
        
        # Transport mode
        transport = hs.to_transport()
        ciphertext = transport.send(b"Hello, Server!")
        writer.write_message(ciphertext)
```

#### Convenience Functions

For single-message operations:

```python
from noiseframework import write_framed_message, read_framed_message

# Write single framed message
write_framed_message(stream, b"Hello")

# Read single framed message
message = read_framed_message(stream)

# With custom max size
message = read_framed_message(stream, max_message_size=1024*1024)
```

#### Error Handling

```python
from noiseframework import FramingError

try:
    message = reader.read_message()
except FramingError as e:
    # Connection closed, oversized message, or invalid frame
    print(f"Framing error: {e}")
except IOError as e:
    # Underlying stream error
    print(f"IO error: {e}")
```

**Key Points:**
- Automatic handling of partial reads
- Protection against oversized messages (configurable limit)
- Message counters for debugging (`messages_sent`, `messages_received`)
- Thread-safe for concurrent read/write on different threads
- Works with any byte stream (sockets, files, pipes, etc.)

See [`examples/framed_tcp_example.py`](examples/framed_tcp_example.py) for a complete working example.

### Async/Await Support

NoiseFramework provides full `asyncio` support for modern async Python applications. All async classes wrap the synchronous implementation using `run_in_executor`, making them safe for use in async contexts without blocking the event loop.

#### Basic Async Usage

```python
import asyncio
from noiseframework import AsyncNoiseHandshake, AsyncNoiseTransport

async def async_handshake():
    # Create async handshake
    handshake = AsyncNoiseHandshake("Noise_XX_25519_ChaChaPoly_SHA256")
    await handshake.set_as_initiator()
    await handshake.generate_static_keypair()
    await handshake.initialize()
    
    # Perform handshake (async)
    msg1 = await handshake.write_message(b"")
    # ... exchange messages over network ...
    
    # Convert to async transport
    transport = await handshake.to_transport()
    
    # Send/receive encrypted messages (async)
    ciphertext = await transport.send(b"Hello, async world!")
    plaintext = await transport.receive(ciphertext)

# Run the async function
asyncio.run(async_handshake())
```

#### Async TCP Server Example

```python
import asyncio
from noiseframework import (
    AsyncNoiseHandshake,
    AsyncFramedReader,
    AsyncFramedWriter,
)

async def handle_client(reader, writer):
    """Handle incoming client connection."""
    # Create responder handshake
    handshake = AsyncNoiseHandshake("Noise_XX_25519_ChaChaPoly_SHA256")
    await handshake.set_as_responder()
    await handshake.generate_static_keypair()
    await handshake.initialize()
    
    # Wrap streams with framing
    framed_reader = AsyncFramedReader(reader)
    framed_writer = AsyncFramedWriter(writer)
    
    # Perform XX handshake
    msg1 = await framed_reader.read_message()
    await handshake.read_message(msg1)
    
    msg2 = await handshake.write_message(b"")
    await framed_writer.write_message(msg2)
    
    msg3 = await framed_reader.read_message()
    await handshake.read_message(msg3)
    
    # Switch to transport mode
    transport = await handshake.to_transport()
    
    # Receive and process encrypted messages
    while True:
        try:
            ciphertext = await framed_reader.read_message()
            plaintext = await transport.receive(ciphertext)
            print(f"Received: {plaintext.decode()}")
            
            # Send encrypted response
            response = await transport.send(b"Message received!")
            await framed_writer.write_message(response)
        except:
            break
    
    await framed_writer.close()

async def main():
    server = await asyncio.start_server(
        handle_client, '127.0.0.1', 9999
    )
    async with server:
        await server.serve_forever()

asyncio.run(main())
```

#### Async TCP Client Example

```python
async def async_client():
    # Connect to server
    reader, writer = await asyncio.open_connection('127.0.0.1', 9999)
    
    # Create initiator handshake
    handshake = AsyncNoiseHandshake("Noise_XX_25519_ChaChaPoly_SHA256")
    await handshake.set_as_initiator()
    await handshake.generate_static_keypair()
    await handshake.initialize()
    
    # Wrap streams with framing
    framed_reader = AsyncFramedReader(reader)
    framed_writer = AsyncFramedWriter(writer)
    
    # Perform XX handshake (3 messages)
    msg1 = await handshake.write_message(b"")
    await framed_writer.write_message(msg1)
    
    msg2 = await framed_reader.read_message()
    await handshake.read_message(msg2)
    
    msg3 = await handshake.write_message(b"")
    await framed_writer.write_message(msg3)
    
    # Switch to transport mode
    transport = await handshake.to_transport()
    
    # Send encrypted messages
    for i in range(3):
        message = f"Message {i+1}".encode()
        ciphertext = await transport.send(message)
        await framed_writer.write_message(ciphertext)
        
        response_ct = await framed_reader.read_message()
        response = await transport.receive(response_ct)
        print(f"Server response: {response.decode()}")
    
    await framed_writer.close()

asyncio.run(async_client())
```

#### Async Framing

For asyncio streams, use `AsyncFramedReader` and `AsyncFramedWriter`:

```python
from noiseframework import AsyncFramedReader, AsyncFramedWriter

async def async_framed_communication(reader, writer):
    framed_writer = AsyncFramedWriter(writer)
    framed_reader = AsyncFramedReader(reader)
    
    # Write framed message
    await framed_writer.write_message(b"Hello, async!")
    
    # Read framed message
    message = await framed_reader.read_message()
    
    # Close when done
    await framed_writer.close()
```

#### Async Convenience Functions

```python
from noiseframework import (
    async_write_framed_message,
    async_read_framed_message,
)

# Write single message
await async_write_framed_message(writer, b"Hello")

# Read single message
message = await async_read_framed_message(reader)
```

#### Async API Classes

**AsyncNoiseHandshake**: Async wrapper for `NoiseHandshake`
- `await set_as_initiator()` - Set role as initiator
- `await set_as_responder()` - Set role as responder
- `await generate_static_keypair()` - Generate static keys
- `await set_static_keypair(private, public)` - Set existing keys
- `await set_remote_static_public_key(public)` - Set remote's public key
- `await initialize()` - Initialize handshake state
- `await write_message(payload)` - Write handshake message
- `await read_message(message)` - Read handshake message
- `await to_transport()` - Get AsyncNoiseTransport after completion
- `await get_handshake_hash()` - Get handshake hash
- `.is_complete` - Check if handshake is complete (property)

**AsyncNoiseTransport**: Async wrapper for `NoiseTransport`
- `await send(plaintext, ad=b"")` - Encrypt and send message
- `await receive(ciphertext, ad=b"")` - Decrypt and receive message
- `.send_nonce` - Current send nonce (property)
- `.receive_nonce` - Current receive nonce (property)

**AsyncFramedReader**: Async framed message reader
- `await read_message()` - Read length-prefixed message
- `await close()` - Close reader
- `.messages_received` - Message counter (property)

**AsyncFramedWriter**: Async framed message writer
- `await write_message(message)` - Write length-prefixed message
- `await close()` - Close writer and wait for completion
- `.messages_sent` - Message counter (property)

**Key Points:**
- All async operations use `run_in_executor` internally
- No blocking calls in the async event loop
- Compatible with `asyncio.StreamReader` and `asyncio.StreamWriter`
- Same security guarantees as synchronous version
- Logging support in all async classes

See [`examples/async_tcp_example.py`](examples/async_tcp_example.py) for a complete working example with server and client.

---

### Pre-Shared Key (PSK) Support

NoiseFramework supports Pre-Shared Key (PSK) patterns for quantum-resistant authentication. PSK patterns add an additional layer of security by mixing a shared secret into the handshake, providing protection against future quantum computer attacks on Diffie-Hellman key exchange.

#### What are PSK Patterns?

PSK patterns combine traditional Noise handshakes with pre-shared keys:
- **Quantum Resistance**: PSKs are immune to quantum computer attacks (unlike DH)
- **Additional Authentication**: Extra authentication layer beyond public keys
- **Pre-computation Resistance**: Attackers can't pre-compute attacks
- **Common Use Cases**: IoT devices, enterprise VPNs, defense systems, embedded systems

#### PSK Pattern Format

PSK patterns use modifiers (`psk0` through `psk4`) that indicate when the PSK is mixed:

```
Noise_XXpsk3_25519_ChaChaPoly_SHA256  # XX with PSK after third message
Noise_NNpsk0_25519_ChaChaPoly_SHA256  # NN with PSK before first message
Noise_IKpsk2_448_AESGCM_BLAKE2b       # IK with PSK after second message
```

**PSK Modifiers:**
- `psk0`: PSK mixed before first message (maximum quantum resistance)
- `psk1`: PSK mixed after first message
- `psk2`: PSK mixed after second message (common for IK patterns)
- `psk3`: PSK mixed after third message (most common - XXpsk3)
- `psk4`: PSK mixed after fourth message (rare)

#### Basic PSK Usage

```python
import os
from noiseframework import NoiseHandshake, NoiseTransport

# Generate or load a 32-byte pre-shared key
psk = os.urandom(32)  # Must be exchanged securely out-of-band

# Initiator
initiator = NoiseHandshake("Noise_NNpsk0_25519_ChaChaPoly_SHA256")
initiator.set_as_initiator()
initiator.set_psk(psk)  # Set PSK before initialize()
initiator.initialize()

# Responder
responder = NoiseHandshake("Noise_NNpsk0_25519_ChaChaPoly_SHA256")
responder.set_as_responder()
responder.set_psk(psk)  # Must use same PSK
responder.initialize()

# Perform handshake (PSK mixed automatically)
msg1 = initiator.write_message(b"")
responder.read_message(msg1)

msg2 = responder.write_message(b"")
initiator.read_message(msg2)

# Create transport - now quantum-resistant!
init_send, init_recv = initiator.to_transport()
resp_send, resp_recv = responder.to_transport()

init_transport = NoiseTransport(init_send, init_recv)
resp_transport = NoiseTransport(resp_send, resp_recv)

# Secure communication
ciphertext = init_transport.send(b"Quantum-resistant message!")
plaintext = resp_transport.receive(ciphertext)
```

#### XXpsk3 - Most Common PSK Pattern

The `XXpsk3` pattern (mutual authentication + late PSK) is the most commonly used PSK pattern:

```python
psk = os.urandom(32)

# Initiator
initiator = NoiseHandshake("Noise_XXpsk3_25519_ChaChaPoly_SHA256")
initiator.set_as_initiator()
initiator.generate_static_keypair()  # XX requires static keys
initiator.set_psk(psk)
initiator.initialize()

# Responder
responder = NoiseHandshake("Noise_XXpsk3_25519_ChaChaPoly_SHA256")
responder.set_as_responder()
responder.generate_static_keypair()
responder.set_psk(psk)
responder.initialize()

# Three-message handshake
msg1 = initiator.write_message(b"")
responder.read_message(msg1)

msg2 = responder.write_message(b"")
initiator.read_message(msg2)

msg3 = initiator.write_message(b"")  # PSK mixed here
responder.read_message(msg3)

# Both parties mutually authenticated + quantum-resistant
assert initiator.remote_static_public == responder.static_public
assert responder.remote_static_public == initiator.static_public
```

#### PSK with Async/Await

PSK works seamlessly with async code:

```python
async def async_psk_example():
    psk = os.urandom(32)
    
    handshake = AsyncNoiseHandshake("Noise_XXpsk3_25519_ChaChaPoly_SHA256")
    await handshake.set_as_initiator()
    await handshake.generate_static_keypair()
    await handshake.set_psk(psk)  # Async PSK setting
    await handshake.initialize()
    
    # Continue with async handshake...
```

#### PSK Security Considerations

**When to Use PSK:**
- You need quantum resistance
- You can securely exchange a shared secret out-of-band
- You're building IoT or embedded systems
- You want additional authentication beyond public keys

**PSK Management:**
- PSKs must be exchanged securely (never over an insecure channel)
- Use strong randomness (32 bytes from `os.urandom()`)
- PSKs can be safely reused across sessions
- Consider key rotation policies for long-lived PSKs

**Placement Trade-offs:**
- **Early PSK (psk0)**: Maximum quantum resistance, protects entire handshake
- **Late PSK (psk3)**: Allows public key exchange first, then adds PSK protection

See [`examples/psk_example.py`](examples/psk_example.py) for complete working examples of NNpsk0, XXpsk3, and IKpsk2 patterns.

---

### Fallback Pattern Support

NoiseFramework supports fallback patterns for graceful handshake degradation when the initiator's first message cannot be decrypted. This implements the **Noise Pipes protocol** (Section 10.2 of the Noise spec).

#### What are Fallback Patterns?

Fallback patterns enable recovery from handshake failures without dropping the connection:
- **Graceful Degradation**: When IK/NK fails, fallback to XX pattern
- **Key Rotation**: Handle responder static key changes
- **PSK Outdated**: Recover from outdated pre-shared keys
- **Role Reversal**: Responder becomes effective initiator in fallback

#### Noise Pipes Protocol (IK → XXfallback)

The most common fallback scenario is **Noise Pipes**: Alice attempts IK, Bob detects wrong static key and falls back to XXfallback.

```python
import os
from noiseframework import NoiseHandshake

# Bob generates his real keys
bob = NoiseHandshake("Noise_XX_25519_ChaChaPoly_SHA256")
bob.set_as_responder()
bob.generate_static_keypair()
bob.initialize()

# Alice attempts IK with WRONG Bob static key (outdated)
alice = NoiseHandshake("Noise_IK_25519_ChaChaPoly_SHA256")
alice.set_as_initiator()
alice.generate_static_keypair()
alice.set_remote_static_public_key(wrong_bob_key)  # Outdated key
alice.initialize()

# Alice sends IK first message - will fail decryption
ik_msg1 = alice.write_message(b"Hello Bob")

# Bob cannot decrypt - extract Alice's ephemeral key (first 32 bytes)
alice_ephemeral = ik_msg1[:32]

# Bob initiates fallback to XXfallback
bob.start_fallback(alice_ephemeral)

# Bob now sends first XXfallback message (role reversal)
fallback_msg1 = bob.write_message(b"Fallback initiated")

# Alice detects IK failure and switches to XXfallback
alice_fallback = NoiseHandshake("Noise_XXfallback_25519_ChaChaPoly_SHA256")
alice_fallback.set_as_initiator()
alice_fallback.set_static_keypair(alice.static_private, alice.static_public)

# Reuse Alice's ephemeral keys (XXfallback uses them as pre-message)
alice_fallback.ephemeral_private = alice.ephemeral_private
alice_fallback.ephemeral_public = alice.ephemeral_public
alice_fallback.initialize()

# Alice reads Bob's fallback message
payload1 = alice_fallback.read_message(fallback_msg1)

# Alice sends her static key (second XXfallback message)
fallback_msg2 = alice_fallback.write_message(b"Acknowledged")

# Bob reads Alice's response - handshake complete!
payload2 = bob.read_message(fallback_msg2)

# Create transport channels
bob_send, bob_recv = bob.to_transport()
alice_send, alice_recv = alice_fallback.to_transport()

# Secure communication established despite initial IK failure!
```

#### Fallback Mechanics

**Pattern Transformation:**
```
XX → XXfallback    # First message "e" becomes pre-message
NN → NNfallback    # First message "e" becomes pre-message
IK → XXfallback    # Cannot use IKfallback (first message contains DH)
NK → XXfallback    # Cannot use NKfallback (first message contains DH)
```

**Valid Fallback Patterns:**
- Fallback requires initiator's first message to be "e", "s", or "e, s"
- Cannot fallback patterns where first message contains DH operations (es, se, ss, ee)

**Fallback Process:**
1. Responder receives initiator's first message but cannot decrypt
2. Responder extracts initiator's ephemeral key from failed message
3. Responder calls `start_fallback(remote_ephemeral_public_key)`
4. Pattern switches (e.g., XX → XXfallback), state re-initialized
5. Responder sends first message (role reversal)
6. Initiator also switches to fallback pattern
7. Handshake completes normally with fallback pattern

#### API Usage

```python
# Responder only - call when decryption fails
handshake.start_fallback(remote_ephemeral_public_key: bytes) -> None

# Async version
await handshake.start_fallback(remote_ephemeral_public_key: bytes)

# Example error handling with fallback
try:
    payload = responder.read_message(initiator_msg1)
except Exception:
    # Extract ephemeral key (first DHLEN bytes)
    alice_ephemeral = initiator_msg1[:32]  # 32 for Curve25519
    responder.start_fallback(alice_ephemeral)
    # Continue with fallback pattern...
```

#### Security Considerations

**When to Use Fallback:**
- Server static key rotation scenarios
- PSK may be outdated but fallback acceptable
- Graceful degradation preferred over connection failure

**Fallback Limitations:**
- Both parties must coordinate fallback (initiator must also switch)
- Ephemeral key must be preserved from failed message
- Only works with patterns where first message can be pre-message

**Noise Pipes Use Case:**
The IK → XXfallback pattern is standardized as "Noise Pipes" and widely used for:
- IoT device provisioning (device has outdated server key)
- Server key rotation (clients gradually update)
- Fallback from authenticated to mutual authentication

See [`examples/fallback_example.py`](examples/fallback_example.py) for a complete working demonstration of the Noise Pipes protocol with IK → XXfallback.

---

### High-Level Connection API

For most applications, the high-level `NoiseConnection` and `AsyncNoiseConnection` classes provide the simplest way to establish secure connections. These classes automatically handle handshakes, transport mode transitions, and message framing.

#### Synchronous Connection Example

```python
from noiseframework import NoiseConnection
import socket
import threading

def server():
    """Responder side."""
    server_sock = socket.socket()
    server_sock.bind(("localhost", 9999))
    server_sock.listen(1)
    client_sock, _ = server_sock.accept()
    
    # Create connection and accept - handshake happens automatically
    with NoiseConnection("Noise_XX_25519_ChaChaPoly_SHA256", "responder") as conn:
        conn.accept(client_sock)