heqiapp/platforms
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

audit full project flows and harden backend

Browse Source
This commit is contained in:
yanweidong
2026-08-03 16:03:44 +08:00
parent d8d62469d6
commit 95b6f2b69c
22 changed files with 505 additions and 890 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
backend/api/cmd/cli/main.go
View File

@@ -196,7 +196,7 @@ func migrateDatabase() error {
if err := prepareAdditiveMigrations(migrationDatabase, driver); err != nil {
return err
}
if err := resetLegacyPaymentSchema(migrationDatabase, driver); err != nil {
if err := rejectLegacyPaymentSchema(migrationDatabase); err != nil {
return err
}
const legacyPhoneIndex = "idx_platform_account_phone"
@@ -216,25 +216,12 @@ func migrateDatabase() error {
return initdb.New(databaseService)
}
// resetLegacyPaymentSchema 执行经业务明确授权的开发期破坏性支付模型重置，不迁移旧支付或退款历史。
func resetLegacyPaymentSchema(databaseService *gorm.DB, driver string) error {
statements := []string{}
if driver == "postgres" {
statements = append(statements,
`DROP TABLE IF EXISTS "wallet_refund" CASCADE`,
`DROP TABLE IF EXISTS "wallet_payment" CASCADE`,
`DROP TABLE IF EXISTS "gasorder_payment" CASCADE`,
)
} else {
statements = append(statements,
`DROP TABLE IF EXISTS wallet_refund`,
`DROP TABLE IF EXISTS wallet_payment`,
`DROP TABLE IF EXISTS gasorder_payment`,
)
}
for _, statement := range statements {
if err := databaseService.Exec(statement).Error; err != nil {
return fmt.Errorf("reset legacy payment schema: %w", err)
// rejectLegacyPaymentSchema 阻止通用迁移物理删除历史资金表。
// 旧支付事实必须通过单独评审的数据迁移保留和对账，不能由 migrate 命令静默重置。
func rejectLegacyPaymentSchema(databaseService *gorm.DB) error {
for _, table := range []string{"wallet_refund", "wallet_payment", "gasorder_payment"} {
if databaseService.Migrator().HasTable(table) {
return fmt.Errorf("检测到历史资金表 %s：请先执行经财务与审计批准的保留式迁移", table)
}
}
return nil

19
backend/api/cmd/cli/main_test.go
View File

@@ -1,18 +1,31 @@
package main
import (
"regexp"
"testing"
"gorm.io/driver/sqlite"
"github.com/DATA-DOG/go-sqlmock"
"gorm.io/driver/postgres"
"gorm.io/gorm"
)
func TestPrepareAdditiveMigrationsSkipsMissingLegacyTable(t *testing.T) {
databaseService, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
sqlDatabase, mock, err := sqlmock.New()
if err != nil {
t.Fatalf("open sqlite database: %v", err)
t.Fatalf("open sql mock: %v", err)
}
defer sqlDatabase.Close()
mock.ExpectQuery(regexp.QuoteMeta("SELECT count(*) FROM information_schema.tables WHERE table_schema = CURRENT_SCHEMA() AND table_name = $1 AND table_type = $2")).
WithArgs("gasorder_track_point", "BASE TABLE").
WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0))
databaseService, err := gorm.Open(postgres.New(postgres.Config{Conn: sqlDatabase}), &gorm.Config{})
if err != nil {
t.Fatalf("open postgres mock: %v", err)
}
if err := prepareAdditiveMigrations(databaseService, "postgres"); err != nil {
t.Fatalf("missing legacy table should not require a backfill: %v", err)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatalf("unexpected migration query: %v", err)
}
}

3
backend/api/internal/config/config.go
View File

@@ -90,6 +90,9 @@ func New(srvKey string) {
Spec.BindIP = conf.CheckIP(Spec.BindIP)
Spec.Addr = net.JoinHostPort(Spec.BindIP, Spec.Port)
conf.NotNil(Spec.Service, Spec.Cache)
if Spec.Databases == nil || len(Spec.Databases.Source) == 0 {
panic("Databases configuration is required")
}
if Spec.Global.ManualRechargeMaxAmount <= 0 {
panic("Global.ManualRechargeMaxAmount must be greater than zero")
}

10
backend/api/internal/logic/iot/iot.go
View File

@@ -4,6 +4,7 @@ package iot
import (
"crypto/subtle"
"encoding/json"
"errors"
"net/http"
"strings"
"time"
@@ -187,7 +188,14 @@ func SaveDeviceMessage(ctx *gin.Context) {
if frame.Control&0x04 != 0 {
status, errorCode = "failed", "DEVICE_REPORTED_FAILURE"
}
return tx.Model(&models.IotCommand{}).Where("device_id = ? AND packet_number = ? AND command_status IN ?", envelope.DeviceID, frame.PacketNumber, []string{"dispatched", "pending_confirmation"}).Updates(map[string]any{"command_status": status, "error_code": errorCode, "acknowledged_at": received, "updated_at": received}).Error
result := tx.Model(&models.IotCommand{}).Where("device_id = ? AND packet_number = ? AND command_status IN ?", envelope.DeviceID, frame.PacketNumber, []string{"dispatched", "pending_confirmation"}).Updates(map[string]any{"command_status": status, "error_code": errorCode, "acknowledged_at": received, "updated_at": received})
if result.Error != nil {
return result.Error
}
if result.RowsAffected > 1 {
return errors.New("ambiguous device acknowledgement")
}
return nil
})
if err != nil {
ctx.JSON(500, gin.H{"code": "IOT_MESSAGE_SAVE_FAILED"})

28
backend/api/internal/logic/payment/callback.go
View File

@@ -6,6 +6,7 @@ import (
"errors"
"net/http"
"strconv"
"strings"
"time"
"git.apinb.com/heqiapp/platforms/backend/api/internal/config"
@@ -80,12 +81,18 @@ func complete(paymentNo, tradeNo string, amount int64, channel, callbackDigest s
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("payment_no = ?", paymentNo).First(&order).Error; err != nil {
return err
}
if order.PaymentStatus == StatusPaid {
return nil
}
if order.Channel != channel || order.Amount != amount {
return errors.New("payment identity or amount mismatch")
}
if strings.TrimSpace(tradeNo) == "" {
return errors.New("channel trade number is required")
}
if order.PaymentStatus == StatusPaid {
if order.ChannelTradeNo != tradeNo {
return errors.New("duplicate callback trade number mismatch")
}
return nil
}
if time.Now().After(order.ExpiresAt) {
return tx.Model(&order).Updates(map[string]any{"payment_status": 50, "channel_trade_no": tradeNo, "callback_digest": callbackDigest, "failure_code": "PAID_AFTER_EXPIRED"}).Error
}
@@ -95,9 +102,10 @@ func complete(paymentNo, tradeNo string, amount int64, channel, callbackDigest s
}
switch order.BusinessType {
case "ec_order":
return tx.Model(&models.EcOrder{}).Where("identity = ? AND order_status = ?", order.BusinessIdentity, 16).Updates(map[string]any{"order_status": 18, "paid_at": &now}).Error
result := tx.Model(&models.EcOrder{}).Where("identity = ? AND order_status = ?", order.BusinessIdentity, 16).Updates(map[string]any{"order_status": 18, "paid_at": &now})
return requireSingleBusinessUpdate(result)
case "gasorder":
return tx.Model(&models.GasorderBasic{}).Where("identity = ? AND order_status IN ?", order.BusinessIdentity, []int{16, 18}).Update("order_status", 35).Error
return requireSingleBusinessUpdate(tx.Model(&models.GasorderBasic{}).Where("identity = ? AND order_status IN ?", order.BusinessIdentity, []int{16, 18}).Update("order_status", 35))
case "recharge":
return completeRecharge(tx, order, now)
}
@@ -105,6 +113,16 @@ func complete(paymentNo, tradeNo string, amount int64, channel, callbackDigest s
})
}
func requireSingleBusinessUpdate(result *gorm.DB) error {
if result.Error != nil {
return result.Error
}
if result.RowsAffected != 1 {
return errors.New("payment business state conflict")
}
return nil
}
func completeRecharge(tx *gorm.DB, payment models.PaymentOrder, now time.Time) error {
var recharge models.WalletRechargeOrder
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("identity = ? AND recharge_status = ?", payment.BusinessIdentity, 10).First(&recharge).Error; err != nil {

21
backend/api/internal/logic/payment/callback_test.go Normal file
View File

@@ -0,0 +1,21 @@
package payment
import (
"errors"
"testing"
"gorm.io/gorm"
)
func TestRequireSingleBusinessUpdate(t *testing.T) {
if err := requireSingleBusinessUpdate(&gorm.DB{RowsAffected: 1}); err != nil {
t.Fatalf("single state update rejected: %v", err)
}
if err := requireSingleBusinessUpdate(&gorm.DB{}); err == nil {
t.Fatal("zero-row state update must be rejected")
}
expected := errors.New("database unavailable")
if err := requireSingleBusinessUpdate(&gorm.DB{Error: expected}); !errors.Is(err, expected) {
t.Fatalf("database error lost: %v", err)
}
}

5
backend/api/internal/logic/payment/close.go
View File

@@ -2,6 +2,7 @@ package payment
import (
"context"
"crypto/subtle"
"fmt"
"net/http"
"time"
@@ -55,7 +56,9 @@ func closeChannelOrder(ctx context.Context, order models.PaymentOrder) error {
// CloseExpiredHandler 只接受 Worker 共享凭证，不暴露为平台用户动作。
func CloseExpiredHandler(ctx *gin.Context) {
if config.Spec.Payment.InternalServiceToken == "" || ctx.GetHeader("X-Heqi-Worker-Token") != config.Spec.Payment.InternalServiceToken {
expected := config.Spec.Payment.InternalServiceToken
actual := ctx.GetHeader("X-Heqi-Worker-Token")
if expected == "" || len(actual) != len(expected) || subtle.ConstantTimeCompare([]byte(actual), []byte(expected)) != 1 {
ctx.AbortWithStatus(http.StatusUnauthorized)
return
}

43
backend/iot-server/internal/service/service.go
View File

@@ -32,6 +32,7 @@ type Service struct {
packet atomic.Uint32
http *http.Server
ready atomic.Bool
client *http.Client
}
type Command struct {
Identity string `json:"identity"`
@@ -97,7 +98,7 @@ func New(cfg config.Config) (*Service, error) {
if err = broker.AddListener(listeners.NewTCP(listenerConfig)); err != nil {
return nil, fmt.Errorf("配置 MQTT 监听器: %w", err)
}
return &Service{cfg: cfg, keys: keys, broker: broker}, nil
return &Service{cfg: cfg, keys: keys, broker: broker, client: &http.Client{Timeout: 12 * time.Second}}, nil
}
func makeTLSConfig(cfg config.Config) (*tls.Config, error) {
@@ -222,10 +223,14 @@ func (s *Service) command(w http.ResponseWriter, r *http.Request) {
func (s *Service) onMessage(_ *mqtt.Client, _ packets.Subscription, message packets.Packet) {
raw := append([]byte(nil), message.Payload...)
envelope := Envelope{Type: "device_message", Topic: message.TopicName, ReceivedAt: time.Now().UTC().Format(time.RFC3339Nano), PayloadHex: hex.EncodeToString(raw)}
expectedDeviceID, topicOK := deviceIDFromTopic(s.cfg, message.TopicName)
envelope := Envelope{Type: "device_message", Topic: message.TopicName, DeviceID: expectedDeviceID, ReceivedAt: time.Now().UTC().Format(time.RFC3339Nano), PayloadHex: hex.EncodeToString(raw)}
if frame, err := protocol.Decode(raw, s.keys); err == nil {
deviceID := hex.EncodeToString(frame.DeviceID[:])
envelope.DeviceID = deviceID
if !topicOK || deviceID != expectedDeviceID {
s.forwardEnvelope(envelope)
return
}
main := byte(0)
if len(frame.Payload) > 0 {
main = frame.Payload[0]
@@ -238,6 +243,10 @@ func (s *Service) onMessage(_ *mqtt.Client, _ packets.Subscription, message pack
}
envelope.Frame = decoded
}
s.forwardEnvelope(envelope)
}
func (s *Service) forwardEnvelope(envelope Envelope) {
data, _ := json.Marshal(envelope)
request, err := http.NewRequest(http.MethodPost, s.cfg.HTTP.CallbackURL, bytes.NewReader(data))
if err != nil {
@@ -245,12 +254,38 @@ func (s *Service) onMessage(_ *mqtt.Client, _ packets.Subscription, message pack
}
request.Header.Set("Content-Type", "application/json")
request.Header.Set("X-Heqi-Iot-Token", s.cfg.HTTP.InternalToken)
response, err := http.DefaultClient.Do(request)
response, err := s.client.Do(request)
if err == nil {
_ = response.Body.Close()
}
}
func deviceIDFromTopic(cfg config.Config, topic string) (string, bool) {
for _, pattern := range []string{cfg.MQTT.UpTopic, cfg.MQTT.AckTopic} {
parts := strings.Split(pattern, "/")
values := strings.Split(topic, "/")
if len(parts) != len(values) {
continue
}
deviceID := ""
matched := true
for index := range parts {
if parts[index] == "+" {
deviceID = values[index]
continue
}
if parts[index] != values[index] {
matched = false
break
}
}
if matched && len(deviceID) == 16 {
return strings.ToLower(deviceID), true
}
}
return "", false
}
func decodeDeviceID(value string) ([8]byte, error) {
var result [8]byte
decoded, err := hex.DecodeString(value)

22
backend/iot-server/internal/service/service_test.go Normal file
View File

@@ -0,0 +1,22 @@
package service
import (
"testing"
"git.apinb.com/heqiapp/platforms/backend/iot-server/internal/config"
)
func TestDeviceIDFromTopic(t *testing.T) {
cfg := config.Config{MQTT: config.MQTT{UpTopic: "devices/+/up", AckTopic: "devices/+/ack"}}
for _, topic := range []string{"devices/0000000000000001/up", "devices/0000000000000001/ack"} {
identity, ok := deviceIDFromTopic(cfg, topic)
if !ok || identity != "0000000000000001" {
t.Fatalf("topic %s: identity=%q ok=%v", topic, identity, ok)
}
}
for _, topic := range []string{"devices/0001/up", "devices/0000000000000001/down", "other/0000000000000001/up"} {
if identity, ok := deviceIDFromTopic(cfg, topic); ok || identity != "" {
t.Fatalf("invalid topic %s accepted as %q", topic, identity)
}
}
}

3
backend/worker/cmd/main/main.go
View File

@@ -31,6 +31,7 @@ func main() {
func closeExpiredPayments(ctx context.Context) {
ticker := time.NewTicker(time.Duration(config.Spec.PaymentAPI.IntervalSeconds) * time.Second)
defer ticker.Stop()
client := &http.Client{Timeout: 15 * time.Second}
for {
select {
case <-ctx.Done():
@@ -41,7 +42,7 @@ func closeExpiredPayments(ctx context.Context) {
continue
}
request.Header.Set("X-Heqi-Worker-Token", config.Spec.PaymentAPI.Token)
response, err := http.DefaultClient.Do(request)
response, err := client.Do(request)
if err == nil {
_ = response.Body.Close()
}

20
backend/worker/internal/config/config.go
View File

@@ -4,6 +4,8 @@ package config
import (
"git.apinb.com/bsm-sdk/core/conf"
"net"
"os"
"strings"
)
var Spec SrvConfig
@@ -30,10 +32,28 @@ type IoTAPIConfig struct {
func New(srvKey string) {
conf.New(srvKey, &Spec)
if databaseDSN := strings.TrimSpace(os.Getenv("HEQI_DATABASE_DSN")); databaseDSN != "" {
if Spec.Databases == nil {
panic("Databases configuration is required")
}
Spec.Databases.Source = []string{databaseDSN}
}
if redisURL := strings.TrimSpace(os.Getenv("HEQI_REDIS_URL")); redisURL != "" {
Spec.Cache = redisURL
}
if token := strings.TrimSpace(os.Getenv("HEQI_PAYMENT_INTERNAL_TOKEN")); token != "" {
Spec.PaymentAPI.Token = token
}
if token := strings.TrimSpace(os.Getenv("HEQI_IOT_INTERNAL_TOKEN")); token != "" {
Spec.IoTAPI.Token = token
}
Spec.Port = conf.CheckPort(Spec.Port)
Spec.BindIP = conf.CheckIP(Spec.BindIP)
Spec.Addr = net.JoinHostPort(Spec.BindIP, Spec.Port)
conf.NotNil(Spec.Service, Spec.Cache)
if Spec.Databases == nil || len(Spec.Databases.Source) == 0 {
panic("Databases configuration is required")
}
if Spec.PaymentAPI.BaseURL == "" || Spec.PaymentAPI.Token == "" || Spec.PaymentAPI.IntervalSeconds <= 0 {
panic("PaymentAPI configuration is required")
}