feat(postgres): extract table access for users (#339)

* feat(postgres): extract table access grants

* feat: return privileges as a array

* fix: missing build plugins tag

* refactor: move split string func to extractor

plugins/extractors/postgres/postgres.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 
 	"github.com/pkg/errors"
+	"google.golang.org/protobuf/types/known/structpb"
 
 	// used to register the postgres driver
 	_ "github.com/lib/pq"
@@ -163,6 +164,11 @@ func (e *Extractor) getTableMetadata(db *sql.DB, dbName string, tableName string
 		return result, nil
 	}
 
+	usrPrivilegeInfo, err := e.userPrivilegesInfo(db, dbName, tableName)
+	if err != nil {
+		return result, nil
+	}
+
 	result = &assetsv1beta1.Table{
 		Resource: &commonv1beta1.Resource{
 			Urn:     models.TableURN("postgres", e.host, dbName, tableName),
@@ -173,6 +179,9 @@ func (e *Extractor) getTableMetadata(db *sql.DB, dbName string, tableName string
 		Schema: &facetsv1beta1.Columns{
 			Columns: columns,
 		},
+		Properties: &facetsv1beta1.Properties{
+			Attributes: usrPrivilegeInfo,
+		},
 	}
 
 	return
@@ -206,6 +215,39 @@ func (e *Extractor) getColumnMetadata(db *sql.DB, dbName string, tableName strin
 	return result, nil
 }
 
+func (e *Extractor) userPrivilegesInfo(db *sql.DB, dbName string, tableName string) (result *structpb.Struct, err error) {
+	query := `SELECT grantee, string_agg(privilege_type, ',') 
+	FROM information_schema.role_table_grants 
+	WHERE table_name='%s' AND table_catalog='%s'
+	GROUP BY grantee;`
+
+	rows, err := db.Query(fmt.Sprintf(query, tableName, dbName))
+	if err != nil {
+		err = errors.Wrap(err, "failed to fetch data from query")
+		return
+	}
+
+	var usrs []interface{}
+	for rows.Next() {
+		var grantee, privilege_type string
+
+		if err = rows.Scan(&grantee, &privilege_type); err != nil {
+			e.logger.Error("failed to get fields", "error", err)
+			continue
+		}
+
+		usrs = append(usrs, map[string]interface{}{
+			"user":            grantee,
+			"privilege_types": ConvertStringListToInterface(strings.Split(privilege_type, ",")),
+		})
+	}
+	grants := map[string]interface{}{
+		"grants": usrs,
+	}
+	result = utils.TryParseMapToProto(grants)
+	return
+}
+
 // Convert nullable string to a boolean
 func isNullable(value string) bool {
 	return value == "YES"
@@ -248,3 +290,11 @@ func init() {
 		panic(err)
 	}
 }
+
+func ConvertStringListToInterface(s []string) []interface{} {
+	out := make([]interface{}, len(s))
+	for i, v := range s {
+		out[i] = v
+	}
+	return out
+}

plugins/extractors/postgres/postgres_test.go

@@ -10,8 +10,12 @@ import (
 	"os"
 	"testing"
 
+	"github.com/odpf/meteor/models"
+	commonv1beta1 "github.com/odpf/meteor/models/odpf/assets/common/v1beta1"
+	facetsv1beta1 "github.com/odpf/meteor/models/odpf/assets/facets/v1beta1"
 	assetsv1beta1 "github.com/odpf/meteor/models/odpf/assets/v1beta1"
 	"github.com/odpf/meteor/test/utils"
+	ut "github.com/odpf/meteor/utils"
 
 	"database/sql"
 
@@ -99,13 +103,7 @@ func TestExtract(t *testing.T) {
 		err = extr.Extract(ctx, emitter.Push)
 		require.NoError(t, err)
 
-		var urns []string
-		for _, record := range emitter.Get() {
-			table := record.Data().(*assetsv1beta1.Table)
-			urns = append(urns, table.Resource.Urn)
-
-		}
-		assert.Equal(t, []string{"postgres::localhost:5438/test_db/article", "postgres::localhost:5438/test_db/post"}, urns)
+		assert.Equal(t, getExpected(), emitter.Get())
 	})
 }
 
@@ -145,3 +143,76 @@ func execute(db *sql.DB, queries []string) (err error) {
 	}
 	return
 }
+
+func getExpected() []models.Record {
+	return []models.Record{
+		models.NewRecord(&assetsv1beta1.Table{
+			Resource: &commonv1beta1.Resource{
+				Urn:     "postgres::localhost:5438/test_db/article",
+				Name:    "article",
+				Service: "postgres",
+				Type:    "table",
+			},
+			Schema: &facetsv1beta1.Columns{
+				Columns: []*facetsv1beta1.Column{
+					{
+						Name:       "id",
+						DataType:   "bigint",
+						IsNullable: false,
+						Length:     0,
+					},
+					{
+						Name:       "name",
+						DataType:   "character varying",
+						IsNullable: false,
+						Length:     20,
+					},
+				},
+			},
+			Properties: &facetsv1beta1.Properties{
+				Attributes: ut.TryParseMapToProto(map[string]interface{}{
+					"grants": []interface{}{
+						map[string]interface{}{
+							"user":            "test_user",
+							"privilege_types": []interface{}{"INSERT", "SELECT", "UPDATE", "DELETE", "TRUNCATE", "REFERENCES", "TRIGGER"},
+						},
+					},
+				}),
+			},
+		}),
+		models.NewRecord(&assetsv1beta1.Table{
+			Resource: &commonv1beta1.Resource{
+				Urn:     "postgres::localhost:5438/test_db/post",
+				Name:    "post",
+				Service: "postgres",
+				Type:    "table",
+			},
+			Schema: &facetsv1beta1.Columns{
+				Columns: []*facetsv1beta1.Column{
+					{
+						Name:       "id",
+						DataType:   "bigint",
+						IsNullable: false,
+						Length:     0,
+					},
+					{
+						Name:       "title",
+						DataType:   "character varying",
+						IsNullable: false,
+						Length:     20,
+					},
+				},
+			},
+			Properties: &facetsv1beta1.Properties{
+				Attributes: ut.TryParseMapToProto(map[string]interface{}{
+					"grants": []interface{}{
+						map[string]interface{}{
+							"user":            "test_user",
+							"privilege_types": []interface{}{"INSERT", "SELECT", "UPDATE", "DELETE", "TRUNCATE", "REFERENCES", "TRIGGER"},
+						},
+					},
+				}),
+			},
+		}),
+	}
+}