cubefs

1
// Copyright 2022 The CubeFS Authors.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
12
// implied. See the License for the specific language governing
13
// permissions and limitations under the License.
14

15
package auditlog
16

17
import (
18
	"encoding/json"
19
	"errors"
20
	"fmt"
21
	"net/url"
22
	"regexp"
23
	"strconv"
24
	"strings"
25

26
	"github.com/cubefs/cubefs/blobstore/util/log"
27
)
28

29
const (
30
	TAG_SERVICE         = 1
31
	TAG_TIME            = 2
32
	TAG_METHOD          = 3
33
	TAG_PATH            = 4
34
	TAG_HEADER          = 5
35
	TAG_PARAMS          = 6
36
	TAG_CODE            = 7
37
	TAG_RESPONSE_HEADER = 8
38
	TAG_RESPONSE        = 9
39
	TAG_RESPONSE_SIZE   = 10
40
	TAG_RESPONSE_TIME   = 11
41
)
42

43
const (
44
	LogTypeAudit = "audit"
45
)
46

47
type LogEntry interface {
48
	Service() string
49
	LogType() string
50
	Code() string
51
	Method() string
52
	Path() string
53
	RespTime() int64 // 100ns
54
	RespLength() int64
55
	ReqLength() int64
56
	ReqHost() string
57
	RemoteAddr() string
58
	Xlogs() []string
59
	UA() string
60
	XRespCode() string
61
	// ReqParams returns params of request, including params in raw query and body
62
	ReqParams() string
63
	Uid() uint32
64
	ApiName() string
65
}
66

67
func ErrInValidFieldCnt(msg string) error {
68
	return errors.New("invalid field count, " + msg)
69
}
70

71
var vre = regexp.MustCompile("^v[0-9]+$")
72

73
type ReqHeader struct {
74
	ContentLength string `json:"Content-Length"`
75
	BodySize      int64  `json:"BodySize"` // body size
76
	RawQuery      string `json:"RawQuery"`
77
	Host          string `json:"Host"`
78
	Token         *Token `json:"Token"`
79
	XRealIp       string `json:"X-Real-Ip"`
80
	XFromCdn      string `json:"X-From-Cdn"`
81
	XSrc          string `json:"X-Src"`
82
	IP            string `json:"IP"`
83
	UA            string `json:"User-Agent"`
84
}
85

86
type Token struct {
87
	AppId uint64 `json:"appid"`
88
	Uid   uint32 `json:"uid"`
89
	Utype uint32 `json:"utype"`
90
}
91

92
type SToken struct {
93
	Uid    uint32 `json:"uid"`
94
	Bucket string `json:"tbl"`
95
}
96

97
type RsInfo struct {
98
	Bucket   string `json:"bucket"`
99
	EntryURI string `json:"entryURI"`
100
}
101

102
type RespHeader struct {
103
	ContentLength     string           `json:"Content-Length"`
104
	Xlog              []string         `json:"X-Log"`
105
	XWarn             []string         `json:"X-Warn"`
106
	Bucket            string           `json:"Tbl"`
107
	Token             *Token           `json:"Token"`
108
	SToken            *SToken          `json:"SToken"`
109
	FileType          int              `json:"FileType"`
110
	BatchOps          map[string]int64 `json:"batchOps"`
111
	PreDelSize        map[string]int64 `json:"preDelSize"`
112
	PreDelArchiveSize map[string]int64 `json:"preDelArchiveSize"`
113
	OUid              uint32           `json:"ouid"` // owner uid
114
	RsInfo            *RsInfo          `json:"rs-info"`
115
	XRespCode         string           `json:"X-Resp-Code"` // return from dora
116
	BillTag           string           `json:"billtag"`     // must be same with definition  in billtag.go
117
	BatchDeletes      map[uint32]int64 `json:"batchDelete"`
118
	ApiName           string           `json:"api"` // api name of this auditlog
119
}
120

121
type RequestRow struct {
122
	data []string
123

124
	// the following fields will be loaded lazily and use get request to obtain them
125
	reqHeader  *ReqHeader
126
	respHeader *RespHeader
127
	rawQuery   *url.Values
128
}
129

