feat: enforce tenant content moderation

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)
+		}
+	}
+}