ModelRouter/internal/core/channels_codearts_oauth_test.go

package core

import (
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/sha256"
	"encoding/base64"
	"encoding/json"
	"math/big"
	"strings"
	"testing"
)

// CodeArts 的 DPoP 证明是标准 ES256 JWS。用 Go 标准库（crypto/ecdsa）当权威
// 校验器验它 —— 只断言"能拼出字符串"是不够的：JWS 的 ES256 签名必须是
// **raw r||s（各 32 字节）**，而 openssl 默认输出 DER；写错格式服务端会拒，
// 而字符串看起来完全正常（这是最容易蒙混过去的一类缺陷）。
func TestCodeArtsDpopJwsVerifiesAsES256(t *testing.T) {
	c := kernelTestCore(t)
	ps := c.Plugins()
	src, err := readBundledChannels(t)
	if err != nil {
		t.Fatal(err)
	}
	if err := ps.LoadSource("channels", src); err != nil {
		t.Fatal(err)
	}

	got, err := ps.Call("channels", "__codearts_dpop_probe",
		map[string]interface{}{"htm": "POST", "htu": "https://sts.cn-north-4.myhuaweicloud.com/v1/oauth2/tokens"})
	if err != nil {
		t.Fatalf("probe: %v", err)
	}
	w, _ := got.(map[string]interface{})
	m, _ := w["json"].(map[string]interface{})
	if m == nil {
		t.Fatalf("无 json: %#v", got)
	}
	if m["error"] != nil {
		t.Fatalf("探针报错: %v", m["error"])
	}
	jws, _ := m["jws"].(string)
	if jws == "" {
		t.Fatal("jws 为空")
	}
	parts := strings.Split(jws, ".")
	if len(parts) != 3 {
		t.Fatalf("JWS 应为三段，实得 %d 段", len(parts))
	}
	dec := func(s string) []byte {
		b, err := base64.RawURLEncoding.DecodeString(s)
		if err != nil {
			t.Fatalf("base64url 解码失败: %v", err)
		}
		return b
	}
	// ---- header ----
	var hdr map[string]interface{}
	if err := json.Unmarshal(dec(parts[0]), &hdr); err != nil {
		t.Fatalf("header 不是 JSON: %v", err)
	}
	if hdr["alg"] != "ES256" {
		t.Errorf("alg 应为 ES256，实得 %v", hdr["alg"])
	}
	if hdr["typ"] != "dpop+jwt" {
		t.Errorf("typ 应为 dpop+jwt，实得 %v", hdr["typ"])
	}
	jwk, _ := hdr["jwk"].(map[string]interface{})
	if jwk == nil {
		t.Fatal("header 必须内嵌 jwk（DPoP 要求公钥随证明走）")
	}
	// ---- 用 jwk 重建公钥 ----
	xb, err := base64.RawURLEncoding.DecodeString(jwk["x"].(string))
	if err != nil {
		t.Fatal(err)
	}
	yb, err := base64.RawURLEncoding.DecodeString(jwk["y"].(string))
	if err != nil {
		t.Fatal(err)
	}
	if len(xb) != 32 || len(yb) != 32 {
		t.Fatalf("P-256 的 x/y 应各 32 字节，实得 %d/%d", len(xb), len(yb))
	}
	pub := &ecdsa.PublicKey{Curve: elliptic.P256(),
		X: new(big.Int).SetBytes(xb), Y: new(big.Int).SetBytes(yb)}

	// ---- payload ----
	var pl map[string]interface{}
	if err := json.Unmarshal(dec(parts[1]), &pl); err != nil {
		t.Fatalf("payload 不是 JSON: %v", err)
	}
	for _, k := range []string{"htm", "htu", "iat", "jti"} {
		if _, ok := pl[k]; !ok {
			t.Errorf("payload 缺 %s（照参考 signDpopJws）", k)
		}
	}
	if pl["htm"] != "POST" {
		t.Errorf("htm = %v", pl["htm"])
	}

	// ---- 验签（关键：raw r||s，不是 DER）----
	sig := dec(parts[2])
	if len(sig) != 64 {
		t.Fatalf("ES256 签名必须是 64 字节 raw r||s，实得 %d 字节（若是 70-72 字节，说明漏了 DER→raw 转换）", len(sig))
	}
	r := new(big.Int).SetBytes(sig[:32])
	s := new(big.Int).SetBytes(sig[32:])
	digest := sha256.Sum256([]byte(parts[0] + "." + parts[1]))
	if !ecdsa.Verify(pub, digest[:], r, s) {
		t.Fatal("ES256 验签失败 —— 签名内容或格式不对，服务端会拒绝这个 DPoP")
	}
}