130
type AdRowParser struct{}
131

132
func (self *AdRowParser) Parse(line string) (*RequestRow, error) {
133
	return ParseReqlogToAdrow(line)
134
}
135

136
func ParseReqlogToAdrow(line string) (*RequestRow, error) {
137
	idx := strings.Index(line, "REQ")
138
	if idx < 0 {
139
		return nil, errors.New("invalid reqlog")
140
	}
141
	line = strings.TrimSuffix(line[idx:], "\n")
142
	a := new(RequestRow)
143
	a.data = strings.Split(line, "\t")
144
	if len(a.data) != 12 && len(a.data) != 14 {
145
		return nil, ErrInValidFieldCnt(fmt.Sprintf("expect 12 or 14 while get %v", len(a.data)))
146
	}
147
	if a.data[0] != "REQ" {
148
		return nil, fmt.Errorf("invalid Head: %v", a.data[0])
149
	}
150
	return a, nil
151
}
152

153
func ParseReqlog(line string) (LogEntry, error) {
154
	return ParseReqlogToAdrow(line)
155
}
156

157
func (a *RequestRow) LogType() string {
158
	return LogTypeAudit
159
}
160

161
func (a *RequestRow) ApiName() string { // s3, apiName can get directly from auditlog
162
	respHeader := a.getRespHeader()
163
	if respHeader == nil {
164
		return ""
165
	}
166
	return respHeader.ApiName
167
}
168

169
func (a *RequestRow) Uid() uint32 {
170
	respSToken := a.RespSToken()
171
	if respSToken != nil {
172
		return respSToken.Uid
173
	}
174
	respToken := a.RespToken()
175
	if respToken != nil {
176
		return respToken.Uid
177
	}
178
	reqToken := a.ReqToken()
179
	if reqToken != nil {
180
		return reqToken.Uid
181
	}
182
	return 0
183
}
184

185
func (a *RequestRow) RespToken() *Token {
186
	respHeader := a.getRespHeader()
187
	if respHeader == nil {
188
		return nil
189
	}
190
	return respHeader.Token
191
}
192

193
func (a *RequestRow) RespSToken() *SToken {
194
	respHeader := a.getRespHeader()
195
	if respHeader == nil {
196
		return nil
197
	}
198
	return respHeader.SToken
199
}
200

201
func (a *RequestRow) UA() string {
202
	reqHeader := a.getReqHeader()
203
	if reqHeader != nil && reqHeader.UA != "" {
204
		return reqHeader.UA
205
	}
206
	return ""
207
}
208

209
func (a *RequestRow) Bucket() string {
210
	respHeader := a.getRespHeader()
211
	if respHeader != nil && respHeader.Bucket != "" {
212
		return respHeader.Bucket
213
	}
214
	return ""
215
}
216

217
func (a *RequestRow) getReqHeader() *ReqHeader {
218
	if a.reqHeader != nil {
219
		return a.reqHeader
220
	}
221
	err := json.Unmarshal([]byte(a.data[TAG_HEADER]), &a.reqHeader)
222
	if err != nil {
223
		log.Warnf("%q, Unmarshal ReqHeader err %v", a.data[TAG_HEADER], err)
224
		return nil
225
	}
226
	return a.reqHeader
227
}
228

229
func (a *RequestRow) getRawquery() *url.Values {
230
	reqHeader := a.getReqHeader()
231
	if reqHeader == nil {
232
		return nil
233
	}
234
	m, err := url.ParseQuery(reqHeader.RawQuery)
235
	if err != nil {
236
		return nil
237
	}
238
	a.rawQuery = &m
239
	return a.rawQuery
240
}
241

242
func (a *RequestRow) getRespHeader() *RespHeader {
243
	if a.respHeader != nil {
244
		return a.respHeader
245
	}
246
	err := json.Unmarshal([]byte(a.data[TAG_RESPONSE_HEADER]), &a.respHeader)
247
	if err != nil {
248
		log.Warnf("%q, Unmarshal RespHeader err %v", a.data[TAG_RESPONSE_HEADER], err)
249
		return nil
250
	}
251
	return a.respHeader
252
}
253

