diff --git a/backend/dataall/core/environment/__init__.py b/backend/dataall/core/environment/__init__.py
index 7007ae59a..eda70ff55 100644
--- a/backend/dataall/core/environment/__init__.py
+++ b/backend/dataall/core/environment/__init__.py
@@ -1,3 +1,3 @@
 """The central package of the application to work with the environment"""
 
-from dataall.core.environment import api, cdk, tasks
+from dataall.core.environment import api, db, cdk, tasks
diff --git a/backend/dataall/core/environment/api/input_types.py b/backend/dataall/core/environment/api/input_types.py
index 076416043..6add6346b 100644
--- a/backend/dataall/core/environment/api/input_types.py
+++ b/backend/dataall/core/environment/api/input_types.py
@@ -1,6 +1,6 @@
 from dataall.base.api import gql
 from dataall.base.api.constants import GraphQLEnumMapper, SortDirection
-
+from dataall.core.environment.db.environment_enums import PolicyManagementOptions
 
 AwsEnvironmentInput = gql.InputType(
     name='AwsEnvironmentInput',
@@ -101,7 +101,7 @@ class EnvironmentSortField(GraphQLEnumMapper):
         gql.Argument('groupUri', gql.NonNullableType(gql.String)),
         gql.Argument('IAMRoleArn', gql.NonNullableType(gql.String)),
         gql.Argument('environmentUri', gql.NonNullableType(gql.String)),
-        gql.Argument('dataallManaged', gql.NonNullableType(gql.Boolean)),
+        gql.Argument('dataallManaged', gql.NonNullableType(PolicyManagementOptions.toGraphQLEnum())),
     ],
 )
 
@@ -120,5 +120,6 @@ class EnvironmentSortField(GraphQLEnumMapper):
     arguments=[
         gql.Argument('consumptionRoleName', gql.String),
         gql.Argument('groupUri', gql.String),
+        gql.Argument('dataallManaged', gql.NonNullableType(PolicyManagementOptions.toGraphQLEnum())),
     ],
 )
diff --git a/backend/dataall/core/environment/api/types.py b/backend/dataall/core/environment/api/types.py
index 4479b7bab..487965889 100644
--- a/backend/dataall/core/environment/api/types.py
+++ b/backend/dataall/core/environment/api/types.py
@@ -8,7 +8,7 @@
     resolve_user_role,
 )
 from dataall.core.environment.api.enums import EnvironmentPermission
-
+from dataall.core.environment.db.environment_enums import PolicyManagementOptions
 
 EnvironmentUserPermission = gql.ObjectType(
     name='EnvironmentUserPermission',
@@ -163,7 +163,7 @@
         gql.Field(name='policy_name', type=gql.String),
         gql.Field(name='policy_type', type=gql.String),
         gql.Field(name='exists', type=gql.Boolean),
-        gql.Field(name='attached', type=gql.Boolean),
+        gql.Field(name='attached', type=gql.String),
     ],
 )
 
@@ -176,7 +176,7 @@
         gql.Field(name='environmentUri', type=gql.String),
         gql.Field(name='IAMRoleArn', type=gql.String),
         gql.Field(name='IAMRoleName', type=gql.String),
-        gql.Field(name='dataallManaged', type=gql.Boolean),
+        gql.Field(name='dataallManaged', type=gql.NonNullableType(PolicyManagementOptions.toGraphQLEnum())),
         gql.Field(name='created', type=gql.String),
         gql.Field(name='updated', type=gql.String),
         gql.Field(name='deleted', type=gql.String),
diff --git a/backend/dataall/core/environment/cdk/environment_stack.py b/backend/dataall/core/environment/cdk/environment_stack.py
index c154bbd52..fd174975c 100644
--- a/backend/dataall/core/environment/cdk/environment_stack.py
+++ b/backend/dataall/core/environment/cdk/environment_stack.py
@@ -451,32 +451,33 @@ def create_group_environment_role(self, group: EnvironmentGroup, id: str):
             permissions=group_permissions,
         ).generate_policies()
 
-        external_managed_policies = []
-        policy_manager = PolicyManager(
-            environmentUri=self._environment.environmentUri,
-            resource_prefix=self._environment.resourcePrefix,
-            role_name=group.environmentIAMRoleName,
-            account=self._environment.AwsAccountId,
-            region=self._environment.region,
-        )
-        try:
-            managed_policies = policy_manager.get_all_policies()
-        except Exception as e:
-            logger.info(f'Not adding any managed policy because of exception: {e}')
-            # Known exception raised in first deployment because pivot role does not exist and cannot be assumed
-            managed_policies = []
-        for policy in managed_policies:
-            # If there is a managed policy that exist it should be attached to the IAM role
-            if policy.get('exists', False):
-                external_managed_policies.append(
-                    iam.ManagedPolicy.from_managed_policy_name(
-                        self,
-                        id=f'{self._environment.resourcePrefix}-managed-policy-{policy.get("policy_name")}',
-                        managed_policy_name=policy.get('policy_name'),
+        with self.engine.scoped_session() as session:
+            external_managed_policies = []
+            policy_manager = PolicyManager(
+                session=session,
+                account=self._environment.AwsAccountId,
+                region=self._environment.region,
+                environmentUri=self._environment.environmentUri,
+                resource_prefix=self._environment.resourcePrefix,
+                role_name=group.environmentIAMRoleName,
+            )
+            try:
+                managed_policies = policy_manager.get_all_policies()
+            except Exception as e:
+                logger.info(f'Not adding any managed policy because of exception: {e}')
+                # Known exception raised in first deployment because pivot role does not exist and cannot be assumed
+                managed_policies = []
+            for policy in managed_policies:
+                # If there is a managed policy that exist it should be attached to the IAM role
+                if policy.get('exists', False):
+                    external_managed_policies.append(
+                        iam.ManagedPolicy.from_managed_policy_name(
+                            self,
+                            id=f'{self._environment.resourcePrefix}-managed-policy-{policy.get("policy_name")}',
+                            managed_policy_name=policy.get('policy_name'),
+                        )
                     )
-                )
 
-        with self.engine.scoped_session() as session:
             data_policy = S3Policy(
                 stack=self,
                 tag_key='Team',
diff --git a/backend/dataall/core/environment/db/__init__.py b/backend/dataall/core/environment/db/__init__.py
index e69de29bb..8036795be 100644
--- a/backend/dataall/core/environment/db/__init__.py
+++ b/backend/dataall/core/environment/db/__init__.py
@@ -0,0 +1 @@
+from . import environment_enums, environment_models, environment_repositories
diff --git a/backend/dataall/core/environment/db/environment_enums.py b/backend/dataall/core/environment/db/environment_enums.py
new file mode 100644
index 000000000..8ac3a3193
--- /dev/null
+++ b/backend/dataall/core/environment/db/environment_enums.py
@@ -0,0 +1,7 @@
+from dataall.base.api import GraphQLEnumMapper
+
+
+class PolicyManagementOptions(GraphQLEnumMapper):
+    FULLY_MANAGED = 'Fully-Managed'
+    PARTIALLY_MANAGED = 'Partially-Managed'
+    EXTERNALLY_MANAGED = 'Externally-Managed'
diff --git a/backend/dataall/core/environment/db/environment_models.py b/backend/dataall/core/environment/db/environment_models.py
index 7f5e9a122..15bd19c55 100644
--- a/backend/dataall/core/environment/db/environment_models.py
+++ b/backend/dataall/core/environment/db/environment_models.py
@@ -107,7 +107,7 @@ class ConsumptionRole(Base):
     groupUri = Column(String, nullable=False)
     IAMRoleName = Column(String, nullable=False)
     IAMRoleArn = Column(String, nullable=False)
-    dataallManaged = Column(Boolean, nullable=False, default=True)
+    dataallManaged = Column(String, nullable=False)
     created = Column(DateTime, default=datetime.datetime.now)
     updated = Column(DateTime, onupdate=datetime.datetime.now)
     deleted = Column(DateTime)
diff --git a/backend/dataall/core/environment/db/environment_repositories.py b/backend/dataall/core/environment/db/environment_repositories.py
index 5d787f7a3..1801b7142 100644
--- a/backend/dataall/core/environment/db/environment_repositories.py
+++ b/backend/dataall/core/environment/db/environment_repositories.py
@@ -160,7 +160,6 @@ def query_user_environment_consumption_roles(session, groups, uri, filter) -> Qu
                 )
             )
         if filter and filter.get('groupUri'):
