8723654472

Committed 17 Apr 2024 02:23PM UTC coverage: 75.676% (-0.3%) from 75.999%

Build # 8723654472

Build Type

Pull #2126

github

Committed by

kubasobon

Commit Message

add add_host_metadata processor for KSPM

Pull Request Pull Request #2126: Add correct host info for cloud provider virtual machines

Run Details

10 of 55 new or added lines in 3 files covered. (18.18%)

2 existing lines in 1 file now uncovered.

7333 of 9690 relevant lines covered (75.68%)

18.14 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.24

/internal/resources/fetching/fetchers/azure/assets_fetcher.go

// Licensed to Elasticsearch B.V. under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Elasticsearch B.V. licenses this file to you under
// the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package fetchers

import (
        "context"
        "errors"
        "fmt"

        "github.com/elastic/elastic-agent-libs/logp"
        "golang.org/x/exp/maps"

        "github.com/elastic/cloudbeat/internal/resources/fetching"
        "github.com/elastic/cloudbeat/internal/resources/fetching/cycle"
        "github.com/elastic/cloudbeat/internal/resources/providers/azurelib"
        "github.com/elastic/cloudbeat/internal/resources/providers/azurelib/governance"
        "github.com/elastic/cloudbeat/internal/resources/providers/azurelib/inventory"
)

type AzureAssetsFetcher struct {
        log        *logp.Logger
        resourceCh chan fetching.ResourceInfo
        provider   azurelib.ProviderAPI
        enrichers  []AssetsEnricher
}

type AzureResource struct {
        Type         string
        SubType      string
        Asset        inventory.AzureAsset `json:"asset,omitempty"`
        Subscription governance.Subscription
}

type typePair struct {
        SubType string
        Type    string
}

func newPair(subType string, tpe string) typePair {
        return typePair{
                SubType: subType,
                Type:    tpe,
        }
}

var AzureAssetTypeToTypePair = map[string]typePair{
        inventory.ClassicStorageAccountAssetType:     newPair(fetching.AzureClassicStorageAccountType, fetching.CloudStorage),
        inventory.DiskAssetType:                      newPair(fetching.AzureDiskType, fetching.CloudCompute),
        inventory.DocumentDBDatabaseAccountAssetType: newPair(fetching.AzureDocumentDBDatabaseAccountType, fetching.CloudDatabase),
        inventory.MySQLDBAssetType:                   newPair(fetching.AzureMySQLDBType, fetching.CloudDatabase),
        inventory.FlexibleMySQLDBAssetType:           newPair(fetching.AzureFlexibleMySQLDBType, fetching.CloudDatabase),
        inventory.NetworkWatchersFlowLogAssetType:    newPair(fetching.AzureNetworkWatchersFlowLogType, fetching.MonitoringIdentity),
        inventory.FlexiblePostgreSQLDBAssetType:      newPair(fetching.AzureFlexiblePostgreSQLDBType, fetching.CloudDatabase),
        inventory.PostgreSQLDBAssetType:              newPair(fetching.AzurePostgreSQLDBType, fetching.CloudDatabase),
        inventory.SQLServersAssetType:                newPair(fetching.AzureSQLServerType, fetching.CloudDatabase),
        inventory.StorageAccountAssetType:            newPair(fetching.AzureStorageAccountType, fetching.CloudStorage),
        inventory.VirtualMachineAssetType:            newPair(fetching.AzureVMType, fetching.CloudCompute),
        inventory.WebsitesAssetType:                  newPair(fetching.AzureWebSiteType, fetching.CloudCompute),
        inventory.VaultAssetType:                     newPair(fetching.AzureVaultType, fetching.KeyManagement),
        inventory.RoleDefinitionsType:                newPair(fetching.AzureRoleDefinitionType, fetching.CloudIdentity),

        // This asset type is used only for enrichment purposes, but is sent to OPA layer, producing no findings.
        inventory.NetworkSecurityGroupAssetType: newPair(fetching.AzureNetworkSecurityGroupType, fetching.MonitoringIdentity),
}

// In order to simplify the mappings, we are trying to query all AzureAssetTypeToTypePair on every asset group
// Because this is done with an "|"" this means that we won't get irrelevant data
var AzureAssetGroups = []string{inventory.AssetGroupResources, inventory.AssetGroupAuthorizationResources}

func NewAzureAssetsFetcher(log *logp.Logger, ch chan fetching.ResourceInfo, provider azurelib.ProviderAPI) *AzureAssetsFetcher {
        return &AzureAssetsFetcher{
                log:        log,
                resourceCh: ch,
                provider:   provider,
                enrichers:  initEnrichers(provider),
        }
}

