Modèles TEE & E2EE | Documentation de l'API Venice

Venice propose des modèles à confidentialité renforcée qui s’exécutent dans des Trusted Execution Environments (TEE) et prennent en charge le chiffrement de bout en bout (E2EE). Ces modèles offrent des garanties cryptographiques que vos données restent privées — même vis-à-vis de Venice.

Comprendre les niveaux de confidentialité

Type	Préfixe	Ce que cela signifie
TEE	`tee-*`	Le modèle s’exécute dans une enclave sécurisée matériellement. Venice ne peut pas accéder au calcul. Vous pouvez le vérifier par attestation.
E2EE	`e2ee-*`	Chiffrement de bout en bout complet. Vos prompts sont chiffrés côté client avant l’envoi. Seul le TEE peut les déchiffrer.

Les modèles E2EE incluent la protection TEE plus le chiffrement côté client. Les modèles TEE fournissent la sécurité de l’enclave sans nécessiter de chiffrement côté client.

Modèles disponibles

Loading…

Consultez la page Modèles pour la liste complète avec les prix et les limites de contexte.

Modèles TEE

Les modèles TEE s’exécutent à l’intérieur d’enclaves sécurisées matériellement (Intel TDX, NVIDIA Confidential Computing). Les poids du modèle et vos données sont protégés du système hôte — y compris l’infrastructure de Venice.

Usage de base

Les modèles TEE fonctionnent exactement comme des modèles réguliers :

from openai import OpenAI

client = OpenAI(
    api_key="your-venice-api-key",
    base_url="https://api.venice.ai/api/v1"
)

response = client.chat.completions.create(
    model="tee-qwen3-5-122b-a10b",
    messages=[{"role": "user", "content": "Explain quantum computing"}]
)

print(response.choices[0].message.content)

import OpenAI from 'openai';

const client = new OpenAI({
    apiKey: 'your-venice-api-key',
    baseURL: 'https://api.venice.ai/api/v1'
});

const response = await client.chat.completions.create({
    model: 'tee-qwen3-5-122b-a10b',
    messages: [{ role: 'user', content: 'Explain quantum computing' }]
});

console.log(response.choices[0].message.content);

curl https://api.venice.ai/api/v1/chat/completions \
  -H "Authorization: Bearer $API_KEY_VENICE" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "tee-qwen3-5-122b-a10b",
    "messages": [{"role": "user", "content": "Explain quantum computing"}]
  }'

Vérifier l’attestation TEE

Vous pouvez vérifier cryptographiquement qu’un modèle s’exécute dans un véritable TEE en récupérant son rapport d’attestation :

# Générer un nonce aléatoire (empêche les attaques par rejeu)
NONCE=$(openssl rand -hex 16)

# Récupérer l'attestation
curl "https://api.venice.ai/api/v1/tee/attestation?model=tee-qwen3-5-122b-a10b&nonce=$NONCE" \
  -H "Authorization: Bearer $API_KEY_VENICE"

import secrets
import requests

nonce = secrets.token_hex(16)

response = requests.get(
    f"https://api.venice.ai/api/v1/tee/attestation",
    params={"model": "tee-qwen3-5-122b-a10b", "nonce": nonce},
    headers={"Authorization": f"Bearer {api_key}"}
)

attestation = response.json()
print(f"Verified: {attestation['verified']}")
print(f"TEE Provider: {attestation['tee_provider']}")
print(f"Model: {attestation['model']}")

La réponse d’attestation inclut :

Champ	Description
`verified`	Indique si l’attestation a passé la vérification côté serveur
`nonce`	Votre nonce, confirmant la fraîcheur
`model`	L’ID du modèle attesté
`tee_provider`	Identifiant du fournisseur TEE
`intel_quote`	Quote Intel TDX brut (base64) pour vérification côté client
`nvidia_payload`	Données d’attestation GPU NVIDIA (le cas échéant)
`signing_key`	Clé publique pour vérifier les signatures de réponse (généralement requise pour les flux E2EE ; peut être omise pour certains modèles TEE simples)
`signing_address`	Adresse Ethereum dérivée de la clé de signature

Pour un usage en production, vérifiez l’attestation côté client en parsant le quote Intel TDX et en vérifiant l’attestation NVIDIA.

Pour la vérification d’un modèle TEE simple, signing_address et les champs de vérification côté serveur suffisent pour des contrôles d’attestation de base. Une signing_key est requise lorsque vous avez besoin d’un accord de clé E2EE côté client et de contrôles stricts de liaison de clé.

Signatures de réponse

Les modèles TEE peuvent signer leurs réponses, prouvant que la sortie provient de l’enclave attestée :

# Après avoir obtenu une completion, vérifier la signature
curl "https://api.venice.ai/api/v1/tee/signature?model=tee-qwen3-5-122b-a10b&request_id=chatcmpl-abc123" \
  -H "Authorization: Bearer $API_KEY_VENICE"

response = requests.get(
    f"https://api.venice.ai/api/v1/tee/signature",
    params={"model": "tee-qwen3-5-122b-a10b", "request_id": completion_id},
    headers={"Authorization": f"Bearer {api_key}"}
)

signature = response.json()
# Vérifier que la signature correspond à signing_address de l'attestation

Modèles E2EE

Les modèles E2EE ajoutent un chiffrement côté client par-dessus la protection TEE. Vos prompts sont chiffrés avant de quitter votre appareil, et seul le TEE peut les déchiffrer. L’E2EE Venice utilise :