254
func (a *RequestRow) String() string {
255
	return strings.Join(a.data, "\t")
256
}
257

258
func (a *RequestRow) Code() string {
259
	if isType(Digital, a.data[TAG_CODE]) {
260
		return a.data[TAG_CODE]
261
	}
262
	return "unknown"
263
}
264

265
func (a *RequestRow) ReqToken() *Token {
266
	reqHeader := a.getReqHeader()
267
	if reqHeader == nil {
268
		return nil
269
	}
270
	return reqHeader.Token
271
}
272

273
// get unix second of time
274
func (a *RequestRow) ReqTime() int64 {
275
	t, err := strconv.ParseInt(a.data[TAG_TIME], 10, 0)
276
	if err != nil {
277
		log.Errorf("cannot parse time %v to int", a.data[TAG_TIME])
278
	}
279
	return t / 10000000
280
}
281

282
func (a *RequestRow) RemoteIp() string {
283
	reqHeader := a.getReqHeader()
284
	if reqHeader != nil {
285
		if reqHeader.XRealIp != "" {
286
			return strings.TrimSpace(reqHeader.XRealIp)
287
		} else {
288
			return strings.TrimSpace(reqHeader.IP)
289
		}
290
	}
291
	return ""
292
}
293

294
func (a *RequestRow) ReqCdn() string {
295
	reqHeader := a.getReqHeader()
296
	if reqHeader == nil {
297
		return ""
298
	}
299
	return reqHeader.XFromCdn
300
}
301

302
func (a *RequestRow) ReqSrc() string {
303
	reqHeader := a.getReqHeader()
304
	if reqHeader == nil {
305
		return ""
306
	}
307
	return reqHeader.XSrc
308
}
309

310
func (a *RequestRow) Service() string {
311
	return a.data[TAG_SERVICE]
312
}
313

314
func (a *RequestRow) Method() string {
315
	return a.data[TAG_METHOD]
316
}
317

318
func (a *RequestRow) Path() string {
319
	return a.data[TAG_PATH]
320
}
321

322
func (a *RequestRow) RawQuery() string {
323
	reqHeader := a.getReqHeader()
324
	if reqHeader == nil {
325
		return ""
326
	}
327
	return a.getReqHeader().RawQuery
328
}
329

330
func (a *RequestRow) RespTime() (respTime int64) {
331
	var err error
332
	respTime, err = strconv.ParseInt(a.data[TAG_RESPONSE_TIME], 10, 64)
333
	if err != nil {
334
		log.Error(err)
335
	}
336
	return respTime
337
}
338

339
func (a *RequestRow) RespLength() (respLength int64) {
340
	respLength, _ = strconv.ParseInt(a.data[TAG_RESPONSE_SIZE], 10, 64)
341
	return
342
}
343

344
func (a *RequestRow) XRespCode() string {
345
	respHeader := a.getRespHeader()
346
	if respHeader == nil {
347
		return ""
348
	}
349
	return respHeader.XRespCode
350
}
351

352
func (a *RequestRow) ReqLength() (reqLength int64) {
353
	reqHeader := a.getReqHeader()
354
	if reqHeader == nil {
355
		return
356
	}
357
	return reqHeader.BodySize
358
}
359

360
func (a *RequestRow) ReqHost() string {
361
	reqHeader := a.getReqHeader()
362
	if reqHeader == nil {
363
		return ""
364
	}
365
	return reqHeader.Host
366
}
367

368
func (a *RequestRow) ReqParams() string {
369
	return a.data[TAG_PARAMS]
370
}
371

372
func (a *RequestRow) ReqFsize() (fsize int64) {
373
	rawQuery := a.getRawquery()
374
	if rawQuery == nil {
375
		return
376
	}
377
	fsize, _ = strconv.ParseInt(rawQuery.Get("fsize"), 10, 64)
378
	return
379
}
380

381
func (a *RequestRow) RemoteAddr() string {
382
	return ""
383
}
384

385
func (a *RequestRow) XWarns() []string {
386
	respHeader := a.getRespHeader()
387
	if respHeader == nil {
388
		return nil
389
	}
390
	return respHeader.XWarn
391
}
392

