dcp-protocol-sdk 1.0.0

DCP Protocol SDK — 9-layer zero-knowledge AES-256-GCM encryption

dcp-protocol-sdk

9-layer zero-knowledge AES-256-GCM encryption SDK for Python.

Install

mkdir -p /opt/dcp-sdk-python && cd /opt/dcp-sdk-python

cat > dcp_sdk/__init__.py << 'PYEOF'
"""DCP Protocol SDK — 9-layer zero-knowledge AES-256-GCM encryption for Python."""

from .client import DCP, encrypt, decrypt, generate_master_key, generate_salt, derive_key_from_password

__version__ = "1.0.0"
__all__ = ["DCP", "encrypt", "decrypt", "generate_master_key", "generate_salt", "derive_key_from_password"]
PYEOF

mkdir -p dcp_sdk

cat > dcp_sdk/client.py << 'PYEOF'
"""
DCP Protocol Python SDK
9-layer AES-256-GCM zero-knowledge encryption

Usage:
    from dcp_sdk import DCP

    dcp = DCP(
        api_url="https://dcprotocol.link",
        api_key=os.environ["DCP_API_KEY"],
        master_key=os.environ["DCP_MASTER_KEY"],
    )

    # Encrypt and store
    record = dcp.store("notes", {"title": "Secret", "body": "Nobody can read this"})

    # Fetch and decrypt
    notes = dcp.fetch("notes")
"""

import os
import json
import base64
import hashlib
import hmac as hmac_mod
import requests
from datetime import datetime, timezone
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives.ciphers.aead import AESGCM


def _b64(data: bytes) -> str:
    return base64.b64encode(data).decode()


def _unb64(s: str) -> bytes:
    return base64.b64decode(s)


def generate_master_key() -> str:
    """Generate a random 256-bit master key (64 hex chars)."""
    return os.urandom(32).hex()


def generate_salt() -> str:
    """Generate a random 128-bit salt (32 hex chars) for PBKDF2."""
    return os.urandom(16).hex()


def derive_key_from_password(password: str, salt: str, iterations: int = 600000) -> str:
    """
    Derive a 256-bit master key from a password using PBKDF2-SHA256.

    Args:
        password: User password (min 8 chars)
        salt: Per-user salt from database (min 16 chars, not secret)
        iterations: PBKDF2 iterations (default 600K, OWASP 2023)

    Returns:
        64-char hex string suitable as master_key
    """
    if len(password) < 8:
        raise ValueError("Password must be at least 8 characters")
    if len(salt) < 16:
        raise ValueError("Salt must be at least 16 characters")

    kdf = PBKDF2HMAC(
        algorithm=hashes.SHA256(),
        length=32,
        salt=salt.encode(),
        iterations=iterations,
    )
    key = kdf.derive(password.encode())
    return key.hex()


def _hkdf_derive(master: bytes, salt: bytes, info: bytes) -> bytes:
    """Derive a 256-bit key using HKDF-SHA256."""
    hkdf = HKDF(algorithm=hashes.SHA256(), length=32, salt=salt, info=info)
    return hkdf.derive(master)


def encrypt(plaintext: str, master_key_hex: str, aad: dict = None) -> dict:
    """
    Encrypt plaintext through 9 layers of AES-256-GCM with HMAC envelope.

    Args:
        plaintext: String to encrypt (typically JSON)
        master_key_hex: 64-char hex master key
        aad: Additional authenticated data (dict)

    Returns:
        DCP envelope dict (version, ciphertext, layers, hmac, etc.)
    """
    if len(master_key_hex) != 64:
        raise ValueError("Master key must be 64 hex characters (32 bytes)")

    master = bytes.fromhex(master_key_hex)
    aad_obj = aad or {}
    aad_json = json.dumps(aad_obj, sort_keys=True)
    aad_bytes = aad_json.encode()

    current = plaintext.encode()
    layers = []

    for i in range(1, 10):
        salt = os.urandom(16)
        nonce = os.urandom(12)
        info = f"dcp-adv9-layer-{i}".encode()
        layer_aad = aad_bytes + f"|layer={i}".encode()

        key = _hkdf_derive(master, salt, info)
        aesgcm = AESGCM(key)
        current = aesgcm.encrypt(nonce, current, layer_aad)

        layers.append({
            "layer": i,
            "salt": _b64(salt),
            "nonce": _b64(nonce),
        })

    envelope = {
        "version": "dcp-v1",
        "alg": "AES-256-GCM",
        "rounds": 9,
        "ciphertext": _b64(current),
        "layers": layers,
        "aad": _b64(aad_bytes),
        "created_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
    }

    # HMAC-SHA256 envelope signature
    hmac_key = _hkdf_derive(master, b"dcp-hmac-salt", b"dcp-hmac-v1")
    envelope_json = json.dumps(envelope, sort_keys=True)
    sig = hmac_mod.new(hmac_key, envelope_json.encode(), hashlib.sha256).digest()
    envelope["hmac"] = _b64(sig)

    return envelope


