diff --git a/ingestion/examples/sample_data/policies/access_control.json b/ingestion/examples/sample_data/policies/access_control.json new file mode 100644 index 00000000000..3824625675b --- /dev/null +++ b/ingestion/examples/sample_data/policies/access_control.json @@ -0,0 +1,58 @@ +{ + "policies": [ + { + "name": "data-steward-role", + "displayName": "Data Steward Role Policy", + "description": "Policy for Data Steward Role to perform operations on metadata entities", + "policyType": "AccessControl", + "enabled": true, + "rules": [ + { + "name": "update-description", + "userRoleAttr": "DataSteward", + "operation": "UpdateDescription", + "allow": true + }, + { + "name": "update-tags", + "userRoleAttr": "DataSteward", + "operation": "UpdateTags", + "allow": true + } + ] + }, + { + "name": "data-consumer-role", + "displayName": "Data Consumer Role Policy", + "description": "Policy for Data Consumer Role to perform operations on metadata entities", + "policyType": "AccessControl", + "enabled": true, + "rules": [ + { + "name": "suggest-description", + "userRoleAttr": "DataConsumer", + "operation": "SuggestDescription", + "allow": true + }, + { + "name": "suggest-tags", + "userRoleAttr": "DataConsumer", + "operation": "SuggestTags", + "allow": true + }, + { + "name": "update-description", + "userRoleAttr": "DataConsumer", + "operation": "UpdateDescription", + "allow": false + }, + { + "name": "update-tags", + "userRoleAttr": "DataConsumer", + "operation": "UpdateTags", + "allow": false + } + ] + } + ] +} \ No newline at end of file diff --git a/ingestion/examples/workflows/access_control.json b/ingestion/examples/workflows/access_control.json new file mode 100644 index 00000000000..29e1df5a6b2 --- /dev/null +++ b/ingestion/examples/workflows/access_control.json @@ -0,0 +1,19 @@ +{ + "source": { + "type": "access_control_policies", + "config": { + "policies_file": "ingestion/examples/sample_data/policies/access_control.json" + } + }, + "sink": { + "type": "metadata-rest", + "config": {} + }, + "metadata_server": { + "type": "metadata-server", + "config": { + "api_endpoint": "http://localhost:8585/api", + "auth_provider_type": "no-auth" + } + } +} \ No newline at end of file diff --git a/ingestion/src/metadata/ingestion/source/access_control_policies.py b/ingestion/src/metadata/ingestion/source/access_control_policies.py new file mode 100644 index 00000000000..42607d6e68f --- /dev/null +++ b/ingestion/src/metadata/ingestion/source/access_control_policies.py @@ -0,0 +1,76 @@ +# Copyright 2021 Collate +# Licensed 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. + +import json +import logging +import uuid +from collections import Iterable +from typing import Optional + +from metadata.generated.schema.entity.policies.policy import Policy +from metadata.ingestion.api.common import Entity, ConfigModel, WorkflowContext +from metadata.ingestion.api.source import Source, SourceStatus +from metadata.ingestion.models.ometa_policy import OMetaPolicy +from metadata.ingestion.ometa.openmetadata_rest import MetadataServerConfig + +logger: logging.Logger = logging.getLogger(__name__) + + +class AccessControlPoliciesConfig(ConfigModel): + policies_file: str + + +class AccessControlPoliciesSource(Source[Entity]): + config: AccessControlPoliciesConfig + status: SourceStatus + policies_data: Optional[dict] = None + + def __init__( + self, + config: AccessControlPoliciesConfig, + metadata_config: MetadataServerConfig, + ctx, + ): + self.config = config + self.status = SourceStatus() + + @classmethod + def create( + cls, config_dict: dict, metadata_config_dict: dict, ctx: WorkflowContext + ): + config = AccessControlPoliciesConfig.parse_obj(config_dict) + metadata_config = MetadataServerConfig.parse_obj(metadata_config_dict) + return cls(config, metadata_config, ctx) + + def prepare(self): + try: + with open(self.config.policies_file, "r") as f: + self.policies_data = json.load(f) + except Exception as e: + logger.fatal( + f"Please ensure that the configured policies_file is set up correctly - {e}" + ) + + def next_record(self) -> Iterable[OMetaPolicy]: + try: + for policy in self.policies_data["policies"]: + # add a generated policy id to reduce overhead of maintaining one for every policy in policies file. + policy["id"] = uuid.uuid4() + self.status.scanned(policy) + yield OMetaPolicy(policy=Policy.parse_obj(policy)) + except Exception as e: + self.status.failure("error", str(e)) + + def get_status(self) -> SourceStatus: + return self.status + + def close(self): + pass # nothing to close. diff --git a/ingestion/src/metadata/ingestion/source/sample_data.py b/ingestion/src/metadata/ingestion/source/sample_data.py index 019a05b162d..6461d9abdad 100644 --- a/ingestion/src/metadata/ingestion/source/sample_data.py +++ b/ingestion/src/metadata/ingestion/source/sample_data.py @@ -26,7 +26,6 @@ from metadata.generated.schema.api.data.createTopic import CreateTopicEntityRequ from metadata.generated.schema.api.lineage.addLineage import AddLineage from metadata.generated.schema.entity.data.database import Database from metadata.generated.schema.entity.data.location import Location, LocationType -from metadata.generated.schema.entity.data.mlmodel import MlModel from metadata.generated.schema.entity.data.pipeline import Pipeline from metadata.generated.schema.entity.data.table import Table from metadata.generated.schema.entity.teams.user import User @@ -39,6 +38,10 @@ from metadata.ingestion.models.ometa_table_db import OMetaDatabaseAndTable from metadata.ingestion.models.table_metadata import Chart, Dashboard from metadata.ingestion.ometa.ometa_api import OpenMetadata from metadata.ingestion.ometa.openmetadata_rest import MetadataServerConfig +from metadata.ingestion.source.access_control_policies import ( + AccessControlPoliciesConfig, + AccessControlPoliciesSource, +) from metadata.utils.helpers import ( get_dashboard_service_or_create, get_database_service_or_create, @@ -261,6 +264,13 @@ class SampleDataSource(Source[Entity]): self.models = json.load( open(self.config.sample_data_folder + "/models/models.json", "r") ) + policies_config = AccessControlPoliciesConfig( + policies_file=self.config.sample_data_folder + + "/policies/access_control.json" + ) + self.policies_source = AccessControlPoliciesSource( + policies_config, metadata_config, ctx + ) @classmethod def create(cls, config_dict, metadata_config_dict, ctx): @@ -269,7 +279,7 @@ class SampleDataSource(Source[Entity]): return cls(config, metadata_config, ctx) def prepare(self): - pass + self.policies_source.prepare() def next_record(self) -> Iterable[Entity]: yield from self.ingest_locations() @@ -282,6 +292,7 @@ class SampleDataSource(Source[Entity]): yield from self.ingest_lineage() yield from self.ingest_users() yield from self.ingest_mlmodels() + yield from self.policies_source.next_record() def ingest_locations(self) -> Iterable[Location]: for location in self.locations["locations"]: