diff -ruN a/controller/relay.go b/controller/relay.go --- a/controller/relay.go +++ b/controller/relay.go @@ -134,8 +134,16 @@ relayInfo, err := relaycommon.GenRelayInfo(c, relayFormat, request, ws) if err != nil { newAPIError = types.NewError(err, types.ErrorCodeGenRelayInfoFailed) + return + } + + // 6bee: per-account output cap and concurrency slot (service/sixbee_limits.go) + releaseSlot, limitErr := service.SixbeeAdmitRequest(c, relayInfo) + if limitErr != nil { + newAPIError = limitErr return } + defer releaseSlot() defer func() { recovered := recover() diff -ruN a/controller/sixbee_epay.go b/controller/sixbee_epay.go --- a/controller/sixbee_epay.go +++ b/controller/sixbee_epay.go @@ -0,0 +1,99 @@ +package controller + +// 6bee: adapter for the owner's private Epay gateway. +// +// The private gateway rejects orders without a signed 10-digit Unix +// `timestamp` (it allows 300 s of clock skew), which go-epay's Purchase does +// not send. Fields and MD5 signing are otherwise the same as go-epay, except +// that `device` is left out so the gateway picks PC or mobile checkout from +// the customer's own browser (the form is posted by that browser). +// +// sixbeeEpayNotifyMatches adds two checks before a signed notify credits a +// balance: the notify is for our PID, and its amount equals the amount the +// order was created with. +// +// With SIXBEE_PAY_ORIGIN set, orders use the gateway's same-domain checkout +// instead of the browser form (controller/sixbee_pay.go), so customers only +// ever see pay.6bee.ai and the payment provider's own pages. + +import ( + "errors" + "fmt" + "path" + "strconv" + "time" + + "github.com/QuantumNous/new-api/model" + "github.com/QuantumNous/new-api/setting/operation_setting" + + "github.com/Calcium-Ion/go-epay/epay" + "github.com/gin-gonic/gin" + "github.com/shopspring/decimal" +) + +// sixbeeEpayPurchase places an Epay order. It returns the URL the console +// posts the customer's browser to and the form fields (none in site mode). +func sixbeeEpayPurchase(c *gin.Context, client *epay.Client, args *epay.PurchaseArgs) (string, map[string]string, error) { + origin, err := sixbeePayOrigin() + if err != nil { + return "", nil, err + } + if origin == nil { + return sixbeeEpayRedirectPurchase(client, args) + } + return sixbeeSitePurchase(c, client, args, origin) +} + +// sixbeeEpayRedirectPurchase is the browser-form order (the gateway's own pages). +func sixbeeEpayRedirectPurchase(client *epay.Client, args *epay.PurchaseArgs) (string, map[string]string, error) { + if client == nil || client.Config == nil || client.BaseUrl == nil || args == nil || + args.NotifyUrl == nil || args.ReturnUrl == nil { + return "", nil, errors.New("epay: incomplete purchase arguments") + } + params := map[string]string{ + "pid": client.Config.PartnerID, + "type": args.Type, + "out_trade_no": args.ServiceTradeNo, + "notify_url": args.NotifyUrl.String(), + "return_url": args.ReturnUrl.String(), + "name": args.Name, + "money": args.Money, + "timestamp": strconv.FormatInt(time.Now().Unix(), 10), + } + u := *client.BaseUrl + u.Path = path.Join(u.Path, epay.PurchaseUrl) + signed, err := epay.GenerateParamsWithError(params, client.Config.Key) + if err != nil { + return "", nil, err + } + return u.String(), signed, nil +} + +// sixbeeEpayNotifyMatches is called after the signature has been verified. +// orderMoney is the amount stored on the local order (0 when the order is not +// found; the caller's normal not-found handling then applies). +func sixbeeEpayNotifyMatches(params map[string]string, orderMoney float64, orderFound bool) error { + if params["pid"] != operation_setting.EpayId { + return fmt.Errorf("pid %q is not ours", params["pid"]) + } + if !orderFound { + return nil + } + paid, err := decimal.NewFromString(params["money"]) + if err != nil { + return fmt.Errorf("invalid money %q", params["money"]) + } + want := decimal.NewFromFloat(orderMoney).Round(2) + if !paid.Round(2).Equal(want) { + return fmt.Errorf("money %s does not match the order amount %s", paid.String(), want.StringFixed(2)) + } + return nil +} + +func sixbeeTopUpMoney(tradeNo string) (float64, bool) { + topUp := model.GetTopUpByTradeNo(tradeNo) + if topUp == nil { + return 0, false + } + return topUp.Money, true +} diff -ruN a/controller/sixbee_epay_test.go b/controller/sixbee_epay_test.go --- a/controller/sixbee_epay_test.go +++ b/controller/sixbee_epay_test.go @@ -0,0 +1,75 @@ +package controller + +import ( + "crypto/md5" + "fmt" + "net/url" + "strconv" + "testing" + "time" + + "github.com/QuantumNous/new-api/setting/operation_setting" + + "github.com/Calcium-Ion/go-epay/epay" +) + +func TestSixbeeEpayPurchaseSignsTimestamp(t *testing.T) { + key := "0123456789abcdef0123456789abcdef" + client, err := epay.NewClient(&epay.Config{PartnerID: "1008", Key: key}, "https://pay.example.net/") + if err != nil { + t.Fatal(err) + } + notify, _ := url.Parse("https://console.example.com/api/user/epay/notify") + ret, _ := url.Parse("https://console.example.com/console/log") + uri, params, err := sixbeeEpayRedirectPurchase(client, &epay.PurchaseArgs{ + Type: "usdc", ServiceTradeNo: "USR1NOabc1234567890", Name: "TUC10", Money: "10.00", + Device: epay.PC, NotifyUrl: notify, ReturnUrl: ret, + }) + if err != nil { + t.Fatal(err) + } + if uri != "https://pay.example.net/submit.php" { + t.Fatalf("uri = %s", uri) + } + ts, err := strconv.ParseInt(params["timestamp"], 10, 64) + if err != nil || len(params["timestamp"]) != 10 || time.Now().Unix()-ts > 5 || ts-time.Now().Unix() > 5 { + t.Fatalf("timestamp = %q", params["timestamp"]) + } + if _, ok := params["device"]; ok { + t.Fatal("device must be left to the gateway") + } + // The same string the PHP gateway signs: sorted non-empty fields without sign/sign_type, then the key. + want := fmt.Sprintf("%x", md5.Sum([]byte( + "money=10.00&name=TUC10¬ify_url="+notify.String()+"&out_trade_no=USR1NOabc1234567890&pid=1008&return_url="+ret.String()+ + "×tamp="+params["timestamp"]+"&type=usdc"+key))) + if params["sign"] != want || params["sign_type"] != "MD5" { + t.Fatalf("sign = %s, want %s", params["sign"], want) + } +} + +func TestSixbeeEpayNotifyMatches(t *testing.T) { + old := operation_setting.EpayId + operation_setting.EpayId = "1008" + defer func() { operation_setting.EpayId = old }() + + cases := []struct { + name string + params map[string]string + money float64 + found bool + ok bool + }{ + {"php float money", map[string]string{"pid": "1008", "money": "10"}, 10, true, true}, + {"two decimals", map[string]string{"pid": "1008", "money": "9.50"}, 9.5, true, true}, + {"wrong pid", map[string]string{"pid": "1003", "money": "10"}, 10, true, false}, + {"lower amount", map[string]string{"pid": "1008", "money": "1"}, 10, true, false}, + {"garbage amount", map[string]string{"pid": "1008", "money": "ten"}, 10, true, false}, + {"order not found is left to RechargeEpay", map[string]string{"pid": "1008", "money": "1"}, 0, false, true}, + } + for _, tc := range cases { + err := sixbeeEpayNotifyMatches(tc.params, tc.money, tc.found) + if (err == nil) != tc.ok { + t.Errorf("%s: err = %v", tc.name, err) + } + } +} diff -ruN a/controller/sixbee_notify.go b/controller/sixbee_notify.go --- a/controller/sixbee_notify.go +++ b/controller/sixbee_notify.go @@ -0,0 +1,156 @@ +package controller + +// 6bee patch: handlers for Telegram linking, the unsubscribe link in broadcast emails and the +// owner's broadcasts (service/sixbee_notify.go, service/sixbee_broadcast.go). + +import ( + "html" + "net/http" + "strconv" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/service" + + "github.com/gin-gonic/gin" +) + +// SixbeeTelegramStatus tells the console whether Telegram notices are available and which chat, +// if any, the signed-in account has linked. +func SixbeeTelegramStatus(c *gin.Context) { + bot := service.SixbeeTelegramBot() + data := gin.H{"available": bot != "", "bot_username": bot, "linked": false} + if link, linked := service.SixbeeTelegramLinkOf(c.GetInt("id")); linked { + data["linked"] = true + data["username"] = link.Username + data["name"] = link.Name + data["linked_at"] = link.CreatedAt + } + common.ApiSuccess(c, data) +} + +// SixbeeTelegramLinkStart issues the one-time t.me link the user opens to link a chat. +func SixbeeTelegramLinkStart(c *gin.Context) { + link, err := service.SixbeeTelegramLinkStart(c.GetInt("id")) + if err != nil { + common.ApiError(c, err) + return + } + common.ApiSuccess(c, gin.H{"url": link, "expires_in": 600}) +} + +func SixbeeTelegramUnlink(c *gin.Context) { + if err := service.SixbeeTelegramUnlink(c.GetInt("id")); err != nil { + common.ApiError(c, err) + return + } + common.ApiSuccess(c, nil) +} + +// SixbeeUnsubscribe is the link at the foot of broadcast emails. GET shows a confirmation page +// (mail scanners that open links must not unsubscribe anyone); POST turns the kind off. +func SixbeeUnsubscribe(c *gin.Context) { + userId, _ := strconv.Atoi(c.Query("u")) + kind, signature := c.Query("k"), c.Query("s") + title, body := "6bee.ai", "" + status := http.StatusOK + switch { + case !service.SixbeeUnsubscribeValid(userId, kind, signature): + status = http.StatusNotFound + body = `

这个退订链接无效。
This unsubscribe link is not valid.

` + case c.Request.Method == http.MethodPost: + if err := service.SixbeeSetNoticePreference(userId, kind, false); err != nil { + common.SysError("6bee: unsubscribe failed: " + err.Error()) + status = http.StatusInternalServerError + body = `

退订失败，请稍后再试。
Unsubscribing failed. Please try again later.

` + break + } + body = `

已退订，你不会再收到这类通知（邮箱和 Telegram）。
You are unsubscribed and will no longer receive these notices, by email or Telegram.

+

想重新接收，可在控制台「个人设置 → 通知」打开。
To receive them again, turn them on under Profile → Notifications in the console.

` + default: + what := "公告 · announcements" + if kind == service.SixbeeBroadcastPromo { + what = "活动通知 · promotions" + } + body = `

不再接收 6bee.ai 的` + html.EscapeString(what) + `？
Stop receiving them from 6bee.ai?

+
+

余额提醒不受影响。
Low balance reminders are not affected.

` + } + page := ` +` + title + `

6bee.ai

` + body + `
` + c.Header("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; form-action 'self'; base-uri 'none'; frame-ancestors 'none'") + c.Header("Referrer-Policy", "no-referrer") + c.Header("X-Content-Type-Options", "nosniff") + c.Data(status, "text/html; charset=utf-8", []byte(page)) +} + +type sixbeeBroadcastRequest struct { + Kind string `json:"kind"` + Title string `json:"title"` + Content string `json:"content"` + LinkUrl string `json:"link_url"` + Test bool `json:"test"` +} + +// SixbeeBroadcastList returns the latest broadcasts with their progress, and how many accounts +// each kind would reach now. +func SixbeeBroadcastList(c *gin.Context) { + broadcasts, err := service.SixbeeBroadcastList() + if err != nil { + common.ApiError(c, err) + return + } + audience, err := service.SixbeeBroadcastAudienceOf() + if err != nil { + common.ApiError(c, err) + return + } + common.ApiSuccess(c, gin.H{"items": broadcasts, "audience": audience, "telegram_bot": service.SixbeeTelegramBot()}) +} + +// SixbeeBroadcastSend sends a test to the author (test: true) or starts sending to everyone. +func SixbeeBroadcastSend(c *gin.Context) { + var req sixbeeBroadcastRequest + if err := c.ShouldBindJSON(&req); err != nil { + common.ApiErrorT(c, "Invalid parameters") + return + } + broadcast := service.SixbeeBroadcast{Kind: req.Kind, Title: req.Title, Content: req.Content, LinkUrl: req.LinkUrl, CreatedBy: c.GetInt("id")} + if req.Test { + emailSent, telegramSent, err := service.SixbeeBroadcastTest(&broadcast, c.GetInt("id")) + if err != nil { + common.ApiError(c, err) + return + } + common.ApiSuccess(c, gin.H{"email": emailSent, "telegram": telegramSent}) + return + } + if err := service.SixbeeBroadcastStart(&broadcast); err != nil { + common.ApiError(c, err) + return + } + common.ApiSuccess(c, broadcast) +} + +func SixbeeBroadcastResume(c *gin.Context) { + id, _ := strconv.Atoi(c.Param("id")) + if err := service.SixbeeBroadcastResume(id); err != nil { + common.ApiError(c, err) + return + } + common.ApiSuccess(c, nil) +} + +func SixbeeBroadcastStop(c *gin.Context) { + id, _ := strconv.Atoi(c.Param("id")) + if err := service.SixbeeBroadcastStop(id); err != nil { + common.ApiError(c, err) + return + } + common.ApiSuccess(c, nil) +} diff -ruN a/controller/sixbee_pay.go b/controller/sixbee_pay.go --- a/controller/sixbee_pay.go +++ b/controller/sixbee_pay.go @@ -0,0 +1,670 @@ +package controller + +// 6bee: same-domain checkout for the owner's private Epay gateway. +// +// With SIXBEE_PAY_ORIGIN set (https://pay.6bee.ai), an Epay order is placed +// server to server: POST mapi.php with checkout_mode=site. The gateway's +// answer is checked against the site key and only its display fields are +// kept, in Redis under a random token: an Alipay QR code or official Alipay +// WAP link, or a USDT/USDC address with the exact amount and expiry. The +// console sends the customer to https://pay.6bee.ai/pay/, rendered +// here, so the browser never receives the gateway's or the wallet service's +// domain. +// +// Alipay and the wallet service post their asynchronous notifications to +// https://pay.6bee.ai/.well-known/epay-provider/{type}/{trade_no}. That +// route forwards the untouched request to the gateway, which verifies it and +// then notifies /api/user/epay/notify on pay.6bee.ai as before. The relay +// never credits anything itself. + +import ( + "bytes" + "context" + "crypto/md5" + "crypto/rand" + "crypto/subtle" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "html/template" + "io" + "net" + "net/http" + "net/url" + "os" + "path" + "regexp" + "sort" + "strconv" + "strings" + "time" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/logger" + "github.com/QuantumNous/new-api/model" + + "github.com/Calcium-Ion/go-epay/epay" + "github.com/boombuler/barcode/qr" + "github.com/gin-gonic/gin" + "github.com/shopspring/decimal" +) + +const ( + sixbeePayCurrency = "USD" // the gateway site for 6bee.ai is priced in USD (Price = 1) + sixbeePayRecordPrefix = "sixbee:pay:" + sixbeePayRecordTTL = 24 * time.Hour + sixbeeAlipayQRLife = 10 * time.Minute // the gateway's Alipay precreate timeout_express + sixbeeAlipayWapLife = 30 * time.Minute + sixbeeGatewayMaxBody = 64 << 10 +) + +var ( + sixbeePayTokenRe = regexp.MustCompile(`^[A-Za-z0-9_-]{22}$`) + sixbeeGatewayTradeRe = regexp.MustCompile(`^[0-9]{19}$`) + sixbeeCryptoAmountRe = regexp.MustCompile(`^(0|[1-9][0-9]{0,11})(\.[0-9]{1,8})?$`) + sixbeeTronAddressRe = regexp.MustCompile(`^T[1-9A-HJ-NP-Za-km-z]{33}$`) + sixbeeSolAddressRe = regexp.MustCompile(`^[1-9A-HJ-NP-Za-km-z]{32,44}$`) + + // USDT and USDC are charged 1:1 with the USD price. The only difference + // allowed is the gateway's collision tail: when another open order on the + // same address has the same amount it adds 0.01 per clash (5.00, 5.01, ...). + sixbeeCryptoCent = decimal.New(1, -2) + sixbeeCryptoTailMax = decimal.New(10, -2) + sixbeeMobileUARe = regexp.MustCompile(`(?i)android|iphone|ipod|mobile|harmonyos|windows phone`) + + // Provider types the relay forwards; the gateway only uses these for 6bee.ai. + sixbeeRelayTypes = map[string]bool{"alipay": true, "usdt": true, "usdc": true} + sixbeeRelayHeaders = []string{ + "Content-Type", + "Wechatpay-Timestamp", "Wechatpay-Nonce", "Wechatpay-Serial", "Wechatpay-Signature", + "Request-Time", "Client-Id", "Signature", + } + // Crypto tokens: payment type → token and network the gateway must answer with. + sixbeeCryptoAssets = map[string][2]string{"usdt": {"USDT", "TRON"}, "usdc": {"USDC", "SOLANA"}} +) + +// Server-to-server calls to the gateway: no redirects, short timeout. +var sixbeeGatewayHTTP = &http.Client{ + Timeout: 20 * time.Second, + CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + }, +} + +// Storage and order lookups, replaceable in tests. +var ( + sixbeePayPut = func(token string, value string) error { + if !common.RedisEnabled { + return errors.New("redis is not enabled") + } + return common.RedisSet(sixbeePayRecordPrefix+token, value, sixbeePayRecordTTL) + } + sixbeePayGet = func(token string) (string, error) { + if !common.RedisEnabled { + return "", errors.New("redis is not enabled") + } + return common.RedisGet(sixbeePayRecordPrefix + token) + } + sixbeeOrderStatus = func(tradeNo string) (string, bool) { + if strings.HasPrefix(tradeNo, "SUB") { + if order := model.GetSubscriptionOrderByTradeNo(tradeNo); order != nil { + return order.Status, true + } + return "", false + } + if topUp := model.GetTopUpByTradeNo(tradeNo); topUp != nil { + return topUp.Status, true + } + return "", false + } +) + +type sixbeePayRecord struct { + TradeNo string `json:"trade_no"` + GatewayTradeNo string `json:"gateway_trade_no"` + Type string `json:"type"` // alipay | usdt | usdc + PayType string `json:"pay_type"` // qrcode | jump | wallet + Code string `json:"code,omitempty"` + Link string `json:"link,omitempty"` + Token string `json:"token,omitempty"` + Network string `json:"network,omitempty"` + CryptoAmount string `json:"crypto_amount,omitempty"` + ExpiresAt int64 `json:"expires_at"` + PricingAmount string `json:"pricing_amount"` + PricingCurrency string `json:"pricing_currency"` + CreatedAt int64 `json:"created_at"` +} + +// sixbeePayOrigin is SIXBEE_PAY_ORIGIN, e.g. https://pay.6bee.ai. Unset: nil +// (browser-form orders). Set but not an https origin: an error, so orders +// fail instead of silently falling back to the gateway's own pages. +func sixbeePayOrigin() (*url.URL, error) { + raw := strings.TrimSpace(os.Getenv("SIXBEE_PAY_ORIGIN")) + if raw == "" { + return nil, nil + } + u, err := url.Parse(raw) + if err != nil || u.Scheme != "https" || u.Hostname() == "" || u.User != nil || + (u.Path != "" && u.Path != "/") || u.RawQuery != "" || u.Fragment != "" { + return nil, fmt.Errorf("SIXBEE_PAY_ORIGIN must be an https origin, got %q", raw) + } + return &url.URL{Scheme: "https", Host: u.Host}, nil +} + +func sixbeeIsPayHost(host string, origin *url.URL) bool { + if origin == nil { + return false + } + if h, _, err := net.SplitHostPort(host); err == nil { + host = h + } + return strings.EqualFold(host, origin.Hostname()) +} + +// sixbeeGatewayBase is the admin's PayAddress, which must be https. +func sixbeeGatewayBase(client *epay.Client) (*url.URL, error) { + if client == nil || client.BaseUrl == nil { + return nil, errors.New("epay: gateway is not configured") + } + u := *client.BaseUrl + if u.Scheme != "https" || u.Host == "" || u.User != nil || u.RawQuery != "" { + return nil, errors.New("epay: PayAddress must be an https URL") + } + return &u, nil +} + +func sixbeePayScene(paymentType string, userAgent string) (string, error) { + switch paymentType { + case "alipay": + if sixbeeMobileUARe.MatchString(userAgent) { + return "wap", nil // official Alipay H5 page, which opens the app + } + return "qrcode", nil + case "usdt", "usdc": + return "wallet", nil + } + return "", fmt.Errorf("epay: %q has no same-domain checkout", paymentType) +} + +func sixbeeNewPayToken() (string, error) { + b := make([]byte, 16) + if _, err := rand.Read(b); err != nil { + return "", err + } + return base64.RawURLEncoding.EncodeToString(b), nil +} + +func sixbeeSitePurchase(c *gin.Context, client *epay.Client, args *epay.PurchaseArgs, origin *url.URL) (string, map[string]string, error) { + if client == nil || client.Config == nil || args == nil || args.NotifyUrl == nil { + return "", nil, errors.New("epay: incomplete purchase arguments") + } + base, err := sixbeeGatewayBase(client) + if err != nil { + return "", nil, err + } + ctx := context.Background() + userAgent, clientIP := "", "" + if c != nil && c.Request != nil { + ctx = c.Request.Context() + userAgent = c.Request.UserAgent() + if ip := net.ParseIP(c.ClientIP()); ip != nil { + clientIP = ip.String() + } + } + scene, err := sixbeePayScene(args.Type, userAgent) + if err != nil { + return "", nil, err + } + token, err := sixbeeNewPayToken() + if err != nil { + return "", nil, err + } + notifyURL := *origin + notifyURL.Path = args.NotifyUrl.Path // /api/user/epay/notify or /api/subscription/epay/notify + pageURL := *origin + pageURL.Path = "/pay/" + token + + params := map[string]string{ + "pid": client.Config.PartnerID, + "type": args.Type, + "out_trade_no": args.ServiceTradeNo, + "notify_url": notifyURL.String(), + "return_url": pageURL.String(), + "name": args.Name, + "money": args.Money, + "currency": sixbeePayCurrency, + "checkout_mode": "site", + "pay_scene": scene, + "timestamp": strconv.FormatInt(time.Now().Unix(), 10), + } + if clientIP != "" { + params["clientip"] = clientIP + } + signed, err := epay.GenerateParamsWithError(params, client.Config.Key) + if err != nil { + return "", nil, err + } + fields, err := sixbeeGatewayPost(ctx, base, "mapi.php", signed) + if err != nil { + return "", nil, err + } + rec, err := sixbeeParseSiteResponse(fields, client.Config.Key, args, scene, base.Hostname(), time.Now()) + if err != nil { + return "", nil, err + } + rec.TradeNo = args.ServiceTradeNo + raw, err := json.Marshal(rec) + if err != nil { + return "", nil, err + } + if err := sixbeePayPut(token, string(raw)); err != nil { + return "", nil, fmt.Errorf("epay: could not store the checkout page: %w", err) + } + return pageURL.String(), map[string]string{}, nil +} + +func sixbeeGatewayPost(ctx context.Context, base *url.URL, endpoint string, form map[string]string) (map[string]string, error) { + u := *base + u.Path = path.Join("/", base.Path, endpoint) + values := url.Values{} + for k, v := range form { + values.Set(k, v) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), strings.NewReader(values.Encode())) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + req.Header.Set("Accept", "application/json") + resp, err := sixbeeGatewayHTTP.Do(req) + if err != nil { + return nil, fmt.Errorf("epay: gateway request failed: %w", err) + } + defer resp.Body.Close() + body, err := io.ReadAll(io.LimitReader(resp.Body, sixbeeGatewayMaxBody+1)) + if err != nil { + return nil, fmt.Errorf("epay: reading the gateway answer failed: %w", err) + } + if len(body) > sixbeeGatewayMaxBody { + return nil, errors.New("epay: gateway answer is too large") + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("epay: gateway answered HTTP %d", resp.StatusCode) + } + return sixbeeDecodeGatewayJSON(body) +} + +// sixbeeDecodeGatewayJSON flattens the gateway's JSON object to the string +// values its signature covers (numbers keep their literal form). +func sixbeeDecodeGatewayJSON(body []byte) (map[string]string, error) { + dec := json.NewDecoder(bytes.NewReader(body)) + dec.UseNumber() + var raw map[string]any + if err := dec.Decode(&raw); err != nil { + return nil, errors.New("epay: gateway answer is not a JSON object") + } + out := make(map[string]string, len(raw)) + for k, v := range raw { + switch t := v.(type) { + case string: + out[k] = t + case json.Number: + out[k] = t.String() + case nil: + default: + return nil, fmt.Errorf("epay: gateway field %q is not a scalar", k) + } + } + return out, nil +} + +// sixbeeGatewaySign is the gateway's MD5 signature: non-empty fields except +// sign/sign_type, sorted by name, joined as k=v&k=v, then the site key. +func sixbeeGatewaySign(fields map[string]string, key string) string { + keys := make([]string, 0, len(fields)) + for k, v := range fields { + if k == "sign" || k == "sign_type" || v == "" { + continue + } + keys = append(keys, k) + } + sort.Strings(keys) + var b strings.Builder + for i, k := range keys { + if i > 0 { + b.WriteByte('&') + } + b.WriteString(k) + b.WriteByte('=') + b.WriteString(fields[k]) + } + b.WriteString(key) + sum := md5.Sum([]byte(b.String())) + return hex.EncodeToString(sum[:]) +} + +func sixbeeParseSiteResponse(f map[string]string, key string, args *epay.PurchaseArgs, scene string, gatewayHost string, now time.Time) (*sixbeePayRecord, error) { + if f["code"] != "0" { + msg := f["msg"] + if len(msg) > 200 { + msg = msg[:200] + } + if msg == "" { + msg = "no message" + } + return nil, fmt.Errorf("epay: gateway refused the order: %s", msg) + } + sign := strings.ToLower(f["sign"]) + if len(sign) != 32 || subtle.ConstantTimeCompare([]byte(sign), []byte(sixbeeGatewaySign(f, key))) != 1 { + return nil, errors.New("epay: gateway answer signature does not verify") + } + ts, err := strconv.ParseInt(f["timestamp"], 10, 64) + if err != nil || ts < now.Unix()-300 || ts > now.Unix()+300 { + return nil, errors.New("epay: gateway answer timestamp is missing or stale") + } + if !sixbeeGatewayTradeRe.MatchString(f["trade_no"]) { + return nil, errors.New("epay: gateway trade_no is not 19 digits") + } + want, err1 := decimal.NewFromString(args.Money) + got, err2 := decimal.NewFromString(f["pricing_amount"]) + if err1 != nil || err2 != nil || !got.Equal(want) || f["pricing_currency"] != sixbeePayCurrency { + return nil, fmt.Errorf("epay: gateway priced the order at %s %s, expected %s %s", + f["pricing_amount"], f["pricing_currency"], args.Money, sixbeePayCurrency) + } + info := f["pay_info"] + if info == "" || len(info) > 4096 { + return nil, errors.New("epay: gateway pay_info is empty or too long") + } + if gatewayHost != "" && strings.Contains(strings.ToLower(info), strings.ToLower(gatewayHost)) { + return nil, errors.New("epay: gateway pay_info names the gateway itself") + } + rec := &sixbeePayRecord{ + GatewayTradeNo: f["trade_no"], + Type: args.Type, + PayType: f["pay_type"], + PricingAmount: f["pricing_amount"], + PricingCurrency: f["pricing_currency"], + CreatedAt: now.Unix(), + } + switch scene { + case "qrcode": + if f["pay_type"] != "qrcode" || !sixbeeIsURL(info, "qr.alipay.com", "") { + return nil, fmt.Errorf("epay: expected an Alipay QR code, got pay_type %q", f["pay_type"]) + } + rec.Code = info + rec.ExpiresAt = now.Add(sixbeeAlipayQRLife).Unix() + case "wap": + if f["pay_type"] != "jump" || !sixbeeIsURL(info, "openapi.alipay.com", "/gateway.do") { + return nil, fmt.Errorf("epay: expected an official Alipay link, got pay_type %q", f["pay_type"]) + } + rec.Link = info + rec.ExpiresAt = now.Add(sixbeeAlipayWapLife).Unix() + case "wallet": + if f["pay_type"] != "wallet" { + return nil, fmt.Errorf("epay: expected wallet details, got pay_type %q", f["pay_type"]) + } + if err := sixbeeParseWallet(info, args.Type, now, rec); err != nil { + return nil, err + } + default: + return nil, fmt.Errorf("epay: unknown checkout scene %q", scene) + } + return rec, nil +} + +func sixbeeIsURL(raw string, host string, pathPrefix string) bool { + u, err := url.Parse(raw) + if err != nil || u.Scheme != "https" || !strings.EqualFold(u.Host, host) || u.User != nil { + return false + } + return pathPrefix == "" || u.Path == pathPrefix +} + +func sixbeeParseWallet(info string, paymentType string, now time.Time, rec *sixbeePayRecord) error { + asset, ok := sixbeeCryptoAssets[paymentType] + if !ok { + return fmt.Errorf("epay: %q is not a crypto payment type", paymentType) + } + dec := json.NewDecoder(strings.NewReader(info)) + dec.UseNumber() + var w map[string]any + if err := dec.Decode(&w); err != nil { + return errors.New("epay: wallet details are not JSON") + } + str := func(k string) string { + switch v := w[k].(type) { + case string: + return v + case json.Number: + return v.String() + } + return "" + } + token, network, amount, address := str("token"), str("network"), str("amount"), str("address") + if str("kind") != "crypto" || token != asset[0] || network != asset[1] { + return fmt.Errorf("epay: wallet is %s on %s, expected %s on %s", token, network, asset[0], asset[1]) + } + if !sixbeeCryptoAmountRe.MatchString(amount) { + return fmt.Errorf("epay: wallet amount %q is not a decimal", amount) + } + d, err := decimal.NewFromString(amount) + if err != nil || !d.IsPositive() { + return fmt.Errorf("epay: wallet amount %q is not positive", amount) + } + usd, err := decimal.NewFromString(rec.PricingAmount) + tail := d.Sub(usd) + if err != nil || tail.IsNegative() || tail.GreaterThan(sixbeeCryptoTailMax) || !tail.Mod(sixbeeCryptoCent).IsZero() { + return fmt.Errorf("epay: wallet amount %s %s is not 1:1 with %s USD", amount, token, rec.PricingAmount) + } + addressOK := sixbeeSolAddressRe.MatchString(address) + if network == "TRON" { + addressOK = sixbeeTronAddressRe.MatchString(address) + } + if !addressOK || str("qr_content") != address { + return errors.New("epay: wallet address is malformed or differs from its QR content") + } + expires, err := strconv.ParseInt(str("expires_at"), 10, 64) + if err != nil || expires <= now.Unix() || expires > now.Add(7*24*time.Hour).Unix() { + return errors.New("epay: wallet expiry is missing or out of range") + } + rec.Token, rec.Network, rec.CryptoAmount, rec.Code, rec.ExpiresAt = token, network, amount, address, expires + return nil +} + +func sixbeeLoadPayRecord(token string) *sixbeePayRecord { + if !sixbeePayTokenRe.MatchString(token) { + return nil + } + raw, err := sixbeePayGet(token) + if err != nil || raw == "" { + return nil + } + var rec sixbeePayRecord + if err := json.Unmarshal([]byte(raw), &rec); err != nil || rec.TradeNo == "" { + return nil + } + return &rec +} + +// sixbeePayState is what the page shows: pending, paid, expired, failed or unknown. +func sixbeePayState(rec *sixbeePayRecord, now time.Time) string { + status, found := sixbeeOrderStatus(rec.TradeNo) + if !found { + return "unknown" + } + switch status { + case common.TopUpStatusSuccess: + return "paid" + case common.TopUpStatusFailed: + return "failed" + case common.TopUpStatusExpired: + return "expired" + } + if rec.ExpiresAt > 0 && now.Unix() > rec.ExpiresAt { + return "expired" + } + return "pending" +} + +func sixbeeQRSVG(content string, label string) (template.HTML, error) { + code, err := qr.Encode(content, qr.M, qr.Auto) + if err != nil { + return "", err + } + bounds := code.Bounds() + n := bounds.Dx() + dark := func(x, y int) bool { + r, _, _, _ := code.At(bounds.Min.X+x, bounds.Min.Y+y).RGBA() + return r < 0x8000 + } + const quiet = 4 + var b strings.Builder + fmt.Fprintf(&b, ``, + n+2*quiet, n+2*quiet, template.HTMLEscapeString(label)) + b.WriteString(``) + return template.HTML(b.String()), nil +} + +// SixbeePayPage renders https://pay.6bee.ai/pay/. The console posts an +// empty form here (it opens Epay checkouts with a form POST); that is +// answered with 303 to the same page. +func SixbeePayPage(c *gin.Context) { + origin, _ := sixbeePayOrigin() + if !sixbeeIsPayHost(c.Request.Host, origin) { + c.String(http.StatusNotFound, "not found") + return + } + token := c.Param("token") + if c.Request.Method == http.MethodPost { + if !sixbeePayTokenRe.MatchString(token) { + c.String(http.StatusNotFound, "not found") + return + } + c.Redirect(http.StatusSeeOther, "/pay/"+token) + return + } + now := time.Now() + lang := sixbeePayLang(c) + rec := sixbeeLoadPayRecord(token) + view := sixbeeBuildPayView(rec, token, lang, now) + nonce, err := sixbeeNewPayToken() + if err != nil { + c.String(http.StatusInternalServerError, "error") + return + } + view.Nonce = nonce + var buf bytes.Buffer + if err := sixbeePayTemplate.Execute(&buf, view); err != nil { + logger.LogError(c.Request.Context(), common.LogText("6bee pay page render failed token=%s error=%q", token, err.Error())) + c.String(http.StatusInternalServerError, "error") + return + } + sixbeePayHeaders(c, nonce) + status := http.StatusOK + if rec == nil { + status = http.StatusNotFound + } + c.Data(status, "text/html; charset=utf-8", buf.Bytes()) +} + +// SixbeePayStatus is polled by the page: {"status": pending|paid|expired|failed|unknown}. +func SixbeePayStatus(c *gin.Context) { + origin, _ := sixbeePayOrigin() + if !sixbeeIsPayHost(c.Request.Host, origin) { + c.String(http.StatusNotFound, "not found") + return + } + c.Header("Cache-Control", "no-store") + c.Header("X-Content-Type-Options", "nosniff") + rec := sixbeeLoadPayRecord(c.Param("token")) + if rec == nil { + c.JSON(http.StatusNotFound, gin.H{"status": "unknown"}) + return + } + c.JSON(http.StatusOK, gin.H{"status": sixbeePayState(rec, time.Now())}) +} + +// SixbeeEpayProviderRelay forwards a payment provider's notification to the +// gateway unchanged and returns the gateway's answer unchanged. +func SixbeeEpayProviderRelay(c *gin.Context) { + origin, _ := sixbeePayOrigin() + if !sixbeeIsPayHost(c.Request.Host, origin) { + c.String(http.StatusNotFound, "not found") + return + } + paymentType, tradeNo := c.Param("type"), c.Param("trade_no") + if !sixbeeRelayTypes[paymentType] || !sixbeeGatewayTradeRe.MatchString(tradeNo) { + c.String(http.StatusNotFound, "not found") + return + } + body, err := io.ReadAll(io.LimitReader(c.Request.Body, sixbeeGatewayMaxBody+1)) + if err != nil || len(body) == 0 || len(body) > sixbeeGatewayMaxBody { + c.String(http.StatusRequestEntityTooLarge, "invalid body") + return + } + base, err := sixbeeGatewayBase(GetEpayClient()) + if err != nil { + c.String(http.StatusServiceUnavailable, "relay unavailable") + return + } + target := *base + target.Path = path.Join("/", base.Path, "pay/notify", tradeNo) + "/" + target.RawQuery = "" + ctx, cancel := context.WithTimeout(c.Request.Context(), 15*time.Second) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodPost, target.String(), bytes.NewReader(body)) + if err != nil { + c.String(http.StatusServiceUnavailable, "relay unavailable") + return + } + for _, name := range sixbeeRelayHeaders { + if v := c.GetHeader(name); v != "" { + req.Header.Set(name, v) + } + } + resp, err := sixbeeGatewayHTTP.Do(req) + if err != nil { + logger.LogWarn(c.Request.Context(), common.LogText("Epay provider relay failed type=%s trade_no=%s client_ip=%s error=%q", paymentType, tradeNo, c.ClientIP(), err.Error())) + c.String(http.StatusBadGateway, "relay failed") + return + } + defer resp.Body.Close() + if resp.StatusCode >= 300 && resp.StatusCode < 400 { + c.String(http.StatusBadGateway, "invalid upstream response") + return + } + out, _ := io.ReadAll(io.LimitReader(resp.Body, sixbeeGatewayMaxBody)) + contentType := resp.Header.Get("Content-Type") + if contentType == "" { + contentType = "text/plain; charset=utf-8" + } + logger.LogInfo(c.Request.Context(), common.LogText("Epay provider relay type=%s trade_no=%s client_ip=%s gateway_status=%d", paymentType, tradeNo, c.ClientIP(), resp.StatusCode)) + c.Header("Cache-Control", "no-store") + c.Data(resp.StatusCode, contentType, out) +} + +func sixbeePayHeaders(c *gin.Context, nonce string) { + c.Header("Content-Security-Policy", "default-src 'none'; style-src 'nonce-"+nonce+"'; script-src 'nonce-"+nonce+ + "'; img-src data:; connect-src 'self'; base-uri 'none'; form-action 'none'; frame-ancestors 'none'") + c.Header("Cache-Control", "no-store") + c.Header("Referrer-Policy", "no-referrer") + c.Header("X-Content-Type-Options", "nosniff") + c.Header("X-Frame-Options", "DENY") + c.Header("X-Robots-Tag", "noindex, nofollow") +} diff -ruN a/controller/sixbee_pay_page.go b/controller/sixbee_pay_page.go --- a/controller/sixbee_pay_page.go +++ b/controller/sixbee_pay_page.go @@ -0,0 +1,371 @@ +package controller + +// 6bee: the pay.6bee.ai checkout page (see controller/sixbee_pay.go). + +import ( + "fmt" + "html/template" + "strings" + "time" + + "github.com/gin-gonic/gin" + "github.com/shopspring/decimal" +) + +type sixbeePayView struct { + Lang string + Nonce string + T map[string]string + Title string + State string // pending | paid | expired | failed | unknown | error | missing + Kind string // qrcode | jump | wallet + ShowPay bool + Amount string + QR template.HTML + AppLink template.URL + Network string + Address string + CryptoAmount string + Token string + Warnings []string + ExpiresAt int64 + Now int64 + StatusURL string + ConsoleURL string + TradeNo string +} + +var sixbeePayText = map[string]map[string]string{ + "zh": { + "title_alipay": "支付宝付款", + "title_crypto": "%s 付款", + "scan_alipay": "打开支付宝，扫一扫付款", + "open_alipay_app": "在手机上？打开支付宝付款", + "open_alipay": "打开支付宝付款", + "jump_hint": "将前往支付宝官方页面完成付款", + "network": "网络", + "address": "收款地址", + "send_amount": "转账金额", + "copy": "复制", + "copied": "已复制", + "warn_network": "只能通过 %[1]s 网络转入 %[2]s，其他网络或币种的转账无法到账。", + "warn_exact": "请按上面的金额精确转账，少转或多转都不会自动到账。", + "warn_expiry": "请在倒计时结束前完成转账；过期后请回到控制台重新下单。", + "time_left": "剩余时间", + "pending": "等待付款…", + "paid": "付款成功，余额已到账。", + "back_soon": "正在返回控制台…", + "expired": "订单已过期，请回到控制台重新下单。", + "failed": "付款未成功，请回到控制台重新下单。", + "unknown": "找不到这笔订单，请回到控制台重新下单。", + "error": "付款信息加载失败，请回到控制台重新下单。", + "missing_title": "付款页面不存在", + "missing": "链接可能已过期，请回到控制台重新下单。", + "order": "订单号", + "back": "返回控制台", + "qr_label": "付款二维码", + "net_TRON": "TRON（TRC20）", + "net_SOLANA": "Solana", + }, + "en": { + "title_alipay": "Pay with Alipay", + "title_crypto": "Pay with %s", + "scan_alipay": "Scan with the Alipay app to pay", + "open_alipay_app": "On your phone? Open Alipay", + "open_alipay": "Open Alipay to pay", + "jump_hint": "You'll finish paying on Alipay's own page", + "network": "Network", + "address": "Address", + "send_amount": "Amount", + "copy": "Copy", + "copied": "Copied", + "warn_network": "Send %[2]s on the %[1]s network only. Other networks or tokens will not arrive.", + "warn_exact": "Send exactly the amount above. A different amount is not credited automatically.", + "warn_expiry": "Finish the transfer before the timer runs out. After that, start a new top-up in the console.", + "time_left": "Time left", + "pending": "Waiting for payment…", + "paid": "Payment received. Your balance has been topped up.", + "back_soon": "Returning to the console…", + "expired": "This order has expired. Start a new top-up in the console.", + "failed": "The payment did not go through. Start a new top-up in the console.", + "unknown": "This order can't be found. Start a new top-up in the console.", + "error": "The payment details could not be loaded. Start a new top-up in the console.", + "missing_title": "Payment page not found", + "missing": "The link may have expired. Start a new top-up in the console.", + "order": "Order", + "back": "Back to console", + "qr_label": "Payment QR code", + "net_TRON": "TRON (TRC20)", + "net_SOLANA": "Solana", + }, +} + +func sixbeePayLang(c *gin.Context) string { + switch c.Query("lang") { + case "en": + return "en" + case "zh": + return "zh" + } + accept := strings.ToLower(strings.TrimSpace(c.GetHeader("Accept-Language"))) + first := strings.TrimSpace(strings.SplitN(accept, ",", 2)[0]) + if accept == "" || strings.HasPrefix(first, "zh") { + return "zh" + } + return "en" +} + +// sixbeeFormatDecimal shows a gateway amount ("6.67000000") for people. fixed >= 0 rounds +// to that many places ("6.67"); fixed < 0 keeps the exact value without trailing zeros but +// with at least two decimals ("1.00", "10.000123"), so a crypto amount is never rounded. +// A value that is not a decimal is shown as given. +func sixbeeFormatDecimal(raw string, fixed int32) string { + d, err := decimal.NewFromString(strings.TrimSpace(raw)) + if err != nil { + return raw + } + if fixed >= 0 { + return d.StringFixed(fixed) + } + s := d.String() + if i := strings.IndexByte(s, '.'); i < 0 || len(s)-i-1 < 2 { + return d.StringFixed(2) + } + return s +} + +func sixbeeBuildPayView(rec *sixbeePayRecord, token string, lang string, now time.Time) sixbeePayView { + t := sixbeePayText[lang] + v := sixbeePayView{ + Lang: map[string]string{"zh": "zh-CN", "en": "en"}[lang], + T: t, + Now: now.Unix(), + ConsoleURL: paymentReturnPath("/wallet"), + } + if rec == nil { + v.State = "missing" + v.Title = t["missing_title"] + return v + } + v.State = sixbeePayState(rec, now) + v.StatusURL = "/pay/" + token + "/status" + v.TradeNo = rec.TradeNo + v.Kind = rec.PayType + v.Amount = sixbeeFormatDecimal(rec.PricingAmount, 2) + " " + rec.PricingCurrency + v.ExpiresAt = rec.ExpiresAt + if rec.Type == "alipay" { + v.Title = t["title_alipay"] + } else { + v.Title = fmt.Sprintf(t["title_crypto"], rec.Token) + } + // Once an order is closed its QR code and address are no longer shown. + v.ShowPay = v.State == "pending" + if !v.ShowPay { + return v + } + switch rec.PayType { + case "qrcode": + svg, err := sixbeeQRSVG(rec.Code, t["qr_label"]) + if err != nil { + v.State, v.ShowPay = "error", false + return v + } + v.QR = svg + v.AppLink = template.URL(rec.Code) // checked: https://qr.alipay.com/… + case "jump": + v.AppLink = template.URL(rec.Link) // checked: https://openapi.alipay.com/gateway.do… + v.ExpiresAt = 0 // Alipay's page keeps its own time + case "wallet": + svg, err := sixbeeQRSVG(rec.Code, t["qr_label"]) + if err != nil { + v.State, v.ShowPay = "error", false + return v + } + v.QR = svg + v.Address = rec.Code + v.CryptoAmount = sixbeeFormatDecimal(rec.CryptoAmount, -1) // exact; trailing zeros dropped + v.Token = rec.Token + v.Network = t["net_"+rec.Network] + if v.Network == "" { + v.Network = rec.Network + } + v.Warnings = []string{ + fmt.Sprintf(t["warn_network"], v.Network, rec.Token), + t["warn_exact"], + t["warn_expiry"], + } + default: + v.State, v.ShowPay = "error", false + } + return v +} + +var sixbeePayTemplate = template.Must(template.New("pay").Parse(` + + + + + + + +{{.Title}} · 6bee.ai + + + + +
+
6bee.ai
+
+{{- if eq .State "missing"}} +