def decrypt(envelope: dict, master_key_hex: str, aad: dict = None) -> str:
    """
    Decrypt a DCP envelope through 9 layers.

    Args:
        envelope: DCP envelope dict
        master_key_hex: 64-char hex master key
        aad: Additional authenticated data (must match what was used for encryption)

    Returns:
        Decrypted plaintext string

    Raises:
        ValueError: If HMAC verification fails (tampering detected)
    """
    if envelope.get("version") != "dcp-v1":
        raise ValueError(f"Unsupported version: {envelope.get('version')}")
    if envelope.get("rounds") != 9:
        raise ValueError(f"Expected 9 rounds, got {envelope.get('rounds')}")
    if len(envelope.get("layers", [])) != 9:
        raise ValueError(f"Expected 9 layers, got {len(envelope.get('layers', []))}")

    master = bytes.fromhex(master_key_hex)

    # Verify HMAC first (fail-fast)
    if "hmac" in envelope:
        hmac_key = _hkdf_derive(master, b"dcp-hmac-salt", b"dcp-hmac-v1")
        unsigned = {k: v for k, v in envelope.items() if k != "hmac"}
        envelope_json = json.dumps(unsigned, sort_keys=True)
        expected = hmac_mod.new(hmac_key, envelope_json.encode(), hashlib.sha256).digest()
        if not hmac_mod.compare_digest(expected, _unb64(envelope["hmac"])):
            raise ValueError("HMAC verification failed — envelope tampered")

    aad_obj = aad or {}
    aad_json = json.dumps(aad_obj, sort_keys=True)
    aad_bytes = aad_json.encode()

    current = _unb64(envelope["ciphertext"])

    for i in range(9, 0, -1):
        layer = envelope["layers"][i - 1]
        salt = _unb64(layer["salt"])
        nonce = _unb64(layer["nonce"])
        info = f"dcp-adv9-layer-{i}".encode()
        layer_aad = aad_bytes + f"|layer={i}".encode()

        key = _hkdf_derive(master, salt, info)
        aesgcm = AESGCM(key)
        current = aesgcm.decrypt(nonce, current, layer_aad)

    return current.decode()