ECDH (Elliptic Curve Diffie-Hellman) sur secp256k1 pour l’échange de clés
HKDF-SHA256 pour la dérivation de clés
AES-256-GCM pour le chiffrement symétrique
Attestation TEE pour vérifier que le modèle s’exécute dans une enclave sécurisée

L’E2EE nécessite une implémentation côté client. Les exemples ci-dessous montrent le protocole complet.

Comment fonctionne l’E2EE

Générer une paire de clés éphémères

Le client génère une paire de clés secp256k1 pour cette session.

Récupérer l'attestation TEE

Le client interroge /api/v1/tee/attestation et reçoit la clé publique du modèle, les preuves d’attestation et le nonce.

Vérifier l'attestation

Le client vérifie la correspondance du nonce, l’absence de mode debug et la validité de l’attestation.

Chiffrer les messages

Le client chiffre les prompts en utilisant le secret partagé ECDH → HKDF → AES-GCM.

Envoyer la requête

Le client envoie la requête avec les en-têtes E2EE (X-Venice-TEE-Client-Pub-Key, X-Venice-TEE-Model-Pub-Key, X-Venice-TEE-Signing-Algo).

Traitement par le TEE

Le TEE déchiffre la requête, la traite et chiffre la réponse.

Déchiffrer la réponse

Le client reçoit les chunks chiffrés et les déchiffre avec sa clé privée.

Prérequis

JavaScript (Node.js ESM) :

npm install elliptic @noble/ciphers @noble/hashes

Python :

pip install cryptography ecdsa requests

Étape 1 : Vérifier la prise en charge E2EE par le modèle

Vérifiez d’abord que le modèle prend en charge l’E2EE en interrogeant l’endpoint /models.

async function getE2EEModels(apiKey) {
  const response = await fetch('https://api.venice.ai/api/v1/models', {
    headers: { Authorization: `Bearer ${apiKey}` },
  })
  const { data } = await response.json()

  return data.filter(model => model.model_spec?.capabilities?.supportsE2EE === true)
}

// Exemple d'utilisation
const models = await getE2EEModels('your-api-key')
console.log('E2EE Models:', models.map(m => m.id))
// Sortie : ['e2ee-qwen3-5-122b-a10b', 'e2ee-glm-5', ...]

import requests

def get_e2ee_models(api_key: str) -> list:
    """Get list of models that support E2EE."""
    response = requests.get(
        'https://api.venice.ai/api/v1/models',
        headers={'Authorization': f'Bearer {api_key}'}
    )
    models = response.json()['data']

    return [
        model for model in models
        if model.get('model_spec', {}).get('capabilities', {}).get('supportsE2EE')
    ]

# Exemple d'utilisation
models = get_e2ee_models('your-api-key')
print('E2EE Models:', [m['id'] for m in models])

Étape 2 : Générer une paire de clés éphémères

Générez une nouvelle paire de clés pour chaque session. La clé privée doit rester en mémoire uniquement et être effacée de manière sécurisée après usage.

import { ec as EC } from 'elliptic'

function generateEphemeralKeyPair() {
  const ec = new EC('secp256k1')
  const keyPair = ec.genKeyPair()

  return {
    privateKey: new Uint8Array(keyPair.getPrivate().toArray('be', 32)),
    publicKeyHex: keyPair.getPublic('hex'), // Format non compressé (65 octets hex)
  }
}

// Sécurité : remplir la clé privée de zéros une fois terminé
function zeroFill(arr) {
  arr.fill(0)
}

from ecdsa import SECP256k1, SigningKey
import secrets

def generate_ephemeral_key_pair():
    """Generate ephemeral secp256k1 key pair for E2EE session."""
    private_key = SigningKey.generate(curve=SECP256k1)
    public_key = private_key.get_verifying_key()

    # Obtenir la clé publique non compressée (04 || x || y)
    public_key_bytes = b'\x04' + public_key.to_string()

    return {
        'private_key': private_key.to_string(),  # 32 octets
        'public_key_hex': public_key_bytes.hex()  # 130 caractères hex
    }

Helpers de validation

Utilisez ces fonctions helper pour valider les clés et le contenu chiffré avant d’envoyer les requêtes.

function validateClientPubkey(pubkeyHex) {
  if (pubkeyHex.length !== 130 || !pubkeyHex.startsWith('04')) {
    throw new Error(`Client pubkey must be 130 hex chars starting with '04' (got ${pubkeyHex.length})`)
  }
}

function isValidEncrypted(s) {
  // Minimum : ephemeral_pub (65) + nonce (12) + tag (16) = 93 octets = 186 caractères hex
  return s.length >= 186 && /^[0-9a-fA-F]+$/.test(s)
}

def validate_client_pubkey(pubkey_hex: str) -> None:
    """Validate client public key format."""
    if len(pubkey_hex) != 130 or not pubkey_hex.startswith('04'):
        raise ValueError(f"Client pubkey must be 130 hex chars starting with '04' (got {len(pubkey_hex)})")

def is_valid_encrypted(s: str) -> bool:
    """Check if string is valid hex-encrypted content."""
    # Minimum : ephemeral_pub (65) + nonce (12) + tag (16) = 93 octets = 186 caractères hex
    return len(s) >= 186 and all(c in '0123456789abcdefABCDEF' for c in s)

Étape 3 : Récupérer et vérifier l’attestation TEE

L’attestation prouve que le modèle s’exécute dans un véritable TEE. Vérifiez toujours l’attestation avant de faire confiance à la clé publique du modèle.

