Fix port aggreation

go/cron/aggregator.go

@@ -90,13 +90,15 @@ loop:
 					continue
 				}
 
+				ports := map[AggregatorSession]struct{}{}
 				for _, e := range events {
 					startTime = min(startTime, e.StartTime)
 					endTime = max(endTime, e.EndTime)
+					ports[e.SessionKey()] = struct{}{}
 					packetCount += cmp.Or(e.PacketCount, 1)
 				}
 
-				networkEvent.Count = int(packetCount)
+				networkEvent.Count = len(ports)
 				if startTime != 0 {
 					networkEvent.StartTime = uint64(startTime)
 				}

go/cron/aggregator_test.go

@@ -11,6 +11,15 @@ func TestAggregator(t *testing.T) {
 	events := []*PfFlows{
 		{
 			Flows: &[]PfFlow{
+				{
+					SrcIp:       netip.AddrFrom4([4]byte{1, 1, 1, 2}),
+					DstIp:       netip.AddrFrom4([4]byte{1, 1, 1, 1}),
+					SrcPort:     80,
+					DstPort:     1025,
+					Proto:       6,
+					BiFlow:      2,
+					PacketCount: 1,
+				},
 				{
 					SrcIp:       netip.AddrFrom4([4]byte{1, 1, 1, 1}),
 					DstIp:       netip.AddrFrom4([4]byte{1, 1, 1, 2}),
@@ -23,25 +32,16 @@ func TestAggregator(t *testing.T) {
 				{
 					SrcIp:       netip.AddrFrom4([4]byte{1, 1, 1, 1}),
 					DstIp:       netip.AddrFrom4([4]byte{1, 1, 1, 2}),
-					SrcPort:     1025,
+					SrcPort:     1024,
 					DstPort:     80,
 					Proto:       6,
 					BiFlow:      1,
 					PacketCount: 1,
 				},
-				{
-					SrcIp:       netip.AddrFrom4([4]byte{1, 1, 1, 2}),
-					DstIp:       netip.AddrFrom4([4]byte{1, 1, 1, 1}),
-					SrcPort:     80,
-					DstPort:     1025,
-					Proto:       6,
-					BiFlow:      2,
-					PacketCount: 1,
-				},
 				{
 					SrcIp:       netip.AddrFrom4([4]byte{1, 1, 1, 1}),
 					DstIp:       netip.AddrFrom4([4]byte{1, 1, 1, 2}),
-					SrcPort:     1024,
+					SrcPort:     1025,
 					DstPort:     80,
 					Proto:       6,
 					BiFlow:      1,
@@ -65,10 +65,14 @@ func TestAggregator(t *testing.T) {
 		t.Fatalf("Not aggreated to a single network event")
 	}
 
-	if ne[0].Count != 4 {
+	if ne[0].Count != 2 {
 		t.Fatalf("Not aggreated properly")
 	}
 
+	if ne[0].DestPort != 80 {
+		t.Fatalf("Not aggreated DestPort")
+	}
+
 	events = []*PfFlows{
 		{
 			Flows: &[]PfFlow{
@@ -105,8 +109,12 @@ func TestAggregator(t *testing.T) {
 		t.Fatalf("Not aggreated to a single network event")
 	}
 
-	if ne[0].Count != 3 {
+	if ne[0].Count != 2 {
 		t.Fatalf("Not aggreated properly")
 	}
 
+	if ne[0].DestPort != 80 {
+		t.Fatalf("Not aggreated DestPort")
+	}
+
 }

go/cron/pfflow.go

@@ -1,7 +1,6 @@
 package maint
 
 import (
-	"cmp"
 	"net/netip"
 	"time"
 )
@@ -102,6 +101,7 @@ func (f *PfFlow) SessionKey() AggregatorSession {
 	if f.BiFlow == 2 {
 		return AggregatorSession{Port: f.DstPort}
 	}
+
 	return AggregatorSession{Port: f.SrcPort}
 }
 
@@ -116,6 +116,14 @@ func (f *PfFlow) NetworkEventDirection() NetworkEventDirection {
 	}
 }
 
+func (f *PfFlow) CalculatedDstPort() int {
+	if f.BiFlow == 2 {
+		return int(f.SrcPort)
+	}
+
+	return int(f.DstPort)
+}
+
 func (f *PfFlow) ToNetworkEvent() *NetworkEvent {
 	if f.DstMac == "00:00:00:00:00:00" && f.SrcMac == "00:00:00:00:00:00" {
 		return nil
@@ -130,11 +138,11 @@ func (f *PfFlow) ToNetworkEvent() *NetworkEvent {
 		EventType:           NetworkEventTypeSuccessful,
 		SourceIp:            f.SrcIp,
 		DestIp:              f.DstIp,
-		DestPort:            int(f.DstPort),
+		DestPort:            f.CalculatedDstPort(),
 		IpProtocol:          ipProto,
 		IpVersion:           IpVersionIpv4,
 		EnforcementState:    EnforcementStateEnforcing,
-		Count:               cmp.Or(int(f.PacketCount), 1),
+		Count:               1,
 		StartTime:           uint64(time.Now().Unix()),
 		Direction:           f.NetworkEventDirection(),
 		DestInventoryitem:   f.DestInventoryitem(),