class DCP:
    """
    DCP Protocol client for Python.

    Example:
        dcp = DCP(
            api_url="https://dcprotocol.link",
            api_key=os.environ["DCP_API_KEY"],
            master_key=os.environ["DCP_MASTER_KEY"],
        )

        record = dcp.store("notes", {"title": "Secret"})
        notes = dcp.fetch("notes")
    """

    def __init__(self, api_url: str, api_key: str, master_key: str):
        self.api_url = api_url.rstrip("/")
        self.api_key = api_key
        self.master_key = master_key

        if len(master_key) != 64:
            raise ValueError("Master key must be 64 hex characters")

        self._session = requests.Session()
        self._session.headers.update({
            "X-API-Key": api_key,
            "Content-Type": "application/json",
        })

    def store(self, collection: str, data, record_id: str = None, aad: dict = None) -> dict:
        """
        Encrypt data and store on DCP server.

        Args:
            collection: Record group name
            data: Any JSON-serializable value
            record_id: Optional custom ID (UUID generated if omitted)
            aad: Additional authenticated data

        Returns:
            {"id": "...", "created_at": "..."}
        """
        plaintext = json.dumps(data)
        final_aad = {"collection": collection}
        if aad:
            final_aad.update(aad)

        envelope = encrypt(plaintext, self.master_key, final_aad)

        body = {"collection": collection, "envelope": envelope}
        if record_id:
            body["id"] = record_id

        resp = self._session.post(f"{self.api_url}/sdk/store", json=body)
        resp.raise_for_status()
        return resp.json()

    def fetch(self, collection: str, record_id: str = None) -> list:
        """
        Fetch and decrypt records from DCP server.

        Args:
            collection: Record group name
            record_id: Optional specific record ID

        Returns:
            List of {"id": "...", "data": {...}, "created_at": "..."}
        """
        params = {"collection": collection}
        if record_id:
            params["id"] = record_id

        resp = self._session.get(f"{self.api_url}/sdk/fetch", params=params)
        resp.raise_for_status()
        records = resp.json()

        results = []
        for r in records:
            try:
                plaintext = decrypt(r["envelope"], self.master_key, {"collection": collection})
                results.append({
                    "id": r["id"],
                    "data": json.loads(plaintext),
                    "created_at": r.get("created_at"),
                })
            except Exception as e:
                # Skip records that can't be decrypted (wrong key)
                import warnings
                warnings.warn(f"DCP: decrypt failed for {r['id']}: {e}")

        return results

    def delete(self, collection: str, record_id: str) -> dict:
        """Delete a record. Irreversible."""
        resp = self._session.post(
            f"{self.api_url}/sdk/delete",
            json={"collection": collection, "id": record_id},
        )
        resp.raise_for_status()
        return resp.json()

    def info(self) -> dict:
        """Get tenant info and health check."""
        resp = self._session.get(f"{self.api_url}/sdk/info")
        resp.raise_for_status()
        return resp.json()
PYEOF

# ─── Setup files ──────────────────────────────────────────────────

cat > setup.py << 'EOF'
from setuptools import setup, find_packages

setup(
    name="dcp-protocol-sdk",
    version="1.0.0",
    description="DCP Protocol SDK — 9-layer zero-knowledge AES-256-GCM encryption",
    long_description=open("README.md").read(),
    long_description_content_type="text/markdown",
    author="D.N.A Ecosystem",
    author_email="[e-mail]",
    url="https://dcprotocol.link",
    packages=find_packages(),
    install_requires=[
        "cryptography>=41.0.0",
        "requests>=2.28.0",
    ],
    python_requires=">=3.8",
    license="MIT",
    classifiers=[
        "Development Status :: 4 - Beta",
        "Intended Audience :: Developers",
        "License :: OSI Approved :: MIT License",
        "Programming Language :: Python :: 3",
        "Topic :: Security :: Cryptography",
    ],
    keywords="encryption aes-256-gcm zero-knowledge cryptography dcp hkdf hmac",
)
EOF

cat > pyproject.toml << 'EOF'
[build-system]
requires = ["setuptools>=64", "wheel"]
build-backend = "setuptools.backends._legacy:_Backend"

[project]
name = "dcp-protocol-sdk"
version = "1.0.0"
description = "DCP Protocol SDK — 9-layer zero-knowledge AES-256-GCM encryption"
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.8"
dependencies = ["cryptography>=41.0.0", "requests>=2.28.0"]
EOF

cat > README.md << 'MDEOF'
# dcp-protocol-sdk

> 9-layer zero-knowledge AES-256-GCM encryption SDK for Python.

## Install

```bash
pip install dcp-protocol-sdk

Quick start

import os
from dcp_sdk import DCP

dcp = DCP(
    api_url="https://dcprotocol.link",
    api_key=os.environ["DCP_API_KEY"],
    master_key=os.environ["DCP_MASTER_KEY"],
)

# Encrypt and store
record = dcp.store("notes", {"title": "Secret", "body": "Nobody can read this"})
print(f"Stored: {record['id']}")

# Fetch and decrypt
notes = dcp.fetch("notes")
for note in notes:
    print(note["data"]["title"], note["data"]["body"])

Features

• 9 layers of AES-256-GCM with independent HKDF-SHA256 derived keys
• HMAC-SHA256 envelope signature (tamper detection)
• PBKDF2 key derivation from password (600K iterations)
• Interoperable with JavaScript SDK (@humanscode/dcp-sdk)
• Zero-knowledge: server never sees plaintext

License

MIT — D.N.A Ecosystem