c4e2e 2.1.0.dev1

True end-to-end encryption for agent-to-agent traffic — No0B@ckSappi3

c4e2e

True end-to-end encryption for agent-to-agent traffic
No0B@ckSappi3 | Offensive Security Tooling

---

Upcoming Improvements

Always check CHANGELOG.md for latest updates.

Major

• Add ability for external key management (USB)

Minor

• Harden policy enforcement for accepted and non-accepted traffic

Patches

• Add functionality to zero memory upon dealloc to protect privkey
• Add functionality to protect session keys while in memory

Crypto Architecture

Layer	Algorithm	Purpose
Identity signing	Ed25519	Long-term identity keypairs, payload signatures
Key exchange	X25519 + HKDF-SHA256	Ephemeral session key derivation
Symmetric encryption	AES-256-GCM	Payload encryption
Key derivation	HKDF-SHA256	Shared secret → session key

Every payload uses a fresh ephemeral X25519 key. Session keys are never reused.
Receiver rejects anything not signed by a trusted sender pubkey.

---

Wire Format

[4 bytes: metadata_len (big-endian uint32)]
[N bytes: base64-encoded metadata JSON — UNENCRYPTED]
[remaining: JSON of encrypted body]

Metadata (visible in transit, base64-encoded, exactly 3 keys):

{
  "name":   "output_filename.json",
  "pubkey": "<sender Ed25519 pubkey, base64>",
  "extra":  { "host_info": {...}, "job_id": "...", "tags": [...] }
}

