Merge branch 'parse-pcap' into 'master'

Parse pcap

See merge request packmate/Packmate!6

.dockerignore

@@ -4,4 +4,5 @@ screenshots
 .*
 docker-compose.yml
 Dockerfile_*
-README*
+docker/postgres_data
+README*

.gitignore

@@ -1,4 +1,6 @@
 src/main/resources/static/*
+*.pcap
+docker/postgres_data
 
 HELP.md
 .gradle

README.md

@@ -7,6 +7,7 @@
 Утилита перехвата и анализа трафика для CTF.
 
 #### Фичи:
+* Поддерживает перехват живого трафика и обработку pcap файлов
 * Поддерживает текстовые и бинарные сервисы
 * Умеет отображать совпадения паттернов в пакетах цветом
   * Подстрока
@@ -42,7 +43,7 @@ git submodule update --init --recursive
 сетевой интерфейс хоста, его название указывается переменной окружения (об этом ниже).
 
 `packmate-db` настроен на прослушивание порта 65001 с локальным IP.  
-При этом файлы БД не монтируются как volume, поэтому при пересоздании контейнера все стримы теряются.
+Файлы БД сохраняются в ./docker/postgres_data, поэтому для обнуления базы нужно удалить эту папку.
 
 ### Настройка
 Программа берет основные настройки из переменных окружения, поэтому для удобства
@@ -51,9 +52,7 @@ git submodule update --init --recursive
 
 В файле необходимо прописать:
 ```bash
-# Интерфейс, на котором производится перехват трафика
+# Локальный IP сервера на указанном интерфейсе или в pcap файле
-PACKMATE_INTERFACE=wlan0
-# Локальный IP сервера на указанном интерфейсе
 PACKMATE_LOCAL_IP=192.168.1.124
 # Имя пользователя для web-авторизации
 PACKMATE_WEB_LOGIN=SomeUser
@@ -61,6 +60,21 @@ PACKMATE_WEB_LOGIN=SomeUser
 PACKMATE_WEB_PASSWORD=SomeSecurePassword
 ```
 
+Если мы перехватываем трафик сервера (лучший вариант, если есть возможность):
+```bash
+# Режим работы - перехват
+PACKMATE_MODE=LIVE
+# Интерфейс, на котором производится перехват трафика
+PACKMATE_INTERFACE=wlan0
+```
+Если мы анализируем pcap дамп:
+```bash
+# Режим работы - анализ файла
+PACKMATE_MODE=FILE
+# Путь до файла от корня проекта
+PACKMATE_PCAP_FILE=dump.pcap
+```
+
 ### Запуск
 После указания нужных настроек в env-файле, можно запустить приложение:
 ```bash

README_EN.md

@@ -7,6 +7,7 @@
 Advanced network traffic flow analyzer for A/D CTFs.
 
 #### Features:
+* Can monitor live traffic or analyze pcap files
 * Supports binary and textual services
 * Can highlight found patterns in packets
   * Substring
@@ -41,7 +42,7 @@ git submodule update --init --recursive
 This program uses Docker and docker-compose.
 
 `packmate-db` will listen to port 65001 at localhost.  
-Database files do not mount as volume, so upon container recreation, all data will be lost.
+Database files are saved in ./docker/postgres_data, so to reset database you have to delete that directory.
 
 ### Settings
 This program retrieves settings from environment variables, 
@@ -50,9 +51,7 @@ It must be called `.env` and located at the root of the project.
 
 Contents of the file:
 ```bash
-# Interface to capture on
+# Local IP on network interface or in pcap file to tell incoming packets from outgoing
-PACKMATE_INTERFACE=wlan0
-# Local IP on said interface to tell incoming packets from outgoing
 PACKMATE_LOCAL_IP=192.168.1.124
 # Username for the web interface
 PACKMATE_WEB_LOGIN=SomeUser
@@ -60,6 +59,21 @@ PACKMATE_WEB_LOGIN=SomeUser
 PACKMATE_WEB_PASSWORD=SomeSecurePassword
 ```
 
+If we are capturing live traffic (best option if possible):
+```bash
+# Mode: capturing
+PACKMATE_MODE=LIVE
+# Interface to capture on
+PACKMATE_INTERFACE=wlan0
+```
+If we are analyzing pcap dump:
+```bash
+# Mode: dump analyzing
+PACKMATE_MODE=FILE
+# Path to pcap file from project root
+PACKMATE_PCAP_FILE=dump.pcap
+```
+
 ### Launch
 After filling in env file you can launch the app:
 ```bash

docker-compose.yml

@@ -7,6 +7,8 @@ services:
       DB_NAME: ${PACKMATE_DB_NAME:-packmate}
       INTERFACE: ${PACKMATE_INTERFACE}
       LOCAL_IP: ${PACKMATE_LOCAL_IP}
+      MODE: ${PACKMATE_MODE:-LIVE}
+      PCAP_FILE: ${PACKMATE_PCAP_FILE}
       WEB_LOGIN: ${PACKMATE_WEB_LOGIN:-BinaryBears}
       WEB_PASSWORD: ${PACKMATE_WEB_PASSWORD:-123456}
     env_file:
@@ -21,6 +23,7 @@ services:
       "java", "-Djava.net.preferIPv4Stack=true", "-Djava.net.preferIPv4Addresses=true",
       "-jar", "/app/app.jar", "--spring.datasource.url=jdbc:postgresql://127.0.0.1:65001/$${DB_NAME}",
       "--spring.datasource.username=$${DB_USER}", "--spring.datasource.password=$${DB_PASSWORD}",
+      "--capture-mode=$${MODE}", "--pcap-file=$${PCAP_FILE}",
       "--interface-name=$${INTERFACE}", "--local-ip=$${LOCAL_IP}", "--account-login=$${WEB_LOGIN}",
       "--account-password=$${WEB_PASSWORD}", "--server.port=65000", "--server.address=0.0.0.0"
     ]
@@ -38,6 +41,8 @@ services:
         POSTGRES_DB: ${PACKMATE_DB_NAME:-packmate}
     env_file:
       - .env
+    volumes:
+    - "./docker/postgres_data:/var/lib/postgresql/data"
     network_mode: "host"
     image: packmate-db:v1
     restart: unless-stopped

src/main/java/ru/serega6531/packmate/PackmateApplication.java

@@ -6,26 +6,27 @@ import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.context.event.ApplicationReadyEvent;
 import org.springframework.context.event.EventListener;
-import org.springframework.scheduling.annotation.EnableScheduling;
+import ru.serega6531.packmate.model.enums.CaptureMode;
-import org.springframework.web.socket.config.annotation.EnableWebSocket;
+import ru.serega6531.packmate.service.PcapService;
 
 @SpringBootApplication
-@EnableScheduling
-@EnableWebSocket
 public class PackmateApplication {
 
     @Value("${enable-capture}")
     private boolean enableCapture;
 
+    @Value("${capture-mode}")
+    private CaptureMode captureMode;
+
     public static void main(String[] args) {
         SpringApplication.run(PackmateApplication.class, args);
     }
 
     @EventListener(ApplicationReadyEvent.class)
     public void afterStartup(ApplicationReadyEvent event) throws PcapNativeException {
-        if (enableCapture) {
+        if (enableCapture && captureMode == CaptureMode.LIVE) {
-            final PcapWorker pcapWorker = event.getApplicationContext().getBean(PcapWorker.class);
+            final PcapService pcapService = event.getApplicationContext().getBean(PcapService.class);
-            pcapWorker.start();
+            pcapService.start();
         }
     }
 

src/main/java/ru/serega6531/packmate/TimeoutStreamsSaver.java

@@ -3,14 +3,17 @@ package ru.serega6531.packmate;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import ru.serega6531.packmate.model.enums.Protocol;
+import ru.serega6531.packmate.pcap.PcapWorker;
 
 import java.util.concurrent.TimeUnit;
 
 @Component
 @Slf4j
+@ConditionalOnProperty(name = "capture-mode", havingValue = "LIVE")
 public class TimeoutStreamsSaver {
 
     private final PcapWorker pcapWorker;

src/main/java/ru/serega6531/packmate/configuration/ApplicationConfiguration.java

@@ -1,20 +1,35 @@
-package ru.serega6531.packmate;
+package ru.serega6531.packmate.configuration;
 
+import org.pcap4j.core.PcapNativeException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.security.config.annotation.authentication.builders.AuthenticationManagerBuilder;
 import org.springframework.security.config.annotation.web.builders.HttpSecurity;
 import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
 import org.springframework.security.config.annotation.web.configuration.WebSecurityConfigurerAdapter;
 import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;
 import org.springframework.security.crypto.password.PasswordEncoder;
+import org.springframework.web.socket.config.annotation.EnableWebSocket;
 import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
 import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
+import ru.serega6531.packmate.WebSocketHandler;
+import ru.serega6531.packmate.model.enums.CaptureMode;
+import ru.serega6531.packmate.pcap.FilePcapWorker;
+import ru.serega6531.packmate.pcap.LivePcapWorker;
+import ru.serega6531.packmate.pcap.PcapWorker;
+import ru.serega6531.packmate.service.ServicesService;
+import ru.serega6531.packmate.service.StreamService;
+import ru.serega6531.packmate.service.SubscriptionService;
+
+import java.net.UnknownHostException;
 
 @Configuration
 @EnableWebSecurity
+@EnableScheduling
+@EnableWebSocket
 public class ApplicationConfiguration extends WebSecurityConfigurerAdapter implements WebSocketConfigurer {
 
     @Value("${account-login}")
@@ -30,6 +45,22 @@ public class ApplicationConfiguration extends WebSecurityConfigurerAdapter imple
         this.webSocketHandler = webSocketHandler;
     }
 
+    @Bean(destroyMethod = "stop")
+    @Autowired
+    public PcapWorker pcapWorker(ServicesService servicesService,
+                                 StreamService streamService,
+                                 SubscriptionService subscriptionService,
+                                 @Value("${local-ip}") String localIpString,
+                                 @Value("${interface-name}") String interfaceName,
+                                 @Value("${pcap-file}") String filename,
+                                 @Value("${capture-mode}") CaptureMode captureMode) throws PcapNativeException, UnknownHostException {
+        if(captureMode == CaptureMode.LIVE) {
+            return new LivePcapWorker(servicesService, streamService, localIpString, interfaceName);
+        } else {
+            return new FilePcapWorker(servicesService, streamService, subscriptionService, localIpString, filename);
+        }
+    }
+
     @Autowired
     public void configureGlobal(AuthenticationManagerBuilder auth) throws Exception {
         auth.inMemoryAuthentication()

src/main/java/ru/serega6531/packmate/controller/PcapController.java

@@ -0,0 +1,32 @@
+package ru.serega6531.packmate.controller;
+
+import org.pcap4j.core.PcapNativeException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+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.service.PcapService;
+
+@RestController
+@RequestMapping("/api/pcap/")
+public class PcapController {
+
+    private final PcapService service;
+
+    @Autowired
+    public PcapController(PcapService service) {
+        this.service = service;
+    }
+
+    @GetMapping("/started")
+    public boolean started() {
+        return service.isStarted();
+    }
+
+    @PostMapping("/start")
+    public void start() throws PcapNativeException {
+        service.start();
+    }
+
+}

src/main/java/ru/serega6531/packmate/model/Packet.java

@@ -24,6 +24,7 @@ import java.util.Set;
 @NoArgsConstructor
 @AllArgsConstructor
 @Builder
+@Table(indexes = { @Index(name = "stream_id_index", columnList = "stream_id") })
 public class Packet {
 
     @Id

src/main/java/ru/serega6531/packmate/model/Stream.java

@@ -35,6 +35,7 @@ public class Stream {
 
     @OneToMany(mappedBy = "stream", cascade = CascadeType.ALL)
     @JsonIgnore
+    @OrderBy("id")
     private List<Packet> packets;
 
     private long startTimestamp;

src/main/java/ru/serega6531/packmate/model/enums/CaptureMode.java

@@ -0,0 +1,7 @@
+package ru.serega6531.packmate.model.enums;
+
+public enum CaptureMode {
+
+    LIVE, FILE
+
+}

src/main/java/ru/serega6531/packmate/model/enums/SubscriptionMessageType.java

@@ -1,5 +1,10 @@
 package ru.serega6531.packmate.model.enums;
 
 public enum SubscriptionMessageType {
-    SAVE_SERVICE, SAVE_PATTERN, DELETE_SERVICE, ENABLE_PATTERN, DISABLE_PATTERN, NEW_STREAM, COUNTERS_UPDATE
+    SAVE_SERVICE, SAVE_PATTERN,
+    DELETE_SERVICE, DELETE_PATTERN,
+    NEW_STREAM,
+    COUNTERS_UPDATE,
+    ENABLE_PATTERN, DISABLE_PATTERN,
+    PCAP_STARTED, PCAP_STOPPED
 }

src/main/java/ru/serega6531/packmate/pcap/AbstractPcapWorker.java

@@ -1,44 +1,43 @@
-package ru.serega6531.packmate;
+package ru.serega6531.packmate.pcap;
 
 import com.google.common.collect.*;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.concurrent.BasicThreadFactory;
 import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.pcap4j.core.*;
+import org.pcap4j.core.PacketListener;
+import org.pcap4j.core.PcapHandle;
 import org.pcap4j.packet.IpV4Packet;
 import org.pcap4j.packet.Packet;
 import org.pcap4j.packet.TcpPacket;
 import org.pcap4j.packet.UdpPacket;
-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.enums.Protocol;
 import ru.serega6531.packmate.model.pojo.UnfinishedStream;
 import ru.serega6531.packmate.service.ServicesService;
 import ru.serega6531.packmate.service.StreamService;
 
-import javax.annotation.PreDestroy;
 import java.net.Inet4Address;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.util.ArrayList;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
 
-@Component
 @Slf4j
-public class PcapWorker implements PacketListener {
+public abstract class AbstractPcapWorker implements PcapWorker, PacketListener {
 
     private final ServicesService servicesService;
     private final StreamService streamService;
 
-    private final PcapNetworkInterface device;
+    protected PcapHandle pcap = null;
-    private PcapHandle pcap = null;
+    protected final ExecutorService loopExecutorService;
-    private final ExecutorService listenerExecutorService;
+
+    // во время работы должен быть != null
+    protected ExecutorService processorExecutorService;
 
     private final InetAddress localIp;
 
@@ -51,11 +50,9 @@ public class PcapWorker implements PacketListener {
     private final SetMultimap<UnfinishedStream, ImmutablePair<Inet4Address, Integer>> fins = HashMultimap.create();
     private final SetMultimap<UnfinishedStream, ImmutablePair<Inet4Address, Integer>> acks = HashMultimap.create();
 
-    @Autowired
+    public AbstractPcapWorker(ServicesService servicesService,
-    public PcapWorker(ServicesService servicesService,
+                              StreamService streamService,
-                      StreamService streamService,
+                              String localIpString) throws UnknownHostException {
-                      @Value("${interface-name}") String interfaceName,
-                      @Value("${local-ip}") String localIpString) throws PcapNativeException, UnknownHostException {
         this.servicesService = servicesService;
         this.streamService = streamService;
 
@@ -65,39 +62,8 @@ public class PcapWorker implements PacketListener {
         }
 
         BasicThreadFactory factory = new BasicThreadFactory.Builder()
-                .namingPattern("pcap-worker-listener").build();
+                .namingPattern("pcap-loop").build();
-        listenerExecutorService = Executors.newSingleThreadExecutor(factory);
+        loopExecutorService = Executors.newSingleThreadExecutor(factory);
-        device = Pcaps.getDevByName(interfaceName);
-    }
-
-    void start() throws PcapNativeException {
-        log.info("Using interface " + device.getName());
-        pcap = device.openLive(65536, PcapNetworkInterface.PromiscuousMode.PROMISCUOUS, 100);
-
-        BasicThreadFactory factory = new BasicThreadFactory.Builder()
-                .namingPattern("pcap-worker-loop").build();
-        ExecutorService loopExecutorService = Executors.newSingleThreadExecutor(factory);
-        try {
-            log.info("Intercept started");
-            pcap.loop(-1, this, loopExecutorService);
-        } catch (InterruptedException ignored) {
-            Thread.currentThread().interrupt();
-            // выходим
-        } catch (Exception e) {
-            log.error("Error while capturing packet", e);
-            stop();
-        }
-    }
-
-    @PreDestroy
-    @SneakyThrows
-    private void stop() {
-        if (pcap != null && pcap.isOpen()) {
-            pcap.breakLoop();
-            pcap.close();
-        }
-
-        log.info("Intercept stopped");
     }
 
     public void gotPacket(Packet rawPacket) {
@@ -105,14 +71,16 @@ public class PcapWorker implements PacketListener {
             return;
         }
 
+        final long time = pcap.getTimestamp().getTime();
+
         if (rawPacket.contains(TcpPacket.class)) {
-            gotTcpPacket(rawPacket);
+            gotTcpPacket(rawPacket, time);
         } else if (rawPacket.contains(UdpPacket.class)) {
-            gotUdpPacket(rawPacket);
+            gotUdpPacket(rawPacket, time);
         }
     }
 
-    private void gotTcpPacket(Packet rawPacket) {
+    private void gotTcpPacket(Packet rawPacket, long time) {
         final IpV4Packet.IpV4Header ipHeader = rawPacket.get(IpV4Packet.class).getHeader();
         Inet4Address sourceIp = ipHeader.getSrcAddr();
         Inet4Address destIp = ipHeader.getDstAddr();
@@ -134,9 +102,7 @@ public class PcapWorker implements PacketListener {
                 servicesService.findService(sourceIp, sourcePort, destIp, destPort);
 
         if (serviceOptional.isPresent()) {
-            final long time = System.currentTimeMillis();
+            processorExecutorService.execute(() -> {
-
-            listenerExecutorService.execute(() -> {
                 UnfinishedStream stream = addNewPacket(sourceIp, destIp, time, sourcePort, destPort, ttl, content, Protocol.TCP);
 
                 if (log.isDebugEnabled()) {
@@ -154,7 +120,7 @@ public class PcapWorker implements PacketListener {
         }
     }
 
-    private void gotUdpPacket(Packet rawPacket) {
+    private void gotUdpPacket(Packet rawPacket, long time) {
         final IpV4Packet.IpV4Header ipHeader = rawPacket.get(IpV4Packet.class).getHeader();
         Inet4Address sourceIp = ipHeader.getSrcAddr();
         Inet4Address destIp = ipHeader.getDstAddr();
@@ -173,9 +139,7 @@ public class PcapWorker implements PacketListener {
                 servicesService.findService(sourceIp, sourcePort, destIp, destPort);
 
         if (serviceOptional.isPresent()) {
-            final long time = System.currentTimeMillis();
+            processorExecutorService.execute(() -> {
-
-            listenerExecutorService.execute(() -> {
                 UnfinishedStream stream = addNewPacket(sourceIp, destIp, time, sourcePort, destPort, ttl, content, Protocol.UDP);
 
                 if (log.isDebugEnabled()) {
@@ -240,9 +204,25 @@ public class PcapWorker implements PacketListener {
         }
     }
 
+    @Override
     @SneakyThrows
-    int closeTimeoutStreams(Protocol protocol, long timeoutMillis) {
+    public void closeAllStreams(Protocol protocol) {
-        return listenerExecutorService.submit(() -> {
+        final var streams = (protocol == Protocol.TCP) ? this.unfinishedTcpStreams : this.unfinishedUdpStreams;
+
+        Multimaps.asMap(streams).forEach((key, value) ->
+                streamService.saveNewStream(key, new ArrayList<>(value)));
+
+        streams.clear();
+        if (protocol == Protocol.TCP) {
+            fins.clear();
+            acks.clear();
+        }
+    }
+
+    @Override
+    @SneakyThrows
+    public int closeTimeoutStreams(Protocol protocol, long timeoutMillis) {
+        return processorExecutorService.submit(() -> {
             int streamsClosed = 0;
 
             final long time = System.currentTimeMillis();

src/main/java/ru/serega6531/packmate/pcap/FilePcapWorker.java

@@ -0,0 +1,80 @@
+package ru.serega6531.packmate.pcap;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.tomcat.util.threads.InlineExecutorService;
+import org.pcap4j.core.PcapNativeException;
+import org.pcap4j.core.Pcaps;
+import org.pcap4j.packet.Packet;
+import ru.serega6531.packmate.model.enums.Protocol;
+import ru.serega6531.packmate.model.enums.SubscriptionMessageType;
+import ru.serega6531.packmate.model.pojo.SubscriptionMessage;
+import ru.serega6531.packmate.service.ServicesService;
+import ru.serega6531.packmate.service.StreamService;
+import ru.serega6531.packmate.service.SubscriptionService;
+
+import java.io.EOFException;
+import java.io.File;
+import java.net.UnknownHostException;
+
+@Slf4j
+public class FilePcapWorker extends AbstractPcapWorker {
+
+    private final SubscriptionService subscriptionService;
+    private final File file;
+
+    public FilePcapWorker(ServicesService servicesService,
+                          StreamService streamService,
+                          SubscriptionService subscriptionService,
+                          String localIpString,
+                          String filename) throws UnknownHostException {
+        super(servicesService, streamService, localIpString);
+        this.subscriptionService = subscriptionService;
+
+        file = new File(filename);
+        if(!file.exists()) {
+            throw new IllegalArgumentException("File " + file.getAbsolutePath() + " does not exist");
+        }
+
+        processorExecutorService = new InlineExecutorService();
+    }
+
+    @SneakyThrows
+    @Override
+    public void start() {
+        log.info("Using file " + file.getAbsolutePath());
+        pcap = Pcaps.openOffline(file.getAbsolutePath());
+        loopExecutorService.execute(this::runScan);
+    }
+
+    @SneakyThrows
+    private void runScan() {
+        while (pcap.isOpen()) {
+            try {
+                final Packet packet = pcap.getNextPacketEx();
+                gotPacket(packet);
+            } catch (PcapNativeException e) {
+                log.error("Pcap read", e);
+                Thread.sleep(100);
+            } catch (EOFException e) {
+                stop();
+
+                log.info("All packets processed");
+                break;
+            }
+        }
+    }
+
+    @SneakyThrows
+    public void stop() {
+        if (pcap != null && pcap.isOpen()) {
+            pcap.close();
+            log.info("Pcap closed");
+        }
+
+        closeAllStreams(Protocol.TCP);
+        closeAllStreams(Protocol.UDP);
+
+        subscriptionService.broadcast(new SubscriptionMessage(SubscriptionMessageType.PCAP_STOPPED, null));
+    }
+}

src/main/java/ru/serega6531/packmate/pcap/LivePcapWorker.java

@@ -0,0 +1,62 @@
+package ru.serega6531.packmate.pcap;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.concurrent.BasicThreadFactory;
+import org.pcap4j.core.PcapNativeException;
+import org.pcap4j.core.PcapNetworkInterface;
+import org.pcap4j.core.Pcaps;
+import ru.serega6531.packmate.service.ServicesService;
+import ru.serega6531.packmate.service.StreamService;
+
+import java.net.UnknownHostException;
+import java.util.concurrent.Executors;
+
+@Slf4j
+public class LivePcapWorker extends AbstractPcapWorker {
+
+    private final PcapNetworkInterface device;
+
+    public LivePcapWorker(ServicesService servicesService,
+                          StreamService streamService,
+                          String localIpString,
+                          String interfaceName) throws PcapNativeException, UnknownHostException {
+        super(servicesService, streamService, localIpString);
+        device = Pcaps.getDevByName(interfaceName);
+
+        if(device == null) {
+            throw new IllegalArgumentException("Device " + interfaceName + " does not exist");
+        }
+
+        BasicThreadFactory factory = new BasicThreadFactory.Builder()
+                .namingPattern("pcap-processor").build();
+        processorExecutorService = Executors.newSingleThreadExecutor(factory);
+    }
+
+    public void start() throws PcapNativeException {
+        log.info("Using interface " + device.getName());
+        pcap = device.openLive(65536, PcapNetworkInterface.PromiscuousMode.PROMISCUOUS, 100);
+
+        try {
+            log.info("Intercept started");
+            pcap.loop(-1, this, loopExecutorService);
+        } catch (InterruptedException ignored) {
+            Thread.currentThread().interrupt();
+            // выходим
+        } catch (Exception e) {
+            log.error("Error while capturing packet", e);
+            stop();
+        }
+    }
+
+    @SneakyThrows
+    public void stop() {
+        if (pcap != null && pcap.isOpen()) {
+            pcap.breakLoop();
+            pcap.close();
+        }
+
+        log.info("Intercept stopped");
+    }
+
+}

src/main/java/ru/serega6531/packmate/pcap/PcapWorker.java

@@ -0,0 +1,21 @@
+package ru.serega6531.packmate.pcap;
+
+import org.pcap4j.core.PcapNativeException;
+import ru.serega6531.packmate.model.enums.Protocol;
+
+public interface PcapWorker {
+
+    void start() throws PcapNativeException;
+    void stop();
+
+    /**
+     * Выполняется в вызывающем потоке
+     */
+    void closeAllStreams(Protocol protocol);
+
+    /**
+     * Выполняется в потоке обработчика
+     */
+    int closeTimeoutStreams(Protocol protocol, long timeoutMillis);
+
+}

