diff --git a/api/prisma/migrations/20260707162000_add_application_cmpp_account/migration.sql b/api/prisma/migrations/20260707162000_add_application_cmpp_account/migration.sql new file mode 100644 index 0000000..42eab0e --- /dev/null +++ b/api/prisma/migrations/20260707162000_add_application_cmpp_account/migration.sql @@ -0,0 +1,13 @@ +ALTER TABLE "SmsApplication" ADD COLUMN "cmppAccount" TEXT; + +WITH numbered AS ( + SELECT id, row_number() OVER (ORDER BY "createdAt", id) AS rn + FROM "SmsApplication" +) +UPDATE "SmsApplication" app +SET "cmppAccount" = lpad(((100000 + numbered.rn) % 1000000)::text, 6, '0') +FROM numbered +WHERE app.id = numbered.id; + +ALTER TABLE "SmsApplication" ALTER COLUMN "cmppAccount" SET NOT NULL; +CREATE UNIQUE INDEX "SmsApplication_cmppAccount_key" ON "SmsApplication"("cmppAccount"); diff --git a/api/prisma/schema.prisma b/api/prisma/schema.prisma index 14fa4fb..13e7541 100644 --- a/api/prisma/schema.prisma +++ b/api/prisma/schema.prisma @@ -336,6 +336,7 @@ model SmsApplication { name String scene String? callbackUrl String? + cmppAccount String @unique secretHash String dailyLimit Int? customerUnitPrice Int @default(0) diff --git a/api/src/send-chain/gateway-events.controller.ts b/api/src/send-chain/gateway-events.controller.ts index b20fec4..3eea273 100644 --- a/api/src/send-chain/gateway-events.controller.ts +++ b/api/src/send-chain/gateway-events.controller.ts @@ -1,6 +1,8 @@ import { Body, Controller, Post } from '@nestjs/common'; import { ApiTags } from '@nestjs/swagger'; import { + GatewayInboundAuthDto, + GatewayInboundSubmitDto, GatewayReceiptEventDto, GatewaySubmitResultDto, GatewayUplinkEventDto, @@ -26,5 +28,14 @@ export class GatewayEventsController { uplink(@Body() body: GatewayUplinkEventDto) { return this.sendChain.handleUplink(body); } -} + @Post('inbound/authenticate') + authenticateInbound(@Body() body: GatewayInboundAuthDto) { + return this.sendChain.authenticateInboundApplication(body); + } + + @Post('inbound/submit') + submitInbound(@Body() body: GatewayInboundSubmitDto) { + return this.sendChain.submitInboundMessage(body); + } +} diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index fc88652..3e97559 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -59,7 +59,17 @@ function createPrismaMock() { findUnique: jest.fn().mockResolvedValue({ id: 'tenant-1', status: 'active', certificationStatus: 'approved' }), }, smsApplication: { - findUnique: jest.fn().mockResolvedValue({ id: 'app-1', tenantId: 'tenant-1', status: 'active', customerUnitPrice: 3, queuePriority: 'normal' }), + findUnique: jest.fn().mockResolvedValue({ id: 'app-1', tenantId: 'tenant-1', cmppAccount: '100001', status: 'active', customerUnitPrice: 3, queuePriority: 'normal' }), + findFirst: jest.fn().mockResolvedValue({ + id: 'app-1', + tenantId: 'tenant-1', + cmppAccount: '100001', + secretHash: 'secret-hash', + status: 'active', + queuePriority: 'normal', + ipAllowlist: [{ ipCidr: '127.0.0.1/32' }], + tenant: { id: 'tenant-1', status: 'active', certificationStatus: 'approved' }, + }), }, smsTemplate: { findUnique: jest.fn().mockResolvedValue({ @@ -69,6 +79,14 @@ function createPrismaMock() { auditStatus: 'approved', signature: { auditStatus: 'approved', reportStatus: 'approved' }, }), + findFirst: jest.fn().mockResolvedValue({ + id: 'tpl-1', + tenantId: 'tenant-1', + applicationId: 'app-1', + content: 'hello', + auditStatus: 'approved', + signature: { auditStatus: 'approved', reportStatus: 'approved' }, + }), }, smsBatchTask: { create: jest.fn().mockResolvedValue(task), diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index 49b3966..6dba57f 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -2,6 +2,8 @@ import { BadRequestException, Injectable, NotFoundException, OnModuleDestroy, On import { Queue, Worker } from 'bullmq'; import IORedis from 'ioredis'; import { randomUUID } from 'node:crypto'; +import { createHash } from 'node:crypto'; +import { isIP } from 'node:net'; import { setTimeout as sleep } from 'node:timers/promises'; import { BillingService } from '../billing/billing.service'; import { PrismaService } from '../prisma/prisma.service'; @@ -20,6 +22,25 @@ export interface CreateBatchTaskDto { createdById?: string; sourceIp?: string; userAgent?: string; + sourceType?: 'client' | 'api' | 'cmpp'; +} + +export interface GatewayInboundAuthDto { + account: string; + password?: string; + authSource?: string; + timestamp?: number; + remoteIp?: string; +} + +export interface GatewayInboundSubmitDto { + account: string; + phoneNumber: string; + content: string; + srcId?: string; + destId?: string; + sequenceId?: number; + remoteIp?: string; } export interface GatewaySubmitResultDto { @@ -176,7 +197,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { applicationId: data.applicationId, templateId: data.templateId, taskNo: `BT-${Date.now()}-${randomUUID().slice(0, 8)}`, - sourceType: 'client', + sourceType: data.sourceType ?? 'client', content: data.content, category: data.category, phoneTotal: phones.length, @@ -628,6 +649,63 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { }); } + async authenticateInboundApplication(data: GatewayInboundAuthDto) { + const application = await this.findInboundApplication(data.account); + if (!application || application.status !== 'active' || application.tenant.status !== 'active') { + throw new BadRequestException('CMPP account is invalid or disabled'); + } + if (application.tenant.certificationStatus !== 'approved') { + throw new BadRequestException('Enterprise certification is not approved'); + } + if (!matchesApplicationSecret(data, application.secretHash)) { + throw new BadRequestException('CMPP account or password is invalid'); + } + if (data.remoteIp && !isApplicationIpAllowed(data.remoteIp, application.ipAllowlist.map((item) => item.ipCidr))) { + throw new BadRequestException('CMPP source IP is not in application allowlist'); + } + return { + applicationId: application.id, + tenantId: application.tenantId, + account: application.cmppAccount, + passwordCipher: application.secretHash, + status: 'authenticated', + }; + } + + async submitInboundMessage(data: GatewayInboundSubmitDto) { + const application = await this.findInboundApplication(data.account); + if (!application || application.status !== 'active' || application.tenant.status !== 'active') { + throw new BadRequestException('CMPP account is invalid or disabled'); + } + if (data.remoteIp && !isApplicationIpAllowed(data.remoteIp, application.ipAllowlist.map((item) => item.ipCidr))) { + throw new BadRequestException('CMPP source IP is not in application allowlist'); + } + if (!/^1[3-9]\d{9}$/.test(data.phoneNumber)) { + throw new BadRequestException('CMPP submit phone number is invalid'); + } + const template = await this.resolveInboundTemplate(application.id, data.content); + const task = await this.createBatchTask({ + tenantId: application.tenantId, + applicationId: application.id, + templateId: template.id, + content: data.content, + phones: [data.phoneNumber], + sourceType: 'cmpp', + sourceIp: data.remoteIp, + userAgent: 'cmpp-gateway', + }); + const message = task?.messages?.[0]; + return { + accepted: true, + tenantId: application.tenantId, + applicationId: application.id, + taskId: task?.id, + messageId: message?.messageId, + messageRecordId: message?.id, + status: message?.status ?? task?.status, + }; + } + async markUnknownTimeout(data: TimeoutUnknownDto) { const olderThanHours = data.olderThanHours ?? 72; const cutoff = new Date(Date.now() - olderThanHours * 60 * 60 * 1000); @@ -911,6 +989,36 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { return normalizeQueuePriority(application.queuePriority); } + private findInboundApplication(account: string) { + return this.prisma.smsApplication.findFirst({ + where: { cmppAccount: account }, + include: { + tenant: true, + ipAllowlist: true, + }, + }); + } + + private async resolveInboundTemplate(applicationId: string, content: string) { + const template = await this.prisma.smsTemplate.findFirst({ + where: { + applicationId, + content, + auditStatus: 'approved', + signature: { + auditStatus: 'approved', + reportStatus: 'approved', + }, + }, + include: { signature: true }, + orderBy: { updatedAt: 'desc' }, + }); + if (!template) { + throw new BadRequestException('CMPP submit content does not match an approved template and signature'); + } + return template; + } + private async validateSendResources(tenantId: string, applicationId?: string, templateId?: string) { const tenant = await this.prisma.tenant.findUnique({ where: { id: tenantId } }); if (!tenant || tenant.status !== 'active') { @@ -1270,3 +1378,64 @@ function bullmqConnection() { maxRetriesPerRequest: null, }; } + +function matchesApplicationSecret(data: GatewayInboundAuthDto, secretHash: string) { + if (data.authSource && data.timestamp !== undefined) { + const expected = createHash('md5') + .update(Buffer.concat([ + Buffer.from(octetString(data.account, 6), 'binary'), + Buffer.alloc(9), + Buffer.from(secretHash), + Buffer.from(String(data.timestamp).padStart(10, '0')), + ])) + .digest('base64'); + return expected === data.authSource; + } + if (!data.password) { + return false; + } + return data.password === secretHash || createHash('sha256').update(data.password).digest('hex') === secretHash; +} + +function octetString(value: string, fixedLength: number) { + if (value.length === fixedLength) { + return value; + } + if (value.length > fixedLength) { + return value.slice(value.length - fixedLength); + } + return value + '\0'.repeat(fixedLength - value.length); +} + +function isApplicationIpAllowed(remoteIp: string, allowlist: string[]) { + const normalizedRemoteIp = normalizeIp(remoteIp); + if (allowlist.length === 0) { + return true; + } + return allowlist.some((rule) => ipMatchesRule(normalizedRemoteIp, rule)); +} + +function ipMatchesRule(remoteIp: string, rule: string) { + const normalizedRule = normalizeIp(rule.trim()); + if (!normalizedRule) { + return false; + } + if (!normalizedRule.includes('/')) { + return remoteIp === normalizedRule; + } + const [network, prefixText] = normalizedRule.split('/'); + const prefix = Number(prefixText); + if (!Number.isInteger(prefix) || prefix < 0 || prefix > 32 || isIP(remoteIp) !== 4 || isIP(network) !== 4) { + return false; + } + const mask = prefix === 0 ? 0 : (0xffffffff << (32 - prefix)) >>> 0; + return (ipv4ToInt(remoteIp) & mask) === (ipv4ToInt(network) & mask); +} + +function normalizeIp(value: string) { + return value.replace(/^::ffff:/, '').trim(); +} + +function ipv4ToInt(value: string) { + return value.split('.').reduce((result, part) => ((result << 8) + Number(part)) >>> 0, 0); +} diff --git a/api/src/sms-config/sms-config.service.spec.ts b/api/src/sms-config/sms-config.service.spec.ts index 5620af8..ad3d011 100644 --- a/api/src/sms-config/sms-config.service.spec.ts +++ b/api/src/sms-config/sms-config.service.spec.ts @@ -8,6 +8,7 @@ function createPrismaMock() { tenantId: 'tenant-1', name: '应用A', status: 'active', + cmppAccount: '100001', queuePriority: 'normal', tenant: { id: 'tenant-1', name: '租户A', code: 'TENANT-A' }, messageRecords: [{ status: 'delivered' }, { status: 'undelivered' }], @@ -17,6 +18,7 @@ function createPrismaMock() { tenantId: 'tenant-1', name: '应用A', status: 'active', + cmppAccount: '100001', queuePriority: 'normal', secretHash: 'secret-hash', tenant: { id: 'tenant-1', name: '租户A', code: 'TENANT-A' }, @@ -164,6 +166,7 @@ describe('SmsConfigService', () => { await expect(service.getApplicationCmppParams('app-1')).resolves.toEqual(expect.objectContaining({ applicationId: 'app-1', tenantName: '租户A', + account: '100001', gatewayHost: '127.0.0.1', gatewayPort: 17890, maxConnections: 2, @@ -172,6 +175,7 @@ describe('SmsConfigService', () => { it('creates enterprise applications with persisted queue priority', async () => { const prisma = createPrismaMock(); + prisma.smsApplication.findUnique.mockResolvedValueOnce(null); const service = new SmsConfigService(prisma as never); await expect(service.createApplication({ @@ -185,6 +189,7 @@ describe('SmsConfigService', () => { data: expect.objectContaining({ tenantId: 'tenant-1', name: '优先应用', + cmppAccount: expect.stringMatching(/^\d{6}$/), queuePriority: 'priority', ipAllowlist: { create: [{ ipCidr: '10.0.0.1/32' }] }, }), @@ -195,11 +200,11 @@ describe('SmsConfigService', () => { const prisma = createPrismaMock(); const service = new SmsConfigService(prisma as never); - expect(() => service.createApplication({ + await expect(service.createApplication({ tenantId: 'tenant-1', name: '异常应用', queuePriority: 'urgent', - })).toThrow('queuePriority must be normal or priority'); + })).rejects.toThrow('queuePriority must be normal or priority'); expect(prisma.smsApplication.create).not.toHaveBeenCalled(); }); diff --git a/api/src/sms-config/sms-config.service.ts b/api/src/sms-config/sms-config.service.ts index ea1664b..566ad78 100644 --- a/api/src/sms-config/sms-config.service.ts +++ b/api/src/sms-config/sms-config.service.ts @@ -1,6 +1,6 @@ import { BadRequestException, Injectable, NotFoundException } from '@nestjs/common'; import { Prisma } from '@prisma/client'; -import { randomBytes, createHash } from 'node:crypto'; +import { randomBytes, randomInt, createHash } from 'node:crypto'; import { PrismaService } from '../prisma/prisma.service'; export interface CreateSmsApplicationDto { @@ -150,15 +150,17 @@ export class SmsConfigService { return application; } - createApplication(data: CreateSmsApplicationDto) { + async createApplication(data: CreateSmsApplicationDto) { const secret = randomBytes(24).toString('hex'); const queuePriority = normalizeApplicationQueuePriority(data.queuePriority); + const cmppAccount = await this.generateCmppAccount(); return this.prisma.smsApplication.create({ data: { tenantId: data.tenantId, name: data.name, scene: data.scene, callbackUrl: data.callbackUrl, + cmppAccount, secretHash: hashSecret(secret), dailyLimit: data.dailyLimit, customerUnitPrice: data.customerUnitPrice ?? 0, @@ -345,8 +347,8 @@ export class SmsConfigService { gatewayHost: channel?.gatewayHost ?? '', gatewayPort: channel?.gatewayPort ?? 0, enterpriseCode: channel?.enterpriseCode ?? application.tenant.code, - account: channel?.account ?? application.tenant.code, - passwordCipher: channel?.passwordCipher ?? application.secretHash, + account: application.cmppAccount, + passwordCipher: application.secretHash, srcId: channel?.srcId ?? '', maxConnections: channel?.config && typeof channel.config === 'object' && 'maxConnections' in channel.config ? Number(channel.config.maxConnections) : 1, heartbeatSeconds: 30, @@ -355,6 +357,17 @@ export class SmsConfigService { }; } + private async generateCmppAccount() { + for (let attempt = 0; attempt < 20; attempt += 1) { + const cmppAccount = String(randomInt(100000, 1000000)); + const exists = await this.prisma.smsApplication.findUnique({ where: { cmppAccount } }); + if (!exists) { + return cmppAccount; + } + } + throw new BadRequestException('Unable to generate unique CMPP account'); + } + async disconnectApplicationConnection(applicationId: string, connectionId: string, data: StatusChangeDto = { status: 'disconnected' }) { const application = await this.prisma.smsApplication.findUnique({ where: { id: applicationId } }); if (!application) { diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index f6e3285..60cabf3 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -221,17 +221,23 @@ #### 4.8.4 当前实现缺口标记 -截至当前版本，Go Gateway 已有 HTTP 控制服务、健康检查、连接上游 SMSC 的 `ConnectChannel` 控制入口、gocmpp 协议 spike 和队列消息结构；但仍缺少生产验收所需的完整能力： +截至当前版本，Go Gateway 已有 HTTP 控制服务、健康检查、连接上游 SMSC 的 `ConnectChannel` 控制入口、gocmpp 协议 spike、队列消息结构，并已补齐第一阶段下游 CMPP 入站能力： -- 未实现 `17890` 入站 CMPP Server 监听。 -- 未实现下游客户 connect/login 鉴权、IP 白名单、应用级连接数限制和连接状态回写。 -- 未实现下游 CMPP Submit 到平台发送请求的转换。 -- 未实现客户侧 SubmitResp、最终 Deliver Receipt 和上行 Deliver 投递。 +- 已实现 `17890` 入站 CMPP Server 监听，生产部署由 `GATEWAY_CMPP_ADDR=0.0.0.0:17890` 启动。 +- 已实现下游客户 connect/login 鉴权：CMPP `Source_Addr` 使用企业应用独立 6 位 `cmppAccount`，密码使用应用 CMPP 参数中的 `passwordCipher`，Gateway 将 CMPP `AuthSource/Timestamp` 交由 NestJS 根据真实数据库校验。 +- 已实现客户端应用 IP 白名单、应用状态、企业状态和企业认证状态校验；校验失败返回 CMPP connect 失败。 +- 已实现下游 CMPP Submit 到平台发送请求的转换：Gateway 解码 CMPP 3.0 submit 内容，调用 NestJS 真实入站接口，NestJS 复用模板/签名/风控/余额/路由/队列优先级发送链路，接受后返回 CMPP submit_resp。 + +仍缺少生产完整闭环能力： + +- 应用级下游连接数限制和连接状态回写尚未完整产品化。 +- 当前下游 CMPP Submit 通过 `sourceType=cmpp` 的系统批次兼容承载，尚未拆成完全独立于批量任务模型的单条发送模型。 +- 未实现客户侧最终 Deliver Receipt 和上行 Deliver 投递。 - 未实现 Gateway 消费 `SubmitCommand` 并真实 submit 到上游通道的 worker。 - 未实现上游 deliver receipt 和普通上行 deliver 的生产解析与事件回传闭环。 - 未实现多连接窗口管理、在途消息恢复、断线重连后的消息状态处理。 -这些缺口未补齐前，不能把 CMPP 对接发送、17890 端口联调、客户账号密码鉴权、客户 IP 白名单或真实网关 submit/receipt/uplink 作为“生产已验收通过”。 +这些缺口未补齐前，只能把 `17890` 端口监听、客户账号密码鉴权、客户 IP 白名单和下游 submit 入平台发送链路作为第一阶段验收通过；不能把客户侧最终回执投递、上游真实运营商 submit/receipt/uplink 或完整多连接窗口恢复作为“生产已验收通过”。 ### 4.9 回执与上行 diff --git a/docs/production-deployment.md b/docs/production-deployment.md index 665df14..c774879 100644 --- a/docs/production-deployment.md +++ b/docs/production-deployment.md @@ -8,7 +8,7 @@ - PostgreSQL：仅本机 `127.0.0.1:5432`。 - MinIO：仅本机 `127.0.0.1:9000/9001`，页面上传下载通过 API 转发。 - Gateway 控制服务：仅本机 `127.0.0.1:8090`。 -- CMPP 入站端口：`17890` 已作为部署变量保留；当前 Go Gateway 代码尚未实现完整入站 CMPP Server，不要把 HTTP 控制服务误当作 CMPP 监听。 +- CMPP 入站端口：`17890`，由 Go Gateway 启动真实 CMPP 3.0 Server，接收企业应用下游 connect/login 和 submit。 ## 首次部署 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 5427f72..ae33c74 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -978,6 +978,22 @@ - 第二条记录历史或被幂等处理。 - 不重复扣费或重复变更最终状态。 +### TC-GW-006 下游客户 CMPP 17890 入站 bind/submit + +- 优先级：P0 +- 前置条件：企业已认证通过；短信应用 active 且存在独立 6 位 `cmppAccount`；应用 IP 白名单包含测试客户端 IP；应用已有审核和报备通过的签名/模板；通道组、余额和 Gateway 均可用。 +- 步骤： + 1. 启动 Go Gateway，确认 `GATEWAY_CMPP_ADDR=0.0.0.0:17890`。 + 2. 使用 gocmpp 或真实 CMPP 客户端连接 `17890`，`Source_Addr` 填应用 `cmppAccount`，密码填应用 CMPP 参数 `passwordCipher`。 + 3. 发送 CMPP 3.0 SubmitReq，手机号和内容匹配已审核模板。 + 4. 查询 NestJS 数据库和运营端短信记录。 +- 预期结果： + - 17890 是真实 CMPP Server 监听，不是 HTTP 端口。 + - bind 阶段调用真实 NestJS API 校验账号、密码、企业状态、认证状态、应用状态和 IP 白名单。 + - 密码错误、应用停用、企业停用、IP 不在白名单时 connect/login 被拒绝。 + - submit 被接受后返回 CMPP SubmitResp 成功，并在真实数据库创建 `sourceType=cmpp` 的发送记录，进入真实发送链路。 + - submit 内容不匹配审核模板、余额不足、无可用通道时返回明确失败，不得伪造成功。 + ### TC-SEND-021 优先队列插队发送 - 优先级：P0 diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 402ae70..cc12047 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -583,6 +583,29 @@ npm run verify:phase8 - `verify:phase8` 当前失败点是独立 BullMQ 性能阈值，不是本轮通道组、计费、报备、回执业务逻辑测试失败。 - 浏览器端完整手工回归仍建议补跑企业应用创建、通道组配置、短信记录详情弹窗中的历史回执展示。 +## 2026-07-07 Gateway 下游 CMPP 入站第一阶段补齐 + +### 本轮修复 + +- Go Gateway 启动时同时监听 `GATEWAY_CMPP_ADDR`，默认生产端口 `0.0.0.0:17890`，不再只是 HTTP `/health` 控制服务。 +- 新增企业应用独立 6 位 `cmppAccount`，Prisma 迁移 `20260707162000_add_application_cmpp_account` 会为存量应用生成账号；客户端/运营端 CMPP 参数接口返回该应用独立账号。 +- Gateway 下游 CMPP bind 使用真实 gocmpp 协议解析 `Source_Addr/AuthSource/Timestamp`，调用 NestJS `/api/gateway/events/inbound/authenticate`，由真实数据库校验应用账号、应用 CMPP 密码、企业状态、企业认证状态、应用状态和 IP 白名单。 +- Gateway 下游 CMPP submit 解码 CMPP 3.0 `SubmitReq`，调用 NestJS `/api/gateway/events/inbound/submit`；NestJS 按 `sourceType=cmpp` 创建发送记录并复用模板/签名/风控/余额/运营商识别/通道组路由/队列优先级链路。 +- Go Gateway 新增入站集成测试，覆盖本地 CMPP 客户端 connect/login、UCS2 submit 和 API 回调。 + +### 验证状态 + +- `npm --prefix api run prisma:generate`：通过。 +- `npm --prefix api test -- sms-config.service.spec.ts send-chain.service.spec.ts --runInBand`：通过。 +- `npm --prefix api run build`：通过。 +- `go test ./...`（Gateway）：通过。 + +### 剩余缺口 + +- 下游连接状态回写、连接数上限、断开/心跳历史日志仍需继续产品化。 +- 下游 submit 当前通过 `sourceType=cmpp` 的系统批次兼容承载，尚未完全拆成独立单条发送模型。 +- 客户侧最终 Deliver Receipt 投递、客户侧上行 Deliver 推送、上游真实 SMSC submit worker、上游 receipt/uplink 生产解析仍未完成。 + ## 2026-07-03 阶段 9：运营端报备回执导入真实上传/解析 ### 本轮修复 diff --git a/gateway/cmd/gateway/main.go b/gateway/cmd/gateway/main.go index 94b2ad5..a6fcb30 100644 --- a/gateway/cmd/gateway/main.go +++ b/gateway/cmd/gateway/main.go @@ -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 { diff --git a/gateway/internal/inbound/server.go b/gateway/internal/inbound/server.go new file mode 100644 index 0000000..164dbfb --- /dev/null +++ b/gateway/internal/inbound/server.go @@ -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 +} diff --git a/gateway/internal/inbound/server_test.go b/gateway/internal/inbound/server_test.go new file mode 100644 index 0000000..504f862 --- /dev/null +++ b/gateway/internal/inbound/server_test.go @@ -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 +}