seqapi masking

docs/en/02-configuration.md

@@ -513,6 +513,95 @@ Config for `/seqapi` API handlers.
 
     Type of field.
 
++ **`masking`** *`Masking`* *`optional`*
+
+  Masking configuration.
+
+  ⚠ **Experimental feature**
+
+  `Masking` fields:
+
+  + **`masks`** *`[]Mask`* *`required`*
+
+    List of masks.
+
+  + **`process_fields`** *`[]string`* *`default=[]`*
+
+    List of processed event fields.
+
+    > It is wrong to set non-empty ignored fields list and non-empty processed fields list at the same time.
+
+  + **`ignore_fields`** *`[]string`* *`default=[]`*
+
+    List of ignored event fields.
+
+    > It is wrong to set non-empty ignored fields list and non-empty processed fields list at the same time.
+
+  `Mask` fields:
+
+  + **`re`** *`string`* *`required`*
+
+    Regular expression for masking.
+
+  + **`groups`** *`[]int`* *`default=[]`*
+
+    Groups are numbers of masking groups in expression. If set to empty list or the list **contains** `0`, the full expression will be masked.
+
+  + **`mode`** *`string`* *`required`* *`options="mask"|"replace"|"cut"`*
+
+    Masking mode:
+      - `mask` - asterisks (`*`) are used for masking
+      - `replace` - `replace_word` is used for masking
+      - `cut` - masking parts will be cut instead of being replaced
+
+  + **`replace_word`** *`string`* *`default=""`*
+
+    Replacement word used in `mode: replace`.
+
+    > Ignored in other mods.
+
+  + **`process_fields`** *`[]string`* *`default=[]`*
+
+    List of mask-specific processed event fields.
+
+    > It is wrong to set non-empty ignored fields list and non-empty processed fields list at the same time.
+
+  + **`ignore_fields`** *`[]string`* *`default=[]`*
+
+    List of mask-specific ignored event fields.
+
+    > It is wrong to set non-empty ignored fields list and non-empty processed fields list at the same time.
+
+  + **`field_filters`** *`FieldFilterSet`* *`optional`*
+
+    Set of field filters to filter events before masking.
+
+  `FieldFilterSet` fields:
+
+  + **`condition`** *`string`* *`required`* *`options="and"|"or"|"not"`*
+
+    Condition for combining filters.
+
+  + **`filters`** *`[]FieldFilter`* *`required`*
+
+    List of filters.
+
+    > Maximum 1 when `condition: not`.
+
+  `FieldFilter` fields:
+
+  + **`field`** *`string`* *`required`*
+
+    Event field.
+
+  + **`mode`** *`string`* *`required`* *`options="equal"|"contains"|"prefix"|"suffix"`*
+
+    Filter mode.
+
+  + **`values`** *`[]string`* *`required`*
+
+    List of event field values to filter.
+
 ### Error groups
 
 **`error_groups`** *`ErrorGroups`* *`optional`*

docs/ru/02-configuration.md

@@ -513,6 +513,95 @@ handlers:
 
     Тип поля.
 
