Skip to content

Commit e5272f0

Browse files
KbayeroKbayero
committed
Merge branch 'release/v11' of https://github.com/utmstack/UTMStack into release/v11
2 parents b5d9568 + b679ec6 commit e5272f0

17 files changed

Lines changed: 299 additions & 76 deletions

File tree

‎agent/modules/syslog.go‎

Lines changed: 109 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99
"io"
1010
"net"
1111
"os"
12+
"strconv"
1213
"strings"
1314
"time"
1415

@@ -20,6 +21,20 @@ import (
2021
"github.com/utmstack/UTMStack/agent/utils"
2122
)
2223

24+
const (
25+
MinBufferSize = 480
26+
RecommendedBufferSize = 2048
27+
MaxBufferSize = 8192
28+
UDPBufferSize = 2048
29+
)
30+
31+
type FramingMethod int
32+
33+
const (
34+
FramingNewline FramingMethod = iota
35+
FramingOctetCounting
36+
)
37+
2338
type SyslogModule struct {
2439
DataType string
2540
TCPListener listenerTCP
@@ -204,7 +219,7 @@ func (m *SyslogModule) enableUDP() {
204219
m.UDPListener.Listener = listener
205220
m.UDPListener.CTX, m.UDPListener.Cancel = context.WithCancel(context.Background())
206221

207-
buffer := make([]byte, 1024)
222+
buffer := make([]byte, UDPBufferSize)
208223
msgChannel := make(chan config.MSGDS)
209224

210225
go m.handleConnectionUDP(msgChannel)
@@ -291,6 +306,88 @@ func (m *SyslogModule) disableUDP() {
291306
}
292307
}
293308

309+
// detectFramingMethod detects the syslog framing method by peeking at the first byte
310+
func detectFramingMethod(reader *bufio.Reader) (FramingMethod, error) {
311+
firstByte, err := reader.Peek(1)
312+
if err != nil {
313+
utils.Logger.ErrorF("failed to peek first byte for framing detection: %v", err)
314+
return 0, fmt.Errorf("failed to peek first byte: %w", err)
315+
}
316+
317+
if firstByte[0] >= '0' && firstByte[0] <= '9' {
318+
return FramingOctetCounting, nil
319+
}
320+
321+
if firstByte[0] == '<' {
322+
return FramingNewline, nil
323+
}
324+
325+
utils.Logger.ErrorF("unknown framing method detected, first byte: 0x%02x", firstByte[0])
326+
return 0, fmt.Errorf("unknown framing method, first byte: 0x%02x", firstByte[0])
327+
}
328+
329+
// readOctetCountingFrame reads a syslog message using octet counting framing method
330+
func readOctetCountingFrame(reader *bufio.Reader) (string, error) {
331+
lengthStr, err := reader.ReadString(' ')
332+
if err != nil {
333+
utils.Logger.ErrorF("failed to read message length in octet counting frame: %v", err)
334+
return "", fmt.Errorf("failed to read message length: %w", err)
335+
}
336+
337+
lengthStr = strings.TrimSuffix(lengthStr, " ")
338+
msgLen, err := strconv.Atoi(lengthStr)
339+
if err != nil {
340+
utils.Logger.ErrorF("invalid message length '%s' in octet counting frame: %v", lengthStr, err)
341+
return "", fmt.Errorf("invalid message length '%s': %w", lengthStr, err)
342+
}
343+
344+
if msgLen < 1 {
345+
utils.Logger.ErrorF("message length %d is too small (minimum 1 byte)", msgLen)
346+
return "", fmt.Errorf("message length %d is too small (minimum 1)", msgLen)
347+
}
348+
if msgLen > MaxBufferSize {
349+
utils.Logger.ErrorF("message length %d exceeds maximum %d bytes", msgLen, MaxBufferSize)
350+
return "", fmt.Errorf("message length %d exceeds maximum %d", msgLen, MaxBufferSize)
351+
}
352+
353+
msgBytes := make([]byte, msgLen)
354+
_, err = io.ReadFull(reader, msgBytes)
355+
if err != nil {
356+
utils.Logger.ErrorF("failed to read %d byte message body: %v", msgLen, err)
357+
return "", fmt.Errorf("failed to read %d byte message body: %w", msgLen, err)
358+
}
359+
360+
return string(msgBytes), nil
361+
}
362+
363+
// readNewlineFrame reads a syslog message using newline-delimited framing method
364+
func readNewlineFrame(reader *bufio.Reader) (string, error) {
365+
message, err := reader.ReadString('\n')
366+
if err != nil {
367+
utils.Logger.ErrorF("failed to read newline-delimited message: %v", err)
368+
return "", fmt.Errorf("failed to read newline-delimited message: %w", err)
369+
}
370+
return message, nil
371+
}
372+
373+
// readSyslogMessage reads a syslog message with automatic framing detection
374+
func readSyslogMessage(reader *bufio.Reader) (string, error) {
375+
method, err := detectFramingMethod(reader)
376+
if err != nil {
377+
return "", err
378+
}
379+
380+
switch method {
381+
case FramingOctetCounting:
382+
return readOctetCountingFrame(reader)
383+
case FramingNewline:
384+
return readNewlineFrame(reader)
385+
default:
386+
utils.Logger.ErrorF("unsupported framing method: %d", method)
387+
return "", fmt.Errorf("unsupported framing method: %d", method)
388+
}
389+
}
390+
294391
func (m *SyslogModule) handleConnectionTCP(c net.Conn) {
295392
defer c.Close()
296393
reader := bufio.NewReader(c)
@@ -336,12 +433,17 @@ func (m *SyslogModule) handleConnectionTCP(c net.Conn) {
336433
case <-m.TCPListener.CTX.Done():
337434
return
338435
default:
339-
message, err := reader.ReadString('\n')
436+
message, err := readSyslogMessage(reader)
340437
if err != nil {
341-
if err == io.EOF || err.(net.Error).Timeout() {
438+
if err == io.EOF {
439+
utils.Logger.Info("TCP connection closed by %s", remoteAddr)
440+
return
441+
}
442+
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
443+
utils.Logger.Info("TCP connection timeout from %s", remoteAddr)
342444
return
343445
}
344-
utils.Logger.ErrorF("error reading tcp data: %v", err)
446+
utils.Logger.ErrorF("error reading syslog message from %s: %v", remoteAddr, err)
345447
return
346448
}
347449
msgChannel <- config.MSGDS{
@@ -398,12 +500,14 @@ func (m *SyslogModule) handleTLSConnection(conn net.Conn) {
398500
default:
399501
// Set read timeout for each message
400502
conn.SetDeadline(time.Now().Add(30 * time.Second))
401-
message, err := reader.ReadString('\n')
503+
message, err := readSyslogMessage(reader)
402504
if err != nil {
403505
if err == io.EOF {
506+
utils.Logger.Info("TLS connection closed by %s", remoteAddr)
404507
return
405508
}
406509
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
510+
utils.Logger.Info("TLS connection timeout from %s", remoteAddr)
407511
return
408512
}
409513
utils.Logger.ErrorF("error reading TLS data from %s: %v", remoteAddr, err)

‎backend/src/main/java/com/park/utmstack/aop/logging/impl/AlertLoggingAspect.java‎

Lines changed: 18 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
import com.park.utmstack.domain.chart_builder.types.query.FilterType;
66
import com.park.utmstack.domain.chart_builder.types.query.OperatorType;
77
import com.park.utmstack.domain.index_pattern.enums.SystemIndexPattern;
8-
import com.park.utmstack.domain.shared_types.alert.AlertType;
8+
import com.park.utmstack.domain.shared_types.alert.UtmAlert;
99
import com.park.utmstack.security.SecurityUtils;
1010
import com.park.utmstack.service.UtmAlertLogService;
1111
import com.park.utmstack.service.elasticsearch.ElasticsearchService;
@@ -87,14 +87,14 @@ public Object logManualAlertStatusChange(ProceedingJoinPoint joinPoint) throws T
8787
Integer status = (Integer) args[1];
8888
String statusObservation = (String) args[2];
8989

90-
List<AlertType> alerts = getAlerts(alertIds);
90+
List<UtmAlert> alerts = getAlerts(alertIds);
9191

9292
joinPoint.proceed();
9393

9494
if (CollectionUtils.isEmpty(alerts))
9595
return null;
9696

97-
for (AlertType alert : alerts) {
97+
for (UtmAlert alert : alerts) {
9898
UtmAlertLog alertLog = new UtmAlertLog();
9999
alertLog.setAlertId(alert.getId());
100100
alertLog.setLogUser(SecurityUtils.getCurrentUserLogin().orElse("system"));
@@ -146,19 +146,19 @@ public Object logAutomaticAlertStatusChange(ProceedingJoinPoint joinPoint) throw
146146
.size(Constants.LOG_ANALYZER_TOTAL_RESULTS)
147147
.query(query));
148148

149-
HitsMetadata<AlertType> response = elasticsearchService.search(sr, AlertType.class).hits();
149+
HitsMetadata<UtmAlert> response = elasticsearchService.search(sr, UtmAlert.class).hits();
150150

151151
joinPoint.proceed();
152152

153153
if (response.total().value() <= 0)
154154
return null;
155155

156-
List<AlertType> alerts = response.hits().stream().map(Hit::source).collect(Collectors.toList());
156+
List<UtmAlert> alerts = response.hits().stream().map(Hit::source).collect(Collectors.toList());
157157

158158
if (CollectionUtils.isEmpty(alerts))
159159
return null;
160160

161-
for (AlertType alert : alerts) {
161+
for (UtmAlert alert : alerts) {
162162
UtmAlertLog alertLog = new UtmAlertLog();
163163
alertLog.setAlertId(alert.getId());
164164
alertLog.setLogUser("system");
@@ -197,7 +197,7 @@ public Object logManualAlertTagsChange(ProceedingJoinPoint joinPoint) throws Thr
197197
List alertIds = (List) args[0];
198198
List tags = (List) args[1];
199199

200-
List<AlertType> alerts = getAlerts(alertIds);
200+
List<UtmAlert> alerts = getAlerts(alertIds);
201201

202202
joinPoint.proceed();
203203

@@ -206,7 +206,7 @@ public Object logManualAlertTagsChange(ProceedingJoinPoint joinPoint) throws Thr
206206

207207
String user = SecurityUtils.getCurrentUserLogin().orElse("system");
208208

209-
for (AlertType alert : alerts) {
209+
for (UtmAlert alert : alerts) {
210210
UtmAlertLog alertLog = new UtmAlertLog();
211211
alertLog.setAlertId(alert.getId());
212212
alertLog.setLogUser(user);
@@ -247,21 +247,21 @@ public Object logAutomaticAlertTagsChange(ProceedingJoinPoint joinPoint) throws
247247
SearchRequest request = SearchRequest.of(s -> s.query(query)
248248
.size(Constants.LOG_ANALYZER_TOTAL_RESULTS).index(indexPattern));
249249

250-
HitsMetadata<AlertType> hits = elasticsearchService.search(request, AlertType.class).hits();
250+
HitsMetadata<UtmAlert> hits = elasticsearchService.search(request, UtmAlert.class).hits();
251251

252252
joinPoint.proceed();
253253

254254
if (hits.total().value() <= 0)
255255
return null;
256256

257-
List<AlertType> alerts = hits.hits().stream().map(Hit::source).collect(Collectors.toList());
257+
List<UtmAlert> alerts = hits.hits().stream().map(Hit::source).collect(Collectors.toList());
258258

259259
if (CollectionUtils.isEmpty(alerts))
260260
return null;
261261

262262
String user = SecurityUtils.getCurrentUserLogin().orElse("system");
263263

264-
for (AlertType alert : alerts) {
264+
for (UtmAlert alert : alerts) {
265265
UtmAlertLog alertLog = new UtmAlertLog();
266266
alertLog.setAlertId(alert.getId());
267267
alertLog.setLogUser(user);
@@ -294,7 +294,7 @@ public Object logManualAlertNotesChange(ProceedingJoinPoint joinPoint) throws Th
294294
String alertId = (String) args[0];
295295
String notes = (String) args[1];
296296

297-
List<AlertType> alerts = getAlerts(Collections.singletonList(alertId));
297+
List<UtmAlert> alerts = getAlerts(Collections.singletonList(alertId));
298298

299299
joinPoint.proceed();
300300

@@ -303,7 +303,7 @@ public Object logManualAlertNotesChange(ProceedingJoinPoint joinPoint) throws Th
303303

304304
String user = SecurityUtils.getCurrentUserLogin().orElse("system");
305305

306-
for (AlertType alert : alerts) {
306+
for (UtmAlert alert : alerts) {
307307
UtmAlertLog alertLog = new UtmAlertLog();
308308
alertLog.setAlertId(alert.getId());
309309
alertLog.setLogUser(user);
@@ -347,19 +347,19 @@ public Object logConvertToIncident(ProceedingJoinPoint joinPoint) throws Throwab
347347
SearchRequest request = SearchRequest.of(s -> s.size(Constants.LOG_ANALYZER_TOTAL_RESULTS)
348348
.query(query).index(indexPattern));
349349

350-
HitsMetadata<AlertType> hits = elasticsearchService.search(request, AlertType.class).hits();
350+
HitsMetadata<UtmAlert> hits = elasticsearchService.search(request, UtmAlert.class).hits();
351351

352352
joinPoint.proceed();
353353

354354
if (hits.total().value() <= 0)
355355
return null;
356356

357-
List<AlertType> alerts = hits.hits().stream().map(Hit::source).collect(Collectors.toList());
357+
List<UtmAlert> alerts = hits.hits().stream().map(Hit::source).collect(Collectors.toList());
358358

359359
if (CollectionUtils.isEmpty(alerts))
360360
return null;
361361

362-
for (AlertType alert : alerts) {
362+
for (UtmAlert alert : alerts) {
363363
UtmAlertLog alertLog = new UtmAlertLog();
364364
alertLog.setAlertId(alert.getId());
365365
alertLog.setLogUser(incidentCreatedBy);
@@ -393,14 +393,14 @@ public Object logConvertToIncident(ProceedingJoinPoint joinPoint) throws Throwab
393393
* @return
394394
* @throws Exception
395395
*/
396-
private List<AlertType> getAlerts(List<?> ids) throws Exception {
396+
private List<UtmAlert> getAlerts(List<?> ids) throws Exception {
397397
final String ctx = CLASS_NAME + ".getAlerts";
398398
try {
399399
List<FilterType> filters = new ArrayList<>();
400400
filters.add(new FilterType(Constants.alertIdKeyword, OperatorType.IS_ONE_OF_TERMS, ids));
401401
SearchRequest request = SearchRequest.of(s -> s.query(SearchUtil.toQuery(filters))
402402
.index(Constants.SYS_INDEX_PATTERN.get(SystemIndexPattern.ALERTS)));
403-
HitsMetadata<AlertType> hits = elasticsearchService.search(request, AlertType.class).hits();
403+
HitsMetadata<UtmAlert> hits = elasticsearchService.search(request, UtmAlert.class).hits();
404404
if (hits.total().value() <= 0)
405405
return Collections.emptyList();
406406
return hits.hits().stream().map(Hit::source).collect(Collectors.toList());

‎backend/src/main/java/com/park/utmstack/domain/reports/types/IncidentType.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,20 @@
11
package com.park.utmstack.domain.reports.types;
22

33
import com.park.utmstack.domain.incident_response.UtmIncidentJob;
4-
import com.park.utmstack.domain.shared_types.alert.AlertType;
4+
import com.park.utmstack.domain.shared_types.alert.UtmAlert;
55

66
import java.util.List;
77

88
public class IncidentType {
9-
private AlertType incident;
9+
private UtmAlert incident;
1010
private List<UtmIncidentJob> srcResponses;
1111
private List<UtmIncidentJob> destResponses;
1212

13-
public AlertType getIncident() {
13+
public UtmAlert getIncident() {
1414
return incident;
1515
}
1616

17-
public void setIncident(AlertType incident) {
17+
public void setIncident(UtmAlert incident) {
1818
this.incident = incident;
1919
}
2020

backend/src/main/java/com/park/utmstack/domain/shared_types/alert/AlertType.java renamed to backend/src/main/java/com/park/utmstack/domain/shared_types/alert/UtmAlert.java

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
@JsonIgnoreProperties(ignoreUnknown = true)
1919
@Getter
2020
@Setter
21-
public class AlertType {
21+
public class UtmAlert {
2222
@JsonProperty("@timestamp")
2323
private String timestamp;
2424

@@ -106,6 +106,10 @@ public class AlertType {
106106
@JsonProperty("logs")
107107
private List<String> logs;
108108

109+
private String assetGroupName;
110+
111+
private Long assetGroupId;
112+
109113
public Instant getTimestampAsInstant() {
110114
if (StringUtils.hasText(timestamp))
111115
return Instant.parse(timestamp);

‎backend/src/main/java/com/park/utmstack/repository/network_scan/UtmNetworkScanRepository.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -123,4 +123,10 @@ void updateGroup(@Param("assetIds") List<Long> assetIds,
123123

124124
@Query(nativeQuery = true, value = "select n.asset_name from utm_network_scan n where n.asset_name is not null and n.is_agent is true and n.asset_alive is true and n.asset_status <> 'MISSING' and n.asset_os_platform = :platform")
125125
List<String> findAgentNamesByPlatform(@Param("platform") String platform);
126+
127+
@Query("SELECT ns.assetName, ns.groupId, ag.groupName " +
128+
"FROM UtmNetworkScan ns " +
129+
"JOIN UtmAssetGroup ag ON ns.groupId = ag.id " +
130+
"WHERE ns.groupId IS NOT NULL")
131+
List<Object[]> findAllAssetGroupMappings();
126132
}

0 commit comments

Comments
 (0)