334 lines
12 KiB
Go
334 lines
12 KiB
Go
package inbound
|
|
|
|
import (
|
|
"context"
|
|
"crypto/md5"
|
|
"errors"
|
|
"fmt"
|
|
cmpp "github.com/bigwhite/gocmpp"
|
|
cmpputils "github.com/bigwhite/gocmpp/utils"
|
|
"log"
|
|
"net"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// One client Submit may contain multiple destinations. The API result retains
|
|
// one internal message mapping per destination while CMPP receives one response.
|
|
|
|
type submitRequest struct {
|
|
Account string `json:"account"`
|
|
PhoneNumber string `json:"phoneNumber,omitempty"`
|
|
PhoneNumbers []string `json:"phoneNumbers,omitempty"`
|
|
Content string `json:"content"`
|
|
SrcID string `json:"srcId,omitempty"`
|
|
DestID string `json:"destId,omitempty"`
|
|
SequenceID uint32 `json:"sequenceId,omitempty"`
|
|
RemoteIP string `json:"remoteIp,omitempty"`
|
|
LongMessage *inboundLongMessageFragment `json:"longMessage,omitempty"`
|
|
}
|
|
|
|
type inboundLongMessageFragment struct {
|
|
Reference int `json:"reference"`
|
|
Total int `json:"total"`
|
|
Index int `json:"index"`
|
|
Format int `json:"format"`
|
|
}
|
|
|
|
type submitResponseMessage struct {
|
|
PhoneNumber string `json:"phoneNumber"`
|
|
MessageID string `json:"messageId"`
|
|
}
|
|
|
|
type submitResponse struct {
|
|
Accepted bool `json:"accepted"`
|
|
Result uint32 `json:"result,omitempty"`
|
|
TenantID string `json:"tenantId,omitempty"`
|
|
ApplicationID string `json:"applicationId,omitempty"`
|
|
MessageID string `json:"messageId"`
|
|
Messages []submitResponseMessage `json:"messages,omitempty"`
|
|
}
|
|
|
|
func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) {
|
|
req, ok := normalizeInboundSubmit(packet.Packer)
|
|
if !ok {
|
|
return true, nil
|
|
}
|
|
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)
|
|
response.AfterSend = s.submitResponseProtocolLogger("", req.protocol, req.sequenceID, "", "", 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)
|
|
response.AfterSend = s.submitResponseProtocolLogger(account, defaultString(session.protocol, req.protocol), req.sequenceID, "", "", 0, 9)
|
|
return false, nil
|
|
}
|
|
phones := make([]string, len(req.destTerminalIDs))
|
|
for index, destination := range req.destTerminalIDs {
|
|
phones[index] = strings.TrimSpace(strings.TrimRight(destination, "\x00"))
|
|
}
|
|
phone := ""
|
|
if len(phones) > 0 {
|
|
phone = phones[0]
|
|
}
|
|
remote := packet.Conn.Conn.RemoteAddr()
|
|
clientProtocol := defaultString(session.protocol, req.protocol)
|
|
logger.Printf(
|
|
"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, longMessage, err := decodeInboundSubmitContent(req)
|
|
if err != nil {
|
|
logger.Printf(
|
|
"cmpp inbound event=submit_rejected protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s result=9 stage=decode reason=%q",
|
|
clientProtocol, req.protocol, account, remote, req.sequenceID, phone, err,
|
|
)
|
|
setInboundSubmitResponse(response.Packer, 0, 9)
|
|
response.AfterSend = s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, "", 0, 9)
|
|
return false, nil
|
|
}
|
|
contentHash := fmt.Sprintf("%x", md5.Sum([]byte(content)))
|
|
startedAt := time.Now()
|
|
releaseSubmitBarrier := beginDownstreamSubmitBarrier(packet.Conn)
|
|
result, err := s.submit(remote, submitRequest{
|
|
Account: account,
|
|
PhoneNumber: phone,
|
|
PhoneNumbers: phones,
|
|
Content: content,
|
|
SrcID: req.srcID,
|
|
DestID: phone,
|
|
SequenceID: req.sequenceID,
|
|
RemoteIP: remoteIP(remote),
|
|
LongMessage: longMessage,
|
|
})
|
|
if err != nil || !result.Accepted {
|
|
reason := "api returned accepted=false"
|
|
if err != nil {
|
|
reason = err.Error()
|
|
}
|
|
responseResult := result.Result
|
|
if responseResult == 0 {
|
|
responseResult = 9
|
|
}
|
|
logger.Printf(
|
|
"cmpp inbound event=submit_rejected protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s result=%d stage=business duration_ms=%d content_chars=%d content_hash=%s reason=%q",
|
|
clientProtocol, req.protocol, account, remote, req.sequenceID, phone, responseResult, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash, reason,
|
|
)
|
|
setInboundSubmitResponse(response.Packer, 0, responseResult)
|
|
protocolLogger := s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, result.MessageID, 0, responseResult)
|
|
response.AfterSend = func(sendErr error) {
|
|
releaseSubmitBarrier()
|
|
protocolLogger(sendErr)
|
|
}
|
|
return false, nil
|
|
}
|
|
gatewayMsgID := messageIDFrom(result.MessageID, req.sequenceID)
|
|
setInboundSubmitResponse(response.Packer, gatewayMsgID, 0)
|
|
responseMessages := result.Messages
|
|
if len(responseMessages) == 0 {
|
|
responseMessages = []submitResponseMessage{{PhoneNumber: phone, MessageID: result.MessageID}}
|
|
}
|
|
for index, acceptedMessage := range responseMessages {
|
|
acceptedPhone := strings.TrimSpace(acceptedMessage.PhoneNumber)
|
|
if acceptedPhone == "" && index < len(phones) {
|
|
acceptedPhone = phones[index]
|
|
}
|
|
rememberDownstream(downstreamSession{
|
|
messageID: acceptedMessage.MessageID,
|
|
account: account,
|
|
tenantID: result.TenantID,
|
|
applicationID: result.ApplicationID,
|
|
enterpriseCode: session.enterpriseCode,
|
|
protocol: clientProtocol,
|
|
srcID: strings.TrimSpace(req.srcID),
|
|
phoneNumber: acceptedPhone,
|
|
gatewayMsgID: gatewayMsgID,
|
|
remoteIP: remoteIP(remote),
|
|
connectedAt: time.Now().UTC(),
|
|
connectionID: session.connectionID,
|
|
conn: packet.Conn,
|
|
mu: &sync.Mutex{},
|
|
presence: s.PresenceStore,
|
|
instanceID: s.gatewayInstanceID(),
|
|
report: session.report,
|
|
deliveryReport: session.deliveryReport,
|
|
protocolLog: session.protocolLog,
|
|
})
|
|
}
|
|
if current := findSessionByConn(packet.Conn); current != nil && current.report != nil {
|
|
go current.report(current, "submit", "")
|
|
}
|
|
response.AfterSend = func(sendErr error) {
|
|
releaseSubmitBarrier()
|
|
s.emitProtocolLog(protocolLogEvent{
|
|
Protocol: "cmpp",
|
|
Direction: "platform_to_client",
|
|
EventType: "submit_resp",
|
|
Status: protocolSendStatus(sendErr),
|
|
TenantID: result.TenantID,
|
|
ApplicationID: result.ApplicationID,
|
|
Account: account,
|
|
MessageID: result.MessageID,
|
|
GatewayMessageID: fmt.Sprint(gatewayMsgID),
|
|
Phone: phone,
|
|
ResultCode: protocolSendResultCode(sendErr, 0),
|
|
Detail: protocolSubmitResponseDetail(req.sequenceID, sendErr),
|
|
})
|
|
if sendErr != nil {
|
|
return
|
|
}
|
|
go func() {
|
|
if _, err := s.flushPending(account, logger); err != nil {
|
|
logger.Printf("cmpp inbound event=post_submit_pending_flush_failed account=%s message_id=%s error=%q", account, result.MessageID, err.Error())
|
|
}
|
|
}()
|
|
}
|
|
logger.Printf(
|
|
"cmpp inbound event=submit_accepted protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s dest_count=%d accepted_count=%d result=0 message_id=%s gateway_message_id=%d duration_ms=%d content_chars=%d content_hash=%s",
|
|
clientProtocol, req.protocol, account, remote, req.sequenceID, phone, len(phones), len(responseMessages), result.MessageID, gatewayMsgID, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash,
|
|
)
|
|
return false, nil
|
|
}
|
|
|
|
type inboundSubmitPacket struct {
|
|
protocol string
|
|
pkTotal uint8
|
|
pkNumber uint8
|
|
tpUdhi uint8
|
|
msgFmt uint8
|
|
msgSrc string
|
|
srcID string
|
|
destTerminalIDs []string
|
|
msgContent string
|
|
sequenceID uint32
|
|
}
|
|
|
|
func normalizeInboundSubmit(packet any) (inboundSubmitPacket, bool) {
|
|
switch req := packet.(type) {
|
|
case *cmpp.Cmpp2SubmitReqPkt:
|
|
return inboundSubmitPacket{
|
|
protocol: "cmpp20", pkTotal: req.PkTotal, pkNumber: req.PkNumber, tpUdhi: req.TpUdhi, msgFmt: req.MsgFmt,
|
|
msgSrc: req.MsgSrc, srcID: req.SrcId, destTerminalIDs: req.DestTerminalId,
|
|
msgContent: req.MsgContent, sequenceID: req.SeqId,
|
|
}, true
|
|
case *cmpp.Cmpp3SubmitReqPkt:
|
|
return inboundSubmitPacket{
|
|
protocol: "cmpp30", pkTotal: req.PkTotal, pkNumber: req.PkNumber, tpUdhi: req.TpUdhi, msgFmt: req.MsgFmt,
|
|
msgSrc: req.MsgSrc, srcID: req.SrcId, destTerminalIDs: req.DestTerminalId,
|
|
msgContent: req.MsgContent, sequenceID: req.SeqId,
|
|
}, true
|
|
default:
|
|
return inboundSubmitPacket{}, false
|
|
}
|
|
}
|
|
|
|
func setInboundSubmitResponse(packet any, messageID uint64, result uint32) {
|
|
switch resp := packet.(type) {
|
|
case *cmpp.Cmpp2SubmitRspPkt:
|
|
resp.MsgId = messageID
|
|
resp.Result = uint8(result)
|
|
case *cmpp.Cmpp3SubmitRspPkt:
|
|
resp.MsgId = messageID
|
|
resp.Result = result
|
|
}
|
|
}
|
|
|
|
func (s Server) submit(remote net.Addr, payload submitRequest) (submitResponse, error) {
|
|
payload.RemoteIP = remoteIP(remote)
|
|
var result submitResponse
|
|
err := s.post(context.Background(), "/gateway/events/inbound/submit", payload, &result)
|
|
return result, err
|
|
}
|
|
|
|
func decodeContent(format uint8, content string) (string, error) {
|
|
switch format {
|
|
case 8:
|
|
return cmpputils.Ucs2ToUtf8(content)
|
|
case 15:
|
|
return cmpputils.GB18030ToUtf8(content)
|
|
default:
|
|
return content, nil
|
|
}
|
|
}
|
|
|
|
func decodeInboundSubmitContent(req inboundSubmitPacket) (string, *inboundLongMessageFragment, error) {
|
|
raw := []byte(req.msgContent)
|
|
if req.tpUdhi == 0 && req.pkTotal <= 1 {
|
|
content, err := decodeContent(req.msgFmt, req.msgContent)
|
|
return content, nil, err
|
|
}
|
|
if len(raw) == 0 {
|
|
return "", nil, errors.New("UDH message content is empty")
|
|
}
|
|
|
|
headerLength := int(raw[0]) + 1
|
|
if headerLength > len(raw) {
|
|
return "", nil, fmt.Errorf("UDH length %d exceeds message content length %d", headerLength, len(raw))
|
|
}
|
|
|
|
var reference, total, index int
|
|
switch {
|
|
case len(raw) >= 6 && raw[0] == 0x05 && raw[1] == 0x00 && raw[2] == 0x03:
|
|
reference = int(raw[3])
|
|
total = int(raw[4])
|
|
index = int(raw[5])
|
|
case len(raw) >= 7 && raw[0] == 0x06 && raw[1] == 0x08 && raw[2] == 0x04:
|
|
reference = int(raw[3])<<8 | int(raw[4])
|
|
total = int(raw[5])
|
|
index = int(raw[6])
|
|
default:
|
|
if req.pkTotal > 1 {
|
|
return "", nil, errors.New("concatenated CMPP submit is missing a supported 8-bit or 16-bit UDH")
|
|
}
|
|
content, err := decodeContent(req.msgFmt, string(raw[headerLength:]))
|
|
return content, nil, err
|
|
}
|
|
if total < 2 || index < 1 || index > total {
|
|
return "", nil, fmt.Errorf("invalid concatenated UDH total/index %d/%d", index, total)
|
|
}
|
|
if req.pkTotal > 0 && int(req.pkTotal) != total {
|
|
return "", nil, fmt.Errorf("PkTotal %d does not match UDH total %d", req.pkTotal, total)
|
|
}
|
|
if req.pkNumber > 0 && int(req.pkNumber) != index {
|
|
return "", nil, fmt.Errorf("PkNumber %d does not match UDH index %d", req.pkNumber, index)
|
|
}
|
|
|
|
content, err := decodeContent(req.msgFmt, string(raw[headerLength:]))
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
return content, &inboundLongMessageFragment{
|
|
Reference: reference,
|
|
Total: total,
|
|
Index: index,
|
|
Format: int(req.msgFmt),
|
|
}, nil
|
|
}
|
|
|
|
func messageIDFrom(value string, seq uint32) uint64 {
|
|
hash := md5.Sum([]byte(value))
|
|
result := uint64(seq)
|
|
for _, item := range hash[:6] {
|
|
result = (result << 8) + uint64(item)
|
|
}
|
|
if result == 0 {
|
|
return uint64(time.Now().UnixNano())
|
|
}
|
|
return result
|
|
}
|