feat: add google workspace extractor

plugins/extractors/googleworkspace/README.md

@@ -0,0 +1,38 @@
+# Google Workspace
+
+## Usage
+
+```yaml
+source:
+  scope: my-scope
+  type: googleworkspace
+  config:     
+    service_account_json: "XXX"
+    user_email: [email protected]
+```
+
+## Inputs
+
+| Key | Value | Example | Description |    |
+| :-- | :---- | :------ | :---------- | :- |
+| `user_email` | `string` | `[email protected]` | User email authorized to access the APIs | *required* |
+| `service_account_json` | `string` | `{"type": "service_account","project_id": "XXXXXX","private_key_id": "XXXXXX","private_key": "XXXXXX","client_email": "XXXXXX","client_id": "XXXXXX","auth_uri": "https://accounts.google.com/o/oauth2/auth","token_uri": "https://oauth2.googleapis.com/token","auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs","client_x509_cert_url": "XXXXXX"}` | Service Account JSON object | *required* |
+
+## Outputs
+
+| Field | Sample Value |
+| :---- | :---- |
+| `resource.urn` | `[email protected]` |
+| `resource.name` | `John Doe` |
+| `email` | `[email protected]` |
+| `full_name` | `John Doe` |
+| `status` | `not suspended` |
+| `properties` | `{"attributes":{"aliases":"[email protected],[email protected]","manager":"[email protected]","org_unit_path":"/"}}`
+
+### Notes
+ - The service account must have a [delegated domain wide authority](https://developers.google.com/admin-sdk/directory/v1/guides/delegation#delegate_domain-wide_authority_to_your_service_account)
+ - User Email : Only users with access to the Admin APIs can access the Admin SDK Directory API, therefore your service account needs to impersonate one of those users to access the Admin SDK Directory API.
+
+## Contributing
+
+Refer to the [contribution guidelines](../../../docs/contribute/guide.md#adding-a-new-extractor) for information on contributing to this module.

plugins/extractors/googleworkspace/googleworkspace.go

@@ -0,0 +1,218 @@
+package googleworkspace
+
+import (
+	"context"
+	_ "embed" // used to print the embedded assets
+	"fmt"
+	"reflect"
+	"strings"
+
+	"github.com/pkg/errors"
+
+	"github.com/odpf/meteor/models"
+	"github.com/odpf/meteor/plugins"
+	"github.com/odpf/meteor/registry"
+	"github.com/odpf/meteor/utils"
+	"github.com/odpf/salt/log"
+	"golang.org/x/oauth2"
+	"golang.org/x/oauth2/google"
+	admin "google.golang.org/api/admin/directory/v1"
+	"google.golang.org/api/option"
+	"google.golang.org/protobuf/types/known/anypb"
+
+	assetsv1beta2 "github.com/odpf/meteor/models/odpf/assets/v1beta2"
+)
+
+//go:embed README.md
+var summary string
+
+type Config struct {
+	ServiceAccountJSON string `mapstructure:"service_account_json" validate:"required"`
+	UserEmail          string `mapstructure:"user_email" validate:"required"`
+}
+
+var sampleConfig = `
+service_account_json: {
+    "type": "service_account",
+    "project_id": "XXXXXX",
+    "private_key_id": "XXXXXX",
+    "private_key": "XXXXXX",
+    "client_email": "XXXXXX",
+    "client_id": "XXXXXX",
+    "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+    "token_uri": "https://oauth2.googleapis.com/token",
+    "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
+    "client_x509_cert_url": "XXXXXX"
+}
+user_email: [email protected]`
+
+var info = plugins.Info{
+	Description:  "User list from Google Workspace",
+	SampleConfig: sampleConfig,
+	Tags:         []string{"platform", "extractor"},
+	Summary:      summary,
+}
+
+// Extractor manages the extraction of data from the extractor
+type Extractor struct {
+	plugins.BaseExtractor
+	logger      log.Logger
+	config      Config
+	TokenSource oauth2.TokenSource
+	Client      *admin.Service
+	emit        plugins.Emit
+}
+
+// New returns a pointer to an initialized Extractor Object
+func New(logger log.Logger) *Extractor {
+	e := &Extractor{
+		logger: logger,
+	}
+	e.BaseExtractor = plugins.NewBaseExtractor(info, &e.config)
+
+	return e
+}
+
+// Init initializes the extractor
+func (e *Extractor) Init(ctx context.Context, config plugins.Config) (err error) {
+	if err = e.BaseExtractor.Init(ctx, config); err != nil {
+		return err
+	}
+
+	jwtConfig, err := google.JWTConfigFromJSON([]byte(e.config.ServiceAccountJSON), admin.AdminDirectoryUserScope)
+	if err != nil {
+		return plugins.InvalidConfigError{}
+	}
+
+	if e.config.UserEmail == "" {
+		return plugins.InvalidConfigError{}
+	}
+	jwtConfig.Subject = e.config.UserEmail
+	ts := jwtConfig.TokenSource(ctx)
+
+	e.TokenSource = ts
+
+	return
+}
+
+// Extract extracts the data from the extractor
+// The data is returned as a list of assets.Asset
+func (e *Extractor) Extract(ctx context.Context, emit plugins.Emit) (err error) {
+	var status string
+	e.emit = emit
+	r, err := FetchUsers(ctx, e.TokenSource)
+	if err != nil {
+		return err
+	}
+
+	if len(r.Users) == 0 {
+		e.logger.Info("No users found.\n")
+		return nil
+	}
+
+	for _, u := range r.Users {
+		if !u.Suspended {
+			status = "not suspended"
+		} else {
+			status = "suspended"
+		}
+
+		var userAttributes = make(map[string]interface{})
+		userAttributes = buildOrgAttributes(userAttributes, u.Organizations)
+		userAttributes = buildRelationsAttributes(userAttributes, u.Relations)
+		userAttributes = buildCustomSchemasAttributes(userAttributes, u.CustomSchemas)
+		userAttributes["org_unit_path"] = u.OrgUnitPath
+		userAttributes["aliases"] = strings.Join(u.Aliases, ",")
+
+		user, err := anypb.New(&assetsv1beta2.User{
+			Email:      u.PrimaryEmail,
+			FullName:   u.Name.FullName,
+			Status:     status,
+			Attributes: utils.TryParseMapToProto(userAttributes),
+		})
+		if err != nil {
+			return err
+		}
+
+		e.emit(models.NewRecord(&assetsv1beta2.Asset{
+			Data: user,
+		}))
+	}
+
+	return nil
+}
+
+// init registers the extractor to catalog
+func init() {
+	if err := registry.Extractors.Register("googleworkspace", func() plugins.Extractor {
+		return New(plugins.GetLog())
+	}); err != nil {
+		panic(err)
+	}
+}
+
+func FetchUsers(ctx context.Context, ts oauth2.TokenSource) (*admin.Users, error) {
+	srv, err := admin.NewService(ctx, option.WithTokenSource(ts))
+	if err != nil {
+		return nil, err
+	}
+
+	r, err := srv.Users.List().Customer("my_customer").Do()
+	if err != nil {
+		return nil, errors.Wrap(err, "unable to retrieve users in domain")
+	}
+
+	return r, nil
+}
+
+func buildOrgAttributes(userAttributes map[string]interface{}, orgsIfc interface{}) map[string]interface{} {
+	if orgsIfc != nil {
+		orgs := reflect.ValueOf(orgsIfc)
+		if orgs.Kind() == reflect.Slice {
+			//based on assumpton that a user shall belong to a single org
+			org := reflect.ValueOf(orgs.Index(0).Interface())
+			for _, key := range org.MapKeys() {
+				value := org.MapIndex(key)
+				userAttributes[fmt.Sprintf("%v", key.Interface())] = value.Interface()
+			}
+		}
+	}
+	return userAttributes
+}
+
+func buildRelationsAttributes(userAttributes map[string]interface{}, relationsIfc interface{}) map[string]interface{} {
+	if relationsIfc != nil {
+		relations := reflect.ValueOf(relationsIfc)
+		if relations.Kind() == reflect.Slice {
+			for idx := 0; idx < relations.Len(); idx++ {
+				var relationType, relationValue string
+
+				relation := reflect.ValueOf(relations.Index(idx).Interface())
+				for _, key := range relation.MapKeys() {
+					value := relation.MapIndex(key)
+
+					if key.Interface().(string) == "type" {
+						relationType = value.Interface().(string)
+					} else if key.Interface().(string) == "value" {
+						relationValue = value.Interface().(string)
+					}
+				}
+				userAttributes[relationType] = relationValue
+			}
+		}
+	}
+	return userAttributes
+}
+
+func buildCustomSchemasAttributes(userAttributes map[string]interface{}, customSchemasIfc interface{}) map[string]interface{} {
+	if customSchemasIfc != nil {
+		customSchema := reflect.ValueOf(customSchemasIfc)
+		if customSchema.Kind() == reflect.Map {
+			for _, key := range customSchema.MapKeys() {
+				value := customSchema.MapIndex(key)
+				userAttributes[fmt.Sprintf("%v", key.Interface())] = value.Interface()
+			}
+		}
+	}
+	return userAttributes
+}

plugins/extractors/googleworkspace/googleworkspace_test.go

@@ -0,0 +1,102 @@
+//go:build plugins
+// +build plugins
+
+package googleworkspace_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/odpf/meteor/models"
+	assetsv1beta2 "github.com/odpf/meteor/models/odpf/assets/v1beta2"
+	"github.com/odpf/meteor/plugins"
+	"github.com/odpf/meteor/plugins/extractors/googleworkspace"
+	"github.com/odpf/meteor/test/mocks"
+	"github.com/odpf/meteor/test/utils"
+	utilities "github.com/odpf/meteor/utils"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"google.golang.org/protobuf/types/known/anypb"
+)
+
+var urnScope string = "test-googleworkspace"
+
+func TestInit(t *testing.T) {
+	t.Run("should return error for empty user email", func(t *testing.T) {
+		err := googleworkspace.New(utils.Logger).Init(context.TODO(), plugins.Config{
+			URNScope: urnScope,
+			RawConfig: map[string]interface{}{
+				"user_email":           "",
+				"service_account_json": "{\"type\":\"service_account\",\"project_id\":\"odpf-meteor\",\"private_key_id\":\"3cb2103ef7883845a5fdcsvdefe6ff83d616757\",\"private_key\":\"-----BEGIN PRIVATE KEY-----\\nMIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggdvdvdEAAoIBAQDF/cDQ++JnH9+9\\n3YBm4APqPbvfj6eHSdAUSjzKdbdfbdbYGgdxC7xPS1PVo+ENw+pBAH3NoRwQWYEin\\nHYj064sMvm8vbR5TcMQpnxYG86TGaPuIh30grz5dI39dtrUjttbdfbdvqRv0qu7I5\\nuELzp2OLUz509Q3AvuqvQVCZc7sDjNr2TPOsLeuCkpmcmBHbdfdi29bhoS+Ac\\n5ipT10yGF0FvT1f5KlJcHfsNoOGPJYePTaGxOW1zk680Z1Wdfbdf1xX9iw5/GUA3XM\\neon4p9X31ASgwbdbdplFZhwvcpoaYpxcuxyvefR44emnfveUY91h6wLvF/mPBElO\\npXOiVJ3lAgMBAAECggEbdbddYz8nSmTWFMW2OtyvojIq+ab864ZGPCpW4zfzF4BI\\n7o5TSIsNOMQMrawFUz0xZkgofJThfOscyXbbdbdbfbfT3wXI9JTWT8l275ssvFQVy1\\nVyAJI/Kize9ru5GnnEzV2sZoYEmOsB2xgqjvKXR90r5wNJ6wFp8Ubp9/+v2lTv1n\\nUCBBYPsPyVmUq677HfMVVa6ZpxCTWvbQga+/ZPaqppgGps5yLDqc434c3A/lDCKBtqk\\njaQXHqKjuYUsoiyl2vbPbwGxc34343c6gQfe7aeCouf8bI4GzCPmoyVPMRFpQJ6Ahp\\nMnCE96KfVVUARh1goxEEwMmSFyBPYFbmvXLPUGNfcQKBgQD3nrDHeWxW+0MjnaYD\\novXKvpnv1NiBCywOAEfc343535dJfgMZX0cfpnTDGXKPBI5ZbUywxk0sewu382JoArM\\n1w2wEIqH+73FGiMVpAuN2DpNX5mOC+z/zjFdOFZ28jkRUy8T+PTkajj7rkB7VDOr\\nIiCZwRrnbQFwhErWS1fZgg2PcQKBgQDMsRgDBfhgJX9sNRX3FHzIEZU94PP1KOc2\\nEUUzcwIV0cNOVzSyOUn2qrcYNg/hZZpGeRBBwyOcDGsqxmz5FAzk0OtbSCaMxybF\\n8NXFDh3ELmnfIyVBjvNBWPckcR1LCZcKGTqVLH/rhPiNhyzH3NQ0c3Gl15GPgzkD\\nboLfFN3jtQKBgG++blpmYkzScNb2wr9rX+5Rm1hOvjFl4EilOb+1rq/WPZ0ig5ZD\\nT5mdQ6ZC+5ppWp8AyjQsgsAYgUG1NoqAFg45OLrrERWMmP6gHBKz3IOkO8CNgzNh\\nUoeV7/cXkkdOObWSqLkXcoWpejHtqq905C9epIyBdZ/YI4mXU3c4343c4QRAoGBAK9F\\nMO9dzFjfouVP63f/Nf3GeIlctuiE1r5IOX4di3qNe/P33iqBvaCWe2Mi36Q78MdJ\\nYK8+3Z4AUD93WtZI4eWIMw+dj0zaNowldZZfSQO0Tnl/yaYCNq8M88pjhRa8pnVC\\nNxSG3x4XZREi3yhgIeCrvXOpS32celRC65MDdiBFAoGAHbURTEkQDZaWPAmVv+0q\\nYaT7x+UzQDGKy/By9QLGM/U2gvLGTw1vzmoeh99BTsQopPB/QuAfJNIHk9h0ohXJ\\nfA/X4T3F2LGhZ9+bujVyCQc0tTxuh41t2ipJPWtDP52rXk1AkCnIeWD+UHI0u5Ba\\nhI1dzLIxZKeq3bESrc/9tmM=\\n-----END PRIVATE KEY-----\\n\",\"client_email\":\"[email protected]\",\"client_id\":\"110059943435984635286\",\"auth_uri\":\"https://accounts.google.com/o/oauth2/auth\",\"token_uri\":\"https://oauth2.googleapis.com/token\",\"auth_provider_x509_cert_url\":\"https://www.googleapis.com/oauth2/v1/certs\",\"client_x509_cert_url\":\"https://www.googleapis.com/robot/v1/metadata/x509/meteor-sa%40odpf-meteor.iam.gserviceaccount.com\"}",
+			}})
+		assert.ErrorAs(t, err, &plugins.InvalidConfigError{})
+	})
+	t.Run("should return error for invalid service account json", func(t *testing.T) {
+		err := googleworkspace.New(utils.Logger).Init(context.TODO(), plugins.Config{
+			URNScope: urnScope,
+			RawConfig: map[string]interface{}{
+				"user_email":           "[email protected]",
+				"service_account_json": "invalide json",
+			}})
+		assert.ErrorAs(t, err, &plugins.InvalidConfigError{})
+	})
+}
+
+func TestExtract(t *testing.T) {
+	t.Run("should extract user details from google workspace", func(t *testing.T) {
+
+		expectedData := []models.Record{
+			models.NewRecord(&assetsv1beta2.Asset{
+				Data: userToAny(assetsv1beta2.User{
+					Email:    "[email protected]",
+					FullName: "John Doe",
+					Status:   "not suspended",
+					Attributes: utilities.TryParseMapToProto(map[string]interface{}{
+						"manager": "[email protected]",
+					}),
+				}),
+			}),
+			models.NewRecord(&assetsv1beta2.Asset{
+				Data: userToAny(assetsv1beta2.User{
+					Email:    "[email protected]",
+					FullName: "Ipsum Lorum",
+					Status:   "not suspended",
+					Attributes: utilities.TryParseMapToProto(map[string]interface{}{
+						"manager": "[email protected]",
+					}),
+				}),
+			}),
+		}
+
+		ctx := context.TODO()
+
+		emitter := mocks.NewEmitter()
+		extractor := mocks.NewExtractor()
+		extractor.On("Init", mock.AnythingOfType("*context.emptyCtx"), mock.AnythingOfType("plugins.Config")).Return(nil)
+		extractor.SetEmit(expectedData)
+		extractor.On("Extract", mock.AnythingOfType("*context.emptyCtx"), mock.AnythingOfType("plugins.Emit")).Return(nil)
+
+		err := extractor.Init(ctx, plugins.Config{
+			URNScope: urnScope,
+			RawConfig: map[string]interface{}{
+				"user_email":           "[email protected]",
+				"service_account_json": "{\"type\":\"service_account\",\"project_id\":\"odpf\",\"private_key_id\":\"3cb2103ef7883455a5f09712befe6ff83d616757\",\"private_key\":\"-----BEGIN PRIVATE KEY-----\\nMIIEvQIBGDSNBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDF/cDQ++JnH9+9\\n3YBm4APqPbvfj6eHSdASSjzKden0lgYGgdxC7xPS1PVo+ENw+pBAH3NoRwQWYEin\\nHYj064sMvm8vbR5TcMQpnxYG8SAGaPuIh30grz5dI39dtrUjttWWvtvqRv0qu7I5\\nuELzp2OLUz509Q3AvuqvQVCZc7sDjNr2TPOsLeuCkpmcmFDyNdOa6vi29bhoS+Ac\\n5ipT10yGF0FvT1f5KlJcHfsNoOGPJYeTRaGxOW1zk680Z1WFyB1xX9iw5/GUA3XM\\neon4p9X31ASgw6WTqplGDhwvcpoaYpxcuxyvefR44emnfveUY91h6wLvF/mPBElO\\npXOiVJ3lAgMBAAECggEALZwVYz8nSmTWFMW2OtyvojIq+ab864BGHFpW4zfzF4BI\\n7o5TSIsNOMQMrawFUz0xZkgofJJHhfOscyXydDHjHXT3wXI9JTWT8l275ssvFQVy1\\nVyAJI/Kize9ru5GnnEzV2sZoTYmOsB2xgqjvKXR90r5wNJ6wFp8Ubp9/+v2lTv1n\\nUCBBYPsPyVmUq677HfMBBa6ZpxCTWvbQga+/ZPaqppgGps5yLDqcp1A/lDCKBtqk\\njaQXHqKjuYDSsoiyl2vbPbwGxIzYSv6gQfe7aeCouf8bI4GzCPmoyVPMRFpQJ6Ahp\\nMnCE96KfVVUARh1goxEEwMmSFyBPYFbmvXLPUGNfcQKBgQD3nrDHeWxW+0MjnaYD\\novXKvpnv1NiBCywOAEfIhxadJfgMZX0cfpnTDGXKPBI5ZbUywxk0sewu382JoArM\\n1w2wEIqH+73FGiMVpAuN2DpNX5mOC+z/zjFdOFZ28jkRUy8T+PTkajj7rkB7VDOr\\nIiCZwRrnbQFwhErWS1fZgg2PcQKBgQDMsRgDBfhgJX9sNRX3FHzIEZU94PP1KOc2\\nEUUzcwIV0cNOVzSyOUn2qrcYNg/hZZpGeRBBwyOcDGsqxmz5FAzk0OtbSCaMxybF\\n8NXFDh3ELmnfIyVBjvNBWPckcR1LCZcKGTqVLH/rhPiNhyzH3NQ0c3Gl15GPgzkD\\nboLfFN3jtQKBgG++blpmYkzScNb2wr9rX+5Rm1hOvjFl4EilOb+1rq/WPZ0ig5ZD\\nT5mdQ6ZC+5ppWp8AyjQsgsAYgUG1NoqAFg45OLrrERWMmP6gHBKz3IOkO8CNgzNh\\nUoeV7/cXkkdOObWSqLkXcoWpejHtqq905C9epIyBdZ/YI4mXUJq4hPQRAoGBAK9F\\nMO9dzFjfouVP63f/Nf3GeIlctuiE1r5IOX4di3qNe/P33iqBvaCWe2Mi36Q78MdJ\\nYK8+3Z4AUD93WtZI4eWIMw+dj0zaNowldZZfSQO0Tnl/yaYCNq8M88pjhRa8pnVC\\nNxSG3x4XZREi3yhgIeCrvXOpS32celRC65MDdiBFAoGAHbURTEkQDZaWPAmVv+0q\\nYaT7x+UzQDGKy/By9QLGM/U2gvLGTw1vzmoeh99BTsQopPB/QuAfJNIHk9h0ohXJ\\nfA/X4T3F2LGhZ9+bujVyCQc0tTxuh41t2ipJPWtDP52rXk1AkCnIeWD+UHI0u5Ba\\nhI1dzLIxZKeq3bESrc/9tmM=\\n-----END PRIVATE KEY-----\\n\",\"client_email\":\"[email protected]\",\"client_id\":\"110059957285984635286\",\"auth_uri\":\"https://accounts.google.com/o/oauth2/auth\",\"token_uri\":\"https://oauth2.googleapis.com/token\",\"auth_provider_x509_cert_url\":\"https://www.googleapis.com/oauth2/v1/certs\",\"client_x509_cert_url\":\"https://www.googleapis.com/robot/v1/metadata/x509/meteor-sa%40odpf.iam.gserviceaccount.com\"}",
+			},
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = extractor.Extract(ctx, emitter.Push)
+
+		assert.NoError(t, err)
+		assert.EqualValues(t, expectedData, emitter.Get())
+	})
+}
+
+func userToAny(user assetsv1beta2.User) *anypb.Any {
+	u, err := anypb.New(&user)
+	if err != nil {
+		return &anypb.Any{}
+	}
+	return u
+}

plugins/extractors/populate.go

@@ -10,6 +10,7 @@ import (
 	_ "github.com/odpf/meteor/plugins/extractors/elastic"
 	_ "github.com/odpf/meteor/plugins/extractors/gcs"
 	_ "github.com/odpf/meteor/plugins/extractors/github"
+	_ "github.com/odpf/meteor/plugins/extractors/googleworkspace"
 	_ "github.com/odpf/meteor/plugins/extractors/grafana"
 	_ "github.com/odpf/meteor/plugins/extractors/kafka"
 	_ "github.com/odpf/meteor/plugins/extractors/mariadb"