简介这是一份SGIP短信网关互操作协议的Python实现源码包面向通信协议开发者、短信网关研究者以及希望在非电信环境中模拟或测试短信服务的Python工程师。资源围绕SGIP协议中MT向手机发送与MO手机发起两类消息展示了TCP/IP网络通信、报文固定头/可变头/数据体解析与构造、请求响应处理等核心环节的实现思路。压缩包共22个文件大小约27KB以7个Python脚本为主覆盖服务端程序、注册服务、网关库、协议常量及调试测试模块另含SVN元数据和工程配置文件便于追溯代码版本。已有295人学习参考。通过学习这些源码读者可快速搭建SGIP模拟服务梳理MT提交/状态报告和MO接收/鉴权的处理流程掌握用socket与struct库完成二进制报文编解码的实践方法也能为短信接口开发或协议测试提供可复用的代码基础。1. SGIP Python 版本短消息网关协议接入的另一种打开方式做运营商短消息网关接入的开发者大概率都被 SGIP 协议折腾过。SGIP 是短消息网关接口协议国内不少省市的行业短消息接入都在用它早期很多方案是基于 C 或 Java 写的Python 版本反而像是个“非主流”选择。但实际做起来你会发现Python 版 SGIP 在联调效率、协议报文调试、快速上线这类场景里优势非常明显——不用编译、改完就能跑、抓包看协议字段也直观。这篇文章适合三类人一是刚接手 SGIP 网关接入、被协议文档绕晕的新手二是想把老项目从其他语言迁移到 Python 的团队三是需要在测试环境快速模拟网关交互的运维或测试同学。接下来说清楚 SGIP 到底是什么、Python 实现的核心模块怎么拆、代码怎么写、参数怎么调以及那些文档里不会写的血泪坑。2. SGIP 协议结构拆解先搞懂报文再写代码2.1 SGIP 在短消息产业链里的位置SGIP 全称 Short Message Gateway Interface Protocol是中国移动定义的短消息网关与业务系统之间的接口协议。一句话概括它的角色你的业务系统比如验证码平台、通知类短信平台通过 SGIP 协议连接到运营商短消息网关网关再负责把消息送到手机用户。这个协议的特点是二进制报文、以 TCP 长连接为基础、消息头固定 12 字节、消息体按消息类型各自定义。和 SMPP、CMPP 相比SGIP 在国内部分省市的行业短信接入中仍然存在。SGIP 的报文封装不像 HTTP 那样有直观的状态码和 JSON 体而是纯字节流字段按偏移量对齐。新手最容易懵的就是这一点看不到明文全是十六进制。所以 Python 做 SGIP 有个天然优势——struct 模块解包、hex 打印调试都方便写工具脚本也比其他语言快。提示SGIP 消息类型里Submit发送短消息、Deliver接收状态报告或上行消息、Bind连接认证是三个最常用的命令先把这三个报文结构吃透其他类型基本都是同一套封装思路。2.2 消息头 12 字节的字段含义与字节序SGIP 每个消息包都以 12 字节的消息头开始字段如下字段名长度字节说明Request ID4消息类型如 1 表示 Bind、2 表示 Bind Resp、3 表示 Submit、4 表示 Submit RespSequence Number816 进制序列号由节点编号 月份 秒数 序号拼接而成注意字节序问题。SGIP 的整数字段默认按网络字节序Big-Endian传输但有些网关节点的实现会在 Sequence Number 的拼接上做二次编码。实际写代码时我建议用struct.pack(!I, ...)和struct.unpack(!I, ...)来处理头部字段避免大小端混乱导致的对端解析失败。Sequence Number 的拼接规则是节点编号3 字节 月份1 字节 秒数4 字节其中秒数是从当日 00:00:00 起算的秒数。这个字段是网关用来去重和关联响应包的同一个 TCP 连接上你发出去的 Submit 和收到的 Submit Resp 必须能通过这个序列号对上。SGIP 消息头定义Python 版用 Python 的struct模块来定义和解析消息头这块代码是后续所有消息处理的基础。做的时候需要区分消息头的字节序以及消息体长度字段是否需要额外校验。2.3 Bind 与 Submit 的报文体字段对照Bind 请求体是连接认证的关键包含客户端编号、认证密码、登录类型等字段。Submit 请求体则是真正发送短消息的部分字段比 Bind 多很多包括 SP 编号、计费类型、用户号码、消息内容等。写代码前先把字段表列清楚。Submit 报文体里Message Coding 字段决定消息内容是 ASCII 还是 UCS2 编码这个字段经常被忽略。很多省份网关要求中文消息必须用 UCS2 编码并且消息内容长度不能超过 62 个字节约 31 个汉字如果你的消息体超长网关会直接返回错误响应而不是帮你截断。3. Python 版 SGIP 消息封装核心类设计与代码实现3.1 消息基类与消息头序列化的最小实现先写一个消息基类负责消息头的打包和解包。这个基类不关心具体业务字段只处理 12 字节头。这样后续 Bind、Submit、Deliver 都继承它各自实现自己的消息体打包逻辑。import struct import time class SGIPMessage: def __init__(self, request_id): self.request_id request_id self.sequence_number self._build_sequence_number() def _build_sequence_number(self): # 节点编号这里用 1 表示本地节点实际接入时由网关分配 node_id 1 # 月份1-12 month time.localtime().tm_mon # 秒数从当日 00:00:00 起算 seconds int(time.time()) % 86400 # SGIP 序列号拼接节点编号(3字节) 月份(1字节) 秒数(4字节) return (node_id 40) | (month 32) | seconds def pack_header(self): # 消息头固定 12 字节Big-Endian 编码 return struct.pack(!II, self.request_id, self.sequence_number) def pack(self): raise NotImplementedError classmethod def unpack_header(cls, data): request_id, sequence_number struct.unpack(!II, data[:12]) return request_id, sequence_number这段代码的核心在_build_sequence_number。按协议规定序列号是 8 字节64 位这里用位运算把它拼成一个整数节点编号占 3 字节月份占 1 字节秒数占 4 字节。注意秒数用的是本地时区如果你的服务器时区和网关不一致可能导致序列号异常不过大部分网关对时区不敏感真正校验的是唯一性。pack_header里用!II表示两个无符号整数网络字节序。这里有个容易犯的错不能用struct.pack(II)不加字节序前缀因为默认是本机字节序x86 是小端拼出来的头在网关侧解析会直接翻车。3.2 Submit 消息封装字段填充与 UCS2 编码Submit 消息体是发送短消息最核心的部分。这里给出一个精简但能跑的版本class SGIPSubmit(SGIPMessage): def __init__(self, sp_id, user_number, content, msg_type1): super().__init__(request_id3) # 3 表示 Submit self.sp_id sp_id self.user_number user_number self.content content self.msg_type msg_type # 默认中文用 UCS2 编码 self.msg_coding 15 # 15 表示 UCS20 表示 ASCII def _encode_content(self): if self.msg_coding 15: # UCS2 编码每个汉字 2 字节 encoded self.content.encode(utf-16-be) return encoded, len(encoded) // 2 else: encoded self.content.encode(ascii) return encoded, len(encoded) def pack(self): body struct.pack(!I, 1) # 消息长度占位后面回填 # 这里简化处理实际需要按 SGIP 文档填写所有字段 return self.pack_header() body这个版本刻意做了简化因为完整字段太多。实际项目中Submit 消息体至少包括 SP 编号、计费类型、计费号码、用户号码、消息编码、消息长度、消息内容等二十多个字段。写代码时我习惯把字段分成固定长度字符串和整数两类固定长度字符串用bytes填充整数用struct.pack。消息编码是重点。msg_coding15对应 UCS2内容编码用utf-16-be因为 SGIP 文档里明确中文短消息采用 UCS2 编码且字节序是大端。这里容易踩的坑是用 UTF-8 编码中文网关收到后解析出来全是乱码因为网关按 UCS2 解码。另外消息内容的长度字段按“字节数”算UCS2 编码下一个汉字占 2 字节所以 31 个汉字就是 62 字节正好是单条短信上限。3.3 连接管理与 Bind 认证流程SGIP 的 TCP 连接不是建立就能收发消息的必须先发 Bind 请求等网关返回 Bind Resp成功后才能 Submit。这个流程和 FTP 协议有点像先认证再干正事。import socket import struct class SGIPClient: def __init__(self, host, port, sp_id, shared_secret): self.host host self.port port self.sp_id sp_id self.shared_secret shared_secret self.sock None def connect(self): self.sock socket.create_connection((self.host, self.port), timeout10) def bind(self): # Bind 请求体客户端编号 认证密码 登录类型 # 认证密码 MD5(SP编号 共享密钥) 的前 16 位具体按各省网关文档 import hashlib md5 hashlib.md5() md5.update((self.sp_id self.shared_secret).encode(utf-8)) auth_password md5.hexdigest()[:16] # 固定部分客户端编号 6 字节认证密码 16 字节登录类型 1 字节 body self.sp_id.encode(utf-8).ljust(6, b\x00) body auth_password.encode(utf-8) body struct.pack(!B, 1) # 1 表示 SP 接入 # 组装并发送 Bind 包 msg SGIPMessage(request_id1) # 1 表示 Bind packet msg.pack_header() body self.sock.send(packet) # 等待 Bind Resp resp_header self.sock.recv(12) request_id, seq SGIPMessage.unpack_header(resp_header) resp_body self.sock.recv(16) # Bind Resp 的消息体固定 16 字节 status struct.unpack(!I, resp_body[:4])[0] if status ! 0: raise Exception(fBind 失败状态码: {status}) def submit(self, user_number, content): # 封装 Submit 并发送 submit SGIPSubmit(self.sp_id, user_number, content) packet submit.pack() self.sock.send(packet)Bind 认证的关键在auth_password的计算。不同省份网关对这个密码的算法有细微差别多数是用 MD5 摘要后取前 16 位但有的会要求先做一次 Hex 转换有的会把共享密钥当作二进制而不是字符串参与计算。我的建议是第一版代码先按最通用的方式写联调时如果收到认证失败的状态码优先怀疑密码算法而不是 TCP 连接。登录类型login_type1表示 SP 接入这个值一般固定不动。如果你对接的是多个 SP 共用的网关需要注意网关侧为你的 SP 编号分配的权限登录类型不对也可能被拒。连接管理上SGIP 长连接需要有心跳保活机制不过 SGIP 协议本身没有定义独立的心跳包常见做法是定时发送一个空消息或者依赖 TCP keepalive。这个在后面“避坑章节”里会细说。4. 收发异步处理从 Submit 到 Submit Resp 的超时与重发机制4.1 同步发送与响应关联的技术方案SGIP 是异步协议你发 Submit 后网关不一定马上回 Submit Resp。有些网关因为内部路由慢几秒后才回有的场景下网关先回 Submit Resp表示已接收之后才通过 Deliver 消息返回真正的状态报告。所以客户端代码不能写成“发了就等响应”那会把连接堵死。我一般会在客户端里维护一个待确认消息字典键是 Sequence Number值是发送时间戳和消息内容。收到 Submit Resp 时用响应包里的序列号去字典里查找到就把这条标记为已确认同时计算发送到确认的耗时。如果超过设定时间比如 30 秒还没收到 Submit Resp就要考虑重发。class SGIPAsyncClient(SGIPClient): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.pending {} # 序列号 - (发送时间, 消息内容) self.timeout 30 # 超时时间单位秒 def submit_with_seq(self, user_number, content): submit SGIPSubmit(self.sp_id, user_number, content) packet submit.pack() # 从消息头里提取序列号 seq submit.sequence_number self.pending[seq] (time.time(), content) self.sock.send(packet) return seq def handle_response(self, data): # 解析响应包 request_id, seq SGIPMessage.unpack_header(data[:12]) if request_id 4: # Submit Resp self.pending.pop(seq, None) return True return Falsepending字典是这块的核心。注意 SGIP 的响应消息类型是 4Submit Resp而请求是 3Submit。你在处理响应时不能看到request_id 3就认为是请求要看网关返回的这个字段。每个消息包的序列号都不同所以字典的键不会冲突。超时重发的策略上我吃过亏。刚开始超时了就直接重发结果网关那边其实已经收到了第一条 Submit只是因为链路拥塞导致 Resp 迟迟没回来。重发之后网关又把两条消息都发给用户用户收到两条一模一样的验证码体验非常差。后来改成第一次超时后先发一个查询请求SGIP 有 Trace 类型的消息可以查状态但不少网关不支持查不到再重发。如果网关不支持 Trace那就只能降低重发频率并且把重发次数限制在 2 次以内。4.2 接收 Deliver 消息的解析与状态报告处理Deliver 消息是网关主动推给你的有两种一种是用户上行短信一种是短消息状态报告。区分它们靠消息体里的字段标识。状态报告里携带着你之前 Submit 消息的最终状态成功、失败、超时等。这个必须做异步接收处理因为它的到达时间完全不可控。def receive_loop(self): while True: try: header self.sock.recv(12) if not header: break request_id, seq SGIPMessage.unpack_header(header) if request_id 5: # 5 表示 Deliver # 先读消息体长度再读完整包 body self._read_full_body(seq) self._handle_deliver(body) except socket.timeout: continue except Exception as e: # 异常情况重新连接 self._reconnect()接收循环要放在独立线程里跑不能和发送主逻辑放在同一个线程。收到 Deliver 后需要回一个 Deliver Resp 给网关告诉它“我收到了”。如果不回网关会按超时重发 Deliver造成重复处理。Deliver 状态报告里最关键的是state字段它表示消息最终状态。我在生产环境里见过最多的是Submit Resp 返回成功但 Deliver 状态报告显示失败。原因可能是网关内部路由出错、用户手机号处于异常状态等。这块数据一定要记录下来最好写进数据库方便后续对账和排查投诉。4.3 TCP 断线重连与序列号连续性长连接最怕断线。运营商网关的 TCP 连接可能因为网络抖动、网关重启等原因断开。断开后怎么处理是有讲究的。不要立刻重连先等待几秒避免网关还没恢复就反复触发连接。重连后要重新 BindBind 成功后之前 pending 的消息怎么处理我的习惯是断线重连后pending 里的消息全部按失败处理由上层业务决定是否重发。为什么这么做因为 TCP 断开意味着你无法确认网关是否收到了那些消息盲目重发可能造成重复短信不重发又可能漏发。交给上层业务结合业务幂等性来决定是最稳妥的。序列号连续性上要注意重连后序列号继续往上走不要重置为 0。如果网关侧对序列号有“必须连续”的校验部分网关有你重置序列号会导致 Bind 成功后所有消息被拒收。5. SGIP Python 版本落地避坑五个高频翻车场景与排查方法5.1 现象Bind 一直返回状态码非 0证书和密码看着都对原因有两个方向。第一密码算法不对。SGIP 文档里写的是“MD5(SP编号 共享密钥)”但有的网关实现是“MD5(共享密钥 SP编号)”顺序反了。第二密码的摘要长度。文档写的是取 16 位但有些网关取 32 位完整摘要。解决先用网关提供的测试工具发一包 Bind把那包的字节流抓下来对比你代码里发的字节流。重点看第 12 到第 28 字节认证密码字段长度和内容是否一致。如果手头没有测试工具就在本地写个脚本把几种密码算法都跑一遍打印出不同算法的字节流写死在代码里逐个试。5.2 现象Submit 消息发出后网关返回“消息长度错误”原因Submit 消息体的长度字段填错了。SGIP 消息头里没有独立的长度字段长度信息是包含在 Submit 消息体内部的。如果内容编码用了 UTF-8 但长度按字符数算必然出错。解决长度务必按字节数算。UCS2 编码下len(content.encode(utf-16-be))才是真实字节长度一个汉字是 2 字节。ASCII 环境下中文会编码失败要先判断消息内容是否含非 ASCII 字符动态切换编码和长度计算。5.3 现象收到重复的状态报告用户投诉收到两条相同短信原因客户端没有正确处理超时重发和 Deliver 去重。消息实际发送成功但 Submit Resp 丢失或延迟导致客户端超时重发网关收到两条 Submit自然产生两条状态报告。解决用业务流水号去重。SGIP 协议里 Submit 消息体中有一个字段是业务编号或用户自定义编号发送同一业务流水号时网关侧如果支持去重就不会重复下发。但如果网关不支持客户端必须在重发前先查 pending确认上次是否已超时。更严格的做法是所有消息统一走业务数据库先查重再发送。5.4 现象TCP 连接一直在但消息越收越少最后卡死原因TCP 接收缓冲区里粘包了。SGIP 没有一个全包长度的头部消息体长度字段在不同消息类型里位置不一样。如果你按固定长度recv可能一次收到两条消息或者一条消息分两次收到。解决写一个拆包函数边接收边解析。根据消息头里的 request_id 确定这个消息体的固定长度不够就继续读。这个函数是 SGIP 客户端里最容易写错的地方但也是必须写对的地方。我建议用recv循环读每次读到的数据先放入缓冲区然后尝试解析缓冲区里的完整消息解析完一条就把它从缓冲区移除。5.5 现象网关侧 IP 白名单配置正确但连接总是被复位原因有些网关要求客户端先发 Bind再允许其他请求。如果你连接建立后直接发 Submit网关会认为你非法直接断开。另外连接的源端口如果频繁变化也可能让网关认为异常。解决严格按“连接 - Bind - Bind Resp - Submit - Submit Resp - Deliver - Deliver Resp”的顺序来。整个交互流程写成一个状态机不要跳步骤。对于源端口问题可以在客户端绑定固定端口或者让网络侧同事检查防火墙是否对高位端口有限制。6. 一个能上生产环境的 SGIP Python 客户端工程化改造清单与验证技巧把demo变成生产代码不只是改改 bug需要把异常处理、日志和配置管理做扎实。我给出的工程化清单有三部分。第一配置外置。网关 IP、端口、SP 编号、共享密钥、超时时间、重发次数全部放到配置文件里不要硬编码。SGIP 接入测试环境和生产环境的参数往往只是 IP 和共享密钥不同代码却要一致。用环境变量或配置文件切换别在代码里写两份。第二日志要结构化。每次收发消息、绑定的成功失败、状态报告的到达都要记录。日志里不要只记录“发送成功”这种话要记录下消息的序列号、SP 编号、用户号码、状态码。出了问题靠日志能还原整条链路的交互过程。血泪经验是没有日志没有排查入口网关侧反馈说没收到消息你却拿不出本地日志这个亏我吃过不止一次。第三监控指标。给客户端暴露几个核心指标连接状态0 断开1 已连接2 已绑定、pending 队列长度、单位时间发消息数、Submit Resp 平均时延、状态报告失败率。这些指标可以通过 HTTP 接口暴露也可以打进日志系统。pending 队列长度特别重要它如果持续增长说明网关处理不过来或者连接已经半开需要告警。配合验证时写一个简单的自测脚本模拟网关端返回固定响应验证客户端的拆包和响应关联逻辑。没有真实网关时这个脚本就是最好的调试工具。def smoke_test(): # 自测本地起一个模拟网关验证消息头序列化和拆包 mock_socket socket.socket() mock_socket.bind((127.0.0.1, 8899)) # 收到 Bind 后回一个成功的 Bind Resp # 收到 Submit 后回一个 Submit Resp pass自测脚本不需要把协议体全部模拟出来但要能验证两点消息头的字节序对不对超时重发逻辑会不会死循环。我跑过很多次烟测最容易暴露的问题就是模拟网关发来的响应和真实网关字段顺序不同导致客户端解析错位。最后说一个关键习惯对接任何新网关前先用 Python 交互式环境手动发一包 Bind把十六进制字节流打印出来跟网关文档上的示例报文逐字节比对。这个习惯帮我拦截了至少三次字段偏移错误。做 SGIP 这东西玄学问题特别多但绝大多数都能靠逐字节比对解决。希望这篇能帮你在接入 SGIP 的路上少踩几个坑顺利的话一天内就能从零跑到通联调。祝你上线顺利。本文还有配套的精品资源点击获取