Important : longueur du nonce — Le nonce client doit faire 32 octets (64 caractères hex). Certains fournisseurs TEE exigent exactement 32 octets et rejettent les nonces plus courts.

import crypto from 'crypto'

async function fetchAndVerifyAttestation(modelId, apiKey) {
  // Générer un nonce client pour la protection anti-rejeu (32 octets = 64 caractères hex)
  const clientNonce = crypto.randomBytes(32).toString('hex')

  const response = await fetch(
    `https://api.venice.ai/api/v1/tee/attestation?model=${encodeURIComponent(modelId)}&nonce=${clientNonce}`,
    { headers: { Authorization: `Bearer ${apiKey}` } }
  )

  const attestation = await response.json()

  // Vérifier l'attestation
  if (attestation.verified !== true) {
    throw new Error('TEE attestation verification failed on server')
  }

  if (attestation.nonce !== clientNonce) {
    throw new Error('Attestation nonce mismatch - possible replay attack')
  }

  // Obtenir la clé publique du modèle pour le chiffrement
  const modelPublicKey = attestation.signing_key || attestation.signing_public_key
  if (!modelPublicKey) {
    throw new Error('No signing key in attestation response')
  }

  return {
    modelPublicKey,
    signingAddress: attestation.signing_address,
    attestation,
  }
}

import secrets
import requests

def fetch_and_verify_attestation(model_id: str, api_key: str) -> dict:
    """Fetch and verify TEE attestation for a model."""
    # Générer un nonce client pour la protection anti-rejeu (32 octets = 64 caractères hex)
    client_nonce = secrets.token_hex(32)

    response = requests.get(
        f'https://api.venice.ai/api/v1/tee/attestation',
        params={'model': model_id, 'nonce': client_nonce},
        headers={'Authorization': f'Bearer {api_key}'}
    )
    attestation = response.json()

    # Vérifier l'attestation
    if attestation.get('verified') != True:
        raise ValueError('TEE attestation verification failed on server')

    if attestation.get('nonce') != client_nonce:
        raise ValueError('Attestation nonce mismatch - possible replay attack')

    # Obtenir la clé publique du modèle pour le chiffrement
    model_public_key = attestation.get('signing_key') or attestation.get('signing_public_key')
    if not model_public_key:
        raise ValueError('No signing key in attestation response')

    return {
        'model_public_key': model_public_key,
        'signing_address': attestation.get('signing_address'),
        'attestation': attestation
    }

Étape 4 : Chiffrer les messages

Chiffrez les messages utilisateur et système avant l’envoi. Seuls les messages des rôles user et system doivent être chiffrés.

Lorsque les en-têtes E2EE sont présents, tous les messages des rôles user et system doivent être chiffrés. Envoyer du contenu en clair dans ces rôles entraînera une erreur « Encrypted field is not valid hex ».

import { gcm } from '@noble/ciphers/aes.js'
import { hkdf } from '@noble/hashes/hkdf.js'
import { sha256 } from '@noble/hashes/sha2.js'
import { ec as EC } from 'elliptic'
import crypto from 'crypto'

const HKDF_INFO = new TextEncoder().encode('ecdsa_encryption')

function encryptMessage(plaintext, modelPublicKeyHex) {
  const ec = new EC('secp256k1')

  // Normaliser la clé publique (ajouter le préfixe 04 si nécessaire)
  let normalizedKey = modelPublicKeyHex
  if (!normalizedKey.startsWith('04') && normalizedKey.length === 128) {
    normalizedKey = '04' + normalizedKey
  }

  const modelPublicKey = ec.keyFromPublic(normalizedKey, 'hex')

  // Générer une paire de clés éphémères pour ce message
  const ephemeralKeyPair = ec.genKeyPair()

  // Secret partagé ECDH
  const sharedSecret = ephemeralKeyPair.derive(modelPublicKey.getPublic())
  const sharedSecretBytes = new Uint8Array(sharedSecret.toArray('be', 32))

  // Dériver la clé AES via HKDF
  const aesKey = hkdf(sha256, sharedSecretBytes, undefined, HKDF_INFO, 32)

  // Générer un nonce aléatoire
  const nonce = crypto.randomBytes(12)

  // Chiffrer avec AES-GCM
  const cipher = gcm(aesKey, nonce)
  const encrypted = cipher.encrypt(new TextEncoder().encode(plaintext))

  // Obtenir la clé publique éphémère (non compressée)
  const ephemeralPublic = new Uint8Array(ephemeralKeyPair.getPublic(false, 'array'))

  // Combiner : ephemeral_public (65 octets) + nonce (12 octets) + ciphertext
  const result = new Uint8Array(65 + 12 + encrypted.length)
  result.set(ephemeralPublic, 0)
  result.set(nonce, 65)
  result.set(encrypted, 65 + 12)

  return Buffer.from(result).toString('hex')
}

function encryptMessagesForE2EE(messages, modelPublicKey) {
  return messages.map(msg => {
    if (msg.role === 'user' || msg.role === 'system') {
      return {
        ...msg,
        content: encryptMessage(msg.content, modelPublicKey),
      }
    }
    return msg
  })
}

from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives import hashes
from ecdsa import SECP256k1, VerifyingKey, SigningKey
import os

HKDF_INFO = b'ecdsa_encryption'

