SOCKS work #1048

src/main/java/app/nzyme/core/database/DatabaseImpl.java

@@ -16,6 +16,7 @@
 import app.nzyme.core.dot11.db.monitoring.*;
 import app.nzyme.core.dot11.tracks.db.TrackDetectorConfigMapper;
 import app.nzyme.core.ethernet.dns.db.*;
+import app.nzyme.core.ethernet.socks.db.SocksTunnelEntryMapper;
 import app.nzyme.core.ethernet.tcp.db.TcpSessionEntryMapper;
 import app.nzyme.core.events.db.EventActionEntryMapper;
 import app.nzyme.core.events.db.EventEntryMapper;
@@ -142,7 +143,8 @@ public void initializeAndMigrate() throws LiquibaseException {
                 .registerRowMapper(new Dot11FrequencyAndChannelWidthEntryMapper())
                 .registerRowMapper(new TcpSessionEntryMapper())
                 .registerRowMapper(new DNSEntropyLogEntryMapper())
-                .registerRowMapper(new DNSLogEntryMapper());
+                .registerRowMapper(new DNSLogEntryMapper())
+                .registerRowMapper(new SocksTunnelEntryMapper());
 
         if (configuration.slowQueryLogThreshold().isPresent()) {
             LOG.info("Slow query log enabled with threshold <{}ms>.", configuration.slowQueryLogThreshold().get());

src/main/java/app/nzyme/core/ethernet/Ethernet.java

@@ -2,16 +2,19 @@
 
 import app.nzyme.core.NzymeNode;
 import app.nzyme.core.ethernet.dns.DNS;
+import app.nzyme.core.ethernet.socks.Socks;
 
 public class Ethernet {
 
     private final NzymeNode nzyme;
 
     private final DNS dns;
+    private final Socks socks;
 
     public Ethernet(NzymeNode nzyme) {
         this.nzyme = nzyme;
         this.dns = new DNS(this);
+        this.socks = new Socks(this);
     }
 
     public NzymeNode getNzyme() {
@@ -22,4 +25,8 @@ public DNS dns() {
         return dns;
     }
 
+    public Socks socks() {
+        return socks;
+    }
+
 }

src/main/java/app/nzyme/core/ethernet/socks/Socks.java

@@ -0,0 +1,58 @@
+package app.nzyme.core.ethernet.socks;
+
+import app.nzyme.core.NzymeNode;
+import app.nzyme.core.ethernet.Ethernet;
+import app.nzyme.core.ethernet.socks.db.SocksTunnelEntry;
+import app.nzyme.core.util.TimeRange;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+public class Socks {
+
+    private final NzymeNode nzyme;
+
+    public Socks(Ethernet ethernet) {
+        this.nzyme = ethernet.getNzyme();
+    }
+
+    public long countAllSocksTunnels(TimeRange timeRange, List<UUID> taps) {
+        if (taps.isEmpty()) {
+            return 0;
+        }
+
+        return nzyme.getDatabase().withHandle(handle ->
+                handle.createQuery("SELECT COUNT(*) FROM socks_tunnels " +
+                                "WHERE most_recent_segment_time >= :tr_from AND most_recent_segment_time <= :tr_to " +
+                                "AND tap_uuid IN (<taps>)")
+                        .bindList("taps", taps)
+                        .bind("tr_from", timeRange.from())
+                        .bind("tr_to", timeRange.to())
+                        .mapTo(Long.class)
+                        .one()
+        );
+    }
+
+    public List<SocksTunnelEntry> findAllSocksTunnels(TimeRange timeRange, int limit, int offset, List<UUID> taps) {
+        if (taps.isEmpty()) {
+            return Collections.emptyList();
+        }
+
+        return nzyme.getDatabase().withHandle(handle ->
+                handle.createQuery("SELECT * FROM socks_tunnels " +
+                                "WHERE most_recent_segment_time >= :tr_from AND most_recent_segment_time <= :tr_to " +
+                                "AND tap_uuid IN (<taps>) " +
+                                "ORDER BY most_recent_segment_time DESC " +
+                                "LIMIT :limit OFFSET :offset")
+                        .bindList("taps", taps)
+                        .bind("tr_from", timeRange.from())
+                        .bind("tr_to", timeRange.to())
+                        .bind("limit", limit)
+                        .bind("offset", offset)
+                        .mapTo(SocksTunnelEntry.class)
+                        .list()
+        );
+    }
+
+}

src/main/java/app/nzyme/core/ethernet/socks/db/SocksTunnelEntry.java

@@ -0,0 +1,101 @@
+package app.nzyme.core.ethernet.socks.db;
+
+import com.google.auto.value.AutoValue;
+import jakarta.annotation.Nullable;
+import org.joda.time.DateTime;
+
+import java.util.UUID;
+
+@AutoValue
+public abstract class SocksTunnelEntry {
+
+    public abstract long id();
+    public abstract UUID uuid();
+    public abstract UUID tapUuid();
+    public abstract String tcpSessionKey();
+    public abstract String socksType();
+    public abstract String authenticationStatus();
+    public abstract String handshakeStatus();
+    public abstract String connectionStatus();
+    @Nullable
+    public abstract String username();
+    public abstract int tunneledBytes();
+    @Nullable
+    public abstract String tunneledDestinationAddress();
+    @Nullable
+    public abstract String tunneledDestinationHost();
+    public abstract int tunneledDestinationPort();
+    public abstract DateTime establishedAt();
+    public abstract DateTime terminatedAt();
+    public abstract DateTime mostRecentSegmentTime();
+    public abstract DateTime updatedAt();
+    public abstract DateTime createdAt();
+
+    public static SocksTunnelEntry create(long id, UUID uuid, UUID tapUuid, String tcpSessionKey, String socksType, String authenticationStatus, String handshakeStatus, String connectionStatus, String username, int tunneledBytes, String tunneledDestinationAddress, String tunneledDestinationHost, int tunneledDestinationPort, DateTime establishedAt, DateTime terminatedAt, DateTime mostRecentSegmentTime, DateTime updatedAt, DateTime createdAt) {
+        return builder()
+                .id(id)
+                .uuid(uuid)
+                .tapUuid(tapUuid)
+                .tcpSessionKey(tcpSessionKey)
+                .socksType(socksType)
+                .authenticationStatus(authenticationStatus)
+                .handshakeStatus(handshakeStatus)
+                .connectionStatus(connectionStatus)
+                .username(username)
+                .tunneledBytes(tunneledBytes)
+                .tunneledDestinationAddress(tunneledDestinationAddress)
+                .tunneledDestinationHost(tunneledDestinationHost)
+                .tunneledDestinationPort(tunneledDestinationPort)
+                .establishedAt(establishedAt)
+                .terminatedAt(terminatedAt)
+                .mostRecentSegmentTime(mostRecentSegmentTime)
+                .updatedAt(updatedAt)
+                .createdAt(createdAt)
+                .build();
+    }
+
+    public static Builder builder() {
+        return new AutoValue_SocksTunnelEntry.Builder();
+    }
+
+    @AutoValue.Builder
+    public abstract static class Builder {
+        public abstract Builder id(long id);
+
+        public abstract Builder uuid(UUID uuid);
+
+        public abstract Builder tapUuid(UUID tapUuid);
+
+        public abstract Builder tcpSessionKey(String tcpSessionKey);
+
+        public abstract Builder socksType(String socksType);
+
+        public abstract Builder authenticationStatus(String authenticationStatus);
+
+        public abstract Builder handshakeStatus(String handshakeStatus);
+
+        public abstract Builder connectionStatus(String connectionStatus);
+
+        public abstract Builder username(String username);
+
+        public abstract Builder tunneledBytes(int tunneledBytes);
+
+        public abstract Builder tunneledDestinationAddress(String tunneledDestinationAddress);
+
+        public abstract Builder tunneledDestinationHost(String tunneledDestinationHost);
+
+        public abstract Builder tunneledDestinationPort(int tunneledDestinationPort);
+
+        public abstract Builder establishedAt(DateTime establishedAt);
+
+        public abstract Builder terminatedAt(DateTime terminatedAt);
+
+        public abstract Builder mostRecentSegmentTime(DateTime mostRecentSegmentTime);
+
+        public abstract Builder updatedAt(DateTime updatedAt);
+
+        public abstract Builder createdAt(DateTime createdAt);
+
+        public abstract SocksTunnelEntry build();
+    }
+}

src/main/java/app/nzyme/core/ethernet/socks/db/SocksTunnelEntryMapper.java

@@ -0,0 +1,37 @@
+package app.nzyme.core.ethernet.socks.db;
+
+import org.jdbi.v3.core.mapper.RowMapper;
+import org.jdbi.v3.core.statement.StatementContext;
+import org.joda.time.DateTime;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.UUID;
+
+public class SocksTunnelEntryMapper implements RowMapper<SocksTunnelEntry> {
+
+    @Override
+    public SocksTunnelEntry map(ResultSet rs, StatementContext ctx) throws SQLException {
+        return SocksTunnelEntry.create(
+                rs.getLong("id"),
+                UUID.fromString(rs.getString("uuid")),
+                UUID.fromString(rs.getString("tap_uuid")),
+                rs.getString("tcp_session_key"),
+                rs.getString("socks_type"),
+                rs.getString("authentication_status"),
+                rs.getString("handshake_status"),
+                rs.getString("connection_status"),
+                rs.getString("username"),
+                rs.getInt("tunneled_bytes"),
+                rs.getString("tunneled_destination_address"),
+                rs.getString("tunneled_destination_host"),
+                rs.getInt("tunneled_destination_port"),
+                new DateTime(rs.getTimestamp("established_at")),
+                new DateTime(rs.getTimestamp("terminated_at")),
+                new DateTime(rs.getTimestamp("most_recent_segment_time")),
+                new DateTime(rs.getTimestamp("updated_at")),
+                new DateTime(rs.getTimestamp("created_at"))
+        );
+    }
+
+}

src/main/java/app/nzyme/core/rest/resources/taps/reports/tables/socks/SocksTunnelReport.java

@@ -30,6 +30,7 @@ public abstract class SocksTunnelReport {
     public abstract DateTime establishedAt();
     @Nullable
     public abstract DateTime terminatedAt();
+    public abstract DateTime mostRecentSegmentTime();
 
     @JsonCreator
     public static SocksTunnelReport create(@JsonProperty("socks_type") String socksType,
@@ -48,7 +49,8 @@ public static SocksTunnelReport create(@JsonProperty("socks_type") String socksT
                                            @JsonProperty("destination_address") String destinationAddress,
                                            @JsonProperty("destination_port") int destinationPort,
                                            @JsonProperty("established_at") DateTime establishedAt,
-                                           @JsonProperty("terminated_at") DateTime terminatedAt) {
+                                           @JsonProperty("terminated_at") DateTime terminatedAt,
+                                           @JsonProperty("most_recent_segment_time") DateTime mostRecentSegmentTime) {
         return builder()
                 .socksType(socksType)
                 .authenticationStatus(authenticationStatus)
@@ -67,6 +69,7 @@ public static SocksTunnelReport create(@JsonProperty("socks_type") String socksT
                 .destinationPort(destinationPort)
                 .establishedAt(establishedAt)
                 .terminatedAt(terminatedAt)
+                .mostRecentSegmentTime(mostRecentSegmentTime)
                 .build();
     }
 
@@ -110,6 +113,8 @@ public abstract static class Builder {
 
         public abstract Builder terminatedAt(DateTime terminatedAt);
 
+        public abstract Builder mostRecentSegmentTime(DateTime mostRecentSegmentTime);
+
         public abstract SocksTunnelReport build();
     }
 }

src/main/java/app/nzyme/core/tables/socks/SocksTable.java

@@ -44,21 +44,23 @@ public void handleReport(UUID tapUuid, DateTime timestamp, SocksTunnelsReport re
 
     private void writeTunnels(Handle handle, UUID tapUuid, DateTime timestamp, List<SocksTunnelReport> tunnels) {
         PreparedBatch insertBatch = handle.prepareBatch("INSERT INTO socks_tunnels(uuid, tap_uuid, " +
-                    "tcp_session_key, socks_type, authentication_status, handshake_status, connection_status, " +
-                    "username, tunneled_bytes, tunneled_destination_address, tunneled_destination_host, " +
-                    "tunneled_destination_port, established_at, terminated_at, updated_at, created_at) " +
-                    "VALUES(:uuid, :tap_uuid, :tcp_session_key, :socks_type, :authentication_status, " +
-                    ":handshake_status, :connection_status, :username, :tunneled_bytes, " +
-                    ":tunneled_destination_address, :tunneled_destination_host, :tunneled_destination_port, " +
-                    ":established_at, :terminated_at, NOW(), NOW())");
+                "tcp_session_key, socks_type, authentication_status, handshake_status, connection_status, " +
+                "username, tunneled_bytes, tunneled_destination_address, tunneled_destination_host, " +
+                "tunneled_destination_port, established_at, terminated_at, most_recent_segment_time, " +
+                "updated_at, created_at) VALUES(:uuid, :tap_uuid, :tcp_session_key, :socks_type, " +
+                ":authentication_status, :handshake_status, :connection_status, :username, :tunneled_bytes, " +
+                ":tunneled_destination_address, :tunneled_destination_host, :tunneled_destination_port, " +
+                ":established_at, :terminated_at, :most_recent_segment_time, NOW(), NOW())");
 
         PreparedBatch updateBatch = handle.prepareBatch("UPDATE socks_tunnels SET " +
                 "authentication_status = :authentication_status, handshake_status = :handshake_status, " +
                 "connection_status = :connection_status, tunneled_bytes = :tunneled_bytes, " +
-                "terminated_at = :terminated_at, updated_at = NOW() WHERE id = :id");
+                "terminated_at = :terminated_at, most_recent_segment_time = :most_recent_segment_time, " +
+                "updated_at = NOW() WHERE id = :id");
 
         for (SocksTunnelReport tunnel : tunnels) {
             String tcpSessionKey = Tools.buildTcpSessionKey(
+                    tunnel.establishedAt(),
                     tunnel.sourceAddress(),
                     tunnel.destinationAddress(),
                     tunnel.sourcePort(),
@@ -71,6 +73,7 @@ private void writeTunnels(Handle handle, UUID tapUuid, DateTime timestamp, List<
                     .bind("established_at", tunnel.establishedAt())
                     .bind("tap_uuid", tapUuid)
                     .bind("connection_status", "Active")
+                    .bind("most_recent_segment_time", tunnel.mostRecentSegmentTime())
                     .mapTo(Long.class)
                     .findOne();
 
@@ -90,6 +93,7 @@ private void writeTunnels(Handle handle, UUID tapUuid, DateTime timestamp, List<
                         .bind("tunneled_destination_port", tunnel.tunneledDestinationPort())
                         .bind("established_at", tunnel.establishedAt())
                         .bind("terminated_at", tunnel.terminatedAt())
+                        .bind("most_recent_segment_time", tunnel.mostRecentSegmentTime())
                         .add();
             } else {
                 // Update existing open tunnel.
@@ -100,6 +104,7 @@ private void writeTunnels(Handle handle, UUID tapUuid, DateTime timestamp, List<
                         .bind("connection_status", tunnel.connectionStatus())
                         .bind("tunneled_bytes", tunnel.tunneledBytes())
                         .bind("terminated_at", tunnel.terminatedAt())
+                        .bind("most_recent_segment_time", tunnel.mostRecentSegmentTime())
                         .add();
             }
         }

src/main/java/app/nzyme/core/tables/tcp/TCPTable.java

@@ -76,6 +76,7 @@ public void handleReport(UUID tapUuid, DateTime timestamp, TcpSessionsReport rep
     private void writeSession(UUID tapUuid, DateTime timestamp, TcpSessionReport session) {
         try {
             String sessionKey = Tools.buildTcpSessionKey(
+                    session.startTime(),
                     session.sourceAddress(),
                     session.destinationAddress(),
                     session.sourcePort(),

src/main/java/app/nzyme/core/util/Tools.java

@@ -96,9 +96,17 @@ public static float round(float f, int decimalPlaces) {
         return bd.floatValue();
     }
 
-    public static String buildTcpSessionKey(String sourceAddress, String destinationAddress, int sourcePort, int destinationPort) {
+    public static String buildTcpSessionKey(DateTime sessionEstablishedAt,
+                                            String sourceAddress,
+                                            String destinationAddress,
+                                            int sourcePort,
+                                            int destinationPort) {
         return Hashing.sha256()
-                .hashString(sourceAddress + destinationAddress + sourcePort + destinationPort, Charsets.UTF_8)
+                .hashString(sessionEstablishedAt.getMillis()
+                        + sourceAddress
+                        + destinationAddress
+                        + sourcePort
+                        + destinationPort, Charsets.UTF_8)
                 .toString();
     }
 

src/main/resources/db/migrations.xml

@@ -4989,4 +4989,12 @@
             </column>
         </createTable>
     </changeSet>
+
+    <changeSet id="add_most_recent_segment_time_to_socks_tunnels" author="lennart koopmann">
+        <addColumn tableName="socks_tunnels">
+            <column name="most_recent_segment_time" type="timestamp with time zone">
+                <constraints nullable="false" />
+            </column>
+        </addColumn>
+    </changeSet>
 </databaseChangeLog>