feat: add cmpp inbound gateway listener

2026-07-07 16:19:14 +08:00
parent ad3b86ba0a
commit cc628d0214
14 changed files with 631 additions and 17 deletions
@@ -7,6 +7,7 @@ import (

 	"cmpp-platform/gateway/internal/control"
 	"cmpp-platform/gateway/internal/health"
+	"cmpp-platform/gateway/internal/inbound"
 )

 func main() {
@@ -14,10 +15,22 @@ func main() {
 	if addr == "" {
 		addr = ":8090"
 	}
+	cmppAddr := os.Getenv("GATEWAY_CMPP_ADDR")
+	if cmppAddr == "" {
+		cmppAddr = ":17890"
+	}
+	apiBaseURL := os.Getenv("API_BASE_URL")
+
+	go func() {
+		log.Printf("cmpp gateway inbound server listening on %s", cmppAddr)
+		if err := (inbound.Server{Addr: cmppAddr, APIBaseURL: apiBaseURL}).ListenAndServe(); err != nil {
+			log.Fatalf("gateway inbound server stopped: %v", err)
+		}
+	}()

 	mux := http.NewServeMux()
 	mux.Handle("/health", health.Handler())
-	control.Register(mux, control.Server{APIBaseURL: os.Getenv("API_BASE_URL")})
+	control.Register(mux, control.Server{APIBaseURL: apiBaseURL})

 	log.Printf("cmpp gateway control server listening on %s", addr)
 	if err := http.ListenAndServe(addr, mux); err != nil {
@@ -0,0 +1,212 @@
+package inbound
+
+import (
+	"bytes"
+	"context"
+	"crypto/md5"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"log"
+	"net"
+	"net/http"
+	"strings"
+	"time"
+
+	cmpp "github.com/bigwhite/gocmpp"
+	cmpputils "github.com/bigwhite/gocmpp/utils"
+)
+
+const defaultHTTPTimeout = 10 * time.Second
+
+type Server struct {
+	Addr       string
+	APIBaseURL string
+	HTTPClient *http.Client
+}
+
+type authRequest struct {
+	Account    string `json:"account"`
+	AuthSource string `json:"authSource"`
+	Timestamp  uint32 `json:"timestamp"`
+	RemoteIP   string `json:"remoteIp,omitempty"`
+}
+
+type submitRequest struct {
+	Account     string `json:"account"`
+	PhoneNumber string `json:"phoneNumber"`
+	Content     string `json:"content"`
+	SrcID       string `json:"srcId,omitempty"`
+	DestID      string `json:"destId,omitempty"`
+	SequenceID  uint32 `json:"sequenceId,omitempty"`
+	RemoteIP    string `json:"remoteIp,omitempty"`
+}
+
+type submitResponse struct {
+	Accepted  bool   `json:"accepted"`
+	MessageID string `json:"messageId"`
+}
+
+type authResponse struct {
+	PasswordCipher string `json:"passwordCipher"`
+}
+
+func (s Server) ListenAndServe() error {
+	addr := s.Addr
+	if addr == "" {
+		addr = ":17890"
+	}
+	return cmpp.ListenAndServe(addr, cmpp.V30, 30*time.Second, 3, nil,
+		cmpp.HandlerFunc(s.handleLogin),
+		cmpp.HandlerFunc(s.handleSubmit),
+	)
+}
+
+func (s Server) handleLogin(response *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) {
+	req, ok := packet.Packer.(*cmpp.CmppConnReqPkt)
+	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)
+		return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnInvalidSrcAddr]
+	}
+	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)
+		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[:])
+	logger.Printf("cmpp inbound account=%s login ok remote=%s", account, packet.Conn.Conn.RemoteAddr())
+	return false, nil
+}
+
+func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) {
+	req, ok := packet.Packer.(*cmpp.Cmpp3SubmitReqPkt)
+	if !ok {
+		return true, nil
+	}
+	resp := response.Packer.(*cmpp.Cmpp3SubmitRspPkt)
+	account := strings.TrimRight(req.MsgSrc, "\x00")
+	phone := ""
+	if len(req.DestTerminalId) > 0 {
+		phone = strings.TrimRight(req.DestTerminalId[0], "\x00")
+	}
+	content, err := decodeContent(req.MsgFmt, req.MsgContent)
+	if err != nil {
+		logger.Printf("cmpp inbound decode submit failed account=%s seq=%d err=%v", account, req.SeqId, err)
+		resp.Result = 9
+		return false, nil
+	}
+	result, err := s.submit(packet.Conn.Conn.RemoteAddr(), submitRequest{
+		Account:     account,
+		PhoneNumber: phone,
+		Content:     content,
+		SrcID:       req.SrcId,
+		DestID:      phone,
+		SequenceID:  req.SeqId,
+		RemoteIP:    remoteIP(packet.Conn.Conn.RemoteAddr()),
+	})
+	if err != nil || !result.Accepted {
+		logger.Printf("cmpp inbound submit rejected account=%s phone=%s seq=%d err=%v", account, phone, req.SeqId, err)
+		resp.Result = 9
+		return false, nil
+	}
+	resp.MsgId = messageIDFrom(result.MessageID, req.SeqId)
+	resp.Result = 0
+	return false, nil
+}
+
+func (s Server) authenticate(remote net.Addr, account string, authSource string, timestamp uint32) (authResponse, error) {
+	payload := authRequest{
+		Account:    account,
+		AuthSource: base64.StdEncoding.EncodeToString([]byte(authSource)),
+		Timestamp:  timestamp,
+		RemoteIP:   remoteIP(remote),
+	}
+	var result authResponse
+	err := s.post(context.Background(), "/gateway/events/inbound/authenticate", payload, &result)
+	return result, err
+}
+
+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 (s Server) post(ctx context.Context, path string, payload any, result any) error {
+	client := s.HTTPClient
+	if client == nil {
+		client = &http.Client{Timeout: defaultHTTPTimeout}
+	}
+	body, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(apiBaseURL(s.APIBaseURL), "/")+path, bytes.NewReader(body))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return fmt.Errorf("api returned %s", resp.Status)
+	}
+	if result != nil {
+		return json.NewDecoder(resp.Body).Decode(result)
+	}
+	return nil
+}
+
+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 apiBaseURL(value string) string {
+	if value == "" {
+		return "http://127.0.0.1:3000/api"
+	}
+	return value
+}
+
+func remoteIP(addr net.Addr) string {
+	if tcp, ok := addr.(*net.TCPAddr); ok {
+		return tcp.IP.String()
+	}
+	host, _, err := net.SplitHostPort(addr.String())
+	if err == nil {
+		return host
+	}
+	return addr.String()
+}
+
+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
+}
@@ -0,0 +1,114 @@
+package inbound
+
+import (
+	"encoding/json"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	cmpp "github.com/bigwhite/gocmpp"
+	cmpputils "github.com/bigwhite/gocmpp/utils"
+)
+
+func TestInboundServerAuthenticatesAndSubmits(t *testing.T) {
+	account := "100001"
+	password := "secret-hash"
+	var gotAuth authRequest
+	var gotSubmit submitRequest
+	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case "/api/gateway/events/inbound/authenticate":
+			if err := json.NewDecoder(r.Body).Decode(&gotAuth); err != nil {
+				t.Fatalf("decode auth: %v", err)
+			}
+			_ = json.NewEncoder(w).Encode(authResponse{PasswordCipher: password})
+		case "/api/gateway/events/inbound/submit":
+			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-1"})
+		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.V30)
+	defer client.Disconnect()
+	if err := client.Connect(addr, account, password, 2*time.Second); err != nil {
+		t.Fatalf("connect inbound cmpp: %v", err)
+	}
+
+	content, err := cmpputils.Utf8ToUcs2("测试入站")
+	if err != nil {
+		t.Fatalf("encode content: %v", err)
+	}
+	_, err = client.SendReqPkt(&cmpp.Cmpp3SubmitReqPkt{
+		PkTotal:            1,
+		PkNumber:           1,
+		RegisteredDelivery: 1,
+		MsgLevel:           1,
+		ServiceId:          "cmpp",
+		FeeUserType:        2,
+		FeeTerminalId:      "13500002696",
+		MsgFmt:             8,
+		MsgSrc:             account,
+		FeeType:            "02",
+		FeeCode:            "0",
+		SrcId:              "10690000",
+		DestUsrTl:          1,
+		DestTerminalId:     []string{"13500002696"},
+		MsgLength:          uint8(len(content)),
+		MsgContent:         content,
+	})
+	if err != nil {
+		t.Fatalf("send submit: %v", err)
+	}
+	rsp := recvSubmitRsp(t, client)
+	if rsp.Result != 0 || rsp.MsgId == 0 {
+		t.Fatalf("unexpected submit response: %+v", rsp)
+	}
+	if gotAuth.Account != account || gotAuth.AuthSource == "" || gotAuth.RemoteIP == "" {
+		t.Fatalf("unexpected auth payload: %+v", gotAuth)
+	}
+	if gotSubmit.Account != account || gotSubmit.PhoneNumber != "13500002696" || gotSubmit.Content != "测试入站" {
+		t.Fatalf("unexpected submit payload: %+v", gotSubmit)
+	}
+}
+
+func reserveTCPAddr(t *testing.T) string {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("reserve tcp addr: %v", err)
+	}
+	addr := listener.Addr().String()
+	if err := listener.Close(); err != nil {
+		t.Fatalf("close reserved listener: %v", err)
+	}
+	return addr
+}
+
+func recvSubmitRsp(t *testing.T, client *cmpp.Client) *cmpp.Cmpp3SubmitRspPkt {
+	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.Cmpp3SubmitRspPkt); ok {
+			return rsp
+		}
+	}
+	t.Fatal("timed out waiting submit response")
+	return nil
+}