def encrypt_message(plaintext: str, model_public_key_hex: str) -> str:
    """Encrypt a message using ECDH + HKDF + AES-GCM."""
    # Normaliser la clé publique
    key_hex = model_public_key_hex
    if not key_hex.startswith('04') and len(key_hex) == 128:
        key_hex = '04' + key_hex

    model_public_key_bytes = bytes.fromhex(key_hex)

    # Parser la clé publique du modèle (sauter le préfixe 04)
    model_verifying_key = VerifyingKey.from_string(
        model_public_key_bytes[1:],
        curve=SECP256k1
    )

    # Générer une paire de clés éphémères pour ce message
    ephemeral_private = SigningKey.generate(curve=SECP256k1)
    ephemeral_public = ephemeral_private.get_verifying_key()

    # ECDH : calculer le secret partagé
    shared_point = model_verifying_key.pubkey.point * ephemeral_private.privkey.secret_multiplier
    shared_secret = shared_point.x().to_bytes(32, 'big')

    # Dériver la clé AES via HKDF
    hkdf = HKDF(
        algorithm=hashes.SHA256(),
        length=32,
        salt=None,
        info=HKDF_INFO,
    )
    aes_key = hkdf.derive(shared_secret)

    # Générer un nonce aléatoire
    nonce = os.urandom(12)

    # Chiffrer avec AES-GCM
    aesgcm = AESGCM(aes_key)
    ciphertext = aesgcm.encrypt(nonce, plaintext.encode('utf-8'), None)

    # Obtenir la clé publique éphémère (non compressée : 04 || x || y)
    ephemeral_public_bytes = b'\x04' + ephemeral_public.to_string()

    # Combiner : ephemeral_public (65 octets) + nonce (12 octets) + ciphertext
    result = ephemeral_public_bytes + nonce + ciphertext

    return result.hex()

def encrypt_messages_for_e2ee(messages: list, model_public_key: str) -> list:
    """Encrypt user and system messages."""
    encrypted_messages = []
    for msg in messages:
        if msg['role'] in ('user', 'system'):
            encrypted_messages.append({
                **msg,
                'content': encrypt_message(msg['content'], model_public_key)
            })
        else:
            encrypted_messages.append(msg)
    return encrypted_messages

Étape 5 : Envoyer la requête avec les en-têtes E2EE

Incluez les en-têtes requis pour activer le traitement E2EE.

En-tête	Description
`X-Venice-TEE-Client-Pub-Key`	Votre clé publique éphémère (hex non compressé, 130 caractères)
`X-Venice-TEE-Model-Pub-Key`	Clé publique du modèle issue de l’attestation
`X-Venice-TEE-Signing-Algo`	Toujours `ecdsa`

