package integration_test import ( "context" "net" "testing" "time" "github.com/activecm/rita/v5/analysis" "github.com/activecm/rita/v5/cmd" "github.com/activecm/rita/v5/config" "github.com/activecm/rita/v5/database" "github.com/activecm/rita/v5/progressbar" "github.com/activecm/rita/v5/util" "github.com/ClickHouse/clickhouse-go/v2" "github.com/charmbracelet/bubbles/progress" "github.com/spf13/afero" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" "golang.org/x/sync/errgroup" ) type FilterTestSuite struct { suite.Suite cfg *config.Config } func TestFilters(t *testing.T) { suite.Run(t, new(FilterTestSuite)) } // Reset config after each test since these tests load the config from a file func (it *FilterTestSuite) SetupSuite() { t := it.T() afs := afero.NewOsFs() cfg, err := config.ReadFileConfig(afs, ConfigPath) require.NoError(t, err) it.cfg = cfg } func (it *FilterTestSuite) SetupTest() { t := it.T() err := it.cfg.Reset() require.NoError(t, err) } func (it *FilterTestSuite) TearDownSuite() { t := it.T() err := it.cfg.Reset() require.NoError(t, err) } func (it *FilterTestSuite) TestNeverIncludeSubnets() { t := it.T() // set up file system interface afs := afero.NewMemMapFs() afs2 := afero.NewOsFs() err := afero.WriteFile(afs, "testsuite_config.hjson", []byte(` { filtering: { filter_external_to_internal: true, internal_subnets: ["10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "fd00::/8"], never_included_subnets: ["10.55.100.0/24"], }, threat_intel: { online_feeds: ["https://feodotracker.abuse.ch/downloads/ipblocklist.txt"], custom_feeds_directory: "./deployment/threat_intel_feeds" }, http_extensions_file_path: "../deployment/http_extensions_list.csv" } `), 0755) require.NoError(t, err) cfg, err := config.ReadFileConfig(afs, "testsuite_config.hjson") require.NoError(t, err) cfg.Env.DBConnection = dockerInfo.clickhouseConnection it.cfg = cfg require.Contains(t, cfg.Filtering.NeverIncludedSubnets, util.NewSubnet(&net.IPNet{IP: net.IP{10, 55, 100, 0}, Mask: net.IPMask{255, 255, 255, 0}})) // // import data _, err = cmd.RunImportCmd(time.Now(), cfg, afs2, "../test_data/valid_tsv", "never_include_subnet", false, true) require.NoError(t, err) // connect to database db, err := database.ConnectToDB(context.Background(), "never_include_subnet", cfg, nil) require.NoError(t, err) var count uint64 // verify that not all connections in 10.0.0.0/8 were filtered err = db.Conn.QueryRow(db.GetContext(), ` SELECT count(DISTINCT hash) FROM conn WHERE isIPAddressInRange(IPv6NumToString(src), '::ffff:10.0.0.0/104') OR isIPAddressInRange(IPv6NumToString(dst), '::ffff:10.0.0.0/104') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 14452-12584, count, "conn table should contain 1868 entries in 10.0.0.0/8, got: %d", count) // 5531 in 10.55.100.0/24 err = db.Conn.QueryRow(db.GetContext(), ` SELECT count(DISTINCT hash) FROM ssl WHERE isIPAddressInRange(IPv6NumToString(src), '::ffff:10.0.0.0/104') OR isIPAddressInRange(IPv6NumToString(dst), '::ffff:10.0.0.0/104') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 5615-(5531), count, "ssl table should contain 63 entries in 10.0.0.0/8, got: %d", count) err = db.Conn.QueryRow(db.GetContext(), ` SELECT count(DISTINCT hash) FROM http WHERE isIPAddressInRange(IPv6NumToString(src), '::ffff:10.0.0.0/104') OR isIPAddressInRange(IPv6NumToString(dst), '::ffff:10.0.0.0/104') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 2004-(1982), count, "http table should contain 22 entries in 10.0.0.0/8, got: %d", count) // verify that all connections in 10.55.100.0/24 were filtered err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM conn WHERE isIPAddressInRange(IPv6NumToString(src), '::ffff:10.55.100.0/120') OR isIPAddressInRange(IPv6NumToString(dst), '::ffff:10.55.100.0/120') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 0, count, "conn table shouldn't contain any entries in 10.55.100.0/24") err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM http WHERE isIPAddressInRange(IPv6NumToString(src), '::ffff:10.55.100.0/120') OR isIPAddressInRange(IPv6NumToString(dst), '::ffff:10.55.100.0/120') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 0, count, "http table shouldn't contain any entries in 10.55.100.0/24") err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM ssl WHERE isIPAddressInRange(IPv6NumToString(src), '::ffff:10.55.100.0/120') OR isIPAddressInRange(IPv6NumToString(dst), '::ffff:10.55.100.0/120') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 0, count, "ssl table shouldn't contain any entries in 10.55.100.0/24") err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM dns WHERE isIPAddressInRange(IPv6NumToString(src), '::ffff:10.55.100.0/120') OR isIPAddressInRange(IPv6NumToString(dst), '::ffff:10.55.100.0/120') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 0, count, "dns table shouldn't contain any entries in 10.55.100.0/24") } func (it *FilterTestSuite) TestNeverIncludeDomains() { t := it.T() // set up file system interface afs := afero.NewMemMapFs() afs2 := afero.NewOsFs() err := afero.WriteFile(afs, "testsuite_config2.hjson", []byte(` { filtering: { filter_external_to_internal: true, internal_subnets: ["10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "fd00::/8"], never_included_domains: ["*.microsoft.com", "businessinsider.com"] }, threat_intel: { online_feeds: ["https://feodotracker.abuse.ch/downloads/ipblocklist.txt"], custom_feeds_directory: "./deployment/threat_intel_feeds" }, http_extensions_file_path: "../deployment/http_extensions_list.csv" } `), 0755) require.NoError(t, err) cfg, err := config.ReadFileConfig(afs, "testsuite_config2.hjson") require.NoError(t, err) cfg.Env.DBConnection = dockerInfo.clickhouseConnection it.cfg = cfg require.Contains(t, cfg.Filtering.NeverIncludedDomains, "*.microsoft.com") require.Contains(t, cfg.Filtering.NeverIncludedDomains, "businessinsider.com") // // import data _, err = cmd.RunImportCmd(time.Now(), cfg, afs2, "../test_data/valid_tsv", "never_include_domain", false, true) require.NoError(t, err) // connect to database db, err := database.ConnectToDB(context.Background(), "never_include_domain", cfg, nil) require.NoError(t, err) var count uint64 // verify that all connections w/ fqdns ending in .microsoft.com are filtered err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM http WHERE cutToFirstSignificantSubdomain(host) = 'microsoft.com' `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 0, count, "http table shouldn't contain any entries with host ending in .microsoft.com") err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM ssl WHERE cutToFirstSignificantSubdomain(server_name) = 'microsoft.com' `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 0, count, "ssl table shouldn't contain any entries with server_name ending in .microsoft.com") err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM dns WHERE cutToFirstSignificantSubdomain(query) = 'microsoft.com' `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 0, count, "dns table shouldn't contain any entries with query ending in .microsoft.com") // verify that not all domains in .businessinsider.com are filtered err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM dns WHERE cutToFirstSignificantSubdomain(query) = 'businessinsider.com' `).Scan(&count) require.NoError(t, err) require.Greater(t, count, uint64(0), "dns table should contain at least one entry with a query ending in .businessinsider.com") err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM http WHERE cutToFirstSignificantSubdomain(host) = 'businessinsider.com' `).Scan(&count) require.NoError(t, err) require.Greater(t, count, uint64(0), "http table should contain at least one entry with a host ending in .businessinsider.com") err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM ssl WHERE cutToFirstSignificantSubdomain(server_name) = 'businessinsider.com' `).Scan(&count) require.NoError(t, err) require.Greater(t, count, uint64(0), "ssl table should contain at least one entry with a server_name ending in .businessinsider.com") // verify that businessinsider.com is filtered err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM dns WHERE query = 'businessinsider.com' `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 0, count, "dns table shouldn't contain any entries with a query of .businessinsider.com") err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM http WHERE host = 'businessinsider.com' `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 0, count, "http table shouldn't contain any entries with a query of .businessinsider.com") err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM ssl WHERE server_name = 'businessinsider.com' `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 0, count, "ssl table shouldn't contain any entries with a query of .businessinsider.com") } func (it *FilterTestSuite) TestAlwaysIncludeSubnets() { t := it.T() // set up file system interface afs := afero.NewMemMapFs() afs2 := afero.NewOsFs() err := afero.WriteFile(afs, "testsuite_config3.hjson", []byte(` { filtering: { filter_external_to_internal: true, internal_subnets: ["10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "fd00::/8"], never_included_subnets: ["10.0.0.0/8"], always_included_subnets: ["10.55.100.0/24"] }, threat_intel: { online_feeds: ["https://feodotracker.abuse.ch/downloads/ipblocklist.txt"], custom_feeds_directory: "./deployment/threat_intel_feeds" }, http_extensions_file_path: "../deployment/http_extensions_list.csv" } `), 0755) require.NoError(t, err) cfg, err := config.ReadFileConfig(afs, "testsuite_config3.hjson") require.NoError(t, err) cfg.Env.DBConnection = dockerInfo.clickhouseConnection it.cfg = cfg require.Contains(t, cfg.Filtering.NeverIncludedSubnets, util.NewSubnet(&net.IPNet{IP: net.IP{10, 0, 0, 0}, Mask: net.IPMask{255, 0, 0, 0}}), "never included subnets should contain 10.0.0.0/8") require.Contains(t, cfg.Filtering.AlwaysIncludedSubnets, util.NewSubnet(&net.IPNet{IP: net.IP{10, 55, 100, 0}, Mask: net.IPMask{255, 255, 255, 0}}), "always included subnets should contain 10.55.100.0/24") // // import data _, err = cmd.RunImportCmd(time.Now(), cfg, afs2, "../test_data/valid_tsv", "always_include_subnet", false, true) require.NoError(t, err) // connect to database db, err := database.ConnectToDB(context.Background(), "always_include_subnet", cfg, nil) require.NoError(t, err) var count uint64 // verify that not all connections in 10.0.0.0/8 were filtered conn := 12591 http := 1982 + 176 // regular http + http missing host ssl := 5531 err = db.Conn.QueryRow(db.GetContext(), ` SELECT count(DISTINCT hash) FROM conn WHERE isIPAddressInRange(IPv6NumToString(src), '::ffff:10.0.0.0/104') OR isIPAddressInRange(IPv6NumToString(dst), '::ffff:10.0.0.0/104') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, conn, count, "conn table should contain %d entries in 10.0.0.0/8, got: %d", conn, count) err = db.Conn.QueryRow(db.GetContext(), ` SELECT count(DISTINCT hash) FROM ssl WHERE isIPAddressInRange(IPv6NumToString(src), '::ffff:10.0.0.0/104') OR isIPAddressInRange(IPv6NumToString(dst), '::ffff:10.0.0.0/104') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, ssl, count, "ssl table should contain %d entries in 10.0.0.0/8, got: %d", ssl, count) err = db.Conn.QueryRow(db.GetContext(), ` SELECT count(DISTINCT hash) FROM http WHERE isIPAddressInRange(IPv6NumToString(src), '::ffff:10.0.0.0/104') OR isIPAddressInRange(IPv6NumToString(dst), '::ffff:10.0.0.0/104') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, http, count, "http table should contain %d entries in 10.0.0.0/8, got: %d", http, count) // verify that all connections in 10.55.100.0/24 were filtered err = db.Conn.QueryRow(db.GetContext(), ` SELECT count(DISTINCT hash) FROM conn WHERE isIPAddressInRange(IPv6NumToString(src), '::ffff:10.55.100.0/120') OR isIPAddressInRange(IPv6NumToString(dst), '::ffff:10.55.100.0/120') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, conn, count, "conn table shouldn contain %d any entries in 10.55.100.0/24, got: %d", conn, count) err = db.Conn.QueryRow(db.GetContext(), ` SELECT count(DISTINCT hash) FROM http WHERE isIPAddressInRange(IPv6NumToString(src), '::ffff:10.55.100.0/120') OR isIPAddressInRange(IPv6NumToString(dst), '::ffff:10.55.100.0/120') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, http, count, "http table shouldn contain %d entries in 10.55.100.0/24, got: %d", http, count) err = db.Conn.QueryRow(db.GetContext(), ` SELECT count(DISTINCT hash) FROM ssl WHERE isIPAddressInRange(IPv6NumToString(src), '::ffff:10.55.100.0/120') OR isIPAddressInRange(IPv6NumToString(dst), '::ffff:10.55.100.0/120') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, ssl, count, "ssl table should contain %d entries in 10.55.100.0/24", ssl, count) err = db.Conn.QueryRow(db.GetContext(), ` SELECT count(DISTINCT hash) FROM dns WHERE isIPAddressInRange(IPv6NumToString(src), '::ffff:10.55.100.0/120') OR isIPAddressInRange(IPv6NumToString(dst), '::ffff:10.55.100.0/120') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 1, count, "dns table should contain 1 entry in 10.55.100.0/24") } func (it *FilterTestSuite) TestAlwaysIncludeDomains() { t := it.T() // set up file system interface afs := afero.NewMemMapFs() afs2 := afero.NewOsFs() err := afero.WriteFile(afs, "testsuite_config4.hjson", []byte(` { filtering: { filter_external_to_internal: true, internal_subnets: ["10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "fd00::/8"], never_included_domains: ["*.microsoft.com", "businessinsider.com"], always_included_domains: ["*.mp.microsoft.com", "analytics.businessinsider.com"] }, threat_intel: { online_feeds: ["https://feodotracker.abuse.ch/downloads/ipblocklist.txt"], custom_feeds_directory: "./deployment/threat_intel_feeds" }, http_extensions_file_path: "../deployment/http_extensions_list.csv" } `), 0755) require.NoError(t, err) cfg, err := config.ReadFileConfig(afs, "testsuite_config4.hjson") require.NoError(t, err) cfg.Env.DBConnection = dockerInfo.clickhouseConnection it.cfg = cfg require.Contains(t, cfg.Filtering.NeverIncludedDomains, "*.microsoft.com") require.Contains(t, cfg.Filtering.NeverIncludedDomains, "businessinsider.com") require.Contains(t, cfg.Filtering.AlwaysIncludedDomains, "*.mp.microsoft.com") require.Contains(t, cfg.Filtering.AlwaysIncludedDomains, "analytics.businessinsider.com") // // import data _, err = cmd.RunImportCmd(time.Now(), cfg, afs2, "../test_data/valid_tsv", "always_include_domain", false, true) require.NoError(t, err) // connect to database db, err := database.ConnectToDB(context.Background(), "always_include_domain", cfg, nil) require.NoError(t, err) var count uint64 // verify that all connections w/ fqdns ending in .microsoft.com (but not *.mp.microsoft.com) are filtered err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM http WHERE cutToFirstSignificantSubdomain(host) = 'microsoft.com' AND NOT endsWith(host, '.mp.microsoft.com') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 0, count, "http table shouldn't contain any entries with host ending in .microsoft.com (but not *.mp.microsoft.com)") err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM ssl WHERE cutToFirstSignificantSubdomain(server_name) = 'microsoft.com' AND NOT endsWith(server_name, '.mp.microsoft.com') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 0, count, "ssl table shouldn't contain any entries with server_name ending in .microsoft.com (but not *.mp.microsoft.com)") err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM dns WHERE cutToFirstSignificantSubdomain(query) = 'microsoft.com' AND NOT endsWith(query, '.mp.microsoft.com') `).Scan(&count) require.NoError(t, err) require.EqualValues(t, 0, count, "dns table shouldn't contain any entries with query ending in .microsoft.com (but not *.mp.microsoft.com)") // verify that connections w/ fqdns ending in *.mp.microsoft.com are not filtered err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM http WHERE endsWith(host, '.mp.microsoft.com') `).Scan(&count) require.NoError(t, err) require.Greater(t, count, uint64(0), "http table should contain at least 1 entry with host ending in .mp.microsoft.com") err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM ssl WHERE endsWith(server_name, '.mp.microsoft.com') `).Scan(&count) require.NoError(t, err) require.Greater(t, count, uint64(0), "ssl table should contain at least 1 entry with server name ending in .microsoft.com") err = db.Conn.QueryRow(db.GetContext(), ` SELECT count() FROM dns WHERE endsWith(query, '.mp.microsoft.com') `).Scan(&count) require.NoError(t, err) require.Greater(t, count, uint64(0), "dns table should contain at least 1 entry with query ending in .microsoft.com") tables := []struct { name string field string }{{name: "dns", field: "query"}, {name: "http", field: "host"}, {name: "ssl", field: "server_name"}} domains := []struct { d string shouldBeFiltered bool }{{d: "businessinsider.com", shouldBeFiltered: true}, {d: "static2.businessinsider.com", shouldBeFiltered: false}} for _, domain := range domains { for _, table := range tables { chCtx := db.QueryParameters(clickhouse.Parameters{ "table": table.name, "field": table.field, "domain": domain.d, }) // verify that not all subdomains in this domain are filtered (wildcard shouldn't apply) err = db.Conn.QueryRow(chCtx, ` SELECT count() FROM {table:Identifier} WHERE endsWith({field:Identifier}, {domain:String}) `).Scan(&count) require.NoError(t, err) require.Greater(t, count, uint64(0), "%s table should contain at least one entry with a domain that ends in .%s", table.name, domain.d) // verify that domain is filtered (if it should be) err = db.Conn.QueryRow(chCtx, ` SELECT count() FROM {table:Identifier} WHERE {field:Identifier} = {domain:String} `).Scan(&count) require.NoError(t, err) if domain.shouldBeFiltered { require.EqualValues(t, 0, count, "%s table shouldn't contain any entries with a domain of %s", table, domain.d) } else { require.Greater(t, count, uint64(0), "%s table should contain at least one entry with a domain of %s", table.name, domain.d) } } } } // TestFilterExternalToInternal also tests ICMP func (it *FilterTestSuite) TestFilterExternalToInternal() { t := it.T() // set up file system interface afs := afero.NewOsFs() cfg, err := config.ReadFileConfig(afs, ConfigPath) require.NoError(t, err) cfg.Env.DBConnection = dockerInfo.clickhouseConnection cfg.Filtering.FilterExternalToInternal = false it.cfg = cfg require.NoError(t, err, "updating config should not return an error") require.False(t, cfg.Filtering.FilterExternalToInternal) // // import data importResults, err := cmd.RunImportCmd(time.Now(), cfg, afs, "../test_data/valid_tsv", "filter_ext_to_int", false, true) require.NoError(t, err) // connect to database db, err := database.ConnectToDB(context.Background(), "filter_ext_to_int", cfg, nil) require.NoError(t, err) // // there are ICMP connections that are only on connections that are external to internal type protoInfo struct { PortProtoService string `ch:"port_proto_service"` ConnCount uint64 `ch:"conn_count"` BytesSent uint64 `ch:"bytes_sent"` BytesReceived uint64 `ch:"bytes_received"` } type testData struct { src string dst string portInfoList []protoInfo } testCases := []testData{ { src: "165.227.88.15", dst: "192.168.88.2", portInfoList: []protoInfo{ {"icmp:3/3", 2, 4858, 0}, }, }, { src: "66.218.84.141", dst: "10.55.100.107", portInfoList: []protoInfo{ {"icmp:3/3", 1, 240, 0}, }, }, { src: "66.218.84.141", dst: "10.55.100.104", portInfoList: []protoInfo{ {"icmp:3/3", 1, 240, 0}, }, }, { src: "66.218.84.141", dst: "10.55.100.108", portInfoList: []protoInfo{ {"icmp:3/3", 1, 240, 0}, }, }, { src: "67.226.210.13", dst: "10.55.100.106", portInfoList: []protoInfo{ {"icmp:3/10", 1, 136, 0}, }, }, { src: "67.226.210.15", dst: "10.55.100.107", portInfoList: []protoInfo{ {"icmp:3/10", 1, 136, 0}, }, }, { src: "67.226.210.14", dst: "10.55.100.108", portInfoList: []protoInfo{ {"icmp:3/10", 1, 136, 0}, }, }, { src: "24.220.6.168", dst: "10.55.200.11", portInfoList: []protoInfo{ {"icmp:3/13", 1, 96, 0}, }, }, } for _, test := range testCases { ctx := clickhouse.Context(context.Background(), clickhouse.WithParameters(clickhouse.Parameters{ "src": test.src, "dst": test.dst, })) var res []protoInfo err = db.Conn.Select(ctx, &res, ` SELECT concat(proto, ':', icmp_type, '/', icmp_code) AS port_proto_service, countMerge(count) AS conn_count, sumMerge(bytes_sent) AS bytes_sent, sumMerge(bytes_received) AS bytes_received FROM port_info WHERE src={src:String} AND dst={dst:String} GROUP BY src, dst, dst_port, proto, service, icmp_type, icmp_code `) require.NoError(t, err, "querying proto table should not produce an error") // ensure that the length of the result list matches the expected value require.Len(t, res, len(test.portInfoList), "length of result list should match expected value") // ensure that the result list matches the expected value require.ElementsMatch(t, test.portInfoList, res, "result list should match expected value") } /* ******* Mixtape Propagation ******* The entries that use ICMP don't have enough connections to qualify as beacons, so they don't appear in the mixtape. In order to test the spagooper query that grabs the ICMP entries within port:proto:service, we have to go through the results of the IP spagooper. */ // set up new analyzer minTSBeacon, maxTSBeacon, notFromConn, err := db.GetBeaconMinMaxTimestamps() require.NoError(t, err) require.False(t, notFromConn, "min and max timestamps should be from conn table") minTS, maxTS, notFromConn, useCurrentTime, err := db.GetTrueMinMaxTimestamps() require.NoError(t, err) require.False(t, notFromConn, "min and max timestamps should be from conn table") require.False(t, useCurrentTime, "first seen analysis should not use the current time") analyzer, err := analysis.NewAnalyzer(db, cfg, importResults.ImportID[0], minTS, maxTS, minTSBeacon, maxTSBeacon, useCurrentTime, false) require.NoError(t, err) ctx := context.Background() queryGroup, ctx := errgroup.WithContext(ctx) // create progress bars bars := progressbar.New(ctx, []*progressbar.ProgressBar{ progressbar.NewBar("IP Connection Analysis ", 2, progress.New(progress.WithDefaultGradient())), }, []progressbar.Spinner{}) type foundEntry struct { PortProtoService string ConnCount uint64 TotalBytes uint64 } type resData struct { src string dst string portInfoList []foundEntry } var foundIPs []resData var expectedResData []resData for _, test := range testCases { d := resData{ src: test.src, dst: test.dst, } var p []foundEntry for _, dd := range test.portInfoList { p = append(p, foundEntry{ PortProtoService: dd.PortProtoService, ConnCount: dd.ConnCount, TotalBytes: (dd.BytesSent + dd.BytesReceived) * 2, // multiply by 2 for openconns }) } d.portInfoList = p expectedResData = append(expectedResData, d) } queryGroup.Go(func() error { for entry := range analyzer.UconnChan { for _, test := range testCases { if entry.Src.String() == test.src && entry.Dst.String() == test.dst { f := resData{ src: entry.Src.String(), dst: entry.Dst.String(), } var portProto []foundEntry for _, p := range entry.PortProtoService { port := foundEntry{ PortProtoService: p, ConnCount: entry.Count, TotalBytes: entry.TotalBytes, } portProto = append(portProto, port) } f.portInfoList = portProto foundIPs = append(foundIPs, f) } } } return nil }) queryGroup.Go(func() error { err := analyzer.ScoopIPConns(ctx, bars) require.NoError(t, err) close(analyzer.UconnChan) return err }) queryGroup.Go(func() error { _, err := bars.Run() require.NoError(t, err) return err }) if err := queryGroup.Wait(); err != nil { require.NoError(t, err) } require.ElementsMatch(t, expectedResData, foundIPs) }