From e3b93113f13d8f6a5a73320d2202753f86475d30 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Mon, 21 Sep 2026 09:52:39 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8E=A5=E5=85=A5=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E5=8A=A0=E5=AF=86=E4=B8=8E=E5=AE=9E=E5=90=8D=E8=AE=A4=E8=AF=81?= =?UTF-8?q?=E8=83=BD=E5=8A=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- go.mod | 11 +- go.sum | 28 +- scripts/IM-租户业务库初始化.sql | 6 + scripts/apply_hotgo_tenant_db.go | 2 + scripts/消息加密字段.sql | 86 ++++++ server/admin_internal_config.go | 50 +++- server/db/adapter.go | 6 + server/db/mysql/adapter.go | 11 +- server/db/mysql/group_message_shards.go | 3 +- server/db/mysql/group_schema.go | 3 +- server/db/mysql/realname_policy_config.go | 57 ++++ server/db/mysql/tenant_adapter.go | 63 +++-- server/im_agora_http.go | 1 + server/im_http.go | 120 +++++++- server/im_http_test.go | 162 +++++++++-- server/internal_admin_query_resources.go | 90 +++++- server/internal_api.go | 7 + server/kafka_config_events.go | 9 + server/kafka_message_persist.go | 10 +- server/realname/aliyun/id2meta.go | 112 ++++++++ server/realname/aliyun/id2meta_live_test.go | 36 +++ server/realtime_dispatcher.go | 108 ++++++++ server/realtime_dispatcher_test.go | 60 ++++ server/session.go | 3 + server/store/config_cache.go | 3 + .../local_message_crypto_integration_test.go | 191 +++++++++++++ server/store/message_crypto.go | 260 ++++++++++++++++++ server/store/message_crypto_test.go | 93 +++++++ server/store/store.go | 170 +++++++++++- server/store/types/message_crypto.go | 19 ++ server/store/types/security.go | 28 ++ server/store/types/types.go | 36 +-- server/topic_test.go | 2 +- 33 files changed, 1747 insertions(+), 99 deletions(-) create mode 100644 scripts/消息加密字段.sql create mode 100644 server/db/mysql/realname_policy_config.go create mode 100644 server/realname/aliyun/id2meta.go create mode 100644 server/realname/aliyun/id2meta_live_test.go create mode 100644 server/store/local_message_crypto_integration_test.go create mode 100644 server/store/message_crypto.go create mode 100644 server/store/message_crypto_test.go create mode 100644 server/store/types/message_crypto.go diff --git a/go.mod b/go.mod index 9689b55..abebad0 100644 --- a/go.mod +++ b/go.mod @@ -5,9 +5,11 @@ go 1.26.0 require ( firebase.google.com/go v3.13.0+incompatible github.com/AgoraIO/Tools/DynamicKey/AgoraDynamicKey/go/src v0.0.0-20260726132416-da06bfb83556 + github.com/alibabacloud-go/cloudauth-20190307/v3 v3.9.2 github.com/alibabacloud-go/darabonba-openapi/v2 v2.2.4 github.com/alibabacloud-go/green-20220302/v3 v3.5.4 github.com/alibabacloud-go/tea v1.5.2 + github.com/alibabacloud-go/tea-utils/v2 v2.0.9 github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible github.com/aws/aws-sdk-go-v2 v1.42.1 github.com/aws/aws-sdk-go-v2/config v1.32.27 @@ -52,7 +54,14 @@ require ( github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.57.0 // indirect github.com/alibabacloud-go/alibabacloud-gateway-spi v0.0.5 // indirect github.com/alibabacloud-go/debug v1.0.1 // indirect - github.com/alibabacloud-go/tea-utils/v2 v2.0.9 // indirect + github.com/alibabacloud-go/endpoint-util v1.1.0 // indirect + github.com/alibabacloud-go/openapi-util v0.1.0 // indirect + github.com/alibabacloud-go/openplatform-20191219/v2 v2.0.1 // indirect + github.com/alibabacloud-go/tea-fileform v1.1.1 // indirect + github.com/alibabacloud-go/tea-oss-sdk v1.1.3 // indirect + github.com/alibabacloud-go/tea-oss-utils v1.1.0 // indirect + github.com/alibabacloud-go/tea-utils v1.3.6 // indirect + github.com/alibabacloud-go/tea-xml v1.1.3 // indirect github.com/aliyun/credentials-go v1.4.5 // indirect github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.14 // indirect github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.30 // indirect diff --git a/go.sum b/go.sum index 22ff6f6..097c2c1 100644 --- a/go.sum +++ b/go.sum @@ -44,12 +44,16 @@ github.com/alibabacloud-go/alibabacloud-gateway-pop v0.0.6/go.mod h1:4EUIoxs/do2 github.com/alibabacloud-go/alibabacloud-gateway-spi v0.0.4/go.mod h1:sCavSAvdzOjul4cEqeVtvlSaSScfNsTQ+46HwlTL1hc= github.com/alibabacloud-go/alibabacloud-gateway-spi v0.0.5 h1:zE8vH9C7JiZLNJJQ5OwjU9mSi4T9ef9u3BURT6LCLC8= github.com/alibabacloud-go/alibabacloud-gateway-spi v0.0.5/go.mod h1:tWnyE9AjF8J8qqLk645oUmVUnFybApTQWklQmi5tY6g= +github.com/alibabacloud-go/cloudauth-20190307/v3 v3.9.2 h1:y4s0WQ1jrBtOJfXGgsv/83brJvkkHbFdORp0WDyVAuw= +github.com/alibabacloud-go/cloudauth-20190307/v3 v3.9.2/go.mod h1:kD75qqMQyjCiz6lssjRzYGTumcli8STLXQstVe6ytxk= github.com/alibabacloud-go/darabonba-array v0.1.0 h1:vR8s7b1fWAQIjEjWnuF0JiKsCvclSRTfDzZHTYqfufY= github.com/alibabacloud-go/darabonba-array v0.1.0/go.mod h1:BLKxr0brnggqOJPqT09DFJ8g3fsDshapUD3C3aOEFaI= github.com/alibabacloud-go/darabonba-encode-util v0.0.2 h1:1uJGrbsGEVqWcWxrS9MyC2NG0Ax+GpOM5gtupki31XE= github.com/alibabacloud-go/darabonba-encode-util v0.0.2/go.mod h1:JiW9higWHYXm7F4PKuMgEUETNZasrDM6vqVr/Can7H8= github.com/alibabacloud-go/darabonba-map v0.0.2 h1:qvPnGB4+dJbJIxOOfawxzF3hzMnIpjmafa0qOTp6udc= github.com/alibabacloud-go/darabonba-map v0.0.2/go.mod h1:28AJaX8FOE/ym8OUFWga+MtEzBunJwQGceGQlvaPGPc= +github.com/alibabacloud-go/darabonba-openapi/v2 v2.0.0/go.mod h1:5JHVmnHvGzR2wNdgaW1zDLQG8kOC4Uec8ubkMogW7OQ= +github.com/alibabacloud-go/darabonba-openapi/v2 v2.0.10/go.mod h1:26a14FGhZVELuz2cc2AolvW4RHmIO3/HRwsdHhaIPDE= github.com/alibabacloud-go/darabonba-openapi/v2 v2.2.4 h1:o6veen0IZ/Fe1JawwhwQMZcbw67CVDY1pQwXcNWPyQo= github.com/alibabacloud-go/darabonba-openapi/v2 v2.2.4/go.mod h1:eHjVxrT9g8uVYN/nyAwOFQEfkVA154ChiqRc2XnNKYU= github.com/alibabacloud-go/darabonba-signature-util v0.0.7 h1:UzCnKvsjPFzApvODDNEYqBHMFt1w98wC7FOo0InLyxg= @@ -64,27 +68,47 @@ github.com/alibabacloud-go/endpoint-util v1.1.0 h1:r/4D3VSw888XGaeNpP994zDUaxdgT github.com/alibabacloud-go/endpoint-util v1.1.0/go.mod h1:O5FuCALmCKs2Ff7JFJMudHs0I5EBgecXXxZRyswlEjE= github.com/alibabacloud-go/green-20220302/v3 v3.5.4 h1:ZnlHri7tTLl28JWB5S2J6MYvD0ewJFI7hedK4E7+gSk= github.com/alibabacloud-go/green-20220302/v3 v3.5.4/go.mod h1:PQTtoEuWOsuq7H7sgTRcXDw4Ut/Hekgy5z2mWGtO9OQ= +github.com/alibabacloud-go/openapi-util v0.0.11/go.mod h1:sQuElr4ywwFRlCCberQwKRFhRzIyG4QTP/P4y1CJ6Ws= github.com/alibabacloud-go/openapi-util v0.1.0 h1:0z75cIULkDrdEhkLWgi9tnLe+KhAFE/r5Pb3312/eAY= github.com/alibabacloud-go/openapi-util v0.1.0/go.mod h1:sQuElr4ywwFRlCCberQwKRFhRzIyG4QTP/P4y1CJ6Ws= +github.com/alibabacloud-go/openplatform-20191219/v2 v2.0.1 h1:L0TIjr9Qh/SLVc1yPhFkcB9+9SbCNK/jPq4ZKB5zmnc= +github.com/alibabacloud-go/openplatform-20191219/v2 v2.0.1/go.mod h1:EKxBRDLcMzwl4VLF/1WJwlByZZECJawPXUvinKMsTTs= github.com/alibabacloud-go/tea v1.1.0/go.mod h1:IkGyUSX4Ba1V+k4pCtJUc6jDpZLFph9QMy2VUPTwukg= github.com/alibabacloud-go/tea v1.1.7/go.mod h1:/tmnEaQMyb4Ky1/5D+SE1BAsa5zj/KeGOFfwYm3N/p4= github.com/alibabacloud-go/tea v1.1.8/go.mod h1:/tmnEaQMyb4Ky1/5D+SE1BAsa5zj/KeGOFfwYm3N/p4= +github.com/alibabacloud-go/tea v1.1.10/go.mod h1:/tmnEaQMyb4Ky1/5D+SE1BAsa5zj/KeGOFfwYm3N/p4= github.com/alibabacloud-go/tea v1.1.11/go.mod h1:/tmnEaQMyb4Ky1/5D+SE1BAsa5zj/KeGOFfwYm3N/p4= +github.com/alibabacloud-go/tea v1.1.12/go.mod h1:/tmnEaQMyb4Ky1/5D+SE1BAsa5zj/KeGOFfwYm3N/p4= github.com/alibabacloud-go/tea v1.1.17/go.mod h1:nXxjm6CIFkBhwW4FQkNrolwbfon8Svy6cujmKFUq98A= +github.com/alibabacloud-go/tea v1.1.19/go.mod h1:nXxjm6CIFkBhwW4FQkNrolwbfon8Svy6cujmKFUq98A= github.com/alibabacloud-go/tea v1.1.20/go.mod h1:nXxjm6CIFkBhwW4FQkNrolwbfon8Svy6cujmKFUq98A= github.com/alibabacloud-go/tea v1.2.2/go.mod h1:CF3vOzEMAG+bR4WOql8gc2G9H3EkH3ZLAQdpmpXMgwk= github.com/alibabacloud-go/tea v1.5.2 h1:DUzV/cRTbJC++rmADfqVu//GuWbaD64ZJ3yW3v2ukNU= github.com/alibabacloud-go/tea v1.5.2/go.mod h1:hgSs82CkOiehSQMoiFN79dL6zsGX7pVGvnn9SIEs8/0= -github.com/alibabacloud-go/tea-utils v1.3.1 h1:iWQeRzRheqCMuiF3+XkfybB3kTgUXkXX+JMrqfLeB2I= +github.com/alibabacloud-go/tea-fileform v1.1.1 h1:1YG6erAP3joQ0XdCXYIotuD7zyOM6qCR49xkp5FZDeU= +github.com/alibabacloud-go/tea-fileform v1.1.1/go.mod h1:ZeCV91o4ISmxidd686f0ebdS5EDHWU+vW+TkjLhrsFE= +github.com/alibabacloud-go/tea-oss-sdk v1.1.3 h1:EhAHI6edMeqgkZEqP7r4nc9iMWAUBKGxJHoBsOSKTtU= +github.com/alibabacloud-go/tea-oss-sdk v1.1.3/go.mod h1:yUnodpR3Bf2rudLE7V/Gft5txjJF30Pk+hH77K/Eab0= +github.com/alibabacloud-go/tea-oss-utils v1.1.0 h1:y65crjjcZ2Pbb6UZtC2deuIZHDVTS3IaDWE7M9nVLRc= +github.com/alibabacloud-go/tea-oss-utils v1.1.0/go.mod h1:PFCF12e9yEKyBUIn7X1IrF/pNjvxgkHy0CgxX4+xRuY= github.com/alibabacloud-go/tea-utils v1.3.1/go.mod h1:EI/o33aBfj3hETm4RLiAxF/ThQdSngxrpF8rKUDJjPE= +github.com/alibabacloud-go/tea-utils v1.3.6 h1:bVjrxHztM8hAs6nOfLWCgxQfAtKb9RgFFMV6J3rdvB4= +github.com/alibabacloud-go/tea-utils v1.3.6/go.mod h1:EI/o33aBfj3hETm4RLiAxF/ThQdSngxrpF8rKUDJjPE= +github.com/alibabacloud-go/tea-utils/v2 v2.0.0/go.mod h1:U5MTY10WwlquGPS34DOeomUGBB0gXbLueiq5Trwu0C4= github.com/alibabacloud-go/tea-utils/v2 v2.0.5/go.mod h1:dL6vbUT35E4F4bFTHL845eUloqaerYBYPsdWR2/jhe4= +github.com/alibabacloud-go/tea-utils/v2 v2.0.6/go.mod h1:qxn986l+q33J5VkialKMqT/TTs3E+U9MJpd001iWQ9I= github.com/alibabacloud-go/tea-utils/v2 v2.0.9 h1:y6pUIlhjxbZl9ObDAcmA1H3c21eaAxADHTDQmBnAIgA= github.com/alibabacloud-go/tea-utils/v2 v2.0.9/go.mod h1:qxn986l+q33J5VkialKMqT/TTs3E+U9MJpd001iWQ9I= +github.com/alibabacloud-go/tea-xml v1.1.1/go.mod h1:Rq08vgCcCAjHyRi/M7xlHKUykZCEtyBy9+DPF6GgEu8= +github.com/alibabacloud-go/tea-xml v1.1.2/go.mod h1:Rq08vgCcCAjHyRi/M7xlHKUykZCEtyBy9+DPF6GgEu8= +github.com/alibabacloud-go/tea-xml v1.1.3 h1:7LYnm+JbOq2B+T/B0fHC4Ies4/FofC4zHzYtqw7dgt0= +github.com/alibabacloud-go/tea-xml v1.1.3/go.mod h1:Rq08vgCcCAjHyRi/M7xlHKUykZCEtyBy9+DPF6GgEu8= github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible h1:8psS8a+wKfiLt1iVDX79F7Y6wUM49Lcha2FMXt4UM8g= github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible/go.mod h1:T/Aws4fEfogEE9v+HPhhw+CntffsBHJ8nXQCwKr0/g8= github.com/aliyun/credentials-go v1.1.2/go.mod h1:ozcZaMR5kLM7pwtCMEpVmQ242suV6qTJya2bDq4X1Tw= github.com/aliyun/credentials-go v1.3.1/go.mod h1:8jKYhQuDawt8x2+fusqa1Y6mPxemTsBEN04dgcAcYz0= github.com/aliyun/credentials-go v1.3.6/go.mod h1:1LxUuX7L5YrZUWzBrRyk0SwSdH4OmPrib8NVePL3fxM= +github.com/aliyun/credentials-go v1.3.10/go.mod h1:Jm6d+xIgwJVLVWT561vy67ZRP4lPTQxMbEYRuT2Ti1U= github.com/aliyun/credentials-go v1.4.5 h1:O76WYKgdy1oQYYiJkERjlA2dxGuvLRrzuO2ScrtGWSk= github.com/aliyun/credentials-go v1.4.5/go.mod h1:Jm6d+xIgwJVLVWT561vy67ZRP4lPTQxMbEYRuT2Ti1U= github.com/aws/aws-sdk-go-v2 v1.42.1 h1:9eOTgu1z/dVtYpNZ3/8/XbbaX0x/BqE3HUzAzs6K0ek= @@ -138,6 +162,7 @@ github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/clbanning/mxj/v2 v2.5.5/go.mod h1:hNiWqW14h+kc+MdF9C6/YoRfjEJoR3ou6tn/Qo+ve2s= github.com/clbanning/mxj/v2 v2.7.0 h1:WA/La7UGCanFe5NpHF0Q3DNtnCsVoxbPKuyBNHWRyME= github.com/clbanning/mxj/v2 v2.7.0/go.mod h1:hNiWqW14h+kc+MdF9C6/YoRfjEJoR3ou6tn/Qo+ve2s= github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw= @@ -494,6 +519,7 @@ golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI= golang.org/x/text v0.38.0 h1:sXmwo9DwP3OK9EZ7PqAdaooSGozfl/3a6/xJcbzPRhE= golang.org/x/text v0.38.0/go.mod h1:YXZt3QhHUKYT53r2lLKFIVi6Ao1jdzrTR/KQ09qyxF4= +golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= diff --git a/scripts/IM-租户业务库初始化.sql b/scripts/IM-租户业务库初始化.sql index db90c5f..d75e9cc 100644 --- a/scripts/IM-租户业务库初始化.sql +++ b/scripts/IM-租户业务库初始化.sql @@ -721,6 +721,8 @@ CREATE TABLE IF NOT EXISTS `im_group_messages` ( `msg_type` varchar(32) COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT 'text' COMMENT 'text/image/file/audio/video/system 等', `status` varchar(32) COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT 'normal' COMMENT 'normal/recalled/deleted/blocked', `content` json DEFAULT NULL COMMENT '消息正文或 Drafty 内容', + `content_nonce` varbinary(12) DEFAULT NULL COMMENT '每条消息独立随机Nonce', + `key_version` int NOT NULL DEFAULT '0' COMMENT '商户消息DEK版本', `head` json DEFAULT NULL COMMENT '消息头，包含 @、引用、客户端能力等', `mentioned_user_ids` json DEFAULT NULL COMMENT '@ 用户列表，便于提醒和检索', `mention_all` tinyint(1) NOT NULL DEFAULT '0' COMMENT '是否 @ 全体', @@ -1088,6 +1090,8 @@ CREATE TABLE IF NOT EXISTS `im_message_revisions` ( `actor` bigint NOT NULL, `head` json DEFAULT NULL, `content` json DEFAULT NULL, + `content_nonce` varbinary(12) DEFAULT NULL, + `key_version` int NOT NULL DEFAULT '0', `reason` varchar(256) NOT NULL DEFAULT '', PRIMARY KEY (`id`), UNIQUE KEY `uk_im_message_revisions_version` (`topic`,`seqid`,`version`) @@ -1108,6 +1112,8 @@ CREATE TABLE IF NOT EXISTS `im_messages` ( `recalled_by` bigint DEFAULT NULL, `head` json DEFAULT NULL, `content` json DEFAULT NULL, + `content_nonce` varbinary(12) DEFAULT NULL, + `key_version` int NOT NULL DEFAULT '0', PRIMARY KEY (`id`), UNIQUE KEY `messages_sender_client_mid` (`from`,`client_msg_id`), UNIQUE KEY `uk_im_messages_topic_seq` (`topic`,`seqid`) diff --git a/scripts/apply_hotgo_tenant_db.go b/scripts/apply_hotgo_tenant_db.go index b8e0550..0375bfe 100644 --- a/scripts/apply_hotgo_tenant_db.go +++ b/scripts/apply_hotgo_tenant_db.go @@ -383,6 +383,7 @@ func main() { profile_visibility VARCHAR(32) NOT NULL DEFAULT 'friends', last_seen_visibility VARCHAR(32) NOT NULL DEFAULT 'friends', read_receipts TINYINT NOT NULL DEFAULT 1, + friend_request_verification TINYINT NOT NULL DEFAULT 1, group_invite_verification TINYINT NOT NULL DEFAULT 0, version INT NOT NULL DEFAULT 1, created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), @@ -540,6 +541,7 @@ func main() { "ALTER TABLE tinode.im_youth_mode_settings ADD COLUMN enabled_at DATETIME(3) AFTER restrictions", "ALTER TABLE tinode.im_users ADD COLUMN phone VARCHAR(32) NULL AFTER useragent", "ALTER TABLE tinode.im_user_profiles ADD COLUMN email VARCHAR(128) NOT NULL DEFAULT '' AFTER region", + "ALTER TABLE tinode.im_user_privacy_settings ADD COLUMN friend_request_verification TINYINT NOT NULL DEFAULT 1 AFTER read_receipts", "ALTER TABLE tinode.im_user_privacy_settings ADD COLUMN group_invite_verification TINYINT NOT NULL DEFAULT 0 AFTER read_receipts", } { if _, err = db.Exec(stmt); err != nil && !strings.Contains(err.Error(), "Duplicate column name") { diff --git a/scripts/消息加密字段.sql b/scripts/消息加密字段.sql new file mode 100644 index 0000000..8bd4a18 --- /dev/null +++ b/scripts/消息加密字段.sql @@ -0,0 +1,86 @@ +-- 在当前商户业务库增加消息加密字段。 +-- 可重复执行，覆盖单聊、消息修订、群聊主表及全部现有群消息月分片。 + +DELIMITER $$ + +DROP PROCEDURE IF EXISTS add_message_crypto_column$$ +CREATE PROCEDURE add_message_crypto_column( + IN p_table VARCHAR(64), + IN p_column VARCHAR(64), + IN p_definition TEXT +) +BEGIN + IF EXISTS ( + SELECT 1 FROM information_schema.TABLES + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = p_table + ) AND NOT EXISTS ( + SELECT 1 FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = p_table AND COLUMN_NAME = p_column + ) THEN + SET @ddl = CONCAT('ALTER TABLE `', REPLACE(p_table, '`', '``'), '` ADD COLUMN `', + REPLACE(p_column, '`', '``'), '` ', p_definition); + PREPARE stmt FROM @ddl; + EXECUTE stmt; + DEALLOCATE PREPARE stmt; + END IF; +END$$ + +DROP PROCEDURE IF EXISTS drop_message_crypto_column$$ +CREATE PROCEDURE drop_message_crypto_column(IN p_table VARCHAR(64), IN p_column VARCHAR(64)) +BEGIN + IF EXISTS ( + SELECT 1 FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = p_table AND COLUMN_NAME = p_column + ) THEN + SET @ddl = CONCAT('ALTER TABLE `', REPLACE(p_table, '`', '``'), '` DROP COLUMN `', + REPLACE(p_column, '`', '``'), '`'); + PREPARE stmt FROM @ddl; + EXECUTE stmt; + DEALLOCATE PREPARE stmt; + END IF; +END$$ + +DROP PROCEDURE IF EXISTS migrate_message_crypto_table$$ +CREATE PROCEDURE migrate_message_crypto_table(IN p_table VARCHAR(64)) +BEGIN + CALL add_message_crypto_column(p_table, 'content_nonce', 'VARBINARY(12) NULL COMMENT ''每条消息独立随机Nonce'' AFTER `content`'); + CALL add_message_crypto_column(p_table, 'key_version', 'INT NOT NULL DEFAULT 0 COMMENT ''商户消息DEK版本'' AFTER `content_nonce`'); + CALL drop_message_crypto_column(p_table, 'content_ciphertext'); + CALL drop_message_crypto_column(p_table, 'encryption_alg'); + CALL drop_message_crypto_column(p_table, 'encrypted'); +END$$ + +DROP PROCEDURE IF EXISTS migrate_message_crypto_schema$$ +CREATE PROCEDURE migrate_message_crypto_schema() +BEGIN + DECLARE done INT DEFAULT 0; + DECLARE table_name_value VARCHAR(64); + DECLARE shard_cursor CURSOR FOR + SELECT TABLE_NAME + FROM information_schema.TABLES + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME REGEXP '^im_group_messages_[0-9]{6}$'; + DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = 1; + + CALL migrate_message_crypto_table('im_messages'); + CALL migrate_message_crypto_table('im_message_revisions'); + CALL migrate_message_crypto_table('im_group_messages'); + + OPEN shard_cursor; + shard_loop: LOOP + FETCH shard_cursor INTO table_name_value; + IF done = 1 THEN + LEAVE shard_loop; + END IF; + CALL migrate_message_crypto_table(table_name_value); + END LOOP; + CLOSE shard_cursor; +END$$ + +CALL migrate_message_crypto_schema()$$ +DROP PROCEDURE migrate_message_crypto_schema$$ +DROP PROCEDURE migrate_message_crypto_table$$ +DROP PROCEDURE drop_message_crypto_column$$ +DROP PROCEDURE add_message_crypto_column$$ + +DELIMITER ; diff --git a/server/admin_internal_config.go b/server/admin_internal_config.go index c1ec9cb..b5d1506 100644 --- a/server/admin_internal_config.go +++ b/server/admin_internal_config.go @@ -90,6 +90,7 @@ func initAdminInternalAPI(raw json.RawMessage) (*adminInternalClient, error) { } store.SetAdminInternalConfigProvider(client) loginIPWhitelistProvider = client + go store.WarmMessageEncryptionKeys() logs.Info.Printf("Admin internal config client enabled: base_url=%s service=%s", cfg.BaseURL, cfg.ServiceName) return client, nil } @@ -211,6 +212,40 @@ func (c *adminInternalClient) SecurityPolicy(tenantCode string) (*types.Security return &result, nil } +func (c *adminInternalClient) MessageKey(tenantCode string, version int) (*types.MessageKey, error) { + path := "/internal/im-config/tenants/" + url.PathEscape(strings.TrimSpace(tenantCode)) + "/message-keys/active" + if version > 0 { + path = "/internal/im-config/tenants/" + url.PathEscape(strings.TrimSpace(tenantCode)) + "/message-keys/" + strconv.Itoa(version) + } + var result types.MessageKey + if err := c.getJSON(path, nil, &result); err != nil { + return nil, err + } + return &result, nil +} + +func (c *adminInternalClient) MessageKeyAck(tenantCode string, version int) error { + if version <= 0 { + return types.ErrMalformed + } + path := "/internal/im-config/tenants/" + url.PathEscape(strings.TrimSpace(tenantCode)) + "/message-keys/ack" + return c.doJSON(http.MethodPost, path, nil, map[string]int{"version": version}, nil) +} + +func (c *adminInternalClient) RealNamePolicy(id types.TenantID) (*types.RealNamePolicy, error) { + if !id.IsValid() { + return nil, types.ErrMalformed + } + var result types.RealNamePolicy + path := "/internal/im-config/tenants/by-id/" + strconv.FormatInt(int64(id), 10) + "/realname-policy" + if err := c.getJSON(path, nil, &result); err != nil { + return nil, err + } + result.TenantID = id + result.Normalize() + return &result, nil +} + func (c *adminInternalClient) TenantLoginIPWhitelist(id types.TenantID) (*tenantLoginIPWhitelistConfig, error) { if !id.IsValid() { return nil, types.ErrMalformed @@ -267,6 +302,10 @@ func (c *adminInternalClient) LatestAppRelease(platform, channel string) (*types } func (c *adminInternalClient) getJSON(path string, query url.Values, out any) error { + return c.doJSON(http.MethodGet, path, query, nil, out) +} + +func (c *adminInternalClient) doJSON(method, path string, query url.Values, payload, out any) error { if c == nil || c.http == nil { return types.ErrUnsupported } @@ -278,11 +317,18 @@ func (c *adminInternalClient) getJSON(path string, query url.Values, out any) er if len(query) > 0 { u.RawQuery = query.Encode() } - req, err := http.NewRequest(http.MethodGet, u.String(), nil) + var requestBody []byte + if payload != nil { + requestBody, err = json.Marshal(payload) + if err != nil { + return err + } + } + req, err := http.NewRequest(method, u.String(), nil) if err != nil { return err } - if err = c.signRequest(req, nil); err != nil { + if err = c.signRequest(req, requestBody); err != nil { return err } resp, err := c.http.Do(req) diff --git a/server/db/adapter.go b/server/db/adapter.go index 7acbdf0..05720b4 100644 --- a/server/db/adapter.go +++ b/server/db/adapter.go @@ -228,6 +228,12 @@ type ThirdPartyConfigAdapter interface { ThirdPartyConfig(group string, keys ...string) (map[string]string, error) } +// RealNamePolicyConfigAdapter reads tenant-level real-name review policy from +// the control database. +type RealNamePolicyConfigAdapter interface { + RealNamePolicy(t.TenantID) (*t.RealNamePolicy, error) +} + // TenantEntryAdapter stores public tenant entry data before a user is authenticated. type TenantEntryAdapter interface { PublicUploadCreate(*t.PublicUpload) (*t.PublicUpload, error) diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index 8218481..5d63e19 100644 --- a/server/db/mysql/adapter.go +++ b/server/db/mysql/adapter.go @@ -594,6 +594,11 @@ func createTenantRegistry(tx *sql.Tx, registryDB, businessDB string) error { max_users BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '0 means unlimited', max_groups BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '0 means unlimited', max_group_members INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '0 means unlimited', + realname_review_policy TINYINT(1) NOT NULL DEFAULT 0 COMMENT 'Real-name review policy: 0 none, 1 auto then manual', + realname_provider_mode VARCHAR(32) NOT NULL DEFAULT 'id_card_2factor' COMMENT 'id_card_2factor or smart_verify', + realname_required TINYINT(1) NOT NULL DEFAULT 0 COMMENT 'Require real-name verification', + realname_allow_skip TINYINT(1) NOT NULL DEFAULT 1 COMMENT 'Allow skipping real-name reminder', + realname_remind_after VARCHAR(32) NOT NULL DEFAULT '24h' COMMENT 'Real-name reminder interval', aliyun_moderation_enabled TINYINT(1) NOT NULL DEFAULT 0 COMMENT 'Enable Aliyun content moderation', local_moderation_enabled TINYINT(1) NOT NULL DEFAULT 0 COMMENT 'Enable local content moderation', created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), @@ -3309,9 +3314,9 @@ func (a *adapter) MessageSave(msg *t.Message) error { // Using a sequential ID provided by the database. clientMsgID := sql.NullString{String: msg.ClientMsgID, Valid: msg.ClientMsgID != ""} res, err := a.db.ExecContext(ctx, - "INSERT INTO im_messages(createdAt,updatedAt,seqid,topic,`from`,client_msg_id,version,head,content) VALUES(?,?,?,?,?,?,?,?,?)", + "INSERT INTO im_messages(createdAt,updatedAt,seqid,topic,`from`,client_msg_id,version,head,content,content_nonce,key_version) VALUES(?,?,?,?,?,?,?,?,?,?,?)", msg.CreatedAt, msg.UpdatedAt, msg.SeqId, msg.Topic, - store.DecodeUid(t.ParseUid(msg.From)), clientMsgID, messageVersion(msg.Version), msg.Head, common.ToJSON(msg.Content)) + store.DecodeUid(t.ParseUid(msg.From)), clientMsgID, messageVersion(msg.Version), msg.Head, common.ToJSON(msg.Content), msg.ContentNonce, msg.KeyVersion) if isDupe(err) { return t.ErrDuplicate } @@ -3364,7 +3369,7 @@ func (a *adapter) MessageGetAll(topic string, forUser t.Uid, opts *t.QueryOpt) ( rows, err := a.db.QueryxContext( ctx, - "SELECT m.createdat,m.updatedat,m.deletedat,m.delid,m.seqid,m.topic,m.`from`,m.head,m.content"+ + "SELECT m.createdat,m.updatedat,m.deletedat,m.delid,m.seqid,m.topic,m.`from`,m.head,m.content,m.content_nonce,m.key_version"+ " FROM im_messages AS m LEFT JOIN im_dellog AS d"+ " ON d.topic=m.topic AND m.seqid BETWEEN d.low AND d.hi-1 AND d.deletedfor=?"+ " WHERE m.delid=0 AND m.topic=? "+seqIdConstraint+" AND d.deletedfor IS NULL"+ diff --git a/server/db/mysql/group_message_shards.go b/server/db/mysql/group_message_shards.go index de92e12..c201fd4 100644 --- a/server/db/mysql/group_message_shards.go +++ b/server/db/mysql/group_message_shards.go @@ -97,7 +97,8 @@ func ensureGroupMessageShard(exec sqlExecutor, start time.Time, status string) e message_id BIGINT NOT NULL, topic VARCHAR(64) NOT NULL, group_id BIGINT NOT NULL, seqid BIGINT NOT NULL, from_user BIGINT NOT NULL, client_msg_id VARCHAR(64), msg_type VARCHAR(32) NOT NULL DEFAULT 'text', status VARCHAR(32) NOT NULL DEFAULT 'normal', - content JSON, head JSON, mentioned_user_ids JSON, mention_all TINYINT(1) NOT NULL DEFAULT 0, + content JSON, content_nonce VARBINARY(12), key_version INT NOT NULL DEFAULT 0, + head JSON, mentioned_user_ids JSON, mention_all TINYINT(1) NOT NULL DEFAULT 0, reply_seqid BIGINT, root_seqid BIGINT, file_count INT NOT NULL DEFAULT 0, content_size INT NOT NULL DEFAULT 0, has_risk TINYINT(1) NOT NULL DEFAULT 0, audit_status VARCHAR(32) NOT NULL DEFAULT 'pass', version INT NOT NULL DEFAULT 1, diff --git a/server/db/mysql/group_schema.go b/server/db/mysql/group_schema.go index 20254ee..0c5b8eb 100644 --- a/server/db/mysql/group_schema.go +++ b/server/db/mysql/group_schema.go @@ -27,7 +27,8 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error { id BIGINT NOT NULL AUTO_INCREMENT, topic VARCHAR(64) NOT NULL, group_id BIGINT NOT NULL, seqid BIGINT NOT NULL, from_user BIGINT NOT NULL, client_msg_id VARCHAR(64), msg_type VARCHAR(32) NOT NULL DEFAULT 'text', status VARCHAR(32) NOT NULL DEFAULT 'normal', - content JSON, head JSON, mentioned_user_ids JSON, mention_all TINYINT(1) NOT NULL DEFAULT 0, + content JSON, content_nonce VARBINARY(12), key_version INT NOT NULL DEFAULT 0, + head JSON, mentioned_user_ids JSON, mention_all TINYINT(1) NOT NULL DEFAULT 0, reply_seqid BIGINT, root_seqid BIGINT, file_count INT NOT NULL DEFAULT 0, content_size INT NOT NULL DEFAULT 0, has_risk TINYINT(1) NOT NULL DEFAULT 0, audit_status VARCHAR(32) NOT NULL DEFAULT 'pass', version INT NOT NULL DEFAULT 1, diff --git a/server/db/mysql/realname_policy_config.go b/server/db/mysql/realname_policy_config.go new file mode 100644 index 0000000..5f04890 --- /dev/null +++ b/server/db/mysql/realname_policy_config.go @@ -0,0 +1,57 @@ +//go:build mysql +// +build mysql + +package mysql + +import ( + "database/sql" + "errors" + + t "github.com/tinode/chat/server/store/types" +) + +func (a *adapter) RealNamePolicy(tenantID t.TenantID) (*t.RealNamePolicy, error) { + if !tenantID.IsValid() { + return nil, t.ErrMalformed + } + db := a.registryDB + if db == nil { + db = a.db + } + if db == nil { + return nil, t.ErrInternal + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + var policy t.RealNamePolicy + err := db.GetContext(ctx, &policy, `SELECT tenant_id, realname_review_policy, realname_provider_mode, + realname_required, realname_allow_skip, realname_remind_after + FROM im_tenant_config WHERE tenant_id=? LIMIT 1`, tenantID) + if errors.Is(err, sql.ErrNoRows) { + policy = *defaultRealNamePolicy(tenantID) + return &policy, nil + } + if isUnknownColumn(err) { + policy = *defaultRealNamePolicy(tenantID) + return &policy, nil + } + if err != nil { + return nil, err + } + policy.Normalize() + return &policy, nil +} + +func defaultRealNamePolicy(tenantID t.TenantID) *t.RealNamePolicy { + policy := &t.RealNamePolicy{ + TenantID: tenantID, + ReviewPolicy: t.RealNameReviewPolicyNone, + ProviderMode: "id_card_2factor", + AllowSkip: true, + RemindAfter: "24h", + } + policy.Normalize() + return policy +} diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index 94f939b..6c15bfe 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -2065,8 +2065,8 @@ func (a *adapter) TenantMessageSave(id t.TenantID, msg *t.Message) error { } clientMsgID := sql.NullString{String: msg.ClientMsgID, Valid: msg.ClientMsgID != ""} res, err := a.db.ExecContext(ctx, - "INSERT INTO im_messages(tenant_id,createdat,updatedat,seqid,topic,`from`,client_msg_id,version,head,content) VALUES(?,?,?,?,?,?,?,?,?,?)", - id, msg.CreatedAt, msg.UpdatedAt, msg.SeqId, msg.Topic, decodeUID(t.ParseUid(msg.From)), clientMsgID, messageVersion(msg.Version), msg.Head, common.ToJSON(msg.Content)) + "INSERT INTO im_messages(tenant_id,createdat,updatedat,seqid,topic,`from`,client_msg_id,version,head,content,content_nonce,key_version) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", + id, msg.CreatedAt, msg.UpdatedAt, msg.SeqId, msg.Topic, decodeUID(t.ParseUid(msg.From)), clientMsgID, messageVersion(msg.Version), msg.Head, common.ToJSON(msg.Content), msg.ContentNonce, msg.KeyVersion) if isDupe(err) { return t.ErrDuplicate } @@ -2139,12 +2139,12 @@ func (a *adapter) TenantGroupMessageSave(id t.TenantID, msg *t.Message) error { } query := fmt.Sprintf(`INSERT INTO %s (message_id,topic,group_id,seqid,from_user,client_msg_id,msg_type,status,content,head,mentioned_user_ids,mention_all, - file_count,content_size,version,created_at,updated_at) - VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, table) + file_count,content_size,version,created_at,updated_at,content_nonce,key_version) + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, table) _, err = tx.ExecContext(ctx, query, messageID, msg.Topic, groupID, msg.SeqId, decodeUID(t.ParseUid(msg.From)), clientMsgID, messageHeadType(msg.Head), "normal", contentJSON, msg.Head, messageHeadJSON(msg.Head, "mentioned_user_ids"), messageHeadBool(msg.Head, "mention_all"), - 0, len(contentJSON), messageVersion(msg.Version), msg.CreatedAt, msg.UpdatedAt) + 0, len(contentJSON), messageVersion(msg.Version), msg.CreatedAt, msg.UpdatedAt, msg.ContentNonce, msg.KeyVersion) if isDupe(err) { return t.ErrDuplicate } @@ -2191,7 +2191,7 @@ func (a *adapter) TenantMessageGetByClientID(id t.TenantID, from t.Uid, clientMs RecalledBy sql.NullInt64 `db:"recalled_by"` }{} err := a.db.QueryRowxContext(ctx, - "SELECT id,createdat,updatedat,deletedat,delid,seqid,topic,`from`,IFNULL(client_msg_id,'') AS client_msg_id,version,recalled_at,recalled_by,head,content "+ + "SELECT id,createdat,updatedat,deletedat,delid,seqid,topic,`from`,IFNULL(client_msg_id,'') AS client_msg_id,version,recalled_at,recalled_by,head,content,content_nonce,key_version "+ "FROM im_messages WHERE `from`=? AND client_msg_id=?", decodeUID(from), clientMsgID).StructScan(&row) if err == sql.ErrNoRows { @@ -2246,7 +2246,7 @@ func (a *adapter) TenantGroupMessageGetByClientID(id t.TenantID, topic string, f t.Message RecalledBy sql.NullInt64 `db:"recalled_by"` }{} - query := fmt.Sprintf("SELECT %s AS id,created_at AS createdat,updated_at AS updatedat,deleted_at AS deletedat,0 AS delid,seqid,topic,from_user AS `from`,IFNULL(client_msg_id,'') AS client_msg_id,version,recalled_at,recalled_by,head,content,'%s' AS shard_month FROM %s WHERE topic=? AND from_user=? AND client_msg_id=?", idColumn, month, table) + query := fmt.Sprintf("SELECT %s AS id,created_at AS createdat,updated_at AS updatedat,deleted_at AS deletedat,0 AS delid,seqid,topic,from_user AS `from`,IFNULL(client_msg_id,'') AS client_msg_id,version,recalled_at,recalled_by,head,content,content_nonce,key_version,'%s' AS shard_month FROM %s WHERE topic=? AND from_user=? AND client_msg_id=?", idColumn, month, table) err = a.db.QueryRowxContext(ctx, query, topic, decodeUID(from), clientMsgID).StructScan(&row) if err == nil { msg := normalizeGroupMessage(id, row.Message, row.RecalledBy) @@ -2442,13 +2442,14 @@ func (a *adapter) TenantMessageEdit(id t.TenantID, topic string, seq int64, acto return nil, t.ErrDuplicate } if _, err = tx.ExecContext(ctx, - "INSERT INTO im_message_revisions(createdat,updatedat,topic,seqid,version,actor,head,content,reason) VALUES(?,?,?,?,?,?,?,?,?)", - editedAt, editedAt, topic, seq, msg.Version, decodeUID(actor), msg.Head, common.ToJSON(msg.Content), strings.TrimSpace(reason)); err != nil { + "INSERT INTO im_message_revisions(createdat,updatedat,topic,seqid,version,actor,head,content,content_nonce,key_version,reason) VALUES(?,?,?,?,?,?,?,?,?,?,?)", + editedAt, editedAt, topic, seq, msg.Version, decodeUID(actor), msg.Head, common.ToJSON(msg.Content), msg.ContentNonce, msg.KeyVersion, strings.TrimSpace(reason)); err != nil { return nil, err } + storedContent, nonce, keyVersion := messageStoredContent(content) _, err = tx.ExecContext(ctx, - "UPDATE im_messages SET updatedat=?,version=version+1,head=?,content=? WHERE topic=? AND seqid=? AND version=?", - editedAt, head, common.ToJSON(content), topic, seq, expectedVersion) + "UPDATE im_messages SET updatedat=?,version=version+1,head=?,content=?,content_nonce=?,key_version=? WHERE topic=? AND seqid=? AND version=?", + editedAt, head, common.ToJSON(storedContent), nonce, keyVersion, topic, seq, expectedVersion) if err != nil { return nil, err } @@ -2496,14 +2497,15 @@ func (a *adapter) TenantGroupMessageEdit(id t.TenantID, topic string, seq int64, return nil, t.ErrDuplicate } if _, err = tx.ExecContext(ctx, - "INSERT INTO im_message_revisions(createdat,updatedat,topic,seqid,version,actor,head,content,reason) VALUES(?,?,?,?,?,?,?,?,?)", - editedAt, editedAt, topic, seq, msg.Version, decodeUID(actor), msg.Head, common.ToJSON(msg.Content), strings.TrimSpace(reason)); err != nil { + "INSERT INTO im_message_revisions(createdat,updatedat,topic,seqid,version,actor,head,content,content_nonce,key_version,reason) VALUES(?,?,?,?,?,?,?,?,?,?,?)", + editedAt, editedAt, topic, seq, msg.Version, decodeUID(actor), msg.Head, common.ToJSON(msg.Content), msg.ContentNonce, msg.KeyVersion, strings.TrimSpace(reason)); err != nil { return nil, err } - contentJSON := common.ToJSON(content) + storedContent, nonce, keyVersion := messageStoredContent(content) + contentJSON := common.ToJSON(storedContent) _, err = tx.ExecContext(ctx, fmt.Sprintf( - "UPDATE %s SET updated_at=?,version=version+1,msg_type=?,head=?,content=?,content_size=? WHERE topic=? AND seqid=? AND version=?", messageTable), - editedAt, messageHeadType(head), head, contentJSON, len(contentJSON), topic, seq, expectedVersion) + "UPDATE %s SET updated_at=?,version=version+1,msg_type=?,head=?,content=?,content_nonce=?,key_version=?,content_size=? WHERE topic=? AND seqid=? AND version=?", messageTable), + editedAt, messageHeadType(head), head, contentJSON, nonce, keyVersion, len(contentJSON), topic, seq, expectedVersion) if err != nil { return nil, err } @@ -2561,7 +2563,7 @@ func (a *adapter) TenantMessageGetAll(id t.TenantID, topic string, uid t.Uid, op defer cancel() } rows, err := a.db.QueryxContext(ctx, - "SELECT m.tenant_id,m.createdat,m.updatedat,m.deletedat,m.delid,m.seqid,m.topic,m.`from`,IFNULL(m.client_msg_id,'') AS client_msg_id,m.version,m.recalled_at,m.recalled_by,m.head,m.content "+ + "SELECT m.tenant_id,m.createdat,m.updatedat,m.deletedat,m.delid,m.seqid,m.topic,m.`from`,IFNULL(m.client_msg_id,'') AS client_msg_id,m.version,m.recalled_at,m.recalled_by,m.head,m.content,m.content_nonce,m.key_version "+ "FROM im_messages AS m LEFT JOIN im_dellog AS d ON d.tenant_id=m.tenant_id AND d.topic=m.topic "+ "AND m.seqid BETWEEN d.low AND d.hi-1 AND d.deletedfor=? WHERE m.tenant_id=? AND m.delid=0 AND m.topic=? "+ seq+" AND d.deletedfor IS NULL ORDER BY m.seqid DESC LIMIT ?", @@ -2640,7 +2642,7 @@ func (a *adapter) TenantGroupMessageGetAll(id t.TenantID, topic string, uid t.Ui args := []any{store.DecodeUid(uid), topic} args = append(args, seqArgs...) args = append(args, limit-len(msgs)) - query := fmt.Sprintf("SELECT m.%s AS id,m.created_at AS createdat,m.updated_at AS updatedat,m.deleted_at AS deletedat,0 AS delid,m.seqid,m.topic,m.from_user AS `from`,IFNULL(m.client_msg_id,'') AS client_msg_id,m.version,m.recalled_at,m.recalled_by,m.head,m.content,'%s' AS shard_month FROM %s AS m LEFT JOIN im_dellog AS d ON d.topic COLLATE utf8mb4_unicode_ci=m.topic AND m.seqid BETWEEN d.low AND d.hi-1 AND d.deletedfor=? WHERE m.status!='deleted' AND m.topic=? %s AND d.deletedfor IS NULL ORDER BY m.seqid DESC LIMIT ?", idColumn, month, table, seqClause) + query := fmt.Sprintf("SELECT m.%s AS id,m.created_at AS createdat,m.updated_at AS updatedat,m.deleted_at AS deletedat,0 AS delid,m.seqid,m.topic,m.from_user AS `from`,IFNULL(m.client_msg_id,'') AS client_msg_id,m.version,m.recalled_at,m.recalled_by,m.head,m.content,m.content_nonce,m.key_version,'%s' AS shard_month FROM %s AS m LEFT JOIN im_dellog AS d ON d.topic COLLATE utf8mb4_unicode_ci=m.topic AND m.seqid BETWEEN d.low AND d.hi-1 AND d.deletedfor=? WHERE m.status!='deleted' AND m.topic=? %s AND d.deletedfor IS NULL ORDER BY m.seqid DESC LIMIT ?", idColumn, month, table, seqClause) rows, queryErr := a.db.QueryxContext(ctx, query, args...) if queryErr != nil { return nil, queryErr @@ -2715,7 +2717,7 @@ func (a *adapter) TenantMessageSearch(id t.TenantID, topic string, uid t.Uid, qu queryArgs := append([]any{decodeUID(uid)}, args...) queryArgs = append(queryArgs, scanLimit) rows, err := a.db.QueryxContext(ctx, - "SELECT m.createdat,m.updatedat,m.deletedat,m.delid,m.seqid,m.topic,m.`from`,IFNULL(m.client_msg_id,'') AS client_msg_id,m.version,m.recalled_at,m.recalled_by,m.head,m.content "+ + "SELECT m.createdat,m.updatedat,m.deletedat,m.delid,m.seqid,m.topic,m.`from`,IFNULL(m.client_msg_id,'') AS client_msg_id,m.version,m.recalled_at,m.recalled_by,m.head,m.content,m.content_nonce,m.key_version "+ "FROM im_messages AS m LEFT JOIN im_dellog AS d ON d.topic=m.topic "+ "AND m.seqid BETWEEN d.low AND d.hi-1 AND d.deletedfor=? WHERE "+strings.Join(where, " AND ")+" AND d.deletedfor IS NULL ORDER BY m.seqid DESC LIMIT ?", queryArgs...) @@ -2796,7 +2798,7 @@ func (a *adapter) tenantGroupMessageSearch(id t.TenantID, topic string, uid t.Ui } queryArgs := append([]any{store.DecodeUid(uid)}, args...) queryArgs = append(queryArgs, scanLimit-len(seen)) - query := fmt.Sprintf("SELECT m.%s AS id,m.created_at AS createdat,m.updated_at AS updatedat,m.deleted_at AS deletedat,0 AS delid,m.seqid,m.topic,m.from_user AS `from`,IFNULL(m.client_msg_id,'') AS client_msg_id,m.version,m.recalled_at,m.recalled_by,m.head,m.content,'%s' AS shard_month FROM %s AS m LEFT JOIN im_dellog AS d ON d.topic COLLATE utf8mb4_unicode_ci=m.topic AND m.seqid BETWEEN d.low AND d.hi-1 AND d.deletedfor=? WHERE %s AND d.deletedfor IS NULL ORDER BY m.seqid DESC LIMIT ?", idColumn, month, table, strings.Join(where, " AND ")) + query := fmt.Sprintf("SELECT m.%s AS id,m.created_at AS createdat,m.updated_at AS updatedat,m.deleted_at AS deletedat,0 AS delid,m.seqid,m.topic,m.from_user AS `from`,IFNULL(m.client_msg_id,'') AS client_msg_id,m.version,m.recalled_at,m.recalled_by,m.head,m.content,m.content_nonce,m.key_version,'%s' AS shard_month FROM %s AS m LEFT JOIN im_dellog AS d ON d.topic COLLATE utf8mb4_unicode_ci=m.topic AND m.seqid BETWEEN d.low AND d.hi-1 AND d.deletedfor=? WHERE %s AND d.deletedfor IS NULL ORDER BY m.seqid DESC LIMIT ?", idColumn, month, table, strings.Join(where, " AND ")) rows, queryErr := a.db.QueryxContext(ctx, query, queryArgs...) if queryErr != nil { return nil, "", queryErr @@ -2863,6 +2865,10 @@ func messageSearchScanLimit(limit int, queryText string) int { } func messageMatchesSearch(msg t.Message, queryText string) bool { + if msg.KeyVersion > 0 { + // Store decrypts and filters encrypted messages after the adapter returns. + return true + } queryText = normalizeMessageSearchText(queryText) if queryText == "" { return true @@ -3144,8 +3150,10 @@ func (a *adapter) TenantCallFinish(id t.TenantID, callID string, seq int64, stat _, err := a.db.ExecContext(ctx, `UPDATE im_call_records SET status=?,end_reason=?,duration_ms=?,finished_at=?,updated_at=? - WHERE call_id=? AND seq_id=?`, - status, status, durationMS, finishedAt, finishedAt, callID, seq) + WHERE call_id=? AND seq_id=? AND finished_at IS NULL + AND NOT (status=? AND ? IN (?,?))`, + status, status, durationMS, finishedAt, finishedAt, callID, seq, + t.CallStatusAccepted, status, t.CallStatusMissed, t.CallStatusDeclined) return err } @@ -4074,7 +4082,7 @@ func tenantMessageGetOne(q sqlx.QueryerContext, ctx context.Context, tenantID t. RecalledBy sql.NullInt64 `db:"recalled_by"` }{} err := sqlx.GetContext(ctx, q, &row, - "SELECT id,createdat,updatedat,deletedat,delid,seqid,topic,`from`,IFNULL(client_msg_id,'') AS client_msg_id,version,recalled_at,recalled_by,head,content "+ + "SELECT id,createdat,updatedat,deletedat,delid,seqid,topic,`from`,IFNULL(client_msg_id,'') AS client_msg_id,version,recalled_at,recalled_by,head,content,content_nonce,key_version "+ "FROM im_messages WHERE topic=? AND seqid=?"+suffix, topic, seq) if err != nil { @@ -4119,7 +4127,7 @@ func tenantGroupMessageGetOneWithTable(q sqlx.QueryerContext, ctx context.Contex t.Message RecalledBy sql.NullInt64 `db:"recalled_by"` }{} - query := fmt.Sprintf("SELECT %s AS id,created_at AS createdat,updated_at AS updatedat,deleted_at AS deletedat,0 AS delid,seqid,topic,from_user AS `from`,IFNULL(client_msg_id,'') AS client_msg_id,version,recalled_at,recalled_by,head,content,'%s' AS shard_month FROM %s WHERE topic=? AND seqid=?%s", idColumn, month, table, suffix) + query := fmt.Sprintf("SELECT %s AS id,created_at AS createdat,updated_at AS updatedat,deleted_at AS deletedat,0 AS delid,seqid,topic,from_user AS `from`,IFNULL(client_msg_id,'') AS client_msg_id,version,recalled_at,recalled_by,head,content,content_nonce,key_version,'%s' AS shard_month FROM %s WHERE topic=? AND seqid=?%s", idColumn, month, table, suffix) err = sqlx.GetContext(ctx, q, &row, query, topic, seq) if err == nil { msg := normalizeGroupMessage(tenantID, row.Message, row.RecalledBy) @@ -4139,6 +4147,13 @@ func messageVersion(version int) int { return version } +func messageStoredContent(value any) (content any, nonce []byte, keyVersion int) { + if stored, ok := value.(t.StoredMessageContent); ok { + return stored.Content, stored.Nonce, stored.KeyVersion + } + return value, nil, 0 +} + func kvMapFromJSON(value any) t.KVMap { switch parsed := common.FromJSON(value).(type) { case map[string]any: diff --git a/server/im_agora_http.go b/server/im_agora_http.go index ca3e8c4..f4e660a 100644 --- a/server/im_agora_http.go +++ b/server/im_agora_http.go @@ -318,6 +318,7 @@ func acceptIMCallFromRTCJoin(principal *imPrincipal, record *types.CallRecord) { } return } + publishRealtimeCallEvent(msg) select { case globals.hub.routeCli <- msg: default: diff --git a/server/im_http.go b/server/im_http.go index 7a089a9..75b98aa 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -26,6 +26,7 @@ import ( "github.com/tinode/chat/server/logs" "github.com/tinode/chat/server/media" "github.com/tinode/chat/server/moderation" + realnamealiyun "github.com/tinode/chat/server/realname/aliyun" "github.com/tinode/chat/server/store" "github.com/tinode/chat/server/store/types" "golang.org/x/crypto/bcrypt" @@ -1691,17 +1692,20 @@ func serveIMRealNameSubmissions(wrt http.ResponseWriter, req *http.Request) { return } submissionID := hashIMSecret(principal.TenantID, principal.Uid.String(), idempotencyKey, "realname")[:32] + status, reviewReason, verifiedAt := initialRealNameSubmissionState(req, principal, name, idType, idNumber) rv := &types.RealNameVerification{ TenantID: principal.TenantID, UserID: principal.Uid, SubmissionID: submissionID, - Status: "pending", + Status: status, NameCipher: protectIMSensitive(principal.TenantID, name), IDType: idType, IDNumberCipher: protectIMSensitive(principal.TenantID, idNumber), IDNumberHash: hashIMSecret(principal.TenantID, idType, idNumber, "realname-id"), MaskedName: maskIMName(name), MaskedIDNumber: maskIMIDNumber(idNumber), + ReviewReason: reviewReason, + VerifiedAt: verifiedAt, } if err := store.Security.CreateRealNameSubmission(principal.TenantID, rv); err != nil { if errors.Is(err, types.ErrDuplicate) { @@ -1711,7 +1715,12 @@ func serveIMRealNameSubmissions(wrt http.ResponseWriter, req *http.Request) { writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) return } - writeIMSuccess(wrt, req, map[string]any{"submission_id": submissionID, "status": "pending"}) + response := map[string]any{"submission_id": submissionID, "status": status} + if status == "manual_review" { + response["status"] = "pending" + response["review_status"] = "manual_review" + } + writeIMSuccess(wrt, req, response) } func serveIMRealNameAppeals(wrt http.ResponseWriter, req *http.Request) { @@ -2563,7 +2572,8 @@ func serveIMFavoriteForward(wrt http.ResponseWriter, req *http.Request) { results = append(results, map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "failed", "reason": relationErrorReason(err)}) continue } - head, content := favoriteForwardMessageHeadContent(item) + clientMid := forwardClientMid(req.Header.Get("Idempotency-Key"), "favorite", target, favoriteID) + head, content := favoriteForwardMessageHeadContent(item, clientMid) msg := &types.Message{ TenantID: principal.TenantID, Topic: target, @@ -2581,6 +2591,7 @@ func serveIMFavoriteForward(wrt http.ResponseWriter, req *http.Request) { if saved != nil && saved.Pending { result := forwardResultUnknownPayload(target, saved.SeqId, relationErrorReason(err)) result["favorite_id"] = favoriteID + result["client_mid"] = clientMid mergeForwardCommentResult(result, commentResult) results = append(results, result) continue @@ -2588,6 +2599,7 @@ func serveIMFavoriteForward(wrt http.ResponseWriter, req *http.Request) { if msg.SeqId > 0 && relationErrorReason(err) == imReasonInternal { result := forwardResultUnknownPayload(target, msg.SeqId, relationErrorReason(err)) result["favorite_id"] = favoriteID + result["client_mid"] = clientMid mergeForwardCommentResult(result, commentResult) results = append(results, result) continue @@ -2597,11 +2609,12 @@ func serveIMFavoriteForward(wrt http.ResponseWriter, req *http.Request) { } if saved != nil && saved.Pending { result := map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "pending", "seq": saved.SeqId, "reason": "MESSAGE_PERSIST_PENDING"} + result["client_mid"] = clientMid mergeForwardCommentResult(result, commentResult) results = append(results, result) continue } - result := map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "sent", "seq": saved.SeqId} + result := map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "sent", "seq": saved.SeqId, "client_mid": clientMid} mergeForwardCommentResult(result, commentResult) results = append(results, result) } @@ -6571,13 +6584,20 @@ func hashIMActionToken(tenantID types.TenantID, uid types.Uid, action, target, t func realNamePayload(rv *types.RealNameVerification) map[string]any { canSubmit := rv.Status == "unverified" || rv.Status == "rejected" + status := rv.Status + if status == "manual_review" { + status = "pending" + } data := map[string]any{ - "status": rv.Status, + "status": status, "submission_id": rv.SubmissionID, "masked_name": rv.MaskedName, "masked_id_number": rv.MaskedIDNumber, "can_submit": canSubmit, } + if rv.Status == "manual_review" { + data["review_status"] = rv.Status + } if rv.VerifiedAt != nil { data["verified_at"] = rv.VerifiedAt.Format(time.RFC3339Nano) } @@ -6604,6 +6624,89 @@ func realNameAppealPayload(appeal *types.RealNameAppeal) map[string]any { } } +func initialRealNameSubmissionState(req *http.Request, principal *imPrincipal, name, idType, idNumber string) (string, *string, *time.Time) { + status := "pending" + policy, err := store.RealNameConfig.Policy(principal.TenantID) + if err != nil || policy == nil { + logs.Warn.Printf("realname policy lookup failed: tenant=%d uid=%s err=%v", principal.TenantID, principal.Uid, err) + return status, nil, nil + } + if policy.ReviewPolicy != types.RealNameReviewPolicyAutoManual { + return status, nil, nil + } + result := reviewIMRealNameAutomatically(req, principal, policy, name, idType, idNumber) + switch result.status { + case "verified": + now := time.Now().UTC().Round(time.Millisecond) + return "verified", nil, &now + default: + reason := strings.TrimSpace(result.reason) + if reason == "" { + reason = "自动审核未通过，已转人工审核" + } + return "manual_review", &reason, nil + } +} + +type imRealNameAutoReviewResult struct { + status string + reason string +} + +func reviewIMRealNameAutomatically(req *http.Request, principal *imPrincipal, policy *types.RealNamePolicy, name, idType, idNumber string) imRealNameAutoReviewResult { + values, err := store.ThirdPartyConfig.Get("im_realname", + "realname_provider", + "aliyun_region_id", + "aliyun_endpoint", + "aliyun_access_key_id", + "aliyun_access_key_secret", + "aliyun_timeout_ms", + "mock_enabled", + "mock_result") + if err != nil { + logs.Warn.Printf("realname auto review config lookup failed: tenant=%d uid=%s err=%v", principal.TenantID, principal.Uid, err) + return imRealNameAutoReviewResult{status: "manual_review", reason: "自动审核配置读取失败，已转人工审核"} + } + if parseIMBool(values["mock_enabled"]) { + switch strings.ToLower(strings.TrimSpace(values["mock_result"])) { + case "verified", "pass", "passed", "success": + return imRealNameAutoReviewResult{status: "verified"} + default: + return imRealNameAutoReviewResult{status: "manual_review", reason: "自动审核未通过，已转人工审核"} + } + } + if strings.TrimSpace(values["realname_provider"]) == "" || + strings.TrimSpace(values["aliyun_access_key_id"]) == "" || + strings.TrimSpace(values["aliyun_access_key_secret"]) == "" { + return imRealNameAutoReviewResult{status: "manual_review", reason: "自动审核配置未完成，已转人工审核"} + } + if !strings.EqualFold(strings.TrimSpace(values["realname_provider"]), "aliyun_cloudauth") { + return imRealNameAutoReviewResult{status: "manual_review", reason: "自动审核供应商暂不支持，已转人工审核"} + } + client, err := realnamealiyun.NewID2MetaClient(realnamealiyun.Config{ + AccessKeyID: values["aliyun_access_key_id"], + AccessKeySecret: values["aliyun_access_key_secret"], + RegionID: values["aliyun_region_id"], + Endpoint: values["aliyun_endpoint"], + TimeoutMS: realnamealiyun.ParseTimeoutMS(values["aliyun_timeout_ms"], 1500), + }) + if err != nil { + logs.Warn.Printf("realname auto review client init failed: tenant=%d uid=%s err=%v", principal.TenantID, principal.Uid, err) + return imRealNameAutoReviewResult{status: "manual_review", reason: "自动审核客户端初始化失败，已转人工审核"} + } + result, err := client.Verify(name, idNumber) + if err != nil { + logs.Warn.Printf("realname auto review call failed: tenant=%d uid=%s trace=%s err=%v", principal.TenantID, principal.Uid, req.Header.Get("X-Request-ID"), err) + return imRealNameAutoReviewResult{status: "manual_review", reason: "自动审核调用失败，已转人工审核"} + } + if result.Passed { + logs.Info.Printf("realname auto review passed: tenant=%d uid=%s request_id=%s latency=%s", principal.TenantID, principal.Uid, result.RequestID, result.Latency) + return imRealNameAutoReviewResult{status: "verified"} + } + logs.Info.Printf("realname auto review mismatch: tenant=%d uid=%s request_id=%s biz_code=%s latency=%s", principal.TenantID, principal.Uid, result.RequestID, result.BizCode, result.Latency) + return imRealNameAutoReviewResult{status: "manual_review", reason: "自动审核未通过，已转人工审核"} +} + func groupInvitationPayload(inv *types.GroupInvitation) map[string]any { if inv == nil { return map[string]any{} @@ -6770,7 +6873,7 @@ func parseIMNonNegativeCursor(raw string) (int, error) { func parseIMBool(raw string) bool { switch strings.ToLower(strings.TrimSpace(raw)) { - case "1", "true", "yes", "y": + case "1", "true", "yes", "y", "on", "enabled": return true default: return false @@ -9038,7 +9141,7 @@ func favoriteForwardContent(item *types.FavoriteItem, comment string) map[string return payload } -func favoriteForwardMessageHeadContent(item *types.FavoriteItem) (types.KVMap, any) { +func favoriteForwardMessageHeadContent(item *types.FavoriteItem, clientMid string) (types.KVMap, any) { itemType := "text" if item != nil && strings.TrimSpace(item.ItemType) != "" { itemType = strings.TrimSpace(item.ItemType) @@ -9059,12 +9162,13 @@ func favoriteForwardMessageHeadContent(item *types.FavoriteItem) (types.KVMap, a if _, ok := head["type"]; !ok { head["type"] = itemType } + head["x-im-client-mid"] = clientMid if content, ok := snapshot["content"]; ok { return head, content } } } - return forwardBundleMessageHead(""), favoriteForwardContent(item, "") + return forwardBundleMessageHead(clientMid), favoriteForwardContent(item, "") } func forwardFailurePayload(topic string, seq int64, target, reason string) map[string]any { diff --git a/server/im_http_test.go b/server/im_http_test.go index e98efd6..c1850b1 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -226,6 +226,24 @@ func (s testThirdPartyConfigStore) Get(_ string, keys ...string) (map[string]str return result, nil } +type testRealNameConfigStore struct { + policy *types.RealNamePolicy + err error +} + +func (s testRealNameConfigStore) Policy(tenantID types.TenantID) (*types.RealNamePolicy, error) { + if s.err != nil { + return nil, s.err + } + if s.policy == nil { + return &types.RealNamePolicy{TenantID: tenantID, ReviewPolicy: types.RealNameReviewPolicyNone, ProviderMode: "id_card_2factor", AllowSkip: true, RemindAfter: "24h"}, nil + } + policy := *s.policy + policy.TenantID = tenantID + policy.Normalize() + return &policy, nil +} + type testConversationsStore struct { topic string settings *types.ConversationSettings @@ -2096,6 +2114,48 @@ func TestServeIMRealNameSubmissionCreatesPending(t *testing.T) { } } +func TestServeIMRealNameSubmissionAutoReviewFallsBackToManual(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + expectIMBearer(t, ctrl, "token-1", 7, 42) + oldSecurity := store.Security + oldRealNameConfig := store.RealNameConfig + oldThirdParty := store.ThirdPartyConfig + security := &testSecurityStore{} + store.Security = security + store.RealNameConfig = testRealNameConfigStore{policy: &types.RealNamePolicy{ + ReviewPolicy: types.RealNameReviewPolicyAutoManual, + ProviderMode: "id_card_2factor", + AllowSkip: true, + RemindAfter: "24h", + }} + store.ThirdPartyConfig = testThirdPartyConfigStore{} + defer func() { + store.Security = oldSecurity + store.RealNameConfig = oldRealNameConfig + store.ThirdPartyConfig = oldThirdParty + }() + + req := httptest.NewRequest(http.MethodPost, "/v1/im/security/realname/submissions", + bytes.NewBufferString(`{"name":"张三","id_type":"cn_id","id_number":"110101199003071234"}`)) + req.Header.Set("Authorization", "Bearer token-1") + req.Header.Set("Idempotency-Key", "idem-1") + rec := httptest.NewRecorder() + + serveIMRealNameSubmissions(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String()) + } + if security.realName == nil || security.realName.Status != "manual_review" || security.realName.ReviewReason == nil { + t.Fatalf("unexpected stored realname: %#v", security.realName) + } + data := decodeIMHTTPResponse(t, rec).Data.(map[string]any) + if data["status"] != "pending" || data["review_status"] != "manual_review" { + t.Fatalf("unexpected response: %#v", data) + } +} + func TestServeIMRealNameAppealCreatesPending(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -5087,6 +5147,7 @@ func TestServeIMFavoriteForwardSavesMessages(t *testing.T) { tokenAuth.EXPECT().IsInitialized().Return(true) tokenAuth.EXPECT().Authenticate(auth.AuthContext{}, []byte("token-1")). Return(&auth.Rec{TenantID: 7, Uid: 42, AuthLevel: auth.LevelAuth}, nil, nil) + mockStore.EXPECT().GetUidString().Return("fav-mid") mockStore.EXPECT().GetUidString().Return("comment-mid") store.Store = mockStore oldFavorites := store.Favorites @@ -5137,6 +5198,9 @@ func TestServeIMFavoriteForwardSavesMessages(t *testing.T) { if msg.Head["x-im-forwarded"] != true { t.Fatalf("forward marker not set: %#v", msg.Head) } + if msg.Head["x-im-client-mid"] != "fwd_fav-mid" { + t.Fatalf("forward client mid not set: %#v", msg.Head) + } content := msg.Content.(map[string]any) if content["text"] != "hello" || content["comment"] != nil || content["snapshot"] != nil { t.Fatalf("unexpected forward content: %#v", content) @@ -5153,11 +5217,82 @@ func TestServeIMFavoriteForwardSavesMessages(t *testing.T) { t.Fatalf("forward status: got %d body=%s", rec.Code, rec.Body.String()) } items := decodeIMHTTPResponse(t, rec).Data.(map[string]any)["items"].([]any) - if got := items[0].(map[string]any); got["status"] != "sent" || got["seq"].(float64) != 11 || got["comment_seq"].(float64) != 10 { + if got := items[0].(map[string]any); got["status"] != "sent" || got["seq"].(float64) != 11 || got["comment_seq"].(float64) != 10 || got["client_mid"] != "fwd_fav-mid" { t.Fatalf("unexpected forward response: %#v", got) } } +func TestServeIMFavoriteForwardMediaUsesNewClientMid(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + oldStore := store.Store + mockStore := mock_store.NewMockPersistentStorageInterface(ctrl) + tokenAuth := mock_auth.NewMockAuthHandler(ctrl) + mockStore.EXPECT().GetLogicalAuthHandler("token").Return(tokenAuth) + tokenAuth.EXPECT().IsInitialized().Return(true) + tokenAuth.EXPECT().Authenticate(auth.AuthContext{}, []byte("token-1")). + Return(&auth.Rec{TenantID: 7, Uid: 42, AuthLevel: auth.LevelAuth}, nil, nil) + mockStore.EXPECT().GetUidString().Return("media-mid") + store.Store = mockStore + oldFavorites := store.Favorites + oldSubs := store.Subs + oldMessages := store.Messages + now := time.Date(2026, 8, 13, 9, 0, 0, 0, time.UTC) + store.Favorites = &testFavoritesStore{items: []types.FavoriteItem{{ + TenantID: 7, + FavoriteID: "fav_img", + UserID: 42, + ItemType: "image", + SourceTopic: "grpAlpha", + SourceSeq: 17, + Snapshot: map[string]any{ + "head": map[string]any{"type": "image", "x-im-type": "image", "x-im-client-mid": "origin-mid"}, + "content": map[string]any{"file_id": "file-img", "mime": "image/png"}, + }, + Availability: "available", + CollectedAt: now, + }}} + subs := mock_store.NewMockSubsPersistenceInterface(ctrl) + messages := mock_store.NewMockMessagesPersistenceInterface(ctrl) + store.Subs = subs + store.Messages = messages + defer func() { + store.Store = oldStore + store.Favorites = oldFavorites + store.Subs = oldSubs + store.Messages = oldMessages + }() + subs.EXPECT().Get(types.TenantID(7), "grpBeta", types.Uid(42), false).Return(&types.Subscription{ + TenantID: 7, User: types.Uid(42).String(), Topic: "grpBeta", ModeWant: types.ModeCPublic, ModeGiven: types.ModeCPublic, + }, nil) + messages.EXPECT().Save(types.TenantID(7), gomock.Any(), gomock.Nil(), true).DoAndReturn( + func(_ types.TenantID, msg *types.Message, _ []string, _ bool) (*types.MessageSaveResult, error) { + if msg.Topic != "grpBeta" || msg.Head["x-im-type"] != "image" { + t.Fatalf("unexpected media forward message: %#v", msg) + } + if msg.Head["x-im-client-mid"] != "fwd_media-mid" { + t.Fatalf("media forward reused original client mid: %#v", msg.Head) + } + content := msg.Content.(map[string]any) + if content["file_id"] != "file-img" || content["mime"] != "image/png" { + t.Fatalf("unexpected media forward content: %#v", content) + } + return &types.MessageSaveResult{SeqId: 22, MarkedReadBySender: true}, nil + }) + + req := httptest.NewRequest(http.MethodPost, "/v1/im/favorites/forward", bytes.NewBufferString(`{"favorite_ids":["fav_img"],"target_topics":["grpBeta"]}`)) + req.Header.Set("Authorization", "Bearer token-1") + rec := httptest.NewRecorder() + serveIMFavoriteResource(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("forward status: got %d body=%s", rec.Code, rec.Body.String()) + } + items := decodeIMHTTPResponse(t, rec).Data.(map[string]any)["items"].([]any) + if got := items[0].(map[string]any); got["status"] != "sent" || got["seq"].(float64) != 22 || got["client_mid"] != "fwd_media-mid" { + t.Fatalf("unexpected media forward response: %#v", got) + } +} + func TestServeIMFavoriteForwardRejectsUnauthorizedTarget(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -5511,7 +5646,7 @@ func TestServeIMMessageForwardKafkaAssignsSeqAndBroadcasts(t *testing.T) { t.Fatalf("unexpected forward response: %#v", data) } got := items[0].(map[string]any) - if got["target_topic"] != targetTopic || got["seq"].(float64) != 35 { + if got["target_topic"] != targetTopic || got["seq"].(float64) != 35 || got["status"] != "pending" || got["reason"] != "MESSAGE_PERSIST_PENDING" { t.Fatalf("unexpected forward item: %#v", got) } if failed := data["failed"].([]any); len(failed) != 0 { @@ -5525,29 +5660,8 @@ func TestServeIMMessageForwardKafkaAssignsSeqAndBroadcasts(t *testing.T) { } select { case delivered := <-globals.hub.routeSrv: - if delivered.RcptTo != targetTopic || delivered.Data == nil || delivered.Data.SeqId != 35 { - t.Fatalf("unexpected broadcast message: %#v", delivered) - } + t.Fatalf("pending forward should not broadcast immediately: %#v", delivered) default: - t.Fatal("expected forwarded message to be broadcast") - } - presByUser := make(map[string]*MsgServerPres) - for range 2 { - select { - case delivered := <-globals.hub.routeSrv: - if delivered.Pres == nil { - t.Fatalf("unexpected non-presence message: %#v", delivered) - } - presByUser[delivered.RcptTo] = delivered.Pres - default: - t.Fatal("expected forwarded message presence notifications") - } - } - if pres := presByUser[types.Uid(42).UserId()]; pres == nil || pres.Topic != "me" || pres.What != "msg" || pres.Src != targetUser.UserId() || pres.SeqId != 35 || pres.AcsActor != "" || pres.SkipTopic != targetTopic { - t.Fatalf("unexpected sender presence: %#v", pres) - } - if pres := presByUser[targetUser.UserId()]; pres == nil || pres.Topic != "me" || pres.What != "msg" || pres.Src != types.Uid(42).UserId() || pres.SeqId != 35 || pres.AcsActor != types.Uid(42).UserId() || pres.SkipTopic != targetTopic { - t.Fatalf("unexpected receiver presence: %#v", pres) } } diff --git a/server/internal_admin_query_resources.go b/server/internal_admin_query_resources.go index dc0a9d4..b9c47d3 100644 --- a/server/internal_admin_query_resources.go +++ b/server/internal_admin_query_resources.go @@ -1,6 +1,7 @@ package main import ( + "encoding/json" "fmt" "net/http" "strconv" @@ -127,11 +128,23 @@ func serveInternalAdminResourceList(wrt http.ResponseWriter, req *http.Request, writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil) return } - result, err := store.InternalAdminQueries.Query(tenantID, query, countQuery, args, page, perPage) + var result *types.InternalAdminQueryResult + keyword := strings.TrimSpace(req.URL.Query().Get("keyword")) + if (resource == "private-messages" || resource == "group-messages") && keyword != "" { + result, err = queryEncryptedAdminMessages(req, tenantID, spec, scope, keyword, page, perPage) + } else { + result, err = store.InternalAdminQueries.Query(tenantID, query, countQuery, args, page, perPage) + } if err != nil { writeInternalQueryError(wrt, req, err) return } + if resource == "private-messages" || resource == "group-messages" { + if err = decryptInternalAdminMessageRows(tenantID, result.Items); err != nil { + writeInternalQueryError(wrt, req, err) + return + } + } writeIMSuccess(wrt, req, map[string]any{ "items": normalizeInternalAdminRowsForAPI(result.Items), "total": result.Total, @@ -140,6 +153,81 @@ func serveInternalAdminResourceList(wrt http.ResponseWriter, req *http.Request, }) } +func queryEncryptedAdminMessages(req *http.Request, tenantID types.TenantID, spec *internalQuerySpec, scope, keyword string, page, perPage int) (*types.InternalAdminQueryResult, error) { + requestCopy := req.Clone(req.Context()) + urlCopy := *req.URL + queryValues := req.URL.Query() + queryValues.Del("keyword") + urlCopy.RawQuery = queryValues.Encode() + requestCopy.URL = &urlCopy + query, countQuery, args, _, _, err := buildInternalAdminListQuery(requestCopy, spec, scope) + if err != nil { + return nil, err + } + + const batchSize = 200 + start := (page - 1) * perPage + end := start + perPage + items := make([]map[string]any, 0, perPage) + matched := 0 + for batchPage := 1; ; batchPage++ { + batch, queryErr := store.InternalAdminQueries.Query(tenantID, query, countQuery, args, batchPage, batchSize) + if queryErr != nil { + return nil, queryErr + } + if err = decryptInternalAdminMessageRows(tenantID, batch.Items); err != nil { + return nil, err + } + for _, row := range batch.Items { + if !strings.Contains(strings.ToLower(fmt.Sprint(row["content_excerpt"])), strings.ToLower(keyword)) { + continue + } + if matched >= start && matched < end { + items = append(items, row) + } + matched++ + } + if len(batch.Items) < batchSize || int64(batchPage*batchSize) >= batch.Total { + break + } + } + return &types.InternalAdminQueryResult{Items: items, Total: int64(matched), Page: page, PerPage: perPage}, nil +} + +func decryptInternalAdminMessageRows(tenantID types.TenantID, rows []map[string]any) error { + for _, row := range rows { + seq, ok := numericID(row["seq_id"]) + if !ok || seq <= 0 { + continue + } + msg, err := store.Messages.GetOne(tenantID, fmt.Sprint(row["topic"]), seq) + if err != nil { + return err + } + row["content_excerpt"] = internalMessageExcerpt(msg.Content) + } + return nil +} + +func internalMessageExcerpt(content any) string { + if value, ok := content.(map[string]any); ok { + for _, key := range []string{"txt", "text", "content", "title"} { + if text := strings.TrimSpace(fmt.Sprint(value[key])); text != "" && text != "" { + if len([]rune(text)) > 500 { + return string([]rune(text)[:500]) + } + return text + } + } + } + raw, _ := json.Marshal(content) + text := string(raw) + if len([]rune(text)) > 500 { + text = string([]rune(text)[:500]) + } + return text +} + func serveInternalAdminResourceDetail(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, resource, id string) { spec, ok := internalAdminQuerySpec(resource) if !ok || spec.DetailKey == "" { diff --git a/server/internal_api.go b/server/internal_api.go index ac3d497..2f5d231 100644 --- a/server/internal_api.go +++ b/server/internal_api.go @@ -607,6 +607,13 @@ func serveInternalMessageQuery(wrt http.ResponseWriter, req *http.Request, tenan writeInternalQueryError(wrt, req, err) return } + msg, err := store.Messages.GetOne(tenantID, topic, seq) + if err != nil { + writeInternalQueryError(wrt, req, err) + return + } + row["content"] = msg.Content + row["content_excerpt"] = internalMessageExcerpt(msg.Content) row["conversation_type"] = strings.TrimSuffix(resource, "-messages") writeIMSuccess(wrt, req, normalizeInternalAdminRowForAPI(row)) } diff --git a/server/kafka_config_events.go b/server/kafka_config_events.go index 1969c10..6ac7fa3 100644 --- a/server/kafka_config_events.go +++ b/server/kafka_config_events.go @@ -23,6 +23,8 @@ const ( configEventAppReleaseUpdated = "app.release.updated" configEventSecurityPolicyUpdated = "security.policy.updated" configEventSensitiveWordsUpdated = "sensitive_words.updated" + configEventMessageKeysUpdated = "message.keys.updated" + configEventMessageKeysActivated = "message.keys.activated" ) type configEventPayload struct { @@ -102,6 +104,13 @@ func executeKafkaConfigEvent(envelope kafkaEnvelope) error { store.InvalidateSecurityPolicyCache(tenantCode) case configEventSensitiveWordsUpdated: store.InvalidateModerationConfigCache(tenantCode, false, false, true) + case configEventMessageKeysUpdated, configEventMessageKeysActivated: + store.InvalidateMessageEncryptionKeys(tenantID) + if strings.TrimSpace(envelope.EventType) == configEventMessageKeysUpdated && envelope.Version > 0 { + if err := store.PrepareMessageEncryptionKey(tenantID, tenantCode, int(envelope.Version)); err != nil { + return err + } + } default: store.InvalidateTenantConfigCache(tenantCode, tenantID) store.InvalidateThirdPartyConfigCache(payload.Group) diff --git a/server/kafka_message_persist.go b/server/kafka_message_persist.go index 4cf15a1..fc11970 100644 --- a/server/kafka_message_persist.go +++ b/server/kafka_message_persist.go @@ -54,10 +54,10 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty if runtimeReady && !claimed { if claim != nil && claim.Seq > 0 { msg.SeqId = claim.Seq - return &types.MessageSaveResult{SeqId: claim.Seq, Duplicate: claim.Status == "done", Pending: claim.Status == "pending"}, nil + return &types.MessageSaveResult{SeqId: claim.Seq, Duplicate: true, Pending: claim.Status == "pending"}, nil } if claim != nil && claim.Status == "pending" { - return &types.MessageSaveResult{Pending: true}, types.ErrDuplicate + return &types.MessageSaveResult{Duplicate: true, Pending: true}, types.ErrDuplicate } return nil, types.ErrDuplicate } @@ -87,8 +87,8 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty return &types.MessageSaveResult{SeqId: msg.SeqId, Pending: true, MarkedReadBySender: readBySender && !fromUid.IsZero()}, err } if clientMsgID != "" { - if err = store.MarkMessageUniqueDone(tenantID, msg.Topic, fromUid, clientMsgID, msg.SeqId, ""); err != nil { - logs.Warn.Printf("kafka message persist: failed to mark message unique key done topic=%s seq=%d: %v", msg.Topic, msg.SeqId, err) + if err = store.MarkMessageUniquePending(tenantID, msg.Topic, fromUid, clientMsgID, msg.SeqId); err != nil { + logs.Warn.Printf("kafka message persist: failed to mark message unique key pending topic=%s seq=%d: %v", msg.Topic, msg.SeqId, err) } } if isGroupMessage { @@ -96,7 +96,7 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty logs.Warn.Printf("kafka message persist: failed to cache pending hot message after publish topic=%s seq=%d: %v", msg.Topic, msg.SeqId, cacheErr) } } - return &types.MessageSaveResult{SeqId: msg.SeqId, MarkedReadBySender: readBySender && !fromUid.IsZero()}, nil + return &types.MessageSaveResult{SeqId: msg.SeqId, Pending: true, MarkedReadBySender: readBySender && !fromUid.IsZero()}, nil } func enqueueKafkaMessagePersist(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) error { diff --git a/server/realname/aliyun/id2meta.go b/server/realname/aliyun/id2meta.go new file mode 100644 index 0000000..11a8c8b --- /dev/null +++ b/server/realname/aliyun/id2meta.go @@ -0,0 +1,112 @@ +package aliyun + +import ( + "errors" + "strconv" + "strings" + "time" + + cloudauth "github.com/alibabacloud-go/cloudauth-20190307/v3/client" + openapi "github.com/alibabacloud-go/darabonba-openapi/v2/client" + util "github.com/alibabacloud-go/tea-utils/v2/service" + "github.com/alibabacloud-go/tea/dara" +) + +type Config struct { + AccessKeyID string + AccessKeySecret string + RegionID string + Endpoint string + TimeoutMS int +} + +type ID2MetaResult struct { + Passed bool + BizCode string + RequestID string + ProviderCode string + ProviderMessage string + ProviderStatusCode int32 + Latency time.Duration +} + +type ID2MetaClient struct { + client *cloudauth.Client + timeoutMS int +} + +func NewID2MetaClient(config Config) (*ID2MetaClient, error) { + if strings.TrimSpace(config.AccessKeyID) == "" || strings.TrimSpace(config.AccessKeySecret) == "" { + return nil, errors.New("missing aliyun realname credentials") + } + if strings.TrimSpace(config.RegionID) == "" { + config.RegionID = "cn-hangzhou" + } + clientConfig := &openapi.Config{ + AccessKeyId: dara.String(strings.TrimSpace(config.AccessKeyID)), + AccessKeySecret: dara.String(strings.TrimSpace(config.AccessKeySecret)), + RegionId: dara.String(strings.TrimSpace(config.RegionID)), + Protocol: dara.String("https"), + } + if endpoint := strings.TrimSpace(config.Endpoint); endpoint != "" { + clientConfig.Endpoint = dara.String(endpoint) + } + client, err := cloudauth.NewClient(clientConfig) + if err != nil { + return nil, err + } + if config.TimeoutMS <= 0 { + config.TimeoutMS = 1500 + } + return &ID2MetaClient{client: client, timeoutMS: config.TimeoutMS}, nil +} + +func (c *ID2MetaClient) Verify(name, idNumber string) (*ID2MetaResult, error) { + if c == nil || c.client == nil { + return nil, errors.New("nil aliyun realname client") + } + name = strings.TrimSpace(name) + idNumber = strings.TrimSpace(idNumber) + if name == "" || idNumber == "" { + return nil, errors.New("missing realname identity fields") + } + started := time.Now() + resp, err := c.client.Id2MetaVerifyWithOptions(&cloudauth.Id2MetaVerifyRequest{ + UserName: dara.String(name), + IdentifyNum: dara.String(idNumber), + ParamType: dara.String("normal"), + }, (&util.RuntimeOptions{}).SetConnectTimeout(c.timeoutMS).SetReadTimeout(c.timeoutMS).SetAutoretry(false)) + latency := time.Since(started) + if err != nil { + return nil, err + } + result := &ID2MetaResult{Latency: latency} + if resp != nil { + result.ProviderStatusCode = dara.Int32Value(resp.StatusCode) + } + if resp == nil || resp.Body == nil { + return result, errors.New("empty aliyun realname response") + } + result.RequestID = dara.StringValue(resp.Body.RequestId) + result.ProviderCode = dara.StringValue(resp.Body.Code) + result.ProviderMessage = dara.StringValue(resp.Body.Message) + if resp.Body.ResultObject != nil { + result.BizCode = strings.TrimSpace(dara.StringValue(resp.Body.ResultObject.BizCode)) + } + if result.ProviderCode != "" && result.ProviderCode != "200" { + return result, errors.New("aliyun realname response code " + result.ProviderCode) + } + if result.BizCode == "" { + return result, errors.New("empty aliyun realname biz code") + } + result.Passed = result.BizCode == "1" + return result, nil +} + +func ParseTimeoutMS(value string, fallback int) int { + parsed, err := strconv.Atoi(strings.TrimSpace(value)) + if err != nil || parsed <= 0 { + return fallback + } + return parsed +} diff --git a/server/realname/aliyun/id2meta_live_test.go b/server/realname/aliyun/id2meta_live_test.go new file mode 100644 index 0000000..0dec1e7 --- /dev/null +++ b/server/realname/aliyun/id2meta_live_test.go @@ -0,0 +1,36 @@ +package aliyun + +import ( + "os" + "testing" +) + +func TestLiveID2MetaVerify(t *testing.T) { + accessKeyID := os.Getenv("ALIYUN_REALNAME_ACCESS_KEY_ID") + accessKeySecret := os.Getenv("ALIYUN_REALNAME_ACCESS_KEY_SECRET") + if accessKeyID == "" || accessKeySecret == "" { + t.Skip("ALIYUN_REALNAME_ACCESS_KEY_ID and ALIYUN_REALNAME_ACCESS_KEY_SECRET are required") + } + client, err := NewID2MetaClient(Config{ + AccessKeyID: accessKeyID, + AccessKeySecret: accessKeySecret, + RegionID: envOrDefault("ALIYUN_REALNAME_REGION_ID", "cn-hangzhou"), + Endpoint: os.Getenv("ALIYUN_REALNAME_ENDPOINT"), + TimeoutMS: ParseTimeoutMS(os.Getenv("ALIYUN_REALNAME_TIMEOUT_MS"), 3000), + }) + if err != nil { + t.Fatal(err) + } + result, err := client.Verify(envOrDefault("ALIYUN_REALNAME_TEST_NAME", "张三"), envOrDefault("ALIYUN_REALNAME_TEST_ID_NUMBER", "110101199003071234")) + if err != nil { + t.Fatalf("live id2meta verify failed: %v result=%+v", err, result) + } + t.Logf("live id2meta result: passed=%v biz_code=%s request_id=%s code=%s message=%s", result.Passed, result.BizCode, result.RequestID, result.ProviderCode, result.ProviderMessage) +} + +func envOrDefault(key, fallback string) string { + if value := os.Getenv(key); value != "" { + return value + } + return fallback +} diff --git a/server/realtime_dispatcher.go b/server/realtime_dispatcher.go index 8018333..25d1482 100644 --- a/server/realtime_dispatcher.go +++ b/server/realtime_dispatcher.go @@ -7,6 +7,7 @@ import ( "strings" "time" + "github.com/tinode/chat/server/auth" "github.com/tinode/chat/server/logs" immetrics "github.com/tinode/chat/server/metrics" "github.com/tinode/chat/server/store" @@ -16,6 +17,7 @@ import ( const ( realtimeKindMessage = "message" realtimeKindSessionControl = "session_control" + realtimeKindCallEvent = "call_event" realtimeTargetTopic = "topic" realtimeTargetUser = "user" @@ -23,6 +25,16 @@ const ( realtimeTargetPlatf = "platform" ) +type realtimeCallEventMessage struct { + Topic string `json:"topic"` + From string `json:"from"` + Sid string `json:"sid,omitempty"` + Event string `json:"event"` + Seq int64 `json:"seq"` + Payload json.RawMessage `json:"payload,omitempty"` + Timestamp time.Time `json:"ts,omitempty"` +} + func startRedisRealtimeFanout(ctx context.Context) { if err := store.StartRealtimeFanoutSubscriber(ctx, currentNodeID(), deliverRealtimeFanoutEnvelope); err != nil { immetrics.IncRealtimeFanout("subscribe", realtimeKindMessage, realtimeTargetTopic, "error") @@ -65,6 +77,35 @@ func publishRealtimeUserMessage(uid types.Uid, msg *ServerComMessage) { immetrics.IncRealtimeFanout("publish", realtimeKindMessage, realtimeTargetUser, "ok") } +func publishRealtimeCallEvent(msg *ClientComMessage) { + if msg == nil || msg.Note == nil || !msg.TenantID.IsValid() || + msg.RcptTo == "" || msg.AsUser == "" || msg.Note.What != "call" || + !isDetachedCallEvent(msg.Note.Event) || globals.cluster != nil { + return + } + sid := "" + if msg.sess != nil { + sid = msg.sess.sid + } + event := realtimeCallEventMessage{ + Topic: msg.RcptTo, + From: msg.AsUser, + Sid: sid, + Event: normalizeCallEvent(msg.Note.Event), + Seq: msg.Note.SeqId, + Payload: msg.Note.Payload, + Timestamp: msg.Timestamp, + } + dedupKey := "call:" + msg.RcptTo + ":" + msg.AsUser + ":" + event.Event + ":" + strconv.FormatInt(msg.Note.SeqId, 10) + if err := store.PublishRealtimeFanout(currentNodeID(), msg.TenantID, realtimeKindCallEvent, realtimeTargetTopic, msg.RcptTo, sid, dedupKey, event); err != nil { + immetrics.IncRealtimeFanout("publish", realtimeKindCallEvent, realtimeTargetTopic, "error") + logs.Warn.Printf("redis realtime fanout: publish call event topic=%s from=%s event=%s seq=%d failed: %v", + msg.RcptTo, msg.AsUser, event.Event, msg.Note.SeqId, err) + return + } + immetrics.IncRealtimeFanout("publish", realtimeKindCallEvent, realtimeTargetTopic, "ok") +} + func publishRealtimeEvictUser(tenantID types.TenantID, uid types.Uid, reason string, skipSid string) { if !tenantID.IsValid() || uid.IsZero() { immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetUser, "skip_invalid") @@ -175,12 +216,79 @@ func deliverRealtimeFanoutEnvelope(envelope *store.RealtimeFanoutEnvelope) { switch envelope.Kind { case realtimeKindSessionControl: deliverRealtimeSessionControl(envelope, &msg) + case realtimeKindCallEvent: + deliverRealtimeCallEvent(envelope) default: immetrics.ObserveRealtimeFanoutDelivery(envelope.Kind, envelope.TargetType, msg.Timestamp) deliverRealtimeMessage(envelope, &msg) } } +func deliverRealtimeCallEvent(envelope *store.RealtimeFanoutEnvelope) { + var event realtimeCallEventMessage + if err := json.Unmarshal(envelope.Message, &event); err != nil { + immetrics.IncRealtimeFanout("deliver", realtimeKindCallEvent, realtimeTargetTopic, "malformed") + logs.Warn.Println("redis realtime fanout: malformed call event", err) + return + } + topic := strings.TrimSpace(event.Topic) + if topic == "" { + topic = strings.TrimSpace(envelope.Target) + } + if topic == "" || event.From == "" || event.Seq <= 0 || !isDetachedCallEvent(event.Event) { + immetrics.IncRealtimeFanout("deliver", realtimeKindCallEvent, realtimeTargetTopic, "skip_invalid") + return + } + key := types.TopicKey{TenantID: envelope.TenantID, Topic: topic} + if globals.hub == nil || globals.hub.topicGet(key) == nil { + immetrics.IncRealtimeFanout("deliver", realtimeKindCallEvent, realtimeTargetTopic, "no_local_topic") + return + } + uid := types.ParseUserId(event.From) + if uid.IsZero() { + immetrics.IncRealtimeFanout("deliver", realtimeKindCallEvent, realtimeTargetTopic, "skip_invalid_user") + return + } + ts := event.Timestamp + if ts.IsZero() { + ts = types.TimeNow() + } + msg := &ClientComMessage{ + Note: &MsgClientNote{ + Topic: topic, + What: "call", + Event: normalizeCallEvent(event.Event), + SeqId: event.Seq, + Payload: event.Payload, + }, + Original: topic, + RcptTo: topic, + AsUser: event.From, + AuthLvl: int(auth.LevelAuth), + Timestamp: ts, + TenantID: envelope.TenantID, + sess: &Session{ + proto: NONE, + sid: "redis-call:" + strings.TrimSpace(event.Sid), + uid: uid, + tenantID: envelope.TenantID, + authLvl: auth.LevelAuth, + lastAction: ts.UnixNano(), + }, + init: true, + } + if msg.sess.sid == "redis-call:" { + msg.sess.sid = "redis-call:" + event.From + } + select { + case globals.hub.routeCli <- msg: + immetrics.IncRealtimeFanout("deliver", realtimeKindCallEvent, realtimeTargetTopic, "hub_queued") + default: + immetrics.IncRealtimeFanout("deliver", realtimeKindCallEvent, realtimeTargetTopic, "hub_full") + logs.Err.Println("redis realtime fanout: hub routeCli queue full for call event", topic) + } +} + func deliverRealtimeMessage(envelope *store.RealtimeFanoutEnvelope, msg *ServerComMessage) { if globals.sessionStore == nil { immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_no_session_store") diff --git a/server/realtime_dispatcher_test.go b/server/realtime_dispatcher_test.go index 5996b52..91e0a22 100644 --- a/server/realtime_dispatcher_test.go +++ b/server/realtime_dispatcher_test.go @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "sync" "testing" "time" @@ -259,6 +260,65 @@ func TestRealtimeP2PTopicFallbackRewritesVisibleInfoTopic(t *testing.T) { } } +func TestRealtimeCallEventRoutesToLocalTopic(t *testing.T) { + oldHub := globals.hub + defer func() { globals.hub = oldHub }() + + tenantID := types.TenantID(1) + caller := types.Uid(1001) + callee := types.Uid(1002) + topic := caller.P2PName(callee) + if topic == "" { + t.Fatal("failed to create p2p topic") + } + topics := &sync.Map{} + topics.Store(types.TopicKey{TenantID: tenantID, Topic: topic}, &Topic{name: topic, tenantID: tenantID}) + globals.hub = &Hub{ + topics: topics, + routeCli: make(chan *ClientComMessage, 1), + } + + payload := json.RawMessage(`{"callId":"call-1"}`) + event := realtimeCallEventMessage{ + Topic: topic, + From: callee.UserId(), + Sid: "callee-sid", + Event: constCallEventAccept, + Seq: 12, + Payload: payload, + } + raw, err := json.Marshal(event) + if err != nil { + t.Fatal(err) + } + + deliverRealtimeFanoutEnvelope(&store.RealtimeFanoutEnvelope{ + TenantID: tenantID, + Kind: realtimeKindCallEvent, + TargetType: realtimeTargetTopic, + Target: topic, + Message: raw, + }) + + select { + case got := <-globals.hub.routeCli: + if got.TenantID != tenantID || got.RcptTo != topic || got.AsUser != callee.UserId() { + t.Fatalf("unexpected routed message: tenant=%d rcpt=%q as=%q", got.TenantID, got.RcptTo, got.AsUser) + } + if got.Note == nil || got.Note.What != "call" || got.Note.Event != constCallEventAccept || got.Note.SeqId != 12 { + t.Fatalf("unexpected call note: %#v", got.Note) + } + if string(got.Note.Payload) != string(payload) { + t.Fatalf("payload = %s, want %s", got.Note.Payload, payload) + } + if got.sess == nil || got.sess.uid != callee || got.sess.sid != "redis-call:callee-sid" { + t.Fatalf("unexpected synthetic session: %#v", got.sess) + } + default: + t.Fatal("expected call event to be routed to local topic") + } +} + func TestRealtimeInfoDedupKeySeparatesReadRecv(t *testing.T) { read := realtimeMessageDedupKey("p2pABC", &ServerComMessage{ Info: &MsgServerInfo{From: "usrA", What: "read", SeqId: 12}, diff --git a/server/session.go b/server/session.go index c1383fe..e202d2e 100644 --- a/server/session.go +++ b/server/session.go @@ -1693,6 +1693,9 @@ func (s *Session) note(msg *ClientComMessage) { } return } + if msg.Note.What == "call" { + publishRealtimeCallEvent(msg) + } select { case globals.hub.routeCli <- msg: default: diff --git a/server/store/config_cache.go b/server/store/config_cache.go index 88ab695..24aae3f 100644 --- a/server/store/config_cache.go +++ b/server/store/config_cache.go @@ -65,6 +65,7 @@ var ( moderationPolicyCache = newConfigCache[*types.ModerationPolicy]() moderationWordsCache = newConfigCache[*types.SensitiveWordSet]() securityPolicyCache = newConfigCache[*types.SecurityPolicy]() + realNamePolicyCache = newConfigCache[*types.RealNamePolicy]() thirdPartyConfigAllKeys = "*" ) @@ -145,8 +146,10 @@ func InvalidateTenantConfigCache(code string, id types.TenantID) { } if id.IsValid() { tenantByIDCache.delete(formatTenantIDForCache(id)) + realNamePolicyCache.delete(formatTenantIDForCache(id)) } else if strings.TrimSpace(code) == "" { tenantByIDCache.clear() + realNamePolicyCache.clear() } } diff --git a/server/store/local_message_crypto_integration_test.go b/server/store/local_message_crypto_integration_test.go new file mode 100644 index 0000000..62b3e7f --- /dev/null +++ b/server/store/local_message_crypto_integration_test.go @@ -0,0 +1,191 @@ +//go:build mysql + +package store_test + +import ( + "database/sql" + "encoding/base64" + "encoding/json" + "fmt" + "os" + "strings" + "testing" + "time" + + _ "github.com/go-sql-driver/mysql" + _ "github.com/tinode/chat/server/db/mysql" + "github.com/tinode/chat/server/store" + "github.com/tinode/chat/server/store/types" +) + +type localCryptoProvider struct { + key string +} + +func (p localCryptoProvider) GetTenantByCode(string) (*types.Tenant, error) { + return nil, types.ErrUnsupported +} +func (p localCryptoProvider) GetTenantByID(types.TenantID) (*types.Tenant, error) { + return nil, types.ErrUnsupported +} +func (p localCryptoProvider) ListActiveTenants(int) ([]types.Tenant, error) { + return nil, types.ErrUnsupported +} +func (p localCryptoProvider) RegistrationSMSEnabled() (bool, error) { + return false, types.ErrUnsupported +} +func (p localCryptoProvider) ThirdPartyConfig(string, ...string) (map[string]string, error) { + return nil, types.ErrUnsupported +} +func (p localCryptoProvider) LatestAppRelease(string, string) (*types.AppRelease, error) { + return nil, types.ErrUnsupported +} +func (p localCryptoProvider) ModerationProvider() (*types.ModerationProviderConfig, error) { + return nil, types.ErrUnsupported +} +func (p localCryptoProvider) ModerationPolicy(string) (*types.ModerationPolicy, error) { + return nil, types.ErrUnsupported +} +func (p localCryptoProvider) SensitiveWords(string) (*types.SensitiveWordSet, error) { + return nil, types.ErrUnsupported +} +func (p localCryptoProvider) SecurityPolicy(string) (*types.SecurityPolicy, error) { + return nil, types.ErrUnsupported +} +func (p localCryptoProvider) RealNamePolicy(types.TenantID) (*types.RealNamePolicy, error) { + return nil, types.ErrUnsupported +} +func (p localCryptoProvider) MessageKey(tenantCode string, version int) (*types.MessageKey, error) { + if tenantCode != "default" || (version != 0 && version != 7) { + return nil, types.ErrNotFound + } + return &types.MessageKey{TenantID: 1, TenantCode: tenantCode, Enabled: true, Version: 7, Algorithm: "AES-256-GCM", Status: "active", Key: p.key}, nil +} +func (p localCryptoProvider) MessageKeyAck(string, int) error { return nil } + +func TestLocalMySQLMessageCryptoFlow(t *testing.T) { + if os.Getenv("IM_LOCAL_CRYPTO_TEST") != "1" { + t.Skip("set IM_LOCAL_CRYPTO_TEST=1 to run against local MySQL") + } + config := json.RawMessage(`{ + "uid_key":"dkXxWLwqUNRfYmdXRzOkZw==", + "max_results":1024, + "use_adapter":"mysql", + "adapters":{"mysql":{ + "dsn":"mysql://root:root@tcp(127.0.0.1:3306)/tinode?parseTime=true&collation=utf8mb4_0900_ai_ci", + "tenant_registry_db":"hotgo", + "tenant_db_routing":true, + "sql_timeout":10 + }} + }`) + if err := store.Store.Open(991, config); err != nil { + t.Fatal(err) + } + defer store.Store.Close() + defer store.SetAdminInternalConfigProvider(nil) + store.SetAdminInternalConfigProvider(localCryptoProvider{key: base64.StdEncoding.EncodeToString([]byte("0123456789abcdef0123456789abcdef"))}) + store.InvalidateMessageEncryptionKeys(1) + + db, err := sql.Open("mysql", "root:root@tcp(127.0.0.1:3306)/tinode?parseTime=true&collation=utf8mb4_0900_ai_ci") + if err != nil { + t.Fatal(err) + } + defer db.Close() + + var topic string + var senderID, originalTopicSeq int64 + err = db.QueryRow(`SELECT s.topic,s.userid,t.seqid FROM im_subscriptions s JOIN im_topics t ON t.name=s.topic + WHERE s.topic LIKE 'p2p%' AND s.deletedat IS NULL ORDER BY t.id DESC LIMIT 1`).Scan(&topic, &senderID, &originalTopicSeq) + if err != nil { + t.Fatal(err) + } + sender := store.EncodeUid(senderID) + marker := fmt.Sprintf("LOCAL-CRYPTO-%d", time.Now().UnixNano()) + msg := &types.Message{ + TenantID: 1, + SeqId: originalTopicSeq + 1, + Topic: topic, + From: sender.String(), + Head: types.KVMap{"x-im-client-mid": marker, "x-im-type": "text"}, + Content: map[string]any{"txt": marker}, + } + msg.InitTimes() + msg.SetUid(store.Store.GetUid()) + fallbackSeq := int64(0) + defer func() { + _, _ = db.Exec("DELETE FROM im_message_revisions WHERE topic=? AND seqid=?", topic, msg.SeqId) + _, _ = db.Exec("DELETE FROM im_messages WHERE topic=? AND seqid=?", topic, msg.SeqId) + if fallbackSeq > 0 { + _, _ = db.Exec("DELETE FROM im_messages WHERE topic=? AND seqid=?", topic, fallbackSeq) + } + _, _ = db.Exec("UPDATE im_topics SET seqid=? WHERE name=? AND seqid IN (?,?)", originalTopicSeq, topic, msg.SeqId, fallbackSeq) + }() + + if _, err = store.PersistPreparedMessage(1, msg, nil, false); err != nil { + t.Fatal(err) + } + var rawContent string + var nonce []byte + var keyVersion int + if err = db.QueryRow("SELECT JSON_UNQUOTE(content),content_nonce,key_version FROM im_messages WHERE topic=? AND seqid=?", topic, msg.SeqId).Scan(&rawContent, &nonce, &keyVersion); err != nil { + t.Fatal(err) + } + if keyVersion != 7 || len(nonce) != 12 || strings.Contains(rawContent, marker) { + t.Fatalf("message not encrypted at rest: key_version=%d nonce=%d content=%q", keyVersion, len(nonce), rawContent) + } + if _, err = base64.StdEncoding.DecodeString(rawContent); err != nil { + t.Fatalf("stored ciphertext is not base64: %v", err) + } + + loaded, err := store.Messages.GetOne(1, topic, msg.SeqId) + if err != nil || fmt.Sprint(loaded.Content.(map[string]any)["txt"]) != marker { + t.Fatalf("GetOne plaintext mismatch: content=%#v err=%v", loaded, err) + } + editedMarker := marker + "-EDITED" + edited, err := store.Messages.Edit(1, topic, msg.SeqId, sender, loaded.Version, loaded.Head, map[string]any{"txt": editedMarker}, "local integration test", types.TimeNow()) + if err != nil || fmt.Sprint(edited.Content.(map[string]any)["txt"]) != editedMarker { + t.Fatalf("Edit plaintext mismatch: content=%#v err=%v", edited, err) + } + var revisionVersion, currentVersion int + if err = db.QueryRow("SELECT key_version FROM im_message_revisions WHERE topic=? AND seqid=? ORDER BY version DESC LIMIT 1", topic, msg.SeqId).Scan(&revisionVersion); err != nil { + t.Fatal(err) + } + if err = db.QueryRow("SELECT key_version FROM im_messages WHERE topic=? AND seqid=?", topic, msg.SeqId).Scan(¤tVersion); err != nil { + t.Fatal(err) + } + if revisionVersion != 7 || currentVersion != 7 { + t.Fatalf("unexpected key versions: revision=%d current=%d", revisionVersion, currentVersion) + } + + history, err := store.Messages.GetAll(1, topic, sender, &types.QueryOpt{Since: msg.SeqId, Before: msg.SeqId + 1, Limit: 10}) + if err != nil || len(history) != 1 || fmt.Sprint(history[0].Content.(map[string]any)["txt"]) != editedMarker { + t.Fatalf("history plaintext mismatch: messages=%#v err=%v", history, err) + } + found, _, err := store.Messages.Search(1, topic, sender, editedMarker, "text", "", 10) + if err != nil || len(found) != 1 || found[0].SeqId != msg.SeqId { + t.Fatalf("search plaintext mismatch: messages=%#v err=%v", found, err) + } + + store.SetAdminInternalConfigProvider(localCryptoProvider{}) + store.InvalidateMessageEncryptionKeys(1) + fallback := &types.Message{ + TenantID: 1, + SeqId: msg.SeqId + 1, + Topic: topic, + From: sender.String(), + Head: types.KVMap{"x-im-client-mid": marker + "-FALLBACK", "x-im-type": "text"}, + Content: map[string]any{"txt": marker + "-PLAINTEXT"}, + } + fallback.InitTimes() + fallback.SetUid(store.Store.GetUid()) + fallbackSeq = fallback.SeqId + if _, err = store.PersistPreparedMessage(1, fallback, nil, false); err != nil { + t.Fatal(err) + } + if err = db.QueryRow("SELECT JSON_UNQUOTE(content),key_version FROM im_messages WHERE topic=? AND seqid=?", topic, fallback.SeqId).Scan(&rawContent, &keyVersion); err != nil { + t.Fatal(err) + } + if keyVersion != 0 || !strings.Contains(rawContent, marker+"-PLAINTEXT") { + t.Fatalf("missing-key fallback was not plaintext: key_version=%d content=%q", keyVersion, rawContent) + } +} diff --git a/server/store/message_crypto.go b/server/store/message_crypto.go new file mode 100644 index 0000000..6992974 --- /dev/null +++ b/server/store/message_crypto.go @@ -0,0 +1,260 @@ +package store + +import ( + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "encoding/base64" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "strconv" + "strings" + "sync" + "time" + + "github.com/tinode/chat/server/logs" + "github.com/tinode/chat/server/store/types" +) + +type cachedMessageKey struct { + enabled bool + version int + status string + key []byte + expires time.Time +} + +const messageKeyCacheTTL = 5 * time.Minute + +var messageKeyCache = struct { + sync.RWMutex + items map[string]cachedMessageKey +}{items: make(map[string]cachedMessageKey)} + +func messageKeyCacheID(tenantID types.TenantID, version int) string { + return strconv.FormatInt(int64(tenantID), 10) + ":" + strconv.Itoa(version) +} + +// WarmMessageEncryptionKeys preloads active keys. Missing keys leave the tenant in plaintext mode. +func WarmMessageEncryptionKeys() { + if adminInternalConfigProvider == nil || Tenants == nil { + return + } + tenants, err := Tenants.ListActive(10000) + if err != nil { + logs.Warn.Printf("message crypto: tenant preload failed: %v", err) + return + } + for i := range tenants { + if _, err = loadMessageKey(&tenants[i], 0); err != nil { + logs.Info.Printf("message crypto: plaintext mode tenant=%s reason=%v", tenants[i].Code, err) + } + } +} + +func loadMessageKey(tenant *types.Tenant, version int) (cachedMessageKey, error) { + var zero cachedMessageKey + if tenant == nil || adminInternalConfigProvider == nil { + return zero, types.ErrNotFound + } + result, err := adminInternalConfigProvider.MessageKey(tenant.Code, version) + if err != nil || result == nil { + if err == nil { + err = types.ErrNotFound + } + return zero, err + } + decoded, err := base64.StdEncoding.DecodeString(result.Key) + if err != nil || len(decoded) != 32 { + return zero, errors.New("invalid tenant message key") + } + entry := cachedMessageKey{enabled: result.Enabled, version: result.Version, status: result.Status, key: decoded, expires: time.Now().Add(messageKeyCacheTTL)} + messageKeyCache.Lock() + messageKeyCache.items[messageKeyCacheID(tenant.ID, result.Version)] = entry + if version == 0 { + messageKeyCache.items[messageKeyCacheID(tenant.ID, 0)] = entry + } + messageKeyCache.Unlock() + return entry, nil +} + +func getMessageKey(tenantID types.TenantID, version int) (cachedMessageKey, error) { + cacheID := messageKeyCacheID(tenantID, version) + messageKeyCache.RLock() + entry, ok := messageKeyCache.items[cacheID] + messageKeyCache.RUnlock() + if ok && time.Now().Before(entry.expires) { + return entry, nil + } + if ok { + messageKeyCache.Lock() + delete(messageKeyCache.items, cacheID) + messageKeyCache.Unlock() + } + tenant, err := Tenants.GetByID(tenantID) + if err != nil { + return cachedMessageKey{}, err + } + entry, err = loadMessageKey(tenant, version) + if err != nil && version == 0 { + entry = cachedMessageKey{expires: time.Now().Add(messageKeyCacheTTL)} + messageKeyCache.Lock() + messageKeyCache.items[cacheID] = entry + messageKeyCache.Unlock() + return entry, nil + } + return entry, err +} + +// PrepareMessageEncryptionKey verifies a pending key in memory and acknowledges +// it to Admin Server. Admin activates the key only after this succeeds. +func PrepareMessageEncryptionKey(tenantID types.TenantID, tenantCode string, version int) error { + if version <= 0 || adminInternalConfigProvider == nil { + return types.ErrMalformed + } + var tenant *types.Tenant + var err error + if tenantID.IsValid() { + tenant, err = Tenants.GetByID(tenantID) + } else { + tenant, err = Tenants.GetByCode(strings.TrimSpace(tenantCode)) + } + if err != nil || tenant == nil { + return err + } + entry, err := loadMessageKey(tenant, version) + if err != nil { + return err + } + if strings.EqualFold(entry.status, "pending") { + if err = adminInternalConfigProvider.MessageKeyAck(tenant.Code, version); err != nil { + return err + } + } + messageKeyCache.Lock() + messageKeyCache.items[messageKeyCacheID(tenant.ID, 0)] = entry + messageKeyCache.Unlock() + return nil +} + +// InvalidateMessageEncryptionKeys forces the next access to reload tenant keys. +func InvalidateMessageEncryptionKeys(tenantID types.TenantID) { + prefix := strconv.FormatInt(int64(tenantID), 10) + ":" + messageKeyCache.Lock() + for key := range messageKeyCache.items { + if len(key) >= len(prefix) && key[:len(prefix)] == prefix { + delete(messageKeyCache.items, key) + } + } + messageKeyCache.Unlock() +} + +func messageAAD(msg *types.Message, keyVersion int) []byte { + topic := []byte(msg.Topic) + aad := make([]byte, 8+8+4+len(topic)) + binary.BigEndian.PutUint64(aad[0:8], uint64(msg.TenantID)) + binary.BigEndian.PutUint64(aad[8:16], uint64(msg.SeqId)) + binary.BigEndian.PutUint32(aad[16:20], uint32(keyVersion)) + copy(aad[20:], topic) + return aad +} + +func encryptMessageContent(msg *types.Message) (*types.Message, error) { + entry, err := getMessageKey(msg.TenantID, 0) + if err != nil || !entry.enabled || entry.version <= 0 { + if err != nil { + logs.Warn.Printf("message crypto: active key unavailable, storing plaintext tenant=%d reason=%v", msg.TenantID, err) + } + copy := *msg + copy.KeyVersion, copy.ContentNonce = 0, nil + return ©, nil + } + plain, err := json.Marshal(msg.Content) + if err != nil { + return nil, err + } + block, err := aes.NewCipher(entry.key) + if err != nil { + return nil, err + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return nil, err + } + nonce := make([]byte, gcm.NonceSize()) + if _, err = io.ReadFull(rand.Reader, nonce); err != nil { + return nil, err + } + copy := *msg + copy.KeyVersion = entry.version + copy.ContentNonce = nonce + copy.Content = base64.StdEncoding.EncodeToString(gcm.Seal(nil, nonce, plain, messageAAD(©, entry.version))) + return ©, nil +} + +func decryptMessageContent(msg *types.Message) error { + if msg == nil || msg.KeyVersion <= 0 { + return nil + } + entry, err := getMessageKey(msg.TenantID, msg.KeyVersion) + if err != nil { + return fmt.Errorf("message key version %d unavailable: %w", msg.KeyVersion, err) + } + ciphertextText, ok := msg.Content.(string) + if !ok { + return errors.New("encrypted message content is not a string") + } + ciphertext, err := base64.StdEncoding.DecodeString(ciphertextText) + if err != nil { + return err + } + block, err := aes.NewCipher(entry.key) + if err != nil { + return err + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return err + } + if len(msg.ContentNonce) != gcm.NonceSize() { + return fmt.Errorf("invalid message nonce length: got %d want %d", len(msg.ContentNonce), gcm.NonceSize()) + } + plain, err := gcm.Open(nil, msg.ContentNonce, ciphertext, messageAAD(msg, msg.KeyVersion)) + if err != nil { + return err + } + return json.Unmarshal(plain, &msg.Content) +} + +func decryptMessages(messages []types.Message) error { + for i := range messages { + if err := decryptMessageContent(&messages[i]); err != nil { + return err + } + } + return nil +} + +func messageContentMatchesSearch(content any, query string) bool { + query = strings.ToLower(strings.TrimSpace(query)) + if query == "" { + return true + } + raw, err := json.Marshal(content) + if err != nil { + return false + } + text := strings.ToLower(string(raw)) + if strings.Contains(text, query) { + return true + } + for _, term := range strings.Fields(query) { + if !strings.Contains(text, term) { + return false + } + } + return true +} diff --git a/server/store/message_crypto_test.go b/server/store/message_crypto_test.go new file mode 100644 index 0000000..651e285 --- /dev/null +++ b/server/store/message_crypto_test.go @@ -0,0 +1,93 @@ +package store + +import ( + "bytes" + "testing" + "time" + + "github.com/tinode/chat/server/store/types" +) + +func TestMessageCryptoRoundTrip(t *testing.T) { + tenantID := types.TenantID(9001) + key := bytes.Repeat([]byte{7}, 32) + messageKeyCache.Lock() + expires := time.Now().Add(time.Minute) + messageKeyCache.items[messageKeyCacheID(tenantID, 0)] = cachedMessageKey{enabled: true, version: 3, key: key, expires: expires} + messageKeyCache.items[messageKeyCacheID(tenantID, 3)] = cachedMessageKey{enabled: true, version: 3, key: key, expires: expires} + messageKeyCache.Unlock() + defer InvalidateMessageEncryptionKeys(tenantID) + + original := &types.Message{TenantID: tenantID, Topic: "p2p-test", SeqId: 42, Content: map[string]any{"txt": "hello"}} + stored, err := encryptMessageContent(original) + if err != nil { + t.Fatal(err) + } + if stored.KeyVersion != 3 || len(stored.ContentNonce) != 12 || stored.Content == original.Content { + t.Fatalf("message was not encrypted: %#v", stored) + } + if err = decryptMessageContent(stored); err != nil { + t.Fatal(err) + } + content, ok := stored.Content.(map[string]any) + if !ok || content["txt"] != "hello" { + t.Fatalf("unexpected plaintext: %#v", stored.Content) + } +} + +func TestMessageCryptoPlaintextWhenDisabled(t *testing.T) { + tenantID := types.TenantID(9002) + messageKeyCache.Lock() + messageKeyCache.items[messageKeyCacheID(tenantID, 0)] = cachedMessageKey{expires: time.Now().Add(time.Minute)} + messageKeyCache.Unlock() + defer InvalidateMessageEncryptionKeys(tenantID) + + original := &types.Message{TenantID: tenantID, Topic: "grp-test", SeqId: 7, Content: "plain"} + stored, err := encryptMessageContent(original) + if err != nil { + t.Fatal(err) + } + if stored.KeyVersion != 0 || stored.Content != "plain" || len(stored.ContentNonce) != 0 { + t.Fatalf("unexpected encrypted message: %#v", stored) + } +} + +func TestMessageCryptoRejectsTampering(t *testing.T) { + tenantID := types.TenantID(9003) + key := bytes.Repeat([]byte{9}, 32) + expires := time.Now().Add(time.Minute) + messageKeyCache.Lock() + messageKeyCache.items[messageKeyCacheID(tenantID, 0)] = cachedMessageKey{enabled: true, version: 4, key: key, expires: expires} + messageKeyCache.items[messageKeyCacheID(tenantID, 4)] = cachedMessageKey{enabled: true, version: 4, key: key, expires: expires} + messageKeyCache.Unlock() + defer InvalidateMessageEncryptionKeys(tenantID) + + original := &types.Message{TenantID: tenantID, Topic: "p2p-secure", SeqId: 8, Content: map[string]any{"txt": "secret"}} + stored, err := encryptMessageContent(original) + if err != nil { + t.Fatal(err) + } + stored.Topic = "p2p-tampered" + if err = decryptMessageContent(stored); err == nil { + t.Fatal("expected authenticated decryption to reject changed topic") + } + + stored, err = encryptMessageContent(original) + if err != nil { + t.Fatal(err) + } + stored.ContentNonce = stored.ContentNonce[:len(stored.ContentNonce)-1] + if err = decryptMessageContent(stored); err == nil { + t.Fatal("expected invalid nonce length to be rejected") + } +} + +func TestMessageCryptoSearchMatchesDecryptedContent(t *testing.T) { + content := map[string]any{"txt": "项目 验收通过", "nested": map[string]any{"title": "日报"}} + if !messageContentMatchesSearch(content, "验收") { + t.Fatal("expected plaintext keyword to match") + } + if messageContentMatchesSearch(content, "不存在") { + t.Fatal("unexpected keyword match") + } +} diff --git a/server/store/store.go b/server/store/store.go index f07ef62..3d08565 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -174,6 +174,9 @@ type AdminInternalConfigProvider interface { ModerationPolicy(tenantCode string) (*types.ModerationPolicy, error) SensitiveWords(tenantCode string) (*types.SensitiveWordSet, error) SecurityPolicy(tenantCode string) (*types.SecurityPolicy, error) + RealNamePolicy(tenantID types.TenantID) (*types.RealNamePolicy, error) + MessageKey(tenantCode string, version int) (*types.MessageKey, error) + MessageKeyAck(tenantCode string, version int) error } var adminInternalConfigProvider AdminInternalConfigProvider @@ -319,6 +322,61 @@ func (thirdPartyConfigMapper) Get(group string, keys ...string) (map[string]stri return result, err } +type RealNameConfigPersistenceInterface interface { + Policy(tenantID types.TenantID) (*types.RealNamePolicy, error) +} + +type realNameConfigMapper struct{} + +var RealNameConfig RealNameConfigPersistenceInterface = realNameConfigMapper{} + +func (realNameConfigMapper) Policy(tenantID types.TenantID) (*types.RealNamePolicy, error) { + if !tenantID.IsValid() { + return nil, types.ErrMalformed + } + cacheKey := formatTenantIDForCache(tenantID) + if cached, ok := realNamePolicyCache.get(cacheKey); ok { + cp := *cached + return &cp, nil + } + if adminInternalConfigProvider != nil { + if result, err := adminInternalConfigProvider.RealNamePolicy(tenantID); err == nil && result != nil { + result.Normalize() + realNamePolicyCache.set(cacheKey, result) + cp := *result + return &cp, nil + } else if err != nil { + logs.Warn.Printf("admin internal realname policy fallback failed: tenant=%d err=%v", tenantID, err) + } + } + configAdp, ok := adp.(adapter.RealNamePolicyConfigAdapter) + if !ok { + policy := defaultRealNamePolicy(tenantID) + realNamePolicyCache.set(cacheKey, policy) + return policy, nil + } + result, err := configAdp.RealNamePolicy(tenantID) + if err != nil { + return nil, err + } + result.Normalize() + realNamePolicyCache.set(cacheKey, result) + cp := *result + return &cp, nil +} + +func defaultRealNamePolicy(tenantID types.TenantID) *types.RealNamePolicy { + policy := &types.RealNamePolicy{ + TenantID: tenantID, + ReviewPolicy: types.RealNameReviewPolicyNone, + ProviderMode: "id_card_2factor", + AllowSkip: true, + RemindAfter: "24h", + } + policy.Normalize() + return policy +} + type TenantEntryPersistenceInterface interface { CreatePublicUpload(upload *types.PublicUpload) (*types.PublicUpload, error) CreateTenantApplication(application *types.TenantApplication) (*types.TenantApplication, error) @@ -1490,10 +1548,16 @@ func (messagesMapper) GetOne(tenantID types.TenantID, topic string, seq int64) ( if err != nil { return nil, err } + var msg *types.Message if types.GetTopicCat(topic) == types.TopicCatGrp { - return tenantAdp.TenantGroupMessageGetOne(tenantID, topic, seq) + msg, err = tenantAdp.TenantGroupMessageGetOne(tenantID, topic, seq) + } else { + msg, err = tenantAdp.TenantMessageGetOne(tenantID, topic, seq) } - return tenantAdp.TenantMessageGetOne(tenantID, topic, seq) + if err == nil { + err = decryptMessageContent(msg) + } + return msg, err } func (messagesMapper) Recall(tenantID types.TenantID, topic string, seq int64, actor types.Uid, expectedVersion int, recalledAt time.Time) (*types.Message, error) { @@ -1503,6 +1567,9 @@ func (messagesMapper) Recall(tenantID types.TenantID, topic string, seq int64, a } if types.GetTopicCat(topic) == types.TopicCatGrp { msg, err := tenantAdp.TenantGroupMessageRecall(tenantID, topic, seq, actor, expectedVersion, recalledAt) + if err == nil { + err = decryptMessageContent(msg) + } if err == nil && groupMsgRuntime != nil { if cacheErr := groupMsgRuntime.CacheHotMessage(tenantID, msg); cacheErr != nil { logs.Warn.Printf("topic[%s]: failed to refresh recalled group message cache: %v", topic, cacheErr) @@ -1510,7 +1577,11 @@ func (messagesMapper) Recall(tenantID types.TenantID, topic string, seq int64, a } return msg, err } - return tenantAdp.TenantMessageRecall(tenantID, topic, seq, actor, expectedVersion, recalledAt) + msg, err := tenantAdp.TenantMessageRecall(tenantID, topic, seq, actor, expectedVersion, recalledAt) + if err == nil { + err = decryptMessageContent(msg) + } + return msg, err } func (messagesMapper) Edit(tenantID types.TenantID, topic string, seq int64, actor types.Uid, expectedVersion int, head types.KVMap, content any, reason string, editedAt time.Time) (*types.Message, error) { @@ -1521,8 +1592,16 @@ func (messagesMapper) Edit(tenantID types.TenantID, topic string, seq int64, act if err != nil { return nil, err } + stored, err := encryptMessageContent(&types.Message{TenantID: tenantID, Topic: topic, SeqId: seq, Content: content}) + if err != nil { + return nil, err + } + storedContent := types.StoredMessageContent{Content: stored.Content, Nonce: stored.ContentNonce, KeyVersion: stored.KeyVersion} if types.GetTopicCat(topic) == types.TopicCatGrp { - msg, err := tenantAdp.TenantGroupMessageEdit(tenantID, topic, seq, actor, expectedVersion, head, content, reason, editedAt) + msg, err := tenantAdp.TenantGroupMessageEdit(tenantID, topic, seq, actor, expectedVersion, head, storedContent, reason, editedAt) + if err == nil { + err = decryptMessageContent(msg) + } if err == nil && groupMsgRuntime != nil { if cacheErr := groupMsgRuntime.CacheHotMessage(tenantID, msg); cacheErr != nil { logs.Warn.Printf("topic[%s]: failed to refresh edited group message cache: %v", topic, cacheErr) @@ -1530,7 +1609,11 @@ func (messagesMapper) Edit(tenantID types.TenantID, topic string, seq int64, act } return msg, err } - return tenantAdp.TenantMessageEdit(tenantID, topic, seq, actor, expectedVersion, head, content, reason, editedAt) + msg, err := tenantAdp.TenantMessageEdit(tenantID, topic, seq, actor, expectedVersion, head, storedContent, reason, editedAt) + if err == nil { + err = decryptMessageContent(msg) + } + return msg, err } type ConversationsPersistenceInterface interface { @@ -3237,10 +3320,14 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag msg.ClientMsgID = clientMsgID fromUid := types.ParseUid(msg.From) isGroupMessage := types.GetTopicCat(msg.Topic) == types.TopicCatGrp + storedMsg, err := encryptMessageContent(msg) + if err != nil { + return nil, err + } if isGroupMessage { - err = tenantAdp.TenantGroupMessageSave(tenantID, msg) + err = tenantAdp.TenantGroupMessageSave(tenantID, storedMsg) } else { - err = tenantAdp.TenantMessageSave(tenantID, msg) + err = tenantAdp.TenantMessageSave(tenantID, storedMsg) } if err != nil { if err == types.ErrDuplicate { @@ -3250,6 +3337,9 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag return nil, dupErr } if duplicate != nil { + if dupErr = decryptMessageContent(duplicate); dupErr != nil { + return nil, dupErr + } msg.SeqId = duplicate.SeqId msg.ShardMonth = duplicate.ShardMonth if groupMsgRuntime != nil { @@ -3264,11 +3354,14 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag return nil, dupErr } if duplicate != nil { + if dupErr = decryptMessageContent(duplicate); dupErr != nil { + return nil, dupErr + } copyPersistedMessage(msg, duplicate) return &types.MessageSaveResult{SeqId: duplicate.SeqId, Duplicate: true}, nil } } - if retryErr := retryTenantMessageSaveAfterSeqConflict(tenantAdp, tenantID, msg, isGroupMessage); retryErr != nil { + if retryErr := retryTenantMessageSaveAfterSeqConflict(tenantAdp, tenantID, storedMsg, isGroupMessage); retryErr != nil { if isGroupMessage && groupMsgRuntime != nil && clientMsgID != "" { _ = groupMsgRuntime.MarkFailed(tenantID, msg.Topic, fromUid, clientMsgID) } @@ -3282,6 +3375,7 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag return nil, err } } + msg.SetUid(storedMsg.Uid()) if err = tenantAdp.TenantTopicUpdateOnMessage(tenantID, msg.Topic, msg); err != nil { return nil, err } @@ -3363,6 +3457,8 @@ func copyPersistedMessage(dst, src *types.Message) { dst.ClientMsgID = src.ClientMsgID dst.Head = src.Head dst.Content = src.Content + dst.ContentNonce = src.ContentNonce + dst.KeyVersion = src.KeyVersion } func PrepareMessagePersist(tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) (*types.MessageSaveResult, error) { @@ -3392,7 +3488,17 @@ func retryTenantMessageSaveAfterSeqConflict(tenantAdp adapter.TenantBusinessAdap return types.ErrDuplicate } prevSeq := msg.SeqId + plainMsg := *msg + if err = decryptMessageContent(&plainMsg); err != nil { + return err + } msg.SeqId = latest[0].SeqId + 1 + plainMsg.SeqId = msg.SeqId + if encryptedMsg, encryptErr := encryptMessageContent(&plainMsg); encryptErr != nil { + return encryptErr + } else { + *msg = *encryptedMsg + } if err = tenantAdp.TenantTopicUpdateOnMessage(tenantID, msg.Topic, msg); err != nil { return err } @@ -3647,9 +3753,17 @@ func (messagesMapper) GetAll(tenantID types.TenantID, topic string, forUser type return msgs, nil } } - return tenantAdp.TenantGroupMessageGetAll(tenantID, topic, forUser, opt) + messages, err := tenantAdp.TenantGroupMessageGetAll(tenantID, topic, forUser, opt) + if err == nil { + err = decryptMessages(messages) + } + return messages, err } - return tenantAdp.TenantMessageGetAll(tenantID, topic, forUser, opt) + messages, err := tenantAdp.TenantMessageGetAll(tenantID, topic, forUser, opt) + if err == nil { + err = decryptMessages(messages) + } + return messages, err } func (messagesMapper) Search(tenantID types.TenantID, topic string, forUser types.Uid, query, messageType, cursor string, limit int) ([]types.Message, string, error) { @@ -3663,7 +3777,40 @@ func (messagesMapper) Search(tenantID types.TenantID, topic string, forUser type if err != nil { return nil, "", err } - return tenantAdp.TenantMessageSearch(tenantID, topic, forUser, query, messageType, cursor, limit) + query = strings.TrimSpace(query) + if query == "" { + messages, nextCursor, searchErr := tenantAdp.TenantMessageSearch(tenantID, topic, forUser, "", messageType, cursor, limit) + if searchErr == nil { + searchErr = decryptMessages(messages) + } + return messages, nextCursor, searchErr + } + + // Encrypted content cannot be filtered by SQL. Walk visible messages in + // sequence order, decrypt them, then apply the same user-facing search. + matches := make([]types.Message, 0, limit) + nextCursor := cursor + for { + batch, batchCursor, searchErr := tenantAdp.TenantMessageSearch(tenantID, topic, forUser, "", messageType, nextCursor, 100) + if searchErr != nil { + return nil, "", searchErr + } + if searchErr = decryptMessages(batch); searchErr != nil { + return nil, "", searchErr + } + for _, msg := range batch { + if messageContentMatchesSearch(msg.Content, query) { + matches = append(matches, msg) + if len(matches) == limit { + return matches, strconv.FormatInt(msg.SeqId, 10), nil + } + } + } + if batchCursor == "" || len(batch) == 0 { + return matches, "", nil + } + nextCursor = batchCursor + } } func (messagesMapper) Mutations(tenantID types.TenantID, topic string, forUser types.Uid, sinceVersion int64, limit int) ([]types.MessageMutation, string, bool, int64, error) { @@ -4661,6 +4808,7 @@ func init() { Store = storeObj{} Tenants = tenantsMapper{} RegistrationConfig = registrationConfigMapper{} + RealNameConfig = realNameConfigMapper{} TenantEntry = tenantEntryMapper{} InternalAdminQueries = internalAdminQueryMapper{} SecurityAudit = securityAuditMapper{} diff --git a/server/store/types/message_crypto.go b/server/store/types/message_crypto.go new file mode 100644 index 0000000..8bed8d4 --- /dev/null +++ b/server/store/types/message_crypto.go @@ -0,0 +1,19 @@ +package types + +// MessageKey is a tenant message data-encryption key returned by Admin Server. +type MessageKey struct { + TenantID TenantID `json:"tenantId"` + TenantCode string `json:"tenantCode"` + Enabled bool `json:"enabled"` + Version int `json:"version"` + Algorithm string `json:"algorithm"` + Status string `json:"status"` + Key string `json:"key"` +} + +// StoredMessageContent carries encrypted content metadata through the edit adapter boundary. +type StoredMessageContent struct { + Content any + Nonce []byte + KeyVersion int +} diff --git a/server/store/types/security.go b/server/store/types/security.go index 643d82a..d6bba7d 100644 --- a/server/store/types/security.go +++ b/server/store/types/security.go @@ -2,6 +2,34 @@ package types import "time" +const ( + RealNameReviewPolicyNone = iota + RealNameReviewPolicyAutoManual +) + +// RealNamePolicy controls tenant-level real-name review behavior. +type RealNamePolicy struct { + TenantID TenantID `db:"tenant_id" json:"tenant_id"` + ReviewPolicy int `db:"realname_review_policy" json:"realname_review_policy"` + ProviderMode string `db:"realname_provider_mode" json:"realname_provider_mode"` + Required bool `db:"realname_required" json:"realname_required"` + AllowSkip bool `db:"realname_allow_skip" json:"realname_allow_skip"` + RemindAfter string `db:"realname_remind_after" json:"realname_remind_after"` +} + +// Normalize fills defaults and clamps unsupported enum values. +func (p *RealNamePolicy) Normalize() { + if p.ProviderMode == "" { + p.ProviderMode = "id_card_2factor" + } + if p.RemindAfter == "" { + p.RemindAfter = "24h" + } + if p.ReviewPolicy != RealNameReviewPolicyAutoManual { + p.ReviewPolicy = RealNameReviewPolicyNone + } +} + // RealNameVerification is the latest real-name verification submission for a user. type RealNameVerification struct { ID int64 `db:"id" json:"-"` diff --git a/server/store/types/types.go b/server/store/types/types.go index c6759e0..9aebacd 100644 --- a/server/store/types/types.go +++ b/server/store/types/types.go @@ -1250,26 +1250,30 @@ type Message struct { Topic string ShardMonth string `db:"shard_month" json:"-" bson:"shard_month,omitempty"` // Sender's user ID as string (without 'usr' prefix), could be empty. - From string - ClientMsgID string `db:"client_msg_id" json:"ClientMsgId,omitempty" bson:"client_msg_id,omitempty"` - Version int `json:"Version,omitempty" bson:",omitempty"` - RecalledAt *time.Time `db:"recalled_at" json:"RecalledAt,omitempty" bson:"recalled_at,omitempty"` - RecalledBy string `json:"RecalledBy,omitempty" bson:"recalled_by,omitempty"` - Head KVMap `json:"Head,omitempty" bson:",omitempty"` - Content any + From string + ClientMsgID string `db:"client_msg_id" json:"ClientMsgId,omitempty" bson:"client_msg_id,omitempty"` + Version int `json:"Version,omitempty" bson:",omitempty"` + RecalledAt *time.Time `db:"recalled_at" json:"RecalledAt,omitempty" bson:"recalled_at,omitempty"` + RecalledBy string `json:"RecalledBy,omitempty" bson:"recalled_by,omitempty"` + Head KVMap `json:"Head,omitempty" bson:",omitempty"` + Content any + ContentNonce []byte `db:"content_nonce" json:"-" bson:"content_nonce,omitempty"` + KeyVersion int `db:"key_version" json:"-" bson:"key_version,omitempty"` } // MessageRevision stores an immutable snapshot of a message before editing. type MessageRevision struct { - ObjHeader `bson:",inline"` - TenantID TenantID `db:"tenant_id" json:"-" bson:"tenant_id"` - Topic string - SeqId int64 - Version int - Actor string - Head KVMap `json:"Head,omitempty" bson:",omitempty"` - Content any - Reason string + ObjHeader `bson:",inline"` + TenantID TenantID `db:"tenant_id" json:"-" bson:"tenant_id"` + Topic string + SeqId int64 + Version int + Actor string + Head KVMap `json:"Head,omitempty" bson:",omitempty"` + Content any + ContentNonce []byte `db:"content_nonce" json:"-" bson:"content_nonce,omitempty"` + KeyVersion int `db:"key_version" json:"-" bson:"key_version,omitempty"` + Reason string } // MessageMutation describes an in-place message change used for offline compensation. diff --git a/server/topic_test.go b/server/topic_test.go index babdc3c..0cc68a7 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -1099,7 +1099,7 @@ func TestSaveMessageMaybeKafkaPublishesWithoutDBPersist(t *testing.T) { if err != nil { t.Fatalf("saveMessageMaybeKafka returned error: %v", err) } - if got == nil || got.SeqId != 1 || !got.MarkedReadBySender { + if got == nil || got.SeqId != 1 || !got.Pending || !got.MarkedReadBySender { t.Fatalf("unexpected save result: %#v", got) } if msg.ClientMsgID != "client-1" {