async function sendE2EERequest(messages, model, e2eeContext, apiKey) {
  // Chiffrer les messages
  const encryptedMessages = encryptMessagesForE2EE(messages, e2eeContext.modelPublicKey)

  const response = await fetch('https://api.venice.ai/api/v1/chat/completions', {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${apiKey}`,
      'Content-Type': 'application/json',
      // En-têtes E2EE
      'X-Venice-TEE-Client-Pub-Key': e2eeContext.publicKeyHex,
      'X-Venice-TEE-Model-Pub-Key': e2eeContext.modelPublicKey,
      'X-Venice-TEE-Signing-Algo': 'ecdsa',
    },
    body: JSON.stringify({
      model,
      messages: encryptedMessages,
      stream: true, // L'E2EE requiert le streaming
    }),
  })

  return response
}

import requests

def send_e2ee_request(
    messages: list,
    model: str,
    e2ee_context: dict,
    api_key: str
) -> requests.Response:
    """Send an E2EE-encrypted chat completion request."""
    # Chiffrer les messages
    encrypted_messages = encrypt_messages_for_e2ee(
        messages,
        e2ee_context['model_public_key']
    )

    response = requests.post(
        'https://api.venice.ai/api/v1/chat/completions',
        headers={
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json',
            # En-têtes E2EE
            'X-Venice-TEE-Client-Pub-Key': e2ee_context['public_key_hex'],
            'X-Venice-TEE-Model-Pub-Key': e2ee_context['model_public_key'],
            'X-Venice-TEE-Signing-Algo': 'ecdsa'
        },
        json={
            'model': model,
            'messages': encrypted_messages,
            'stream': True  # L'E2EE requiert le streaming
        },
        stream=True
    )

    return response

Étape 6 : Déchiffrer les chunks de réponse

Les réponses des modèles E2EE sont des chunks chiffrés encodés en hex. Déchiffrez chaque chunk avec votre clé privée.

import { gcm } from '@noble/ciphers/aes.js'
import { hkdf } from '@noble/hashes/hkdf.js'
import { sha256 } from '@noble/hashes/sha2.js'
import { ec as EC } from 'elliptic'

const HKDF_INFO = new TextEncoder().encode('ecdsa_encryption')

function hexToBytes(hex) {
  const h = hex.startsWith('0x') ? hex.slice(2) : hex
  const bytes = new Uint8Array(h.length / 2)
  for (let i = 0; i < bytes.length; i++) {
    bytes[i] = parseInt(h.substring(i * 2, i * 2 + 2), 16)
  }
  return bytes
}

function isHexEncrypted(s) {
  // Minimum : ephemeral_pub (65) + nonce (12) + tag (16) = 93 octets = 186 caractères hex
  if (s.length < 186) return false
  return /^[0-9a-fA-F]+$/.test(s)
}

function decryptChunk(ciphertextHex, clientPrivateKey) {
  const raw = hexToBytes(ciphertextHex)

  // Parser les composants
  const serverEphemeralPubKey = raw.slice(0, 65)
  const nonce = raw.slice(65, 65 + 12)
  const ciphertext = raw.slice(65 + 12)

  // ECDH avec la clé éphémère du serveur
  const ec = new EC('secp256k1')
  const clientKey = ec.keyFromPrivate(Buffer.from(clientPrivateKey))
  const serverKey = ec.keyFromPublic(Buffer.from(serverEphemeralPubKey))
  const sharedSecret = clientKey.derive(serverKey.getPublic())
  const sharedSecretBytes = new Uint8Array(sharedSecret.toArray('be', 32))

  // Dériver la clé AES
  const aesKey = hkdf(sha256, sharedSecretBytes, undefined, HKDF_INFO, 32)

  // Déchiffrer
  const cipher = gcm(aesKey, nonce)
  const plaintext = cipher.decrypt(ciphertext)

  return new TextDecoder().decode(plaintext)
}

// Traiter la réponse en streaming
async function processE2EEStream(response, clientPrivateKey) {
  const reader = response.body.getReader()
  const decoder = new TextDecoder()
  let fullContent = ''

  while (true) {
    const { done, value } = await reader.read()
    if (done) break

    const text = decoder.decode(value)
    const lines = text.split('\n')

    for (const line of lines) {
      if (!line.startsWith('data: ')) continue
      const data = line.slice(6)
      if (data === '[DONE]') continue

      try {
        const chunk = JSON.parse(data)
        const content = chunk.choices?.[0]?.delta?.content

        if (content && isHexEncrypted(content)) {
          const decrypted = decryptChunk(content, clientPrivateKey)
          fullContent += decrypted
          process.stdout.write(decrypted) // Sortie en temps réel
        } else if (content) {
          fullContent += content
          process.stdout.write(content)
        }
      } catch (e) {
        // Ignorer les chunks malformés
      }
    }
  }

  return fullContent
}

from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives import hashes
from ecdsa import SECP256k1, VerifyingKey, SigningKey
import json
import re

HKDF_INFO = b'ecdsa_encryption'

def is_hex_encrypted(s: str) -> bool:
    """Check if string looks like hex-encrypted content."""
    if len(s) < 186:  # Minimum : 65 + 12 + 16 = 93 octets = 186 hex
        return False
    return bool(re.match(r'^[0-9a-fA-F]+$', s))

def decrypt_chunk(ciphertext_hex: str, client_private_key: bytes) -> str:
    """Decrypt an E2EE response chunk."""
    raw = bytes.fromhex(ciphertext_hex)

    # Parser les composants
    server_ephemeral_pub = raw[:65]
    nonce = raw[65:77]
    ciphertext = raw[77:]

    # Parser la clé publique éphémère du serveur (sauter le préfixe 04)
    server_verifying_key = VerifyingKey.from_string(
        server_ephemeral_pub[1:],
        curve=SECP256k1
    )

    # Reconstruire la clé privée du client
    client_signing_key = SigningKey.from_string(client_private_key, curve=SECP256k1)

    # ECDH : calculer le secret partagé
    shared_point = server_verifying_key.pubkey.point * client_signing_key.privkey.secret_multiplier
    shared_secret = shared_point.x().to_bytes(32, 'big')

    # Dériver la clé AES
    hkdf = HKDF(
        algorithm=hashes.SHA256(),
        length=32,
        salt=None,
        info=HKDF_INFO,
    )
    aes_key = hkdf.derive(shared_secret)

    # Déchiffrer
    aesgcm = AESGCM(aes_key)
    plaintext = aesgcm.decrypt(nonce, ciphertext, None)

    return plaintext.decode('utf-8')

def process_e2ee_stream(response, client_private_key: bytes) -> str:
    """Process streaming E2EE response."""
    full_content = ''

    for line in response.iter_lines():
        if not line:
            continue

        line_str = line.decode('utf-8')
        if not line_str.startswith('data: '):
            continue

        data = line_str[6:]
        if data == '[DONE]':
            continue

        try:
            chunk = json.loads(data)
            content = chunk.get('choices', [{}])[0].get('delta', {}).get('content', '')

            if content and is_hex_encrypted(content):
                decrypted = decrypt_chunk(content, client_private_key)
                full_content += decrypted
                print(decrypted, end='', flush=True)  # Sortie en temps réel
            elif content:
                full_content += content
                print(content, end='', flush=True)
        except json.JSONDecodeError:
            pass

    print()  # Saut de ligne final
    return full_content

Exemple complet fonctionnel

JavaScript
Python

import elliptic from 'elliptic';
import { gcm } from '@noble/ciphers/aes.js';
import { hkdf } from '@noble/hashes/hkdf.js';
import { sha256 } from '@noble/hashes/sha2.js';
import crypto from 'crypto';

const EC = elliptic.ec;

const API_KEY = process.env.API_KEY_VENICE;
const BASE_URL = 'https://api.venice.ai/api/v1';
const MODEL = 'e2ee-qwen3-5-122b-a10b';
const HKDF_INFO = new TextEncoder().encode('ecdsa_encryption');

function hexToBytes(hex) {
  const bytes = new Uint8Array(hex.length / 2);
  for (let i = 0; i < bytes.length; i++) {
    bytes[i] = parseInt(hex.substring(i * 2, i * 2 + 2), 16);
  }
  return bytes;
}

async function main() {
  // Étape 1 : Générer une paire de clés éphémères
  console.log('🔑 Generating ephemeral key pair...');
  const ec = new EC('secp256k1');
  const keyPair = ec.genKeyPair();
  const clientPublicKeyHex = keyPair.getPublic('hex');

  // Étape 2 : Récupérer et vérifier l'attestation
  console.log('🔍 Fetching TEE attestation...');
  const clientNonce = crypto.randomBytes(32).toString('hex'); // 32 octets requis
  const attestationRes = await fetch(
    `${BASE_URL}/tee/attestation?model=${MODEL}&nonce=${clientNonce}`,
    { headers: { Authorization: `Bearer ${API_KEY}` } }
  );
  const attestation = await attestationRes.json();

  if (attestation.verified !== true || attestation.nonce !== clientNonce) {
    throw new Error('Attestation verification failed');
  }

  const modelPublicKey = attestation.signing_key || attestation.signing_public_key;
  console.log('✅ TEE attestation verified');

  // Étape 3 : Chiffrer le message
  console.log('🔐 Encrypting message...');
  const plaintext = 'What is 2+2? Answer briefly.';

  // Normaliser et parser la clé publique du modèle
  let normalizedKey = modelPublicKey;
  if (!normalizedKey.startsWith('04') && normalizedKey.length === 128) {
    normalizedKey = '04' + normalizedKey;
  }

  const modelKey = ec.keyFromPublic(normalizedKey, 'hex');
  const ephemeralKeyPair = ec.genKeyPair();
  const sharedSecret = ephemeralKeyPair.derive(modelKey.getPublic());
  const sharedSecretBytes = new Uint8Array(sharedSecret.toArray('be', 32));
  const aesKey = hkdf(sha256, sharedSecretBytes, undefined, HKDF_INFO, 32);
  const nonce = crypto.randomBytes(12);
  const cipher = gcm(aesKey, nonce);
  const encrypted = cipher.encrypt(new TextEncoder().encode(plaintext));
  const ephemeralPublic = new Uint8Array(ephemeralKeyPair.getPublic(false, 'array'));

  const result = new Uint8Array(65 + 12 + encrypted.length);
  result.set(ephemeralPublic, 0);
  result.set(nonce, 65);
  result.set(encrypted, 77);

  const encryptedContent = Buffer.from(result).toString('hex');
  const messages = [{ role: 'user', content: encryptedContent }];

  // Étape 4 : Envoyer la requête E2EE
  console.log('📤 Sending encrypted request...');
  const response = await fetch(`${BASE_URL}/chat/completions`, {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${API_KEY}`,
      'Content-Type': 'application/json',
      'X-Venice-TEE-Client-Pub-Key': clientPublicKeyHex,
      'X-Venice-TEE-Model-Pub-Key': modelPublicKey,
      'X-Venice-TEE-Signing-Algo': 'ecdsa',
    },
    body: JSON.stringify({ model: MODEL, messages, stream: true }),
  });

  // Étape 5 : Déchiffrer la réponse
  console.log('📥 Decrypting response...\n');
  const reader = response.body.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    const text = decoder.decode(value);
    for (const line of text.split('\n')) {
      if (!line.startsWith('data: ') || line.includes('[DONE]')) continue;

      try {
        const chunk = JSON.parse(line.slice(6));
        const content = chunk.choices?.[0]?.delta?.content;
        if (!content) continue;

        if (/^[0-9a-fA-F]+$/.test(content) && content.length >= 186) {
          // Déchiffrer
          const raw = hexToBytes(content);
          const serverEphemeralPub = raw.slice(0, 65);
          const nonce = raw.slice(65, 77);
          const ciphertext = raw.slice(77);

          const serverKey = ec.keyFromPublic(Buffer.from(serverEphemeralPub));
          const sharedSecret = keyPair.derive(serverKey.getPublic());
          const aesKey = hkdf(sha256, new Uint8Array(sharedSecret.toArray('be', 32)), undefined, HKDF_INFO, 32);
          const cipher = gcm(aesKey, nonce);
          const plaintext = new TextDecoder().decode(cipher.decrypt(ciphertext));
          process.stdout.write(plaintext);
        } else {
          process.stdout.write(content);
        }
      } catch {}
    }
  }

  console.log('\n\n🔐 Response decrypted end-to-end');
}

