Webhook

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 公钥。实际代码中请加上缓存，
// 以及未命中时重新获取一次（应对密钥轮换）。
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", "")
    # 以字节比较：对 str 使用 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))
}