웹훅

curl -X POST https://api.atlascloud.ai/api/v1/model/generateVideo \
  -H "Authorization: Bearer your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
        "model": "bytedance/seedance-2.0/text-to-video",
        "prompt": "A calico kitten chasing a butterfly in a garden, cinematic",
        "duration": 5,
        "resolution": "1080p",
        "webhook_url": "https://your-app.example.com/hooks/atlascloud"
      }'

import requests

response = requests.post(
    "https://api.atlascloud.ai/api/v1/model/generateVideo",
    headers={
        "Authorization": "Bearer your-api-key",
        "Content-Type": "application/json",
    },
    json={
        "model": "bytedance/seedance-2.0/text-to-video",
        "prompt": "A calico kitten chasing a butterfly in a garden, cinematic",
        "duration": 5,
        "resolution": "1080p",
        "webhook_url": "https://your-app.example.com/hooks/atlascloud",
    },
)

print(response.json()["data"]["id"])  # 작업 id(session_id)

const res = await fetch("https://api.atlascloud.ai/api/v1/model/generateVideo", {
  method: "POST",
  headers: {
    Authorization: "Bearer your-api-key",
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    model: "bytedance/seedance-2.0/text-to-video",
    prompt: "A calico kitten chasing a butterfly in a garden, cinematic",
    duration: 5,
    resolution: "1080p",
    webhook_url: "https://your-app.example.com/hooks/atlascloud",
  }),
});

const { data } = await res.json();
console.log(data.id); // 작업 id(session_id)

const crypto = require("crypto");

const JWKS_URL = "https://api.atlascloud.ai/api/v1/webhooks/jwks.json";
let jwks = {}; // kid -> jwk 매핑

async function publicKey(kid) {
  if (!jwks[kid]) {
    const { keys } = await (await fetch(JWKS_URL)).json();
    jwks = Object.fromEntries(keys.map((k) => [k.kid, k]));
  }
  const jwk = jwks[kid];
  return jwk && crypto.createPublicKey({ key: jwk, format: "jwk" });
}

// req.body는 반드시 원본 본문 Buffer여야 합니다(예: express.raw()).
async function verifyEd25519(req) {
  const ts = req.get("X-AtlasCloud-Webhook-Timestamp");
  const kid = req.get("X-AtlasCloud-Webhook-Key-Id");
  const sig =
    req.get("X-AtlasCloud-Webhook-Signature-Ed25519") ||
    req.get("X-AtlasCloud-Webhook-Signature");

  if (Math.abs(Date.now() / 1000 - Number(ts)) > 300) return false; // 재생 허용 시간

  const pub = await publicKey(kid);
  if (!pub) return false;
  const msg = Buffer.concat([Buffer.from(`${ts}.`), req.body]);
  return crypto.verify(null, msg, pub, Buffer.from(sig, "base64url"));
}

import time, json, base64, urllib.request
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from cryptography.exceptions import InvalidSignature

JWKS_URL = "https://api.atlascloud.ai/api/v1/webhooks/jwks.json"
_jwks = {}  # kid -> jwk 매핑

def _b64u(s: str) -> bytes:
    return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))

def _public_key(kid: str):
    if kid not in _jwks:
        keys = json.load(urllib.request.urlopen(JWKS_URL, timeout=5))["keys"]
        _jwks.clear()
        _jwks.update({k["kid"]: k for k in keys})
    jwk = _jwks.get(kid)
    return Ed25519PublicKey.from_public_bytes(_b64u(jwk["x"])) if jwk else None

def verify_ed25519(headers, raw_body: bytes) -> bool:
    ts = headers["X-AtlasCloud-Webhook-Timestamp"]
    kid = headers["X-AtlasCloud-Webhook-Key-Id"]
    sig = headers.get("X-AtlasCloud-Webhook-Signature-Ed25519") \
        or headers["X-AtlasCloud-Webhook-Signature"]

    if abs(time.time() - int(ts)) > 300:  # 재생 허용 시간
        return False
    pub = _public_key(kid)
    if pub is None:
        return False
    try:
        pub.verify(_b64u(sig), ts.encode() + b"." + raw_body)
        return True
    except InvalidSignature:
        return False

const jwksURL = "https://api.atlascloud.ai/api/v1/webhooks/jwks.json"

