c4e2e 1.0.1

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 Improvemnents:

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

pip install cryptography
# clone repo then:
pip install -e .

---

Quick Start

Generate keypair

from c4e2e import keygen, pubkey_to_b64, export_ed25519_private

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

# Or via CLI
c4e2e keygen --out-dir ./keys
# → ./keys/identity.pem  (chmod 600)
# → ./keys/identity.pub  (base64 pubkey)

---

Modes

Transmitter Mode

Only encrypts and signs outgoing payloads. Does not decrypt.

from c4e2e import keygen, load_config, create_node, pubkey_to_b64

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

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=pubkey_to_b64(receiver_pub),
    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
frame = node_a.encrypt("task.json", {"cmd": "run"}, pubkey_to_b64(pub_b))
result = node_b.decrypt(frame)

# B → A
ack = node_b.encrypt("ack.json", {"status": "ok"}, pubkey_to_b64(pub_a))
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,
    build_metadata, build_extra,
    pack_frame, unpack_frame,
    encrypt_for_recipient, decrypt_from_sender,
    sign, verify,
)
import json

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

# 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
encrypted_body = encrypt_for_recipient(job_bytes, receiver_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
plaintext = decrypt_from_sender(enc_body, receiver_priv, sender_pub)
job = json.loads(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       ← Ed25519, X25519, AES-256-GCM, HKDF
│   ├── 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
├── examples/
│   ├── transmitter_agent.py
│   ├── receiver_agent.py
│   └── hybrid_and_payload_crafting.py
├── tests/
│   └── test_c4e2e.py   ← 24 tests
└── pyproject.toml