diff --git a/docs/sql/IM-后台消息与登录审计SQL.sql b/docs/sql/IM-后台消息与登录审计SQL.sql new file mode 100644 index 0000000..5985fad --- /dev/null +++ b/docs/sql/IM-后台消息与登录审计SQL.sql @@ -0,0 +1,88 @@ +-- MySQL 8.0. Admin message governance and authentication audit tables. +-- Execute this script in each tenant business database. + +CREATE TABLE IF NOT EXISTS im_user_login_logs ( + id BIGINT NOT NULL AUTO_INCREMENT, + user_id BIGINT NOT NULL, + username VARCHAR(64) NOT NULL DEFAULT '', + account_type VARCHAR(24) NOT NULL DEFAULT '', + account_hash CHAR(64) NOT NULL DEFAULT '', + account_mask VARCHAR(128) NOT NULL DEFAULT '', + login_type VARCHAR(24) NOT NULL, + event_type VARCHAR(24) NOT NULL DEFAULT 'login', + session_id VARCHAR(96) NOT NULL DEFAULT '', + device_id VARCHAR(255) NOT NULL DEFAULT '', + platform VARCHAR(32) NOT NULL DEFAULT 'unknown', + app_version VARCHAR(32) NOT NULL DEFAULT '', + ip VARCHAR(64) NOT NULL DEFAULT '', + region VARCHAR(128) NOT NULL DEFAULT '', + user_agent VARCHAR(512) NOT NULL DEFAULT '', + risk_level VARCHAR(16) NOT NULL DEFAULT 'low', + request_id VARCHAR(64) NOT NULL DEFAULT '', + login_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + logout_at DATETIME(3) NULL, + metadata JSON NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + PRIMARY KEY (id), + INDEX idx_im_user_login_user_time (user_id, login_at DESC, id DESC), + INDEX idx_im_user_login_ip_time (ip, login_at DESC, id DESC), + INDEX idx_im_user_login_device_time (device_id(64), login_at DESC), + INDEX idx_im_user_login_request (request_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +CREATE TABLE IF NOT EXISTS im_login_failure_logs ( + id BIGINT NOT NULL AUTO_INCREMENT, + user_id BIGINT NULL, + account_type VARCHAR(24) NOT NULL DEFAULT '', + account_hash CHAR(64) NOT NULL DEFAULT '', + account_mask VARCHAR(128) NOT NULL DEFAULT '', + failure_type VARCHAR(32) NOT NULL, + reason VARCHAR(96) NOT NULL DEFAULT '', + detail VARCHAR(500) NOT NULL DEFAULT '', + ip VARCHAR(64) NOT NULL DEFAULT '', + region VARCHAR(128) NOT NULL DEFAULT '', + device_id VARCHAR(255) NOT NULL DEFAULT '', + platform VARCHAR(32) NOT NULL DEFAULT 'unknown', + app_version VARCHAR(32) NOT NULL DEFAULT '', + user_agent VARCHAR(512) NOT NULL DEFAULT '', + risk_level VARCHAR(16) NOT NULL DEFAULT 'medium', + risk_action VARCHAR(24) NOT NULL DEFAULT 'audit_only', + request_id VARCHAR(64) NOT NULL DEFAULT '', + attempt_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + metadata JSON NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + PRIMARY KEY (id), + INDEX idx_im_login_failure_account_time (account_hash, attempt_at DESC, id DESC), + INDEX idx_im_login_failure_user_time (user_id, attempt_at DESC, id DESC), + INDEX idx_im_login_failure_ip_time (ip, attempt_at DESC, id DESC), + INDEX idx_im_login_failure_type_time (failure_type, attempt_at DESC, id DESC), + INDEX idx_im_login_failure_request (request_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; + +CREATE TABLE IF NOT EXISTS im_message_admin_actions ( + id BIGINT NOT NULL AUTO_INCREMENT, + action_id VARCHAR(96) NOT NULL, + command_id VARCHAR(96) NOT NULL, + conversation_type VARCHAR(16) NOT NULL, + topic VARCHAR(64) NOT NULL, + seqid BIGINT NOT NULL, + sender_user_id BIGINT NULL, + action VARCHAR(16) NOT NULL, + action_status VARCHAR(16) NOT NULL DEFAULT 'pending', + reason VARCHAR(500) NOT NULL DEFAULT '', + actor_admin_id BIGINT NOT NULL DEFAULT 0, + request_id VARCHAR(64) NOT NULL DEFAULT '', + before_snapshot JSON NULL, + after_snapshot JSON NULL, + failure_reason VARCHAR(96) NOT NULL DEFAULT '', + failure_message VARCHAR(1000) NOT NULL DEFAULT '', + processed_at DATETIME(3) NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY (id), + UNIQUE INDEX uk_im_message_admin_action_id (action_id), + UNIQUE INDEX uk_im_message_admin_command (command_id), + INDEX idx_im_message_admin_target (conversation_type, topic, seqid, id DESC), + INDEX idx_im_message_admin_status (action_status, created_at, id), + INDEX idx_im_message_admin_actor (actor_admin_id, created_at DESC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; diff --git a/server/auth_audit.go b/server/auth_audit.go new file mode 100644 index 0000000..56b4c37 --- /dev/null +++ b/server/auth_audit.go @@ -0,0 +1,207 @@ +package main + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "net" + "strings" + "time" + + "github.com/tinode/chat/server/auth" + "github.com/tinode/chat/server/logs" + "github.com/tinode/chat/server/store" + "github.com/tinode/chat/server/store/types" +) + +func recordSessionLoginSuccess(s *Session, rec *auth.Rec, scheme, suppliedAccount, requestID string, at time.Time) { + if s == nil || rec == nil || rec.Uid.IsZero() || !s.tenantID.IsValid() { + return + } + username := "" + if user, err := store.Users.Get(s.tenantID, rec.Uid); err == nil && user != nil { + username = user.Username + } + accountType, account := loginAuditAccount(scheme, suppliedAccount, rec.Credential) + if account == "" { + accountType, account = classifyLoginAccount(username) + } + event := &types.LoginAuditEvent{ + UserID: rec.Uid, Username: username, AccountType: accountType, + AccountHash: hashLoginAccount(account), AccountMask: maskLoginAccount(account), + LoginType: loginAuditType(scheme), EventType: "login", SessionID: truncateAuditValue(s.sid, 96), + DeviceID: truncateAuditValue(s.deviceID, 255), Platform: normalizeAuditPlatform(s.platf), + AppVersion: truncateAuditValue(versionToString(s.ver), 32), IP: sessionAuditIP(s.remoteAddr), + UserAgent: truncateAuditValue(s.userAgent, 512), RiskLevel: "low", + RequestID: truncateAuditValue(requestID, 64), LoginAt: at.UTC().Round(time.Millisecond), + Metadata: map[string]any{"auth_level": rec.AuthLevel.String(), "tenant_code": s.tenantCode}, + } + if err := store.SecurityAudit.RecordLogin(s.tenantID, event); err != nil { + logs.Warn.Printf("login audit write failed tenant:%d uid:%s sid:%s err:%v", s.tenantID, rec.Uid, s.sid, err) + } +} + +func recordSessionLoginFailure(s *Session, scheme string, secret []byte, rec *auth.Rec, requestID string, cause error, at time.Time) { + if s == nil || !s.tenantID.IsValid() { + return + } + suppliedAccount := "" + if strings.EqualFold(strings.TrimSpace(scheme), "basic") { + suppliedAccount = strings.SplitN(string(secret), ":", 2)[0] + } + credential := "" + uid := types.ZeroUid + if rec != nil { + credential = rec.Credential + uid = rec.Uid + } + accountType, account := loginAuditAccount(scheme, suppliedAccount, credential) + failureType, reason, riskLevel, riskAction := classifyLoginFailure(scheme, cause) + event := &types.LoginFailureEvent{ + UserID: uid, AccountType: accountType, AccountHash: hashLoginAccount(account), AccountMask: maskLoginAccount(account), + FailureType: failureType, Reason: reason, Detail: truncateAuditValue(loginFailureDetail(cause), 500), + IP: sessionAuditIP(s.remoteAddr), DeviceID: truncateAuditValue(s.deviceID, 255), Platform: normalizeAuditPlatform(s.platf), + AppVersion: truncateAuditValue(versionToString(s.ver), 32), UserAgent: truncateAuditValue(s.userAgent, 512), + RiskLevel: riskLevel, RiskAction: riskAction, RequestID: truncateAuditValue(requestID, 64), + AttemptAt: at.UTC().Round(time.Millisecond), Metadata: map[string]any{"scheme": strings.TrimSpace(scheme), "tenant_code": s.tenantCode}, + } + if err := store.SecurityAudit.RecordLoginFailure(s.tenantID, event); err != nil { + logs.Warn.Printf("login failure audit write failed tenant:%d sid:%s err:%v", s.tenantID, s.sid, err) + } +} + +func loginAuditAccount(scheme, supplied, credential string) (string, string) { + if parts := strings.SplitN(strings.TrimSpace(credential), ":", 2); len(parts) == 2 && parts[1] != "" { + kind := strings.ToLower(strings.TrimSpace(parts[0])) + if kind == "tel" { + kind = "phone" + } + return kind, strings.TrimSpace(parts[1]) + } + if strings.EqualFold(strings.TrimSpace(scheme), "basic") { + return classifyLoginAccount(supplied) + } + return "", "" +} + +func classifyLoginAccount(account string) (string, string) { + account = strings.TrimSpace(account) + switch { + case account == "": + return "", "" + case strings.Contains(account, "@"): + return "email", account + case isPhoneLike(account): + return "phone", account + default: + return "username", account + } +} + +func isPhoneLike(value string) bool { + digits := 0 + for i, r := range value { + if r >= '0' && r <= '9' { + digits++ + continue + } + if !(i == 0 && r == '+') { + return false + } + } + return digits >= 7 +} + +func maskLoginAccount(account string) string { + account = strings.TrimSpace(account) + if account == "" { + return "" + } + if at := strings.LastIndex(account, "@"); at > 0 { + return string([]rune(account[:at])[:1]) + "***" + account[at:] + } + runes := []rune(account) + if isPhoneLike(account) && len(runes) > 7 { + return string(runes[:3]) + "****" + string(runes[len(runes)-4:]) + } + if len(runes) <= 2 { + return strings.Repeat("*", len(runes)) + } + return string(runes[:1]) + strings.Repeat("*", len(runes)-2) + string(runes[len(runes)-1:]) +} + +func hashLoginAccount(account string) string { + account = strings.ToLower(strings.TrimSpace(account)) + if account == "" { + return "" + } + sum := sha256.Sum256([]byte(account)) + return hex.EncodeToString(sum[:]) +} + +func loginAuditType(scheme string) string { + switch strings.ToLower(strings.TrimSpace(scheme)) { + case "basic": + return "password" + case "token": + return "token_refresh" + case "code", "sms", "sms_code": + return "sms_code" + case "sso": + return "sso" + case "qr", "qr_code": + return "qr_code" + default: + return "password" + } +} + +func classifyLoginFailure(scheme string, cause error) (failureType, reason, riskLevel, riskAction string) { + failureType, reason, riskLevel, riskAction = "password_error", "AUTH_FAILED", "medium", "audit_only" + detail := strings.ToLower(loginFailureDetail(cause)) + switch { + case strings.Contains(detail, "rate") || strings.Contains(detail, "too many"): + return "rate_limited", "RATE_LIMITED", "high", "blocked" + case errors.Is(cause, types.ErrPolicy) || strings.Contains(detail, "risk"): + return "risk_blocked", "RISK_BLOCKED", "high", "blocked" + case errors.Is(cause, types.ErrPermissionDenied): + return "account_disabled", "ACCOUNT_DISABLED", "high", "blocked" + case errors.Is(cause, types.ErrUserNotFound) || errors.Is(cause, types.ErrNotFound): + return "account_not_found", "ACCOUNT_NOT_FOUND", "medium", "audit_only" + case loginAuditType(scheme) == "sms_code": + return "code_error", "CODE_ERROR", "medium", "audit_only" + default: + return + } +} + +func loginFailureDetail(cause error) string { + if cause == nil { + return "authentication failed" + } + return cause.Error() +} + +func sessionAuditIP(remoteAddr string) string { + remoteAddr = strings.TrimSpace(remoteAddr) + if host, _, err := net.SplitHostPort(remoteAddr); err == nil { + return truncateAuditValue(host, 64) + } + return truncateAuditValue(strings.Trim(remoteAddr, "[]"), 64) +} + +func normalizeAuditPlatform(platform string) string { + platform = strings.ToLower(strings.TrimSpace(platform)) + if platform == "" { + return "unknown" + } + return truncateAuditValue(platform, 32) +} + +func truncateAuditValue(value string, max int) string { + runes := []rune(strings.TrimSpace(value)) + if len(runes) <= max { + return string(runes) + } + return string(runes[:max]) +} diff --git a/server/auth_audit_test.go b/server/auth_audit_test.go new file mode 100644 index 0000000..183cf76 --- /dev/null +++ b/server/auth_audit_test.go @@ -0,0 +1,89 @@ +package main + +import ( + "errors" + "strings" + "testing" + + "github.com/tinode/chat/server/store/types" +) + +func TestLoginAuditAccountDoesNotTreatTokenAsAccount(t *testing.T) { + kind, account := loginAuditAccount("token", "secret-token", "") + if kind != "" || account != "" { + t.Fatalf("token must not be persisted as an account: kind=%q account=%q", kind, account) + } + + kind, account = loginAuditAccount("basic", "13800138000", "") + if kind != "phone" || account != "13800138000" { + t.Fatalf("unexpected basic account classification: kind=%q account=%q", kind, account) + } +} + +func TestMaskAndHashLoginAccount(t *testing.T) { + tests := map[string]string{ + "13800138000": "138****8000", + "alice@example.com": "a***@example.com", + "alice": "a***e", + "ab": "**", + } + for input, want := range tests { + if got := maskLoginAccount(input); got != want { + t.Errorf("maskLoginAccount(%q)=%q want %q", input, got, want) + } + } + hash := hashLoginAccount(" Alice ") + if len(hash) != 64 || hash != hashLoginAccount("alice") || strings.Contains(hash, "alice") { + t.Fatalf("unexpected account hash %q", hash) + } +} + +func TestClassifyLoginFailure(t *testing.T) { + tests := []struct { + scheme string + err error + want string + }{ + {"basic", types.ErrFailed, "password_error"}, + {"sms_code", types.ErrFailed, "code_error"}, + {"basic", types.ErrUserNotFound, "account_not_found"}, + {"basic", types.ErrPermissionDenied, "account_disabled"}, + {"basic", types.ErrPolicy, "risk_blocked"}, + {"basic", errors.New("too many attempts"), "rate_limited"}, + } + for _, tc := range tests { + got, _, _, _ := classifyLoginFailure(tc.scheme, tc.err) + if got != tc.want { + t.Errorf("classifyLoginFailure(%q,%v)=%q want %q", tc.scheme, tc.err, got, tc.want) + } + } +} + +func TestMessageQuerySpecsExposeAdminFields(t *testing.T) { + privateSpec := privateMessagesInternalQuerySpec() + groupSpec := groupMessagesInternalQuerySpec() + for label, spec := range map[string]*internalQuerySpec{"private": privateSpec, "group": groupSpec} { + joined := strings.ToLower(spec.Select + " " + spec.From) + for _, field := range []string{"from_user_id", "content_type", "content_excerpt", "status", "im_message_admin_actions"} { + if !strings.Contains(joined, field) { + t.Errorf("%s query missing %s", label, field) + } + } + } + if strings.Contains(strings.ToLower(privateSpec.Select), "m.content as content") || strings.Contains(strings.ToLower(groupSpec.Select), "m.content as content") { + t.Fatal("message list query must not expose full message content") + } +} + +func TestRedactClientMessageForLog(t *testing.T) { + raw := []byte(`{"login":{"id":"1","scheme":"basic","secret":"YWxpY2U6YWxpY2UxMjM="},"acc":{"secret":"account-secret","tmpsecret":"temp-secret"}}`) + redacted := string(redactClientMessageForLog(raw)) + for _, secret := range []string{"YWxpY2U6YWxpY2UxMjM=", "account-secret", "temp-secret"} { + if strings.Contains(redacted, secret) { + t.Fatalf("redacted log still contains secret %q: %s", secret, redacted) + } + } + if strings.Count(redacted, "[REDACTED]") != 3 { + t.Fatalf("unexpected redacted payload: %s", redacted) + } +} diff --git a/server/db/adapter.go b/server/db/adapter.go index 5cc623b..81b9479 100644 --- a/server/db/adapter.go +++ b/server/db/adapter.go @@ -240,6 +240,13 @@ type TenantInternalAdminQueryAdapter interface { TenantInternalAdminQueryOne(tenantID t.TenantID, query string, args []any) (map[string]any, error) } +type TenantSecurityAuditAdapter interface { + TenantLoginAuditCreate(t.TenantID, *t.LoginAuditEvent) error + TenantLoginFailureCreate(t.TenantID, *t.LoginFailureEvent) error + TenantLoginAuditClose(t.TenantID, string, time.Time) error + TenantMessageAdminApply(t.TenantID, *t.MessageAdminAction) error +} + // TenantRoutedAdapter can return a tenant-specific adapter, e.g. for the // "one tenant, one database" deployment model. type TenantRoutedAdapter interface { diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index 3520094..8b43747 100644 --- a/server/db/mysql/adapter.go +++ b/server/db/mysql/adapter.go @@ -272,6 +272,9 @@ func (a *adapter) Open(jsonconfig json.RawMessage) error { if err = ensureIMHTTPInfrastructureTables(a.db); err != nil { return errors.New("mysql adapter failed to ensure HTTP infrastructure schema: " + err.Error()) } + if err = ensureIMAdminGovernanceTables(a.db); err != nil { + return errors.New("mysql adapter failed to ensure admin governance schema: " + err.Error()) + } if err = ensureIMFileReviewTables(a.db); err != nil { return errors.New("mysql adapter failed to ensure file review schema: " + err.Error()) } @@ -512,6 +515,10 @@ func (a *adapter) TenantAdapterFor(tenantID t.TenantID) (dbadapter.TenantBusines db.Close() return nil, nil, err } + if err = ensureIMAdminGovernanceTables(db); err != nil { + db.Close() + return nil, nil, err + } if err = ensureIMFileReviewTables(db); err != nil { db.Close() return nil, nil, err @@ -950,6 +957,9 @@ func (a *adapter) CreateDb(reset bool) error { if err = ensureIMHTTPInfrastructureTables(tx); err != nil { return err } + if err = ensureIMAdminGovernanceTables(tx); err != nil { + return err + } if err = ensureIMFileReviewTables(tx); err != nil { return err } diff --git a/server/db/mysql/admin_governance_schema.go b/server/db/mysql/admin_governance_schema.go new file mode 100644 index 0000000..8c07882 --- /dev/null +++ b/server/db/mysql/admin_governance_schema.go @@ -0,0 +1,64 @@ +//go:build mysql +// +build mysql + +package mysql + +func ensureIMAdminGovernanceTables(exec sqlExecutor) error { + statements := []string{ + `CREATE TABLE IF NOT EXISTS im_user_login_logs ( + id BIGINT NOT NULL AUTO_INCREMENT, user_id BIGINT NOT NULL, + username VARCHAR(64) NOT NULL DEFAULT '', account_type VARCHAR(24) NOT NULL DEFAULT '', + account_hash CHAR(64) NOT NULL DEFAULT '', account_mask VARCHAR(128) NOT NULL DEFAULT '', + login_type VARCHAR(24) NOT NULL, event_type VARCHAR(24) NOT NULL DEFAULT 'login', + session_id VARCHAR(96) NOT NULL DEFAULT '', device_id VARCHAR(255) NOT NULL DEFAULT '', + platform VARCHAR(32) NOT NULL DEFAULT 'unknown', app_version VARCHAR(32) NOT NULL DEFAULT '', + ip VARCHAR(64) NOT NULL DEFAULT '', region VARCHAR(128) NOT NULL DEFAULT '', + user_agent VARCHAR(512) NOT NULL DEFAULT '', risk_level VARCHAR(16) NOT NULL DEFAULT 'low', + request_id VARCHAR(64) NOT NULL DEFAULT '', login_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + logout_at DATETIME(3), metadata JSON, created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + PRIMARY KEY(id), INDEX idx_im_user_login_user_time(user_id, login_at DESC, id DESC), + INDEX idx_im_user_login_ip_time(ip, login_at DESC, id DESC), + INDEX idx_im_user_login_device_time(device_id(64), login_at DESC), + INDEX idx_im_user_login_request(request_id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, + `CREATE TABLE IF NOT EXISTS im_login_failure_logs ( + id BIGINT NOT NULL AUTO_INCREMENT, user_id BIGINT, + account_type VARCHAR(24) NOT NULL DEFAULT '', account_hash CHAR(64) NOT NULL DEFAULT '', + account_mask VARCHAR(128) NOT NULL DEFAULT '', failure_type VARCHAR(32) NOT NULL, + reason VARCHAR(96) NOT NULL DEFAULT '', detail VARCHAR(500) NOT NULL DEFAULT '', + ip VARCHAR(64) NOT NULL DEFAULT '', region VARCHAR(128) NOT NULL DEFAULT '', + device_id VARCHAR(255) NOT NULL DEFAULT '', platform VARCHAR(32) NOT NULL DEFAULT 'unknown', + app_version VARCHAR(32) NOT NULL DEFAULT '', user_agent VARCHAR(512) NOT NULL DEFAULT '', + risk_level VARCHAR(16) NOT NULL DEFAULT 'medium', risk_action VARCHAR(24) NOT NULL DEFAULT 'audit_only', + request_id VARCHAR(64) NOT NULL DEFAULT '', attempt_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + metadata JSON, created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + PRIMARY KEY(id), INDEX idx_im_login_failure_account_time(account_hash, attempt_at DESC, id DESC), + INDEX idx_im_login_failure_user_time(user_id, attempt_at DESC, id DESC), + INDEX idx_im_login_failure_ip_time(ip, attempt_at DESC, id DESC), + INDEX idx_im_login_failure_type_time(failure_type, attempt_at DESC, id DESC), + INDEX idx_im_login_failure_request(request_id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, + `CREATE TABLE IF NOT EXISTS im_message_admin_actions ( + id BIGINT NOT NULL AUTO_INCREMENT, action_id VARCHAR(96) NOT NULL, + command_id VARCHAR(96) NOT NULL, conversation_type VARCHAR(16) NOT NULL, + topic VARCHAR(64) NOT NULL, seqid BIGINT NOT NULL, sender_user_id BIGINT, + action VARCHAR(16) NOT NULL, action_status VARCHAR(16) NOT NULL DEFAULT 'pending', + reason VARCHAR(500) NOT NULL DEFAULT '', actor_admin_id BIGINT NOT NULL DEFAULT 0, + request_id VARCHAR(64) NOT NULL DEFAULT '', before_snapshot JSON, after_snapshot JSON, + failure_reason VARCHAR(96) NOT NULL DEFAULT '', failure_message VARCHAR(1000) NOT NULL DEFAULT '', + processed_at DATETIME(3), created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY(id), UNIQUE INDEX uk_im_message_admin_action_id(action_id), + UNIQUE INDEX uk_im_message_admin_command(command_id), + INDEX idx_im_message_admin_target(conversation_type, topic, seqid, id DESC), + INDEX idx_im_message_admin_status(action_status, created_at, id), + INDEX idx_im_message_admin_actor(actor_admin_id, created_at DESC) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, + } + for _, statement := range statements { + if _, err := exec.Exec(statement); err != nil { + return err + } + } + return nil +} diff --git a/server/db/mysql/admin_governance_schema_test.go b/server/db/mysql/admin_governance_schema_test.go new file mode 100644 index 0000000..e2075a0 --- /dev/null +++ b/server/db/mysql/admin_governance_schema_test.go @@ -0,0 +1,28 @@ +//go:build mysql +// +build mysql + +package mysql + +import ( + "strings" + "testing" +) + +func TestEnsureIMAdminGovernanceTablesUsesSelectedTenantDatabase(t *testing.T) { + exec := &recordingGroupSchemaExecutor{} + if err := ensureIMAdminGovernanceTables(exec); err != nil { + t.Fatal(err) + } + if len(exec.statements) != 3 { + t.Fatalf("created %d admin governance tables, want 3", len(exec.statements)) + } + joined := strings.ToLower(strings.Join(exec.statements, "\n")) + for _, table := range []string{"im_user_login_logs", "im_login_failure_logs", "im_message_admin_actions"} { + if !strings.Contains(joined, table) { + t.Fatalf("missing table %s", table) + } + } + if strings.Contains(joined, "tinode.") || strings.Contains(joined, "create database") || strings.Contains(joined, "tenant_id") { + t.Fatalf("admin governance schema must stay inside the selected tenant database: %s", joined) + } +} diff --git a/server/db/mysql/tenant_security_audit.go b/server/db/mysql/tenant_security_audit.go new file mode 100644 index 0000000..31ab894 --- /dev/null +++ b/server/db/mysql/tenant_security_audit.go @@ -0,0 +1,166 @@ +//go:build mysql +// +build mysql + +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "strings" + "time" + + "github.com/jmoiron/sqlx" + t "github.com/tinode/chat/server/store/types" +) + +func (a *adapter) TenantLoginAuditCreate(id t.TenantID, event *t.LoginAuditEvent) error { + tenantAdp, err := a.routedTenantAdapter(id) + if err != nil || event == nil || event.UserID.IsZero() { + return t.ErrMalformed + } + if tenantAdp != a { + return tenantAdp.TenantLoginAuditCreate(id, event) + } + metadata, err := json.Marshal(event.Metadata) + if err != nil { + return err + } + _, err = a.db.Exec(`INSERT INTO im_user_login_logs + (user_id,username,account_type,account_hash,account_mask,login_type,event_type,session_id,device_id,platform,app_version,ip,region,user_agent,risk_level,request_id,login_at,logout_at,metadata) + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, + decodeUID(event.UserID), event.Username, event.AccountType, event.AccountHash, event.AccountMask, + event.LoginType, event.EventType, event.SessionID, event.DeviceID, event.Platform, event.AppVersion, + event.IP, event.Region, event.UserAgent, event.RiskLevel, event.RequestID, event.LoginAt, event.LogoutAt, nullableJSON(metadata)) + return err +} + +func (a *adapter) TenantLoginFailureCreate(id t.TenantID, event *t.LoginFailureEvent) error { + tenantAdp, err := a.routedTenantAdapter(id) + if err != nil || event == nil { + return t.ErrMalformed + } + if tenantAdp != a { + return tenantAdp.TenantLoginFailureCreate(id, event) + } + metadata, err := json.Marshal(event.Metadata) + if err != nil { + return err + } + var userID any + if !event.UserID.IsZero() { + userID = decodeUID(event.UserID) + } + _, err = a.db.Exec(`INSERT INTO im_login_failure_logs + (user_id,account_type,account_hash,account_mask,failure_type,reason,detail,ip,region,device_id,platform,app_version,user_agent,risk_level,risk_action,request_id,attempt_at,metadata) + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, + userID, event.AccountType, event.AccountHash, event.AccountMask, event.FailureType, event.Reason, event.Detail, + event.IP, event.Region, event.DeviceID, event.Platform, event.AppVersion, event.UserAgent, + event.RiskLevel, event.RiskAction, event.RequestID, event.AttemptAt, nullableJSON(metadata)) + return err +} + +func (a *adapter) TenantLoginAuditClose(id t.TenantID, sessionID string, at time.Time) error { + tenantAdp, err := a.routedTenantAdapter(id) + if err != nil || strings.TrimSpace(sessionID) == "" { + return t.ErrMalformed + } + if tenantAdp != a { + return tenantAdp.TenantLoginAuditClose(id, sessionID, at) + } + _, err = a.db.Exec("UPDATE im_user_login_logs SET logout_at=? WHERE session_id=? AND logout_at IS NULL", at, sessionID) + return err +} + +func (a *adapter) TenantMessageAdminApply(id t.TenantID, action *t.MessageAdminAction) error { + tenantAdp, err := a.routedTenantAdapter(id) + if err != nil || action == nil || action.Topic == "" || action.SeqID <= 0 || (action.Action != "hide" && action.Action != "delete") { + return t.ErrMalformed + } + if tenantAdp != a { + return tenantAdp.TenantMessageAdminApply(id, action) + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + tx, err := a.db.BeginTxx(ctx, nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + var existing int64 + if err = tx.GetContext(ctx, &existing, "SELECT COUNT(*) FROM im_message_admin_actions WHERE command_id=?", action.CommandID); err != nil { + return err + } + if existing > 0 { + return tx.Commit() + } + + sender, beforeStatus, err := lockAdminMessage(ctx, tx, action.ConversationType, action.Topic, action.SeqID) + if err != nil { + return err + } + before, _ := json.Marshal(map[string]any{"status": beforeStatus, "sender_user_id": sender}) + afterStatus := "hidden" + if action.Action == "delete" { + afterStatus = "deleted" + } + if action.ConversationType == "group" { + _, err = tx.ExecContext(ctx, "UPDATE im_group_messages SET status=?,deleted_at=?,updated_at=? WHERE topic=? AND seqid=?", afterStatus, action.ProcessedAt, action.ProcessedAt, action.Topic, action.SeqID) + } else { + _, err = tx.ExecContext(ctx, "UPDATE im_messages SET deletedat=?,updatedat=? WHERE topic=? AND seqid=?", action.ProcessedAt, action.ProcessedAt, action.Topic, action.SeqID) + } + if err != nil { + return err + } + after, _ := json.Marshal(map[string]any{"status": afterStatus, "sender_user_id": sender}) + _, err = tx.ExecContext(ctx, `INSERT INTO im_message_admin_actions + (action_id,command_id,conversation_type,topic,seqid,sender_user_id,action,action_status,reason,actor_admin_id,request_id,before_snapshot,after_snapshot,processed_at) + VALUES(?,?,?,?,?,?,?,?,?,?,?,CAST(? AS JSON),CAST(? AS JSON),?)`, + action.ActionID, action.CommandID, action.ConversationType, action.Topic, action.SeqID, sender, + action.Action, "success", action.Reason, action.ActorAdminID, action.RequestID, string(before), string(after), action.ProcessedAt) + if err != nil { + return err + } + return tx.Commit() +} + +func lockAdminMessage(ctx context.Context, tx *sqlx.Tx, conversationType, topic string, seqID int64) (int64, string, error) { + if conversationType == "group" { + var row struct { + Sender int64 `db:"from_user"` + Status string `db:"status"` + } + err := tx.GetContext(ctx, &row, "SELECT from_user,status FROM im_group_messages WHERE topic=? AND seqid=? FOR UPDATE", topic, seqID) + if errors.Is(err, sql.ErrNoRows) { + return 0, "", t.ErrNotFound + } + return row.Sender, row.Status, err + } + var row struct { + Sender int64 `db:"from"` + DeletedAt sql.NullTime `db:"deletedat"` + RecalledAt sql.NullTime `db:"recalled_at"` + } + err := tx.GetContext(ctx, &row, "SELECT `from`,deletedat,recalled_at FROM im_messages WHERE topic=? AND seqid=? FOR UPDATE", topic, seqID) + if errors.Is(err, sql.ErrNoRows) { + return 0, "", t.ErrNotFound + } + status := "normal" + if row.RecalledAt.Valid { + status = "recalled" + } + if row.DeletedAt.Valid { + status = "deleted" + } + return row.Sender, status, err +} + +func nullableJSON(raw []byte) any { + if len(raw) == 0 || string(raw) == "null" { + return nil + } + return string(raw) +} diff --git a/server/internal_admin_query_resources.go b/server/internal_admin_query_resources.go index fd472c2..df34ea5 100644 --- a/server/internal_admin_query_resources.go +++ b/server/internal_admin_query_resources.go @@ -60,6 +60,18 @@ var internalGroupResourceSpecs = map[string]*internalQuerySpec{ } var internalTenantResourceSpecs = map[string]*internalQuerySpec{ + "private-messages": privateMessagesInternalQuerySpec(), + "group-messages": groupMessagesInternalQuerySpec(), + "user-login-logs": { + Select: `SELECT id,CAST(user_id AS CHAR) AS user_id,username,account_type,account_mask,login_type,event_type,session_id,device_id,platform,app_version,ip,region,user_agent,risk_level,request_id,login_at,logout_at,metadata`, + From: "FROM im_user_login_logs", DefaultSort: "ORDER BY id DESC", DetailKey: "id", + Filters: []internalQueryFilter{{"user_id", "user_id", "=", "uid"}, {"account", "account_mask", "LIKE", "string"}, {"login_type", "login_type", "=", "string"}, {"event_type", "event_type", "=", "string"}, {"platform", "platform", "=", "string"}, {"ip", "ip", "LIKE", "string"}, {"login_at_start", "login_at", ">=", "time"}, {"login_at_end", "login_at", "<=", "time"}}, + }, + "login-failure-logs": { + Select: `SELECT id,CAST(user_id AS CHAR) AS user_id,account_type,account_mask,failure_type,reason,detail,ip,region,device_id,platform,app_version,user_agent,risk_level,risk_action,request_id,attempt_at,metadata`, + From: "FROM im_login_failure_logs", DefaultSort: "ORDER BY id DESC", DetailKey: "id", + Filters: []internalQueryFilter{{"user_id", "user_id", "=", "uid"}, {"account", "account_mask", "LIKE", "string"}, {"failure_type", "failure_type", "=", "string"}, {"platform", "platform", "=", "string"}, {"ip", "ip", "LIKE", "string"}, {"attempt_at_start", "attempt_at", ">=", "time"}, {"attempt_at_end", "attempt_at", "<=", "time"}}, + }, "audit-logs": { Select: "SELECT id,request_id,actor_user_id,action,target_type,target_id,ip,user_agent,detail,created_at", From: "FROM im_audit_logs", DefaultSort: "ORDER BY id DESC", DetailKey: "id", @@ -278,6 +290,86 @@ func groupsInternalQuerySpec() *internalQuerySpec { } } +func privateMessagesInternalQuerySpec() *internalQuerySpec { + rawContentType := `COALESCE(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(m.head,'$."x-im-type"')),''),NULLIF(JSON_UNQUOTE(JSON_EXTRACT(m.head,'$.type')),''),CASE WHEN JSON_EXTRACT(m.content,'$.callId') IS NOT NULL THEN 'call' ELSE 'text' END)` + contentType := `CASE ` + rawContentType + ` WHEN 'call' THEN 'system' WHEN 'system_notice' THEN 'system' WHEN 'contact_card' THEN 'card' WHEN 'forward_bundle' THEN 'card' WHEN 'link' THEN 'text' ELSE ` + rawContentType + ` END` + contentExcerpt := `LEFT(COALESCE(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(m.content,'$.txt')),''),CASE JSON_TYPE(m.content) WHEN 'STRING' THEN JSON_UNQUOTE(m.content) ELSE CONCAT('[',` + contentType + `,']') END,''),500)` + status := `CASE WHEN aa.action='delete' THEN 'deleted' WHEN aa.action='hide' THEN 'hidden' WHEN m.recalled_at IS NOT NULL THEN 'recalled' WHEN m.deletedat IS NOT NULL THEN 'deleted' ELSE 'normal' END` + return &internalQuerySpec{ + Select: `SELECT m.topic,m.seqid AS seq_id,CAST(m.` + "`from`" + ` AS CHAR) AS from_user_id, + COALESCE(su.username,'') AS sender_username,COALESCE(sp.nickname,'') AS sender_nickname, + CAST(rs.userid AS CHAR) AS to_user_id,COALESCE(ru.username,'') AS receiver_username,COALESCE(rp.nickname,'') AS receiver_nickname, + ` + contentType + ` AS content_type,` + contentExcerpt + ` AS content_excerpt, + ` + status + ` AS status,m.createdat AS created_at,m.updatedat AS updated_at`, + From: `FROM (SELECT * FROM im_messages WHERE topic LIKE 'p2p%') AS m + LEFT JOIN im_users AS su ON su.id=m.` + "`from`" + ` + LEFT JOIN im_user_profiles AS sp ON sp.user_id=m.` + "`from`" + ` + LEFT JOIN im_subscriptions AS rs ON rs.id=(SELECT MIN(s2.id) FROM im_subscriptions AS s2 WHERE s2.topic=m.topic AND s2.userid<>m.` + "`from`" + `) + LEFT JOIN im_users AS ru ON ru.id=rs.userid + LEFT JOIN im_user_profiles AS rp ON rp.user_id=rs.userid + LEFT JOIN im_message_admin_actions AS aa ON aa.id=(SELECT MAX(a2.id) FROM im_message_admin_actions AS a2 WHERE a2.conversation_type='private' AND a2.topic=m.topic AND a2.seqid=m.seqid AND a2.action_status='success')`, + DefaultSort: "ORDER BY m.id DESC", + Filters: []internalQueryFilter{{"topic", "m.topic", "LIKE", "string"}, {"from_user_id", "m.`from`", "=", "uid"}, {"to_user_id", "rs.userid", "=", "uid"}, {"content_type", contentType, "=", "string"}, {"status", status, "=", "string"}, {"keyword", "CAST(m.content AS CHAR)", "LIKE", "string"}, {"created_at_start", "m.createdat", ">=", "time"}, {"created_at_end", "m.createdat", "<=", "time"}}, + } +} + +func groupMessagesInternalQuerySpec() *internalQuerySpec { + status := `CASE WHEN aa.action='delete' THEN 'deleted' WHEN aa.action='hide' THEN 'hidden' WHEN m.status='recalled' THEN 'recalled' WHEN m.status IN ('deleted','blocked') THEN m.status ELSE 'normal' END` + contentType := `CASE m.msg_type WHEN 'call' THEN 'system' WHEN 'system_notice' THEN 'system' WHEN 'contact_card' THEN 'card' WHEN 'forward_bundle' THEN 'card' WHEN 'link' THEN 'text' ELSE m.msg_type END` + contentExcerpt := `LEFT(COALESCE(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(m.content,'$.txt')),''),CASE JSON_TYPE(m.content) WHEN 'STRING' THEN JSON_UNQUOTE(m.content) ELSE CONCAT('[',` + contentType + `,']') END,''),500)` + return &internalQuerySpec{ + Select: `SELECT m.topic,m.seqid AS seq_id,CAST(m.from_user AS CHAR) AS from_user_id, + COALESCE(u.username,'') AS sender_username,COALESCE(p.nickname,'') AS sender_nickname, + COALESCE(gs.name,'') AS group_name,m.topic AS group_no,` + contentType + ` AS content_type, + ` + contentExcerpt + ` AS content_excerpt, + ` + status + ` AS status,COALESCE(m.reply_seqid,0) AS reply_to_seq,m.mentioned_user_ids AS at_users, + m.created_at,m.updated_at`, + From: `FROM im_group_messages AS m + LEFT JOIN im_users AS u ON u.id=m.from_user + LEFT JOIN im_user_profiles AS p ON p.user_id=m.from_user + LEFT JOIN im_group_settings AS gs ON gs.topic=m.topic COLLATE utf8mb4_0900_ai_ci + LEFT JOIN im_message_admin_actions AS aa ON aa.id=(SELECT MAX(a2.id) FROM im_message_admin_actions AS a2 WHERE a2.conversation_type='group' AND a2.topic=m.topic COLLATE utf8mb4_0900_ai_ci AND a2.seqid=m.seqid AND a2.action_status='success')`, + DefaultSort: "ORDER BY m.id DESC", + Filters: []internalQueryFilter{{"topic", "m.topic", "LIKE", "string"}, {"sender_user_id", "m.from_user", "=", "uid"}, {"content_type", contentType, "=", "string"}, {"status", status, "=", "string"}, {"keyword", "CAST(m.content AS CHAR)", "LIKE", "string"}, {"created_at_start", "m.created_at", ">=", "time"}, {"created_at_end", "m.created_at", "<=", "time"}}, + } +} + +func serveInternalDashboardOverview(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID) { + row, err := store.InternalAdminQueries.QueryOne(tenantID, `SELECT + (SELECT COUNT(*) FROM im_users WHERE state<>20) AS total_users, + (SELECT COUNT(*) FROM im_users WHERE state=0 AND lastseen>=UTC_DATE()) AS active_users, + (SELECT COUNT(*) FROM im_group_settings WHERE group_status='active') AS total_groups, + (SELECT COUNT(*) FROM im_topics WHERE state<>20 AND (name LIKE 'p2p%' OR name LIKE 'grp%')) AS total_conversations, + ((SELECT COUNT(*) FROM im_messages WHERE topic LIKE 'p2p%')+(SELECT COUNT(*) FROM im_group_messages)) AS total_messages, + ((SELECT COUNT(*) FROM im_messages WHERE topic LIKE 'p2p%' AND createdat>=UTC_DATE())+(SELECT COUNT(*) FROM im_group_messages WHERE created_at>=UTC_DATE())) AS today_messages`, nil) + if err != nil { + writeInternalQueryError(wrt, req, err) + return + } + writeIMSuccess(wrt, req, normalizeInternalAdminRowForAPI(row)) +} + +func serveInternalDashboardTrends(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID) { + days := parseIMPositiveInt(req.URL.Query().Get("days"), 7) + if days > 31 { + days = 31 + } + query := fmt.Sprintf(`WITH RECURSIVE dates AS ( + SELECT UTC_DATE()-INTERVAL %d DAY AS day + UNION ALL SELECT day+INTERVAL 1 DAY FROM dates WHERE day=d.day AND u.createdat=d.day AND pm.createdat=d.day AND gm.created_at 96 { + actionID = actionID[:96] + } + requestID := strings.TrimSpace(envelope.TraceID) + if len(requestID) > 64 { + requestID = requestID[:64] + } + if err := store.SecurityAudit.ApplyMessageAction(tenantID, &types.MessageAdminAction{ + ActionID: actionID, CommandID: commandID, ConversationType: conversationType, + Topic: topic, SeqID: payload.SeqID, Action: action, Reason: strings.TrimSpace(payload.Reason), + ActorAdminID: actorAdminID, RequestID: requestID, ProcessedAt: now, + }); err != nil { + return nil, err + } + if conversationType == "group" { + if err := store.InvalidateHotMessages(tenantID, topic, []types.Range{{Low: payload.SeqID}}); err != nil { + logs.Warn.Printf("message admin cache invalidation failed tenant:%d topic:%s seq:%d err:%v", tenantID, topic, payload.SeqID, err) + } + } + return map[string]any{ + "conversation_type": conversationType, "topic": topic, "seq_id": payload.SeqID, + "action": action, "status": "success", "processed_at": now, + }, nil +} + func executeKafkaUserGovernanceCommand(tenantID types.TenantID, envelope kafkaEnvelope, payload adminCommandPayload, actor types.Uid) (map[string]any, error) { target := parseKafkaCommandUser(firstNonEmpty(payload.TargetID, payload.UserID)) if target.IsZero() { diff --git a/server/kafka_admin_message_test.go b/server/kafka_admin_message_test.go new file mode 100644 index 0000000..348b23a --- /dev/null +++ b/server/kafka_admin_message_test.go @@ -0,0 +1,57 @@ +package main + +import ( + "testing" + "time" + + "github.com/tinode/chat/server/store" + "github.com/tinode/chat/server/store/types" +) + +type messageAdminAuditRecorder struct { + action *types.MessageAdminAction +} + +func (r *messageAdminAuditRecorder) RecordLogin(types.TenantID, *types.LoginAuditEvent) error { + return nil +} + +func (r *messageAdminAuditRecorder) RecordLoginFailure(types.TenantID, *types.LoginFailureEvent) error { + return nil +} + +func (r *messageAdminAuditRecorder) CloseLogin(types.TenantID, string, time.Time) error { + return nil +} + +func (r *messageAdminAuditRecorder) ApplyMessageAction(_ types.TenantID, action *types.MessageAdminAction) error { + r.action = action + return nil +} + +func TestExecuteKafkaMessageAdminCommand(t *testing.T) { + previous := store.SecurityAudit + recorder := &messageAdminAuditRecorder{} + store.SecurityAudit = recorder + t.Cleanup(func() { store.SecurityAudit = previous }) + + result, err := executeKafkaMessageAdminCommand(1, kafkaEnvelope{ + EventID: "event-1", TraceID: "trace-1", EventType: adminCommandMessageHide, + Actor: &kafkaActor{Type: "admin", ID: "42", DisplayName: "operator"}, + }, adminCommandPayload{ + CommandID: "command-1", ConversationType: "private", Topic: "p2ptest", SeqID: 7, Reason: "sensitive content", + }) + if err != nil { + t.Fatal(err) + } + if recorder.action == nil { + t.Fatal("message action was not persisted") + } + if recorder.action.ActionID != "action_event-1" || recorder.action.CommandID != "command-1" || + recorder.action.ActorAdminID != 42 || recorder.action.Action != "hide" || recorder.action.Topic != "p2ptest" || recorder.action.SeqID != 7 { + t.Fatalf("unexpected persisted action: %+v", recorder.action) + } + if result["status"] != "success" || result["action"] != "hide" { + t.Fatalf("unexpected command result: %+v", result) + } +} diff --git a/server/session.go b/server/session.go index 28c7d25..6a180ea 100644 --- a/server/session.go +++ b/server/session.go @@ -461,6 +461,9 @@ func (s *Session) purgeChannels() { func (s *Session) cleanUp(expired bool) { atomic.StoreInt32(&s.terminating, 1) if !s.uid.IsZero() && s.tenantID.IsValid() { + if err := store.SecurityAudit.CloseLogin(s.tenantID, s.sid, time.Now().UTC().Round(time.Millisecond)); err != nil { + logs.Warn.Println("s.cleanUp: failed to close login audit", s.uid, err, s.sid) + } if err := store.MarkUserOffline(s.tenantID, s.uid, s.sid, currentNodeID()); err != nil { logs.Warn.Println("s.cleanUp: failed to mark user offline", s.uid, err, s.sid) } @@ -498,24 +501,46 @@ func (s *Session) dispatchRaw(raw []byte) { return } - toLog := raw - truncated := "" - if len(raw) > 512 { - toLog = raw[:512] - truncated = "<...>" - } - logs.Info.Printf("in: '%s%s' sid='%s' uid='%s'", toLog, truncated, s.sid, s.uid) - if err := json.Unmarshal(raw, &msg); err != nil { // Malformed message logs.Warn.Println("s.dispatch", err, s.sid) s.queueOut(ErrMalformed("", "", now)) return } + toLog := redactClientMessageForLog(raw) + truncated := "" + if len(toLog) > 512 { + toLog = toLog[:512] + truncated = "<...>" + } + logs.Info.Printf("in: '%s%s' sid='%s' uid='%s'", toLog, truncated, s.sid, s.uid) s.dispatch(&msg) } +func redactClientMessageForLog(raw []byte) []byte { + var payload map[string]any + if json.Unmarshal(raw, &payload) != nil { + return []byte("[malformed payload]") + } + for _, section := range []string{"login", "acc"} { + body, ok := payload[section].(map[string]any) + if !ok { + continue + } + for _, field := range []string{"secret", "tmpsecret"} { + if _, exists := body[field]; exists { + body[field] = "[REDACTED]" + } + } + } + redacted, err := json.Marshal(payload) + if err != nil { + return []byte("[unprintable payload]") + } + return redacted +} + func (s *Session) dispatch(msg *ClientComMessage) { now := types.TimeNow() atomic.StoreInt64(&s.lastAction, now.UnixNano()) @@ -1089,6 +1114,7 @@ func (s *Session) login(msg *ClientComMessage) { handler := store.Store.GetLogicalAuthHandler(msg.Login.Scheme) if handler == nil { + recordSessionLoginFailure(s, msg.Login.Scheme, msg.Login.Secret, nil, msg.Id, types.ErrUnsupported, msg.Timestamp) logs.Warn.Println("s.login: unknown authentication scheme", msg.Login.Scheme, s.sid) s.queueOut(ErrAuthUnknownScheme(msg.Id, "", msg.Timestamp)) return @@ -1097,6 +1123,7 @@ func (s *Session) login(msg *ClientComMessage) { rec, challenge, err := handler.Authenticate( auth.AuthContext{TenantID: s.tenantID, RemoteAddr: s.remoteAddr}, msg.Login.Secret) if err != nil { + recordSessionLoginFailure(s, msg.Login.Scheme, msg.Login.Secret, rec, msg.Id, err, msg.Timestamp) resp := decodeStoreError(err, msg.Id, msg.Timestamp, nil) if resp.Ctrl.Code >= 500 { // Log internal errors @@ -1107,6 +1134,7 @@ func (s *Session) login(msg *ClientComMessage) { } if rec == nil || rec.TenantID != s.tenantID { + recordSessionLoginFailure(s, msg.Login.Scheme, msg.Login.Secret, rec, msg.Id, types.ErrFailed, msg.Timestamp) logs.Warn.Println("s.login: authenticator returned invalid tenant", s.sid) s.queueOut(decodeStoreError(types.ErrFailed, msg.Id, msg.Timestamp, nil)) return @@ -1121,6 +1149,7 @@ func (s *Session) login(msg *ClientComMessage) { } if err != nil { + recordSessionLoginFailure(s, msg.Login.Scheme, msg.Login.Secret, rec, msg.Id, err, msg.Timestamp) logs.Warn.Println("s.login: user state check failed", rec.Uid, err, s.sid) s.queueOut(decodeStoreError(err, msg.Id, msg.Timestamp, nil)) return @@ -1142,10 +1171,11 @@ func (s *Session) login(msg *ClientComMessage) { } } if err != nil { + recordSessionLoginFailure(s, msg.Login.Scheme, msg.Login.Secret, rec, msg.Id, err, msg.Timestamp) logs.Warn.Println("s.login: failed to validate credentials:", err, s.sid) s.queueOut(decodeStoreError(err, msg.Id, msg.Timestamp, nil)) } else { - s.queueOut(s.onLogin(msg.Id, msg.Timestamp, rec, missing)) + s.queueOut(s.onLogin(msg.Id, msg.Timestamp, rec, missing, msg.Login.Scheme, strings.SplitN(string(msg.Login.Secret), ":", 2)[0])) } } @@ -1209,7 +1239,7 @@ func (s *Session) authSecretReset(params []byte) error { } // onLogin performs steps after successful authentication. -func (s *Session) onLogin(msgID string, timestamp time.Time, rec *auth.Rec, missing []string) *ServerComMessage { +func (s *Session) onLogin(msgID string, timestamp time.Time, rec *auth.Rec, missing []string, scheme, account string) *ServerComMessage { var reply *ServerComMessage var params map[string]any @@ -1266,6 +1296,9 @@ func (s *Session) onLogin(msgID string, timestamp time.Time, rec *auth.Rec, miss logs.Warn.Println("failed to update device record", err) } } + if !s.uid.IsZero() { + recordSessionLoginSuccess(s, rec, scheme, account, msgID, timestamp) + } } // GenSecret fails only if tokenLifetime is < 0. It can't be < 0 here, diff --git a/server/store/security_audit.go b/server/store/security_audit.go new file mode 100644 index 0000000..7026e95 --- /dev/null +++ b/server/store/security_audit.go @@ -0,0 +1,63 @@ +package store + +import ( + "time" + + adapter "github.com/tinode/chat/server/db" + "github.com/tinode/chat/server/store/types" +) + +type SecurityAuditPersistenceInterface interface { + RecordLogin(types.TenantID, *types.LoginAuditEvent) error + RecordLoginFailure(types.TenantID, *types.LoginFailureEvent) error + CloseLogin(types.TenantID, string, time.Time) error + ApplyMessageAction(types.TenantID, *types.MessageAdminAction) error +} + +type securityAuditMapper struct{} + +var SecurityAudit SecurityAuditPersistenceInterface = securityAuditMapper{} + +func tenantSecurityAuditAdapter(tenantID types.TenantID) (adapter.TenantSecurityAuditAdapter, error) { + business, err := tenantBusinessAdapter(tenantID) + if err != nil { + return nil, err + } + audit, ok := business.(adapter.TenantSecurityAuditAdapter) + if !ok { + return nil, types.ErrUnsupported + } + return audit, nil +} + +func (securityAuditMapper) RecordLogin(tenantID types.TenantID, event *types.LoginAuditEvent) error { + audit, err := tenantSecurityAuditAdapter(tenantID) + if err != nil { + return err + } + return audit.TenantLoginAuditCreate(tenantID, event) +} + +func (securityAuditMapper) RecordLoginFailure(tenantID types.TenantID, event *types.LoginFailureEvent) error { + audit, err := tenantSecurityAuditAdapter(tenantID) + if err != nil { + return err + } + return audit.TenantLoginFailureCreate(tenantID, event) +} + +func (securityAuditMapper) CloseLogin(tenantID types.TenantID, sessionID string, at time.Time) error { + audit, err := tenantSecurityAuditAdapter(tenantID) + if err != nil { + return err + } + return audit.TenantLoginAuditClose(tenantID, sessionID, at) +} + +func (securityAuditMapper) ApplyMessageAction(tenantID types.TenantID, action *types.MessageAdminAction) error { + audit, err := tenantSecurityAuditAdapter(tenantID) + if err != nil { + return err + } + return audit.TenantMessageAdminApply(tenantID, action) +} diff --git a/server/store/store.go b/server/store/store.go index 3f64198..65b9b96 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -4714,6 +4714,7 @@ func init() { RegistrationConfig = registrationConfigMapper{} TenantEntry = tenantEntryMapper{} InternalAdminQueries = internalAdminQueryMapper{} + SecurityAudit = securityAuditMapper{} Users = usersMapper{} Topics = topicsMapper{} Subs = subsMapper{} diff --git a/server/store/types/admin_governance.go b/server/store/types/admin_governance.go new file mode 100644 index 0000000..55d47d9 --- /dev/null +++ b/server/store/types/admin_governance.go @@ -0,0 +1,59 @@ +package types + +import "time" + +type LoginAuditEvent struct { + UserID Uid + Username string + AccountType string + AccountHash string + AccountMask string + LoginType string + EventType string + SessionID string + DeviceID string + Platform string + AppVersion string + IP string + Region string + UserAgent string + RiskLevel string + RequestID string + LoginAt time.Time + LogoutAt *time.Time + Metadata map[string]any +} + +type LoginFailureEvent struct { + UserID Uid + AccountType string + AccountHash string + AccountMask string + FailureType string + Reason string + Detail string + IP string + Region string + DeviceID string + Platform string + AppVersion string + UserAgent string + RiskLevel string + RiskAction string + RequestID string + AttemptAt time.Time + Metadata map[string]any +} + +type MessageAdminAction struct { + ActionID string + CommandID string + ConversationType string + Topic string + SeqID int64 + Action string + Reason string + ActorAdminID int64 + RequestID string + ProcessedAt time.Time +} diff --git a/server/user.go b/server/user.go index ed6dffa..8d575c9 100644 --- a/server/user.go +++ b/server/user.go @@ -307,7 +307,7 @@ func replyCreateUser(s *Session, msg *ClientComMessage, rec *auth.Rec) { if msg.Acc.Login { // Process user's login request. _, missing, _ := stringSliceDelta(globals.authValidators[rec.AuthLevel], validated) - reply = s.onLogin(msg.Id, msg.Timestamp, rec, missing) + reply = s.onLogin(msg.Id, msg.Timestamp, rec, missing, msg.Acc.Scheme, account) } else { // Not using the new account for logging in. reply = NoErrCreated(msg.Id, "", msg.Timestamp)