// fetchKey는 kid에 해당하는 base64url 공개 키를 반환합니다. 실제 코드에서는
// 캐싱과 미스 시 1회 재조회(키 교체 대응)를 추가하세요.
func fetchKey(kid string) (string, bool) {
    resp, err := http.Get(jwksURL)
    if err != nil {
        return "", false
    }
    defer resp.Body.Close()
    var set struct {
        Keys []struct{ Kid, X string } `json:"keys"`
    }
    if json.NewDecoder(resp.Body).Decode(&set) != nil {
        return "", false
    }
    for _, k := range set.Keys {
        if k.Kid == kid {
            return k.X, true
        }
    }
    return "", false
}

func verifyEd25519(h http.Header, body []byte) bool {
    ts := h.Get("X-AtlasCloud-Webhook-Timestamp")
    sigB64 := h.Get("X-AtlasCloud-Webhook-Signature-Ed25519")
    if sigB64 == "" {
        sigB64 = h.Get("X-AtlasCloud-Webhook-Signature")
    }
    t, _ := strconv.ParseInt(ts, 10, 64)
    if math.Abs(float64(time.Now().Unix()-t)) > 300 { // 재생 허용 시간
        return false
    }
    xB64, ok := fetchKey(h.Get("X-AtlasCloud-Webhook-Key-Id"))
    if !ok {
        return false
    }
    pub, e1 := base64.RawURLEncoding.DecodeString(xB64)
    sig, e2 := base64.RawURLEncoding.DecodeString(sigB64)
    if e1 != nil || e2 != nil || len(pub) != ed25519.PublicKeySize {
        return false
    }
    return ed25519.Verify(ed25519.PublicKey(pub), append([]byte(ts+"."), body...), sig)
}

const crypto = require("crypto");
const express = require("express");
const app = express();

const SIGNING_SECRET = process.env.ATLASCLOUD_WEBHOOK_SECRET;

// 원본 본문을 확보하세요 — JSON 파서가 먼저 실행되지 않도록 합니다.
app.use("/hooks/atlascloud", express.raw({ type: "*/*" }));

app.post("/hooks/atlascloud", (req, res) => {
  const sig = req.get("X-AtlasCloud-Webhook-Signature") || "";
  const expected = crypto
    .createHmac("sha256", SIGNING_SECRET)
    .update(req.body)               // req.body는 Buffer(원본 바이트)입니다
    .digest("hex");

  // Buffer로 비교하고 바이트 길이를 확인하세요. timingSafeEqual은 길이가 다르면
  // 예외를 던지며, 잘못된 멀티바이트 헤더는 JS 문자열 길이는 같아도
  // 바이트 길이가 다를 수 있습니다.
  const sigBuf = Buffer.from(sig);
  const expBuf = Buffer.from(expected);
  const ok =
    sigBuf.length === expBuf.length &&
    crypto.timingSafeEqual(sigBuf, expBuf);
  if (!ok) return res.status(401).send("invalid signature");

  const event = JSON.parse(req.body.toString("utf8"));
  // ... event.session_id 기준으로 큐에 넣고, 즉시 응답합니다 ...
  res.status(200).send("ok");
});

import hmac, hashlib, os
from flask import Flask, request, abort

SIGNING_SECRET = os.environ["ATLASCLOUD_WEBHOOK_SECRET"].encode()
app = Flask(__name__)

@app.post("/hooks/atlascloud")
def atlascloud_webhook():
    raw = request.get_data()  # JSON 파싱 이전의 원본 바이트
    expected = hmac.new(SIGNING_SECRET, raw, hashlib.sha256).hexdigest()
    received = request.headers.get("X-AtlasCloud-Webhook-Signature", "")
    # 바이트로 비교하세요. compare_digest는 비 ASCII 문자열에서 예외를 던집니다.
    if not hmac.compare_digest(expected.encode(), received.encode()):
        abort(401)

    event = request.get_json()
    # ... event["session_id"] 기준으로 큐에 넣고, 200을 빠르게 반환합니다 ...
    return "ok", 200

func verify(secret string, body []byte, sigHeader string) bool {
    mac := hmac.New(sha256.New, []byte(secret))
    mac.Write(body)
    expected := hex.EncodeToString(mac.Sum(nil))
    return hmac.Equal([]byte(expected), []byte(sigHeader))
}