393
func (a *RequestRow) BillTag() string {
394
	respHeader := a.getRespHeader()
395
	if respHeader == nil {
396
		return ""
397
	}
398
	return respHeader.BillTag
399
}
400

401
func (a *RequestRow) BatchDelete() map[uint32]int64 {
402
	respHeader := a.getRespHeader()
403
	if respHeader == nil {
404
		return nil
405
	}
406
	return respHeader.BatchDeletes
407
}
408

409
type Xlog struct {
410
	Name    string
411
	Err     string
412
	MsSpend uint64
413
}
414

415
func (a *RequestRow) Xlogs() []string {
416
	respHeader := a.getRespHeader()
417
	if respHeader == nil {
418
		return nil
419
	}
420
	return respHeader.Xlog
421
}
422

423
func (a *RequestRow) XlogSearch(name string) (rets []Xlog) {
424
	xlogs := a.Xlogs()
425
	for _, log := range xlogs {
426
		items := strings.Split(log, ";")
427
		for _, item := range items {
428
			if !strings.HasPrefix(item, name) {
429
				continue
430
			}
431
			ret := parseXlog(item)
432
			if ret.Name == name {
433
				rets = append(rets, ret)
434
			}
435
		}
436
	}
437
	return
438
}
439

440
func parseXlog(s string) (ret Xlog) {
441
	idx := strings.Index(s, "/")
442
	if idx > 0 {
443
		ret.Err = s[idx+1:]
444
		s = s[:idx]
445
	}
446
	idx = strings.LastIndex(s, ":")
447
	if idx > 0 {
448
		ret.MsSpend, _ = strconv.ParseUint(s[idx+1:], 10, 64)
449
		s = s[:idx]
450
	}
451
	ret.Name = s
452
	return
453
}
454

455
func (a *RequestRow) XlogTime(name string) (msSpeed uint64) {
456
	rets := a.XlogSearch(name)
457
	for _, ret := range rets {
458
		if ret.MsSpend > msSpeed {
459
			msSpeed = ret.MsSpend
460
		}
461
	}
462
	return
463
}
464

465
func (a *RequestRow) XlogsTime(names []string) (msSpeedTotal uint64) {
466
	for _, name := range names {
467
		msSpeedTotal += a.XlogTime(name)
468
	}
469
	return
470
}
471

472
// apiWithParams returns api information with maxApiLevel( default 2).
473
func apiWithParams(service, method, path, host, params string, maxApiLevel int) (api string) {
474
	const unknown = ".unknown"
475
	if service == "" || method == "" {
476
		return "unknown.unknown"
477
	}
478
	stype := strings.ToLower(service)
479
	fields := strings.Split(strings.ToLower(path), "/")
480
	if len(fields) <= 1 {
481
		return stype + unknown
482
	}
483

484
	firstPath := fields[1]
485
	firstPathIndex := 1
486

487
	if (vre.MatchString(firstPath) || firstPath == "admin") && len(fields) > 2 && fields[2] != "" {
488
		firstPath = firstPath + "-" + fields[2]
489
		firstPathIndex = 2
490
	}
491

492
	// for defy api from apiserver
493
	if firstPath == "v2-tune" {
494
		return stype + ".v2-tune." + strings.Join(fields[firstPathIndex+1:], ".")
495
	}
496
	if !isValidApi(firstPath) {
497
		return stype + unknown
498
	}
499

500
	api = firstPath
501
	if maxApiLevel > 2 {
502
		level := 3
503
		index := firstPathIndex + 1
504
		length := len(fields)
505
		for level <= maxApiLevel && index < length {
506
			api += "." + fields[index]
507
			level++
508
			index++
509
		}
510
		if !isValidMultiPathApi(api) {
511
			return stype + unknown
512
		}
513
	}
514

515
	return stype + "." + api
516
}
517

518
func isValidApi(api string) bool {
519
	return isType(ALPHA|Digital|SubLetter|Underline, api)
520
}
521

522
func isValidMultiPathApi(api string) bool {
523
	return isType(ALPHA|Digital|DotLetter|SubLetter|Underline, api)
524
}
525