in pce/validator/validation_suite.py [0:0]
def validate_roles(self, pce: PCE) -> ValidationResult:
"""
Ensure that the container task execution role has the proper policy (`TASK_POLICY`) among those attached to it.
"""
c = pce.pce_compute.container_definition
if not c:
return ValidationResult(
ValidationResultCode.ERROR,
ValidationErrorDescriptionTemplate.CLUSTER_DEFINITION_NOT_SET.value,
)
policies = self.iam_gateway.get_policies_for_role(c.task_role_id)
if not policies:
pce_id = c.tags[PCE_ID_KEY]
return ValidationResult(
ValidationResultCode.ERROR,
ValidationErrorDescriptionTemplate.ROLE_POLICIES_NOT_FOUND.value.format(
role_names=c.task_role_id
),
ValidationErrorSolutionHintTemplate.ROLE_POLICIES_NOT_FOUND.value.format(
role_names=c.task_role_id, pce_id=pce_id
),
)
policy_name_found = None
for policy_name, policy_contents in policies.attached_policy_contents.items():
if TASK_POLICY == policy_contents:
policy_name_found = policy_name
break
if not policy_name_found:
return ValidationResult(
ValidationResultCode.ERROR,
ValidationErrorDescriptionTemplate.ROLE_WRONG_POLICY.value.format(
role_name=c.task_role_id,
policy_names=",".join(policies.attached_policy_contents.keys()),
),
ValidationErrorSolutionHintTemplate.ROLE_WRONG_POLICY.value.format(
role_name=c.task_role_id,
role_policy=TASK_POLICY,
),
)
if len(policies.attached_policy_contents.values()) > 1:
return ValidationResult(
ValidationResultCode.WARNING,
ValidationWarningDescriptionTemplate.MORE_POLICIES_THAN_EXPECTED.value.format(
policy_names=",".join(
policies.attached_policy_contents.keys() - {policy_name_found}
),
role_id=c.task_role_id,
),
ValidationWarningSolutionHintTemplate.MORE_POLICIES_THAN_EXPECTED.value,
)
return ValidationResult(ValidationResultCode.SUCCESS)