Add cloud services to topology

deepfence_server/handler/agent_report.go

@@ -13,6 +13,7 @@ import (
 	jsoniter "github.com/json-iterator/go"
 
 	"github.com/deepfence/ThreatMapper/deepfence_server/ingesters"
+	"github.com/deepfence/ThreatMapper/deepfence_server/model"
 	"github.com/deepfence/ThreatMapper/deepfence_server/pkg/scope/report"
 	"github.com/deepfence/ThreatMapper/deepfence_utils/controls"
 	"github.com/deepfence/ThreatMapper/deepfence_utils/directory"
@@ -40,7 +41,13 @@ func getAgentReportIngester(ctx context.Context) (*ingesters.Ingester[report.Com
 	if has {
 		return ing.(*ingesters.Ingester[report.CompressedReport]), nil
 	}
-	newEntry, err := ingesters.NewNeo4jCollector(ctx)
+
+	token, err := model.SimpleFetchLicense(ctx)
+	if err != nil {
+		log.Error().Err(err).Msg("Continuing without license key")
+	}
+
+	newEntry, err := ingesters.NewNeo4jCollector(ctx, token)
 	if err != nil {
 		return nil, err
 	}

deepfence_server/handler/topology.go

@@ -278,7 +278,7 @@ func graphToSummaries(
 		return res
 	}
 
-	for cp, crs := range graph.CloudServices {
+	for cp, crs := range graph.InternalCloudServices {
 		for _, crStub := range crs {
 			cr := string(crStub.ID)
 			nodes[cr] = detailed.NodeSummary{
@@ -293,6 +293,14 @@ func graphToSummaries(
 
 	nodes["in-the-internet"] = inboundInternetNode
 	nodes["out-the-internet"] = outboundInternetNode
+	for _, n := range graph.ExternalCloudServices {
+		nodes[string(n.ID)] = detailed.NodeSummary{
+			ID:                string(n.ID),
+			Label:             n.Name,
+			ImmediateParentID: "",
+			Type:              "pseudo",
+		}
+	}
 
 	for h, n := range graph.Processes {
 		for _, idStub := range n {

deepfence_server/ingesters/agent.go

@@ -3,14 +3,20 @@ package ingesters
 import (
 	"bytes"
 	"context"
+	"crypto/tls"
+	"encoding/json"
 	"fmt"
+	"io"
+	"net"
+	"net/http"
 	"os"
 	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/deepfence/ThreatMapper/deepfence_server/pkg/constants"
 	"github.com/deepfence/ThreatMapper/deepfence_server/pkg/scope/report"
 	"github.com/deepfence/ThreatMapper/deepfence_utils/directory"
 	"github.com/deepfence/ThreatMapper/deepfence_utils/log"
@@ -409,6 +415,7 @@ type Connection struct {
 	localPort   int
 	leftIP      *string
 	rightIP     *string
+	resolved    bool
 }
 
 func connections2maps(connections []Connection, buf *bytes.Buffer) []map[string]interface{} {
@@ -485,7 +492,7 @@ func NewReportIngestionData() ReportIngestionData {
 	}
 }
 
-func prepareNeo4jIngestion(rpt *report.Report, resolvers *EndpointResolversCache, buf *bytes.Buffer) ReportIngestionData {
+func prepareNeo4jIngestion(rpt *report.Report, resolvers *EndpointResolversCache, buf *bytes.Buffer, token string) ReportIngestionData {
 
 	res := ReportIngestionData{
 		Hosts:                      make([]map[string]interface{}, 0, len(rpt.Host)),
@@ -620,6 +627,8 @@ func prepareNeo4jIngestion(rpt *report.Report, resolvers *EndpointResolversCache
 			}
 		}
 	}
+	connections = resolveCloudServices(connections, token)
+	connections = resolveReverseDNS(connections)
 	res.EndpointEdgesBatch = connections2maps(connections, buf)
 
 	processEdgesBatch := map[string][]string{}
@@ -762,7 +771,8 @@ func (nc *neo4jIngester) PushToDBSeq(ctx context.Context, batches ReportIngestio
 	if _, err := tx.Run(ctx, `
 		UNWIND $batch as row
 		MATCH (n:Node{node_id: row.source})
-		MATCH (m:Node{node_id: row.destination})
+		MERGE (m:Node{node_id: row.destination})
+		ON CREATE SET m.pseudo = true, m.active = true, m.updated_at = TIMESTAMP(), m.cloud_region = 'internet', m.cloud_provider = 'internet'
 		MERGE (n)-[r:CONNECTS]->(m)
 		WITH n, r, m, row.pids as rpids
 		UNWIND rpids as pids
@@ -1038,11 +1048,11 @@ func (nc *neo4jIngester) runDBPusher(
 	log.Info().Msgf("runDBPusher ended")
 }
 
-func (nc *neo4jIngester) runPreparer() {
+func (nc *neo4jIngester) runPreparer(token string) {
 	var buf bytes.Buffer
 	for rpt := range nc.preparersInput {
 		r := computeResolvers(&rpt, &buf)
-		data := prepareNeo4jIngestion(&rpt, nc.resolvers, &buf)
+		data := prepareNeo4jIngestion(&rpt, nc.resolvers, &buf, token)
 		select {
 		case nc.batcher <- data:
 			nc.resolversUpdate <- r
@@ -1052,7 +1062,7 @@ func (nc *neo4jIngester) runPreparer() {
 	}
 }
 
-func NewNeo4jCollector(ctx context.Context) (Ingester[report.CompressedReport], error) {
+func NewNeo4jCollector(ctx context.Context, token string) (Ingester[report.CompressedReport], error) {
 	rdb, err := newEndpointResolversCache(ctx)
 
 	if err != nil {
@@ -1112,7 +1122,7 @@ func NewNeo4jCollector(ctx context.Context) (Ingester[report.CompressedReport],
 	}()
 
 	for i := 0; i < preparerWorkersNum; i++ {
-		go nc.runPreparer()
+		go nc.runPreparer(token)
 	}
 
 	go func() {
@@ -1332,3 +1342,126 @@ func GetPushBack(ctx context.Context, driver neo4j.DriverWithContext) (int32, er
 	}
 	return int32(rec.Values[0].(int64)), nil
 }
+
+const (
+	threatIntelResolverURL = "https://threat-intel.deepfence.io/threat-intel"
+)
+
+type CloudInfo struct {
+	Type     string `json:"type"`
+	Region   string `json:"region"`
+	Provider string `json:"provider"`
+}
+
+func (ci *CloudInfo) NodeID() string {
+	return fmt.Sprintf("%s-%s", ci.Provider, ci.Type)
+}
+
+type IPResponse struct {
+	Infos []CloudInfo `json:"Infos"`
+}
+
+type IPRequest struct {
+	IPv4s []string `json:"ipv4s"`
+	IPv6s []string `json:"ipv6s"`
+}
+
+func requestCloudInfo(ctx context.Context, strIps []string, token string) ([]CloudInfo, error) {
+	// check if token is present
+	var infos []CloudInfo
+
+	bodyReq := IPRequest{
+		IPv4s: strIps,
+	}
+	b, err := json.Marshal(bodyReq)
+	if err != nil {
+		return infos, err
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, threatIntelResolverURL+"/cloud-ips", bytes.NewReader(b))
+	if err != nil {
+		return infos, err
+	}
+
+	req.Header.Set("x-license-key", token)
+
+	q := req.URL.Query()
+	q.Add("version", constants.Version)
+	q.Add("product", utils.Project)
+	req.URL.RawQuery = q.Encode()
+
+	log.Info().Msgf("query threatintel at %s", req.URL.String())
+
+	tr := http.DefaultTransport.(*http.Transport).Clone()
+	tr.TLSClientConfig = &tls.Config{
+		InsecureSkipVerify: true,
+	}
+	hc := http.Client{
+		Timeout:   1 * time.Minute,
+		Transport: tr,
+	}
+	resp, err := hc.Do(req)
+	if err != nil {
+		log.Error().Err(err).Msg("failed http request")
+		return infos, err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return infos, fmt.Errorf("%d invaid response code", resp.StatusCode)
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		log.Error().Err(err).Msg("failed read response body")
+		return infos, err
+	}
+	defer resp.Body.Close()
+
+	var res IPResponse
+	if err := json.Unmarshal(body, &res); err != nil {
+		log.Error().Err(err).Msg("failed to decode response body")
+		return infos, err
+	}
+
+	return res.Infos, nil
+}
+
+func resolveCloudServices(connections []Connection, token string) []Connection {
+	ips := []string{}
+	ids := []int{}
+	for i := range connections {
+		if connections[i].rightIP != nil {
+			ips = append(ips, *connections[i].rightIP)
+			ids = append(ids, i)
+		}
+	}
+	if len(ips) == 0 {
+		return connections
+	}
+	infos, err := requestCloudInfo(context.Background(), ips, token)
+	if err != nil || len(ids) != len(infos) {
+		log.Error().Err(err).Msgf("issue fetching cloud infos %v/%v", infos, connections)
+		return connections
+	}
+	for i := range infos {
+		if !strings.Contains(infos[i].NodeID(), "unknown") {
+			connections[ids[i]].destination = infos[i].NodeID()
+			connections[ids[i]].resolved = true
+		}
+	}
+	return connections
+}
+
+func resolveReverseDNS(connections []Connection) []Connection {
+	for i := range connections {
+		if connections[i].rightIP != nil && !connections[i].resolved {
+			names, err := net.LookupAddr(*connections[i].rightIP)
+			if err != nil || len(names) == 0 {
+				log.Error().Err(err).Msgf("Reverse DNS failed for %s", *connections[i].rightIP)
+				continue
+			}
+			connections[i].destination = fmt.Sprintf("out-%s", names[0])
+		}
+	}
+	return connections
+}

deepfence_server/model/license.go

@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/deepfence/ThreatMapper/deepfence_utils/directory"
 	"github.com/deepfence/ThreatMapper/deepfence_utils/log"
 	postgresqlDb "github.com/deepfence/ThreatMapper/deepfence_utils/postgresql/postgresql-db"
 	"github.com/deepfence/ThreatMapper/deepfence_utils/utils"
@@ -371,3 +372,16 @@ func formatDate(dt time.Time) string {
 func parseDate(dateString string) (time.Time, error) {
 	return time.Parse(DateLayout1, dateString)
 }
+
+// FetchLicense gets license key from database
+func SimpleFetchLicense(ctx context.Context) (string, error) {
+	pgClient, err := directory.PostgresClient(ctx)
+	if err != nil {
+		return "", err
+	}
+	license, err := GetLicense(ctx, pgClient)
+	if err != nil {
+		return "", err
+	}
+	return license.LicenseKey, nil
+}

deepfence_server/reporters/graph/topology_reporter.go

@@ -436,7 +436,7 @@ func extractResourceNodeIds(ids []interface{}) []NodeID {
 	return res
 }
 
-func (ntp *neo4jTopologyReporter) GetCloudServices(
+func (ntp *neo4jTopologyReporter) GetInternalCloudServices(
 	ctx context.Context,
 	tx neo4j.ExplicitTransaction,
 	cloudProvider []string,
@@ -491,6 +491,46 @@ func (ntp *neo4jTopologyReporter) GetCloudServices(
 
 }
 
+func (ntp *neo4jTopologyReporter) GetExternalCloudServices(
+	ctx context.Context,
+	tx neo4j.ExplicitTransaction,
+	cloudProvider []string,
+	cloudRegions []string,
+	fieldfilters mo.Option[reporters.FieldsFilters]) ([]NodeStub, error) {
+
+	ctx, span := telemetry.NewSpan(ctx, "toploogy", "get-cloud-services")
+	defer span.End()
+
+	res := []NodeStub{}
+
+	r, err := tx.Run(ctx, `
+		MATCH (n:Node) -[:CONNECTS]- (:Node)
+		WHERE n.cloud_provider = "internet"
+		RETURN n.node_id`,
+		map[string]interface{}{},
+	)
+
+	if err != nil {
+		return res, err
+	}
+	records, err := r.Collect(ctx)
+
+	if err != nil {
+		return res, err
+	}
+
+	for _, record := range records {
+		nodeID := record.Values[0].(string)
+		res = append(res,
+			NodeStub{
+				ID:   NodeID(nodeID),
+				Name: nodeID,
+			})
+	}
+	return res, nil
+
+}
+
 func (ntp *neo4jTopologyReporter) GetPublicCloudResources(
 	ctx context.Context,
 	tx neo4j.ExplicitTransaction,
@@ -921,8 +961,9 @@ type RenderedGraph struct {
 	Connections []ConnectionSummary   `json:"connections" required:"true"`
 	// PublicCloudResources    map[NodeID][]ResourceStub `json:"public-cloud-resources" required:"true"`
 	// NonPublicCloudResources map[NodeID][]ResourceStub `json:"non-public-cloud-resources" required:"true"`
-	CloudServices      map[NodeID][]ResourceStub `json:"cloud-services" required:"true"`
-	SkippedConnections bool                      `json:"skipped_connections" required:"true"`
+	InternalCloudServices map[NodeID][]ResourceStub `json:"cloud-services" required:"true"`
+	ExternalCloudServices []NodeStub                `json:"external-cloud-services" required:"true"`
+	SkippedConnections    bool                      `json:"skipped_connections" required:"true"`
 }
 
 type TopologyFilters struct {
@@ -1239,7 +1280,11 @@ func (ntp *neo4jTopologyReporter) getGraph(ctx context.Context, filters Topology
 	if err != nil {
 		return res, err
 	}
-	res.CloudServices, err = ntp.GetCloudServices(ctx, tx, cloudFilter, regionFilter, mo.None[reporters.FieldsFilters]())
+	res.InternalCloudServices, err = ntp.GetInternalCloudServices(ctx, tx, cloudFilter, regionFilter, mo.None[reporters.FieldsFilters]())
+	if err != nil {
+		return res, err
+	}
+	res.ExternalCloudServices, err = ntp.GetExternalCloudServices(ctx, tx, cloudFilter, regionFilter, mo.None[reporters.FieldsFilters]())
 	if err != nil {
 		return res, err
 	}

deepfence_utils/utils/constants.go

@@ -67,7 +67,8 @@ const (
 	UpdateLicenseTask      = "update_license"
 	ReportLicenseUsageTask = "report_license_usage"
 
-	ThreatIntelUpdateTask = "threat_intel_update"
+	ThreatIntelUpdateTask   = "threat_intel_update"
+	CloudServicesUpdateTask = "cloud_services_update"
 )
 
 const (
@@ -238,6 +239,7 @@ var Tasks = []string{
 	ReportLicenseUsageTask,
 
 	ThreatIntelUpdateTask,
+	CloudServicesUpdateTask,
 }
 
 type ReportType string