From bb4992f0f9aace473e26dde5b9efe106f4c05fad Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Sat, 11 Jul 2026 11:00:42 +0800 Subject: [PATCH] fix: negotiate downstream cmpp protocol versions --- api/src/send-chain/send-chain.service.spec.ts | 15 ++ api/src/send-chain/send-chain.service.ts | 1 + .../first-version-development-requirements.md | 4 +- docs/system-functional-test-cases.md | 2 + docs/testing-progress.md | 3 + gateway/internal/inbound/server.go | 192 +++++++++++------- gateway/internal/inbound/server_test.go | 136 ++++++++++++- gateway/third_party/gocmpp/server.go | 6 +- 8 files changed, 276 insertions(+), 83 deletions(-) diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index 6281936..7ff9f6c 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -69,6 +69,7 @@ function createPrismaMock() { id: 'app-1', tenantId: 'tenant-1', cmppAccount: '100001', + cmppEnterpriseCode: 'SP0001', secretHash: 'secret-hash', status: 'active', interfaceEnabled: true, @@ -439,6 +440,20 @@ describe('SendChainService', () => { expect(prisma.smsBatchTask.create).not.toHaveBeenCalled(); }); + it('returns the application enterprise code after Gateway authentication', async () => { + const { service } = createService(); + + await expect(service.authenticateInboundApplication({ + account: '100001', + password: 'secret-hash', + remoteIp: '127.0.0.1', + })).resolves.toEqual(expect.objectContaining({ + account: '100001', + enterpriseCode: 'SP0001', + status: 'authenticated', + })); + }); + it('rejects Gateway authentication when application interface is disabled', async () => { const { service, prisma } = createService(); prisma.smsApplication.findFirst.mockResolvedValue({ diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index d6db05b..5f580d5 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -1414,6 +1414,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { applicationId: application.id, tenantId: application.tenantId, account: application.cmppAccount, + enterpriseCode: application.cmppEnterpriseCode, passwordCipher: application.secretHash, status: 'authenticated', }; diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index 96b7761..c80680d 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -210,11 +210,11 @@ #### 4.8.2 下游客户 CMPP 接入能力 1. Gateway 必须监听生产 CMPP 端口 `17890`，作为平台侧 CMPP Server 接收企业客户系统连接；该端口不是 HTTP 健康检查或控制接口。 -2. Gateway 必须按企业应用生成的 CMPP 接入参数校验客户 connect/login，包括客户侧企业代码、账号、密码、CMPP 版本、源 IP 白名单、短信接口开关、应用状态、企业状态、连接数上限；客户侧企业代码必须来自 `SmsApplication.cmppEnterpriseCode`，且 `interfaceEnabled=false` 时必须拒绝鉴权和后续 submit。 +2. Gateway 必须按企业应用生成的 CMPP 接入参数校验客户 connect/login，包括客户侧企业代码、账号、密码、CMPP 版本、源 IP 白名单、短信接口开关、应用状态、企业状态、连接数上限；客户侧企业代码必须来自 `SmsApplication.cmppEnterpriseCode`，且 `interfaceEnabled=false` 时必须拒绝鉴权和后续 submit。Gateway 必须根据 CONNECT `Version` 为每条 TCP 连接独立协商 CMPP2.0/2.1/3.0 解包与响应类型，不得用固定 CMPP3.0 结构解析 CMPP2.0 Submit。 3. 客户端应用的 IP 白名单必须对 CMPP 下游连接生效；未命中白名单、应用停用、企业停用、密码错误、超过连接数上限时必须拒绝连接并记录系统日志。 4. Gateway 必须维护应用级下游连接状态，回写 applicationId、tenantId、connectionId、currentConnections、desiredConnections、lastHeartbeatAt、lastError，运营端企业应用列表和连接详情必须来自这些真实状态。 5. Gateway 必须实现下游 ActiveTest、Terminate、异常断开处理；断开后连接数和状态必须及时回写。 -6. Gateway 必须处理客户提交的 CMPP Submit，将手机号、内容、源地址、企业应用、客户消息序号等转换为平台发送请求。 +6. Gateway 必须处理客户提交的 CMPP Submit，将手机号、内容、源地址、企业应用、客户消息序号等转换为平台发送请求。应用身份必须来自当前已鉴权 TCP 连接绑定的 `cmppAccount`；Submit `MsgSrc` 是独立的企业代码，必须与该应用 `cmppEnterpriseCode` 匹配，不得把 `MsgSrc` 当作登录账号查找应用。 7. 下游 CMPP Submit 进入平台后，不创建客户端批量任务，但必须按手机号维度创建 `sms_message_record`，source 标记为 `cmpp`，并保留客户侧 sequence/msgId 映射。 8. 下游 CMPP Submit 必须复用 NestJS 发送前校验：企业/应用状态、IP 白名单、签名/模板报备、模板匹配策略、风控、黑名单、余额/授信、运营商识别、通道组路由。 9. 对客户 Submit 的响应必须符合 CMPP 协议：参数错误、鉴权失败、余额不足、模板或签名未通过、无可用通道、风控拒绝等应映射为明确失败状态；已接收进入平台发送链路时返回成功并生成可追踪平台 messageId。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index e8f7f9c..8cc4c4c 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -1005,8 +1005,10 @@ - 预期结果： - 17890 是真实 CMPP Server 监听，不是 HTTP 端口。 - bind 阶段调用真实 NestJS API 校验账号、密码、企业状态、认证状态、应用状态、短信接口开关和 IP 白名单。 + - CMPP2.0 和 CMPP3.0 连接分别返回对应版本 ConnectResp，后续 Submit/Deliver 按该 TCP 连接协商版本解包和组包，不发生字段错位。 - 密码错误、应用停用、企业停用、短信接口关闭、IP 不在白名单时 connect/login 被拒绝。 - submit 被接受后返回 CMPP SubmitResp 成功，并在真实数据库创建 `sourceType=cmpp` 的发送记录，进入真实发送链路。 + - Submit 应用身份使用 bind 已鉴权账号；`MsgSrc` 使用应用级企业代码并独立校验。企业代码与登录账号不同时仍能正确定位应用，企业代码不匹配时返回失败。 - submit 内容不匹配审核模板、余额不足、短信接口关闭、无可用通道时返回明确失败，不得伪造成功。 - Gateway 对每次 submit 记录 `submit_received` 和 `submit_accepted`/`submit_rejected`；日志可按账号、IP、sequenceId、号码和 messageId 定位，拒绝时包含 NestJS 真实业务原因和 CMPP result，但不包含明文短信正文。 - CMPP 包在进入 handler 前因长度、命令字、读包或 Unpack 失败时，Gateway 记录 `read/unpack packet failed`、远端地址、协议模式、错误类型和原始错误，不得静默断开。 diff --git a/docs/testing-progress.md b/docs/testing-progress.md index d6750f2..fbdec7a 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -658,6 +658,8 @@ npm run verify:phase8 - Gateway HTTP 回调在 NestJS 返回非 2xx 时保留最多 64KB 响应体，客户 Submit 失败日志可直接显示模板不匹配、IP 白名单、余额或路由等真实业务原因，不再只显示 HTTP 状态码。 - 日志不记录明文短信正文，仅记录字符数和 MD5 哈希，便于比对同一内容且避免日志泄露。 - 将当前 gocmpp 版本固定为仓库内小型 fork，仅在 server 循环补充底层诊断：包在进入业务 handler 之前发生长度、命令字、包体读取或 Unpack 失败时，记录 `read/unpack packet failed`、远端地址、库解析协议模式、Go 错误类型和原始错误；正常 EOF 断开不记为解包失败。 +- 生产复现确认 CMPP2.0 客户 Submit 被固定 CMPP3.0 解包导致 `MsgSrc/手机号/内容` 错位为空。现在 gocmpp server 按 CONNECT `Version` 将每条连接切换到 CMPP2.0/2.1/3.0 解包模式，并返回同版本 ConnectResp、SubmitResp 和 Deliver。 +- Gateway 使用 bind 时已鉴权会话账号调用 NestJS 入站接口，Submit `MsgSrc` 改为与鉴权返回的应用级 `cmppEnterpriseCode` 独立比对，不再将企业代码误当登录账号。 ### 验证状态 @@ -665,6 +667,7 @@ npm run verify:phase8 - `go test ./... -count=1`：通过。 - `go build ./cmd/gateway`：通过。 - 真实 TCP 非法包用例：向入站端口写入非法 `total_length`，确认业务 handler 未执行时仍产生 `read/unpack packet failed` 日志。 +- CMPP2.0 真实集成用例：客户使用与登录账号不同的 `MsgSrc=SP0001`，完成 V20 ConnectResp、Cmpp2SubmitReq/Resp 和 Cmpp2Deliver Receipt，NestJS 收到的 account 仍为 bind 账号：通过。 ## 2026-07-07 Gateway 上游提交与下游 Deliver 闭环补齐 diff --git a/gateway/internal/inbound/server.go b/gateway/internal/inbound/server.go index 942db36..4d11086 100644 --- a/gateway/internal/inbound/server.go +++ b/gateway/internal/inbound/server.go @@ -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 { diff --git a/gateway/internal/inbound/server_test.go b/gateway/internal/inbound/server_test.go index 31339fe..0af4e48 100644 --- a/gateway/internal/inbound/server_test.go +++ b/gateway/internal/inbound/server_test.go @@ -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 +} diff --git a/gateway/third_party/gocmpp/server.go b/gateway/third_party/gocmpp/server.go index 1b5a975..d8467bf 100644 --- a/gateway/third_party/gocmpp/server.go +++ b/gateway/third_party/gocmpp/server.go @@ -134,12 +134,16 @@ func (c *conn) readPacket() (*Response, error) { if err != nil { return nil, err } - typ := c.server.Typ + typ := c.Conn.Typ var pkt *Packet var rsp *Response switch p := i.(type) { case *CmppConnReqPkt: + if p.Version == V20 || p.Version == V21 || p.Version == V30 { + c.Conn.Typ = p.Version + typ = p.Version + } pkt = &Packet{ Packer: p, Conn: c.Conn,