src/main/java/ru/serega6531/packmate/service/PatternService.java

@@ -79,7 +79,7 @@ public class PatternService {
 
         final Pattern saved = repository.save(pattern);
         patterns.put(saved.getId(), saved);
-        log.info("Added new pattern {} with value {}", pattern.getName(), pattern.getValue());
+        log.info("Added new pattern '{}' with value '{}'", pattern.getName(), pattern.getValue());
         subscriptionService.broadcast(new SubscriptionMessage(SubscriptionMessageType.SAVE_PATTERN, saved));
         return saved;
     }

src/main/java/ru/serega6531/packmate/service/PcapService.java

@@ -0,0 +1,34 @@
+package ru.serega6531.packmate.service;
+
+import lombok.Getter;
+import org.pcap4j.core.PcapNativeException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import ru.serega6531.packmate.model.enums.SubscriptionMessageType;
+import ru.serega6531.packmate.model.pojo.SubscriptionMessage;
+import ru.serega6531.packmate.pcap.PcapWorker;
+
+@Service
+public class PcapService {
+
+    @Getter
+    private boolean started = false;
+
+    private final SubscriptionService subscriptionService;
+    private final PcapWorker worker;
+
+    @Autowired
+    public PcapService(SubscriptionService subscriptionService, PcapWorker worker) {
+        this.subscriptionService = subscriptionService;
+        this.worker = worker;
+    }
+
+    public synchronized void start() throws PcapNativeException {
+        if(!started) {
+            started = true;
+            subscriptionService.broadcast(new SubscriptionMessage(SubscriptionMessageType.PCAP_STARTED, null));
+            worker.start();
+        }
+    }
+
+}