{{.T.missing_title}}

+

{{.T.missing}}

+{{- else}} +

{{.Title}}

+

{{.Amount}}

+{{- if .ShowPay}} +
+{{- if eq .Kind "qrcode"}} +
{{.QR}}
+

{{.T.scan_alipay}}

+{{.T.open_alipay_app}} +{{- else if eq .Kind "jump"}} +{{.T.open_alipay}} +

{{.T.jump_hint}}

+{{- else if eq .Kind "wallet"}} +

{{.T.network}}{{.Network}}

+
{{.QR}}
+
{{.T.address}}{{.Address}}
+
{{.T.send_amount}}{{.CryptoAmount}} {{.Token}}
+
    {{range .Warnings}}
  • {{.}}
    • {{end}}
+{{- end}} +{{- if .ExpiresAt}} +

{{.T.time_left}} --:--

+{{- end}} +
+{{- end}} +

{{index .T .State}}

+{{- end}} +
+
{{if .TradeNo}}{{.T.order}} {{.TradeNo}}{{end}}{{.T.back}}
+
+ + + +`)) diff -ruN a/controller/sixbee_pay_test.go b/controller/sixbee_pay_test.go --- a/controller/sixbee_pay_test.go +++ b/controller/sixbee_pay_test.go @@ -0,0 +1,399 @@ +package controller + +import ( + "crypto/md5" + "encoding/hex" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "strings" + "testing" + "time" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/setting/operation_setting" + + "github.com/Calcium-Ion/go-epay/epay" + "github.com/gin-gonic/gin" +) + +const sixbeeTestKey = "0123456789abcdef0123456789abcdef" + +func sixbeeSigned(fields map[string]string) map[string]string { + fields["sign_type"] = "MD5" + fields["sign"] = sixbeeGatewaySign(fields, sixbeeTestKey) + return fields +} + +func TestSixbeeGatewaySignMatchesThePHPGateway(t *testing.T) { + f := map[string]string{"trade_no": "2026110920391191756", "code": "0", "pay_type": "qrcode", + "pay_info": "https://qr.alipay.com/bax1", "empty": "", "sign": "x", "sign_type": "MD5", "timestamp": "1794227949"} + sum := md5.Sum([]byte("code=0&pay_info=https://qr.alipay.com/bax1&pay_type=qrcode×tamp=1794227949&trade_no=2026110920391191756" + sixbeeTestKey)) + if got := sixbeeGatewaySign(f, sixbeeTestKey); got != hex.EncodeToString(sum[:]) { + t.Fatalf("sign = %s", got) + } +} + +func sixbeeQRAnswer(now time.Time) map[string]string { + return sixbeeSigned(map[string]string{ + "code": "0", "trade_no": "2026110920391191756", "pay_type": "qrcode", "pay_info": "https://qr.alipay.com/bax08812abc", + "pricing_amount": "5.00", "pricing_currency": "USD", "payment_amount": "33.37", "payment_currency": "CNY", + "fx_rate": "6.6740", "timestamp": strconv.FormatInt(now.Unix(), 10), + }) +} + +func sixbeeWalletAnswer(now time.Time, wallet map[string]any) map[string]string { + raw, _ := json.Marshal(wallet) + return sixbeeSigned(map[string]string{ + "code": "0", "trade_no": "2026110920391191757", "pay_type": "wallet", "pay_info": string(raw), + "pricing_amount": "5.00", "pricing_currency": "USD", "payment_currency": "USDT", + "timestamp": strconv.FormatInt(now.Unix(), 10), + }) +} + +func TestSixbeeParseSiteResponse(t *testing.T) { + now := time.Unix(1794227949, 0) + args := &epay.PurchaseArgs{Type: "alipay", Money: "5.00"} + rec, err := sixbeeParseSiteResponse(sixbeeQRAnswer(now), sixbeeTestKey, args, "qrcode", "pay.gateway.example", now) + if err != nil { + t.Fatal(err) + } + if rec.Code != "https://qr.alipay.com/bax08812abc" || rec.ExpiresAt != now.Add(10*time.Minute).Unix() { + t.Fatalf("record = %+v", rec) + } + + // 1:1 with the 5.00 USD price, plus the gateway's 0.01 collision tail at most ten times. + for _, amount := range []string{"5", "5.00000000", "5.01", "5.10"} { + wallet := map[string]any{"kind": "crypto", "token": "USDT", "network": "TRON", "amount": amount, + "address": "TTestAddressForUnitTestsXXXXXXXXXX", "qr_content": "TTestAddressForUnitTestsXXXXXXXXXX", "expires_at": now.Unix() + 1800} + rec, err = sixbeeParseSiteResponse(sixbeeWalletAnswer(now, wallet), sixbeeTestKey, &epay.PurchaseArgs{Type: "usdt", Money: "5.00"}, "wallet", "pay.gateway.example", now) + if err != nil { + t.Fatalf("amount %s: %v", amount, err) + } + if rec.Code != wallet["address"] || rec.CryptoAmount != amount || rec.Network != "TRON" || rec.ExpiresAt != now.Unix()+1800 { + t.Fatalf("wallet record = %+v", rec) + } + } + + resign := func(f map[string]string) map[string]string { + delete(f, "sign") + return sixbeeSigned(f) + } + bad := []struct { + name string + f map[string]string + args *epay.PurchaseArgs + scene string + }{ + {"refused", map[string]string{"code": "-1", "msg": "IP not allowed"}, args, "qrcode"}, + {"bad signature", func() map[string]string { + f := sixbeeQRAnswer(now) + f["pay_info"] = "https://qr.alipay.com/other" + return f + }(), args, "qrcode"}, + {"stale", resign(func() map[string]string { + f := sixbeeQRAnswer(now) + f["timestamp"] = strconv.FormatInt(now.Unix()-900, 10) + return f + }()), args, "qrcode"}, + {"other amount", sixbeeQRAnswer(now), &epay.PurchaseArgs{Type: "alipay", Money: "50.00"}, "qrcode"}, + {"other currency", resign(func() map[string]string { f := sixbeeQRAnswer(now); f["pricing_currency"] = "CNY"; return f }()), args, "qrcode"}, + {"gateway page", resign(func() map[string]string { + f := sixbeeQRAnswer(now) + f["pay_type"], f["pay_info"] = "qrcode", "https://pay.gateway.example/pay/qrcode/2026110920391191756/" + return f + }()), args, "qrcode"}, + {"not alipay", resign(func() map[string]string { + f := sixbeeQRAnswer(now) + f["pay_info"] = "https://evil.example/qr" + return f + }()), args, "qrcode"}, + {"jump for qrcode", resign(func() map[string]string { f := sixbeeQRAnswer(now); f["pay_type"] = "jump"; return f }()), args, "qrcode"}, + {"wap not official", resign(func() map[string]string { + f := sixbeeQRAnswer(now) + f["pay_type"], f["pay_info"] = "jump", "https://openapi.alipay.com.evil.example/gateway.do" + return f + }()), args, "wap"}, + {"wrong network", sixbeeWalletAnswer(now, map[string]any{"kind": "crypto", "token": "USDT", "network": "SOLANA", "amount": "5", + "address": "TTestAddressForUnitTestsXXXXXXXXXX", "qr_content": "TTestAddressForUnitTestsXXXXXXXXXX", "expires_at": now.Unix() + 60}), + &epay.PurchaseArgs{Type: "usdt", Money: "5.00"}, "wallet"}, + {"qr differs from address", sixbeeWalletAnswer(now, map[string]any{"kind": "crypto", "token": "USDT", "network": "TRON", "amount": "5", + "address": "TTestAddressForUnitTestsXXXXXXXXXX", "qr_content": "TTestAddressForUnitTestsYYYYYYYYYY", "expires_at": now.Unix() + 60}), + &epay.PurchaseArgs{Type: "usdt", Money: "5.00"}, "wallet"}, + {"below 1:1", sixbeeWalletAnswer(now, map[string]any{"kind": "crypto", "token": "USDT", "network": "TRON", "amount": "4.99", + "address": "TTestAddressForUnitTestsXXXXXXXXXX", "qr_content": "TTestAddressForUnitTestsXXXXXXXXXX", "expires_at": now.Unix() + 60}), + &epay.PurchaseArgs{Type: "usdt", Money: "5.00"}, "wallet"}, + {"rate applied", sixbeeWalletAnswer(now, map[string]any{"kind": "crypto", "token": "USDT", "network": "TRON", "amount": "36.50", + "address": "TTestAddressForUnitTestsXXXXXXXXXX", "qr_content": "TTestAddressForUnitTestsXXXXXXXXXX", "expires_at": now.Unix() + 60}), + &epay.PurchaseArgs{Type: "usdt", Money: "5.00"}, "wallet"}, + {"sub-cent tail", sixbeeWalletAnswer(now, map[string]any{"kind": "crypto", "token": "USDT", "network": "TRON", "amount": "5.0012", + "address": "TTestAddressForUnitTestsXXXXXXXXXX", "qr_content": "TTestAddressForUnitTestsXXXXXXXXXX", "expires_at": now.Unix() + 60}), + &epay.PurchaseArgs{Type: "usdt", Money: "5.00"}, "wallet"}, + {"tail too long", sixbeeWalletAnswer(now, map[string]any{"kind": "crypto", "token": "USDT", "network": "TRON", "amount": "5.11", + "address": "TTestAddressForUnitTestsXXXXXXXXXX", "qr_content": "TTestAddressForUnitTestsXXXXXXXXXX", "expires_at": now.Unix() + 60}), + &epay.PurchaseArgs{Type: "usdt", Money: "5.00"}, "wallet"}, + {"expired wallet", sixbeeWalletAnswer(now, map[string]any{"kind": "crypto", "token": "USDT", "network": "TRON", "amount": "5", + "address": "TTestAddressForUnitTestsXXXXXXXXXX", "qr_content": "TTestAddressForUnitTestsXXXXXXXXXX", "expires_at": now.Unix() - 1}), + &epay.PurchaseArgs{Type: "usdt", Money: "5.00"}, "wallet"}, + } + for _, tc := range bad { + if _, err := sixbeeParseSiteResponse(tc.f, sixbeeTestKey, tc.args, tc.scene, "pay.gateway.example", now); err == nil { + t.Errorf("%s: accepted", tc.name) + } + } +} + +func sixbeeTestGateway(t *testing.T, handler http.HandlerFunc) (*epay.Client, func()) { + t.Helper() + srv := httptest.NewTLSServer(handler) + oldClient := sixbeeGatewayHTTP + sixbeeGatewayHTTP = srv.Client() + sixbeeGatewayHTTP.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse } + client, err := epay.NewClient(&epay.Config{PartnerID: "1008", Key: sixbeeTestKey}, srv.URL) + if err != nil { + t.Fatal(err) + } + oldAddress, oldID, oldKey := operation_setting.PayAddress, operation_setting.EpayId, operation_setting.EpayKey + operation_setting.PayAddress, operation_setting.EpayId, operation_setting.EpayKey = srv.URL, "1008", sixbeeTestKey + return client, func() { + srv.Close() + sixbeeGatewayHTTP = oldClient + operation_setting.PayAddress, operation_setting.EpayId, operation_setting.EpayKey = oldAddress, oldID, oldKey + } +} + +func sixbeeMemoryStore(t *testing.T) map[string]string { + store := map[string]string{} + oldPut, oldGet := sixbeePayPut, sixbeePayGet + sixbeePayPut = func(token, value string) error { store[token] = value; return nil } + sixbeePayGet = func(token string) (string, error) { return store[token], nil } + t.Cleanup(func() { sixbeePayPut, sixbeePayGet = oldPut, oldGet }) + return store +} + +func TestSixbeeSitePurchase(t *testing.T) { + t.Setenv("SIXBEE_PAY_ORIGIN", "https://pay.example.com") + store := sixbeeMemoryStore(t) + var got url.Values + client, done := sixbeeTestGateway(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/mapi.php" || r.Method != http.MethodPost { + http.NotFound(w, r) + return + } + _ = r.ParseForm() + got = r.PostForm + _ = json.NewEncoder(w).Encode(sixbeeQRAnswer(time.Now())) + }) + defer done() + + gin.SetMode(gin.TestMode) + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Request = httptest.NewRequest(http.MethodPost, "/api/user/pay", nil) + c.Request.Header.Set("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Chrome/141.0") + notify, _ := url.Parse("https://console.example.com/api/user/epay/notify") + ret, _ := url.Parse("https://console.example.com/usage-logs") + uri, params, err := sixbeeEpayPurchase(c, client, &epay.PurchaseArgs{ + Type: "alipay", ServiceTradeNo: "USR1NOabc1231794227949", Name: "TUC5", Money: "5.00", + Device: epay.PC, NotifyUrl: notify, ReturnUrl: ret, + }) + if err != nil { + t.Fatal(err) + } + if len(params) != 0 || !strings.HasPrefix(uri, "https://pay.example.com/pay/") { + t.Fatalf("uri = %s params = %v", uri, params) + } + token := strings.TrimPrefix(uri, "https://pay.example.com/pay/") + if !sixbeePayTokenRe.MatchString(token) || store[token] == "" { + t.Fatalf("token %q not stored", token) + } + want := map[string]string{"checkout_mode": "site", "pay_scene": "qrcode", "currency": "USD", "pid": "1008", "type": "alipay", + "notify_url": "https://pay.example.com/api/user/epay/notify", "return_url": uri, "money": "5.00"} + for k, v := range want { + if got.Get(k) != v { + t.Errorf("%s = %q, want %q", k, got.Get(k), v) + } + } + fields := map[string]string{} + for k := range got { + fields[k] = got.Get(k) + } + if got.Get("sign") != sixbeeGatewaySign(fields, sixbeeTestKey) { + t.Error("order signature does not verify") + } + rec := sixbeeLoadPayRecord(token) + if rec == nil || rec.TradeNo != "USR1NOabc1231794227949" || rec.GatewayTradeNo != "2026110920391191756" { + t.Fatalf("record = %+v", rec) + } + + c.Request.Header.Set("User-Agent", "Mozilla/5.0 (iPhone; CPU iPhone OS 18_0 like Mac OS X) Mobile/15E148") + if scene, _ := sixbeePayScene("alipay", c.Request.UserAgent()); scene != "wap" { + t.Errorf("mobile scene = %s", scene) + } + if _, err := sixbeePayScene("wxpay", ""); err == nil { + t.Error("wxpay has no site checkout here") + } + + t.Setenv("SIXBEE_PAY_ORIGIN", "http://pay.example.com") + if _, _, err := sixbeeEpayPurchase(c, client, &epay.PurchaseArgs{Type: "alipay", NotifyUrl: notify, ReturnUrl: ret}); err == nil { + t.Error("an http origin must not fall back to the gateway's own pages") + } +} + +func TestSixbeePayPageAndStatus(t *testing.T) { + t.Setenv("SIXBEE_PAY_ORIGIN", "https://pay.example.com") + store := sixbeeMemoryStore(t) + orderStatus := common.TopUpStatusPending + oldStatus := sixbeeOrderStatus + sixbeeOrderStatus = func(string) (string, bool) { return orderStatus, true } + defer func() { sixbeeOrderStatus = oldStatus }() + + now := time.Now() + rec := sixbeePayRecord{TradeNo: "USR1NOabc1231794227949", Type: "usdt", PayType: "wallet", Code: "TTestAddressForUnitTestsXXXXXXXXXX", + Token: "USDT", Network: "TRON", CryptoAmount: "5.0012", ExpiresAt: now.Unix() + 1800, PricingAmount: "5.00", PricingCurrency: "USD"} + raw, _ := json.Marshal(rec) + token := "AAAAAAAAAAAAAAAAAAAAAA" + store[token] = string(raw) + + gin.SetMode(gin.TestMode) + r := gin.New() + SetSixbeePayRoutesForTest(r) + get := func(method, target, host string) *httptest.ResponseRecorder { + w := httptest.NewRecorder() + req := httptest.NewRequest(method, target, nil) + req.Host = host + req.Header.Set("Accept-Language", "zh-CN,zh;q=0.9") + r.ServeHTTP(w, req) + return w + } + + w := get(http.MethodGet, "/pay/"+token, "pay.example.com") + body := w.Body.String() + if w.Code != 200 || !strings.Contains(body, "TTestAddressForUnitTestsXXXXXXXXXX") || !strings.Contains(body, "5.0012 USDT") || + !strings.Contains(body, "TRON（TRC20）") || !strings.Contains(body, "= common.RoleAdminUser { if req.UpstreamModelUpdateNotifyEnabled != nil { settings.UpstreamModelUpdateNotifyEnabled = *req.UpstreamModelUpdateNotifyEnabled diff -ruN a/go.mod b/go.mod --- a/go.mod +++ b/go.mod @@ -46,7 +46,7 @@ github.com/tcolgate/mp3 v0.0.0-20170426193717-e79c5a46d300 github.com/thanhpk/randstr v1.0.6 github.com/tidwall/gjson v1.19.0 - github.com/tidwall/sjson v1.2.5 // indirect + github.com/tidwall/sjson v1.2.5 github.com/tiktoken-go/tokenizer v0.8.1 // indirect github.com/waffo-com/waffo-go v1.3.2 github.com/yapingcat/gomedia v0.0.0-20240906162731-17feea57090c @@ -100,7 +100,7 @@ github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.21 // indirect github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.21 // indirect github.com/beorn7/perks v1.0.1 // indirect - github.com/boombuler/barcode v1.1.0 // indirect + github.com/boombuler/barcode v1.1.0 github.com/bytedance/sonic v1.14.1 // indirect github.com/bytedance/sonic/loader v0.3.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect diff -ruN a/middleware/sixbee_visitor.go b/middleware/sixbee_visitor.go --- a/middleware/sixbee_visitor.go +++ b/middleware/sixbee_visitor.go @@ -0,0 +1,97 @@ +package middleware + +// 6bee patch: link anonymous visitors to accounts for traffic attribution. +// +// The first-party analytics script on www.6bee.ai and console.6bee.ai keeps an +// anonymous visitor ID in the sb_vid cookie on .6bee.ai. When a signed-in console request +// carries that cookie, this records (account, SHA-256 of the visitor ID) in +// sixbee.user_visitors, so we can tell which channel an +// account first came from. Without that table (or on a +// database other than PostgreSQL) this does nothing. The raw cookie value is never stored. + +import ( + "crypto/sha256" + "encoding/hex" + "regexp" + "strconv" + "sync" + "sync/atomic" + "time" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/model" + + "github.com/gin-gonic/gin" +) + +var ( + sixbeeVisitorIDRe = regexp.MustCompile(`^[A-Za-z0-9_-]{16,64}$`) + sixbeeVisitorSeen sync.Map // "user:hash" -> unix time of the last write + sixbeeVisitorOn atomic.Bool // sixbee.user_visitors exists + sixbeeVisitorChecked atomic.Int64 // unix time of the last existence check + sixbeeVisitorPruned atomic.Int64 // unix time of the last cache prune + sixbeeVisitorLogOnce sync.Once + sixbeeVisitorRecheck = int64(300) // look for the table again every 5 min until it exists + sixbeeVisitorRewrite = int64(3600) +) + +func sixbeeVisitorTableReady(now int64) bool { + if sixbeeVisitorOn.Load() { + return true + } + last := sixbeeVisitorChecked.Load() + if now-last < sixbeeVisitorRecheck || !sixbeeVisitorChecked.CompareAndSwap(last, now) { + return false + } + if model.DB == nil || !common.UsingMainDatabase(common.DatabaseTypePostgreSQL) { + return false + } + var ok bool + if err := model.DB.Raw("select to_regclass('sixbee.user_visitors') is not null").Scan(&ok).Error; err == nil && ok { + sixbeeVisitorOn.Store(true) + sixbeeVisitorLogOnce.Do(func() { common.SysLog("6bee: visitor attribution enabled (sixbee.user_visitors)") }) + } + return sixbeeVisitorOn.Load() +} + +// SixbeeVisitorLink runs after the handler, so the user ID set by UserAuth/AdminAuth is known. +func SixbeeVisitorLink() gin.HandlerFunc { + return func(c *gin.Context) { + c.Next() + uid := c.GetInt("id") + if uid <= 0 || c.Writer.Status() >= 400 { + return + } + raw, err := c.Cookie("sb_vid") + if err != nil || !sixbeeVisitorIDRe.MatchString(raw) { + return + } + sum := sha256.Sum256([]byte(raw)) + hash := hex.EncodeToString(sum[:]) + key := strconv.Itoa(uid) + ":" + hash + now := time.Now().Unix() + if last, ok := sixbeeVisitorSeen.Load(key); ok && now-last.(int64) < sixbeeVisitorRewrite { + return + } + if !sixbeeVisitorTableReady(now) { + return + } + sixbeeVisitorSeen.Store(key, now) + if last := sixbeeVisitorPruned.Load(); now-last >= sixbeeVisitorRewrite && sixbeeVisitorPruned.CompareAndSwap(last, now) { + sixbeeVisitorSeen.Range(func(k, v any) bool { // keep the cache to the last hour's browsers + if now-v.(int64) >= sixbeeVisitorRewrite { + sixbeeVisitorSeen.Delete(k) + } + return true + }) + } + go func() { + err := model.DB.Exec(`insert into sixbee.user_visitors (user_id, visitor_hash) values (?, ?) + on conflict (user_id, visitor_hash) do update set last_seen = now()`, uid, hash).Error + if err != nil { + sixbeeVisitorSeen.Delete(key) + common.SysLog("6bee: visitor link failed: " + err.Error()) + } + }() + } +} diff -ruN a/relay/common/override.go b/relay/common/override.go --- a/relay/common/override.go +++ b/relay/common/override.go @@ -201,7 +201,7 @@ } func ApplyParamOverrideWithRelayInfo(jsonData []byte, info *RelayInfo) ([]byte, error) { - paramOverride := getParamOverrideMap(info) + paramOverride := sixbeeExpandAccountPlaceholder(getParamOverrideMap(info), info) // 6bee: {{sixbee_account}} if len(paramOverride) == 0 { return jsonData, nil } diff -ruN a/relay/common/sixbee_account.go b/relay/common/sixbee_account.go --- a/relay/common/sixbee_account.go +++ b/relay/common/sixbee_account.go @@ -0,0 +1,136 @@ +package common + +// 6bee: per-account pseudonymous ID for upstream requests. +// +// Any string value in a channel's param override may contain the placeholder +// {{sixbee_account}}. Before the override is applied, the placeholder is replaced +// with a stable, non-reversible ID of the requesting account, for example +// +// {"operations":[{"mode":"set","path":"user_id","value":"{{sixbee_account}}"}]} +// {"safety_identifier":"{{sixbee_account}}","cache_salt":"{{sixbee_account}}"} +// +// ID = "6b_" + first 32 hex chars of HMAC-SHA256(key, "account:"), +// key = SHA-256("6bee-account-pseudonym|" + SIXBEE_ACCOUNT_ID_SECRET), falling back to +// CRYPTO_SECRET when the variable is unset. Keep the secret fixed: changing it gives +// every account a new ID upstream. +// +// Only the channel's override config is rewritten (copy-on-write; the cached channel +// map is never modified). The client request body is not searched, so a prompt that +// contains the placeholder text is forwarded unchanged. + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "os" + "strconv" + "strings" + "sync" + + "github.com/QuantumNous/new-api/common" +) + +const SixbeeAccountPlaceholder = "{{sixbee_account}}" + +var ( + sixbeeAccountKeyOnce sync.Once + sixbeeAccountKey []byte +) + +func sixbeeAccountKeyBytes() []byte { + sixbeeAccountKeyOnce.Do(func() { + secret, source := strings.TrimSpace(os.Getenv("SIXBEE_ACCOUNT_ID_SECRET")), "SIXBEE_ACCOUNT_ID_SECRET" + if secret == "" { + secret, source = common.CryptoSecret, "CRYPTO_SECRET" + } + sum := sha256.Sum256([]byte("6bee-account-pseudonym|" + secret)) + sixbeeAccountKey = sum[:] + common.SysLog("6bee account IDs: key derived from " + source) + }) + return sixbeeAccountKey +} + +// SixbeeAccountID returns the upstream-facing ID of an account. +func SixbeeAccountID(userID int) string { + mac := hmac.New(sha256.New, sixbeeAccountKeyBytes()) + mac.Write([]byte("account:" + strconv.Itoa(userID))) + return "6b_" + hex.EncodeToString(mac.Sum(nil))[:32] +} + +// sixbeeExpandAccountPlaceholder returns paramOverride with every {{sixbee_account}} +// replaced by the account's ID. Without a placeholder it returns the same map. +func sixbeeExpandAccountPlaceholder(paramOverride map[string]any, info *RelayInfo) map[string]any { + if len(paramOverride) == 0 || info == nil { + return paramOverride + } + id := "" + accountID := func() string { + if id == "" { + id = SixbeeAccountID(info.UserId) + } + return id + } + if out, changed := sixbeeSubstitute(paramOverride, accountID); changed { + return out.(map[string]any) + } + return paramOverride +} + +// sixbeeSubstitute copies only the maps and slices on the path to a changed string. +func sixbeeSubstitute(v any, accountID func() string) (any, bool) { + switch t := v.(type) { + case string: + if strings.Contains(t, SixbeeAccountPlaceholder) { + return strings.ReplaceAll(t, SixbeeAccountPlaceholder, accountID()), true + } + case map[string]any: + var out map[string]any + for k, e := range t { + ne, changed := sixbeeSubstitute(e, accountID) + if !changed { + continue + } + if out == nil { + out = make(map[string]any, len(t)) + for k2, e2 := range t { + out[k2] = e2 + } + } + out[k] = ne + } + if out != nil { + return out, true + } + case []map[string]any: + var out []map[string]any + for i, e := range t { + ne, changed := sixbeeSubstitute(e, accountID) + if !changed { + continue + } + if out == nil { + out = append([]map[string]any(nil), t...) + } + out[i] = ne.(map[string]any) + } + if out != nil { + return out, true + } + case []any: + var out []any + for i, e := range t { + ne, changed := sixbeeSubstitute(e, accountID) + if !changed { + continue + } + if out == nil { + out = append([]any(nil), t...) + } + out[i] = ne + } + if out != nil { + return out, true + } + } + return v, false +} diff -ruN a/relay/request_billing.go b/relay/request_billing.go --- a/relay/request_billing.go +++ b/relay/request_billing.go @@ -56,15 +56,31 @@ } info.SetEstimatePromptTokens(tokens) + // 6bee: context budget and per-account input TPM (service/sixbee_limits.go) + if apiErr := service.SixbeeAdmitTokens(c, info, tokens); apiErr != nil { + return apiErr + } + priceData, err := helper.ModelPriceHelper(c, info, tokens, meta) if err != nil { + service.SixbeeReleaseTokens(c, info) return types.NewError(err, types.ErrorCodeModelPriceError, types.ErrOptionWithStatusCode(http.StatusBadRequest)) } if priceData.FreeModel { logger.LogInfo(c, common.LogText("model %s is free, skipping pre-consume", info.OriginModelName)) return nil } - return service.PreConsumeBilling(c, priceData.QuotaToPreConsume, info) + // 6bee: reserve the maximum billable output as well as the input (plan A07/A10) + outputReserve, err := service.SixbeeOutputReserve(info, priceData) + if err != nil { + service.SixbeeReleaseTokens(c, info) + return types.NewError(err, types.ErrorCodeModelPriceError, types.ErrOptionWithStatusCode(http.StatusBadRequest)) + } + if apiErr := service.PreConsumeBilling(c, priceData.QuotaToPreConsume+outputReserve, info); apiErr != nil { + service.SixbeeReleaseTokens(c, info) + return apiErr + } + return nil } // RefundFailedRequestBilling applies the common final-failure policy after all diff -ruN a/relaykit/dto/user_settings.go b/relaykit/dto/user_settings.go --- a/relaykit/dto/user_settings.go +++ b/relaykit/dto/user_settings.go @@ -18,6 +18,11 @@ SidebarModules string `json:"sidebar_modules,omitempty"` // SidebarModules 左侧边栏模块配置 BillingPreference string `json:"billing_preference,omitempty"` // BillingPreference 扣费策略（订阅/钱包） Language string `json:"language,omitempty"` // Language 用户语言偏好 (zh, en) + // 6bee: where notices go and which broadcasts the user accepts; unset means on + EmailNotifyEnabled *bool `json:"email_notify_enabled,omitempty"` + TelegramNotifyEnabled *bool `json:"telegram_notify_enabled,omitempty"` + AnnouncementNotifyEnabled *bool `json:"announcement_notify_enabled,omitempty"` + PromoNotifyEnabled *bool `json:"promo_notify_enabled,omitempty"` } var ( diff -ruN a/router/api-router.go b/router/api-router.go --- a/router/api-router.go +++ b/router/api-router.go @@ -17,6 +17,7 @@ func SetApiRouter(router *gin.Engine) { apiRouter := router.Group("/api") apiRouter.Use(middleware.RouteTag("api")) + apiRouter.Use(middleware.SixbeeVisitorLink()) // 6bee: link sb_vid visitor IDs to signed-in accounts apiRouter.Use(gzip.Gzip(gzip.DefaultCompression)) apiRouter.Use(middleware.AccessTokenAudit()) apiRouter.Use(middleware.BodyStorageCleanup()) // 清理请求体存储 @@ -439,4 +440,5 @@ deploymentsRoute.DELETE("/:id", controller.DeleteDeployment) } } + setSixbeeNotifyRoutes(apiRouter) // 6bee: Telegram linking, unsubscribe link, owner broadcasts } diff -ruN a/router/main.go b/router/main.go --- a/router/main.go +++ b/router/main.go @@ -19,6 +19,7 @@ SetTaskPluginProtocolRouter(router) SetVideoRouter(router) SetTaskRouter(router) + SetSixbeePayRouter(router) // 6bee: pay.6bee.ai checkout pluginDispatcher := SetPluginRouter(router) frontendBaseUrl := os.Getenv("FRONTEND_BASE_URL") if common.IsMasterNode && frontendBaseUrl != "" { diff -ruN a/router/sixbee-notify-router.go b/router/sixbee-notify-router.go --- a/router/sixbee-notify-router.go +++ b/router/sixbee-notify-router.go @@ -0,0 +1,35 @@ +package router + +import ( + "github.com/QuantumNous/new-api/controller" + "github.com/QuantumNous/new-api/middleware" + "github.com/QuantumNous/new-api/service" + + "github.com/gin-gonic/gin" +) + +// setSixbeeNotifyRoutes adds the 6bee notification routes (service/sixbee_notify.go, +// service/sixbee_broadcast.go) to the /api group: +// +// /api/user/sixbee/telegram the signed-in user's Telegram link (browser session only) +// /api/sixbee/notify/unsubscribe the signed link in broadcast emails, no session +// /api/sixbee/broadcast announcements and promotions, root only +func setSixbeeNotifyRoutes(api *gin.RouterGroup) { + service.SixbeeNotifyStart() + + telegram := api.Group("/user/sixbee/telegram") + telegram.Use(middleware.UserAuth(), middleware.DisableCache()) + telegram.GET("", controller.SixbeeTelegramStatus) + telegram.POST("/link", middleware.CriticalRateLimit(), controller.SixbeeTelegramLinkStart) + telegram.DELETE("", controller.SixbeeTelegramUnlink) + + api.GET("/sixbee/notify/unsubscribe", middleware.CriticalRateLimit(), middleware.DisableCache(), controller.SixbeeUnsubscribe) + api.POST("/sixbee/notify/unsubscribe", middleware.CriticalRateLimit(), middleware.DisableCache(), controller.SixbeeUnsubscribe) + + broadcast := api.Group("/sixbee/broadcast") + broadcast.Use(middleware.RootAuth(), middleware.DisableCache()) + broadcast.GET("", controller.SixbeeBroadcastList) + broadcast.POST("", controller.SixbeeBroadcastSend) + broadcast.POST("/:id/resume", controller.SixbeeBroadcastResume) + broadcast.POST("/:id/stop", controller.SixbeeBroadcastStop) +} diff -ruN a/router/sixbee-pay-router.go b/router/sixbee-pay-router.go --- a/router/sixbee-pay-router.go +++ b/router/sixbee-pay-router.go @@ -0,0 +1,18 @@ +package router + +import ( + "github.com/QuantumNous/new-api/controller" + "github.com/QuantumNous/new-api/middleware" + + "github.com/gin-gonic/gin" +) + +// SetSixbeePayRouter adds the pay.6bee.ai checkout pages and the payment +// provider notification relay (6bee: controller/sixbee_pay.go). The handlers +// answer 404 unless the request is for the SIXBEE_PAY_ORIGIN host. +func SetSixbeePayRouter(router *gin.Engine) { + router.GET("/pay/:token", controller.SixbeePayPage) + router.POST("/pay/:token", middleware.AnonymousRequestBodyLimit(), controller.SixbeePayPage) + router.GET("/pay/:token/status", controller.SixbeePayStatus) + router.POST("/.well-known/epay-provider/:type/:trade_no", controller.SixbeeEpayProviderRelay) +} diff -ruN a/service/sixbee_broadcast.go b/service/sixbee_broadcast.go --- a/service/sixbee_broadcast.go +++ b/service/sixbee_broadcast.go @@ -0,0 +1,430 @@ +package service + +// 6bee patch: announcements and promotions sent by the site owner to every account, by email and +// Telegram (service/sixbee_notify.go). One broadcast runs at a time, in account-id order, and +// records how far it got, so one that a restart interrupts can be resumed without sending twice. +// +// SIXBEE_BROADCAST_EMAIL_INTERVAL_MS pause after each email, default 600 (Resend allows 2 a second) + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/base64" + "html" + "net/url" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + "unicode/utf8" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/model" + "github.com/QuantumNous/new-api/relaykit/dto" + + "github.com/bytedance/gopkg/util/gopool" +) + +const ( + SixbeeBroadcastAnnouncement = "announcement" + SixbeeBroadcastPromo = "promo" + + SixbeeBroadcastRunning = "running" + SixbeeBroadcastDone = "done" + SixbeeBroadcastCancelled = "cancelled" + SixbeeBroadcastInterrupted = "interrupted" +) + +// SixbeeBroadcast is one announcement or promotion and how far its delivery got. +type SixbeeBroadcast struct { + Id int `json:"id" gorm:"primaryKey"` + Kind string `json:"kind" gorm:"type:varchar(16)"` + Title string `json:"title" gorm:"type:varchar(255)"` + Content string `json:"content" gorm:"type:text"` + LinkUrl string `json:"link_url" gorm:"type:varchar(512)"` + Status string `json:"status" gorm:"type:varchar(16);index"` + Recipients int `json:"recipients"` // accounts that received it on at least one channel + EmailSent int `json:"email_sent"` // emails accepted by the mail provider + TelegramSent int `json:"telegram_sent"` // Telegram messages accepted + Failed int `json:"failed"` // sends that returned an error + LastUserId int `json:"last_user_id"` // every account up to this id has been handled + CreatedBy int `json:"created_by"` + CreatedAt int64 `json:"created_at"` + FinishedAt int64 `json:"finished_at"` +} + +func (SixbeeBroadcast) TableName() string { return "sixbee_broadcasts" } + +// SixbeeBroadcastAudience counts who a broadcast of one kind would reach right now. +type SixbeeBroadcastAudience struct { + Accounts int `json:"accounts"` // enabled accounts that accept this kind on some channel + Email int `json:"email"` + Telegram int `json:"telegram"` +} + +var ( + sixbeeBroadcastMutex sync.Mutex + sixbeeBroadcastCancel atomic.Bool +) + +// sixbeeBroadcastRecover marks a broadcast that the previous process left running. +func sixbeeBroadcastRecover() { + if !common.IsMasterNode { + return + } + model.DB.Model(&SixbeeBroadcast{}).Where("status = ?", SixbeeBroadcastRunning).Update("status", SixbeeBroadcastInterrupted) +} + +func sixbeeBroadcastValidate(broadcast *SixbeeBroadcast) error { + broadcast.Title = strings.TrimSpace(broadcast.Title) + broadcast.Content = strings.TrimSpace(broadcast.Content) + broadcast.LinkUrl = strings.TrimSpace(broadcast.LinkUrl) + if broadcast.Kind != SixbeeBroadcastAnnouncement && broadcast.Kind != SixbeeBroadcastPromo { + return common.NewMessage("Choose announcement or promotion") + } + if broadcast.Title == "" || utf8.RuneCountInString(broadcast.Title) > 100 { + return common.NewMessage("The title must be 1 to 100 characters") + } + if broadcast.Content == "" || utf8.RuneCountInString(broadcast.Content) > 3000 { + return common.NewMessage("The content must be 1 to 3000 characters") + } + if broadcast.LinkUrl != "" { + link, err := url.Parse(broadcast.LinkUrl) + if err != nil || link.Scheme != "https" || link.Host == "" || len(broadcast.LinkUrl) > 500 { + return common.NewMessage("The link must be an https:// address") + } + } + return nil +} + +// sixbeeUnsubscribeSignature signs "account stops receiving kind". It is the only credential of +// the unsubscribe link in broadcast emails, so it is keyed by the server's secret. +func sixbeeUnsubscribeSignature(userId int, kind string) string { + key := sha256.Sum256([]byte("6bee-notice-unsubscribe|" + common.CryptoSecret)) + mac := hmac.New(sha256.New, key[:]) + mac.Write([]byte(strconv.Itoa(userId) + "|" + kind)) + return base64.RawURLEncoding.EncodeToString(mac.Sum(nil)[:20]) +} + +// SixbeeUnsubscribeValid checks an unsubscribe link's signature. +func SixbeeUnsubscribeValid(userId int, kind string, signature string) bool { + if userId <= 0 || (kind != SixbeeBroadcastAnnouncement && kind != SixbeeBroadcastPromo) { + return false + } + return hmac.Equal([]byte(signature), []byte(sixbeeUnsubscribeSignature(userId, kind))) +} + +func sixbeeChinese(language string) bool { + return language == "" || strings.HasPrefix(strings.ToLower(language), "zh") +} + +// sixbeeBroadcastEmail renders the email body. The owner's text is plain text: it is escaped, +// and line breaks are kept. +func sixbeeBroadcastEmail(broadcast *SixbeeBroadcast, userId int, language string) string { + zh := sixbeeChinese(language) + var body strings.Builder + body.WriteString(`
`) + body.WriteString(`
6bee.ai
`) + body.WriteString(`