main().catch(console.error);

#!/usr/bin/env python3
"""Complete E2EE implementation example for Venice AI API."""

import os
import json
import secrets
import requests
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives import hashes
from ecdsa import SECP256k1, VerifyingKey, SigningKey

API_KEY = os.environ.get('API_KEY_VENICE')
BASE_URL = 'https://api.venice.ai/api/v1'
MODEL = 'e2ee-qwen3-5-122b-a10b'
HKDF_INFO = b'ecdsa_encryption'

def main():
    # Étape 1 : Générer une paire de clés éphémères
    print('🔑 Generating ephemeral key pair...')
    private_key = SigningKey.generate(curve=SECP256k1)
    public_key = private_key.get_verifying_key()
    client_public_key_hex = (b'\x04' + public_key.to_string()).hex()

    # Étape 2 : Récupérer et vérifier l'attestation
    print('🔍 Fetching TEE attestation...')
    client_nonce = secrets.token_hex(32)  # 32 octets requis
    attestation_res = requests.get(
        f'{BASE_URL}/tee/attestation',
        params={'model': MODEL, 'nonce': client_nonce},
        headers={'Authorization': f'Bearer {API_KEY}'},
        timeout=30
    )
    attestation = attestation_res.json()

    if attestation.get('verified') != True or attestation.get('nonce') != client_nonce:
        raise ValueError('Attestation verification failed')

    model_public_key = attestation.get('signing_key') or attestation.get('signing_public_key')
    print(f'✅ TEE attestation verified (provider: {attestation.get("tee_provider", "unknown")})')

    # Étape 3 : Chiffrer le message
    print('🔐 Encrypting message...')
    plaintext = 'What is 2+2? Answer briefly.'

    # Normaliser la clé publique
    key_hex = model_public_key
    if not key_hex.startswith('04') and len(key_hex) == 128:
        key_hex = '04' + key_hex

    model_key_bytes = bytes.fromhex(key_hex)
    model_verifying_key = VerifyingKey.from_string(model_key_bytes[1:], curve=SECP256k1)

    # ECDH
    ephemeral_private = SigningKey.generate(curve=SECP256k1)
    ephemeral_public = ephemeral_private.get_verifying_key()
    shared_point = model_verifying_key.pubkey.point * ephemeral_private.privkey.secret_multiplier
    shared_secret = shared_point.x().to_bytes(32, 'big')

    # Dériver la clé AES
    hkdf = HKDF(algorithm=hashes.SHA256(), length=32, salt=None, info=HKDF_INFO)
    aes_key = hkdf.derive(shared_secret)

    # Chiffrer
    nonce = os.urandom(12)
    aesgcm = AESGCM(aes_key)
    ciphertext = aesgcm.encrypt(nonce, plaintext.encode('utf-8'), None)

    ephemeral_public_bytes = b'\x04' + ephemeral_public.to_string()
    result = ephemeral_public_bytes + nonce + ciphertext
    encrypted_content = result.hex()

    messages = [{'role': 'user', 'content': encrypted_content}]

    # Étape 4 : Envoyer la requête E2EE
    print('📤 Sending encrypted request...')
    response = requests.post(
        f'{BASE_URL}/chat/completions',
        headers={
            'Authorization': f'Bearer {API_KEY}',
            'Content-Type': 'application/json',
            'X-Venice-TEE-Client-Pub-Key': client_public_key_hex,
            'X-Venice-TEE-Model-Pub-Key': model_public_key,
            'X-Venice-TEE-Signing-Algo': 'ecdsa'
        },
        json={'model': MODEL, 'messages': messages, 'stream': True},
        stream=True,
        timeout=60
    )

    # Étape 5 : Déchiffrer la réponse
    print('📥 Decrypting response...\n')

    for line in response.iter_lines():
        if not line:
            continue
        line_str = line.decode('utf-8')
        if not line_str.startswith('data: ') or '[DONE]' in line_str:
            continue

        try:
            chunk = json.loads(line_str[6:])
            content = chunk.get('choices', [{}])[0].get('delta', {}).get('content', '')
            if not content:
                continue

            # Vérifier si chiffré
            if len(content) >= 186 and all(c in '0123456789abcdefABCDEF' for c in content):
                raw = bytes.fromhex(content)
                server_ephemeral_pub = raw[:65]
                nonce = raw[65:77]
                ciphertext = raw[77:]

                server_verifying_key = VerifyingKey.from_string(server_ephemeral_pub[1:], curve=SECP256k1)
                shared_point = server_verifying_key.pubkey.point * private_key.privkey.secret_multiplier
                shared_secret = shared_point.x().to_bytes(32, 'big')

                hkdf = HKDF(algorithm=hashes.SHA256(), length=32, salt=None, info=HKDF_INFO)
                aes_key = hkdf.derive(shared_secret)

                aesgcm = AESGCM(aes_key)
                plaintext = aesgcm.decrypt(nonce, ciphertext, None)
                print(plaintext.decode('utf-8'), end='', flush=True)
            else:
                print(content, end='', flush=True)
        except Exception:
            pass

    print('\n\n🔐 Response decrypted end-to-end')

