From 0f8e3b82b74547f3bdaee341de5ff98c5533e079 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Wed, 16 Sep 2026 22:21:25 +0800 Subject: [PATCH] feat: enforce tenant content moderation --- docs/sql/IM-内容审核日志SQL.sql | 22 +++ go.mod | 1 + go.sum | 2 + server/admin_internal_config.go | 29 ++++ server/datamodel.go | 16 ++ server/db/adapter.go | 1 + server/db/mysql/adapter.go | 2 + server/db/mysql/admin_governance_schema.go | 12 ++ server/db/mysql/tenant_security_audit.go | 24 +++ server/im_http.go | 54 +++++++ server/im_http_test.go | 4 +- server/kafka_admin_message_test.go | 4 + server/kafka_config_events.go | 13 ++ server/moderation/aliyun/text.go | 100 ++++++++++++ server/moderation/extractor.go | 34 ++++ server/moderation/local.go | 74 +++++++++ server/moderation/moderation_test.go | 67 ++++++++ server/moderation/provider.go | 62 ++++++++ server/moderation/service.go | 172 +++++++++++++++++++++ server/store/config_cache.go | 23 +++ server/store/security_audit.go | 9 ++ server/store/store.go | 57 +++++++ server/store/types/admin_governance.go | 17 ++ server/store/types/moderation.go | 45 ++++++ server/topic.go | 43 ++++++ server/topic_test.go | 5 + 26 files changed, 891 insertions(+), 1 deletion(-) create mode 100644 docs/sql/IM-内容审核日志SQL.sql create mode 100644 server/moderation/aliyun/text.go create mode 100644 server/moderation/extractor.go create mode 100644 server/moderation/local.go create mode 100644 server/moderation/moderation_test.go create mode 100644 server/moderation/provider.go create mode 100644 server/moderation/service.go create mode 100644 server/store/types/moderation.go diff --git a/docs/sql/IM-内容审核日志SQL.sql b/docs/sql/IM-内容审核日志SQL.sql new file mode 100644 index 0000000..a42d800 --- /dev/null +++ b/docs/sql/IM-内容审核日志SQL.sql @@ -0,0 +1,22 @@ +CREATE TABLE IF NOT EXISTS `im_content_moderation_logs` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `request_id` VARCHAR(64) NOT NULL, + `scene` VARCHAR(32) NOT NULL, + `topic` VARCHAR(96) NOT NULL DEFAULT '', + `seqid` BIGINT DEFAULT NULL, + `actor_id` BIGINT DEFAULT NULL, + `provider` VARCHAR(32) NOT NULL DEFAULT '', + `provider_request_id` VARCHAR(96) NOT NULL DEFAULT '', + `risk_level` VARCHAR(16) NOT NULL DEFAULT 'none', + `risk_labels` JSON DEFAULT NULL, + `action` VARCHAR(16) NOT NULL, + `content_hash` CHAR(64) NOT NULL, + `latency_ms` BIGINT NOT NULL DEFAULT 0, + `error_code` VARCHAR(64) NOT NULL DEFAULT '', + `created_at` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + PRIMARY KEY (`id`), + UNIQUE KEY `uk_im_content_moderation_request` (`request_id`), + KEY `idx_im_content_moderation_scene_time` (`scene`,`created_at` DESC,`id` DESC), + KEY `idx_im_content_moderation_actor_time` (`actor_id`,`created_at` DESC,`id` DESC), + KEY `idx_im_content_moderation_topic_seq` (`topic`,`seqid`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; diff --git a/go.mod b/go.mod index e9dc302..e9856a6 100644 --- a/go.mod +++ b/go.mod @@ -21,6 +21,7 @@ require ( github.com/gorilla/websocket v1.5.3 github.com/jackc/pgx/v5 v5.10.0 github.com/jmoiron/sqlx v1.4.0 + github.com/mozillazg/go-pinyin v0.20.0 github.com/nyaruka/phonenumbers v1.8.0 github.com/prometheus/client_golang v1.23.2 github.com/prometheus/common v0.69.0 diff --git a/go.sum b/go.sum index 4b52576..22ff6f6 100644 --- a/go.sum +++ b/go.sum @@ -265,6 +265,8 @@ github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9G github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= github.com/montanaflynn/stats v0.9.0 h1:tsBJ0RXwph9BmAuFoCmqGv6e8xa0MENQ8m0ptKq29mQ= github.com/montanaflynn/stats v0.9.0/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow= +github.com/mozillazg/go-pinyin v0.20.0 h1:BtR3DsxpApHfKReaPO1fCqF4pThRwH9uwvXzm+GnMFQ= +github.com/mozillazg/go-pinyin v0.20.0/go.mod h1:iR4EnMMRXkfpFVV5FMi4FNB6wGq9NV6uDWbUuPhP4Yc= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno= diff --git a/server/admin_internal_config.go b/server/admin_internal_config.go index a5aed8f..8deb907 100644 --- a/server/admin_internal_config.go +++ b/server/admin_internal_config.go @@ -167,6 +167,35 @@ func (c *adminInternalClient) ThirdPartyConfig(group string, keys ...string) (ma return wrapper.Values, nil } +func (c *adminInternalClient) ModerationProvider() (*types.ModerationProviderConfig, error) { + var result types.ModerationProviderConfig + if err := c.getJSON("/internal/im-config/moderation/provider", nil, &result); err != nil { + return nil, err + } + return &result, nil +} + +func (c *adminInternalClient) ModerationPolicy(tenantCode string) (*types.ModerationPolicy, error) { + var result types.ModerationPolicy + path := "/internal/im-config/tenants/" + url.PathEscape(strings.TrimSpace(tenantCode)) + "/moderation-policy" + if err := c.getJSON(path, nil, &result); err != nil { + return nil, err + } + return &result, nil +} + +func (c *adminInternalClient) SensitiveWords(tenantCode string) (*types.SensitiveWordSet, error) { + var result types.SensitiveWordSet + path := "/internal/im-config/tenants/" + url.PathEscape(strings.TrimSpace(tenantCode)) + "/sensitive-words" + if err := c.getJSON(path, nil, &result); err != nil { + return nil, err + } + if result.Items == nil { + result.Items = []*types.SensitiveWordRule{} + } + return &result, nil +} + func (c *adminInternalClient) TenantLoginIPWhitelist(id types.TenantID) (*tenantLoginIPWhitelistConfig, error) { if !id.IsValid() { return nil, types.ErrMalformed diff --git a/server/datamodel.go b/server/datamodel.go index cf887c7..ea952b9 100644 --- a/server/datamodel.go +++ b/server/datamodel.go @@ -10,11 +10,13 @@ package main import ( "encoding/json" + "errors" "net/http" "strconv" "strings" "time" + "github.com/tinode/chat/server/moderation" "github.com/tinode/chat/server/store/types" ) @@ -1619,6 +1621,20 @@ func ErrPolicyReply(msg *ClientComMessage, ts time.Time) *ServerComMessage { return ErrPolicyExplicitTs(msg.Id, msg.Original, ts, msg.Timestamp) } +func ErrContentModerationReply(msg *ClientComMessage, ts time.Time, err error, result *moderation.Result) *ServerComMessage { + reply := ErrPolicyReply(msg, ts) + reason := "CONTENT_REJECTED" + if errors.Is(err, moderation.ErrUnavailable) { + reason = "CONTENT_REVIEW_UNAVAILABLE" + } + reply.Ctrl.Text = "content rejected" + reply.Ctrl.Params = map[string]any{"reason": reason} + if result != nil && strings.TrimSpace(result.BlockMessage) != "" { + reply.Ctrl.Params.(map[string]any)["message"] = result.BlockMessage + } + return reply +} + // ErrCallBusyExplicitTs indicates a "busy" reply to a video call request (486). func ErrCallBusyExplicitTs(id, topic string, serverTs, incomingReqTs time.Time) *ServerComMessage { return &ServerComMessage{ diff --git a/server/db/adapter.go b/server/db/adapter.go index c025f62..91a0c4a 100644 --- a/server/db/adapter.go +++ b/server/db/adapter.go @@ -245,6 +245,7 @@ type TenantSecurityAuditAdapter interface { TenantLoginFailureCreate(t.TenantID, *t.LoginFailureEvent) error TenantLoginAuditClose(t.TenantID, string, time.Time) error TenantMessageAdminApply(t.TenantID, *t.MessageAdminAction) error + TenantContentModerationLogCreate(t.TenantID, *t.ContentModerationLog) error } // TenantRoutedAdapter can return a tenant-specific adapter, e.g. for the diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index ddc951a..118684c 100644 --- a/server/db/mysql/adapter.go +++ b/server/db/mysql/adapter.go @@ -587,6 +587,8 @@ 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', + 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), created_by BIGINT NOT NULL, updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), diff --git a/server/db/mysql/admin_governance_schema.go b/server/db/mysql/admin_governance_schema.go index f05b503..ce2502b 100644 --- a/server/db/mysql/admin_governance_schema.go +++ b/server/db/mysql/admin_governance_schema.go @@ -63,6 +63,18 @@ func ensureIMAdminGovernanceTables(exec sqlExecutor) error { PRIMARY KEY(id), UNIQUE INDEX uk_im_user_ip_whitelist(user_id,ip_cidr), INDEX idx_im_user_ip_whitelist_user(user_id,status), INDEX idx_im_user_ip_whitelist_deleted(deleted_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, + `CREATE TABLE IF NOT EXISTS im_content_moderation_logs ( + id BIGINT NOT NULL AUTO_INCREMENT, request_id VARCHAR(64) NOT NULL, + scene VARCHAR(32) NOT NULL, topic VARCHAR(96) NOT NULL DEFAULT '', seqid BIGINT, + actor_id BIGINT, provider VARCHAR(32) NOT NULL DEFAULT '', provider_request_id VARCHAR(96) NOT NULL DEFAULT '', + risk_level VARCHAR(16) NOT NULL DEFAULT 'none', risk_labels JSON, action VARCHAR(16) NOT NULL, + content_hash CHAR(64) NOT NULL, latency_ms BIGINT NOT NULL DEFAULT 0, error_code VARCHAR(64) NOT NULL DEFAULT '', + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), PRIMARY KEY(id), + UNIQUE INDEX uk_im_content_moderation_request(request_id), + INDEX idx_im_content_moderation_scene_time(scene,created_at DESC,id DESC), + INDEX idx_im_content_moderation_actor_time(actor_id,created_at DESC,id DESC), + INDEX idx_im_content_moderation_topic_seq(topic,seqid) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, } for _, statement := range statements { if _, err := exec.Exec(statement); err != nil { diff --git a/server/db/mysql/tenant_security_audit.go b/server/db/mysql/tenant_security_audit.go index 9d1078a..de6e74b 100644 --- a/server/db/mysql/tenant_security_audit.go +++ b/server/db/mysql/tenant_security_audit.go @@ -36,6 +36,30 @@ func (a *adapter) TenantLoginAuditCreate(id t.TenantID, event *t.LoginAuditEvent return err } +func (a *adapter) TenantContentModerationLogCreate(id t.TenantID, event *t.ContentModerationLog) error { + tenantAdp, err := a.routedTenantAdapter(id) + if err != nil || event == nil || strings.TrimSpace(event.RequestID) == "" { + return t.ErrMalformed + } + if tenantAdp != a { + return tenantAdp.TenantContentModerationLogCreate(id, event) + } + labels, err := json.Marshal(event.RiskLabels) + if err != nil { + return err + } + var actorID any + if !event.ActorID.IsZero() { + actorID = decodeUID(event.ActorID) + } + _, err = a.db.Exec(`INSERT INTO im_content_moderation_logs + (request_id,scene,topic,seqid,actor_id,provider,provider_request_id,risk_level,risk_labels,action,content_hash,latency_ms,error_code,created_at) + VALUES(?,?,?,?,?,?,?,?,CAST(? AS JSON),?,?,?,?,?)`, event.RequestID, event.Scene, event.Topic, event.SeqID, actorID, + event.Provider, event.ProviderRequestID, event.RiskLevel, string(labels), event.Action, event.ContentHash, + event.LatencyMS, event.ErrorCode, event.CreatedAt) + return err +} + func (a *adapter) TenantLoginFailureCreate(id t.TenantID, event *t.LoginFailureEvent) error { tenantAdp, err := a.routedTenantAdapter(id) if err != nil || event == nil { diff --git a/server/im_http.go b/server/im_http.go index 595a9dc..847e6a8 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -25,6 +25,7 @@ import ( "github.com/tinode/chat/server/auth" "github.com/tinode/chat/server/logs" "github.com/tinode/chat/server/media" + "github.com/tinode/chat/server/moderation" "github.com/tinode/chat/server/store" "github.com/tinode/chat/server/store/types" "golang.org/x/crypto/bcrypt" @@ -57,8 +58,36 @@ const ( imReasonResourceNotFound = "RESOURCE_NOT_FOUND" imReasonGroupDissolved = "GROUP_DISSOLVED" imReasonValidationFailed = "VALIDATION_FAILED" + imReasonContentRejected = "CONTENT_REJECTED" + imReasonReviewUnavailable = "CONTENT_REVIEW_UNAVAILABLE" ) +func reviewIMHTTPText(wrt http.ResponseWriter, req *http.Request, principal *imPrincipal, scene, topic, text string) bool { + text = strings.TrimSpace(text) + if text == "" { + return true + } + tenant, err := store.Tenants.GetByID(principal.TenantID) + if err != nil || tenant == nil { + logs.Warn.Printf("content moderation skipped because tenant code is unavailable: tenant=%d scene=%s err=%v", principal.TenantID, scene, err) + return true + } + result, err := moderation.Review(req.Context(), moderation.Request{TenantID: principal.TenantID, TenantCode: tenant.Code, Scene: scene, ActorID: principal.Uid, Topic: topic, Text: text, TraceID: req.Header.Get("X-Request-ID")}) + if err == nil { + return true + } + reason, message := imReasonContentRejected, "内容不符合平台规范，请修改后重试" + status := http.StatusUnprocessableEntity + if errors.Is(err, moderation.ErrUnavailable) { + reason, message, status = imReasonReviewUnavailable, "内容审核暂不可用，请稍后重试", http.StatusServiceUnavailable + } + if result != nil && strings.TrimSpace(result.BlockMessage) != "" { + message = result.BlockMessage + } + writeIMError(wrt, req, status, 30060, reason, message, nil) + return false +} + const ( imVerificationCodeLength = 6 imVerificationExpiresInSeconds = 300 @@ -3183,6 +3212,12 @@ func serveIMMeProfile(wrt http.ResponseWriter, req *http.Request) { return } } + if body.Nickname != nil && !reviewIMHTTPText(wrt, req, principal, moderation.SceneUserNickname, "", *body.Nickname) { + return + } + if body.Signature != nil && !reviewIMHTTPText(wrt, req, principal, moderation.SceneUserSignature, "", *body.Signature) { + return + } if body.AvatarFileID != nil { if avatar := strings.TrimSpace(*body.AvatarFileID); avatar != "" { if err := store.Files.LinkAttachments(principal.TenantID, principal.Uid.UserId(), types.ZeroUid, []string{avatar}); err != nil { @@ -4585,6 +4620,13 @@ func serveIMMessageRevision(wrt http.ResponseWriter, req *http.Request) { if body.ClientMid != "" { head["x-im-client-mid"] = body.ClientMid } + scene := moderation.ScenePrivateMessage + if strings.HasPrefix(topic, "grp") { + scene = moderation.SceneGroupMessage + } + if !reviewIMHTTPText(wrt, req, principal, scene, topic, moderation.ExtractText(body.Content)) { + return + } editedAt := time.Now().UTC().Round(time.Millisecond) updated, err := store.Messages.Edit(principal.TenantID, topic, seq, principal.Uid, body.Version, head, body.Content, body.Reason, editedAt) if err != nil { @@ -4996,6 +5038,9 @@ func serveIMGroupCreate(wrt http.ResponseWriter, req *http.Request) { writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "description"}) return } + if !reviewIMHTTPText(wrt, req, principal, moderation.SceneGroupName, "", name) { + return + } topicName := genTopicName() now := time.Now().UTC().Round(time.Millisecond) public := map[string]any{"name": name, "fn": name, "version": 1} @@ -5307,6 +5352,9 @@ func serveIMGroupDetail(wrt http.ResponseWriter, req *http.Request, topic string writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "name"}) return } + if !reviewIMHTTPText(wrt, req, principal, moderation.SceneGroupName, topic, name) { + return + } public["name"] = name public["fn"] = name } @@ -5493,6 +5541,9 @@ func serveIMGroupMyProfile(wrt http.ResponseWriter, req *http.Request, topic str writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "nickname"}) return } + if !reviewIMHTTPText(wrt, req, principal, moderation.SceneGroupNickname, topic, nickname) { + return + } if body.Version > 0 && body.Version != currentVersion { writeIMError(wrt, req, http.StatusConflict, 30006, "VERSION_CONFLICT", "version conflict", map[string]any{"current_version": currentVersion}) return @@ -6115,6 +6166,9 @@ func serveIMGroupAnnouncements(wrt http.ResponseWriter, req *http.Request, topic writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "content"}) return } + if !reviewIMHTTPText(wrt, req, principal, moderation.SceneGroupAnnouncement, topic, content) { + return + } now := time.Now().UTC().Round(time.Millisecond) announcement, err := store.GroupBusiness.PutAnnouncement(principal.TenantID, &types.GroupAnnouncement{ AnnouncementID: "ann_" + store.Store.GetUidString(), diff --git a/server/im_http_test.go b/server/im_http_test.go index 66d506b..1aa5073 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -6612,6 +6612,7 @@ func TestServeIMMessageForwardKafkaAssignsSeqAndBroadcasts(t *testing.T) { defer ctrl.Finish() mockStore := expectIMBearer(t, ctrl, "token-1", 7, 42) mockStore.EXPECT().GetUidString().Return("mid-3") + mockStore.EXPECT().GetUid().Return(types.Uid(103)) oldKafkaBus := kafkaBus oldPublisher := publishKafkaMessagePersist @@ -6736,7 +6737,8 @@ func TestServeIMMessageForwardKafkaAssignsSeqAndBroadcasts(t *testing.T) { func TestServeIMMessageForwardKafkaPublishUnknownReturnsPending(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() - expectIMBearer(t, ctrl, "token-1", 7, 42) + mockStore := expectIMBearer(t, ctrl, "token-1", 7, 42) + mockStore.EXPECT().GetUid().Return(types.Uid(104)) oldKafkaBus := kafkaBus oldPublisher := publishKafkaMessagePersist diff --git a/server/kafka_admin_message_test.go b/server/kafka_admin_message_test.go index 348b23a..645cc90 100644 --- a/server/kafka_admin_message_test.go +++ b/server/kafka_admin_message_test.go @@ -29,6 +29,10 @@ func (r *messageAdminAuditRecorder) ApplyMessageAction(_ types.TenantID, action return nil } +func (r *messageAdminAuditRecorder) RecordContentModeration(types.TenantID, *types.ContentModerationLog) error { + return nil +} + func TestExecuteKafkaMessageAdminCommand(t *testing.T) { previous := store.SecurityAudit recorder := &messageAdminAuditRecorder{} diff --git a/server/kafka_config_events.go b/server/kafka_config_events.go index 12a5e63..a479ae3 100644 --- a/server/kafka_config_events.go +++ b/server/kafka_config_events.go @@ -21,6 +21,8 @@ const ( configEventThirdPartyUpdated = "third_party.updated" configEventRegistrationUpdated = "registration.config.updated" configEventAppReleaseUpdated = "app.release.updated" + configEventSecurityPolicyUpdated = "security.policy.updated" + configEventSensitiveWordsUpdated = "sensitive_words.updated" ) type configEventPayload struct { @@ -83,17 +85,28 @@ func executeKafkaConfigEvent(envelope kafkaEnvelope) error { switch strings.TrimSpace(envelope.EventType) { case configEventTenantUpdated, configEventTenantStatusChanged, configEventTenantLimitsUpdated, configEventTenantFeaturesUpdated: store.InvalidateTenantConfigCache(tenantCode, tenantID) + if strings.TrimSpace(envelope.EventType) == configEventTenantFeaturesUpdated { + store.InvalidateModerationConfigCache(tenantCode, false, true, false) + } case configEventDictionaryUpdated, configEventThirdPartyUpdated: store.InvalidateThirdPartyConfigCache(payload.Group) + if payload.Group == "moderation" { + store.InvalidateModerationConfigCache(tenantCode, true, false, false) + } case configEventRegistrationUpdated: store.InvalidateRegistrationConfigCache() case configEventAppReleaseUpdated: store.InvalidateAppReleaseConfigCache() + case configEventSecurityPolicyUpdated: + store.InvalidateModerationConfigCache(tenantCode, false, true, false) + case configEventSensitiveWordsUpdated: + store.InvalidateModerationConfigCache(tenantCode, false, false, true) default: store.InvalidateTenantConfigCache(tenantCode, tenantID) store.InvalidateThirdPartyConfigCache(payload.Group) store.InvalidateRegistrationConfigCache() store.InvalidateAppReleaseConfigCache() + store.InvalidateModerationConfigCache(tenantCode, true, true, true) } logs.Info.Printf("kafka config event applied: event=%s type=%s tenant=%s group=%s", envelope.EventID, envelope.EventType, tenantCode, payload.Group) return nil diff --git a/server/moderation/aliyun/text.go b/server/moderation/aliyun/text.go new file mode 100644 index 0000000..3d34257 --- /dev/null +++ b/server/moderation/aliyun/text.go @@ -0,0 +1,100 @@ +package aliyun + +import ( + "context" + "encoding/json" + "errors" + "strings" + "time" + + openapi "github.com/alibabacloud-go/darabonba-openapi/v2/client" + green "github.com/alibabacloud-go/green-20220302/v3/client" + "github.com/alibabacloud-go/tea/dara" + "github.com/tinode/chat/server/store/types" +) + +type TextClient struct { + client *green.Client + timeout time.Duration +} + +type Result struct { + ProviderRequestID string + RiskLevel string + RiskLabels []string + Summary string + Latency time.Duration +} + +func NewTextClient(config *types.ModerationProviderConfig) (*TextClient, error) { + if config == nil || strings.TrimSpace(config.AccessKeyID) == "" || strings.TrimSpace(config.AccessKeySecret) == "" { + return nil, errors.New("missing aliyun moderation credentials") + } + clientConfig := &openapi.Config{AccessKeyId: dara.String(config.AccessKeyID), AccessKeySecret: dara.String(config.AccessKeySecret), RegionId: dara.String(config.RegionID), Protocol: dara.String("https")} + if endpoint := strings.TrimSpace(config.Endpoint); endpoint != "" { + clientConfig.Endpoint = dara.String(endpoint) + } + client, err := green.NewClient(clientConfig) + if err != nil { + return nil, err + } + timeout := time.Duration(config.TimeoutMS) * time.Millisecond + if timeout <= 0 { + timeout = 1500 * time.Millisecond + } + return &TextClient{client: client, timeout: timeout}, nil +} + +func (c *TextClient) Review(ctx context.Context, service, text, dataID string) (*Result, error) { + ctx, cancel := context.WithTimeout(ctx, c.timeout) + defer cancel() + params, err := json.Marshal(map[string]string{"content": text, "dataId": dataID}) + if err != nil { + return nil, err + } + started := time.Now() + resp, err := c.client.TextModerationPlusWithContext(ctx, &green.TextModerationPlusRequest{Service: dara.String(service), ServiceParameters: dara.String(string(params))}, &dara.RuntimeOptions{}) + latency := time.Since(started) + if err != nil { + return nil, err + } + if resp == nil || resp.Body == nil || dara.Int32Value(resp.Body.Code) != 200 || resp.Body.Data == nil { + return nil, errors.New("aliyun text moderation failed") + } + result := &Result{ProviderRequestID: dara.StringValue(resp.Body.RequestId), RiskLevel: moderationRisk(dara.StringValue(resp.Body.Data.RiskLevel)), Latency: latency} + seen := map[string]bool{} + for _, item := range resp.Body.Data.Result { + if item == nil { + continue + } + label := strings.TrimSpace(dara.StringValue(item.Label)) + if label != "" && !seen[label] { + seen[label] = true + result.RiskLabels = append(result.RiskLabels, label) + } + } + if ext := resp.Body.Data.Ext; ext != nil && ext.LlmContent != nil { + result.Summary = strings.TrimSpace(dara.StringValue(ext.LlmContent.OutputText)) + if llmIndicatesRisk(result.Summary) && result.RiskLevel == "none" { + result.RiskLevel = "medium" + result.RiskLabels = append(result.RiskLabels, "llm_risk") + } + } + return result, nil +} + +func moderationRisk(value string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case "high", "medium", "low": + return strings.ToLower(strings.TrimSpace(value)) + default: + return "none" + } +} +func llmIndicatesRisk(value string) bool { + value = strings.TrimSpace(value) + if value == "" { + return false + } + return !strings.Contains(value, "无风险") && !strings.HasPrefix(value, "正常") && (strings.Contains(value, "风险") || strings.Contains(value, "违规") || strings.Contains(value, "诈骗") || strings.Contains(value, "引流")) +} diff --git a/server/moderation/extractor.go b/server/moderation/extractor.go new file mode 100644 index 0000000..d8d7b21 --- /dev/null +++ b/server/moderation/extractor.go @@ -0,0 +1,34 @@ +package moderation + +import "strings" + +var visibleTextKeys = map[string]bool{ + "txt": true, "text": true, "title": true, "caption": true, + "description": true, "file_name": true, "filename": true, "name": true, +} + +func ExtractText(content any) string { + parts := make([]string, 0, 4) + var walk func(any, int, bool) + walk = func(value any, depth int, includeString bool) { + if depth > 6 || value == nil { + return + } + switch typed := value.(type) { + case string: + if text := strings.TrimSpace(typed); includeString && text != "" { + parts = append(parts, text) + } + case map[string]any: + for key, item := range typed { + walk(item, depth+1, visibleTextKeys[strings.ToLower(strings.TrimSpace(key))]) + } + case []any: + for _, item := range typed { + walk(item, depth+1, includeString) + } + } + } + walk(content, 0, true) + return strings.TrimSpace(strings.Join(parts, "\n")) +} diff --git a/server/moderation/local.go b/server/moderation/local.go new file mode 100644 index 0000000..286cbe3 --- /dev/null +++ b/server/moderation/local.go @@ -0,0 +1,74 @@ +package moderation + +import ( + "regexp" + "strings" + + "github.com/mozillazg/go-pinyin" + "github.com/tinode/chat/server/store/types" +) + +func matchLocalWords(text, scene string, words *types.SensitiveWordSet) *ProviderResult { + normalized := strings.ToLower(strings.TrimSpace(text)) + if normalized == "" || words == nil { + return nil + } + for _, rule := range words.Items { + if rule == nil || strings.TrimSpace(rule.Word) == "" || !sceneEnabled(rule.Scenes, scene) { + continue + } + matched := false + switch strings.ToLower(strings.TrimSpace(rule.MatchType)) { + case "exact": + matched = normalized == strings.ToLower(strings.TrimSpace(rule.Word)) + case "regex": + pattern, err := regexp.Compile(rule.Word) + matched = err == nil && pattern.MatchString(text) + case "pinyin": + args := pinyin.NewArgs() + args.Style = pinyin.Normal + textPinyin := strings.Join(pinyin.LazyConvert(text, &args), "") + wordPinyin := strings.Join(pinyin.LazyConvert(rule.Word, &args), "") + if wordPinyin == "" { + wordPinyin = strings.ToLower(strings.TrimSpace(rule.Word)) + } + matched = wordPinyin != "" && strings.Contains(strings.ToLower(textPinyin), strings.ToLower(wordPinyin)) + default: + matched = strings.Contains(normalized, strings.ToLower(strings.TrimSpace(rule.Word))) + } + if matched { + return &ProviderResult{RiskLevel: normalizedRisk(rule.RiskLevel), RiskLabels: []string{"local_sensitive_word"}, Summary: normalizedAction(rule.Action)} + } + } + return nil +} + +func sceneEnabled(scenes []string, scene string) bool { + if len(scenes) == 0 { + return true + } + for _, value := range scenes { + if value == "*" || value == scene { + return true + } + } + return false +} + +func normalizedRisk(value string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case "high", "medium", "low": + return strings.ToLower(strings.TrimSpace(value)) + default: + return "none" + } +} + +func normalizedAction(value string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case ActionBlock, ActionAuditOnly: + return strings.ToLower(strings.TrimSpace(value)) + default: + return ActionAllow + } +} diff --git a/server/moderation/moderation_test.go b/server/moderation/moderation_test.go new file mode 100644 index 0000000..c0f3ec2 --- /dev/null +++ b/server/moderation/moderation_test.go @@ -0,0 +1,67 @@ +package moderation + +import ( + "context" + "testing" + "time" + + "github.com/tinode/chat/server/store" + "github.com/tinode/chat/server/store/types" +) + +type testConfig struct { + policy *types.ModerationPolicy + words *types.SensitiveWordSet +} + +func (c testConfig) Provider() (*types.ModerationProviderConfig, error) { + return &types.ModerationProviderConfig{}, nil +} +func (c testConfig) Policy(string) (*types.ModerationPolicy, error) { return c.policy, nil } +func (c testConfig) Words(string) (*types.SensitiveWordSet, error) { return c.words, nil } + +type testAudit struct{} + +func (testAudit) RecordLogin(types.TenantID, *types.LoginAuditEvent) error { return nil } +func (testAudit) RecordLoginFailure(types.TenantID, *types.LoginFailureEvent) error { return nil } +func (testAudit) CloseLogin(types.TenantID, string, time.Time) error { return nil } +func (testAudit) ApplyMessageAction(types.TenantID, *types.MessageAdminAction) error { return nil } +func (testAudit) RecordContentModeration(types.TenantID, *types.ContentModerationLog) error { + return nil +} + +func TestExtractText(t *testing.T) { + content := map[string]any{"txt": "正文", "ent": []any{map[string]any{"data": map[string]any{"title": "查看详情", "url": "https://example.com"}}}, "meta": "ignore"} + if got := ExtractText(content); got != "正文\n查看详情" && got != "查看详情\n正文" { + t.Fatalf("ExtractText()=%q", got) + } +} + +func TestReviewLocalBlock(t *testing.T) { + oldConfig, oldAudit := store.ModerationConfig, store.SecurityAudit + defer func() { store.ModerationConfig, store.SecurityAudit = oldConfig, oldAudit }() + store.ModerationConfig = testConfig{policy: &types.ModerationPolicy{Enabled: true, LocalWordsEnabled: true, Scenes: map[string]bool{ScenePrivateMessage: true}, RiskActions: map[string]string{"high": ActionBlock}}, words: &types.SensitiveWordSet{Items: []*types.SensitiveWordRule{{Word: "违规词", MatchType: "contains", Scenes: []string{ScenePrivateMessage}, Action: ActionBlock, RiskLevel: "high"}}}} + store.SecurityAudit = testAudit{} + result, err := Review(context.Background(), Request{TenantID: 1, TenantCode: "default", Scene: ScenePrivateMessage, Text: "包含违规词的消息"}) + if err != ErrRejected || result.Action != ActionBlock || result.Provider != "local" { + t.Fatalf("result=%+v err=%v", result, err) + } +} + +func TestReviewOtherTenantUsesOwnPolicy(t *testing.T) { + oldConfig, oldAudit := store.ModerationConfig, store.SecurityAudit + defer func() { store.ModerationConfig, store.SecurityAudit = oldConfig, oldAudit }() + store.ModerationConfig = testConfig{policy: &types.ModerationPolicy{Enabled: true, LocalWordsEnabled: true, Scenes: map[string]bool{ScenePrivateMessage: true}, RiskActions: map[string]string{"high": ActionBlock}}, words: &types.SensitiveWordSet{Items: []*types.SensitiveWordRule{{Word: "违规词", MatchType: "contains", Scenes: []string{ScenePrivateMessage}, Action: ActionBlock, RiskLevel: "high"}}}} + store.SecurityAudit = testAudit{} + result, err := Review(context.Background(), Request{TenantID: 2, TenantCode: "team", Scene: ScenePrivateMessage, Text: "违规词"}) + if err != ErrRejected || result.Action != ActionBlock { + t.Fatalf("result=%+v err=%v", result, err) + } +} + +func TestMatchLocalWordsPinyin(t *testing.T) { + result := matchLocalWords("这里包含敏感内容", SceneGroupMessage, &types.SensitiveWordSet{Items: []*types.SensitiveWordRule{{Word: "mingan", MatchType: "pinyin", Scenes: []string{SceneGroupMessage}, RiskLevel: "medium"}}}) + if result == nil || result.RiskLevel != "medium" { + t.Fatalf("result=%+v", result) + } +} diff --git a/server/moderation/provider.go b/server/moderation/provider.go new file mode 100644 index 0000000..18c9e91 --- /dev/null +++ b/server/moderation/provider.go @@ -0,0 +1,62 @@ +package moderation + +import ( + "context" + "errors" + "time" + + "github.com/tinode/chat/server/store/types" +) + +const ( + SceneUserNickname = "user_nickname" + SceneUserSignature = "user_signature" + SceneGroupName = "group_name" + SceneGroupNickname = "group_nickname" + ScenePrivateMessage = "private_message" + SceneGroupMessage = "group_message" + SceneGroupAnnouncement = "group_announcement" + + ActionAllow = "allow" + ActionBlock = "block" + ActionAuditOnly = "audit_only" +) + +var ( + ErrRejected = errors.New("content rejected") + ErrUnavailable = errors.New("content review unavailable") +) + +type Request struct { + TenantID types.TenantID + TenantCode string + Scene string + ActorID types.Uid + Topic string + Text string + TraceID string +} + +type ProviderResult struct { + ProviderRequestID string + RiskLevel string + RiskLabels []string + Summary string + Latency time.Duration +} + +type Result struct { + RequestID string + Action string + RiskLevel string + RiskLabels []string + Provider string + ProviderRequestID string + Latency time.Duration + BlockMessage string + Err error +} + +type Provider interface { + Review(context.Context, string, string, string) (*ProviderResult, error) +} diff --git a/server/moderation/service.go b/server/moderation/service.go new file mode 100644 index 0000000..8af9039 --- /dev/null +++ b/server/moderation/service.go @@ -0,0 +1,172 @@ +package moderation + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "strings" + "sync" + "time" + + "github.com/tinode/chat/server/logs" + "github.com/tinode/chat/server/moderation/aliyun" + "github.com/tinode/chat/server/store" + "github.com/tinode/chat/server/store/types" +) + +type circuitState struct { + sync.Mutex + failures int + openUntil time.Time +} + +var aliyunCircuit circuitState + +func Review(ctx context.Context, req Request) (*Result, error) { + text := strings.TrimSpace(req.Text) + result := &Result{RequestID: moderationRequestID(), Action: ActionAllow, RiskLevel: "none"} + if text == "" || strings.TrimSpace(req.TenantCode) == "" { + return result, nil + } + policy, err := store.ModerationConfig.Policy(req.TenantCode) + if err != nil { + return failureResult(result, nil, err) + } + if policy == nil || !policy.Enabled || !policy.Scenes[req.Scene] { + return result, nil + } + defer func() { recordDecision(req, text, result) }() + result.BlockMessage = policy.BlockMessage + if policy.LocalWordsEnabled { + words, wordsErr := store.ModerationConfig.Words(req.TenantCode) + if wordsErr != nil { + if logs.Warn != nil { + logs.Warn.Printf("moderation local words unavailable: tenant=%s scene=%s err=%v", req.TenantCode, req.Scene, wordsErr) + } + } else if localResult := matchLocalWords(text, req.Scene, words); localResult != nil { + result.Provider, result.RiskLevel, result.RiskLabels = "local", localResult.RiskLevel, localResult.RiskLabels + result.Action = actionForRisk(policy, localResult.RiskLevel) + if localResult.Summary == ActionBlock { + result.Action = ActionBlock + } + if result.Action == ActionBlock { + result.Err = ErrRejected + return result, ErrRejected + } + } + } + if !policy.AliyunEnabled { + return result, nil + } + providerConfig, err := store.ModerationConfig.Provider() + if err != nil || providerConfig == nil || !providerConfig.Enabled { + return failureResult(result, policy, firstError(err, ErrUnavailable)) + } + if circuitOpen() { + return failureResult(result, policy, ErrUnavailable) + } + provider, err := aliyun.NewTextClient(providerConfig) + if err != nil { + recordFailure() + return failureResult(result, policy, err) + } + service := providerConfig.ChatService + if req.Scene == SceneUserNickname || req.Scene == SceneUserSignature || req.Scene == SceneGroupName || req.Scene == SceneGroupNickname { + service = providerConfig.NicknameService + } + providerResult, err := provider.Review(ctx, service, text, result.RequestID) + if err != nil { + recordFailure() + return failureResult(result, policy, err) + } + recordSuccess() + result.Provider, result.ProviderRequestID, result.RiskLevel, result.RiskLabels, result.Latency = "aliyun_green", providerResult.ProviderRequestID, providerResult.RiskLevel, providerResult.RiskLabels, providerResult.Latency + result.Action = actionForRisk(policy, result.RiskLevel) + if result.Action == ActionBlock { + result.Err = ErrRejected + return result, ErrRejected + } + return result, nil +} + +func failureResult(result *Result, policy *types.ModerationPolicy, err error) (*Result, error) { + result.Err = err + failureMode := "allow" + if policy != nil && policy.FailureMode == "block" { + failureMode = "block" + result.BlockMessage = policy.BlockMessage + } + if failureMode == "block" { + result.Action = ActionBlock + return result, ErrUnavailable + } + result.Action = ActionAuditOnly + return result, nil +} + +func actionForRisk(policy *types.ModerationPolicy, risk string) string { + if policy != nil { + if action := normalizedAction(policy.RiskActions[normalizedRisk(risk)]); action != "" { + return action + } + } + return ActionAllow +} +func firstError(err, fallback error) error { + if err != nil { + return err + } + return fallback +} +func moderationRequestID() string { + var raw [12]byte + if _, err := rand.Read(raw[:]); err != nil { + return hex.EncodeToString([]byte(time.Now().Format(time.RFC3339Nano))) + } + return hex.EncodeToString(raw[:]) +} +func circuitOpen() bool { + aliyunCircuit.Lock() + defer aliyunCircuit.Unlock() + return time.Now().Before(aliyunCircuit.openUntil) +} +func recordFailure() { + aliyunCircuit.Lock() + defer aliyunCircuit.Unlock() + aliyunCircuit.failures++ + if aliyunCircuit.failures >= 5 { + aliyunCircuit.openUntil = time.Now().Add(30 * time.Second) + aliyunCircuit.failures = 0 + } +} +func recordSuccess() { + aliyunCircuit.Lock() + aliyunCircuit.failures = 0 + aliyunCircuit.openUntil = time.Time{} + aliyunCircuit.Unlock() +} +func logDecision(req Request, result *Result) { + if logs.Info != nil { + logs.Info.Printf("content moderation: request=%s tenant=%s scene=%s actor=%s provider=%s risk=%s action=%s latency_ms=%d", result.RequestID, req.TenantCode, req.Scene, req.ActorID.UserId(), result.Provider, result.RiskLevel, result.Action, result.Latency.Milliseconds()) + } +} + +func recordDecision(req Request, text string, result *Result) { + logDecision(req, result) + sum := sha256.Sum256([]byte(text)) + errorCode := "" + if result.Err != nil { + errorCode = "PROVIDER_ERROR" + } + if err := store.SecurityAudit.RecordContentModeration(req.TenantID, &types.ContentModerationLog{ + RequestID: result.RequestID, Scene: req.Scene, Topic: req.Topic, ActorID: req.ActorID, + Provider: result.Provider, ProviderRequestID: result.ProviderRequestID, RiskLevel: result.RiskLevel, + RiskLabels: result.RiskLabels, Action: result.Action, ContentHash: hex.EncodeToString(sum[:]), + LatencyMS: result.Latency.Milliseconds(), ErrorCode: errorCode, CreatedAt: time.Now().UTC().Round(time.Millisecond), + }); err != nil { + if logs.Warn != nil { + logs.Warn.Printf("content moderation audit write failed: request=%s tenant=%s err=%v", result.RequestID, req.TenantCode, err) + } + } +} diff --git a/server/store/config_cache.go b/server/store/config_cache.go index 14bd88a..423639c 100644 --- a/server/store/config_cache.go +++ b/server/store/config_cache.go @@ -61,9 +61,32 @@ var ( registrationSMSCache = newConfigCache[bool]() thirdPartyConfigCache = newConfigCache[map[string]string]() appReleaseConfigCache = newConfigCache[*types.AppRelease]() + moderationProviderCache = newConfigCache[*types.ModerationProviderConfig]() + moderationPolicyCache = newConfigCache[*types.ModerationPolicy]() + moderationWordsCache = newConfigCache[*types.SensitiveWordSet]() thirdPartyConfigAllKeys = "*" ) +func InvalidateModerationConfigCache(tenantCode string, provider, policy, words bool) { + if provider { + moderationProviderCache.clear() + } + if policy { + if strings.TrimSpace(tenantCode) == "" { + moderationPolicyCache.clear() + } else { + moderationPolicyCache.delete(strings.TrimSpace(tenantCode)) + } + } + if words { + if strings.TrimSpace(tenantCode) == "" { + moderationWordsCache.clear() + } else { + moderationWordsCache.delete(strings.TrimSpace(tenantCode)) + } + } +} + func cloneTenant(t *types.Tenant) *types.Tenant { if t == nil { return nil diff --git a/server/store/security_audit.go b/server/store/security_audit.go index 7026e95..34c7d69 100644 --- a/server/store/security_audit.go +++ b/server/store/security_audit.go @@ -12,6 +12,15 @@ type SecurityAuditPersistenceInterface interface { RecordLoginFailure(types.TenantID, *types.LoginFailureEvent) error CloseLogin(types.TenantID, string, time.Time) error ApplyMessageAction(types.TenantID, *types.MessageAdminAction) error + RecordContentModeration(types.TenantID, *types.ContentModerationLog) error +} + +func (securityAuditMapper) RecordContentModeration(tenantID types.TenantID, event *types.ContentModerationLog) error { + audit, err := tenantSecurityAuditAdapter(tenantID) + if err != nil { + return err + } + return audit.TenantContentModerationLogCreate(tenantID, event) } type securityAuditMapper struct{} diff --git a/server/store/store.go b/server/store/store.go index 54d66c2..880b51f 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -170,6 +170,9 @@ type AdminInternalConfigProvider interface { RegistrationSMSEnabled() (bool, error) ThirdPartyConfig(group string, keys ...string) (map[string]string, error) LatestAppRelease(platform, channel string) (*types.AppRelease, error) + ModerationProvider() (*types.ModerationProviderConfig, error) + ModerationPolicy(tenantCode string) (*types.ModerationPolicy, error) + SensitiveWords(tenantCode string) (*types.SensitiveWordSet, error) } var adminInternalConfigProvider AdminInternalConfigProvider @@ -178,6 +181,60 @@ func SetAdminInternalConfigProvider(provider AdminInternalConfigProvider) { adminInternalConfigProvider = provider } +type ModerationConfigPersistenceInterface interface { + Provider() (*types.ModerationProviderConfig, error) + Policy(tenantCode string) (*types.ModerationPolicy, error) + Words(tenantCode string) (*types.SensitiveWordSet, error) +} + +type moderationConfigMapper struct{} + +var ModerationConfig ModerationConfigPersistenceInterface = moderationConfigMapper{} + +func (moderationConfigMapper) Provider() (*types.ModerationProviderConfig, error) { + if value, ok := moderationProviderCache.get("provider"); ok { + return value, nil + } + if adminInternalConfigProvider == nil { + return nil, types.ErrUnsupported + } + value, err := adminInternalConfigProvider.ModerationProvider() + if err == nil && value != nil { + moderationProviderCache.set("provider", value) + } + return value, err +} + +func (moderationConfigMapper) Policy(tenantCode string) (*types.ModerationPolicy, error) { + tenantCode = strings.TrimSpace(tenantCode) + if value, ok := moderationPolicyCache.get(tenantCode); ok { + return value, nil + } + if adminInternalConfigProvider == nil { + return nil, types.ErrUnsupported + } + value, err := adminInternalConfigProvider.ModerationPolicy(tenantCode) + if err == nil && value != nil { + moderationPolicyCache.set(tenantCode, value) + } + return value, err +} + +func (moderationConfigMapper) Words(tenantCode string) (*types.SensitiveWordSet, error) { + tenantCode = strings.TrimSpace(tenantCode) + if value, ok := moderationWordsCache.get(tenantCode); ok { + return value, nil + } + if adminInternalConfigProvider == nil { + return nil, types.ErrUnsupported + } + value, err := adminInternalConfigProvider.SensitiveWords(tenantCode) + if err == nil && value != nil { + moderationWordsCache.set(tenantCode, value) + } + return value, err +} + // RegistrationConfigPersistenceInterface exposes global registration policy. type RegistrationConfigPersistenceInterface interface { SMSEnabled() (bool, error) diff --git a/server/store/types/admin_governance.go b/server/store/types/admin_governance.go index 55d47d9..f0226ad 100644 --- a/server/store/types/admin_governance.go +++ b/server/store/types/admin_governance.go @@ -57,3 +57,20 @@ type MessageAdminAction struct { RequestID string ProcessedAt time.Time } + +type ContentModerationLog struct { + RequestID string + Scene string + Topic string + SeqID *int64 + ActorID Uid + Provider string + ProviderRequestID string + RiskLevel string + RiskLabels []string + Action string + ContentHash string + LatencyMS int64 + ErrorCode string + CreatedAt time.Time +} diff --git a/server/store/types/moderation.go b/server/store/types/moderation.go new file mode 100644 index 0000000..dcaf72b --- /dev/null +++ b/server/store/types/moderation.go @@ -0,0 +1,45 @@ +package types + +type ModerationProviderConfig struct { + Provider string `json:"provider"` + Endpoint string `json:"endpoint"` + Enabled bool `json:"enabled"` + RegionID string `json:"region_id"` + NicknameService string `json:"nickname_service"` + ChatService string `json:"chat_service"` + TimeoutMS int `json:"timeout_ms"` + FailureMode string `json:"failure_mode"` + AccessKeyID string `json:"access_key_id"` + AccessKeySecret string `json:"access_key_secret"` +} + +type ModerationPolicy struct { + TenantID TenantID `json:"tenant_id"` + TenantCode string `json:"tenant_code"` + PolicyVersion uint `json:"policy_version"` + Enabled bool `json:"enabled"` + AliyunEnabled bool `json:"aliyun_enabled"` + LocalWordsEnabled bool `json:"local_words_enabled"` + Scenes map[string]bool `json:"scenes"` + RiskActions map[string]string `json:"risk_actions"` + FailureMode string `json:"failure_mode"` + BlockMessage string `json:"block_message"` +} + +type SensitiveWordRule struct { + ID int64 `json:"id"` + Word string `json:"word"` + MatchType string `json:"match_type"` + Scenes []string `json:"scenes"` + Action string `json:"action"` + Replacement string `json:"replacement"` + RiskLevel string `json:"risk_level"` + Priority int `json:"priority"` +} + +type SensitiveWordSet struct { + TenantID TenantID `json:"tenant_id"` + TenantCode string `json:"tenant_code"` + Version int64 `json:"version"` + Items []*SensitiveWordRule `json:"items"` +} diff --git a/server/topic.go b/server/topic.go index 5da320e..b7596f6 100644 --- a/server/topic.go +++ b/server/topic.go @@ -20,6 +20,7 @@ import ( "github.com/tinode/chat/server/auth" "github.com/tinode/chat/server/logs" + "github.com/tinode/chat/server/moderation" "github.com/tinode/chat/server/store" "github.com/tinode/chat/server/store/types" ) @@ -1494,6 +1495,22 @@ func (t *Topic) handlePubBroadcast(msg *ClientComMessage) { msg.Pub.Head = head msg.Pub.Content = content } + if msg.sess != nil && msg.sess.authLvl != auth.LevelRoot && !isCall { + scene := moderation.ScenePrivateMessage + if t.cat == types.TopicCatGrp { + scene = moderation.SceneGroupMessage + } + if text := moderation.ExtractText(msg.Pub.Content); text != "" { + result, reviewErr := moderation.Review(context.Background(), moderation.Request{ + TenantID: t.tenantID, TenantCode: msg.sess.tenantCode, Scene: scene, + ActorID: asUid, Topic: t.name, Text: text, TraceID: msg.Id, + }) + if reviewErr != nil { + msg.sess.queueOut(ErrContentModerationReply(msg, types.TimeNow(), reviewErr, result)) + return + } + } + } // Save to DB at master topic. attachments := messageAttachmentURLs(msg) @@ -2880,6 +2897,32 @@ func (t *Topic) replySetDesc(sess *Session, asUid types.Uid, asChan bool, sess.queueOut(ErrPermissionDeniedReply(msg, now)) return errors.New("attempt to change Trusted by non-root") } + if sess != nil && authLevel != auth.LevelRoot { + reviewMeta := func(scene string, value any) error { + text := moderation.ExtractText(value) + if text == "" { + return nil + } + result, reviewErr := moderation.Review(context.Background(), moderation.Request{TenantID: t.tenantID, TenantCode: sess.tenantCode, Scene: scene, ActorID: asUid, Topic: t.name, Text: text, TraceID: msg.Id}) + if reviewErr != nil { + sess.queueOut(ErrContentModerationReply(msg, now, reviewErr, result)) + return reviewErr + } + return nil + } + if t.cat == types.TopicCatMe { + if err = reviewMeta(moderation.SceneUserNickname, set.Desc.Public); err != nil { + return err + } + } else if t.cat == types.TopicCatGrp { + if err = reviewMeta(moderation.SceneGroupName, set.Desc.Public); err != nil { + return err + } + if err = reviewMeta(moderation.SceneGroupNickname, set.Desc.Private); err != nil { + return err + } + } + } switch t.cat { case types.TopicCatMe: diff --git a/server/topic_test.go b/server/topic_test.go index 438e541..babdc3c 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -1061,15 +1061,20 @@ func TestSaveMessageMaybeKafkaPublishesWithoutDBPersist(t *testing.T) { defer ctrl.Finish() oldKafkaBus := kafkaBus + oldStore := store.Store oldMessages := store.Messages oldPublisher := publishKafkaMessagePersist defer func() { kafkaBus = oldKafkaBus + store.Store = oldStore store.Messages = oldMessages publishKafkaMessagePersist = oldPublisher }() kafkaBus = &kafkaRuntime{} + mockStore := mock_store.NewMockPersistentStorageInterface(ctrl) + mockStore.EXPECT().GetUid().Return(types.Uid(101)) + store.Store = mockStore store.Messages = mock_store.NewMockMessagesPersistenceInterface(ctrl) var publishedTopic string var publishedKey string