Encrypted body (opaque without receiver's key):

{
  "eph_pub":    "<base64 X25519 ephemeral public key>",
  "ciphertext": "<base64 AES-256-GCM ciphertext>",
  "signature":  "<base64 Ed25519 signature over ciphertext>"
}

The receiver decrypts → writes the job JSON to a file named metadata.name.

---

Install

From wheel (recommended)

pip install c4e2e

Pre-built wheels are available for Python 3.11 and 3.12:

Platform	Requirement
Linux x86_64	glibc ≥ 2.28 (Ubuntu 18.04+, Debian Buster+, RHEL/AlmaLinux 8+)
Windows AMD64	Windows 10 or later
macOS arm64	macOS 12+ (Apple Silicon)

Older Linux: systems with glibc < 2.28 (e.g. CentOS 7, Ubuntu 16.04) are not supported by the pre-built wheel. Build from source on those systems.

From source

Requires CMake ≥ 3.15 and OpenSSL ≥ 1.1.1 installed on the build host.

# Install build tools
pip install scikit-build-core cmake

# Build and install in editable mode
pip install -e .

# Or build the native library manually first (dev/debugging):
#   Linux/macOS:
#     cmake -B native/build native -DCMAKE_BUILD_TYPE=Release
#     cmake --build native/build
#
#   Windows (requires OpenSSL installed via choco/vcpkg):
#     cmake -B native\build native -DOPENSSL_ROOT_DIR="C:\Program Files\OpenSSL-Win64"
#     cmake --build native\build --config Release
#
#   macOS — pass Homebrew OpenSSL root so cmake can find it:
#     OPENSSL_ROOT_DIR=$(brew --prefix openssl@3) \
#       cmake -B native/build native -DCMAKE_BUILD_TYPE=Release
#     cmake --build native/build

Windows note: the C++ library links against OpenSSL. For dev builds (running tests directly from source), OpenSSL's DLL directory must be findable at runtime. crypto_bindings.py automatically registers the common chocolatey install paths (C:\Program Files\OpenSSL-Win64\bin etc.) via os.add_dll_directory() so you do not need to add anything to PATH manually. Wheel-installed builds are self-contained and have no such requirement.

---

Quick Start

Generate keypair

from c4e2e import keygen, pubkey_to_b64, export_ed25519_private_pem

priv, pub = keygen()                         # returns (KeyHandle, KeyHandle)
pub_b64 = pubkey_to_b64(pub)                 # share this with peers
pem = export_ed25519_private_pem(priv)       # write to disk, keep secret

# Or via CLI
c4e2e keygen --out-dir ./keys
# → ./keys/identity.pem   (chmod 600, Ed25519 private key)
# → ./keys/identity.pub   (base64 pubkey)
# → ./keys/transport.pem  (chmod 600, X25519 transport private key)

---

Modes

Transmitter Mode

Only encrypts and signs outgoing payloads. Does not decrypt.

from c4e2e import keygen, pubkey_to_b64, load_config, create_node

sender_priv, sender_pub = keygen()
receiver_priv, receiver_pub = keygen()

sender_pub_b64   = pubkey_to_b64(sender_pub)
receiver_pub_b64 = pubkey_to_b64(receiver_pub)

# Build the receiver node first so we can get its X25519 transport key.
# recipient_pubkey must be the receiver's X25519 transport key, NOT their Ed25519
# identity key. The two serve different roles:
#   Ed25519 → authentication / trusted-key allowlist
#   X25519  → ECDH session key derivation (what senders encrypt to)
rx_cfg = load_config(
    mode="receiver",
    private_key=receiver_priv,
    trusted_keys=[sender_pub_b64],
    output_dir="./out",
)
rx = create_node(rx_cfg)

cfg = load_config(
    mode="transmitter",
    private_key=sender_priv,       # or private_key_path="/path/to/key.pem"
    output_dir="./out",
)
tx = create_node(cfg)

frame = tx.encrypt(
    name="job_001.json",           # receiver writes a file with this name
    job={
        "task": "port_scan",
        "target": "10.10.10.0/24",
        "ports": [22, 80, 443],
    },
    recipient_pubkey=rx.transport_public_key_b64,   # X25519 transport key
    job_id="job-001",
    tags=["recon", "external"],
)

# frame is raw bytes — send over socket, HTTP, queue, write to file, etc.

---

Receiver Mode

Only decrypts incoming payloads. Rejects anything not signed by a trusted key.

from c4e2e import load_config, create_node, UntrustedSenderError, SignatureError

cfg = load_config(
    mode="receiver",
    private_key=receiver_priv,
    trusted_keys=[pubkey_to_b64(sender_pub)],   # allowlist
    output_dir="./decrypted",
)
rx = create_node(cfg)

try:
    result = rx.decrypt(frame)
    print(result["name"])         # "job_001.json"
    print(result["job"])          # {"task": "port_scan", ...}
    print(result["output_path"])  # Path("./decrypted/job_001.json")
    print(result["metadata"])     # full metadata dict

except UntrustedSenderError:
    print("Sender not in trusted set — rejected")
except SignatureError:
    print("Signature invalid — payload tampered or wrong key")

The decrypted job is automatically written to output_dir / metadata["name"].

---

Hybrid Mode

Both encrypt and decrypt. Typical for peer agents.

from c4e2e import load_config, create_node

cfg_a = load_config(
    mode="hybrid",
    private_key=priv_a,
    trusted_keys=[pubkey_to_b64(pub_b)],
    output_dir="./agent_a_out",
)
node_a = create_node(cfg_a)

cfg_b = load_config(
    mode="hybrid",
    private_key=priv_b,
    trusted_keys=[pubkey_to_b64(pub_a)],
    output_dir="./agent_b_out",
)
node_b = create_node(cfg_b)

# A → B: use node_b's X25519 transport key, not its Ed25519 identity key
frame = node_a.encrypt("task.json", {"cmd": "run"}, node_b.transport_public_key_b64)
result = node_b.decrypt(frame)

# B → A
ack = node_b.encrypt("ack.json", {"status": "ok"}, node_a.transport_public_key_b64)
node_a.decrypt(ack)

---

Key Configuration Sources

Priority: explicit kwarg > env variable > config file > default

Option A: Hardcode (dev/testing)

cfg = load_config(mode="transmitter", private_key=my_priv_key_object)

Option B: Environment variables

export C4E2E_MODE=receiver
export C4E2E_PRIVATE_KEY_PATH=/etc/c4e2e/identity.pem
export C4E2E_TRUSTED_KEYS="base64key1,base64key2"
export C4E2E_OUTPUT_DIR=/var/c4e2e/out
export C4E2E_RECIPIENT_PUBKEY=base64key   # used by CLI encrypt

cfg = load_config()  # reads all C4E2E_* vars automatically

Option C: Config file (JSON)

{
  "c4e2e": {
    "mode": "hybrid",
    "output_dir": "/var/c4e2e/out",
    "private_key_path": "/etc/c4e2e/identity.pem",
    "trusted_keys": ["base64key1", "base64key2"]
  }
}

Option D: Config file (TOML)

[c4e2e]
mode = "hybrid"
output_dir = "/var/c4e2e/out"
private_key_path = "/etc/c4e2e/identity.pem"
trusted_keys = ["base64key1", "base64key2"]

cfg = load_config(config_file="/etc/c4e2e/config.toml")

Option E: CLI flags (argparse integration)

import argparse
from c4e2e import add_cli_args, config_from_args

parser = argparse.ArgumentParser()
parser.add_argument("--target")
add_cli_args(parser)   # injects --c4e2e-mode, --c4e2e-key, --c4e2e-trusted, etc.

args = parser.parse_args()
cfg = config_from_args(args)
node = create_node(cfg)

./agent.py --target 10.0.0.0/8 \
  --c4e2e-mode transmitter \
  --c4e2e-key ./keys/identity.pem \
  --c4e2e-trusted <recipient_b64_pubkey>

---

Standalone CLI

# Generate keypair
c4e2e keygen --out-dir ./keys

# Encrypt a payload to a file
c4e2e encrypt \
  --key ./keys/identity.pem \
  --recipient-key <RECIPIENT_PUBKEY_B64> \
  --name "recon_001.json" \
  --job '{"task":"port_scan","target":"10.0.0.0/8"}' \
  --job-id "job-001" \
  --tags "recon,external" \
  --out ./payload.bin

# Encrypt from a job file
c4e2e encrypt \
  --key ./keys/identity.pem \
  --recipient-key <RECIPIENT_PUBKEY_B64> \
  --name "bigjob.json" \
  --job-file ./job.json \
  --out ./payload.bin

# Decrypt a payload
c4e2e decrypt \
  --key ./keys/identity.pem \
  --trusted-key <SENDER_PUBKEY_B64> \
  --payload ./payload.bin \
  --output-dir ./decrypted \
  --print-job

# Inspect a payload (metadata only, no decryption needed)
c4e2e inspect --payload ./payload.bin

# Watch a directory for incoming .bin payloads (daemon mode)
c4e2e watch \
  --key ./keys/identity.pem \
  --trusted-key <SENDER_PUBKEY_B64> \
  --watch-dir ./inbox \
  --output-dir ./decrypted \
  --delete-after \
  --interval 0.5

# Use config file instead of flags
c4e2e --config /etc/c4e2e/config.toml decrypt --payload ./payload.bin

---

Manual Payload Crafting (Low-Level API)

from c4e2e import (
    keygen, pubkey_to_b64,
    generate_x25519_keypair, x25519_pub_to_b64,
    build_metadata, build_extra,
    pack_frame, unpack_frame,
    encrypt_for_recipient, decrypt_from_sender,
)
import json

sender_priv, sender_pub = keygen()
recv_x_priv, recv_x_pub = generate_x25519_keypair()   # X25519 transport keypair

# 1. Build metadata
metadata = build_metadata(
    name="custom_output.json",
    pubkey_b64=pubkey_to_b64(sender_pub),
    extra=build_extra(
        job_id="op-nightfall-001",
        tags=["c2", "persistence"],
        include_host=True,
    ),
)

# 2. Serialize job
job_bytes = json.dumps({"task": "beacon", "interval": 300}).encode()

# 3. Encrypt to receiver's X25519 transport public key
encrypted_body = encrypt_for_recipient(job_bytes, recv_x_pub, sender_priv)

# 4. Pack into wire frame
frame = pack_frame(metadata, encrypted_body)

# ── On the receiving end ──

# 5. Unpack (metadata visible without keys)
meta, enc_body = unpack_frame(frame)
print(meta["name"])    # custom_output.json
print(meta["pubkey"])  # sender's pubkey

# 6. Decrypt — returns MsgHandle; plaintext stays in locked C++ memory
with decrypt_from_sender(enc_body, recv_x_priv, sender_pub) as msg:
    plaintext = msg.to_bytes()   # orchestration-shim crossing; minimize lifetime
    job = json.loads(plaintext)
    del plaintext
print(job)  # {"task": "beacon", "interval": 300}

---

Adding Trust at Runtime

rx = create_node(cfg)

# Add a new trusted key without restarting
rx.trust("base64newpubkey...")

# Remove a key
rx.untrust("base64oldpubkey...")

# List trusted keys
print(rx.trusted_keys)

---

Output File Format

When a receiver decrypts a payload, it writes a JSON file to output_dir:

output_dir/
└── job_001.json          ← filename from metadata.name

File contents:

{
  "metadata": {
    "name": "job_001.json",
    "pubkey": "<sender pubkey b64>",
    "extra": {
      "host_info": {
        "hostname": "agent-box",
        "ip": "10.0.0.5",
        "platform": "Linux",
        "arch": "x86_64",
        "timestamp": "2025-01-15T04:20:00Z"
      },
      "job_id": "job-001",
      "tags": ["recon", "external"]
    }
  },
  "job": {
    "task": "port_scan",
    "target": "10.10.10.0/24",
    "ports": [22, 80, 443]
  }
}

---

Error Handling

from c4e2e import UntrustedSenderError, SignatureError, C4NodeError, ModeError

try:
    result = rx.decrypt(frame)
except UntrustedSenderError:
    # sender pubkey not in trusted set — drop
    pass
except SignatureError:
    # signature invalid — payload tampered or wrong sender key
    pass
except C4NodeError:
    # malformed frame, decryption failure, etc.
    pass

---

Security Notes

• Directory traversal protected: metadata.name is sanitized to basename before writing
• Signature-first: receiver verifies Ed25519 signature before attempting decryption
• Ephemeral keys: every payload uses a fresh X25519 key — no session key reuse
• Metadata is plaintext: name, pubkey, and extra are visible to a passive observer. Don't put secrets in extra
• Trusted key allowlist: receiver drops any payload whose sender pubkey isn't pre-registered
• PEM keys can be password-protected: use export_ed25519_private(priv, password=b"...")

---

Package Structure

c4e2e/
├── c4e2e/
│   ├── __init__.py          ← public API
│   ├── crypto.py            ← native-backed shim (re-exports _native.crypto_wrapper)
│   ├── payload.py           ← wire format, metadata, frame pack/unpack
│   ├── config.py            ← config loading (env, file, kwargs, CLI)
│   ├── node.py              ← Transmitter, Receiver, Hybrid classes
│   ├── cli.py               ← standalone CLI tool
│   └── _native/
│       ├── __init__.py      ← deferred load docstring
│       ├── crypto_bindings.py  ← raw ctypes ABI (one-to-one with c4e2e_crypto.h)
│       ├── crypto_wrapper.py   ← KeyHandle, MsgHandle, high-level wrappers
│       └── c4e2e_crypto.dll    ← (or .so / .dylib) — installed by wheel
├── native/
│   ├── CMakeLists.txt       ← C++ build config
│   ├── include/
│   │   └── c4e2e_crypto.h   ← public C API
│   └── src/
│       └── c4e2e_crypto.cpp ← full implementation
├── tests/
│   └── test_c4e2e.py
└── pyproject.toml           ← scikit-build-core build backend

License

Apache License 2.0 — see LICENSE for the full text.

---

Security Model (v2.1)

All crypto executes in c4e2e_crypto (C++ shared library):

• Private keys live in C++ KeyRegistry (mutex-protected). Python holds only an opaque uint64_t handle (KeyHandle).
• Decrypted plaintext lives in a page-locked HeapSecureBuf in C++ PlaintextRegistry. Python receives a MsgHandle, not bytes.
• MsgHandle.to_bytes() is the single documented crossing point for orchestration code (e.g., node.py) that needs raw bytes.
• No raw X25519 private key export: use export_x25519_private_pem for stored keys.
• Memory zeroization uses OPENSSL_cleanse (non-Windows) / SecureZeroMemory (Windows) — guaranteed not to be optimised away.
• PEM passwords are length-delimited (not null-terminated) on both import and export paths — binary passwords with embedded null bytes are handled correctly.