if __name__ == '__main__':
    main()

Limitations de l’E2EE

L’E2EE comporte certaines contraintes liées aux exigences de chiffrement :

Fonctionnalité	Statut
Streaming	Requis (non-streaming non pris en charge)
Recherche web	Désactivée (divulguerait le contenu)
Uploads de fichiers	Non pris en charge
Function calling	Non pris en charge
Prompt système Venice	Désactivé (doit être chiffré côté client)

Bonnes pratiques de sécurité

Générez de nouvelles paires de clés à chaque session — Ne réutilisez pas les clés éphémères
Effacez les clés privées par remplissage de zéros — Effacez les octets de clé privée de la mémoire une fois terminé
Vérifiez l’attestation — Vérifiez toujours verified: true et la correspondance du nonce
Cherchez le mode debug — Rejetez les attestations provenant d’enclaves en debug
Utilisez le streaming — L’E2EE requiert le streaming pour un chunking correct du chiffrement
Gérez les erreurs avec élégance — N’exposez pas les erreurs de déchiffrement aux utilisateurs
Utilisez des nonces de 32 octets — Les fournisseurs TEE exigent exactement 32 octets

Bonnes pratiques

Vérifiez toujours l'attestation en production

Ne vous contentez pas de faire confiance à la réponse verified: true. Parsez le quote Intel TDX côté client et vérifiez que les mesures correspondent aux valeurs attendues. Pour les GPU NVIDIA, vérifiez l’attestation via le service de vérification de NVIDIA.

Utilisez des nonces frais

Générez toujours un nouveau nonce aléatoire pour chaque requête d’attestation. Cela empêche les attaques par rejeu, où un attaquant pourrait servir une attestation périmée.

Vérifiez la liaison de clé

La clé de signature doit être liée au champ REPORTDATA de TDX. Cela prouve que la clé a été générée à l’intérieur de l’enclave.

Cherchez le mode debug