-            print('filter group')
             group = filter['groupUri']
             query = query.filter(
                 or_(
diff --git a/backend/dataall/core/environment/services/environment_service.py b/backend/dataall/core/environment/services/environment_service.py
index b064d759f..9645943fb 100644
--- a/backend/dataall/core/environment/services/environment_service.py
+++ b/backend/dataall/core/environment/services/environment_service.py
@@ -13,6 +13,7 @@
 from dataall.base.aws.sts import SessionHelper
 from dataall.base.context import get_context
 from dataall.base.db.exceptions import AWSResourceNotFound
+from dataall.core.environment.db.environment_enums import PolicyManagementOptions
 from dataall.core.organizations.db.organization_repositories import OrganizationRepository
 from dataall.core.permissions.services.environment_permissions import (
     ENABLE_ENVIRONMENT_SUBSCRIPTIONS,
@@ -46,6 +47,7 @@
 from dataall.core.permissions.services.tenant_permissions import MANAGE_ENVIRONMENTS
 from dataall.core.stacks.db.stack_repositories import StackRepository
 from dataall.core.vpc.db.vpc_repositories import VpcRepository
+from dataall.modules.shares_base.services.shares_enums import PrincipalType
 
 log = logging.getLogger(__name__)
 
@@ -397,15 +399,21 @@ def invite_group(uri, data=None) -> (Environment, EnvironmentGroup):
                 env_group_iam_role_arn = f'arn:aws:iam::{environment.AwsAccountId}:role/{env_group_iam_role_name}'
                 env_role_imported = False
 
-            # If environment role is imported, then data.all should attach the policies at import time
-            # If environment role is created in environment stack, then data.all should attach the policies in the env stack
+            # If environment role is imported, then data.all should attach the policies at import time ( Fully Managed )
+            # If environment role is created in environment stack, then data.all should attach the policies in the env stack ( Partially Managed - Here policy will be created but won't be attached )
+            policy_management: str = (
+                PolicyManagementOptions.FULLY_MANAGED.value
+                if env_role_imported is True
+                else PolicyManagementOptions.PARTIALLY_MANAGED.value
+            )
             PolicyManager(
+                session=session,
                 role_name=env_group_iam_role_name,
                 environmentUri=environment.environmentUri,
                 account=environment.AwsAccountId,
                 region=environment.region,
                 resource_prefix=environment.resourcePrefix,
-            ).create_all_policies(managed=env_role_imported)
+            ).create_all_policies(policy_management=policy_management)
 
             athena_workgroup = NamingConventionService(
                 target_uri=environment.environmentUri,
@@ -470,6 +478,7 @@ def remove_group(uri, group):
             group_membership = EnvironmentService.find_environment_group(session, group, environment.environmentUri)
 
             PolicyManager(
+                session=session,
                 role_name=group_membership.environmentIAMRoleName,
                 environmentUri=environment.environmentUri,
                 account=environment.AwsAccountId,
@@ -590,16 +599,17 @@ def add_consumption_role(uri, data=None) -> (Environment, EnvironmentGroup):
                 groupUri=group,
                 IAMRoleArn=IAMRoleArn,
                 IAMRoleName=IAMRoleArn.split('/')[-1],
-                dataallManaged=data.get('dataallManaged', True),
+                dataallManaged=data.get('dataallManaged'),
             )
 
             PolicyManager(
+                session=session,
                 role_name=consumption_role.IAMRoleName,
                 environmentUri=environment.environmentUri,
                 account=environment.AwsAccountId,
                 region=environment.region,
                 resource_prefix=environment.resourcePrefix,
-            ).create_all_policies(managed=consumption_role.dataallManaged)
+            ).create_all_policies(policy_management=consumption_role.dataallManaged)
 
             session.add(consumption_role)
             session.commit()
@@ -630,6 +640,7 @@ def remove_consumption_role(uri, env_uri):
 
             if consumption_role:
                 PolicyManager(
+                    session=session,
                     role_name=consumption_role.IAMRoleName,
                     environmentUri=environment.environmentUri,
                     account=environment.AwsAccountId,
@@ -668,6 +679,26 @@ def update_consumption_role(uri, env_uri, input):
                 for key, value in input.items():
                     setattr(consumption_role, key, value)
                 session.commit()
+
+                # If the input consumption role is not Fully-Managed then attach the share policy if it exists
+                if consumption_role.dataallManaged == PolicyManagementOptions.FULLY_MANAGED.value:
+                    environment: Environment = EnvironmentService.get_environment_by_uri(session, env_uri)
+                    share_policy_manager = PolicyManager(
+                        session=session,
+                        account=environment.AwsAccountId,
+                        region=environment.region,
+                        environmentUri=environment.environmentUri,
+                        resource_prefix=environment.resourcePrefix,
+                        role_name=consumption_role.IAMRoleName,
+                    )
+                    for policy_manager in [
+                        Policy
+                        for Policy in share_policy_manager.initializedPolicies
+                        if Policy.policy_type == 'SharePolicy'
+                    ]:
+                        managed_policy_list = policy_manager.get_policies_unattached_to_role()
+                        policy_manager.attach_policies(managed_policy_list)
+
             return consumption_role
 
     @staticmethod
@@ -808,6 +839,15 @@ def paginated_all_environment_consumption_roles(uri, data=None) -> dict:
     def get_consumption_role(session, uri) -> Query:
         return EnvironmentRepository.get_consumption_role(session, uri)
 
+    @staticmethod
+    def get_role_policy_management_type(principal_type: str, principal_id: str):
+        with get_context().db_engine.scoped_session() as session:
+            if principal_type == PrincipalType.ConsumptionRole.value:
+                consumption_role: ConsumptionRole = EnvironmentService.get_consumption_role(session, uri=principal_id)
+                return consumption_role.dataallManaged
+
+        return PolicyManagementOptions.FULLY_MANAGED.value
+
     @staticmethod
     @ResourcePolicyService.has_resource_permission(environment_permissions.LIST_ENVIRONMENT_NETWORKS)
     def paginated_environment_networks(uri, data=None) -> dict:
@@ -893,6 +933,7 @@ def delete_environment(uri):
                     StackStatus.DELETE_COMPLETE.value,
                 ]:
                     PolicyManager(
+                        session=session,
                         role_name=environment.EnvironmentDefaultIAMRoleName,
                         environmentUri=environment.environmentUri,
                         account=environment.AwsAccountId,
@@ -1111,16 +1152,12 @@ def get_template_from_resource_bucket(uri, template_name):
     @ResourcePolicyService.has_resource_permission(environment_permissions.GET_ENVIRONMENT)
     def resolve_consumption_role_policies(uri, IAMRoleName):
         environment = EnvironmentService.find_environment_by_uri(uri=uri)
-        return PolicyManager(
-            role_name=IAMRoleName,
-            environmentUri=uri,
-            account=environment.AwsAccountId,
-            region=environment.region,
-            resource_prefix=environment.resourcePrefix,
-        ).get_all_policies()
-
-    @staticmethod
-    @ResourcePolicyService.has_resource_permission(environment_permissions.GET_ENVIRONMENT)
-    def get_consumption_role_by_name(uri, IAMRoleName):
         with get_context().db_engine.scoped_session() as session:
-            return EnvironmentRepository.get_environment_consumption_role_by_name(session, uri, IAMRoleName)
+            return PolicyManager(
+                session=session,
+                account=environment.AwsAccountId,
+                region=environment.region,
+                environmentUri=uri,
+                resource_prefix=environment.resourcePrefix,
+                role_name=IAMRoleName,
+            ).get_all_policies()
diff --git a/backend/dataall/core/environment/services/managed_iam_policies.py b/backend/dataall/core/environment/services/managed_iam_policies.py
index ebf7a90ce..434974a0e 100644
--- a/backend/dataall/core/environment/services/managed_iam_policies.py
+++ b/backend/dataall/core/environment/services/managed_iam_policies.py
@@ -3,6 +3,8 @@
 import json
 from abc import ABC, abstractmethod
 from dataall.base.aws.iam import IAM
+from dataall.core.environment.db.environment_enums import PolicyManagementOptions
+from dataall.core.environment.db.environment_repositories import EnvironmentRepository
 
 logger = logging.getLogger(__name__)
 
@@ -93,14 +95,8 @@ def _get_policy_names(self, base_policy_name):
 
 
 class PolicyManager(object):
-    def __init__(
-        self,
-        role_name,
-        account,
-        region,
-        environmentUri,
-        resource_prefix,
-    ):
+    def __init__(self, session, account, region, environmentUri, resource_prefix, role_name):
+        self.session = session
         self.role_name = role_name
         self.account = account
         self.region = region
@@ -119,7 +115,7 @@ def _initialize_policy(self, managedPolicy):
             resource_prefix=self.resource_prefix,
         )
 
-    def create_all_policies(self, managed) -> bool:
+    def create_all_policies(self, policy_management: str) -> bool:
         """
         Manager that registers and calls all policies created by data.all modules and that
         need to be created for consumption roles and team roles
@@ -136,7 +132,7 @@ def create_all_policies(self, managed) -> bool:
                     policy=json.dumps(empty_policy),
                 )
 
-                if managed:
+                if policy_management == PolicyManagementOptions.FULLY_MANAGED.value:
                     IAM.attach_role_policy(
                         account_id=self.account,
                         region=self.region,
@@ -199,12 +195,27 @@ def get_all_policies(self) -> List[dict]:
                 if policy_manager.check_if_policy_exists(policy_name=old_managed_policy_name):
                     policy_name_list.append(old_managed_policy_name)
 
+            # Check if the role_name is registered as a consumption role.
+            # If its a consumption role with a "Externally Managed" policy management then 'attached' will be marked as 'N/A'
+            externally_managed_role: bool = False
+            role_arn = f'arn:aws:iam::{self.account}:role/{self.role_name}'
+            consumption_role_details = EnvironmentRepository.find_consumption_roles_by_IAMArn(
+                session=self.session, uri=self.environmentUri, arn=role_arn
+            )
+            if (
+                consumption_role_details
+                and consumption_role_details.dataallManaged == PolicyManagementOptions.EXTERNALLY_MANAGED.value
+            ):
+                externally_managed_role = True
+
             for policy_name in policy_name_list:
                 policy_dict = {
                     'policy_name': policy_name,
                     'policy_type': policy_manager.policy_type,
                     'exists': policy_manager.check_if_policy_exists(policy_name=policy_name),
-                    'attached': policy_manager.check_if_policy_attached(policy_name=policy_name),
+                    'attached': 'N/A'
+                    if externally_managed_role
+                    else policy_manager.check_if_policy_attached(policy_name=policy_name),
                 }
                 all_policies.append(policy_dict)
         logger.info(f'All policies currently added to role: {self.role_name} are: {str(all_policies)}')
diff --git a/backend/dataall/modules/s3_datasets_shares/services/s3_share_validator.py b/backend/dataall/modules/s3_datasets_shares/services/s3_share_validator.py
index 644591fec..e5dcf3e10 100644
--- a/backend/dataall/modules/s3_datasets_shares/services/s3_share_validator.py
+++ b/backend/dataall/modules/s3_datasets_shares/services/s3_share_validator.py
@@ -1,5 +1,6 @@
 from dataall.base.db.exceptions import UnauthorizedOperation, InvalidInput
 from dataall.base.aws.iam import IAM
+from dataall.core.environment.db.environment_enums import PolicyManagementOptions
 from dataall.core.environment.services.environment_service import EnvironmentService
 from dataall.core.environment.db.environment_models import EnvironmentGroup, ConsumptionRole
 from dataall.core.environment.services.managed_iam_policies import PolicyManager
@@ -105,7 +106,7 @@ def _validate_iam_role_and_policy(
                 session, principal_id, environment.environmentUri
             )
             principal_role_name = consumption_role.IAMRoleName
-            managed = consumption_role.dataallManaged
+            managed = consumption_role.dataallManaged == PolicyManagementOptions.FULLY_MANAGED.value
 
         else:
             env_group: EnvironmentGroup = EnvironmentService.get_environment_group(
@@ -125,6 +126,7 @@ def _validate_iam_role_and_policy(
 
         log.info('Verifying data.all managed share IAM policy is attached to IAM role...')
         share_policy_manager = PolicyManager(
+            session=session,
             role_name=principal_role_name,
             environmentUri=environment.environmentUri,
             account=environment.AwsAccountId,
@@ -156,10 +158,6 @@ def _validate_iam_role_and_policy(
             # End of backwards compatibility
 
             unattached = policy_manager.get_policies_unattached_to_role()
-            if unattached and not managed and not attachMissingPolicies:
-                raise Exception(
-                    f'Required customer managed policies {policy_manager.get_policies_unattached_to_role()} are not attached to role {principal_role_name}'
-                )
-            elif unattached:
+            if unattached and (managed or attachMissingPolicies):
                 managed_policy_list = policy_manager.get_policies_unattached_to_role()
                 policy_manager.attach_policies(managed_policy_list)
diff --git a/backend/dataall/modules/s3_datasets_shares/services/share_managers/s3_access_point_share_manager.py b/backend/dataall/modules/s3_datasets_shares/services/share_managers/s3_access_point_share_manager.py
index 791960ff6..9f6e8f825 100644
--- a/backend/dataall/modules/s3_datasets_shares/services/share_managers/s3_access_point_share_manager.py
+++ b/backend/dataall/modules/s3_datasets_shares/services/share_managers/s3_access_point_share_manager.py
@@ -6,6 +6,9 @@
 from warnings import warn
 
 from dataall.base.db.exceptions import AWSServiceQuotaExceeded
+from dataall.core.environment.db.environment_enums import PolicyManagementOptions
+from dataall.core.environment.db.environment_models import ConsumptionRole
+from dataall.core.environment.db.environment_repositories import EnvironmentRepository
 from dataall.core.environment.services.environment_service import EnvironmentService
 from dataall.base.db import utils
 from dataall.base.aws.sts import SessionHelper
@@ -164,6 +167,15 @@ def check_target_role_access_policy(self) -> None:
         """
         logger.info(f'Check target role {self.target_requester_IAMRoleName} access policy')
 
+        is_managed_role: bool = True
+        if self.share.principalType == PrincipalType.ConsumptionRole.value:
+            # When principalType is a consumptionRole type then principalId contains the uri of the consumption role
+            consumption_role: ConsumptionRole = EnvironmentRepository.get_consumption_role(
+                self.session, self.share.principalId
+            )
+            if consumption_role.dataallManaged == PolicyManagementOptions.EXTERNALLY_MANAGED.value:
+                is_managed_role = False
+
         key_alias = f'alias/{self.dataset.KmsAlias}'
         kms_client = KmsClient(self.dataset_account_id, self.source_environment.region)
         kms_key_id = kms_client.get_key_id(key_alias)
@@ -201,13 +213,14 @@ def check_target_role_access_policy(self) -> None:
                 self.folder_errors.append(ShareErrorFormatter.dne_error_msg('IAM Policy', share_resource_policy_name))
                 return
 
-        unattached_policies: List[str] = share_policy_service.get_policies_unattached_to_role()
-        if len(unattached_policies) > 0:
-            logger.info(
-                f'IAM Policies {unattached_policies} exists but are not attached to role {self.share.principalRoleName}'
-            )
-            self.folder_errors.append(ShareErrorFormatter.dne_error_msg('IAM Policy attached', unattached_policies))
-            return
+        if is_managed_role:
+            unattached_policies: List[str] = share_policy_service.get_policies_unattached_to_role()
+            if len(unattached_policies) > 0:
+                logger.info(
+                    f'IAM Policies {unattached_policies} exists but are not attached to role {self.share.principalRoleName}'
+                )
+                self.folder_errors.append(ShareErrorFormatter.dne_error_msg('IAM Policy attached', unattached_policies))
+                return
 
         s3_target_resources = [
             f'arn:aws:s3:::{self.bucket_name}',
@@ -399,7 +412,7 @@ def grant_target_role_access_policy(self):
                 consumption_role = EnvironmentService.get_consumption_role(
                     session=self.session, uri=self.share.principalId
                 )
-                if consumption_role.dataallManaged:
+                if consumption_role.dataallManaged == PolicyManagementOptions.FULLY_MANAGED.value:
                     share_managed_policies = share_policy_service.get_managed_policies()
                     share_policy_service.attach_policies(share_managed_policies)
 
diff --git a/backend/dataall/modules/s3_datasets_shares/services/share_managers/s3_bucket_share_manager.py b/backend/dataall/modules/s3_datasets_shares/services/share_managers/s3_bucket_share_manager.py
index aac6fea78..e44aff335 100644
--- a/backend/dataall/modules/s3_datasets_shares/services/share_managers/s3_bucket_share_manager.py
+++ b/backend/dataall/modules/s3_datasets_shares/services/share_managers/s3_bucket_share_manager.py
@@ -6,7 +6,9 @@
 from dataall.base.aws.iam import IAM
 from dataall.base.aws.sts import SessionHelper
 from dataall.base.db.exceptions import AWSServiceQuotaExceeded
-from dataall.core.environment.db.environment_models import Environment
+from dataall.core.environment.db.environment_enums import PolicyManagementOptions
+from dataall.core.environment.db.environment_models import Environment, ConsumptionRole
+from dataall.core.environment.db.environment_repositories import EnvironmentRepository
 from dataall.core.environment.services.environment_service import EnvironmentService
 from dataall.modules.s3_datasets.db.dataset_models import DatasetBucket
 from dataall.modules.s3_datasets_shares.aws.kms_client import (
@@ -74,6 +76,15 @@ def check_s3_iam_access(self) -> None:
         """
         logger.info(f'Check target role {self.target_requester_IAMRoleName} access policy')
 
+        is_managed_role: bool = True
+        if self.share.principalType == PrincipalType.ConsumptionRole.value:
+            # When principalType is a consumptionRole type then principalId contains the uri of the consumption role
+            consumption_role: ConsumptionRole = EnvironmentRepository.get_consumption_role(
+                self.session, self.share.principalId
+            )
+            if consumption_role.dataallManaged == PolicyManagementOptions.EXTERNALLY_MANAGED.value:
+                is_managed_role = False
+
         key_alias = f'alias/{self.target_bucket.KmsAlias}'
         kms_client = KmsClient(self.source_account_id, self.source_environment.region)
         kms_key_id = kms_client.get_key_id(key_alias)
@@ -113,13 +124,14 @@ def check_s3_iam_access(self) -> None:
                 self.bucket_errors.append(ShareErrorFormatter.dne_error_msg('IAM Policy', share_resource_policy_name))
                 return
 
-        unattached_policies: List[str] = share_policy_service.get_policies_unattached_to_role()
-        if len(unattached_policies) > 0:
-            logger.info(
-                f'IAM Policies {unattached_policies} exists but are not attached to role {self.share.principalRoleName}'
-            )
-            self.bucket_errors.append(ShareErrorFormatter.dne_error_msg('IAM Policy attached', unattached_policies))
-            return
+        if is_managed_role:
+            unattached_policies: List[str] = share_policy_service.get_policies_unattached_to_role()
+            if len(unattached_policies) > 0:
+                logger.info(
+                    f'IAM Policies {unattached_policies} exists but are not attached to role {self.share.principalRoleName}'
+                )
+                self.bucket_errors.append(ShareErrorFormatter.dne_error_msg('IAM Policy attached', unattached_policies))
+                return
 
         s3_target_resources = [f'arn:aws:s3:::{self.bucket_name}', f'arn:aws:s3:::{self.bucket_name}/*']
 
@@ -304,7 +316,7 @@ def grant_s3_iam_access(self):
                 consumption_role = EnvironmentService.get_consumption_role(
                     session=self.session, uri=self.share.principalId
                 )
-                if consumption_role.dataallManaged:
+                if consumption_role.dataallManaged == PolicyManagementOptions.FULLY_MANAGED.value:
                     share_managed_policies = share_policy_service.get_managed_policies()
                     share_policy_service.attach_policies(share_managed_policies)
 
diff --git a/backend/dataall/modules/s3_datasets_shares/services/share_processors/s3_access_point_share_processor.py b/backend/dataall/modules/s3_datasets_shares/services/share_processors/s3_access_point_share_processor.py
index d522bb79d..d28679c67 100644
--- a/backend/dataall/modules/s3_datasets_shares/services/share_processors/s3_access_point_share_processor.py
+++ b/backend/dataall/modules/s3_datasets_shares/services/share_processors/s3_access_point_share_processor.py
@@ -11,6 +11,7 @@
     ShareItemStatus,
     ShareObjectActions,
     ShareItemActions,
+    PrincipalType,
 )
 from dataall.modules.s3_datasets.db.dataset_models import DatasetStorageLocation
 from dataall.modules.shares_base.db.share_object_repositories import ShareObjectRepository
diff --git a/backend/dataall/modules/s3_datasets_shares/services/share_processors/s3_bucket_share_processor.py b/backend/dataall/modules/s3_datasets_shares/services/share_processors/s3_bucket_share_processor.py
index 447988156..af8e9f4ab 100644
--- a/backend/dataall/modules/s3_datasets_shares/services/share_processors/s3_bucket_share_processor.py
+++ b/backend/dataall/modules/s3_datasets_shares/services/share_processors/s3_bucket_share_processor.py
@@ -1,6 +1,5 @@
 import logging
 from datetime import datetime
-from logging import exception
 from typing import List
 
 from dataall.modules.shares_base.services.share_exceptions import PrincipalRoleNotFound
@@ -11,6 +10,7 @@
     ShareItemStatus,
     ShareObjectActions,
     ShareItemActions,
+    PrincipalType,
 )
 from dataall.modules.shares_base.db.share_object_repositories import ShareObjectRepository
 from dataall.modules.shares_base.db.share_state_machines_repositories import ShareStatusRepository
diff --git a/backend/dataall/modules/shares_base/api/resolvers.py b/backend/dataall/modules/shares_base/api/resolvers.py
index 285326e46..c8fc29749 100644
--- a/backend/dataall/modules/shares_base/api/resolvers.py
+++ b/backend/dataall/modules/shares_base/api/resolvers.py
@@ -288,6 +288,12 @@ def resolve_existing_shared_items(context: Context, source: ShareObject, **kwarg
     return ShareItemService.check_existing_shared_items(source)
 
 
+def resolve_role_policy_management(context, source: ShareObject):
+    if not source:
+        return 'Not Available'
+    return EnvironmentService.get_role_policy_management_type(source.principalType, source.principalId)
+
+
 def list_shareable_objects(context: Context, source: ShareObject, filter: dict = None):
     if not source:
         return None
diff --git a/backend/dataall/modules/shares_base/api/types.py b/backend/dataall/modules/shares_base/api/types.py
index a6f92d00d..570564ca4 100644
--- a/backend/dataall/modules/shares_base/api/types.py
+++ b/backend/dataall/modules/shares_base/api/types.py
@@ -1,4 +1,5 @@
 from dataall.base.api import gql
+from dataall.core.environment.db.environment_enums import PolicyManagementOptions
 from dataall.modules.shares_base.services.shares_enums import (
     ShareableType,
     PrincipalType,
@@ -14,6 +15,7 @@
     list_shareable_objects,
     resolve_user_role,
     resolve_can_view_logs,
+    resolve_role_policy_management,
 )
 from dataall.core.environment.api.resolvers import resolve_environment
 
@@ -171,6 +173,11 @@
             resolver=resolve_user_role,
         ),
         gql.Field('permissions', gql.ArrayType(ShareObjectDataPermission.toGraphQLEnum())),
+        gql.Field(
+            name='policyManagement',
+            type=gql.NonNullableType(PolicyManagementOptions.toGraphQLEnum()),
+            resolver=resolve_role_policy_management,
+        ),
     ],
 )
 
diff --git a/backend/migrations/versions/0d1653ee4dc3_merging_disjoint_heads.py b/backend/migrations/versions/0d1653ee4dc3_merging_disjoint_heads.py
new file mode 100644
index 000000000..918e71ebb
--- /dev/null
+++ b/backend/migrations/versions/0d1653ee4dc3_merging_disjoint_heads.py
@@ -0,0 +1,24 @@
+"""merging disjoint heads
+
+Revision ID: 0d1653ee4dc3
+Revises: 77c3f1b2bec8, ba2da94739ab
+Create Date: 2025-05-08 14:07:07.096840
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '0d1653ee4dc3'
+down_revision = ('77c3f1b2bec8', 'ba2da94739ab')
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    pass
+
+
+def downgrade():
+    pass
diff --git a/backend/migrations/versions/77c3f1b2bec8_consumption_role_column_update.py b/backend/migrations/versions/77c3f1b2bec8_consumption_role_column_update.py
new file mode 100644
index 000000000..977e123f2
--- /dev/null
+++ b/backend/migrations/versions/77c3f1b2bec8_consumption_role_column_update.py
@@ -0,0 +1,79 @@
+"""Consumption role schema change and backfilling
+
+Revision ID: 77c3f1b2bec8
+Revises: af2e1362d4cb
+Create Date: 2025-02-05 11:05:55.782419
+
+"""
+
+from typing import List, Dict
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy import orm
+
+from dataall.core.environment.db.environment_enums import PolicyManagementOptions
+from dataall.core.environment.db.environment_models import ConsumptionRole
+
+# revision identifiers, used by Alembic.
+revision = '77c3f1b2bec8'
+down_revision = 'af2e1362d4cb'
+branch_labels = None
+depends_on = None
+
+
+def get_session():
+    bind = op.get_bind()
+    session = orm.Session(bind=bind)
+    return session
+
+
+def upgrade():
+    # Update the column type to String and also remove the DEFAULT True
+    op.alter_column(
+        table_name='consumptionrole',
+        column_name='dataallManaged',
+        nullable=False,
+        existing_type=sa.Boolean(),
+        type_=sa.String(),
+        server_default=None,
+    )
+
+    session = get_session()
+
+    # For all consumption roles, set the dataallManaged column value with the PolicyManagementOptions types
+    consumption_roles: List[ConsumptionRole] = session.query(ConsumptionRole).all()
+    for consumption_role in consumption_roles:
+        if consumption_role.dataallManaged == 'true':
+            consumption_role.dataallManaged = PolicyManagementOptions.FULLY_MANAGED.value
+        else:
+            consumption_role.dataallManaged = PolicyManagementOptions.PARTIALLY_MANAGED.value
+    session.add_all(consumption_roles)
+    session.commit()
+
+
+def downgrade():
+    session = get_session()
+    consumption_roles: List[ConsumptionRole] = session.query(ConsumptionRole).all()
+    # For each consumption role, get the policy management options and set value to True if FullyManaged else False
+    consumption_role_policy_mgmt_map: Dict[str, str] = {
+        consumption_role.consumptionRoleUri: True
+        if consumption_role.dataallManaged == PolicyManagementOptions.FULLY_MANAGED.value
+        else False
+        for consumption_role in consumption_roles
+    }
+
+    op.drop_column(table_name='consumptionrole', column_name='dataallManaged')
+    op.add_column(
+        'consumptionrole',
+        sa.Column('dataallManaged', sa.Boolean(), nullable=False, server_default=sa.sql.expression.true()),
+    )
+
+    # Update all the consumption.dataallManaged column with boolean values by using consumption_role_policy_mgmt_map map
+    consumption_roles: List[ConsumptionRole] = session.query(ConsumptionRole).all()
+    for consumption_role in consumption_roles:
+        consumption_role.dataallManaged = consumption_role_policy_mgmt_map.get(
+            consumption_role.consumptionRoleUri, True
+        )
+    session.add_all(consumption_roles)
+    session.commit()
diff --git a/frontend/src/design/components/InfoIconWithToolTip.js b/frontend/src/design/components/InfoIconWithToolTip.js
new file mode 100644
index 000000000..e2ce78beb
--- /dev/null
+++ b/frontend/src/design/components/InfoIconWithToolTip.js
@@ -0,0 +1,25 @@
+import { Tooltip } from '@mui/material';
+import InfoIcon from '@mui/icons-material/Info';
+import PropTypes from 'prop-types';
+
+export const InfoIconWithToolTip = (props) => {
+  const { title, size, placement } = props;
+
+  return (
+    <Tooltip title={title} placement={placement}>
+      <InfoIcon sx={{ fontSize: `${size}rem` }}></InfoIcon>
+    </Tooltip>
+  );
+};
+
+InfoIconWithToolTip.propTypes = {
+  title: PropTypes.any,
+  size: PropTypes.number,
+  placement: PropTypes.string
+};
+
+InfoIconWithToolTip.defaultProps = {
+  title: '',
+  size: 1,
+  placement: 'bottom'
+};
diff --git a/frontend/src/design/components/index.js b/frontend/src/design/components/index.js
index 724562cd1..ffa320b85 100644
--- a/frontend/src/design/components/index.js
+++ b/frontend/src/design/components/index.js
@@ -32,3 +32,4 @@ export * from './popovers';
 export * from './SanitizedHTML';
 export * from './NoAccessMaintenanceWindow';
 export * from './UserModal';
+export * from './InfoIconWithToolTip';
diff --git a/frontend/src/modules/Catalog/components/RequestAccessModal.js b/frontend/src/modules/Catalog/components/RequestAccessModal.js
index 316df29a2..573ff8f84 100644
--- a/frontend/src/modules/Catalog/components/RequestAccessModal.js
+++ b/frontend/src/modules/Catalog/components/RequestAccessModal.js
@@ -720,7 +720,9 @@ export const RequestAccessModal = (props) => {
                       </Box>
                     )}
                     {!values.consumptionRole ||
-                    values.consumptionRole.dataallManaged ||
+                    values.consumptionRole.dataallManaged === 'Fully-Managed' ||
+                    values.consumptionRole.dataallManaged ===
+                      'Externally-Managed' ||
                     isSharePolicyAttached ? (
                       <Box />
                     ) : (
@@ -742,25 +744,24 @@ export const RequestAccessModal = (props) => {
                                 color="textSecondary"
                                 component="p"
                                 variant="caption"
-                              ></Typography>
-                              {values.consumptionRole &&
-                              !(
-                                values.consumptionRole.dataallManaged ||
-                                isSharePolicyAttached ||
-                                values.attachMissingPolicies
-                              ) ? (
-                                <FormHelperText error>
-                                  Selected consumption role is managed by
-                                  customer, but the share policy{' '}
-                                  <strong>{unAttachedPolicyNames}</strong> is
-                                  not attached.
-                                  <br />
-                                  Please attach it or let Data.all attach it for
-                                  you.
-                                </FormHelperText>
-                              ) : (
-                                ''
-                              )}
+                              >
+                                {values.consumptionRole &&
+                                values.consumptionRole.dataallManaged ===
+                                  'Partially-Managed' &&
+                                !isSharePolicyAttached ? (
+                                  <FormHelperText error>
+                                    Selected consumption role is partially
+                                    managed by customer and the share policy{' '}
+                                    <strong>{unAttachedPolicyNames}</strong> is
+                                    not attached.
+                                    <br />
+                                    Please attach it or let Data.all attach it
+                                    for you.
+                                  </FormHelperText>
+                                ) : (
+                                  ''
+                                )}
+                              </Typography>
                             </div>
                           }
                         />
diff --git a/frontend/src/modules/Environments/components/EnvironmentRoleAddForm.js b/frontend/src/modules/Environments/components/EnvironmentRoleAddForm.js
index 8926de4dc..5a105ada6 100644
--- a/frontend/src/modules/Environments/components/EnvironmentRoleAddForm.js
+++ b/frontend/src/modules/Environments/components/EnvironmentRoleAddForm.js
@@ -1,27 +1,34 @@
 import { GroupAddOutlined } from '@mui/icons-material';
 import { LoadingButton } from '@mui/lab';
 import {
+  Alert,
   Autocomplete,
   Box,
   CardContent,
   CircularProgress,
   Dialog,
-  FormControlLabel,
-  Switch,
   TextField,
   Typography
 } from '@mui/material';
 import { Formik } from 'formik';
 import { useSnackbar } from 'notistack';
 import PropTypes from 'prop-types';
-import React from 'react';
 import * as Yup from 'yup';
 import { SET_ERROR, useDispatch } from 'globalErrors';
 import { useClient, useFetchGroups } from 'services';
 import { addConsumptionRoleToEnvironment } from '../services';
+import { policyManagementInfoMap } from '../../constants';
+import { InfoIconWithToolTip } from '../../../design';
 
 export const EnvironmentRoleAddForm = (props) => {
-  const { environment, onClose, open, reloadRoles, ...other } = props;
+  const {
+    environment,
+    onClose,
+    open,
+    reloadRoles,
+    policyManagementOptions,
+    ...other
+  } = props;
   const { enqueueSnackbar } = useSnackbar();
   const dispatch = useDispatch();
   const client = useClient();
@@ -95,7 +102,7 @@ export const EnvironmentRoleAddForm = (props) => {
           <Formik
             initialValues={{
               groupUri: '',
-              dataallManaged: true
+              dataallManaged: ''
             }}
             validationSchema={Yup.object().shape({
               groupUri: Yup.string()
@@ -104,7 +111,12 @@ export const EnvironmentRoleAddForm = (props) => {
               consumptionRoleName: Yup.string()
                 .max(255)
                 .required('*Consumption Role Name is required'),
-              IAMRoleArn: Yup.string().required('*IAM Role Arn is required')
+              IAMRoleArn: Yup.string().required('*IAM Role Arn is required'),
+              dataallManaged: Yup.string()
+                .required(
+                  'Policy Management option required. Please select a valid option'
+                )
+                .oneOf(policyManagementOptions.map((obj) => obj.key))
             })}
             onSubmit={async (
               values,
@@ -181,30 +193,68 @@ export const EnvironmentRoleAddForm = (props) => {
                   />
                 </CardContent>
                 <CardContent>
-                  <FormControlLabel
-                    control={
-                      <Switch
-                        checked={values.dataallManaged}
-                        onChange={handleChange}
-                        color="primary"
-                        edge="start"
+                  <Autocomplete
+                    id="PolicyManagement"
+                    disablePortal
+                    options={policyManagementOptions}
+                    onChange={(event, value) => {
+                      if (value && value.key) {
+                        setFieldValue('dataallManaged', value.key);
+                      } else {
+                        setFieldValue('dataallManaged', '');
+                      }
+                    }}
+                    renderOption={(props, option) => {
+                      const { key, ...propOptions } = props;
+                      return (
+                        <Box key={key} {...propOptions}>
+                          {option.label}
+                          <InfoIconWithToolTip
+                            title={
+                              <span style={{ fontSize: 'small' }}>
+                                {policyManagementInfoMap[option.key] != null
+                                  ? policyManagementInfoMap[option.key]
+                                  : 'Invalid Option for policy management.'}
+                              </span>
+                            }
+                            placement={'right-start'}
+                            size={1}
+                          />
+                        </Box>
+                      );
+                    }}
+                    renderInput={(params) => (
+                      <TextField
+                        {...params}
+                        fullWidth
+                        error={Boolean(
+                          touched.dataallManaged && errors.dataallManaged
+                        )}
+                        helperText={
+                          touched.dataallManaged && errors.dataallManaged
+                        }
+                        label="Policy Management"
                         name="dataallManaged"
+                        variant="outlined"
+                        value={values.dataallManaged}
                       />
-                    }
-                    label={
-                      <div>
-                        Data.all managed
-                        <Typography
-                          color="textSecondary"
-                          component="p"
-                          variant="caption"
-                        >
-                          Allow Data.all to attach IAM policies to this role
-                        </Typography>
-                      </div>
-                    }
+                    )}
                   />
                 </CardContent>
+                {values.dataallManaged === 'EXTERNALLY_MANAGED' ? (
+                  <CardContent>
+                    <Alert severity="error" sx={{ mr: 1 }}>
+                      With "Externally-Managed" policy management, you are
+                      completely responsible for attaching / giving your
+                      consumption role appropriate permissions. Please select
+                      "Externally-Managed" if you know that your role has some
+                      super-user permissions or if you are completely managing
+                      the role and its policies.
+                    </Alert>
+                  </CardContent>
+                ) : (
+                  <div></div>
+                )}
                 <Box>
                   <CardContent>
                     <LoadingButton
diff --git a/frontend/src/modules/Environments/components/EnvironmentTeams.js b/frontend/src/modules/Environments/components/EnvironmentTeams.js
index f2f6bacad..9e47566ec 100644
--- a/frontend/src/modules/Environments/components/EnvironmentTeams.js
+++ b/frontend/src/modules/Environments/components/EnvironmentTeams.js
@@ -35,6 +35,7 @@ import { VscChecklist } from 'react-icons/vsc';
 import {
   Defaults,
   DeleteObjectWithFrictionModal,
+  InfoIconWithToolTip,
   Label,
   Pager,
   RefreshTableMenu,
@@ -45,6 +46,7 @@ import {
 import { SET_ERROR, useDispatch } from 'globalErrors';
 import { isFeatureEnabled } from 'utils';
 import {
+  fetchEnums,
   getConsumptionRolePolicies,
   useClient,
   useFetchGroups
@@ -62,6 +64,7 @@ import { EnvironmentRoleAddForm } from './EnvironmentRoleAddForm';
 import { EnvironmentTeamInviteEditForm } from './EnvironmentTeamInviteEditForm';
 import { EnvironmentTeamInviteForm } from './EnvironmentTeamInviteForm';
 import { DataGrid, GridActionsCellItem, GridRowModes } from '@mui/x-data-grid';
+import { policyManagementInfoMap } from '../../constants';
 
 function TeamRow({
   team,
@@ -298,6 +301,7 @@ export const IAMRolePolicyDataGridCell = ({ environmentUri, IAMRoleName }) => {
   const dispatch = useDispatch();
   const { enqueueSnackbar } = useSnackbar();
   const client = useClient();
+  const [policyAttachStatus, setPolicyAttachStatus] = useState('Not Attached');
 
   useEffect(() => {
     if (client) {
@@ -318,6 +322,11 @@ export const IAMRolePolicyDataGridCell = ({ environmentUri, IAMRoleName }) => {
       );
       if (!response.errors) {
         setManagedPolicyDetails(response.data.getConsumptionRolePolicies);
+        setPolicyAttachStatus(
+          getIAMPolicyAttachementStatus(
+            response.data.getConsumptionRolePolicies
+          )
+        );
       } else {
         dispatch({ type: SET_ERROR, error: response.errors[0].message });
       }
@@ -328,27 +337,37 @@ export const IAMRolePolicyDataGridCell = ({ environmentUri, IAMRoleName }) => {
     }
   };
 
+  const getIAMPolicyTagColour = () => {
+    if (policyAttachStatus === 'N/A') return 'warning';
+    if (policyAttachStatus === 'Not Attached') return 'error';
+    if (policyAttachStatus === 'No Policies Present') return 'error';
+    return 'success';
+  };
+
+  const getIAMPolicyAttachementStatus = (managedPolicyDetails) => {
+    if (managedPolicyDetails.length === 0) {
+      return 'No Policies Present';
+    }
+    const is_policy_attach = managedPolicyDetails.map(
+      (policy) => policy.attached
+    );
+    if (is_policy_attach.every((policy) => policy === 'N/A')) {
+      return 'N/A';
+    }
+    if (is_policy_attach.some((policy) => policy === 'false')) {
+      return 'Not Attached';
+    }
+    return 'Attached';
+  };
+
   return (
     <Box>
       {isLoading ? (
         <CircularProgress size={30} />
       ) : (
         <Box>
-          <Label
-            sx={{ ml: 5 }}
-            color={
-              managedPolicyDetails
-                .map((policy) => policy.attached)
-                .includes(false)
-                ? 'error'
-                : 'success'
-            }
-          >
-            {managedPolicyDetails
-              .map((policy) => policy.attached)
-              .includes(false)
-              ? 'Not Attached'
-              : 'Attached'}
+          <Label sx={{ ml: 5 }} color={getIAMPolicyTagColour()}>
+            {policyAttachStatus}
           </Label>
           <LoadingButton
             onClick={async () => {
@@ -394,6 +413,7 @@ export const EnvironmentTeams = ({ environment }) => {
   const [isDeleteRoleModalOpenId, setIsDeleteRoleModalOpen] = useState(0);
   const [isTeamEditModalOpenId, setIsTeamEditModalOpen] = useState('');
   const [isDeleteTeamModalOpenId, setIsDeleteTeamModalOpen] = useState('');
+  const [policyManagementOptions, setPolicyManagementOptions] = useState([]);
 
   const handleDeleteTeamModalOpen = (groupUri) => {
     setIsDeleteTeamModalOpen(groupUri);
@@ -505,7 +525,8 @@ export const EnvironmentTeams = ({ environment }) => {
         consumptionRoleUri: newRow.consumptionRoleUri,
         input: {
           groupUri: newRow.groupUri,
-          consumptionRoleName: newRow.consumptionRoleName
+          consumptionRoleName: newRow.consumptionRoleName,
+          dataallManaged: newRow.dataallManaged
         }
       })
     );
@@ -523,6 +544,25 @@ export const EnvironmentTeams = ({ environment }) => {
     }
   };
 
+  const fetchPolicyManagementOptions = async () => {
+    const response = await fetchEnums(client, ['PolicyManagementOptions']);
+    if (response['PolicyManagementOptions'].length > 0) {
+      setPolicyManagementOptions(
+        response['PolicyManagementOptions'].map((elem) => {
+          return {
+            label: elem.value,
+            key: elem.name
+          };
+        })
+      );
+    } else {
+      dispatch({
+        type: SET_ERROR,
+        error: 'Could not fetch consumption role policy management options'
+      });
+    }
+  };
+
   useEffect(() => {
     if (client) {
       fetchItems().catch((e) =>
@@ -531,6 +571,9 @@ export const EnvironmentTeams = ({ environment }) => {
       fetchRoles().catch((e) =>
         dispatch({ type: SET_ERROR, error: e.message })
       );
+      fetchPolicyManagementOptions().catch((e) =>
+        dispatch({ type: SET_ERROR, error: e.message })
+      );
     }
   }, [client, filter.page, filterRoles.page, fetchItems, fetchRoles, dispatch]);
 
@@ -603,6 +646,16 @@ export const EnvironmentTeams = ({ environment }) => {
     return newRow;
   };
 
+  const formattedName = (unformattedPolicyMgmtName) => {
+    // Split name by "_"
+    return unformattedPolicyMgmtName
+      .split('_')
+      .map((word) => {
+        return word.charAt(0).toUpperCase() + word.slice(1).toLowerCase();
+      })
+      .join('-');
+  };
+
   let { groupOptions } = useFetchGroups(environment);
 
   return (
@@ -787,6 +840,7 @@ export const EnvironmentTeams = ({ environment }) => {
                   open
                   reloadRoles={fetchRoles}
                   onClose={handleAddRoleModalClose}
+                  policyManagementOptions={policyManagementOptions}
                 />
               )}
             </Grid>
@@ -821,14 +875,35 @@ export const EnvironmentTeams = ({ environment }) => {
                   {
                     field: 'dataallManaged',
                     headerName: 'Policy Management',
-                    valueGetter: (params) => {
-                      return `${
-                        params.row.dataallManaged
-                          ? 'Data.all managed'
-                          : 'Customer managed'
-                      }`;
+                    renderCell: (params) => {
+                      return (
+                        <span>
+                          {formattedName(params.row.dataallManaged)}
+                          <InfoIconWithToolTip
+                            title={
+                              <span style={{ fontSize: 'small' }}>
+                                {policyManagementInfoMap[
+                                  params.row.dataallManaged
+                                ] != null
+                                  ? policyManagementInfoMap[
+                                      params.row.dataallManaged
+                                    ]
+                                  : 'Invalid Option for policy management.'}
+                              </span>
+                            }
+                            size={1}
+                            placement={'right-start'}
+                          />
+                        </span>
+                      );
                     },
-                    flex: 0.6
+                    flex: 0.6,
+                    editable: true,
+                    type: 'singleSelect',
+                    valueOptions: policyManagementOptions.map((obj) => obj.key),
+                    valueFormatter: (value) => {
+                      return formattedName(value.value);
+                    }
                   },
                   {
                     field: 'policiesNames',
diff --git a/frontend/src/modules/Shares/services/getShareObject.js b/frontend/src/modules/Shares/services/getShareObject.js
index f634b68c5..986bdc6bc 100644
--- a/frontend/src/modules/Shares/services/getShareObject.js
+++ b/frontend/src/modules/Shares/services/getShareObject.js
@@ -31,6 +31,7 @@ export const getShareObject = ({ shareUri, filter }) => ({
           SamlGroupName
           environmentName
         }
+        policyManagement
         items(filter: $filter) {
           count
           page
diff --git a/frontend/src/modules/Shares/views/ShareView.js b/frontend/src/modules/Shares/views/ShareView.js
index 35d338fc4..1c65c34a2 100644
--- a/frontend/src/modules/Shares/views/ShareView.js
+++ b/frontend/src/modules/Shares/views/ShareView.js
@@ -55,7 +55,8 @@ import {
   useSettings,
   Label,
   SanitizedHTML,
-  UserModal
+  UserModal,
+  InfoIconWithToolTip
 } from 'design';
 import { SET_ERROR, useDispatch } from 'globalErrors';
 import { useClient } from 'services';
@@ -88,6 +89,7 @@ import { ShareSubmitModal } from '../components/ShareSubmitModal';
 import { useTheme } from '@mui/styles';
 import { UpdateExtensionReason } from '../components/ShareUpdateExtension';
 import CancelIcon from '@mui/icons-material/Close';
+import { policyManagementInfoMap } from '../../constants';
 
 const isReadOnlyShare = (share) => share.permissions.every((p) => p === 'Read');
 
@@ -959,6 +961,15 @@ const ShareView = () => {
     return null;
   }
 
+  const formatPolicyManagmentType = (unformattedPolicyMgmtName) => {
+    return unformattedPolicyMgmtName
+      .split('_') // Split string "FULLY_MANAGED", etc on "_"
+      .map((word) => {
+        return word.charAt(0).toUpperCase() + word.slice(1).toLowerCase(); // Capitalize first word and lower case others
+      })
+      .join('-');
+  };
+
   return (
     <>
       <Helmet>
@@ -1378,6 +1389,77 @@ const ShareView = () => {
                               </Tooltip>
                             </Typography>
                           </ListItem>
+                          <ListItem
+                            disableGutters
+                            divider
+                            sx={{
+                              justifyContent: 'space-between',
+                              padding: 2
+                            }}
+                          >
+                            <Typography
+                              color="textSecondary"
+                              variant="subtitle2"
+                            >
+                              <Tooltip
+                                title={
+                                  <span>
+                                    IAM policy management indicates how much
+                                    control data.all has into creating and
+                                    attaching policies
+                                    <br />
+                                    <br />
+                                    1. Data.all fully managed - Data.all manages
+                                    creating, maintaining and also attaching the
+                                    policy <br />
+                                    <br />
+                                    2. Data.all partially managed - Data.all
+                                    will create the IAM policy but won't attach
+                                    policy to your consumption role. With this
+                                    option, data.all will indicate share to be
+                                    unhealthy if the data.all created policy is
+                                    not attached.
+                                    <br />
+                                    <br />
+                                    3. Externally Managed - Data.all will create
+                                    the IAM policy required for any share but it
+                                    will be incumbent on role owners to attach
+                                    it or use their own policy. With this
+                                    option, data.all will not indicate the share
+                                    to be unhealthy even if the policy is not
+                                    attached.
+                                  </span>
+                                }
+                              >
+                                Principal IAM Policy Management
+                              </Tooltip>
+                            </Typography>
+                            <Typography
+                              color="textPrimary"
+                              variant="body2"
+                              sx={{
+                                whiteSpace: 'nowrap',
+                                overflow: 'hidden',
+                                textOverflow: 'ellipsis',
+                                WebkitBoxOrient: 'vertical',
+                                WebkitLineClamp: 2
+                              }}
+                            >
+                              <span>
+                                {formatPolicyManagmentType(
+                                  share.policyManagement
+                                )}{' '}
+                              </span>
+                              <InfoIconWithToolTip
+                                title={
+                                  policyManagementInfoMap[
+                                    share.policyManagement
+                                  ]
+                                }
+                                size={1}
+                              />
+                            </Typography>
+                          </ListItem>
                           <ListItem
                             disableGutters
                             divider
diff --git a/frontend/src/modules/constants.js b/frontend/src/modules/constants.js
index 6268f7a4b..4835cf136 100644
--- a/frontend/src/modules/constants.js
+++ b/frontend/src/modules/constants.js
@@ -14,3 +14,12 @@ export const Topics = [
 ];
 
 export const ConfidentialityList = ['Unclassified', 'Official', 'Secret'];
+
+export const policyManagementInfoMap = {
+  FULLY_MANAGED:
+    'Data.all manages creating, maintaining and also attaching the policy',
+  PARTIALLY_MANAGED:
+    'Data.all will create the IAM policy but will not attach policy to your consumption role. With this option, data.all will indicate share to be unhealthy if the data.all created policy is not attached.',
+  EXTERNALLY_MANAGED:
+    'Data.all will create the IAM policy required for any share but it will be incumbent on role owners to attach it or use their own policy. With this option, data.all will not indicate the share to be unhealthy even if the policy is not attached.'
+};
diff --git a/tests/core/environments/conftest.py b/tests/core/environments/conftest.py
index 853cbfbb4..a18970d58 100644
--- a/tests/core/environments/conftest.py
+++ b/tests/core/environments/conftest.py
@@ -1,5 +1,6 @@
 import uuid
 
+from dataall.core.environment.db.environment_enums import PolicyManagementOptions
 from tests.core.permissions.test_permission import *
 
 
@@ -33,7 +34,7 @@ def consumption_role(mock_aws_client, client, org_fixture, env_fixture, user, gr
             'groupUri': str(uuid.uuid4())[:8],
             'IAMRoleArn': test_arn,
             'environmentUri': env_fixture.environmentUri,
-            'dataallManaged': False,
+            'dataallManaged': 'FULLY_MANAGED',
         },
     )
     assert not response.errors
diff --git a/tests/core/environments/test_environment.py b/tests/core/environments/test_environment.py
index 4cf5aa7d9..969287525 100644
--- a/tests/core/environments/test_environment.py
+++ b/tests/core/environments/test_environment.py
@@ -749,6 +749,7 @@ def test_update_consumption_role(client, org_fixture, env_fixture, user, group,
                 groupUri
                 IAMRoleName
                 IAMRoleArn
+                dataallManaged
             }
         }
     """
@@ -770,9 +771,15 @@ def test_update_consumption_role(client, org_fixture, env_fixture, user, group,
         groups=[group.name],
         environmentUri=env_fixture.environmentUri,
         consumptionRoleUri=consumption_role_uri,
-        input={'consumptionRoleName': 'testRoleName', 'groupUri': 'testGroupUri'},
+        # Update consumptionRoleName, groupUri and also the policy management ( dataallManaged )
+        input={
+            'consumptionRoleName': 'testRoleName',
+            'groupUri': 'testGroupUri',
+            'dataallManaged': 'PARTIALLY_MANAGED',
+        },
     )
 
     assert not response.errors
     assert response.data.updateConsumptionRole.consumptionRoleName == 'testRoleName'
     assert response.data.updateConsumptionRole.groupUri == 'testGroupUri'
+    assert response.data.updateConsumptionRole.dataallManaged == 'PARTIALLY_MANAGED'
diff --git a/tests/modules/conftest.py b/tests/modules/conftest.py
index 6f76a7a56..895d53c42 100644
--- a/tests/modules/conftest.py
+++ b/tests/modules/conftest.py
@@ -2,8 +2,15 @@
 
 import pytest
 
-from dataall.core.environment.db.environment_models import Environment, EnvironmentGroup, EnvironmentParameter
+from dataall.core.environment.db.environment_enums import PolicyManagementOptions
+from dataall.core.environment.db.environment_models import (
+    Environment,
+    EnvironmentGroup,
+    EnvironmentParameter,
+    ConsumptionRole,
+)
 from dataall.core.organizations.db.organization_models import Organization
+from dataall.core.permissions.services import environment_permissions
 from dataall.core.permissions.services.environment_permissions import ENVIRONMENT_ALL
 from dataall.core.permissions.services.organization_permissions import ORGANIZATION_ALL
 from dataall.core.permissions.services.resource_policy_service import ResourcePolicyService
@@ -45,6 +52,40 @@ def factory(environment: Environment, group: str) -> EnvironmentGroup:
     yield factory
 
 
+@pytest.fixture(scope='module')
+def consumption_role(db):
+    def factory(
+        environment: Environment,
+        group: str,
+        consumption_role_name='test123',
+        datallManaged=PolicyManagementOptions.FULLY_MANAGED.value,
+    ) -> EnvironmentGroup:
+        with db.scoped_session() as session:
+            IAMRoleArn = f'arn:aws:iam::{environment.AwsAccountId}:role/{consumption_role_name}'
+            consumption_role: ConsumptionRole = ConsumptionRole(
+                consumptionRoleName=consumption_role_name,
+                environmentUri=environment.environmentUri,
+                groupUri=group,
+                IAMRoleArn=IAMRoleArn,
+                IAMRoleName=IAMRoleArn.split('/')[-1],
+                dataallManaged=datallManaged,
+            )
+            session.add(consumption_role)
+            session.commit()
+
+            ResourcePolicyService.attach_resource_policy(
+                session=session,
+                group=group,
+                resource_uri=consumption_role.consumptionRoleUri,
+                permissions=environment_permissions.CONSUMPTION_ROLE_ALL,
+                resource_type=ConsumptionRole.__name__,
+            )
+            session.commit()
+            return consumption_role
+
+    yield factory
+
+
 def _create_env_params(session, env: Environment, params: Dict[str, str]):
     if params:
         for key, value in params.items():
diff --git a/tests/modules/s3_datasets_shares/tasks/conftest.py b/tests/modules/s3_datasets_shares/tasks/conftest.py
index 253663b1d..7c83d17d1 100644
--- a/tests/modules/s3_datasets_shares/tasks/conftest.py
+++ b/tests/modules/s3_datasets_shares/tasks/conftest.py
@@ -1,7 +1,7 @@
 import pytest
 
 from dataall.core.organizations.db.organization_models import Organization
-from dataall.core.environment.db.environment_models import Environment, EnvironmentGroup
+from dataall.core.environment.db.environment_models import Environment, EnvironmentGroup, ConsumptionRole
 from dataall.modules.shares_base.services.shares_enums import (
     ShareableType,
     ShareItemStatus,
@@ -128,15 +128,24 @@ def factory(dataset: S3Dataset, name) -> DatasetBucket:
 
 @pytest.fixture(scope='module')
 def share(db):
-    def factory(dataset: S3Dataset, environment: Environment, env_group: EnvironmentGroup) -> ShareObject:
+    def factory(
+        dataset: S3Dataset,
+        environment: Environment,
+        env_group: EnvironmentGroup,
+        consumption_role: ConsumptionRole = None,
+    ) -> ShareObject:
         with db.scoped_session() as session:
             share = ShareObject(
                 datasetUri=dataset.datasetUri,
                 environmentUri=environment.environmentUri,
                 owner='bob',
-                principalId=environment.SamlGroupName,
-                principalType=PrincipalType.Group.value,
-                principalRoleName=env_group.environmentIAMRoleName,
+                principalId=environment.SamlGroupName if not consumption_role else consumption_role.consumptionRoleUri,
+                principalType=PrincipalType.Group.value
+                if not consumption_role
+                else PrincipalType.ConsumptionRole.value,
+                principalRoleName=env_group.environmentIAMRoleName
+                if not consumption_role
+                else consumption_role.IAMRoleName,
                 status=ShareObjectStatus.Approved.value,
                 groupUri=env_group.groupUri,
                 permissions=[ShareObjectDataPermission.Read.value],
diff --git a/tests/modules/s3_datasets_shares/tasks/test_s3_bucket_share_manager.py b/tests/modules/s3_datasets_shares/tasks/test_s3_bucket_share_manager.py
index b0db7668f..0bb3130fb 100644
--- a/tests/modules/s3_datasets_shares/tasks/test_s3_bucket_share_manager.py
+++ b/tests/modules/s3_datasets_shares/tasks/test_s3_bucket_share_manager.py
@@ -4,8 +4,9 @@
 
 from typing import Callable
 
+from dataall.core.environment.db.environment_enums import PolicyManagementOptions
 from dataall.core.groups.db.group_models import Group
-from dataall.core.environment.db.environment_models import Environment, EnvironmentGroup
+from dataall.core.environment.db.environment_models import Environment, EnvironmentGroup, ConsumptionRole
 from dataall.core.organizations.db.organization_models import Organization
 from dataall.modules.shares_base.db.share_object_models import ShareObject
 from dataall.modules.s3_datasets_shares.services.share_managers import S3BucketShareManager
@@ -51,6 +52,16 @@ def source_environment_group(environment_group: Callable, source_environment: En
     yield source_environment_group
 
 
+@pytest.fixture(scope='module')
+def source_consumption_role(
+    consumption_role: Callable, source_environment_group, source_environment: Environment, group: Group
+):
+    source_consumption_role = consumption_role(
+        source_environment, group.name, 'test123', PolicyManagementOptions.FULLY_MANAGED.value
+    )
+    yield source_consumption_role
+
+
 @pytest.fixture(scope='module')
 def target_environment(env: Callable, org_fixture: Organization, group2: Group):
     target_environment = env(
@@ -100,6 +111,18 @@ def share2(
     yield share2
 
 
+@pytest.fixture(scope='module')
+def share2_consumption_role(
+    share: Callable,
+    dataset2: S3Dataset,
+    target_environment: Environment,
+    target_environment_group: EnvironmentGroup,
+    source_consumption_role,
+) -> ShareObject:
+    share2_consumption_role = share(dataset2, target_environment, target_environment_group, source_consumption_role)
+    yield share2_consumption_role
+
+
 @pytest.fixture(scope='module')
 def share3(
     share: Callable,
@@ -125,6 +148,25 @@ def share_data2(
     )
 
 
+@pytest.fixture(scope='module')
+def share_data2_consumption_role(
+    share2_consumption_role,
+    dataset2,
+    source_environment,
+    target_environment,
+    source_environment_group,
+    target_environment_group,
+):
+    yield ShareData(
+        share=share2_consumption_role,
+        dataset=dataset2,
+        source_environment=source_environment,
+        target_environment=target_environment,
+        source_env_group=source_environment_group,
+        env_group=target_environment_group,
+    )
+
+
 @pytest.fixture(scope='module')
 def share_data3(
     share3, dataset_imported, source_environment, target_environment, source_environment_group, target_environment_group
@@ -167,6 +209,17 @@ def share2_manager(
     yield manager
 
 
+@pytest.fixture(scope='function')
+def share2_consumption_role_manager(
+    db,
+    bucket2,
+    share_data2_consumption_role,
+):
+    with db.scoped_session() as session:
+        manager = S3BucketShareManager(session, share_data2_consumption_role, bucket2)
+    yield manager
+
+
 @pytest.fixture(scope='function')
 def share3_manager(
     db,
@@ -263,6 +316,14 @@ def mock_iam_client(mocker, account_id, role_name):
     return mock_client
 
 
+def convert_role_management_to_type(db_engine, consumption_role: ConsumptionRole, policy_management_type: str):
+    with db_engine.scoped_session() as session:
+        consumption_role.dataallManaged = policy_management_type
+        session.add(consumption_role)
+        session.commit()
+    return consumption_role
+
+
 # For below test cases, dataset2, share2, src, target env and src group , env group remain the same
 def test_grant_role_bucket_policy_with_no_policy_present(
     mocker, dataset2, share2: ShareObject, target_environment: Environment, share2_manager
@@ -715,7 +776,7 @@ def test_grant_s3_iam_access_with_complete_policy_present(mocker, dataset2, shar
         return_value=['policy-0'],
     )
 
-    mocker.patch(
+    policy_attacher_mock = mocker.patch(
         'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.attach_policies',
         return_value=True,
     )
@@ -754,6 +815,153 @@ def test_grant_s3_iam_access_with_complete_policy_present(mocker, dataset2, shar
         and policy['Statement'][kms_index]['Action'] == created_iam_policy['Statement'][kms_index]['Action']
     )
 
+    # Assert that the policy was attached at the end
+    # TODO
+    policy_attacher_mock.assert_called_once()
+
+
+def test_grant_s3_iam_access_with_fully_managed_consumption_role_complete_policy_but_unattached(
+    mocker, dataset2, share2_consumption_role_manager
+):
+    # Given
+    # The IAM Policy for sharing for the IAM role exists (check_if_policy_exists returns False but get_managed_policies returns ['policy-0'], indicating that indexed IAM policies exist)
+    # And the IAM Policy is NOT empty and already contains all target resources (get_managed_policy_default_version returns policy)
+    # Check if the IAM role ( i.e. the consumption role ) gets attached at the end
+
+    policy = {
+        'Version': '2012-10-17',
+        'Statement': [
+            {
+                'Sid': f'{IAM_S3_BUCKETS_STATEMENT_SID}S31',
+                'Effect': 'Allow',
+                'Action': S3_ALLOWED_ACTIONS,
+                'Resource': [
+                    f'arn:aws:s3:::{dataset2.S3BucketName}',
+                    f'arn:aws:s3:::{dataset2.S3BucketName}/*',
+                    f'arn:aws:s3:::S3Bucket',
+                    f'arn:aws:s3:::S3Bucket/*',
+                ],
+            },
+            {
+                'Sid': f'{IAM_S3_BUCKETS_STATEMENT_SID}KMS1',
+                'Effect': 'Allow',
+                'Action': ['kms:*'],
+                'Resource': [f'arn:aws:kms:{dataset2.region}:{dataset2.AwsAccountId}:key/kms-key'],
+            },
+        ],
+    }
+
+    mocker.patch('dataall.base.aws.iam.IAM.create_managed_policy', return_value=True)
+
+    mocker.patch('dataall.base.aws.iam.IAM.get_attached_managed_policies_to_role', return_value=['policy-0'])
+
+    mocker.patch('dataall.base.aws.iam.IAM.list_policy_names_by_policy_pattern', return_value=['policy-0'])
+
+    mocker.patch('dataall.base.aws.iam.IAM.delete_managed_policy_non_default_versions', return_value=True)
+
+    mocker.patch('dataall.base.aws.iam.IAM.delete_managed_policy_by_name', return_value=True)
+
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService._get_managed_policy_quota',
+        return_value=10,
+    )
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.check_if_policy_exists',
+        return_value=False,
+    )
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.get_policies_unattached_to_role',
+        return_value=['policy-0'],
+    )
+
+    policy_attacher_mock = mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.attach_policies',
+        return_value=True,
+    )
+
+    kms_client = mock_kms_client(mocker)
+    kms_client().get_key_id.return_value = 'kms-key'
+
+    mocker.patch('dataall.base.aws.iam.IAM.get_managed_policy_default_version', return_value=('v1', policy))
+    mocker.patch('dataall.base.aws.iam.IAM.update_managed_policy_default_version', return_value=None)
+    share2_consumption_role_manager.grant_s3_iam_access()
+
+    policy_attacher_mock.assert_called_once()
+
+
+def test_grant_s3_iam_access_with_partially_managed_consumption_role_complete_policy_but_unattached(
+    db, mocker, dataset2, source_consumption_role, share2_consumption_role_manager
+):
+    # Given
+    # The IAM Policy for sharing for the IAM role exists (check_if_policy_exists returns False but get_managed_policies returns ['policy-0'], indicating that indexed IAM policies exist)
+    # And the IAM Policy is NOT empty and already contains all target resources (get_managed_policy_default_version returns policy)
+    # The consumption role is partially-managed consumption role
+    # Check if the IAM role ( i.e. the consumption role ) gets attached at the end
+    with db.scoped_session() as session:
+        source_consumption_role.dataallManaged = PolicyManagementOptions.PARTIALLY_MANAGED.value
+        session.add(source_consumption_role)
+        session.commit()
+
+    policy = {
+        'Version': '2012-10-17',
+        'Statement': [
+            {
+                'Sid': f'{IAM_S3_BUCKETS_STATEMENT_SID}S31',
+                'Effect': 'Allow',
+                'Action': S3_ALLOWED_ACTIONS,
+                'Resource': [
+                    f'arn:aws:s3:::{dataset2.S3BucketName}',
+                    f'arn:aws:s3:::{dataset2.S3BucketName}/*',
+                    f'arn:aws:s3:::S3Bucket',
+                    f'arn:aws:s3:::S3Bucket/*',
+                ],
+            },
+            {
+                'Sid': f'{IAM_S3_BUCKETS_STATEMENT_SID}KMS1',
+                'Effect': 'Allow',
+                'Action': ['kms:*'],
+                'Resource': [f'arn:aws:kms:{dataset2.region}:{dataset2.AwsAccountId}:key/kms-key'],
+            },
+        ],
+    }
+
+    mocker.patch('dataall.base.aws.iam.IAM.create_managed_policy', return_value=True)
+
+    mocker.patch('dataall.base.aws.iam.IAM.get_attached_managed_policies_to_role', return_value=['policy-0'])
+
+    mocker.patch('dataall.base.aws.iam.IAM.list_policy_names_by_policy_pattern', return_value=['policy-0'])
+
+    mocker.patch('dataall.base.aws.iam.IAM.delete_managed_policy_non_default_versions', return_value=True)
+
+    mocker.patch('dataall.base.aws.iam.IAM.delete_managed_policy_by_name', return_value=True)
+
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService._get_managed_policy_quota',
+        return_value=10,
+    )
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.check_if_policy_exists',
+        return_value=False,
+    )
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.get_policies_unattached_to_role',
+        return_value=['policy-0'],
+    )
+
+    policy_attacher_mock = mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.attach_policies',
+        return_value=True,
+    )
+
+    kms_client = mock_kms_client(mocker)
+    kms_client().get_key_id.return_value = 'kms-key'
+
+    mocker.patch('dataall.base.aws.iam.IAM.get_managed_policy_default_version', return_value=('v1', policy))
+    mocker.patch('dataall.base.aws.iam.IAM.update_managed_policy_default_version', return_value=None)
+    share2_consumption_role_manager.grant_s3_iam_access()
+
+    policy_attacher_mock.assert_not_called()
+
 
 def test_grant_dataset_bucket_key_policy_with_complete_policy_present(
     mocker, share2: ShareObject, target_environment: Environment, share2_manager
@@ -1667,8 +1875,8 @@ def test_check_s3_iam_access_no_policy(mocker, dataset2, share2_manager):
 
 def test_check_s3_iam_access_policy_not_attached(mocker, dataset2, share2_manager):
     # Given
-    # There is not existing IAM policy in the requesters account for the dataset's S3bucket
-    # Check if the update_role_policy func is called and policy statements are added
+    # a share is made on a role and the policy is not attached
+    # Check if IAM policy not attached errors are raised
 
     policy = {
         'Version': '2012-10-17',
@@ -1714,6 +1922,161 @@ def test_check_s3_iam_access_policy_not_attached(mocker, dataset2, share2_manage
     assert 'IAM Policy attached Target Resource' in share2_manager.bucket_errors[0]
 
 
+def test_check_s3_iam_access_policy_fully_managed_consumption_role_not_attached(
+    mocker, dataset2, share2_consumption_role_manager
+):
+    # Given
+    # a share is made on a fully managed  consumption role and the policy is not attached
+    # Check if IAM policy not attached errors are raised
+
+    policy = {
+        'Version': '2012-10-17',
+        'Statement': [
+            {
+                'Sid': f'{IAM_S3_BUCKETS_STATEMENT_SID}S31',
+                'Effect': 'Allow',
+                'Action': ['s3:*'],
+                'Resource': [f'arn:aws:s3:::{dataset2.S3BucketName}', f'arn:aws:s3:::{dataset2.S3BucketName}/*'],
+            },
+            {
+                'Sid': f'{IAM_S3_BUCKETS_STATEMENT_SID}KMS1',
+                'Effect': 'Allow',
+                'Action': ['kms:*'],
+                'Resource': [f'arn:aws:kms:{dataset2.region}:{dataset2.AwsAccountId}:key/kms-key'],
+            },
+        ],
+    }
+
+    mocker.patch('dataall.base.aws.iam.IAM.get_managed_policy_default_version', return_value=('v1', policy))
+
+    # When policy does not exist
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.check_if_policy_exists',
+        return_value=True,
+    )
+
+    iam_update_role_policy_mock_1 = mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.get_policies_unattached_to_role',
+        return_value=['policy-0'],
+    )
+    mocker.patch('dataall.base.aws.iam.IAM.list_policy_names_by_policy_pattern', return_value=['policy-0'])
+
+    mocker.patch('dataall.base.aws.iam.IAM.get_attached_managed_policies_to_role', return_value=[])
+
+    kms_client = mock_kms_client(mocker)
+    kms_client().get_key_id.return_value = 'kms-key'
+    # When
+    share2_consumption_role_manager.check_s3_iam_access()
+    # Then
+    iam_update_role_policy_mock_1.assert_called_once()
+    assert (len(share2_consumption_role_manager.bucket_errors)) == 1
+    assert 'IAM Policy attached Target Resource' in share2_consumption_role_manager.bucket_errors[0]
+
+
+def test_check_s3_iam_access_policy_externally_managed_consumption_role_not_attached(
+    db, mocker, dataset2, source_consumption_role, share2_consumption_role_manager
+):
+    # Given
+    # a share is made on a externally managed  consumption role and the policy is not attached
+    # Check if IAM policy not attached errors are not raised
+
+    convert_role_management_to_type(db, source_consumption_role, PolicyManagementOptions.EXTERNALLY_MANAGED.value)
+
+    policy = {
+        'Version': '2012-10-17',
+        'Statement': [
+            {
+                'Sid': f'{IAM_S3_BUCKETS_STATEMENT_SID}S31',
+                'Effect': 'Allow',
+                'Action': ['s3:List*', 's3:GetObject', 's3:Describe*'],
+                'Resource': [f'arn:aws:s3:::{dataset2.S3BucketName}', f'arn:aws:s3:::{dataset2.S3BucketName}/*'],
+            },
+            {
+                'Sid': f'{IAM_S3_BUCKETS_STATEMENT_SID}KMS1',
+                'Effect': 'Allow',
+                'Action': ['kms:*'],
+                'Resource': [f'arn:aws:kms:{dataset2.region}:{dataset2.AwsAccountId}:key/kms-key'],
+            },
+        ],
+    }
+
+    mocker.patch('dataall.base.aws.iam.IAM.get_managed_policy_default_version', return_value=('v1', policy))
+
+    # When policy does not exist
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.check_if_policy_exists',
+        return_value=True,
+    )
+
+    iam_update_role_policy_mock_1 = mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.get_policies_unattached_to_role',
+        return_value=['policy-0'],
+    )
+    mocker.patch('dataall.base.aws.iam.IAM.list_policy_names_by_policy_pattern', return_value=['policy-0'])
+
+    mocker.patch('dataall.base.aws.iam.IAM.get_attached_managed_policies_to_role', return_value=[])
+
+    kms_client = mock_kms_client(mocker)
+    kms_client().get_key_id.return_value = 'kms-key'
+    # When
+    share2_consumption_role_manager.check_s3_iam_access()
+    # Then
+    iam_update_role_policy_mock_1.assert_not_called()
+
+
+def test_check_s3_iam_access_policy_partially_managed_consumption_role_not_attached(
+    db, mocker, dataset2, source_consumption_role, share2_consumption_role_manager
+):
+    # Given
+    # a share is made on a fully managed  consumption role and the policy is not attached
+    # Check if IAM policy not attached errors are raised
+
+    convert_role_management_to_type(db, source_consumption_role, PolicyManagementOptions.PARTIALLY_MANAGED.value)
+
+    policy = {
+        'Version': '2012-10-17',
+        'Statement': [
+            {
+                'Sid': f'{IAM_S3_BUCKETS_STATEMENT_SID}S31',
+                'Effect': 'Allow',
+                'Action': ['s3:List*', 's3:GetObject', 's3:Describe*'],
+                'Resource': [f'arn:aws:s3:::{dataset2.S3BucketName}', f'arn:aws:s3:::{dataset2.S3BucketName}/*'],
+            },
+            {
+                'Sid': f'{IAM_S3_BUCKETS_STATEMENT_SID}KMS1',
+                'Effect': 'Allow',
+                'Action': ['kms:*'],
+                'Resource': [f'arn:aws:kms:{dataset2.region}:{dataset2.AwsAccountId}:key/kms-key'],
+            },
+        ],
+    }
+
+    mocker.patch('dataall.base.aws.iam.IAM.get_managed_policy_default_version', return_value=('v1', policy))
+
+    # When policy does not exist
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.check_if_policy_exists',
+        return_value=True,
+    )
+
+    iam_update_role_policy_mock_1 = mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.get_policies_unattached_to_role',
+        return_value=['policy-0'],
+    )
+    mocker.patch('dataall.base.aws.iam.IAM.list_policy_names_by_policy_pattern', return_value=['policy-0'])
+
+    mocker.patch('dataall.base.aws.iam.IAM.get_attached_managed_policies_to_role', return_value=[])
+
+    kms_client = mock_kms_client(mocker)
+    kms_client().get_key_id.return_value = 'kms-key'
+    # When
+    share2_consumption_role_manager.check_s3_iam_access()
+    # Then
+    iam_update_role_policy_mock_1.assert_called_once()
+    assert (len(share2_consumption_role_manager.bucket_errors)) == 1
+    assert 'IAM Policy attached Target Resource' in share2_consumption_role_manager.bucket_errors[0]
+
+
 def test_check_s3_iam_access_missing_policy_statement(mocker, dataset2, share2_manager):
     # Given policy with some other bucket as resource
     # Check if the needed {IAM_S3_BUCKETS_STATEMENT_SID}S3 statement exists in the IAM policy
diff --git a/tests/modules/s3_datasets_shares/test_share.py b/tests/modules/s3_datasets_shares/test_share.py
index 1a1892333..e8ebeee39 100644
--- a/tests/modules/s3_datasets_shares/test_share.py
+++ b/tests/modules/s3_datasets_shares/test_share.py
@@ -9,6 +9,7 @@
 
 from dataall.base.utils.expiration_util import ExpirationUtils
 from dataall.base.utils.naming_convention import NamingConventionPattern, NamingConventionService
+from dataall.core.environment.db.environment_enums import PolicyManagementOptions
 from dataall.core.environment.db.environment_models import Environment, EnvironmentGroup, ConsumptionRole
 from dataall.core.organizations.db.organization_models import Organization
 from dataall.modules.shares_base.services.share_object_service import ShareObjectService
@@ -1472,12 +1473,12 @@ def test_create_share_object_share_policy_not_attached_attachMissingPolicies_ena
     assert create_share_object_response.data.createShareObject.requestPurpose == 'testShare'
 
 
-def test_create_share_object_share_policy_not_attached_attachMissingPolicies_disabled_dataallManaged(
+def test_create_share_object_share_policy_not_attached_attachMissingPolicies_fully_managed_dataallManaged(
     mocker, client, user2, group2, env2group, env2, dataset1
 ):
     # Given
     # Existing dataset, target environment and group
-    # SharePolicy exists and is NOT attached, attachMissingPolicies=True but principal=Group so managed=Trye
+    # SharePolicy exists and is NOT attached, attachMissingPolicies=False but principal=Group so managed=Fully-Managed
     # When a correct user creates request, data.all attaches the policy and the share creates successfully
     mocker.patch(
         'dataall.base.aws.iam.IAM.get_role_arn_by_name',
@@ -1515,12 +1516,61 @@ def test_create_share_object_share_policy_not_attached_attachMissingPolicies_dis
     assert create_share_object_response.data.createShareObject.requestPurpose == 'testShare'
 
 
-def test_create_share_object_share_policy_not_attached_attachMissingPolicies_disabled_dataallNotManaged(
+def test_create_share_object_share_policy_not_attached_attachMissingPolicies_not_fully_managed_dataallNotManaged(
     mocker, client, user2, group2, env2group, env2, dataset1
 ):
     # Given
     # Existing dataset, target environment and group
-    # SharePolicy exists and is NOT attached, attachMissingPolicies=True
+    # SharePolicy exists and is NOT attached, attachMissingPolicies=False and consumption role policy management = Partially-Managed ( thus managed = false )
+    # When a correct user creates request, data.all doesn't attach the policy and the share creates successfully
+    mocker.patch(
+        'dataall.base.aws.iam.IAM.get_role_arn_by_name',
+        return_value='role_arn',
+    )
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.check_if_policy_exists',
+        return_value=False,
+    )
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.get_policies_unattached_to_role',
+        return_value='policy-0',
+    )
+    attach_mocker = mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.attach_policies',
+        return_value=True,
+    )
+    mocker.patch('dataall.base.aws.iam.IAM.list_policy_names_by_policy_pattern', return_value=['policy-0'])
+
+    consumption_role = MagicMock(spec_set=ConsumptionRole)
+    consumption_role.IAMRoleName = 'randomName'
+    consumption_role.IAMRoleArn = 'randomArn'
+    consumption_role.dataallManaged = PolicyManagementOptions.PARTIALLY_MANAGED.value
+    mocker.patch(
+        'dataall.core.environment.services.environment_service.EnvironmentService.get_environment_consumption_role',
+        return_value=consumption_role,
+    )
+    create_share_object_response = create_share_object(
+        mocker=mocker,
+        client=client,
+        username=user2.username,
+        group=group2,
+        groupUri=env2group.groupUri,
+        environmentUri=env2.environmentUri,
+        datasetUri=dataset1.datasetUri,
+        attachMissingPolicies=False,
+        principalId=consumption_role.IAMRoleName,
+        principalType=PrincipalType.ConsumptionRole.value,
+    )
+
+    attach_mocker.assert_not_called()
+
+
+def test_create_share_object_share_policy_not_attached_attachMissingPolicies_fully_managed_dataallNotManaged_consumption_role(
+    mocker, client, user2, group2, env2group, env2, dataset1
+):
+    # Given
+    # Existing dataset, target environment and group
+    # SharePolicy exists and is NOT attached, attachMissingPolicies=False and consumption role policy management = Fully-Managed ( thus managed = True )
     # When a correct user creates request, data.all attaches the policy and the share creates successfully
     mocker.patch(
         'dataall.base.aws.iam.IAM.get_role_arn_by_name',
@@ -1534,12 +1584,16 @@ def test_create_share_object_share_policy_not_attached_attachMissingPolicies_dis
         'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.get_policies_unattached_to_role',
         return_value='policy-0',
     )
+    attach_mocker = mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.attach_policies',
+        return_value=True,
+    )
     mocker.patch('dataall.base.aws.iam.IAM.list_policy_names_by_policy_pattern', return_value=['policy-0'])
 
     consumption_role = MagicMock(spec_set=ConsumptionRole)
     consumption_role.IAMRoleName = 'randomName'
     consumption_role.IAMRoleArn = 'randomArn'
-    consumption_role.dataallManaged = False
+    consumption_role.dataallManaged = PolicyManagementOptions.FULLY_MANAGED.value
     mocker.patch(
         'dataall.core.environment.services.environment_service.EnvironmentService.get_environment_consumption_role',
         return_value=consumption_role,
@@ -1556,9 +1610,8 @@ def test_create_share_object_share_policy_not_attached_attachMissingPolicies_dis
         principalId=consumption_role.IAMRoleName,
         principalType=PrincipalType.ConsumptionRole.value,
     )
-    # Then share object is not created and an error appears
-    assert 'Required customer managed policies' in create_share_object_response.errors[0].message
-    assert 'are not attached to role randomName' in create_share_object_response.errors[0].message
+
+    attach_mocker.assert_called_once()
 
 
 def test_create_share_object_with_share_expiration_added(
@@ -1757,7 +1810,7 @@ def test_list_shares_to_me_approver(client, user, group, share1_draft):
     # When a user from the Approvers group lists the share objects sent to him
     get_share_requests_to_me_response = get_share_requests_to_me(client=client, user=user, group=group)
     # Then he sees the 2 shares
-    assert get_share_requests_to_me_response.data.getShareRequestsToMe.count == 4
+    assert get_share_requests_to_me_response.data.getShareRequestsToMe.count == 5
 
 
 def test_list_shares_to_me_requester(client, user2, group2, share1_draft):
@@ -1784,7 +1837,7 @@ def test_list_shares_from_me_requester(client, user2, group2, share1_draft):
     # When a user from the Requesters group lists the share objects sent from him
     get_share_requests_from_me_response = get_share_requests_from_me(client=client, user=user2, group=group2)
     # Then he sees the 2 shares
-    assert get_share_requests_from_me_response.data.getShareRequestsFromMe.count == 4
+    assert get_share_requests_from_me_response.data.getShareRequestsFromMe.count == 5
 
 
 def test_add_share_item(client, user2, group2, share1_draft, mock_glue_client):
@@ -2604,7 +2657,7 @@ def test_verify_dataset_share_objects_request(db, client, user, group, share3_pr
         client=client, user=user, group=group, filter={'datasets_uris': [dataset1.datasetUri]}
     )
 
-    assert list_dataset_shares.data.getShareRequestsToMe.count == 2
+    assert list_dataset_shares.data.getShareRequestsToMe.count == 3
     shareUris = [share.shareUri for share in list_dataset_shares.data.getShareRequestsToMe.nodes]
     assert len(shareUris)
     assert share3_processed.shareUri in shareUris
diff --git a/tests/permissions.py b/tests/permissions.py
index 0abded8fc..744cae9a1 100644
--- a/tests/permissions.py
+++ b/tests/permissions.py
@@ -1127,6 +1127,9 @@ def __post_init__(self):
     field_id('ShareObject', 'userRoleForShareObject'): TestData(
         resource_ignore=IgnoreReason.USERROLEINRESOURCE, tenant_ignore=IgnoreReason.NOTREQUIRED
     ),
+    field_id('ShareObject', 'policyManagement'): TestData(
+        resource_ignore=IgnoreReason.USERLIMITED, tenant_ignore=IgnoreReason.NOTREQUIRED
+    ),
     field_id('SharedDatabaseTableItem', 'sharedGlueDatabaseName'): TestData(
         resource_ignore=IgnoreReason.INTRAMODULE, tenant_ignore=IgnoreReason.NOTREQUIRED
     ),
diff --git a/tests_new/integration_tests/core/environment/queries.py b/tests_new/integration_tests/core/environment/queries.py
index c817815e2..bbcaf9e93 100644
--- a/tests_new/integration_tests/core/environment/queries.py
+++ b/tests_new/integration_tests/core/environment/queries.py
@@ -1,3 +1,5 @@
+from dataall.core.environment.db.environment_enums import PolicyManagementOptions
+
 ENV_TYPE = """
 environmentUri
 created
@@ -210,7 +212,9 @@ def remove_group_from_env(client, env_uri, group_uri):
     return response.data.removeGroupFromEnvironment
 
 
-def add_consumption_role(client, env_uri, group_uri, consumption_role_name, iam_role_arn, is_managed=True):
+def add_consumption_role(
+    client, env_uri, group_uri, consumption_role_name, iam_role_arn, is_managed=PolicyManagementOptions.FULLY_MANAGED
+):
     query = {
         'operationName': 'addConsumptionRoleToEnvironment',
         'variables': {
diff --git a/tests_new/integration_tests/modules/shares/s3_datasets_shares/global_conftest.py b/tests_new/integration_tests/modules/shares/s3_datasets_shares/global_conftest.py
index 69aae09fe..1fd94534a 100644
--- a/tests_new/integration_tests/modules/shares/s3_datasets_shares/global_conftest.py
+++ b/tests_new/integration_tests/modules/shares/s3_datasets_shares/global_conftest.py
@@ -50,13 +50,7 @@ def create_consumption_role(client, group, environment, environment_client, iam_
         iam_role_name,
         f'dataall-integration-tests-role-{environment.region}',
     )
-    return add_consumption_role(
-        client,
-        environment.environmentUri,
-        group,
-        cons_role_name,
-        role['Role']['Arn'],
-    )
+    return add_consumption_role(client, environment.environmentUri, group, cons_role_name, role['Role']['Arn'])
 
 
 # --------------SESSION PARAM FIXTURES----------------------------