++ **`masking`** *`Masking`* *`optional`*
+
+  Конфигурация маскирования данных.
+
+  ⚠ **Экспериментальная функция**
+
+  Поля `Masking`:
+
+  + **`masks`** *`[]Mask`* *`required`*
+
+    Список масок.
+
+  + **`process_fields`** *`[]string`* *`default=[]`*
+
+    Список полей события, которые необходимо маскировать.
+
+    > Параметры `process_fields` и `ignore_fields` не должны задаваться одновременно.
+
+  + **`ignore_fields`** *`[]string`* *`default=[]`*
+
+    Список полей события, которые нет необходимости маскировать.
+
+    > Параметры `process_fields` и `ignore_fields` не должны задаваться одновременно.
+
+  Поля `Mask`:
+
+  + **`re`** *`string`* *`required`*
+
+    Регулярное выражение для маскирования.
+
+  + **`groups`** *`[]int`* *`default=[]`*
+
+    Номера маскируемых групп в выражении. Если задан пустой массив или массив **содержит** `0`, то замаскировано будет все выражение.
+
+  + **`mode`** *`string`* *`required`* *`options="mask"|"replace"|"cut"`*
+
+    Режим маскирования:
+      - `mask` - используются звездочки (`*`)
+      - `replace` - используется `replace_word`
+      - `cut` - маскируемые группы будут вырезаны, а не заменены
+
+  + **`replace_word`** *`string`* *`default=""`*
+
+    Заменяющее слово, используемое в режиме `mode: replace`.
+
+    > Игнорируется в других режимах .
+
+  + **`process_fields`** *`[]string`* *`default=[]`*
+
+    Список специфичных для маски полей события, которые необходимо маскировать.
+
+    > Параметры `process_fields` и `ignore_fields` не должны задаваться одновременно.
+
+  + **`ignore_fields`** *`[]string`* *`default=[]`*
+
+    Список специфичных для маски полей события, которые нет необходимости маскировать.
+
+    > Параметры `process_fields` и `ignore_fields` не должны задаваться одновременно.
+
+  + **`field_filters`** *`FieldFilterSet`* *`optional`*
+
+    Набор фильтров полей для фильтрации событий перед маскировкой.
+
+  Поля `FieldFilterSet`:
+
+  + **`condition`** *`string`* *`required`* *`options="and"|"or"|"not"`*
+
+    Условие для объединения фильтров.
+
+  + **`filters`** *`[]FieldFilter`* *`required`*
+
+    Список фильтров.
+
+    > Максимум 1 при `condition: not`.
+
+  Поля `FieldFilter`:
+
+  + **`field`** *`string`* *`required`*
+
+    Поле события.
+
+  + **`mode`** *`string`* *`required`* *`options="equal"|"contains"|"prefix"|"suffix"`*
+
+    Режим фильтрации.
+
+  + **`values`** *`[]string`* *`required`*
+
+    Список значений поля события для фильтрации.
+
 ### Error groups
 
 **`error_groups`** *`ErrorGroups`* *`optional`*

internal/api/seqapi/v1/grpc/aggregation.go

@@ -61,5 +61,34 @@ func (a *API) GetAggregation(ctx context.Context, req *seqapi.GetAggregationRequ
 	if err != nil {
 		return nil, err
 	}
+
+	if a.masker != nil {
+		buf := make([]string, 0)
+		for i, agg := range resp.Aggregations {
+			if agg == nil {
+				continue
+			}
+
+			buf = buf[:0]
+			for _, b := range agg.Buckets {
+				buf = append(buf, b.GetKey())
+			}
+
+			aggReq := req.Aggregations[i]
+			field := aggReq.Field
+			if aggReq.GroupBy != "" {
+				field = aggReq.GroupBy
+			}
+
+			buf = a.masker.MaskAgg(field, buf)
+
+			for j, key := range buf {
+				if agg.Buckets[j] != nil {
+					agg.Buckets[j].Key = key
+				}
+			}
+		}
+	}
+
 	return resp, nil
 }

internal/api/seqapi/v1/grpc/api.go

@@ -3,12 +3,16 @@ package grpc
 import (
 	"time"
 
+	"go.uber.org/zap"
+
 	"github.com/ozontech/seq-ui/internal/api/profiles"
 	"github.com/ozontech/seq-ui/internal/app/config"
 	"github.com/ozontech/seq-ui/internal/app/types"
 	"github.com/ozontech/seq-ui/internal/pkg/cache"
 	"github.com/ozontech/seq-ui/internal/pkg/client/seqdb"
+	"github.com/ozontech/seq-ui/internal/pkg/mask"
 	asyncsearches "github.com/ozontech/seq-ui/internal/pkg/service/async_searches"
+	"github.com/ozontech/seq-ui/logger"
 	"github.com/ozontech/seq-ui/pkg/seqapi/v1"
 )
 
@@ -24,6 +28,7 @@ type API struct {
 	pinnedFields        []*seqapi.Field
 	asyncSearches       *asyncsearches.Service
 	profiles            *profiles.Profiles
+	masker              *mask.Masker
 }
 
 func New(
@@ -39,6 +44,11 @@ func New(
 		fCache = newFieldsCache(cfg.FieldsCacheTTL)
 	}
 
+	masker, err := mask.New(cfg.Masking)
+	if err != nil {
+		logger.Fatal("failed to init masking", zap.Error(err))
+	}
+
 	return &API{
 		config:              cfg,
 		seqDB:               seqDB,
@@ -49,6 +59,7 @@ func New(
 		pinnedFields:        parsePinnedFields(cfg.PinnedFields),
 		asyncSearches:       asyncSearches,
 		profiles:            p,
+		masker:              masker,
 	}
 }
 

internal/api/seqapi/v1/grpc/events.go

@@ -37,5 +37,9 @@ func (a *API) GetEvent(ctx context.Context, req *seqapi.GetEventRequest) (*seqap
 		logger.Error("failed to marshal event proto for caching", zap.String("id", req.Id), zap.Error(err))
 	}
 
+	if a.masker != nil && resp.Event != nil {
+		a.masker.Mask(resp.Event.Data)
+	}
+
 	return resp, nil
 }