Vérifiez que l’attestation TDX n’a pas de flags debug activés. Une enclave en debug peut être inspectée et ne doit pas être considérée comme fiable pour la production.

Utilisez nos SDKs pour l'E2EE

L’E2EE nécessite une implémentation cryptographique minutieuse. Utilisez nos SDK officiels plutôt que d’implémenter le protocole vous-même.

Vérifier les capacités d’un modèle

Vous pouvez vérifier si un modèle prend en charge TEE ou E2EE via l’endpoint models :

curl https://api.venice.ai/api/v1/models \
  -H "Authorization: Bearer $API_KEY_VENICE" | jq '.data[] | select(.model_spec.capabilities.supportsTeeAttestation == true or .model_spec.capabilities.supportsE2EE == true) | {id, tee: .model_spec.capabilities.supportsTeeAttestation, e2ee: .model_spec.capabilities.supportsE2EE}'

models = client.models.list()

for model in models.data:
    caps = getattr(model, 'model_spec', {}).get('capabilities', {})
    if caps.get('supportsTeeAttestation') or caps.get('supportsE2EE'):
        print(f"{model.id}: TEE={caps.get('supportsTeeAttestation')}, E2EE={caps.get('supportsE2EE')}")

Gestion des erreurs

Erreur	Cause	Solution
`TEE attestation verification failed`	L’attestation n’a pas passé la validation	Réessayez ou contactez le support
`Attestation nonce mismatch`	Attaque par rejeu possible	Générez un nonce frais
`TDX debug mode detected`	L’enclave est en mode debug	Ne l’utilisez pas pour la production
`Failed to decrypt field`	Déchiffrement E2EE échoué côté serveur	Vérifiez votre implémentation de chiffrement
`E2EE requires streaming`	Requête non-streaming sur un modèle E2EE	Définissez `stream: true`
`Encrypted field is not valid hex`	Texte en clair envoyé avec des en-têtes E2EE	Chiffrez tous les messages user/system
`Invalid public key`	Mauvais format de clé	Utilisez 130 caractères hex commençant par `04`

Dépannage

502 Bad Gateway ou 'Nonce must be exactly 32 bytes'

La longueur du nonce est incorrecte. Les fournisseurs TEE requièrent exactement 32 octets (64 caractères hex).

Utilisez crypto.randomBytes(32).toString('hex') (JS) ou secrets.token_hex(32) (Python)
Erreur courante : secrets.token_hex(16) produit 32 caractères hex (16 octets), pas 32 octets

Attestation verification failed

Vérifiez que le modèle prend en charge l’E2EE (supportsE2EE: true)
Vérifiez que votre clé API est valide et a accès au modèle demandé
Vérifiez la connectivité réseau vers l’API Venice

Decryption failed

Assurez-vous d’utiliser la même clé privée qui a généré la clé publique envoyée dans les en-têtes
Vérifiez que le contenu de la réponse est bien encodé en hex (E2EE actif)
Vérifiez que la clé publique du modèle correspond à celle utilisée pour le chiffrement

Encrypted field is not valid hex

Tous les messages de rôle user et system doivent être chiffrés lorsque les en-têtes E2EE sont présents
Vérifiez que votre contenu chiffré passe la validation isValidEncrypted() (minimum 186 caractères hex)
Vérifiez que la sortie du chiffrement est en hex minuscule sans préfixe

Erreurs de clé publique invalide

La clé publique client doit faire exactement 130 caractères hex et commencer par 04
Utilisez le helper validateClientPubkey() pour vérifier le format avant l’envoi
Assurez-vous d’utiliser le format de clé publique non compressée (65 octets = 130 caractères hex)

Modèle introuvable

Vérifiez que l’ID du modèle est correct et qu’il prend en charge l’E2EE
Utilisez l’endpoint /models pour vérifier les modèles E2EE disponibles

​Comprendre les niveaux de confidentialité

​Modèles disponibles

​Modèles TEE

​Usage de base

​Vérifier l’attestation TEE

​Signatures de réponse

​Modèles E2EE

​Comment fonctionne l’E2EE

​Prérequis

​Étape 1 : Vérifier la prise en charge E2EE par le modèle

​Étape 2 : Générer une paire de clés éphémères

​Helpers de validation

​Étape 3 : Récupérer et vérifier l’attestation TEE

​Étape 4 : Chiffrer les messages

​Étape 5 : Envoyer la requête avec les en-têtes E2EE

​Étape 6 : Déchiffrer les chunks de réponse

​Exemple complet fonctionnel

​Limitations de l’E2EE

​Bonnes pratiques de sécurité

​Bonnes pratiques

​Vérifier les capacités d’un modèle

​Gestion des erreurs

​Dépannage

​Ressources

Comprendre les niveaux de confidentialité

Modèles disponibles

Modèles TEE

Usage de base

Vérifier l’attestation TEE

Signatures de réponse

Modèles E2EE

Comment fonctionne l’E2EE

Prérequis

Étape 1 : Vérifier la prise en charge E2EE par le modèle

Étape 2 : Générer une paire de clés éphémères

Helpers de validation

Étape 3 : Récupérer et vérifier l’attestation TEE

Étape 4 : Chiffrer les messages

Étape 5 : Envoyer la requête avec les en-têtes E2EE

Étape 6 : Déchiffrer les chunks de réponse

Exemple complet fonctionnel

Limitations de l’E2EE

Bonnes pratiques de sécurité

Bonnes pratiques

Vérifier les capacités d’un modèle

Gestion des erreurs

Dépannage

Ressources