611 lines
22 KiB
Go
611 lines
22 KiB
Go
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"crypto/hmac"
|
|
"crypto/sha256"
|
|
"encoding/base64"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/url"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/tinode/chat/server/logs"
|
|
"github.com/tinode/chat/server/store"
|
|
"github.com/tinode/chat/server/store/types"
|
|
)
|
|
|
|
const (
|
|
internalCapabilityAdminQuery = "admin_query"
|
|
|
|
internalHeaderService = "X-Service-Name"
|
|
internalHeaderTimestamp = "X-Timestamp"
|
|
internalHeaderNonce = "X-Nonce"
|
|
internalHeaderSignature = "X-Signature"
|
|
)
|
|
|
|
type internalAPIConfig struct {
|
|
Enabled bool `json:"enabled"`
|
|
TimeWindowSeconds int `json:"time_window_seconds"`
|
|
Services map[string]internalAPIServiceConfig `json:"services"`
|
|
}
|
|
|
|
type internalAPIServiceConfig struct {
|
|
Secret string `json:"secret"`
|
|
Allow []string `json:"allow"`
|
|
}
|
|
|
|
type internalAPIRuntime struct {
|
|
config internalAPIConfig
|
|
nonces sync.Map
|
|
}
|
|
|
|
var internalAPI internalAPIRuntime
|
|
|
|
func initInternalAPI(raw json.RawMessage) error {
|
|
internalAPI = internalAPIRuntime{}
|
|
if len(raw) == 0 || string(raw) == "null" {
|
|
return nil
|
|
}
|
|
var cfg internalAPIConfig
|
|
if err := json.Unmarshal(raw, &cfg); err != nil {
|
|
return err
|
|
}
|
|
if cfg.TimeWindowSeconds <= 0 {
|
|
cfg.TimeWindowSeconds = 300
|
|
}
|
|
if cfg.Services == nil {
|
|
cfg.Services = map[string]internalAPIServiceConfig{}
|
|
}
|
|
internalAPI.config = cfg
|
|
if cfg.Enabled {
|
|
logs.Info.Printf("Internal API enabled for %d service(s)", len(cfg.Services))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func serveInternalAdminQuery(wrt http.ResponseWriter, req *http.Request, prefix string) {
|
|
if req.Method != http.MethodGet {
|
|
writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil)
|
|
return
|
|
}
|
|
if !authorizeInternalAPI(wrt, req, internalCapabilityAdminQuery) {
|
|
return
|
|
}
|
|
|
|
tail := strings.Trim(strings.TrimPrefix(req.URL.Path, prefix), "/")
|
|
parts := splitInternalPath(tail)
|
|
if len(parts) < 3 || parts[0] != "tenants" {
|
|
writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil)
|
|
return
|
|
}
|
|
tenant, ok := resolveInternalTenant(wrt, req, parts[1])
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
switch {
|
|
case len(parts) == 3 && parts[2] == "users":
|
|
serveInternalAdminResourceList(wrt, req, tenant.ID, "users", "")
|
|
case len(parts) == 4 && parts[2] == "users":
|
|
serveInternalUserQuery(wrt, req, tenant.ID, parts[3])
|
|
case len(parts) == 5 && parts[2] == "users" && parts[4] == "devices":
|
|
serveInternalAdminResourceList(wrt, req, tenant.ID, "user_devices", parts[3])
|
|
case len(parts) == 3 && parts[2] == "groups":
|
|
serveInternalAdminResourceList(wrt, req, tenant.ID, "groups", "")
|
|
case len(parts) == 4 && parts[2] == "groups":
|
|
serveInternalGroupQuery(wrt, req, tenant.ID, parts[3])
|
|
case len(parts) == 5 && parts[2] == "groups" && parts[4] == "members":
|
|
serveInternalGroupMembersQuery(wrt, req, tenant.ID, parts[3])
|
|
case len(parts) == 5 && parts[2] == "groups" && internalGroupResourceSpecs[parts[4]] != nil:
|
|
serveInternalAdminResourceList(wrt, req, tenant.ID, "group_"+parts[4], parts[3])
|
|
case len(parts) == 5 && parts[2] == "messages":
|
|
serveInternalMessageQuery(wrt, req, tenant.ID, parts[3], parts[4])
|
|
case len(parts) == 3 && parts[2] == "files":
|
|
serveInternalAdminResourceList(wrt, req, tenant.ID, "files", "")
|
|
case len(parts) == 4 && parts[2] == "files":
|
|
serveInternalFileQuery(wrt, req, tenant.ID, parts[3])
|
|
case len(parts) == 5 && parts[2] == "files" && parts[4] == "links":
|
|
serveInternalAdminResourceList(wrt, req, tenant.ID, "file_links", parts[3])
|
|
case len(parts) == 3 && internalTenantResourceSpecs[parts[2]] != nil:
|
|
serveInternalAdminResourceList(wrt, req, tenant.ID, parts[2], "")
|
|
case len(parts) == 4 && internalTenantResourceSpecs[parts[2]] != nil:
|
|
serveInternalAdminResourceDetail(wrt, req, tenant.ID, parts[2], parts[3])
|
|
case len(parts) == 4 && parts[2] == "reports" && parts[3] == "abuse":
|
|
writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil)
|
|
case len(parts) == 6 && parts[2] == "reports" && parts[3] == "abuse" && parts[5] == "target":
|
|
serveInternalAbuseTargetQuery(wrt, req, tenant.ID, parts[4])
|
|
case len(parts) == 4 && parts[2] == "outbox" && internalOutboxResourceSpecs[parts[3]] != nil:
|
|
serveInternalAdminResourceList(wrt, req, tenant.ID, "outbox_"+parts[3], "")
|
|
case len(parts) == 5 && parts[2] == "outbox" && internalOutboxResourceSpecs[parts[3]] != nil:
|
|
serveInternalAdminResourceDetail(wrt, req, tenant.ID, "outbox_"+parts[3], parts[4])
|
|
case len(parts) == 4 && parts[2] == "org" && internalOrgResourceSpecs[parts[3]] != nil:
|
|
serveInternalAdminResourceList(wrt, req, tenant.ID, "org_"+parts[3], "")
|
|
case len(parts) == 4 && parts[2] == "workbench" && parts[3] == "apps":
|
|
serveInternalAdminResourceList(wrt, req, tenant.ID, "workbench/apps", "")
|
|
case len(parts) == 4 && parts[2] == "dashboard" && parts[3] == "summary":
|
|
serveInternalDashboardSummary(wrt, req, tenant.ID)
|
|
case len(parts) == 4 && parts[2] == "kafka" && parts[3] == "dead-letters":
|
|
serveInternalKafkaDeadLettersQuery(wrt, req, tenant.ID)
|
|
case len(parts) == 5 && parts[2] == "kafka" && parts[3] == "dead-letters":
|
|
serveInternalKafkaDeadLetterQuery(wrt, req, tenant.ID, parts[4])
|
|
default:
|
|
writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil)
|
|
}
|
|
}
|
|
|
|
func authorizeInternalAPI(wrt http.ResponseWriter, req *http.Request, capability string) bool {
|
|
if !internalAPI.config.Enabled {
|
|
writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil)
|
|
return false
|
|
}
|
|
serviceName := strings.TrimSpace(req.Header.Get(internalHeaderService))
|
|
timestampRaw := strings.TrimSpace(req.Header.Get(internalHeaderTimestamp))
|
|
nonce := strings.TrimSpace(req.Header.Get(internalHeaderNonce))
|
|
signature := strings.TrimSpace(req.Header.Get(internalHeaderSignature))
|
|
service, ok := internalAPI.config.Services[serviceName]
|
|
if !ok || service.Secret == "" || timestampRaw == "" || nonce == "" || signature == "" {
|
|
writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "internal signature required", nil)
|
|
return false
|
|
}
|
|
if !internalServiceAllows(service, capability) {
|
|
writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", nil)
|
|
return false
|
|
}
|
|
timestamp, err := strconv.ParseInt(timestampRaw, 10, 64)
|
|
if err != nil {
|
|
writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "invalid internal timestamp", nil)
|
|
return false
|
|
}
|
|
window := time.Duration(internalAPI.config.TimeWindowSeconds) * time.Second
|
|
if delta := time.Since(time.Unix(timestamp, 0)); delta > window || delta < -window {
|
|
writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "internal signature expired", nil)
|
|
return false
|
|
}
|
|
bodyHash, err := hashInternalRequestBody(req)
|
|
if err != nil {
|
|
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
|
|
return false
|
|
}
|
|
signingText := strings.Join([]string{
|
|
req.Method,
|
|
req.URL.RequestURI(),
|
|
bodyHash,
|
|
timestampRaw,
|
|
nonce,
|
|
}, "\n")
|
|
if !verifyInternalSignature(service.Secret, signingText, signature) {
|
|
writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "invalid internal signature", nil)
|
|
return false
|
|
}
|
|
if req.Method != http.MethodGet {
|
|
requestClaim := internalRequestClaimValue(signingText)
|
|
if !claimInternalNonce(serviceName, nonce, requestClaim, time.Now().Add(window)) {
|
|
writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "internal nonce replayed", nil)
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func splitInternalPath(tail string) []string {
|
|
if tail == "" {
|
|
return nil
|
|
}
|
|
rawParts := strings.Split(tail, "/")
|
|
parts := make([]string, 0, len(rawParts))
|
|
for _, part := range rawParts {
|
|
if part == "" {
|
|
continue
|
|
}
|
|
decoded, err := url.PathUnescape(part)
|
|
if err != nil {
|
|
decoded = part
|
|
}
|
|
parts = append(parts, decoded)
|
|
}
|
|
return parts
|
|
}
|
|
|
|
func resolveInternalTenant(wrt http.ResponseWriter, req *http.Request, tenantCode string) (*types.Tenant, bool) {
|
|
if tenantCode == "" || store.Tenants == nil {
|
|
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "tenant_code"})
|
|
return nil, false
|
|
}
|
|
tenant, err := store.Tenants.GetByCode(tenantCode)
|
|
if err != nil {
|
|
if errors.Is(err, types.ErrNotFound) {
|
|
writeIMError(wrt, req, http.StatusNotFound, 30010, imReasonTenantUnavailable, "tenant unavailable", nil)
|
|
return nil, false
|
|
}
|
|
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
|
|
return nil, false
|
|
}
|
|
if !tenant.IsActive() {
|
|
writeIMError(wrt, req, http.StatusForbidden, 30010, imReasonTenantUnavailable, "tenant unavailable", nil)
|
|
return nil, false
|
|
}
|
|
return tenant, true
|
|
}
|
|
|
|
func serveInternalUserQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, userID string) {
|
|
userDBID, ok := parseInternalAdminDBID(userID)
|
|
if !ok {
|
|
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "user_id"})
|
|
return
|
|
}
|
|
if spec, ok := internalAdminQuerySpec("users"); ok {
|
|
query := spec.Select + " " + spec.From + " WHERE u.id=?"
|
|
if row, err := store.InternalAdminQueries.QueryOne(tenantID, query, []any{userDBID}); err == nil {
|
|
writeIMSuccess(wrt, req, normalizeInternalAdminRowForAPI(row))
|
|
return
|
|
}
|
|
}
|
|
uid := store.EncodeUid(userDBID)
|
|
user, err := store.Users.Get(tenantID, uid)
|
|
if err != nil {
|
|
writeInternalQueryError(wrt, req, err)
|
|
return
|
|
}
|
|
writeIMSuccess(wrt, req, map[string]any{
|
|
"user_id": userDBID,
|
|
"state": user.State,
|
|
"state_at": user.StateAt,
|
|
"username": user.Username,
|
|
"phone": user.Phone,
|
|
"public": user.Public,
|
|
"trusted": user.Trusted,
|
|
"last_seen": user.LastSeen,
|
|
"user_agent": user.UserAgent,
|
|
"tags": user.Tags,
|
|
"created_at": user.CreatedAt,
|
|
"updated_at": user.UpdatedAt,
|
|
})
|
|
}
|
|
|
|
func serveInternalGroupQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, topic string) {
|
|
if topic == "" || store.GroupBusiness == nil {
|
|
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "topic"})
|
|
return
|
|
}
|
|
if spec, ok := internalAdminQuerySpec("groups"); ok {
|
|
query := spec.Select + " " + spec.From + " WHERE t.name=?"
|
|
if row, err := store.InternalAdminQueries.QueryOne(tenantID, query, []any{topic}); err == nil {
|
|
writeIMSuccess(wrt, req, normalizeInternalAdminRowForAPI(row))
|
|
return
|
|
}
|
|
}
|
|
settings, err := store.GroupBusiness.GetSettings(tenantID, topic)
|
|
if err != nil {
|
|
writeInternalQueryError(wrt, req, err)
|
|
return
|
|
}
|
|
writeIMSuccess(wrt, req, internalGroupSettingsPayload(settings))
|
|
}
|
|
|
|
func serveInternalGroupMembersQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, topic string) {
|
|
if topic == "" || store.GroupBusiness == nil {
|
|
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "topic"})
|
|
return
|
|
}
|
|
spec := &internalQuerySpec{
|
|
Select: `SELECT gm.id,gm.topic,gm.user_id,u.state,u.public,p.nickname,p.public_account,s.modewant AS mode_want,s.modegiven AS mode_given,s.recvseqid AS recv_seq_id,s.readseqid AS read_seq_id,s.delid AS del_id,
|
|
gm.role,gm.member_status,gm.nickname AS group_nickname,gm.pinned,gm.mute_notifications,gm.join_source,gm.joined_by,gm.joined_at,gm.left_at,gm.removed_by,gm.version,gm.created_at,gm.updated_at`,
|
|
From: "FROM im_group_members AS gm LEFT JOIN im_users AS u ON u.id=gm.user_id LEFT JOIN im_user_profiles AS p ON p.user_id=gm.user_id LEFT JOIN im_subscriptions AS s ON s.topic=gm.topic AND s.userid=gm.user_id",
|
|
DefaultSort: "ORDER BY gm.id DESC",
|
|
ScopeColumn: "gm.topic",
|
|
Filters: []internalQueryFilter{
|
|
{"user_id", "gm.user_id", "=", "uid"},
|
|
{"role", "gm.role", "=", "string"},
|
|
{"member_status", "gm.member_status", "=", "string"},
|
|
{"q", "gm.nickname", "LIKE", "string"},
|
|
{"eligible_role", "gm.role", "=", "string"},
|
|
},
|
|
}
|
|
query, countQuery, args, page, perPage, err := buildInternalAdminListQuery(req, spec, topic)
|
|
if err == nil {
|
|
if result, queryErr := store.InternalAdminQueries.Query(tenantID, query, countQuery, args, page, perPage); queryErr == nil {
|
|
writeIMSuccess(wrt, req, map[string]any{
|
|
"items": normalizeInternalAdminRowsForAPI(result.Items),
|
|
"total": result.Total,
|
|
"page": result.Page,
|
|
"per_page": result.PerPage,
|
|
"next_cursor": "",
|
|
"has_more": false,
|
|
})
|
|
return
|
|
}
|
|
}
|
|
limit := parseIMLimit(req.URL.Query().Get("limit"), 50, 200)
|
|
members, nextCursor, err := store.GroupBusiness.ListMembers(tenantID, topic, types.GroupMemberListOptions{
|
|
Role: strings.TrimSpace(req.URL.Query().Get("role")),
|
|
EligibleRole: strings.TrimSpace(req.URL.Query().Get("eligible_role")),
|
|
Query: strings.TrimSpace(req.URL.Query().Get("q")),
|
|
Cursor: strings.TrimSpace(req.URL.Query().Get("cursor")),
|
|
Limit: limit,
|
|
})
|
|
if err != nil {
|
|
writeInternalQueryError(wrt, req, err)
|
|
return
|
|
}
|
|
items := make([]any, 0, len(members))
|
|
for _, member := range members {
|
|
items = append(items, map[string]any{
|
|
"user_id": internalAdminUIDRaw(member.UserID),
|
|
"role": member.Role,
|
|
"member_status": member.MemberStatus,
|
|
"nickname": member.Nickname,
|
|
"pinned": member.Pinned,
|
|
"mute_notifications": member.MuteNotifications,
|
|
"join_source": member.JoinSource,
|
|
"joined_by": internalAdminUIDRaw(member.JoinedBy),
|
|
"joined_at": member.JoinedAt,
|
|
"left_at": member.LeftAt,
|
|
"removed_by": internalAdminUIDRaw(member.RemovedBy),
|
|
"version": member.Version,
|
|
"created_at": member.CreatedAt,
|
|
"updated_at": member.UpdatedAt,
|
|
})
|
|
}
|
|
writeIMSuccess(wrt, req, map[string]any{"items": items, "next_cursor": nextCursor})
|
|
}
|
|
|
|
func serveInternalMessageQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, topic, seqRaw string) {
|
|
seq, err := strconv.ParseInt(seqRaw, 10, 64)
|
|
if topic == "" || err != nil || seq <= 0 {
|
|
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
|
|
return
|
|
}
|
|
msg, err := store.Messages.GetOne(tenantID, topic, seq)
|
|
if err != nil {
|
|
writeInternalQueryError(wrt, req, err)
|
|
return
|
|
}
|
|
writeIMSuccess(wrt, req, map[string]any{
|
|
"topic": msg.Topic,
|
|
"seq_id": msg.SeqId,
|
|
"from": internalAdminUIDStringRaw(msg.From),
|
|
"client_msg_id": msg.ClientMsgID,
|
|
"version": msg.Version,
|
|
"head": msg.Head,
|
|
"content": msg.Content,
|
|
"deleted_at": msg.DeletedAt,
|
|
"recalled_at": msg.RecalledAt,
|
|
"recalled_by": internalAdminUIDStringRaw(msg.RecalledBy),
|
|
"created_at": msg.CreatedAt,
|
|
"updated_at": msg.UpdatedAt,
|
|
})
|
|
}
|
|
|
|
func serveInternalFileQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, fileID string) {
|
|
if fileID == "" || store.Files == nil {
|
|
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "file_id"})
|
|
return
|
|
}
|
|
file, err := store.Files.Get(tenantID, fileID)
|
|
if err != nil {
|
|
writeInternalQueryError(wrt, req, err)
|
|
return
|
|
}
|
|
writeIMSuccess(wrt, req, map[string]any{
|
|
"file_id": file.Id,
|
|
"status": file.Status,
|
|
"user_id": internalAdminUIDStringRaw(file.User),
|
|
"mime_type": file.MimeType,
|
|
"size": file.Size,
|
|
"location": file.Location,
|
|
"etag": file.ETag,
|
|
"created_at": file.CreatedAt,
|
|
"updated_at": file.UpdatedAt,
|
|
})
|
|
}
|
|
|
|
func serveInternalKafkaDeadLettersQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID) {
|
|
if store.HTTPInfrastructure == nil {
|
|
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
|
|
return
|
|
}
|
|
limit := parseIMLimit(req.URL.Query().Get("limit"), 50, 200)
|
|
page, err := store.HTTPInfrastructure.ListKafkaDeadLetters(tenantID, types.KafkaDeadLetterListOptions{
|
|
Status: strings.TrimSpace(req.URL.Query().Get("status")),
|
|
BizType: strings.TrimSpace(req.URL.Query().Get("biz_type")),
|
|
BizID: strings.TrimSpace(req.URL.Query().Get("biz_id")),
|
|
EventType: strings.TrimSpace(req.URL.Query().Get("event_type")),
|
|
Cursor: strings.TrimSpace(req.URL.Query().Get("cursor")),
|
|
Limit: limit,
|
|
})
|
|
if err != nil {
|
|
writeInternalQueryError(wrt, req, err)
|
|
return
|
|
}
|
|
items := make([]any, 0, len(page.Items))
|
|
for _, dead := range page.Items {
|
|
items = append(items, internalKafkaDeadLetterPayload(&dead, false))
|
|
}
|
|
writeIMSuccess(wrt, req, map[string]any{
|
|
"items": items,
|
|
"next_cursor": page.NextCursor,
|
|
"has_more": page.NextCursor != "",
|
|
})
|
|
}
|
|
|
|
func serveInternalKafkaDeadLetterQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, deadID string) {
|
|
if store.HTTPInfrastructure == nil || strings.TrimSpace(deadID) == "" {
|
|
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "dead_id"})
|
|
return
|
|
}
|
|
dead, err := store.HTTPInfrastructure.GetKafkaDeadLetter(tenantID, deadID)
|
|
if err != nil {
|
|
writeInternalQueryError(wrt, req, err)
|
|
return
|
|
}
|
|
writeIMSuccess(wrt, req, internalKafkaDeadLetterPayload(dead, true))
|
|
}
|
|
|
|
func internalGroupSettingsPayload(settings *types.GroupSettings) map[string]any {
|
|
if settings == nil {
|
|
return nil
|
|
}
|
|
return map[string]any{
|
|
"topic": settings.Topic,
|
|
"owner_user_id": internalAdminUIDRaw(settings.OwnerUserID),
|
|
"name": settings.Name,
|
|
"avatar_file_id": settings.AvatarFileID,
|
|
"description": settings.Description,
|
|
"group_status": settings.GroupStatus,
|
|
"join_approval": settings.JoinApproval,
|
|
"mute_all": settings.MuteAll,
|
|
"member_upload_files": settings.MemberUploadFiles,
|
|
"member_private_chat": settings.MemberPrivateChat,
|
|
"member_invite": settings.MemberInvite,
|
|
"max_member_count": settings.MaxMemberCount,
|
|
"max_admin_count": settings.MaxAdminCount,
|
|
"member_count": settings.MemberCount,
|
|
"version": settings.Version,
|
|
"created_by": internalAdminUIDRaw(settings.CreatedBy),
|
|
"dissolved_by": internalAdminUIDRaw(settings.DissolvedBy),
|
|
"dissolved_at": settings.DissolvedAt,
|
|
"created_at": settings.CreatedAt,
|
|
"updated_at": settings.UpdatedAt,
|
|
}
|
|
}
|
|
|
|
func internalKafkaDeadLetterPayload(dead *types.KafkaDeadLetter, includeEnvelope bool) map[string]any {
|
|
if dead == nil {
|
|
return nil
|
|
}
|
|
payload := map[string]any{
|
|
"id": dead.ID,
|
|
"dead_id": dead.DeadID,
|
|
"event_id": dead.EventID,
|
|
"parent_event_id": dead.ParentEventID,
|
|
"source_topic": dead.SourceTopic,
|
|
"consumer_group": dead.ConsumerGroup,
|
|
"event_type": dead.EventType,
|
|
"biz_type": dead.BizType,
|
|
"biz_id": dead.BizID,
|
|
"retry_count": dead.RetryCount,
|
|
"last_reason": dead.LastReason,
|
|
"last_error": dead.LastError,
|
|
"status": dead.Status,
|
|
"replay_count": dead.ReplayCount,
|
|
"created_at": dead.CreatedAt,
|
|
"updated_at": dead.UpdatedAt,
|
|
}
|
|
if includeEnvelope {
|
|
var envelope any
|
|
if len(dead.Envelope) > 0 && json.Unmarshal(dead.Envelope, &envelope) == nil {
|
|
payload["envelope"] = envelope
|
|
} else {
|
|
payload["envelope_raw"] = string(dead.Envelope)
|
|
}
|
|
}
|
|
return payload
|
|
}
|
|
|
|
func writeInternalQueryError(wrt http.ResponseWriter, req *http.Request, err error) {
|
|
switch {
|
|
case errors.Is(err, types.ErrUserNotFound), errors.Is(err, types.ErrTopicNotFound), errors.Is(err, types.ErrNotFound):
|
|
writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil)
|
|
case errors.Is(err, types.ErrMalformed):
|
|
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
|
|
case errors.Is(err, types.ErrPermissionDenied):
|
|
writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", nil)
|
|
default:
|
|
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
|
|
}
|
|
}
|
|
|
|
func parseInternalUid(raw string) types.Uid {
|
|
if uid := types.ParseUserId(raw); !uid.IsZero() {
|
|
return uid
|
|
}
|
|
id, ok := parseInternalAdminDBID(raw)
|
|
if !ok {
|
|
return types.ZeroUid
|
|
}
|
|
return store.EncodeUid(id)
|
|
}
|
|
|
|
func internalServiceAllows(service internalAPIServiceConfig, capability string) bool {
|
|
for _, allowed := range service.Allow {
|
|
allowed = strings.TrimSpace(allowed)
|
|
if allowed == "*" || allowed == capability {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func hashInternalRequestBody(req *http.Request) (string, error) {
|
|
if req.Body == nil {
|
|
sum := sha256.Sum256(nil)
|
|
return hex.EncodeToString(sum[:]), nil
|
|
}
|
|
body, err := io.ReadAll(io.LimitReader(req.Body, 1<<20))
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if err := req.Body.Close(); err != nil {
|
|
return "", err
|
|
}
|
|
req.Body = io.NopCloser(bytes.NewReader(body))
|
|
sum := sha256.Sum256(body)
|
|
return hex.EncodeToString(sum[:]), nil
|
|
}
|
|
|
|
func verifyInternalSignature(secret, signingText, signature string) bool {
|
|
signature = strings.TrimPrefix(signature, "sha256=")
|
|
mac := hmac.New(sha256.New, []byte(secret))
|
|
_, _ = mac.Write([]byte(signingText))
|
|
expected := mac.Sum(nil)
|
|
if decoded, err := hex.DecodeString(signature); err == nil {
|
|
return hmac.Equal(decoded, expected)
|
|
}
|
|
if decoded, err := base64.StdEncoding.DecodeString(signature); err == nil {
|
|
return hmac.Equal(decoded, expected)
|
|
}
|
|
return false
|
|
}
|
|
|
|
func claimInternalNonce(serviceName, nonce, requestClaim string, expiresAt time.Time) bool {
|
|
ttl := time.Until(expiresAt)
|
|
nonceKey := fmt.Sprintf("%s:%s", serviceName, nonce)
|
|
if claimed, sameValue, enabled, err := store.ClaimShortStateValue("internal_api_nonce", nonceKey, requestClaim, ttl); enabled {
|
|
if err != nil {
|
|
logs.Warn.Println("internal api: failed to claim nonce in redis", err)
|
|
return false
|
|
}
|
|
return claimed || sameValue
|
|
}
|
|
now := time.Now()
|
|
internalAPI.nonces.Range(func(key, value any) bool {
|
|
if claim, ok := value.(internalNonceClaim); ok && claim.expiresAt.Before(now) {
|
|
internalAPI.nonces.Delete(key)
|
|
}
|
|
return true
|
|
})
|
|
if existing, loaded := internalAPI.nonces.LoadOrStore(nonceKey, internalNonceClaim{value: requestClaim, expiresAt: expiresAt}); loaded {
|
|
claim, ok := existing.(internalNonceClaim)
|
|
return ok && claim.value == requestClaim
|
|
}
|
|
return true
|
|
}
|
|
|
|
type internalNonceClaim struct {
|
|
value string
|
|
expiresAt time.Time
|
|
}
|
|
|
|
func internalRequestClaimValue(signingText string) string {
|
|
sum := sha256.Sum256([]byte(signingText))
|
|
return hex.EncodeToString(sum[:])
|
|
}
|