` + html.EscapeString(broadcast.Title) + `

`) + body.WriteString(`
` + strings.ReplaceAll(html.EscapeString(broadcast.Content), "\n", "
") + `
`) + if broadcast.LinkUrl != "" { + label := "View details" + if zh { + label = "查看详情" + } + body.WriteString(`

` + label + `

`) + } + query := url.Values{"u": {strconv.Itoa(userId)}, "k": {broadcast.Kind}, "s": {sixbeeUnsubscribeSignature(userId, broadcast.Kind)}} + unsubscribe := html.EscapeString(PaymentReturnURL("/api/sixbee/notify/unsubscribe?" + query.Encode())) + settings := html.EscapeString(PaymentReturnURL("/profile")) + body.WriteString(`

`) + switch { + case zh && broadcast.Kind == SixbeeBroadcastPromo: + body.WriteString(`你收到这封邮件，是因为你注册了 6bee.ai 并接收活动通知。退订活动通知 · 通知设置`) + case zh: + body.WriteString(`你收到这封邮件，是因为你注册了 6bee.ai。不再接收公告 · 通知设置`) + case broadcast.Kind == SixbeeBroadcastPromo: + body.WriteString(`You are receiving this because you have a 6bee.ai account and accept promotions. Unsubscribe from promotions · Notification settings`) + default: + body.WriteString(`You are receiving this because you have a 6bee.ai account. Stop announcements · Notification settings`) + } + body.WriteString(`