func (f *AzureAssetsFetcher) Fetch(ctx context.Context, cycleMetadata cycle.Metadata) error {
        f.log.Info("Starting AzureAssetsFetcher.Fetch")
        var errAgg error
        // This might be relevant if we'd like to fetch assets in parallel in order to evaluate a rule that uses multiple resources
        var assets []inventory.AzureAsset
        for _, assetGroup := range AzureAssetGroups {
                // Fetching all types even if non-existent in asset group for simplicity
                r, err := f.provider.ListAllAssetTypesByName(ctx, assetGroup, maps.Keys(AzureAssetTypeToTypePair))
                if err != nil {
                        f.log.Errorf("AzureAssetsFetcher.Fetch failed to fetch asset group %s: %s", assetGroup, err.Error())
                        errAgg = errors.Join(errAgg, err)
                        continue
                }
                assets = append(assets, r...)
        }

        subscriptions, err := f.provider.GetSubscriptions(ctx, cycleMetadata)
        if err != nil {
                f.log.Errorf("Error fetching subscription information: %v", err)
        }

        for _, e := range f.enrichers {
                if err := e.Enrich(ctx, cycleMetadata, assets); err != nil {
                        errAgg = errors.Join(errAgg, fmt.Errorf("error while enriching assets: %w", err))
                }
        }

        for _, asset := range assets {
                select {
                case <-ctx.Done():
                        err := ctx.Err()
                        f.log.Infof("AzureAssetsFetcher.Fetch context err: %s", err.Error())
                        errAgg = errors.Join(errAgg, err)
                        return errAgg
                case f.resourceCh <- resourceFromAsset(asset, cycleMetadata, subscriptions):
                }
        }

        return errAgg
}

func resourceFromAsset(asset inventory.AzureAsset, cycleMetadata cycle.Metadata, subscriptions map[string]governance.Subscription) fetching.ResourceInfo {
        pair := AzureAssetTypeToTypePair[asset.Type]
        subscription, ok := subscriptions[asset.SubscriptionId]
        if !ok {
                subscription = governance.Subscription{
                        FullyQualifiedID: asset.SubscriptionId,
                        ShortID:          "",
                        DisplayName:      "",
                        ManagementGroup: governance.ManagementGroup{
                                FullyQualifiedID: "",
                                DisplayName:      "",
                        },
                }
        }
        return fetching.ResourceInfo{
                CycleMetadata: cycleMetadata,
                Resource: &AzureResource{
                        Type:         pair.Type,
                        SubType:      pair.SubType,
                        Asset:        asset,
                        Subscription: subscription,
                },
        }
}

func (f *AzureAssetsFetcher) Stop() {}

func (r *AzureResource) GetData() any {
        return r.Asset
}

func (r *AzureResource) GetMetadata() (fetching.ResourceMetadata, error) {
        return fetching.ResourceMetadata{
                ID:                   r.Asset.Id,
                Type:                 r.Type,
                SubType:              r.SubType,
                Name:                 r.Asset.Name,
                Region:               r.Asset.Location,
                CloudAccountMetadata: r.Subscription.GetCloudAccountMetadata(),
        }, nil
}

func (r *AzureResource) GetElasticCommonData() (map[string]any, error) {
        if r.Asset.Type == inventory.VirtualMachineAssetType {
                m := map[string]any{
                        "host.name": r.Asset.Name,
                }

                // "host.hostname" = "properties.osProfile.computerName" if it exists
                osProfileRaw, ok := r.Asset.Properties["osProfile"]
                if !ok {
                        return m, nil
                }
                osProfile, ok := osProfileRaw.(map[string]any)
                if !ok {
                        return m, nil
                }
                computerNameRaw, ok := osProfile["computerName"]
                if !ok {
                        return m, nil
                }
                computerName, ok := computerNameRaw.(string)
                if !ok {
                        return m, nil
                }
                m["host.hostname"] = computerName
                return m, nil
        }
        return nil, nil
}