src/main/java/ru/serega6531/packmate/service/ServicesService.java

@@ -66,7 +66,7 @@ public class ServicesService {
     }
 
     public CtfService save(CtfService service) {
-        log.info("Added or edited service {} at port {}", service.getName(), service.getPort());
+        log.info("Added or edited service '{}' at port {}", service.getName(), service.getPort());
         final CtfService saved = repository.save(service);
         services.put(saved.getPort(), saved);
         subscriptionService.broadcast(new SubscriptionMessage(SubscriptionMessageType.SAVE_SERVICE, saved));

src/main/java/ru/serega6531/packmate/service/StreamOptimizer.java

@@ -122,8 +122,12 @@ public class StreamOptimizer {
                 }
 
                 if (httpStarted) {
-                    content = URLDecoder.decode(content, StandardCharsets.UTF_8.toString());
+                    try {
-                    packet.setContent(content.getBytes());
+                        content = URLDecoder.decode(content, StandardCharsets.UTF_8.toString());
+                        packet.setContent(content.getBytes());
+                    } catch (IllegalArgumentException e) {
+                        log.warn("urldecode", e);
+                    }
                 }
             } else {
                 httpStarted = false;

src/main/java/ru/serega6531/packmate/service/SubscriptionService.java

@@ -2,6 +2,7 @@ package ru.serega6531.packmate.service;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -39,10 +40,15 @@ public class SubscriptionService {
         log.info("User unsubscribed: {}", Objects.requireNonNull(session.getRemoteAddress()).getHostName());
     }
 
-    void broadcast(SubscriptionMessage message) {
+    /**
+     * Вызов потокобезопасный
+     */
+    @SneakyThrows
+    public void broadcast(SubscriptionMessage message) {
+        final TextMessage messageJson = objectToTextMessage(message);
         subscribers.forEach(s -> {
             try {
-                s.sendMessage(objectToTextMessage(message));
+                s.sendMessage(messageJson);
             } catch (IOException | SockJsTransportFailureException e) {
                 log.warn("WS", e);
             }

src/main/resources/application.yml

@@ -15,7 +15,9 @@ spring:
 
 
 enable-capture: true
+capture-mode: LIVE  # LIVE, FILE
 interface-name: enp0s31f6
+pcap-file: file.pcap
 local-ip: "192.168.0.125"
 account-login: BinaryBears
 account-password: 123456