Изменен формат сообщений в ws, рефакторинг
This commit is contained in:
@@ -23,7 +23,7 @@ public class PackmateApplication {
|
||||
|
||||
@EventListener(ApplicationReadyEvent.class)
|
||||
public void afterStartup(ApplicationReadyEvent event) throws PcapNativeException {
|
||||
if(enableCapture) {
|
||||
if (enableCapture) {
|
||||
final PcapWorker pcapWorker = event.getApplicationContext().getBean(PcapWorker.class);
|
||||
pcapWorker.start();
|
||||
}
|
||||
|
||||
@@ -12,8 +12,8 @@ import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Component;
|
||||
import ru.serega6531.packmate.model.CtfService;
|
||||
import ru.serega6531.packmate.model.Protocol;
|
||||
import ru.serega6531.packmate.model.UnfinishedStream;
|
||||
import ru.serega6531.packmate.model.enums.Protocol;
|
||||
import ru.serega6531.packmate.model.pojo.UnfinishedStream;
|
||||
import ru.serega6531.packmate.service.ServicesService;
|
||||
import ru.serega6531.packmate.service.StreamService;
|
||||
|
||||
@@ -208,13 +208,13 @@ public class PcapWorker implements PacketListener {
|
||||
if (stream.getProtocol() == protocol) {
|
||||
final List<ru.serega6531.packmate.model.Packet> packets = entry.getValue();
|
||||
if (System.currentTimeMillis() - packets.get(packets.size() - 1).getTimestamp() > timeoutMillis) {
|
||||
if(streamService.saveNewStream(stream, packets)) {
|
||||
if (streamService.saveNewStream(stream, packets)) {
|
||||
streamsClosed++;
|
||||
}
|
||||
|
||||
iterator.remove();
|
||||
|
||||
if(protocol == Protocol.TCP) {
|
||||
if (protocol == Protocol.TCP) {
|
||||
fins.remove(stream);
|
||||
acks.remove(stream);
|
||||
}
|
||||
|
||||
@@ -5,7 +5,7 @@ import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
import ru.serega6531.packmate.model.Protocol;
|
||||
import ru.serega6531.packmate.model.enums.Protocol;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
@@ -29,12 +29,12 @@ public class TimeoutStreamsSaver {
|
||||
@Scheduled(fixedRateString = "PT${timeout-stream-check-interval}S", initialDelayString = "PT${timeout-stream-check-interval}S")
|
||||
public void saveStreams() {
|
||||
int streamsClosed = pcapWorker.closeTimeoutStreams(Protocol.UDP, udpStreamTimeoutMillis);
|
||||
if(streamsClosed > 0) {
|
||||
if (streamsClosed > 0) {
|
||||
log.info("Закрыто {} udp стримов", streamsClosed);
|
||||
}
|
||||
|
||||
streamsClosed = pcapWorker.closeTimeoutStreams(Protocol.TCP, tcpStreamTimeoutMillis);
|
||||
if(streamsClosed > 0) {
|
||||
if (streamsClosed > 0) {
|
||||
log.info("Закрыто {} tcp стримов", streamsClosed);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
package ru.serega6531.packmate.controller;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.web.bind.annotation.*;
|
||||
import org.springframework.web.bind.annotation.PathVariable;
|
||||
import org.springframework.web.bind.annotation.PostMapping;
|
||||
import org.springframework.web.bind.annotation.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
import ru.serega6531.packmate.model.Packet;
|
||||
import ru.serega6531.packmate.model.Pagination;
|
||||
import ru.serega6531.packmate.model.Stream;
|
||||
import ru.serega6531.packmate.service.PacketService;
|
||||
import ru.serega6531.packmate.service.StreamService;
|
||||
@@ -28,7 +30,7 @@ public class PacketController {
|
||||
@PostMapping("/{streamId}")
|
||||
public List<Packet> getPacketsForStream(@PathVariable long streamId) {
|
||||
final Optional<Stream> stream = streamService.find(streamId);
|
||||
if(stream.isPresent()) {
|
||||
if (stream.isPresent()) {
|
||||
return packetService.getPacketsForStream(stream.get());
|
||||
} else {
|
||||
return Collections.emptyList();
|
||||
|
||||
@@ -2,8 +2,8 @@ package ru.serega6531.packmate.controller;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.web.bind.annotation.*;
|
||||
import ru.serega6531.packmate.model.Pagination;
|
||||
import ru.serega6531.packmate.model.Stream;
|
||||
import ru.serega6531.packmate.model.pojo.Pagination;
|
||||
import ru.serega6531.packmate.service.StreamService;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
@@ -4,6 +4,7 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
|
||||
import lombok.Data;
|
||||
import lombok.ToString;
|
||||
import org.hibernate.annotations.GenericGenerator;
|
||||
import ru.serega6531.packmate.model.enums.PatternType;
|
||||
|
||||
import javax.persistence.*;
|
||||
import java.util.List;
|
||||
|
||||
@@ -1,5 +0,0 @@
|
||||
package ru.serega6531.packmate.model;
|
||||
|
||||
public enum Protocol {
|
||||
TCP, UDP
|
||||
}
|
||||
@@ -4,6 +4,7 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
|
||||
import lombok.Data;
|
||||
import lombok.ToString;
|
||||
import org.hibernate.annotations.GenericGenerator;
|
||||
import ru.serega6531.packmate.model.enums.Protocol;
|
||||
|
||||
import javax.persistence.*;
|
||||
import java.util.List;
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
package ru.serega6531.packmate.model;
|
||||
package ru.serega6531.packmate.model.enums;
|
||||
|
||||
public enum PatternType {
|
||||
INPUT, OUTPUT, BOTH
|
||||
@@ -0,0 +1,5 @@
|
||||
package ru.serega6531.packmate.model.enums;
|
||||
|
||||
public enum Protocol {
|
||||
TCP, UDP
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package ru.serega6531.packmate.model.enums;
|
||||
|
||||
public enum SubscriptionMessageType {
|
||||
|
||||
SAVE_SERVICE, SAVE_PATTERN, DELETE_SERVICE, DELETE_PATTERN, NEW_STREAM
|
||||
|
||||
}
|
||||
@@ -1,7 +1,8 @@
|
||||
package ru.serega6531.packmate.model;
|
||||
package ru.serega6531.packmate.model.pojo;
|
||||
|
||||
import lombok.Data;
|
||||
import org.springframework.data.domain.Sort;
|
||||
import ru.serega6531.packmate.model.Pattern;
|
||||
|
||||
@Data
|
||||
public class Pagination {
|
||||
@@ -0,0 +1,14 @@
|
||||
package ru.serega6531.packmate.model.pojo;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Data;
|
||||
import ru.serega6531.packmate.model.enums.SubscriptionMessageType;
|
||||
|
||||
@Data
|
||||
@AllArgsConstructor
|
||||
public class SubscriptionMessage {
|
||||
|
||||
private SubscriptionMessageType type;
|
||||
private Object value;
|
||||
|
||||
}
|
||||
@@ -1,7 +1,8 @@
|
||||
package ru.serega6531.packmate.model;
|
||||
package ru.serega6531.packmate.model.pojo;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Getter;
|
||||
import ru.serega6531.packmate.model.enums.Protocol;
|
||||
|
||||
import java.net.Inet4Address;
|
||||
|
||||
@@ -17,7 +18,7 @@ public class UnfinishedStream {
|
||||
|
||||
@Override
|
||||
public boolean equals(Object obj) {
|
||||
if(!(obj instanceof UnfinishedStream)) {
|
||||
if (!(obj instanceof UnfinishedStream)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -2,7 +2,7 @@ package ru.serega6531.packmate.repository;
|
||||
|
||||
import org.springframework.data.jpa.repository.JpaRepository;
|
||||
import ru.serega6531.packmate.model.Pattern;
|
||||
import ru.serega6531.packmate.model.PatternType;
|
||||
import ru.serega6531.packmate.model.enums.PatternType;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
|
||||
@@ -8,8 +8,10 @@ import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import ru.serega6531.packmate.model.FoundPattern;
|
||||
import ru.serega6531.packmate.model.Pattern;
|
||||
import ru.serega6531.packmate.model.PatternType;
|
||||
import ru.serega6531.packmate.model.enums.PatternType;
|
||||
import ru.serega6531.packmate.model.Stream;
|
||||
import ru.serega6531.packmate.model.enums.SubscriptionMessageType;
|
||||
import ru.serega6531.packmate.model.pojo.SubscriptionMessage;
|
||||
import ru.serega6531.packmate.repository.PatternRepository;
|
||||
|
||||
import java.util.*;
|
||||
@@ -22,14 +24,19 @@ public class PatternService {
|
||||
|
||||
private final PatternRepository repository;
|
||||
private final StreamService streamService;
|
||||
private final StreamSubscriptionService subscriptionService;
|
||||
|
||||
private final Map<Integer, Pattern> patterns = new HashMap<>();
|
||||
private final Map<String, java.util.regex.Pattern> compiledPatterns = new HashMap<>();
|
||||
|
||||
@Autowired
|
||||
public PatternService(PatternRepository repository, @Lazy StreamService streamService) {
|
||||
public PatternService(PatternRepository repository,
|
||||
@Lazy StreamService streamService,
|
||||
StreamSubscriptionService subscriptionService) {
|
||||
this.repository = repository;
|
||||
this.streamService = streamService;
|
||||
this.subscriptionService = subscriptionService;
|
||||
|
||||
repository.findAll().forEach(p -> patterns.put(p.getId(), p));
|
||||
log.info("Loaded {} patterns", patterns.size());
|
||||
}
|
||||
@@ -108,6 +115,7 @@ public class PatternService {
|
||||
patterns.remove(id);
|
||||
compiledPatterns.remove(pattern.getValue());
|
||||
repository.delete(pattern);
|
||||
subscriptionService.broadcast(new SubscriptionMessage(SubscriptionMessageType.DELETE_PATTERN, id));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -115,6 +123,7 @@ public class PatternService {
|
||||
log.info("Добавлен новый паттерн {} со значением {}", pattern.getName(), pattern.getValue());
|
||||
final Pattern saved = repository.save(pattern);
|
||||
patterns.put(saved.getId(), pattern);
|
||||
subscriptionService.broadcast(new SubscriptionMessage(SubscriptionMessageType.SAVE_PATTERN, saved));
|
||||
return saved;
|
||||
}
|
||||
|
||||
|
||||
@@ -4,6 +4,8 @@ import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Service;
|
||||
import ru.serega6531.packmate.model.CtfService;
|
||||
import ru.serega6531.packmate.model.enums.SubscriptionMessageType;
|
||||
import ru.serega6531.packmate.model.pojo.SubscriptionMessage;
|
||||
import ru.serega6531.packmate.repository.ServiceRepository;
|
||||
|
||||
import java.util.Collection;
|
||||
@@ -16,11 +18,15 @@ import java.util.Optional;
|
||||
public class ServicesService {
|
||||
|
||||
private final ServiceRepository repository;
|
||||
private final StreamSubscriptionService subscriptionService;
|
||||
|
||||
private final Map<Integer, CtfService> services = new HashMap<>();
|
||||
|
||||
@Autowired
|
||||
public ServicesService(ServiceRepository repository) {
|
||||
public ServicesService(ServiceRepository repository, StreamSubscriptionService subscriptionService) {
|
||||
this.repository = repository;
|
||||
this.subscriptionService = subscriptionService;
|
||||
|
||||
repository.findAll().forEach(s -> services.put(s.getPort(), s));
|
||||
log.info("Loaded {} services", services.size());
|
||||
}
|
||||
@@ -47,12 +53,14 @@ public class ServicesService {
|
||||
log.info("Удален сервис на порту {}", port);
|
||||
services.remove(port);
|
||||
repository.deleteById(port);
|
||||
subscriptionService.broadcast(new SubscriptionMessage(SubscriptionMessageType.DELETE_SERVICE, port));
|
||||
}
|
||||
|
||||
public CtfService save(CtfService service) {
|
||||
log.info("Добавлен или изменен сервис {} на порту {}", service.getName(), service.getPort());
|
||||
final CtfService saved = repository.save(service);
|
||||
services.put(saved.getPort(), service);
|
||||
subscriptionService.broadcast(new SubscriptionMessage(SubscriptionMessageType.SAVE_SERVICE, saved));
|
||||
return saved;
|
||||
}
|
||||
|
||||
|
||||
@@ -11,6 +11,10 @@ import org.springframework.data.domain.Sort;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import ru.serega6531.packmate.model.*;
|
||||
import ru.serega6531.packmate.model.enums.SubscriptionMessageType;
|
||||
import ru.serega6531.packmate.model.pojo.Pagination;
|
||||
import ru.serega6531.packmate.model.pojo.SubscriptionMessage;
|
||||
import ru.serega6531.packmate.model.pojo.UnfinishedStream;
|
||||
import ru.serega6531.packmate.repository.StreamRepository;
|
||||
|
||||
import java.io.ByteArrayInputStream;
|
||||
@@ -97,7 +101,7 @@ public class StreamService {
|
||||
}
|
||||
}
|
||||
|
||||
if(unpackGzippedHttp) {
|
||||
if (unpackGzippedHttp) {
|
||||
boolean gzipStarted = false;
|
||||
int gzipStartPacket = 0;
|
||||
int gzipEndPacket;
|
||||
@@ -111,7 +115,7 @@ public class StreamService {
|
||||
List<Packet> cut = packets.subList(gzipStartPacket, gzipEndPacket + 1);
|
||||
|
||||
Packet decompressed = decompressGzipPackets(cut);
|
||||
if(decompressed != null) {
|
||||
if (decompressed != null) {
|
||||
packets.removeAll(cut);
|
||||
packets.add(gzipStartPacket, decompressed);
|
||||
gzipStarted = false;
|
||||
@@ -128,7 +132,7 @@ public class StreamService {
|
||||
List<Packet> cut = packets.subList(gzipStartPacket, gzipEndPacket + 1);
|
||||
|
||||
Packet decompressed = decompressGzipPackets(cut);
|
||||
if(decompressed != null) {
|
||||
if (decompressed != null) {
|
||||
packets.removeAll(cut);
|
||||
packets.add(gzipStartPacket, decompressed);
|
||||
gzipStarted = false;
|
||||
@@ -152,7 +156,7 @@ public class StreamService {
|
||||
List<Packet> cut = packets.subList(gzipStartPacket, gzipEndPacket + 1);
|
||||
|
||||
Packet decompressed = decompressGzipPackets(cut);
|
||||
if(decompressed != null) {
|
||||
if (decompressed != null) {
|
||||
packets.removeAll(cut);
|
||||
packets.add(gzipStartPacket, decompressed);
|
||||
}
|
||||
@@ -163,13 +167,13 @@ public class StreamService {
|
||||
for (Packet packet : packets) {
|
||||
String content = new String(packet.getContent());
|
||||
final Matcher matcher = userAgentPattern.matcher(content);
|
||||
if(matcher.find()) {
|
||||
if (matcher.find()) {
|
||||
ua = matcher.group(1);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if(ua != null) {
|
||||
if (ua != null) {
|
||||
stream.setUserAgentHash(calculateUserAgentHash(ua));
|
||||
}
|
||||
|
||||
@@ -191,7 +195,7 @@ public class StreamService {
|
||||
savedStream.setPackets(packetService.saveAll(packets));
|
||||
savedStream = save(savedStream);
|
||||
|
||||
subscriptionService.broadcastNewStream(savedStream);
|
||||
subscriptionService.broadcast(new SubscriptionMessage(SubscriptionMessageType.NEW_STREAM, savedStream));
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
@@ -8,7 +8,7 @@ import org.springframework.stereotype.Service;
|
||||
import org.springframework.web.socket.TextMessage;
|
||||
import org.springframework.web.socket.WebSocketSession;
|
||||
import org.springframework.web.socket.sockjs.SockJsTransportFailureException;
|
||||
import ru.serega6531.packmate.model.Stream;
|
||||
import ru.serega6531.packmate.model.pojo.SubscriptionMessage;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
@@ -28,26 +28,26 @@ public class StreamSubscriptionService {
|
||||
}
|
||||
|
||||
public void addSubscriber(WebSocketSession session) {
|
||||
log.info("Подписан пользователь {}", session.getRemoteAddress().getHostName());
|
||||
subscribers.add(session);
|
||||
log.info("Подписан пользователь {}", session.getRemoteAddress().getHostName());
|
||||
}
|
||||
|
||||
public void removeSubscriber(WebSocketSession session) {
|
||||
log.info("Отписан пользователь {}", session.getRemoteAddress().getHostName());
|
||||
subscribers.remove(session);
|
||||
log.info("Отписан пользователь {}", session.getRemoteAddress().getHostName());
|
||||
}
|
||||
|
||||
public void broadcastNewStream(Stream stream) {
|
||||
public void broadcast(SubscriptionMessage message) {
|
||||
subscribers.forEach(s -> {
|
||||
try {
|
||||
s.sendMessage(objectToTextMessage(stream));
|
||||
s.sendMessage(objectToTextMessage(message));
|
||||
} catch (IOException | SockJsTransportFailureException e) {
|
||||
e.printStackTrace();
|
||||
log.warn("WS", e);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private TextMessage objectToTextMessage(Object object) throws JsonProcessingException {
|
||||
private TextMessage objectToTextMessage(SubscriptionMessage object) throws JsonProcessingException {
|
||||
return new TextMessage(mapper.writeValueAsString(object));
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user