internal/api/seqapi/v1/grpc/search.go

@@ -95,5 +95,11 @@ func (a *API) Search(ctx context.Context, req *seqapi.SearchRequest) (*seqapi.Se
 		}
 	}
 
+	if a.masker != nil {
+		for _, e := range resp.Events {
+			a.masker.Mask(e.Data)
+		}
+	}
+
 	return resp, nil
 }

internal/api/seqapi/v1/http/aggregation.go

@@ -73,7 +73,31 @@ func (a *API) serveGetAggregation(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	wr.WriteJson(getAggregationResponseFromProto(resp))
+	getAggResp := getAggregationResponseFromProto(resp)
+
+	if a.masker != nil {
+		buf := make([]string, 0)
+		for i, agg := range getAggResp.Aggregations {
+			buf = buf[:0]
+			for _, b := range agg.Buckets {
+				buf = append(buf, b.Key)
+			}
+
+			aggReq := httpReq.Aggregations[i]
+			field := aggReq.Field
+			if aggReq.GroupBy != "" {
+				field = aggReq.GroupBy
+			}
+
+			buf = a.masker.MaskAgg(field, buf)
+
+			for j, key := range buf {
+				getAggResp.Aggregations[i].Buckets[j].Key = key
+			}
+		}
+	}
+
+	wr.WriteJson(getAggResp)
 }
 
 type aggregationFunc string // @name seqapi.v1.AggregationFunc

internal/api/seqapi/v1/http/api.go

@@ -5,14 +5,17 @@ import (
 
 	"github.com/go-chi/chi/v5"
 	"github.com/gofrs/uuid"
+	"go.uber.org/zap"
 
 	"github.com/ozontech/seq-ui/internal/api/profiles"
 	"github.com/ozontech/seq-ui/internal/app/config"
 	"github.com/ozontech/seq-ui/internal/app/tokenlimiter"
 	"github.com/ozontech/seq-ui/internal/app/types"
 	"github.com/ozontech/seq-ui/internal/pkg/cache"
 	"github.com/ozontech/seq-ui/internal/pkg/client/seqdb"
+	"github.com/ozontech/seq-ui/internal/pkg/mask"
 	asyncsearches "github.com/ozontech/seq-ui/internal/pkg/service/async_searches"
+	"github.com/ozontech/seq-ui/logger"
 	"github.com/ozontech/seq-ui/pkg/seqapi/v1"
 )
 
@@ -27,6 +30,7 @@ type API struct {
 	exportLimiter       *tokenlimiter.Limiter
 	asyncSearches       *asyncsearches.Service
 	profiles            *profiles.Profiles
+	masker              *mask.Masker
 }
 
 func New(
@@ -42,6 +46,15 @@ func New(
 		fCache = newFieldsCache(cfg.FieldsCacheTTL)
 	}
 
+	masker, err := mask.New(cfg.Masking)
+	if err != nil {
+		logger.Fatal("failed to init masking", zap.Error(err))
+	}
+	// for export
+	if masker != nil {
+		seqDB.WithMasking(masker)
+	}
+
 	return &API{
 		config:              cfg,
 		seqDB:               seqDB,
@@ -53,6 +66,7 @@ func New(
 		exportLimiter:       tokenlimiter.New(cfg.MaxParallelExportRequests),
 		asyncSearches:       asyncSearches,
 		profiles:            p,
+		masker:              masker,
 	}
 }
 

internal/api/seqapi/v1/http/events.go

@@ -58,7 +58,12 @@ func (a *API) serveGetEvent(w http.ResponseWriter, r *http.Request) {
 		logger.Error("failed to marshal event proto for caching", zap.String("id", id), zap.Error(err))
 	}
 
-	wr.WriteJson(getEventResponseFromProto(resp))
+	eventResp := getEventResponseFromProto(resp)
+	if a.masker != nil {
+		a.masker.Mask(eventResp.Event.Data)
+	}
+
+	wr.WriteJson(eventResp)
 }
 
 type event struct {