`) + return body.String() +} + +func sixbeeBroadcastTelegram(broadcast *SixbeeBroadcast, language string) string { + zh := sixbeeChinese(language) + text := "" + html.EscapeString(broadcast.Title) + "\n\n" + html.EscapeString(broadcast.Content) + if broadcast.LinkUrl != "" { + label := "View details" + if zh { + label = "查看详情" + } + text += "\n\n" + label + " →" + } + if broadcast.Kind == SixbeeBroadcastPromo { + if zh { + text += "\n\n不想再收到活动通知：/nopromo" + } else { + text += "\n\nStop promotions: /nopromo" + } + } + return text +} + +// sixbeeBroadcastChannels says where one account accepts a broadcast of this kind: the email +// address ("" for none) and the Telegram chat (0 for none). +func sixbeeBroadcastChannels(kind string, user *model.User, setting dto.UserSetting, chats map[int]int64) (string, int64) { + accepts := setting.AnnouncementNotifyEnabled + if kind == SixbeeBroadcastPromo { + accepts = setting.PromoNotifyEnabled + } + if !sixbeeSwitchOn(accepts) { + return "", 0 + } + address := "" + if sixbeeSwitchOn(setting.EmailNotifyEnabled) { + address = setting.NotificationEmail + if address == "" { + address = user.Email + } + } + var chat int64 + if sixbeeSwitchOn(setting.TelegramNotifyEnabled) && SixbeeTelegramBot() != "" { + chat = chats[user.Id] + } + return address, chat +} + +// sixbeeBroadcastDeliver sends one broadcast to one account and reports what was accepted. +func sixbeeBroadcastDeliver(broadcast *SixbeeBroadcast, user *model.User, address string, chat int64, emailPause time.Duration) (emailSent bool, telegramSent bool, failed int) { + language := user.GetSetting().Language + if address != "" { + if err := common.SendEmail(broadcast.Title, address, sixbeeBroadcastEmail(broadcast, user.Id, language)); err != nil { + failed++ + common.SysError("6bee: broadcast " + strconv.Itoa(broadcast.Id) + " email to user " + strconv.Itoa(user.Id) + " failed: " + err.Error()) + } else { + emailSent = true + } + time.Sleep(emailPause) + } + if chat != 0 { + if err := sixbeeTelegramSend(chat, sixbeeBroadcastTelegram(broadcast, language)); err != nil { + failed++ + common.SysError("6bee: broadcast " + strconv.Itoa(broadcast.Id) + " Telegram to user " + strconv.Itoa(user.Id) + " failed: " + err.Error()) + } else { + telegramSent = true + } + time.Sleep(40 * time.Millisecond) + } + return emailSent, telegramSent, failed +} + +func sixbeeTelegramChats() map[int]int64 { + var links []SixbeeTelegramLink + model.DB.Find(&links) + chats := make(map[int]int64, len(links)) + for _, link := range links { + chats[link.UserId] = link.ChatId + } + return chats +} + +// SixbeeBroadcastAudienceOf counts the accounts each kind of broadcast would reach. +func SixbeeBroadcastAudienceOf() (map[string]SixbeeBroadcastAudience, error) { + audience := map[string]SixbeeBroadcastAudience{} + chats := sixbeeTelegramChats() + lastId := 0 + for { + var users []model.User + if err := model.DB.Select("id", "email", "setting").Where("id > ? AND status = ?", lastId, common.UserStatusEnabled).Order("id").Limit(500).Find(&users).Error; err != nil { + return nil, err + } + if len(users) == 0 { + return audience, nil + } + for i := range users { + lastId = users[i].Id + setting := users[i].GetSetting() + for _, kind := range []string{SixbeeBroadcastAnnouncement, SixbeeBroadcastPromo} { + address, chat := sixbeeBroadcastChannels(kind, &users[i], setting, chats) + count := audience[kind] + if address != "" || chat != 0 { + count.Accounts++ + } + if address != "" { + count.Email++ + } + if chat != 0 { + count.Telegram++ + } + audience[kind] = count + } + } + } +} + +// SixbeeBroadcastTest sends the broadcast to its author only, whatever their own switches say, +// and reports which channels took it. +func SixbeeBroadcastTest(broadcast *SixbeeBroadcast, userId int) (emailSent bool, telegramSent bool, err error) { + if err := sixbeeBroadcastValidate(broadcast); err != nil { + return false, false, err + } + user, err := model.GetUserById(userId, false) + if err != nil { + return false, false, err + } + setting := user.GetSetting() + address := setting.NotificationEmail + if address == "" { + address = user.Email + } + var chat int64 + if link, linked := SixbeeTelegramLinkOf(userId); linked && SixbeeTelegramBot() != "" { + chat = link.ChatId + } + if address == "" && chat == 0 { + return false, false, common.NewMessage("Your account has no email address or linked Telegram to send a test to") + } + preview := *broadcast + preview.Title = "[TEST] " + broadcast.Title + emailSent, telegramSent, failed := sixbeeBroadcastDeliver(&preview, user, address, chat, 0) + if failed > 0 && !emailSent && !telegramSent { + return false, false, common.NewMessage("The test could not be sent; see the server log") + } + return emailSent, telegramSent, nil +} + +// SixbeeBroadcastStart records a broadcast and starts sending it to every account. +func SixbeeBroadcastStart(broadcast *SixbeeBroadcast) error { + if err := sixbeeBroadcastValidate(broadcast); err != nil { + return err + } + sixbeeBroadcastMutex.Lock() + defer sixbeeBroadcastMutex.Unlock() + var running int64 + if err := model.DB.Model(&SixbeeBroadcast{}).Where("status = ?", SixbeeBroadcastRunning).Count(&running).Error; err != nil { + return err + } + if running > 0 { + return common.NewMessage("Another broadcast is still being sent") + } + broadcast.Id = 0 + broadcast.Status = SixbeeBroadcastRunning + broadcast.CreatedAt = time.Now().Unix() + broadcast.Recipients, broadcast.EmailSent, broadcast.TelegramSent, broadcast.Failed, broadcast.LastUserId, broadcast.FinishedAt = 0, 0, 0, 0, 0, 0 + if err := model.DB.Create(broadcast).Error; err != nil { + return err + } + sixbeeBroadcastCancel.Store(false) + id := broadcast.Id + gopool.Go(func() { sixbeeBroadcastRun(id) }) + return nil +} + +// SixbeeBroadcastResume continues a cancelled or interrupted broadcast after the last account +// it handled. +func SixbeeBroadcastResume(id int) error { + sixbeeBroadcastMutex.Lock() + defer sixbeeBroadcastMutex.Unlock() + var running int64 + if err := model.DB.Model(&SixbeeBroadcast{}).Where("status = ?", SixbeeBroadcastRunning).Count(&running).Error; err != nil { + return err + } + if running > 0 { + return common.NewMessage("Another broadcast is still being sent") + } + result := model.DB.Model(&SixbeeBroadcast{}).Where("id = ? AND status IN ?", id, []string{SixbeeBroadcastCancelled, SixbeeBroadcastInterrupted}). + Updates(map[string]any{"status": SixbeeBroadcastRunning, "finished_at": 0}) + if result.Error != nil { + return result.Error + } + if result.RowsAffected == 0 { + return common.NewMessage("This broadcast cannot be resumed") + } + sixbeeBroadcastCancel.Store(false) + gopool.Go(func() { sixbeeBroadcastRun(id) }) + return nil +} + +// SixbeeBroadcastStop asks the running broadcast to stop after the account it is on. +func SixbeeBroadcastStop(id int) error { + var broadcast SixbeeBroadcast + if err := model.DB.Where("id = ? AND status = ?", id, SixbeeBroadcastRunning).Limit(1).Find(&broadcast).Error; err != nil { + return err + } + if broadcast.Id == 0 { + return common.NewMessage("This broadcast is not being sent") + } + sixbeeBroadcastCancel.Store(true) + return nil +} + +// SixbeeBroadcastList returns the latest broadcasts, newest first. +func SixbeeBroadcastList() ([]SixbeeBroadcast, error) { + var broadcasts []SixbeeBroadcast + err := model.DB.Order("id desc").Limit(30).Find(&broadcasts).Error + return broadcasts, err +} + +func sixbeeBroadcastRun(id int) { + var broadcast SixbeeBroadcast + if err := model.DB.First(&broadcast, id).Error; err != nil { + common.SysError("6bee: broadcast " + strconv.Itoa(id) + " not found: " + err.Error()) + return + } + emailPause := time.Duration(common.GetEnvOrDefault("SIXBEE_BROADCAST_EMAIL_INTERVAL_MS", 600)) * time.Millisecond + chats := sixbeeTelegramChats() + status := SixbeeBroadcastDone + var failure error + save := func() { + model.DB.Model(&SixbeeBroadcast{}).Where("id = ?", broadcast.Id).Updates(map[string]any{ + "recipients": broadcast.Recipients, "email_sent": broadcast.EmailSent, "telegram_sent": broadcast.TelegramSent, + "failed": broadcast.Failed, "last_user_id": broadcast.LastUserId, + }) + } +batches: + for { + var users []model.User + if err := model.DB.Select("id", "email", "setting").Where("id > ? AND status = ?", broadcast.LastUserId, common.UserStatusEnabled).Order("id").Limit(200).Find(&users).Error; err != nil { + status, failure = SixbeeBroadcastInterrupted, err + break + } + if len(users) == 0 { + break + } + for i := range users { + if sixbeeBroadcastCancel.Load() { + status = SixbeeBroadcastCancelled + break batches + } + user := &users[i] + address, chat := sixbeeBroadcastChannels(broadcast.Kind, user, user.GetSetting(), chats) + if address != "" || chat != 0 { + emailSent, telegramSent, failed := sixbeeBroadcastDeliver(&broadcast, user, address, chat, emailPause) + broadcast.Failed += failed + if emailSent { + broadcast.EmailSent++ + } + if telegramSent { + broadcast.TelegramSent++ + } + if emailSent || telegramSent { + broadcast.Recipients++ + } + } + broadcast.LastUserId = user.Id + save() + } + } + save() + model.DB.Model(&SixbeeBroadcast{}).Where("id = ?", broadcast.Id).Updates(map[string]any{"status": status, "finished_at": time.Now().Unix()}) + summary := "6bee: broadcast " + strconv.Itoa(broadcast.Id) + " " + status + ": " + strconv.Itoa(broadcast.Recipients) + " accounts, " + + strconv.Itoa(broadcast.EmailSent) + " emails, " + strconv.Itoa(broadcast.TelegramSent) + " Telegram, " + strconv.Itoa(broadcast.Failed) + " failed" + if failure != nil { + common.SysError(summary + ": " + failure.Error()) + return + } + common.SysLog(summary) +} diff -ruN a/service/sixbee_limits.go b/service/sixbee_limits.go --- a/service/sixbee_limits.go +++ b/service/sixbee_limits.go @@ -0,0 +1,515 @@ +package service + +// 6bee account limits — local patch on top of New API v1.0.0-rc.47 (not upstream). +// Implements Appendix A of the 6bee platform plan, keyed by user id so that extra +// API keys cannot bypass an account's totals: +// +// SIXBEE_MAX_CONCURRENCY in-flight relay requests per account (default 2, 0 = off) +// SIXBEE_INPUT_TPM uncached input tokens per rolling 60 s per account; +// a request is admitted while the window is below it (default 1000000, 0 = off) +// SIXBEE_MODEL_LIMITS JSON list, first match wins ("name" or "prefix*"): +// [{"model":"glm-5.3","context":1048576,"max_output":65536,"default_output":32768}] +// SIXBEE_MAX_CONTEXT_TOKENS context budget of models not in the list (default 131072, 0 = off) +// SIXBEE_MAX_OUTPUT_TOKENS output cap of models not in the list (default 16384, 0 = off) +// SIXBEE_DEFAULT_OUTPUT_TOKENS output injected when a request sets none (unlisted) (default 8192) +// +// Output: a max_tokens above the model's cap is lowered to the cap (header X-6bee-Output-Cap); +// a request without one gets the model's default, shrunk to fit the context budget if needed. +// The resulting output ceiling is added to the pre-consumed quota (A07/A10): rc.47 reserves only +// the input estimate and lets settlement drive the balance negative. +// TPM: the input estimate is reserved at admission and replaced by the upstream-reported +// uncached input tokens when the request is billed (cache hits do not count). +// State lives in Redis when Redis is enabled (fail closed on Redis errors), otherwise in +// process memory. Single-node assumption: concurrency slots are cleared on process start. + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "os" + "strconv" + "strings" + "sync" + "time" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/logger" + relaycommon "github.com/QuantumNous/new-api/relay/common" + "github.com/QuantumNous/new-api/relaykit/dto" + "github.com/QuantumNous/new-api/relaykit/types" + hosttypes "github.com/QuantumNous/new-api/types" + + "github.com/gin-gonic/gin" + "github.com/go-redis/redis/v8" +) + +const ( + sixbeeTPMWindow = 60 * time.Second + sixbeeSlotStaleAfter = 30 * time.Minute // longer than any allowed request; evicts slots of crashed requests + sixbeeMinOutput = 1024 // smallest output a shrunk default may leave + sixbeeCtxTPMMember = "sixbee_tpm_member" + sixbeeCtxTPMCommitted = "sixbee_tpm_committed" + sixbeeCtxOutputInjected = "sixbee_output_injected" +) + +type sixbeeModelLimit struct { + Model string `json:"model"` + Context int `json:"context"` + MaxOutput int `json:"max_output"` + DefaultOutput int `json:"default_output"` +} + +var ( + sixbeeOnce sync.Once + sixbeeMaxConcurrency int + sixbeeInputTPM int + sixbeeFallback sixbeeModelLimit + sixbeeModelTable []sixbeeModelLimit + + sixbeeMu sync.Mutex + sixbeeMemSlots = map[int]int{} + sixbeeMemTPM = map[int][]sixbeeTPMEntry{} +) + +type sixbeeTPMEntry struct { + member string + at time.Time + tokens int +} + +func sixbeeInit() { + sixbeeOnce.Do(func() { + sixbeeMaxConcurrency = common.GetEnvOrDefault("SIXBEE_MAX_CONCURRENCY", 2) + sixbeeInputTPM = common.GetEnvOrDefault("SIXBEE_INPUT_TPM", 1000000) + sixbeeFallback = sixbeeModelLimit{ + Model: "*", + Context: common.GetEnvOrDefault("SIXBEE_MAX_CONTEXT_TOKENS", 131072), + MaxOutput: common.GetEnvOrDefault("SIXBEE_MAX_OUTPUT_TOKENS", 16384), + DefaultOutput: common.GetEnvOrDefault("SIXBEE_DEFAULT_OUTPUT_TOKENS", 8192), + } + sixbeeFallback = sixbeeNormalize(sixbeeFallback) + if raw := strings.TrimSpace(os.Getenv("SIXBEE_MODEL_LIMITS")); raw != "" { + var table []sixbeeModelLimit + if err := json.Unmarshal([]byte(raw), &table); err != nil { + common.SysError("6bee limits: SIXBEE_MODEL_LIMITS is not valid JSON, using the fallback for every model: " + err.Error()) + } else { + for _, l := range table { + if strings.TrimSpace(l.Model) == "" { + continue + } + sixbeeModelTable = append(sixbeeModelTable, sixbeeNormalize(l)) + } + } + } + if common.RedisEnabled && common.RDB != nil { + // Single node: any slot left in Redis belongs to a previous process. + ctx := context.Background() + iter := common.RDB.Scan(ctx, 0, "sixbee:conc:*", 100).Iterator() + for iter.Next(ctx) { + common.RDB.Del(ctx, iter.Val()) + } + } + models := make([]string, 0, len(sixbeeModelTable)+1) + for _, l := range append(sixbeeModelTable, sixbeeFallback) { + models = append(models, fmt.Sprintf("%s=%d/%d/%d", l.Model, l.Context, l.MaxOutput, l.DefaultOutput)) + } + common.SysLog(fmt.Sprintf("6bee limits: concurrency=%d input_tpm=%d (uncached) models[context/max_output/default_output]=%s redis=%v", + sixbeeMaxConcurrency, sixbeeInputTPM, strings.Join(models, " "), common.RedisEnabled)) + }) +} + +func sixbeeNormalize(l sixbeeModelLimit) sixbeeModelLimit { + l.Model = strings.TrimSpace(l.Model) + if l.DefaultOutput <= 0 || (l.MaxOutput > 0 && l.DefaultOutput > l.MaxOutput) { + l.DefaultOutput = l.MaxOutput + } + return l +} + +// sixbeeLimitFor returns the limits for the model name the client asked for. +func sixbeeLimitFor(model string) sixbeeModelLimit { + for _, l := range sixbeeModelTable { + if p, ok := strings.CutSuffix(l.Model, "*"); ok { + if strings.HasPrefix(model, p) { + return l + } + } else if l.Model == model { + return l + } + } + return sixbeeFallback +} + +func sixbeeErr(status int, code string, msg string) *types.NewAPIError { + return types.NewErrorWithStatusCode(errors.New(msg), types.ErrorCode(code), status, + types.ErrOptionWithSkipRetry(), types.ErrOptionWithNoRecordErrorLog()) +} + +func sixbeeLimiterDown(c *gin.Context, err error) *types.NewAPIError { + logger.LogError(c, "6bee limiter unavailable: "+err.Error()) + return sixbeeErr(http.StatusServiceUnavailable, "rate_limiter_unavailable", + "Request admission is temporarily unavailable. Please retry shortly.") +} + +// sixbeeMaxOutputOf returns the output ceiling the request asks for (0 = not set). +func sixbeeMaxOutputOf(req dto.Request) int { + switch r := req.(type) { + case *dto.GeneralOpenAIRequest: + return int(max(derefUint(r.MaxTokens), derefUint(r.MaxCompletionTokens))) + case *dto.ClaudeRequest: + return int(derefUint(r.MaxTokens)) + case *dto.OpenAIResponsesRequest: + return int(derefUint(r.MaxOutputTokens)) + } + return 0 +} + +// sixbeeSetInjectedOutput rewrites the output ceiling this patch injected. +func sixbeeSetInjectedOutput(req dto.Request, v uint) { + switch r := req.(type) { + case *dto.GeneralOpenAIRequest: + r.MaxTokens = &v + case *dto.ClaudeRequest: + r.MaxTokens = &v + sixbeeFitClaudeThinking(r) + case *dto.OpenAIResponsesRequest: + r.MaxOutputTokens = &v + } +} + +// sixbeeFitClaudeThinking keeps an Anthropic thinking budget below max_tokens after 6bee lowered +// max_tokens. Anthropic rejects budget_tokens >= max_tokens, and New API's reasoning conversion +// would otherwise raise max_tokens back above the cap to make room for the budget. +func sixbeeFitClaudeThinking(r *dto.ClaudeRequest) { + if r.Thinking == nil || r.Thinking.BudgetTokens == nil || r.MaxTokens == nil { + return + } + if limit := int(*r.MaxTokens) - 1; *r.Thinking.BudgetTokens > limit && limit >= sixbeeMinOutput { + r.Thinking.BudgetTokens = &limit + } +} + +func derefUint(p *uint) uint { + if p == nil { + return 0 + } + return *p +} + +// sixbeeApplyOutputCap lowers output ceilings above the model's cap and injects the model's +// default when absent, so both the upstream call and the pre-consumed quota are bounded. +func sixbeeApplyOutputCap(c *gin.Context, req dto.Request, lim sixbeeModelLimit) *types.NewAPIError { + if r, ok := req.(*dto.GeneralOpenAIRequest); ok && r.N != nil && *r.N > 1 { + return sixbeeErr(http.StatusBadRequest, "n_not_supported", "n > 1 is not supported; send separate requests instead.") + } + if lim.MaxOutput <= 0 { + return nil + } + capv := uint(lim.MaxOutput) + clamped := false + clamp := func(p **uint) { + if *p != nil && **p > capv { + v := capv + *p = &v + clamped = true + } + } + injected := false + switch r := req.(type) { + case *dto.GeneralOpenAIRequest: + clamp(&r.MaxTokens) + clamp(&r.MaxCompletionTokens) + injected = derefUint(r.MaxTokens) == 0 && derefUint(r.MaxCompletionTokens) == 0 + case *dto.ClaudeRequest: + clamp(&r.MaxTokens) + if clamped { + sixbeeFitClaudeThinking(r) + } + injected = derefUint(r.MaxTokens) == 0 + case *dto.OpenAIResponsesRequest: + clamp(&r.MaxOutputTokens) + injected = derefUint(r.MaxOutputTokens) == 0 + default: + return nil + } + if injected { + sixbeeSetInjectedOutput(req, uint(lim.DefaultOutput)) + c.Set(sixbeeCtxOutputInjected, true) + } + if clamped { + c.Header("X-6bee-Output-Cap", strconv.Itoa(lim.MaxOutput)) + } + return nil +} + +var sixbeeAcquireScript = redis.NewScript(` +redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', ARGV[2]) +if redis.call('ZCARD', KEYS[1]) >= tonumber(ARGV[3]) then return 0 end +redis.call('ZADD', KEYS[1], ARGV[1], ARGV[4]) +redis.call('EXPIRE', KEYS[1], ARGV[5]) +return 1`) + +// SixbeeAdmitRequest applies the model's output cap and takes one of the account's concurrency +// slots. Call right after GenRelayInfo; always call the returned release (it is never nil). +func SixbeeAdmitRequest(c *gin.Context, info *relaycommon.RelayInfo) (func(), *types.NewAPIError) { + sixbeeInit() + noop := func() {} + if info == nil || info.UserId == 0 { + return noop, nil + } + if apiErr := sixbeeApplyOutputCap(c, info.Request, sixbeeLimitFor(info.OriginModelName)); apiErr != nil { + return noop, apiErr + } + limit := sixbeeMaxConcurrency + if limit <= 0 { + return noop, nil + } + uid := info.UserId + tooMany := sixbeeErr(http.StatusTooManyRequests, "account_concurrency_limit", + fmt.Sprintf("This account already has %d requests in progress (limit %d). Retry when one finishes.", limit, limit)) + + if !common.RedisEnabled || common.RDB == nil { + sixbeeMu.Lock() + defer sixbeeMu.Unlock() + if sixbeeMemSlots[uid] >= limit { + c.Header("Retry-After", "1") + return noop, tooMany + } + sixbeeMemSlots[uid]++ + var once sync.Once + return func() { + once.Do(func() { + sixbeeMu.Lock() + if sixbeeMemSlots[uid]--; sixbeeMemSlots[uid] <= 0 { + delete(sixbeeMemSlots, uid) + } + sixbeeMu.Unlock() + }) + }, nil + } + + key := "sixbee:conc:" + strconv.Itoa(uid) + member := c.GetString(common.RequestIdKey) + if member == "" { + member = strconv.FormatInt(time.Now().UnixNano(), 36) + } + now := time.Now() + ok, err := sixbeeAcquireScript.Run(context.Background(), common.RDB, []string{key}, + now.UnixMilli(), now.Add(-sixbeeSlotStaleAfter).UnixMilli(), limit, member, int(sixbeeSlotStaleAfter.Seconds())).Int() + if err != nil { + return noop, sixbeeLimiterDown(c, err) + } + if ok != 1 { + c.Header("Retry-After", "1") + return noop, tooMany + } + var once sync.Once + return func() { + once.Do(func() { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := common.RDB.ZRem(ctx, key, member).Err(); err != nil { + common.SysLog("6bee limiter: failed to release slot: " + err.Error()) + } + }) + }, nil +} + +// Admission while the window is below the limit; when it is not, also returns the time (ms) at +// which enough of the oldest entries will have expired to bring it below the limit. +var sixbeeTPMScript = redis.NewScript(` +local old = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', '(' .. ARGV[2]) +for _, m in ipairs(old) do redis.call('HDEL', KEYS[2], m) end +redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', '(' .. ARGV[2]) +local live = redis.call('ZRANGE', KEYS[1], 0, -1, 'WITHSCORES') +local limit = tonumber(ARGV[3]) +local vals, sum = {}, 0 +for i = 1, #live, 2 do + local v = tonumber(redis.call('HGET', KEYS[2], live[i])) or 0 + vals[#vals + 1] = {v, tonumber(live[i + 1])} + sum = sum + v +end +if sum >= limit then + local rem, at = sum, 0 + for _, e in ipairs(vals) do + rem = rem - e[1] + at = e[2] + if rem < limit then break end + end + return {0, sum, at} +end +local t = tonumber(ARGV[5]) +redis.call('ZADD', KEYS[1], ARGV[1], ARGV[4]) +redis.call('HSET', KEYS[2], ARGV[4], t) +redis.call('EXPIRE', KEYS[1], ARGV[6]) +redis.call('EXPIRE', KEYS[2], ARGV[6]) +return {1, sum + t, 0}`) + +var sixbeeTPMSettleScript = redis.NewScript(` +if redis.call('ZSCORE', KEYS[1], ARGV[1]) then redis.call('HSET', KEYS[2], ARGV[1], ARGV[2]) end +return 1`) + +// SixbeeAdmitTokens checks the model's context budget and reserves the input estimate in the +// account's rolling one-minute input-token window. Call after the prompt estimate is known. +func SixbeeAdmitTokens(c *gin.Context, info *relaycommon.RelayInfo, promptTokens int) *types.NewAPIError { + sixbeeInit() + if info == nil || info.UserId == 0 { + return nil + } + lim := sixbeeLimitFor(info.OriginModelName) + maxOut := sixbeeMaxOutputOf(info.Request) + if lim.Context > 0 && promptTokens+maxOut > lim.Context { + // An output ceiling we injected ourselves shrinks to fit instead of failing the request. + if room := lim.Context - promptTokens; c.GetBool(sixbeeCtxOutputInjected) && room >= sixbeeMinOutput { + sixbeeSetInjectedOutput(info.Request, uint(room)) + } else { + return sixbeeErr(http.StatusBadRequest, "context_budget_exceeded", + fmt.Sprintf("Input (about %d tokens) plus max output (%d) exceeds the %d-token budget for %s. Shorten the input or lower max_tokens.", + promptTokens, maxOut, lim.Context, info.OriginModelName)) + } + } + limit := sixbeeInputTPM + if limit <= 0 || promptTokens <= 0 { + return nil + } + uid := info.UserId + member := c.GetString(common.RequestIdKey) + if member == "" { + member = strconv.FormatInt(time.Now().UnixNano(), 36) + } + now := time.Now() + tooMany := func(retryAfter time.Duration) *types.NewAPIError { + secs := int(retryAfter.Seconds()) + 1 + if secs < 1 || secs > 60 { + secs = 60 + } + c.Header("Retry-After", strconv.Itoa(secs)) + return sixbeeErr(http.StatusTooManyRequests, "account_input_tpm_limit", + fmt.Sprintf("This account has used its input token limit (%d uncached tokens per minute). Retry in %d s.", limit, secs)) + } + + if !common.RedisEnabled || common.RDB == nil { + sixbeeMu.Lock() + defer sixbeeMu.Unlock() + live := sixbeeMemTPM[uid][:0] + sum := 0 + for _, e := range sixbeeMemTPM[uid] { + if now.Sub(e.at) < sixbeeTPMWindow { + live = append(live, e) + sum += e.tokens + } + } + sixbeeMemTPM[uid] = live + if sum >= limit { + rem := sum + for _, e := range live { + rem -= e.tokens + if rem < limit { + return tooMany(sixbeeTPMWindow - now.Sub(e.at)) + } + } + return tooMany(sixbeeTPMWindow) + } + sixbeeMemTPM[uid] = append(sixbeeMemTPM[uid], sixbeeTPMEntry{member: member, at: now, tokens: promptTokens}) + c.Set(sixbeeCtxTPMMember, member) + return nil + } + + zkey, hkey := "sixbee:tpm:"+strconv.Itoa(uid), "sixbee:tpmv:"+strconv.Itoa(uid) + res, err := sixbeeTPMScript.Run(context.Background(), common.RDB, []string{zkey, hkey}, + now.UnixMilli(), now.Add(-sixbeeTPMWindow).UnixMilli(), limit, member, promptTokens, int(sixbeeTPMWindow.Seconds())+5).Int64Slice() + if err != nil || len(res) != 3 { + if err == nil { + err = fmt.Errorf("unexpected TPM script result %v", res) + } + return sixbeeLimiterDown(c, err) + } + if res[0] != 1 { + return tooMany(time.UnixMilli(res[2]).Add(sixbeeTPMWindow).Sub(now)) + } + c.Set(sixbeeCtxTPMMember, member) + return nil +} + +// SixbeeReleaseTokens drops this request's reservation when it ends before reaching an upstream +// (price error, insufficient balance). Safe to call more than once. +func SixbeeReleaseTokens(c *gin.Context, info *relaycommon.RelayInfo) { + member := c.GetString(sixbeeCtxTPMMember) + if member == "" || info == nil || c.GetBool(sixbeeCtxTPMCommitted) { + return + } + c.Set(sixbeeCtxTPMMember, "") + uid := info.UserId + if !common.RedisEnabled || common.RDB == nil { + sixbeeMu.Lock() + defer sixbeeMu.Unlock() + entries := sixbeeMemTPM[uid] + for i, e := range entries { + if e.member == member { + sixbeeMemTPM[uid] = append(entries[:i], entries[i+1:]...) + break + } + } + return + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + sid := strconv.Itoa(uid) + common.RDB.ZRem(ctx, "sixbee:tpm:"+sid, member) + common.RDB.HDel(ctx, "sixbee:tpmv:"+sid, member) +} + +// SixbeeSettleTokens replaces the estimate with the upstream-reported uncached input tokens. +func SixbeeSettleTokens(c *gin.Context, info *relaycommon.RelayInfo, uncachedPromptTokens int) { + member := c.GetString(sixbeeCtxTPMMember) + if member == "" || info == nil || uncachedPromptTokens < 0 { + return + } + c.Set(sixbeeCtxTPMCommitted, true) + uid := info.UserId + if !common.RedisEnabled || common.RDB == nil { + sixbeeMu.Lock() + defer sixbeeMu.Unlock() + for i, e := range sixbeeMemTPM[uid] { + if e.member == member { + sixbeeMemTPM[uid][i].tokens = uncachedPromptTokens + break + } + } + return + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + sid := strconv.Itoa(uid) + if err := sixbeeTPMSettleScript.Run(ctx, common.RDB, []string{"sixbee:tpm:" + sid, "sixbee:tpmv:" + sid}, member, uncachedPromptTokens).Err(); err != nil { + common.SysLog("6bee limiter: failed to settle TPM: " + err.Error()) + } +} + +// SixbeeUncachedPromptTokens is the input that counts toward the TPM window. OpenAI-style usage +// includes cache hits in prompt_tokens; Anthropic-style input_tokens already excludes them. +func SixbeeUncachedPromptTokens(usage *dto.Usage, anthropicSemantic bool) int { + if usage == nil { + return -1 + } + n := usage.PromptTokens + if !anthropicSemantic { + n -= usage.PromptTokensDetails.CachedTokens + } + return max(n, 0) +} + +// SixbeeOutputReserve is the quota for the request's maximum billable output, added to the +// pre-consumed quota so a low balance cannot be overrun by the response (plan A07/A10). +func SixbeeOutputReserve(info *relaycommon.RelayInfo, priceData hosttypes.PriceData) (int, error) { + if info == nil || priceData.UsePrice || priceData.FreeModel { + return 0, nil + } + maxOut := sixbeeMaxOutputOf(info.Request) + if maxOut <= 0 { + return 0, nil + } + return common.QuotaFromFloatStrict(float64(maxOut) * priceData.ModelRatio * priceData.CompletionRatio * priceData.GroupRatioInfo.GroupRatio) +} diff -ruN a/service/sixbee_notify.go b/service/sixbee_notify.go --- a/service/sixbee_notify.go +++ b/service/sixbee_notify.go @@ -0,0 +1,465 @@ +package service + +// 6bee patch: account notices go to email and Telegram. +// +// A user receives notices (low balance, announcements, promotions) at the account's email +// address (or the notification address they set) and, once they link it, in a private chat +// with the site's Telegram bot. Webhook, Bark and Gotify are no longer offered. +// +// Linking: the console asks for a one-time link (SixbeeTelegramLinkStart), the user opens +// https://t.me/?start= and presses Start, the bot receives "/start " by long +// polling (no public webhook) and stores the chat in sixbee_telegram_links. +// +// SIXBEE_TG_BOT_TOKEN bot token; empty = Telegram off (email only) +// SIXBEE_TG_API API base, default https://api.telegram.org (tests point it at a mock) + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "html" + "io" + "net/http" + "os" + "regexp" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/model" + "github.com/QuantumNous/new-api/relaykit/dto" + + "github.com/bytedance/gopkg/util/gopool" +) + +// SixbeeTelegramLink is the private Telegram chat an account receives notices in. +type SixbeeTelegramLink struct { + UserId int `json:"user_id" gorm:"primaryKey;autoIncrement:false"` + ChatId int64 `json:"chat_id" gorm:"index"` + Username string `json:"username" gorm:"type:varchar(64)"` + Name string `json:"name" gorm:"type:varchar(128)"` + CreatedAt int64 `json:"created_at"` +} + +func (SixbeeTelegramLink) TableName() string { return "sixbee_telegram_links" } + +const ( + sixbeeTelegramLinkTTL = 10 * time.Minute + sixbeeTelegramLinkPrefix = "sixbee:tgbind:" +) + +var ( + sixbeeNotifyOnce sync.Once + sixbeeTelegramToken string + sixbeeTelegramAPI string + sixbeeTelegramUsername atomic.Value // string, set once getMe answers + sixbeeTelegramReplies sync.Map // chat id -> *sixbeeReplyWindow + sixbeeTelegramPollHTTP = &http.Client{Timeout: 70 * time.Second} + sixbeeTelegramSendHTTP = &http.Client{Timeout: 15 * time.Second} + + sixbeeLineBreakRe = regexp.MustCompile(`(?i)|

||`) + sixbeeAnchorRe = regexp.MustCompile(`(?i)]*?href\s*=\s*['"]([^'"]*)['"][^>]*>`) + sixbeeTagRe = regexp.MustCompile(`]*>`) +) + +type sixbeeTelegramError struct { + Code int + Description string +} + +func (e *sixbeeTelegramError) Error() string { + return fmt.Sprintf("telegram: %d %s", e.Code, e.Description) +} + +type sixbeeTelegramUpdate struct { + UpdateId int64 `json:"update_id"` + Message *sixbeeTelegramMessage `json:"message"` +} + +type sixbeeTelegramMessage struct { + Text string `json:"text"` + Chat struct { + Id int64 `json:"id"` + Type string `json:"type"` + } `json:"chat"` + From *struct { + IsBot bool `json:"is_bot"` + FirstName string `json:"first_name"` + LastName string `json:"last_name"` + Username string `json:"username"` + LanguageCode string `json:"language_code"` + } `json:"from"` +} + +type sixbeeReplyWindow struct { + sync.Mutex + start time.Time + count int +} + +// SixbeeNotifyStart creates the 6bee notification tables and, when a bot token is set, starts +// the Telegram listener. Called once while the API routes are registered. +func SixbeeNotifyStart() { + sixbeeNotifyOnce.Do(func() { + if model.DB == nil { + return + } + if err := model.DB.AutoMigrate(&SixbeeTelegramLink{}, &SixbeeBroadcast{}); err != nil { + common.SysError("6bee: notification tables: " + err.Error()) + return + } + sixbeeBroadcastRecover() + sixbeeTelegramToken = strings.TrimSpace(os.Getenv("SIXBEE_TG_BOT_TOKEN")) + sixbeeTelegramAPI = strings.TrimRight(common.GetEnvOrDefaultString("SIXBEE_TG_API", "https://api.telegram.org"), "/") + if sixbeeTelegramToken == "" { + common.SysLog("6bee: Telegram notices off (SIXBEE_TG_BOT_TOKEN is empty); email only") + return + } + if !common.RedisEnabled { + common.SysLog("6bee: Telegram notices off (linking needs Redis)") + return + } + if !common.IsMasterNode { + return + } + gopool.Go(sixbeeTelegramPoll) + }) +} + +// SixbeeTelegramBot returns the bot's username once Telegram answered getMe; "" while Telegram +// notices are unavailable. +func SixbeeTelegramBot() string { + name, _ := sixbeeTelegramUsername.Load().(string) + return name +} + +// sixbeeTelegramCall posts one Bot API method. Errors never contain the bot token. +func sixbeeTelegramCall(client *http.Client, method string, payload any, result any) error { + body, err := common.Marshal(payload) + if err != nil { + return err + } + resp, err := client.Post(sixbeeTelegramAPI+"/bot"+sixbeeTelegramToken+"/"+method, "application/json", bytes.NewReader(body)) + if err != nil { + return errors.New("telegram: " + strings.ReplaceAll(err.Error(), sixbeeTelegramToken, "")) + } + defer resp.Body.Close() + raw, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20)) + if err != nil { + return errors.New("telegram: reading the answer failed") + } + var answer struct { + Ok bool `json:"ok"` + ErrorCode int `json:"error_code"` + Description string `json:"description"` + Result json.RawMessage `json:"result"` + } + if err := common.Unmarshal(raw, &answer); err != nil { + return &sixbeeTelegramError{Code: resp.StatusCode, Description: "unreadable answer"} + } + if !answer.Ok { + return &sixbeeTelegramError{Code: answer.ErrorCode, Description: answer.Description} + } + if result == nil { + return nil + } + return common.Unmarshal(answer.Result, result) +} + +// sixbeeTelegramSend sends one HTML message to a private chat. A chat that blocked the bot or no +// longer exists is unlinked, so later notices skip it. +func sixbeeTelegramSend(chatId int64, text string) error { + err := sixbeeTelegramCall(sixbeeTelegramSendHTTP, "sendMessage", map[string]any{ + "chat_id": chatId, + "text": text, + "parse_mode": "HTML", + "link_preview_options": map[string]any{"is_disabled": true}, + }, nil) + var tgErr *sixbeeTelegramError + if errors.As(err, &tgErr) && tgErr.Code == http.StatusForbidden { + model.DB.Where("chat_id = ?", chatId).Delete(&SixbeeTelegramLink{}) + } + return err +} + +// sixbeeTelegramHTML turns notice HTML written for email into the subset Telegram accepts: +// line breaks become newlines, links and basic emphasis stay, every other tag is dropped. +func sixbeeTelegramHTML(content string) string { + content = sixbeeLineBreakRe.ReplaceAllString(content, "\n") + content = sixbeeAnchorRe.ReplaceAllString(content, ``) + content = sixbeeTagRe.ReplaceAllStringFunc(content, func(tag string) string { + name := strings.ToLower(sixbeeTagRe.FindStringSubmatch(tag)[1]) + switch name { + case "a", "b", "strong", "i", "em", "u", "s", "code", "pre": + return tag + } + return "" + }) + return strings.TrimSpace(content) +} + +func sixbeeTelegramPoll() { + for { + var me struct { + Username string `json:"username"` + } + err := sixbeeTelegramCall(sixbeeTelegramSendHTTP, "getMe", map[string]any{}, &me) + if err == nil && me.Username != "" { + sixbeeTelegramUsername.Store(me.Username) + common.SysLog("6bee: Telegram notices on, bot @" + me.Username) + break + } + if err != nil { + common.SysError("6bee: Telegram getMe failed, retrying in 60 s: " + err.Error()) + } + time.Sleep(time.Minute) + } + var offset int64 + var lastErrorLog time.Time + for { + var updates []sixbeeTelegramUpdate + err := sixbeeTelegramCall(sixbeeTelegramPollHTTP, "getUpdates", map[string]any{ + "offset": offset, + "timeout": 50, + "allowed_updates": []string{"message"}, + }, &updates) + if err != nil { + if time.Since(lastErrorLog) > 10*time.Minute { + lastErrorLog = time.Now() + common.SysError("6bee: Telegram getUpdates failed (retrying): " + err.Error()) + } + pause := 5 * time.Second + var tgErr *sixbeeTelegramError + if errors.As(err, &tgErr) && (tgErr.Code == http.StatusConflict || tgErr.Code == http.StatusTooManyRequests || tgErr.Code == http.StatusUnauthorized) { + pause = 30 * time.Second + } + time.Sleep(pause) + continue + } + for _, update := range updates { + offset = update.UpdateId + 1 + sixbeeTelegramHandle(update.Message) + } + } +} + +// sixbeeTelegramHandle answers one message sent to the bot in a private chat. +func sixbeeTelegramHandle(message *sixbeeTelegramMessage) { + if message == nil || message.From == nil || message.From.IsBot || message.Chat.Type != "private" { + return + } + chatId := message.Chat.Id + // At most 10 answers a minute per chat, so the bot cannot be made to flood anyone. + value, _ := sixbeeTelegramReplies.LoadOrStore(chatId, &sixbeeReplyWindow{}) + window := value.(*sixbeeReplyWindow) + window.Lock() + if time.Since(window.start) > time.Minute { + window.start, window.count = time.Now(), 0 + } + window.count++ + over := window.count > 10 + window.Unlock() + if over { + return + } + + zh := strings.HasPrefix(strings.ToLower(message.From.LanguageCode), "zh") + say := func(zhText string, enText string) { + text := enText + if zh { + text = zhText + } + if err := sixbeeTelegramSend(chatId, text); err != nil { + common.SysError("6bee: Telegram reply failed: " + err.Error()) + } + } + console := html.EscapeString(PaymentReturnURL("/profile")) + command, argument, _ := strings.Cut(strings.TrimSpace(message.Text), " ") + command, _, _ = strings.Cut(command, "@") + argument = strings.TrimSpace(argument) + + switch { + case command == "/start" && argument != "": + userId := sixbeeTelegramLinkClaim(argument) + user, err := model.GetUserById(userId, false) + if userId == 0 || err != nil || user.Status != common.UserStatusEnabled { + say("这个绑定链接已失效或已使用过。请回到控制台「个人设置 → 通知」重新点「绑定 Telegram」：\n"+console, + "This link has expired or was already used. Open Profile → Notifications in the console and press “Link Telegram” again:\n"+console) + return + } + link := SixbeeTelegramLink{ + UserId: user.Id, + ChatId: chatId, + Username: message.From.Username, + Name: strings.TrimSpace(message.From.FirstName + " " + message.From.LastName), + CreatedAt: time.Now().Unix(), + } + if err := model.DB.Save(&link).Error; err != nil { + common.SysError("6bee: saving Telegram link failed: " + err.Error()) + say("绑定失败，请稍后再试。", "Linking failed. Please try again later.") + return + } + account := html.EscapeString(user.Username) + say("✅ 已绑定 6bee.ai 账户 "+account+"。\n余额提醒、公告和活动通知会发到这里。\n\n/nopromo 不再接收活动通知\n/stop 解除绑定", + "✅ Linked to 6bee.ai account "+account+".\nLow balance reminders, announcements and promotions will arrive here.\n\n/nopromo stop promotions\n/stop unlink") + case command == "/stop": + result := model.DB.Where("chat_id = ?", chatId).Delete(&SixbeeTelegramLink{}) + if result.Error == nil && result.RowsAffected > 0 { + say("已解除绑定，这里不会再收到 6bee.ai 的通知。", "Unlinked. You will no longer receive 6bee.ai notices here.") + return + } + say("这个聊天没有绑定 6bee.ai 账户。", "This chat is not linked to a 6bee.ai account.") + case command == "/nopromo": + var links []SixbeeTelegramLink + model.DB.Where("chat_id = ?", chatId).Find(&links) + for _, link := range links { + if err := SixbeeSetNoticePreference(link.UserId, SixbeeBroadcastPromo, false); err != nil { + common.SysError("6bee: /nopromo failed: " + err.Error()) + } + } + if len(links) == 0 { + say("这个聊天没有绑定 6bee.ai 账户。", "This chat is not linked to a 6bee.ai account.") + return + } + say("已关闭活动通知（邮箱和 Telegram 都不再发送）。余额提醒和公告不受影响，可在控制台「个人设置 → 通知」重新开启。", + "Promotions are off, by email and Telegram. Low balance reminders and announcements still arrive. Turn promotions back on under Profile → Notifications in the console.") + default: + say("这是 6bee.ai 的通知机器人。要在这里接收余额提醒和公告，请在控制台「个人设置 → 通知」点「绑定 Telegram」：\n"+console+"\n\n/nopromo 不再接收活动通知\n/stop 解除绑定", + "This is the 6bee.ai notification bot. To receive low balance reminders and announcements here, open Profile → Notifications in the console and press “Link Telegram”:\n"+console+"\n\n/nopromo stop promotions\n/stop unlink") + } +} + +// SixbeeTelegramLinkStart returns a one-time https://t.me link that ties the chat that opens it +// to userId. It is valid for 10 minutes; asking again cancels the previous link. +func SixbeeTelegramLinkStart(userId int) (string, error) { + bot := SixbeeTelegramBot() + if bot == "" || !common.RedisEnabled { + return "", common.NewMessage("Telegram notifications are not available right now") + } + random := make([]byte, 16) + if _, err := rand.Read(random); err != nil { + return "", err + } + token := base64.RawURLEncoding.EncodeToString(random) + sum := sha256.Sum256([]byte(token)) + key := sixbeeTelegramLinkPrefix + hex.EncodeToString(sum[:]) + userKey := sixbeeTelegramLinkPrefix + "user:" + strconv.Itoa(userId) + ctx := context.Background() + if previous, err := common.RDB.Get(ctx, userKey).Result(); err == nil && previous != "" { + common.RDB.Del(ctx, previous) + } + pipe := common.RDB.TxPipeline() + pipe.Set(ctx, key, userId, sixbeeTelegramLinkTTL) + pipe.Set(ctx, userKey, key, sixbeeTelegramLinkTTL) + if _, err := pipe.Exec(ctx); err != nil { + return "", err + } + return "https://t.me/" + bot + "?start=" + token, nil +} + +// sixbeeTelegramLinkClaim spends a link token and returns the account it was issued for, 0 when +// the token is unknown, expired or already used. +func sixbeeTelegramLinkClaim(token string) int { + if len(token) != 22 { + return 0 + } + sum := sha256.Sum256([]byte(token)) + value, err := common.RDB.GetDel(context.Background(), sixbeeTelegramLinkPrefix+hex.EncodeToString(sum[:])).Result() + if err != nil { + return 0 + } + userId, _ := strconv.Atoi(value) + return userId +} + +// SixbeeTelegramLinkOf returns the chat linked to userId. +func SixbeeTelegramLinkOf(userId int) (SixbeeTelegramLink, bool) { + var link SixbeeTelegramLink + if err := model.DB.Where("user_id = ?", userId).Limit(1).Find(&link).Error; err != nil || link.ChatId == 0 { + return link, false + } + return link, true +} + +// SixbeeTelegramUnlink removes userId's chat and tells that chat. +func SixbeeTelegramUnlink(userId int) error { + link, linked := SixbeeTelegramLinkOf(userId) + if !linked { + return nil + } + if err := model.DB.Where("user_id = ?", userId).Delete(&SixbeeTelegramLink{}).Error; err != nil { + return err + } + if SixbeeTelegramBot() != "" { + gopool.Go(func() { + _ = sixbeeTelegramSend(link.ChatId, "已在控制台解除绑定，这里不会再收到 6bee.ai 的通知。\nUnlinked in the console. No more 6bee.ai notices will arrive here.") + }) + } + return nil +} + +func sixbeeSwitchOn(value *bool) bool { return value == nil || *value } + +// sixbeeNotifyUser delivers one notice by email and Telegram, whichever the user has on. It +// fails only when nothing could be delivered and something went wrong. +func sixbeeNotifyUser(userId int, userEmail string, setting dto.UserSetting, data dto.Notify) error { + var failures []error + delivered := 0 + if sixbeeSwitchOn(setting.EmailNotifyEnabled) { + address := setting.NotificationEmail + if address == "" { + address = userEmail + } + if address != "" { + if err := sendEmailNotify(address, data); err != nil { + failures = append(failures, err) + } else { + delivered++ + } + } + } + if sixbeeSwitchOn(setting.TelegramNotifyEnabled) && SixbeeTelegramBot() != "" { + if link, linked := SixbeeTelegramLinkOf(userId); linked { + content := data.Content + for _, value := range data.Values { + content = strings.Replace(content, dto.ContentValueParam, fmt.Sprintf("%v", value), 1) + } + text := "" + html.EscapeString(data.Title) + "\n\n" + sixbeeTelegramHTML(content) + if err := sixbeeTelegramSend(link.ChatId, text); err != nil { + failures = append(failures, err) + } else { + delivered++ + } + } + } + if delivered == 0 && len(failures) > 0 { + return errors.Join(failures...) + } + return nil +} + +// SixbeeSetNoticePreference turns announcements or promotions on or off for one account. +func SixbeeSetNoticePreference(userId int, kind string, enabled bool) error { + setting, err := model.GetUserSetting(userId, true) + if err != nil { + return err + } + switch kind { + case SixbeeBroadcastPromo: + setting.PromoNotifyEnabled = &enabled + case SixbeeBroadcastAnnouncement: + setting.AnnouncementNotifyEnabled = &enabled + default: + return errors.New("unknown notice kind") + } + return model.UpdateUserSetting(userId, setting) +} diff -ruN a/service/sixbee_notify_test.go b/service/sixbee_notify_test.go --- a/service/sixbee_notify_test.go +++ b/service/sixbee_notify_test.go @@ -0,0 +1,81 @@ +package service + +import ( + "strings" + "testing" + + "github.com/QuantumNous/new-api/common" +) + +func TestSixbeeTelegramHTML(t *testing.T) { + got := sixbeeTelegramHTML("您的额度即将用尽，当前剩余额度为 $1.00。
充值链接：https://console.6bee.ai/wallet

bye

") + want := "您的额度即将用尽，当前剩余额度为 $1.00。\n充值链接：https://console.6bee.ai/walletxbye" + if got != want { + t.Fatalf("got %q\nwant %q", got, want) + } +} + +func TestSixbeeUnsubscribeSignature(t *testing.T) { + common.CryptoSecret = "test-secret" + sig := sixbeeUnsubscribeSignature(7, SixbeeBroadcastPromo) + if !SixbeeUnsubscribeValid(7, SixbeeBroadcastPromo, sig) { + t.Fatal("own signature refused") + } + for name, ok := range map[string]bool{ + "other user": SixbeeUnsubscribeValid(8, SixbeeBroadcastPromo, sig), + "other kind": SixbeeUnsubscribeValid(7, SixbeeBroadcastAnnouncement, sig), + "bad kind": SixbeeUnsubscribeValid(7, "quota", sig), + "empty": SixbeeUnsubscribeValid(7, SixbeeBroadcastPromo, ""), + } { + if ok { + t.Errorf("%s: accepted", name) + } + } + common.CryptoSecret = "another-secret" + if SixbeeUnsubscribeValid(7, SixbeeBroadcastPromo, sig) { + t.Error("signature survived a secret change") + } +} + +func TestSixbeeBroadcastValidate(t *testing.T) { + ok := SixbeeBroadcast{Kind: SixbeeBroadcastPromo, Title: " 双十一 ", Content: "充 100 送 10", LinkUrl: "https://www.6bee.ai/"} + if err := sixbeeBroadcastValidate(&ok); err != nil || ok.Title != "双十一" { + t.Fatalf("valid broadcast refused: %v %q", err, ok.Title) + } + for name, b := range map[string]SixbeeBroadcast{ + "kind": {Kind: "quota", Title: "t", Content: "c"}, + "no title": {Kind: SixbeeBroadcastPromo, Content: "c"}, + "no content": {Kind: SixbeeBroadcastPromo, Title: "t"}, + "long": {Kind: SixbeeBroadcastPromo, Title: "t", Content: strings.Repeat("字", 3001)}, + "http link": {Kind: SixbeeBroadcastPromo, Title: "t", Content: "c", LinkUrl: "http://example.com"}, + "js link": {Kind: SixbeeBroadcastPromo, Title: "t", Content: "c", LinkUrl: "javascript:alert(1)"}, + } { + if sixbeeBroadcastValidate(&b) == nil { + t.Errorf("%s: accepted", name) + } + } +} + +func TestSixbeeBroadcastRender(t *testing.T) { + common.CryptoSecret = "test-secret" + b := SixbeeBroadcast{Kind: SixbeeBroadcastPromo, Title: "Sale", Content: "line 1\n", LinkUrl: "https://www.6bee.ai/?a=1&b=2"} + email := sixbeeBroadcastEmail(&b, 7, "en") + for _, want := range []string{"<b>Sale</b>", "line 1
<script>", `href="https://www.6bee.ai/?a=1&b=2"`, "Unsubscribe from promotions", "/api/sixbee/notify/unsubscribe?k=promo&s=", "u=7"} { + if !strings.Contains(email, want) { + t.Errorf("email lacks %q", want) + } + } + if strings.Contains(email, "