hectorzhao/lislgosms
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: negotiate downstream cmpp protocol versions

Browse Source
This commit is contained in:
hectorzhao
2026-07-11 11:00:42 +08:00
parent a4d42cf702
commit bb4992f0f9
8 changed files with 276 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/internal/inbound/server.go
+119 -73
View File
@@ -59,6 +59,7 @@ type authResponse struct {
ApplicationID string `json:"applicationId"`
TenantID string `json:"tenantId"`
Account string `json:"account"`
EnterpriseCode string `json:"enterpriseCode"`
}
type DownstreamReceipt struct {
@@ -86,27 +87,30 @@ type DownstreamUplink struct {
}
type downstreamSession struct {
messageID string
account string
protocol string
srcID string
phoneNumber string
gatewayMsgID uint64
remoteIP string
connectedAt time.Time
conn *cmpp.Conn
mu *sync.Mutex
presence PresenceStore
instanceID string
messageID string
account string
enterpriseCode string
protocol string
srcID string
phoneNumber string
gatewayMsgID uint64
remoteIP string
connectedAt time.Time
conn *cmpp.Conn
mu *sync.Mutex
presence PresenceStore
instanceID string
}
var downstreamRegistry = struct {
sync.RWMutex
byMessageID map[string]*downstreamSession
byAccount map[string]*downstreamSession
byConn map[*cmpp.Conn]*downstreamSession
}{
byMessageID: make(map[string]*downstreamSession),
byAccount: make(map[string]*downstreamSession),
byConn: make(map[*cmpp.Conn]*downstreamSession),
}
func (s Server) ListenAndServe() error {
@@ -128,38 +132,39 @@ func (s Server) handleLogin(response *cmpp.Response, packet *cmpp.Packet, logger
if !ok {
return true, nil
}
resp := response.Packer.(*cmpp.Cmpp3ConnRspPkt)
resp.Version = 0x30
account := strings.TrimRight(req.SrcAddr, "\x00")
if account == "" {
resp.Status = uint32(cmpp.ErrnoConnInvalidSrcAddr)
setInboundConnectResponse(response.Packer, cmpp.ErrnoConnInvalidSrcAddr, req.AuthSrc, "", req.Version)
return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnInvalidSrcAddr]
}
if req.Version != cmpp.V20 && req.Version != cmpp.V21 && req.Version != cmpp.V30 {
setInboundConnectResponse(response.Packer, cmpp.ErrnoConnVerTooHigh, req.AuthSrc, "", cmpp.V30)
return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnVerTooHigh]
}
auth, err := s.authenticate(packet.Conn.Conn.RemoteAddr(), account, req.AuthSrc, req.Timestamp)
if err != nil {
logger.Printf("cmpp inbound auth failed account=%s remote=%s err=%v", account, packet.Conn.Conn.RemoteAddr(), err)
resp.Status = uint32(cmpp.ErrnoConnAuthFailed)
setInboundConnectResponse(response.Packer, cmpp.ErrnoConnAuthFailed, req.AuthSrc, "", req.Version)
return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnAuthFailed]
}
authSource := []byte(req.AuthSrc)
authISMG := md5.Sum(bytes.Join([][]byte{{byte(resp.Status)}, authSource, []byte(auth.PasswordCipher)}, nil))
resp.AuthIsmg = string(authISMG[:])
setInboundConnectResponse(response.Packer, 0, req.AuthSrc, auth.PasswordCipher, req.Version)
session := downstreamSession{
account: strings.TrimSpace(defaultString(auth.Account, account)),
protocol: cmppVersionName(req.Version),
srcID: strings.TrimSpace(auth.Account),
remoteIP: remoteIP(packet.Conn.Conn.RemoteAddr()),
connectedAt: time.Now().UTC(),
conn: packet.Conn,
mu: &sync.Mutex{},
presence: s.PresenceStore,
instanceID: s.gatewayInstanceID(),
account: strings.TrimSpace(defaultString(auth.Account, account)),
enterpriseCode: strings.TrimSpace(auth.EnterpriseCode),
protocol: cmppVersionName(req.Version),
srcID: strings.TrimSpace(auth.Account),
remoteIP: remoteIP(packet.Conn.Conn.RemoteAddr()),
connectedAt: time.Now().UTC(),
conn: packet.Conn,
mu: &sync.Mutex{},
presence: s.PresenceStore,
instanceID: s.gatewayInstanceID(),
}
rememberAccount(session)
go s.flushPending(defaultString(auth.Account, account), logger)
logger.Printf(
"cmpp inbound event=login_accepted protocol=%s requested_version=0x%02x response_version=0x30 account=%s remote=%s",
cmppVersionName(req.Version), uint8(req.Version), account, packet.Conn.Conn.RemoteAddr(),
"cmpp inbound event=login_accepted protocol=%s requested_version=0x%02x response_version=0x%02x account=%s remote=%s",
cmppVersionName(req.Version), uint8(req.Version), uint8(req.Version), account, packet.Conn.Conn.RemoteAddr(),
)
return false, nil
}
@@ -169,16 +174,35 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge
if !ok {
return true, nil
}
account := strings.TrimRight(req.msgSrc, "\x00")
session := findSessionByConn(packet.Conn)
if session == nil || strings.TrimSpace(session.account) == "" {
logger.Printf(
"cmpp inbound event=submit_rejected protocol=%s packet_type=%s remote=%s seq=%d result=9 stage=session reason=%q",
req.protocol, req.protocol, packet.Conn.Conn.RemoteAddr(), req.sequenceID, "authenticated connection session not found",
)
setInboundSubmitResponse(response.Packer, 0, 9)
return false, nil
}
account := session.account
enterpriseCode := strings.TrimRight(req.msgSrc, "\x00")
if session.enterpriseCode != "" && enterpriseCode != session.enterpriseCode {
logger.Printf(
"cmpp inbound event=submit_rejected protocol=%s packet_type=%s account=%s enterprise_code=%s remote=%s seq=%d result=9 stage=protocol reason=%q",
defaultString(session.protocol, req.protocol), req.protocol, account, enterpriseCode, packet.Conn.Conn.RemoteAddr(), req.sequenceID,
fmt.Sprintf("enterprise code mismatch: expected %s", session.enterpriseCode),
)
setInboundSubmitResponse(response.Packer, 0, 9)
return false, nil
}
phone := ""
if len(req.destTerminalIDs) > 0 {
phone = strings.TrimRight(req.destTerminalIDs[0], "\x00")
}
remote := packet.Conn.Conn.RemoteAddr()
clientProtocol := inboundClientProtocol(account, packet.Conn, req.protocol)
clientProtocol := defaultString(session.protocol, req.protocol)
logger.Printf(
"cmpp inbound event=submit_received protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s src_id=%s msg_fmt=%d pk=%d/%d dest_count=%d content_bytes=%d",
clientProtocol, req.protocol, account, remote, req.sequenceID, phone, strings.TrimSpace(req.srcID), req.msgFmt,
"cmpp inbound event=submit_received protocol=%s packet_type=%s account=%s enterprise_code=%s remote=%s seq=%d phone=%s src_id=%s msg_fmt=%d pk=%d/%d dest_count=%d content_bytes=%d",
clientProtocol, req.protocol, account, enterpriseCode, remote, req.sequenceID, phone, strings.TrimSpace(req.srcID), req.msgFmt,
req.pkNumber, req.pkTotal, len(req.destTerminalIDs), len(req.msgContent),
)
content, err := decodeContent(req.msgFmt, req.msgContent)
@@ -216,18 +240,19 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge
gatewayMsgID := messageIDFrom(result.MessageID, req.sequenceID)
setInboundSubmitResponse(response.Packer, gatewayMsgID, 0)
rememberDownstream(downstreamSession{
messageID: result.MessageID,
account: account,
protocol: clientProtocol,
srcID: strings.TrimSpace(req.srcID),
phoneNumber: phone,
gatewayMsgID: gatewayMsgID,
remoteIP: remoteIP(remote),
connectedAt: time.Now().UTC(),
conn: packet.Conn,
mu: &sync.Mutex{},
presence: s.PresenceStore,
instanceID: s.gatewayInstanceID(),
messageID: result.MessageID,
account: account,
enterpriseCode: session.enterpriseCode,
protocol: clientProtocol,
srcID: strings.TrimSpace(req.srcID),
phoneNumber: phone,
gatewayMsgID: gatewayMsgID,
remoteIP: remoteIP(remote),
connectedAt: time.Now().UTC(),
conn: packet.Conn,
mu: &sync.Mutex{},
presence: s.PresenceStore,
instanceID: s.gatewayInstanceID(),
})
logger.Printf(
"cmpp inbound event=submit_accepted protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s result=0 message_id=%s gateway_message_id=%d duration_ms=%d content_chars=%d content_hash=%s",
@@ -278,14 +303,25 @@ func setInboundSubmitResponse(packet any, messageID uint64, result uint32) {
}
}
func inboundClientProtocol(account string, conn *cmpp.Conn, fallback string) string {
func setInboundConnectResponse(packet any, status uint8, authSource string, secret string, version cmpp.Type) {
switch resp := packet.(type) {
case *cmpp.Cmpp2ConnRspPkt:
resp.Status = status
resp.AuthSrc = authSource
resp.Secret = secret
resp.Version = version
case *cmpp.Cmpp3ConnRspPkt:
resp.Status = uint32(status)
resp.AuthSrc = authSource
resp.Secret = secret
resp.Version = version
}
}
func findSessionByConn(conn *cmpp.Conn) *downstreamSession {
downstreamRegistry.RLock()
defer downstreamRegistry.RUnlock()
session := downstreamRegistry.byAccount[account]
if session != nil && session.conn == conn && session.protocol != "" {
return session.protocol
}
return fallback
return downstreamRegistry.byConn[conn]
}
func cmppVersionName(version cmpp.Type) string {
@@ -483,6 +519,7 @@ func rememberDownstream(session downstreamSession) {
session.touchPresence("connected", true, false)
downstreamRegistry.Lock()
downstreamRegistry.byMessageID[session.messageID] = &session
downstreamRegistry.byConn[session.conn] = &session
if session.account != "" {
downstreamRegistry.byAccount[session.account] = &session
}
@@ -496,6 +533,7 @@ func rememberAccount(session downstreamSession) {
session.touchPresence("connected", false, false)
downstreamRegistry.Lock()
downstreamRegistry.byAccount[session.account] = &session
downstreamRegistry.byConn[session.conn] = &session
downstreamRegistry.Unlock()
}
@@ -514,6 +552,9 @@ func forgetDownstream(session *downstreamSession) {
delete(downstreamRegistry.byAccount, session.account)
}
}
if current := downstreamRegistry.byConn[session.conn]; current == session {
delete(downstreamRegistry.byConn, session.conn)
}
downstreamRegistry.Unlock()
_ = session.removePresence()
}
@@ -700,16 +741,7 @@ func PushReceipt(event DownstreamReceipt) (bool, error) {
if err != nil {
return false, err
}
deliver := &cmpp.Cmpp3DeliverReqPkt{
MsgId: session.gatewayMsgID,
DestId: session.srcID,
ServiceId: "cmpp",
MsgFmt: 0,
SrcTerminalId: defaultString(event.PhoneNumber, session.phoneNumber),
RegisterDelivery: 1,
MsgLength: uint8(cmpp.CmppReceiptPktLen),
MsgContent: string(receiptBytes),
}
deliver := downstreamDeliverPacket(session, session.gatewayMsgID, session.srcID, defaultString(event.PhoneNumber, session.phoneNumber), 0, 1, string(receiptBytes))
return sendDownstream(session, deliver)
}
@@ -722,19 +754,33 @@ func PushUplink(event DownstreamUplink) (bool, error) {
if err != nil {
return false, err
}
deliver := &cmpp.Cmpp3DeliverReqPkt{
MsgId: messageIDFrom(defaultString(event.MessageID, event.Account), uint32(time.Now().UnixNano())),
DestId: defaultString(event.DestID, session.srcID),
ServiceId: "cmpp",
MsgFmt: 8,
SrcTerminalId: event.PhoneNumber,
RegisterDelivery: 0,
MsgLength: uint8(len(content)),
MsgContent: content,
}
deliver := downstreamDeliverPacket(
session,
messageIDFrom(defaultString(event.MessageID, event.Account), uint32(time.Now().UnixNano())),
defaultString(event.DestID, session.srcID),
event.PhoneNumber,
8,
0,
content,
)
return sendDownstream(session, deliver)
}
func downstreamDeliverPacket(session *downstreamSession, messageID uint64, destID string, sourceTerminalID string, msgFmt uint8, registerDelivery uint8, content string) cmpp.Packer {
if session != nil && (session.protocol == "cmpp20" || session.protocol == "cmpp21") {
return &cmpp.Cmpp2DeliverReqPkt{
MsgId: messageID, DestId: destID, ServiceId: "cmpp", MsgFmt: msgFmt,
SrcTerminalId: sourceTerminalID, RegisterDelivery: registerDelivery,
MsgLength: uint8(len(content)), MsgContent: content,
}
}
return &cmpp.Cmpp3DeliverReqPkt{
MsgId: messageID, DestId: destID, ServiceId: "cmpp", MsgFmt: msgFmt,
SrcTerminalId: sourceTerminalID, RegisterDelivery: registerDelivery,
MsgLength: uint8(len(content)), MsgContent: content,
}
}
func findSession(messageID string, account string) *downstreamSession {
downstreamRegistry.RLock()
defer downstreamRegistry.RUnlock()
@@ -749,7 +795,7 @@ func findSession(messageID string, account string) *downstreamSession {
return nil
}
func sendDownstream(session *downstreamSession, deliver *cmpp.Cmpp3DeliverReqPkt) (bool, error) {
func sendDownstream(session *downstreamSession, deliver cmpp.Packer) (bool, error) {
session.mu.Lock()
defer session.mu.Unlock()
if err := session.conn.SendPkt(deliver, <-session.conn.SeqId); err != nil {
gateway/internal/inbound/server_test.go
+129 -7
View File
@@ -93,6 +93,8 @@ func (m *memoryRecoveryStore) GetAccountRecoveryStatus(_ context.Context, accoun
}
func TestInboundServerAuthenticatesAndSubmits(t *testing.T) {
resetDownstreamRegistry()
defer resetDownstreamRegistry()
account := "100001"
password := "secret-hash"
var gotAuth authRequest
@@ -103,7 +105,7 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) {
if err := json.NewDecoder(r.Body).Decode(&gotAuth); err != nil {
t.Fatalf("decode auth: %v", err)
}
_ = json.NewEncoder(w).Encode(authResponse{PasswordCipher: password})
_ = json.NewEncoder(w).Encode(authResponse{PasswordCipher: password, Account: account, EnterpriseCode: account})
case "/api/gateway/events/inbound/submit":
if err := json.NewDecoder(r.Body).Decode(&gotSubmit); err != nil {
t.Fatalf("decode submit: %v", err)
@@ -186,6 +188,92 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) {
}
}
func TestInboundServerNegotiatesCMPP2AndUsesAuthenticatedAccountForSubmit(t *testing.T) {
resetDownstreamRegistry()
defer resetDownstreamRegistry()
account := "100001"
password := "secret-hash"
var gotSubmit submitRequest
submitCalls := 0
api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/api/gateway/events/inbound/authenticate":
_ = json.NewEncoder(w).Encode(authResponse{PasswordCipher: password, Account: account, EnterpriseCode: "SP0001"})
case "/api/gateway/events/inbound/submit":
submitCalls++
if err := json.NewDecoder(r.Body).Decode(&gotSubmit); err != nil {
t.Fatalf("decode submit: %v", err)
}
_ = json.NewEncoder(w).Encode(submitResponse{Accepted: true, MessageID: "MSG-CMPP2"})
case "/api/gateway/events/downstream/pending":
_ = json.NewEncoder(w).Encode([]pendingDelivery{})
default:
t.Fatalf("unexpected api path: %s", r.URL.Path)
}
}))
defer api.Close()
addr := reserveTCPAddr(t)
go func() {
_ = (Server{Addr: addr, APIBaseURL: api.URL + "/api"}).ListenAndServe()
}()
time.Sleep(300 * time.Millisecond)
client := cmpp.NewClient(cmpp.V20)
defer client.Disconnect()
if err := client.Connect(addr, account, password, 2*time.Second); err != nil {
t.Fatalf("connect CMPP2 inbound: %v", err)
}
content, err := cmpputils.Utf8ToUcs2("测试CMPP2")
if err != nil {
t.Fatalf("encode content: %v", err)
}
_, err = client.SendReqPkt(&cmpp.Cmpp2SubmitReqPkt{
PkTotal: 1, PkNumber: 1, RegisteredDelivery: 1, MsgLevel: 1,
ServiceId: "cmpp", FeeUserType: 2, FeeTerminalId: "13500002696",
MsgFmt: 8, MsgSrc: "SP0001", FeeType: "02", FeeCode: "0",
SrcId: "10690000", DestUsrTl: 1, DestTerminalId: []string{"13500002696"},
MsgLength: uint8(len(content)), MsgContent: content,
})
if err != nil {
t.Fatalf("send CMPP2 submit: %v", err)
}
rsp := recvSubmitRsp20(t, client)
if rsp.Result != 0 || rsp.MsgId == 0 {
t.Fatalf("unexpected CMPP2 submit response: %+v", rsp)
}
if gotSubmit.Account != account || gotSubmit.PhoneNumber != "13500002696" || gotSubmit.Content != "测试CMPP2" {
t.Fatalf("unexpected CMPP2 submit payload: %+v", gotSubmit)
}
_, err = client.SendReqPkt(&cmpp.Cmpp2SubmitReqPkt{
PkTotal: 1, PkNumber: 1, RegisteredDelivery: 1, MsgLevel: 1,
ServiceId: "cmpp", FeeUserType: 2, FeeTerminalId: "13500002696",
MsgFmt: 8, MsgSrc: "BAD001", FeeType: "02", FeeCode: "0",
SrcId: "10690000", DestUsrTl: 1, DestTerminalId: []string{"13500002696"},
MsgLength: uint8(len(content)), MsgContent: content,
})
if err != nil {
t.Fatalf("send mismatched enterprise code: %v", err)
}
if rejected := recvSubmitRsp20(t, client); rejected.Result != 9 {
t.Fatalf("expected enterprise code rejection, got %+v", rejected)
}
if submitCalls != 1 {
t.Fatalf("submit API calls = %d, want 1", submitCalls)
}
delivered, err := PushReceipt(DownstreamReceipt{
MessageID: "MSG-CMPP2", PhoneNumber: "13500002696", ReceiptStatus: "delivered",
DeliveredAt: time.Now().UTC().Format(time.RFC3339Nano),
})
if err != nil || !delivered {
t.Fatalf("push CMPP2 receipt delivered=%v err=%v", delivered, err)
}
deliver := recvDeliver20(t, client)
if deliver.RegisterDelivery != 1 {
t.Fatalf("expected CMPP2 receipt deliver, got %+v", deliver)
}
}
func TestPostIncludesAPIErrorResponseBody(t *testing.T) {
api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
@@ -259,20 +347,21 @@ func TestSetInboundSubmitResponseSupportsCMPP2AndCMPP3(t *testing.T) {
}
}
func TestInboundClientProtocolUsesConnectRequestVersion(t *testing.T) {
func TestFindSessionByConnUsesAuthenticatedConnection(t *testing.T) {
resetDownstreamRegistry()
defer resetDownstreamRegistry()
conn := &cmpp.Conn{}
downstreamRegistry.byAccount["100001"] = &downstreamSession{
session := &downstreamSession{
account: "100001",
protocol: "cmpp20",
conn: conn,
}
if got := inboundClientProtocol("100001", conn, "cmpp30"); got != "cmpp20" {
t.Fatalf("protocol = %s, want cmpp20", got)
downstreamRegistry.byConn[conn] = session
if got := findSessionByConn(conn); got != session {
t.Fatalf("unexpected session: %+v", got)
}
if got := inboundClientProtocol("missing", conn, "cmpp30"); got != "cmpp30" {
t.Fatalf("fallback protocol = %s, want cmpp30", got)
if got := findSessionByConn(&cmpp.Conn{}); got != nil {
t.Fatalf("expected missing session, got %+v", got)
}
}
@@ -446,11 +535,28 @@ func recvDeliver(t *testing.T, client *cmpp.Client) *cmpp.Cmpp3DeliverReqPkt {
return nil
}
func recvDeliver20(t *testing.T, client *cmpp.Client) *cmpp.Cmpp2DeliverReqPkt {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
packet, err := client.RecvAndUnpackPkt(200 * time.Millisecond)
if err != nil {
continue
}
if deliver, ok := packet.(*cmpp.Cmpp2DeliverReqPkt); ok {
return deliver
}
}
t.Fatal("timed out waiting CMPP2 deliver request")
return nil
}
func resetDownstreamRegistry() {
downstreamRegistry.Lock()
defer downstreamRegistry.Unlock()
downstreamRegistry.byAccount = make(map[string]*downstreamSession)
downstreamRegistry.byMessageID = make(map[string]*downstreamSession)
downstreamRegistry.byConn = make(map[*cmpp.Conn]*downstreamSession)
}
func reserveTCPAddr(t *testing.T) string {
@@ -481,3 +587,19 @@ func recvSubmitRsp(t *testing.T, client *cmpp.Client) *cmpp.Cmpp3SubmitRspPkt {
t.Fatal("timed out waiting submit response")
return nil
}
func recvSubmitRsp20(t *testing.T, client *cmpp.Client) *cmpp.Cmpp2SubmitRspPkt {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
packet, err := client.RecvAndUnpackPkt(200 * time.Millisecond)
if err != nil {
continue
}
if rsp, ok := packet.(*cmpp.Cmpp2SubmitRspPkt); ok {
return rsp
}
}
t.Fatal("timed out waiting CMPP2 submit response")
return nil
}