1	// Licensed to Elasticsearch B.V. under one or more contributor
2	// license agreements. See the NOTICE file distributed with
3	// this work for additional information regarding copyright
4	// ownership. Elasticsearch B.V. licenses this file to you under
5	// the Apache License, Version 2.0 (the "License"); you may
6	// not use this file except in compliance with the License.
7	// You may obtain a copy of the License at
8	//
9	// http://www.apache.org/licenses/LICENSE-2.0
10	//
11	// Unless required by applicable law or agreed to in writing,
12	// software distributed under the License is distributed on an
13	// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14	// KIND, either express or implied. See the License for the
15	// specific language governing permissions and limitations
16	// under the License.
17
18	package fetchers
19
20	import (
21	"context"
22	"errors"
23	"fmt"
24
25	"github.com/elastic/elastic-agent-libs/logp"
26	"golang.org/x/exp/maps"
27
28	"github.com/elastic/cloudbeat/internal/resources/fetching"
29	"github.com/elastic/cloudbeat/internal/resources/fetching/cycle"
30	"github.com/elastic/cloudbeat/internal/resources/providers/azurelib"
31	"github.com/elastic/cloudbeat/internal/resources/providers/azurelib/governance"
32	"github.com/elastic/cloudbeat/internal/resources/providers/azurelib/inventory"
33	)
34
35	type AzureAssetsFetcher struct {
36	log *logp.Logger
37	resourceCh chan fetching.ResourceInfo
38	provider azurelib.ProviderAPI
39	enrichers []AssetsEnricher
40	}
41
42	type AzureResource struct {
43	Type string
44	SubType string
45	Asset inventory.AzureAsset `json:"asset,omitempty"`
46	Subscription governance.Subscription
47	}
48
49	type typePair struct {
50	SubType string
51	Type string
52	}
53
54	func newPair(subType string, tpe string) typePair {	146✔
55	return typePair{	146✔
56	SubType: subType,	146✔
57	Type: tpe,	146✔
58	}	146✔
59	}	146✔
60
61	var AzureAssetTypeToTypePair = map[string]typePair{
62	inventory.ClassicStorageAccountAssetType: newPair(fetching.AzureClassicStorageAccountType, fetching.CloudStorage),
63	inventory.DiskAssetType: newPair(fetching.AzureDiskType, fetching.CloudCompute),
64	inventory.DocumentDBDatabaseAccountAssetType: newPair(fetching.AzureDocumentDBDatabaseAccountType, fetching.CloudDatabase),
65	inventory.MySQLDBAssetType: newPair(fetching.AzureMySQLDBType, fetching.CloudDatabase),
66	inventory.FlexibleMySQLDBAssetType: newPair(fetching.AzureFlexibleMySQLDBType, fetching.CloudDatabase),
67	inventory.NetworkWatchersFlowLogAssetType: newPair(fetching.AzureNetworkWatchersFlowLogType, fetching.MonitoringIdentity),
68	inventory.FlexiblePostgreSQLDBAssetType: newPair(fetching.AzureFlexiblePostgreSQLDBType, fetching.CloudDatabase),
69	inventory.PostgreSQLDBAssetType: newPair(fetching.AzurePostgreSQLDBType, fetching.CloudDatabase),
70	inventory.SQLServersAssetType: newPair(fetching.AzureSQLServerType, fetching.CloudDatabase),
71	inventory.StorageAccountAssetType: newPair(fetching.AzureStorageAccountType, fetching.CloudStorage),
72	inventory.VirtualMachineAssetType: newPair(fetching.AzureVMType, fetching.CloudCompute),
73	inventory.WebsitesAssetType: newPair(fetching.AzureWebSiteType, fetching.CloudCompute),
74	inventory.VaultAssetType: newPair(fetching.AzureVaultType, fetching.KeyManagement),
75	inventory.RoleDefinitionsType: newPair(fetching.AzureRoleDefinitionType, fetching.CloudIdentity),
76
77	// This asset type is used only for enrichment purposes, but is sent to OPA layer, producing no findings.
78	inventory.NetworkSecurityGroupAssetType: newPair(fetching.AzureNetworkSecurityGroupType, fetching.MonitoringIdentity),
79	}
80
81	// In order to simplify the mappings, we are trying to query all AzureAssetTypeToTypePair on every asset group
82	// Because this is done with an "\|"" this means that we won't get irrelevant data
83	var AzureAssetGroups = []string{inventory.AssetGroupResources, inventory.AssetGroupAuthorizationResources}
84
85	func NewAzureAssetsFetcher(log logp.Logger, ch chan fetching.ResourceInfo, provider azurelib.ProviderAPI) AzureAssetsFetcher {	1✔
86	return &AzureAssetsFetcher{	1✔
87	log: log,	1✔
88	resourceCh: ch,	1✔
89	provider: provider,	1✔
90	enrichers: initEnrichers(provider),	1✔
91	}	1✔
92	}	1✔
93
94	func (f *AzureAssetsFetcher) Fetch(ctx context.Context, cycleMetadata cycle.Metadata) error {	2✔
95	f.log.Info("Starting AzureAssetsFetcher.Fetch")	2✔
96	var errAgg error	2✔
97	// This might be relevant if we'd like to fetch assets in parallel in order to evaluate a rule that uses multiple resources	2✔
98	var assets []inventory.AzureAsset	2✔
99	for _, assetGroup := range AzureAssetGroups {	6✔
100	// Fetching all types even if non-existent in asset group for simplicity	4✔
101	r, err := f.provider.ListAllAssetTypesByName(ctx, assetGroup, maps.Keys(AzureAssetTypeToTypePair))	4✔
102	if err != nil {	5✔
103	f.log.Errorf("AzureAssetsFetcher.Fetch failed to fetch asset group %s: %s", assetGroup, err.Error())	1✔
104	errAgg = errors.Join(errAgg, err)	1✔
105	continue	1✔
106	}
107	assets = append(assets, r...)	3✔
108	}
109
110	subscriptions, err := f.provider.GetSubscriptions(ctx, cycleMetadata)	2✔
111	if err != nil {	3✔
112	f.log.Errorf("Error fetching subscription information: %v", err)	1✔
113	}	1✔
114
115	for _, e := range f.enrichers {	16✔
116	if err := e.Enrich(ctx, cycleMetadata, assets); err != nil {	15✔
117	errAgg = errors.Join(errAgg, fmt.Errorf("error while enriching assets: %w", err))	1✔
118	}	1✔
119	}
120
121	for _, asset := range assets {	33✔
122	select {	31✔
123	case <-ctx.Done():	×
124	err := ctx.Err()	×
125	f.log.Infof("AzureAssetsFetcher.Fetch context err: %s", err.Error())	×
126	errAgg = errors.Join(errAgg, err)	×
127	return errAgg	×
128	case f.resourceCh <- resourceFromAsset(asset, cycleMetadata, subscriptions):	31✔
129	}
130	}
131
132	return errAgg	2✔
133	}
134
135	func resourceFromAsset(asset inventory.AzureAsset, cycleMetadata cycle.Metadata, subscriptions map[string]governance.Subscription) fetching.ResourceInfo {	31✔
136	pair := AzureAssetTypeToTypePair[asset.Type]	31✔
137	subscription, ok := subscriptions[asset.SubscriptionId]	31✔
138	if !ok {	32✔
139	subscription = governance.Subscription{	1✔
140	FullyQualifiedID: asset.SubscriptionId,	1✔
141	ShortID: "",	1✔
142	DisplayName: "",	1✔
143	ManagementGroup: governance.ManagementGroup{	1✔
144	FullyQualifiedID: "",	1✔
145	DisplayName: "",	1✔
146	},	1✔
147	}	1✔
148	}	1✔
149	return fetching.ResourceInfo{	31✔
150	CycleMetadata: cycleMetadata,	31✔
151	Resource: &AzureResource{	31✔
152	Type: pair.Type,	31✔
153	SubType: pair.SubType,	31✔
154	Asset: asset,	31✔
155	Subscription: subscription,	31✔
156	},	31✔
157	}	31✔
158	}
159
160	func (f *AzureAssetsFetcher) Stop() {}	1✔
161
162	func (r *AzureResource) GetData() any {	31✔
163	return r.Asset	31✔
164	}	31✔
165
166	func (r *AzureResource) GetMetadata() (fetching.ResourceMetadata, error) {	31✔
167	return fetching.ResourceMetadata{	31✔
168	ID: r.Asset.Id,	31✔
169	Type: r.Type,	31✔
170	SubType: r.SubType,	31✔
171	Name: r.Asset.Name,	31✔
172	Region: r.Asset.Location,	31✔
173	CloudAccountMetadata: r.Subscription.GetCloudAccountMetadata(),	31✔
174	}, nil	31✔
175	}	31✔
176
177	func (r *AzureResource) GetElasticCommonData() (map[string]any, error) {	30✔
178	if r.Asset.Type == inventory.VirtualMachineAssetType {	32✔
179	m := map[string]any{	2✔
180	"host.name": r.Asset.Name,	2✔
181	}	2✔
182		2✔
183	// "host.hostname" = "properties.osProfile.computerName" if it exists	2✔
184	osProfileRaw, ok := r.Asset.Properties["osProfile"]	2✔
185	if !ok {	4✔
186	return m, nil	2✔
187	}	2✔
NEW 188	osProfile, ok := osProfileRaw.(map[string]any)	×
NEW 189	if !ok {	×
NEW 190	return m, nil	×
NEW 191	}	×
NEW 192	computerNameRaw, ok := osProfile["computerName"]	×
NEW 193	if !ok {	×
NEW 194	return m, nil	×
NEW 195	}	×
NEW 196	computerName, ok := computerNameRaw.(string)	×
NEW 197	if !ok {	×
NEW 198	return m, nil	×
NEW 199	}	×
NEW 200	m["host.hostname"] = computerName	×
NEW 201	return m, nil	×
202	}
203	return nil, nil	28✔
204	}

elastic / cloudbeat / 8723654472

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous