TEE 및 E2EE 모델

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"}]
  }'

# 무작위 nonce 생성(재전송 공격 방지)
NONCE=$(openssl rand -hex 16)

# 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']}")

# completion을 받은 후 서명을 검증
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()
# 서명이 attestation의 signing_address와 일치하는지 검증

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)
}

// 사용 예시
const models = await getE2EEModels('your-api-key')
console.log('E2EE Models:', models.map(m => m.id))
// 출력: ['e2ee-qwen3-5-122b-a10b', 'e2ee-glm-5', ...]

import requests

def get_e2ee_models(api_key: str) -> list:
    """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')
    ]

# 사용 예시
models = get_e2ee_models('your-api-key')
print('E2EE Models:', [m['id'] for m in models])

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'), // 비압축 포맷 (65바이트 hex)
  }
}

// 보안: 사용 종료 후 private key를 0으로 채우기
function zeroFill(arr) {
  arr.fill(0)
}

from ecdsa import SECP256k1, SigningKey
import secrets

def generate_ephemeral_key_pair():
    """E2EE 세션을 위한 임시 secp256k1 키 쌍을 생성합니다."""
    private_key = SigningKey.generate(curve=SECP256k1)
    public_key = private_key.get_verifying_key()

    # 비압축 공개 키 가져오기 (04 || x || y)
    public_key_bytes = b'\x04' + public_key.to_string()

    return {
        'private_key': private_key.to_string(),  # 32 바이트
        'public_key_hex': public_key_bytes.hex()  # 130 hex 문자
    }

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) {
  // 최소: ephemeral_pub (65) + nonce (12) + tag (16) = 93 바이트 = 186 hex 문자
  return s.length >= 186 && /^[0-9a-fA-F]+$/.test(s)
}

def validate_client_pubkey(pubkey_hex: str) -> None:
    """클라이언트 공개 키 포맷 검증."""
    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:
    """유효한 hex 암호 콘텐츠인지 확인."""
    # 최소: ephemeral_pub (65) + nonce (12) + tag (16) = 93 바이트 = 186 hex 문자
    return len(s) >= 186 and all(c in '0123456789abcdefABCDEF' for c in s)

import crypto from 'crypto'

async function fetchAndVerifyAttestation(modelId, apiKey) {
  // 재전송 방지를 위한 클라이언트 nonce 생성 (32바이트 = 64 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()

  // 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')
  }

  // 암호화용 모델 공개 키 가져오기
  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:
    """모델에 대한 TEE attestation을 가져오고 검증합니다."""
    # 재전송 방지를 위한 클라이언트 nonce 생성 (32바이트 = 64 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()

    # 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')

    # 암호화용 모델 공개 키 가져오기
    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
    }

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')

  // 공개 키 정규화 (필요 시 04 prefix 추가)
  let normalizedKey = modelPublicKeyHex
  if (!normalizedKey.startsWith('04') && normalizedKey.length === 128) {
    normalizedKey = '04' + normalizedKey
  }

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

  // 이 메시지를 위한 임시 키 쌍 생성
  const ephemeralKeyPair = ec.genKeyPair()

  // ECDH 공유 비밀
  const sharedSecret = ephemeralKeyPair.derive(modelPublicKey.getPublic())
  const sharedSecretBytes = new Uint8Array(sharedSecret.toArray('be', 32))

  // HKDF로 AES 키 파생
  const aesKey = hkdf(sha256, sharedSecretBytes, undefined, HKDF_INFO, 32)

  // 무작위 nonce 생성
  const nonce = crypto.randomBytes(12)

  // AES-GCM 으로 암호화
  const cipher = gcm(aesKey, nonce)
  const encrypted = cipher.encrypt(new TextEncoder().encode(plaintext))

  // 임시 공개 키 가져오기 (비압축)
  const ephemeralPublic = new Uint8Array(ephemeralKeyPair.getPublic(false, 'array'))

  // 결합: ephemeral_public (65바이트) + nonce (12바이트) + 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:
    """ECDH + HKDF + AES-GCM을 사용해 메시지를 암호화."""
    # 공개 키 정규화
    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)

    # 모델의 공개 키 파싱 (04 prefix 생략)
    model_verifying_key = VerifyingKey.from_string(
        model_public_key_bytes[1:],
        curve=SECP256k1
    )

    # 이 메시지를 위한 임시 키 쌍 생성
    ephemeral_private = SigningKey.generate(curve=SECP256k1)
    ephemeral_public = ephemeral_private.get_verifying_key()

    # ECDH: 공유 비밀 계산
    shared_point = model_verifying_key.pubkey.point * ephemeral_private.privkey.secret_multiplier
    shared_secret = shared_point.x().to_bytes(32, 'big')

    # HKDF로 AES 키 파생
    hkdf = HKDF(
        algorithm=hashes.SHA256(),
        length=32,
        salt=None,
        info=HKDF_INFO,
    )
    aes_key = hkdf.derive(shared_secret)

    # 무작위 nonce 생성
    nonce = os.urandom(12)

    # AES-GCM 으로 암호화
    aesgcm = AESGCM(aes_key)
    ciphertext = aesgcm.encrypt(nonce, plaintext.encode('utf-8'), None)

    # 임시 공개 키 가져오기 (비압축: 04 || x || y)
    ephemeral_public_bytes = b'\x04' + ephemeral_public.to_string()

    # 결합: ephemeral_public (65바이트) + nonce (12바이트) + ciphertext
    result = ephemeral_public_bytes + nonce + ciphertext

    return result.hex()

def encrypt_messages_for_e2ee(messages: list, model_public_key: str) -> list:
    """user 및 system 메시지를 암호화."""
    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

async function sendE2EERequest(messages, model, e2eeContext, apiKey) {
  // 메시지 암호화
  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',
      // 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, // E2EE는 스트리밍이 필요
    }),
  })

  return response
}

import requests

def send_e2ee_request(
    messages: list,
    model: str,
    e2ee_context: dict,
    api_key: str
) -> requests.Response:
    """E2EE 암호화 chat completion 요청을 전송."""
    # 메시지 암호화
    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',
            # 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  # E2EE는 스트리밍이 필요
        },
        stream=True
    )

    return response

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) {
  // 최소: ephemeral_pub (65) + nonce (12) + tag (16) = 93 바이트 = 186 hex 문자
  if (s.length < 186) return false
  return /^[0-9a-fA-F]+$/.test(s)
}

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

  // 구성요소 파싱
  const serverEphemeralPubKey = raw.slice(0, 65)
  const nonce = raw.slice(65, 65 + 12)
  const ciphertext = raw.slice(65 + 12)

  // 서버 임시 키와 ECDH
  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))

  // AES 키 파생
  const aesKey = hkdf(sha256, sharedSecretBytes, undefined, HKDF_INFO, 32)

  // 복호화
  const cipher = gcm(aesKey, nonce)
  const plaintext = cipher.decrypt(ciphertext)

  return new TextDecoder().decode(plaintext)
}

// 스트리밍 응답 처리
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) // 실시간 출력
        } else if (content) {
          fullContent += content
          process.stdout.write(content)
        }
      } catch (e) {
        // 비정상 청크 건너뛰기
      }
    }
  }

  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:
    """hex 암호화된 콘텐츠처럼 보이는지 확인."""
    if len(s) < 186:  # 최소: 65 + 12 + 16 = 93 바이트 = 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:
    """E2EE 응답 청크를 복호화."""
    raw = bytes.fromhex(ciphertext_hex)

    # 구성요소 파싱
    server_ephemeral_pub = raw[:65]
    nonce = raw[65:77]
    ciphertext = raw[77:]

    # 서버 임시 공개 키 파싱 (04 prefix 생략)
    server_verifying_key = VerifyingKey.from_string(
        server_ephemeral_pub[1:],
        curve=SECP256k1
    )

    # 클라이언트의 private key 재구성
    client_signing_key = SigningKey.from_string(client_private_key, curve=SECP256k1)

    # ECDH: 공유 비밀 계산
    shared_point = server_verifying_key.pubkey.point * client_signing_key.privkey.secret_multiplier
    shared_secret = shared_point.x().to_bytes(32, 'big')

    # AES 키 파생
    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)

    return plaintext.decode('utf-8')

def process_e2ee_stream(response, client_private_key: bytes) -> str:
    """스트리밍 E2EE 응답을 처리."""
    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)  # 실시간 출력
            elif content:
                full_content += content
                print(content, end='', flush=True)
        except json.JSONDecodeError:
            pass

    print()  # 마지막 줄바꿈
    return full_content

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')}")