feat(rolesanywhere): flag profiles with unscoped sessions (#12416)

This commit is contained in:
Pedro Martín
2026-08-13 17:06:24 +02:00
committed by GitHub
parent dd882c70e7
commit 0758c3585d
17 changed files with 1812 additions and 0 deletions
@@ -42,6 +42,7 @@
"lightsail:GetRelationalDatabases",
"macie2:GetMacieSession",
"macie2:GetAutomatedDiscoveryConfiguration",
"rolesanywhere:ListProfiles",
"rolesanywhere:ListTagsForResource",
"rolesanywhere:ListTrustAnchors",
"s3:GetAccountPublicAccessBlock",
@@ -213,6 +213,7 @@ Resources:
- "lightsail:GetRelationalDatabases"
- "macie2:GetMacieSession"
- "macie2:GetAutomatedDiscoveryConfiguration"
- "rolesanywhere:ListProfiles"
- "rolesanywhere:ListTagsForResource"
- "rolesanywhere:ListTrustAnchors"
- "s3:GetAccountPublicAccessBlock"
@@ -0,0 +1 @@
Add the `iam_workload_identity_pool_provider_attribute_condition` check to flag GCP Workload Identity Federation providers that trust a multi-tenant issuer without an attribute condition restricting which external identities can impersonate federated principals
@@ -0,0 +1 @@
Add the `rolesanywhere_profile_restricts_session_permissions` check to flag AWS IAM Roles Anywhere profiles that reference an administrative role without scoping down the vended session with a session policy or managed policies
@@ -0,0 +1,41 @@
{
"Provider": "aws",
"CheckID": "rolesanywhere_profile_restricts_session_permissions",
"CheckTitle": "IAM Roles Anywhere profiles scope down the vended session permissions",
"CheckType": [
"Software and Configuration Checks/AWS Security Best Practices"
],
"ServiceName": "rolesanywhere",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "AwsRolesAnywhereProfile",
"ResourceGroup": "security",
"Description": "**IAM Roles Anywhere profiles** that reference an administrative role are assessed for **session scoping**. A profile defining neither an inline `sessionPolicy` nor `managedPolicyArns` vends credentials with the full permissions of its roles. It is flagged only when a referenced role is administrative, since an unscoped session on a least-privilege role is already constrained.",
"Risk": "Roles Anywhere profiles bind X.509 certificates to IAM roles. When a profile references an administrative role and does not scope the session, vended credentials carry full administrative permissions. An attacker presenting a valid certificate - or planting a rogue trust anchor and profile - gains durable privileged access that rotating IAM keys does not revoke.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://docs.aws.amazon.com/rolesanywhere/latest/userguide/introduction.html",
"https://docs.aws.amazon.com/rolesanywhere/latest/APIReference/API_CreateProfile.html",
"https://docs.aws.amazon.com/rolesanywhere/latest/userguide/security-best-practices.html"
],
"Remediation": {
"Code": {
"CLI": "aws rolesanywhere update-profile --profile-id <profile_id> --session-policy '{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Action\":[\"<least_privilege_actions>\"],\"Resource\":[\"<scoped_resources>\"]}]}'",
"NativeIaC": "```yaml\nResources:\n <example_resource_name>:\n Type: AWS::RolesAnywhere::Profile\n Properties:\n Name: scoped-profile\n Enabled: true\n RoleArns:\n - <role_arn>\n SessionPolicy: '{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Action\":[\"<least_privilege_actions>\"],\"Resource\":[\"<scoped_resources>\"]}]}' # FIX: scope down the vended session\n```",
"Other": "1. Identify the least-privilege actions the workload actually needs\n2. Attach a sessionPolicy or managedPolicyArns to the Roles Anywhere profile that grants only those actions\n3. Prefer purpose-built roles per workload over broad roles referenced by many profiles\n4. Review trust anchors and profiles regularly for entries you did not create",
"Terraform": "```hcl\nresource \"aws_rolesanywhere_profile\" \"<example_resource_name>\" {\n name = \"scoped-profile\"\n enabled = true\n role_arns = [<role_arn>]\n session_policy = jsonencode({\n Version = \"2012-10-17\"\n Statement = [{\n Effect = \"Allow\"\n Action = [<least_privilege_actions>]\n Resource = [<scoped_resources>]\n }]\n }) # FIX: scope down the vended session\n}\n```"
},
"Recommendation": {
"Text": "Attach a session policy or managed policies to every enabled IAM Roles Anywhere profile so the vended credentials are scoped below the referenced role's permissions. Pair each profile with a purpose-built least-privilege role and audit trust anchors and profiles regularly for unexpected entries.",
"Url": "https://hub.prowler.com/check/rolesanywhere_profile_restricts_session_permissions"
}
},
"Categories": [
"identity-access",
"trust-boundaries"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "A profile is failed only when it is enabled, its session is unscoped, and a referenced role effectively grants administrative (*:*) access. The session-policy set (inline sessionPolicy plus every managedPolicyArns entry, resolved to its policy document) is evaluated as a union: any member granting *:* leaves the session unrestricted. Role classification merges all attached and inline identity-policy documents so explicit denies negate allows across policies, excludes condition-guarded statements (not statically provable), and intersects the result with the role's permissions boundary: a role whose boundary does not grant *:* - or whose boundary document cannot be resolved - is not classified as administrative. Disabled profiles, scoped profiles, and profiles with no role identified as administrative are reported as PASS. Referenced roles absent from the IAM inventory (for example cross-account roles or denied ListRoles) and policy documents that could not be collected are treated as non-administrative to avoid false positives."
}
@@ -0,0 +1,272 @@
import json
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.lib.logger import logger
from prowler.providers.aws.services.iam.iam_client import iam_client
from prowler.providers.aws.services.iam.lib.policy import check_full_service_access
from prowler.providers.aws.services.rolesanywhere.rolesanywhere_client import (
rolesanywhere_client,
)
# AWS-managed AdministratorAccess ARN suffix, partition-agnostic
# (arn:aws:..., arn:aws-cn:..., arn:aws-us-gov:...).
ADMIN_POLICY_ARN_SUFFIX = ":iam::aws:policy/AdministratorAccess"
# Synthetic statement equivalent to the AWS-managed AdministratorAccess
# document, used when a policy is identified by that ARN but its document was
# not collected by the IAM service.
_ADMIN_STATEMENT = {"Effect": "Allow", "Action": "*", "Resource": "*"}
def _grants_full_access(document) -> bool:
"""Return True when a policy document grants administrative (``*:*``) access.
Args:
document: Decoded IAM policy document, or None when unavailable.
"""
if not document:
return False
try:
return check_full_service_access("*", document)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
def _full_access_status(documents) -> bool | None:
"""Return whether a policy-document set grants ``*:*``, or None when unknown.
Merges every condition-free statement across the given documents into a
single evaluation so an explicit deny in one document negates an allow in
another (Deny > Allow via the shared policy-evaluation helpers).
Unresolved or malformed documents and context-dependent semantics cannot
produce a definitive classification and propagate as None:
- a document that is missing or not a well-formed statement container;
- a ``Condition``-guarded Deny that could negate an otherwise proven
full-access grant;
- a ``Condition``-guarded Allow that could grant full access not proven
by the unconditional statements.
Args:
documents: Iterable of decoded IAM policy documents (None members mark
documents that could not be resolved).
"""
statements = []
conditional_effects = set()
for document in documents:
if not isinstance(document, dict) or "Statement" not in document:
return None
document_statements = document.get("Statement", [])
if not isinstance(document_statements, list):
document_statements = [document_statements]
for statement in document_statements:
if not isinstance(statement, dict):
return None
effect = str(statement.get("Effect", "")).lower()
if (
effect not in {"allow", "deny"}
or not ("Action" in statement or "NotAction" in statement)
or not ("Resource" in statement or "NotResource" in statement)
):
return None
if statement.get("Condition"):
conditional_effects.add(effect)
else:
statements.append(statement)
grants_full_access = _grants_full_access({"Statement": statements})
if (grants_full_access and "deny" in conditional_effects) or (
not grants_full_access and "allow" in conditional_effects
):
return None
return grants_full_access
def _role_is_privileged(role, policies) -> bool | None:
"""Return whether an IAM role is administrative, or None when unknown.
Effective permissions are the intersection of the role's identity policies
(attached and inline, evaluated together) and its permissions boundary.
Unresolved policy documents, malformed policies, and condition-guarded
statements that could change the outcome propagate as None instead of
being collapsed into a definitive classification.
Args:
role: An ``iam_service.Role`` referenced by a Roles Anywhere profile.
policies: Mapping of policy ARN to ``iam_service.Policy`` from iam_client.
"""
documents = []
for attached in role.attached_policies:
policy_arn = attached.get("PolicyArn", "")
document = getattr(policies.get(policy_arn), "document", None)
if policy_arn.endswith(ADMIN_POLICY_ARN_SUFFIX) and not document:
documents.append({"Statement": [_ADMIN_STATEMENT]})
else:
documents.append(document)
for inline_name in role.inline_policies:
policy = policies.get(f"{role.arn}:policy/{inline_name}")
documents.append(getattr(policy, "document", None))
identity_status = _full_access_status(documents)
if identity_status is False:
# Identity policies provably do not grant *:*; no boundary can widen them.
return False
boundary = getattr(role, "permissions_boundary", None)
if not boundary:
return identity_status
boundary_arn = (
boundary.get("PermissionsBoundaryArn", "") if isinstance(boundary, dict) else ""
)
if boundary_arn.endswith(ADMIN_POLICY_ARN_SUFFIX):
# An AdministratorAccess boundary restricts nothing.
return identity_status
boundary_status = _full_access_status(
[getattr(policies.get(boundary_arn), "document", None)]
)
if boundary_status is False:
# The boundary provably does not grant *:*: the intersection cannot be
# administrative regardless of the identity policies.
return False
if boundary_status is None:
return None
return identity_status
def _session_is_scoped(profile, policies) -> bool | None:
"""Return whether session policies restrict permissions, or None when unknown.
AWS evaluates the inline ``sessionPolicy`` and every ``managedPolicyArns``
entry together as a single session-policy category, so the complete set is
merged into one evaluation: the session is scoped only when at least one
session policy exists and the set does not grant ``*:*``. Managed entries
are resolved through the collected IAM policies. An invalid inline policy
or an unresolved managed policy does not prove that the session is
restricted and propagates as None.
Args:
profile: A ``rolesanywhere_service.Profile``.
policies: Mapping of policy ARN to ``iam_service.Policy`` from iam_client.
"""
if not profile.session_policy and not profile.managed_policy_arns:
return False
documents = []
if profile.session_policy:
try:
documents.append(json.loads(profile.session_policy))
except (ValueError, TypeError):
return None
for arn in profile.managed_policy_arns or []:
if arn.endswith(ADMIN_POLICY_ARN_SUFFIX):
documents.append({"Statement": [_ADMIN_STATEMENT]})
else:
documents.append(getattr(policies.get(arn), "document", None))
grants_full_access = _full_access_status(documents)
return None if grants_full_access is None else not grants_full_access
class rolesanywhere_profile_restricts_session_permissions(Check):
"""Flag Roles Anywhere profiles that vend an unscoped session on a privileged role.
A Roles Anywhere profile that does not restrict the session with an inline
``sessionPolicy`` or ``managedPolicyArns`` vends temporary credentials
carrying the full permissions of every role it references. This is only a
real risk when a referenced role is itself administrative: any certificate
accepted by the trust anchor then wields administrative permissions, turning
the profile into a durable privileged-access path that surviving key rotation
does not remove. Profiles that scope the session, whose referenced roles were
proven not administrative, or that are disabled are reported as PASS. When
session scoping or role permissions cannot be evaluated (unresolved or
invalid policy documents, condition-guarded grants, unknown roles), the
report is MANUAL rather than a proven outcome.
"""
def execute(self) -> list[Check_Report_AWS]:
"""Evaluate session-permission scoping for Roles Anywhere profiles.
Returns:
list[Check_Report_AWS]: One report per Roles Anywhere profile. FAIL
for enabled, unscoped profiles that reference a proven administrative
role; MANUAL when session scoping or role permissions could not be
evaluated; PASS for scoped profiles, profiles whose roles were proven
not administrative, and disabled profiles.
"""
findings = []
roles_by_arn = {role.arn: role for role in iam_client.roles}
for profile in rolesanywhere_client.profiles.values():
report = Check_Report_AWS(metadata=self.metadata(), resource=profile)
role_statuses = {
arn: (
_role_is_privileged(roles_by_arn[arn], iam_client.policies)
if arn in roles_by_arn
else None
)
for arn in profile.role_arns
}
privileged_role_arns = [
arn for arn, status in role_statuses.items() if status is True
]
unknown_role_arns = [
arn for arn, status in role_statuses.items() if status is None
]
session_scoped = _session_is_scoped(profile, iam_client.policies)
if not profile.enabled:
report.status = "PASS"
report.status_extended = (
f"IAM Roles Anywhere profile {profile.name} is disabled and "
"cannot vend session credentials."
)
elif session_scoped is True:
report.status = "PASS"
report.status_extended = (
f"IAM Roles Anywhere profile {profile.name} restricts vended "
"session permissions with a session policy or managed policies."
)
elif session_scoped is None:
report.status = "MANUAL"
report.status_extended = (
f"IAM Roles Anywhere profile {profile.name} session scoping "
"could not be evaluated because an inline or managed session "
"policy was invalid or unresolved."
)
elif privileged_role_arns:
report.status = "FAIL"
report.status_extended = (
f"IAM Roles Anywhere profile {profile.name} does not scope down "
"sessions and references administrative role(s) "
f"{', '.join(privileged_role_arns)}; certificates authenticated "
"through it inherit administrative permissions, enabling durable "
"privileged access."
)
elif unknown_role_arns:
report.status = "MANUAL"
report.status_extended = (
f"IAM Roles Anywhere profile {profile.name} does not scope down "
"sessions, and the effective permissions of referenced role(s) "
f"{', '.join(unknown_role_arns)} could not be evaluated."
)
else:
report.status = "PASS"
report.status_extended = (
f"IAM Roles Anywhere profile {profile.name} does not scope down "
"sessions, but no referenced role was identified as "
"administrative; scoping the session is recommended as "
"defense-in-depth."
)
findings.append(report)
return findings
@@ -11,7 +11,9 @@ class RolesAnywhere(AWSService):
def __init__(self, provider):
super().__init__(__class__.__name__, provider)
self.trust_anchors = {}
self.profiles = {}
self.__threading_call__(self._list_trust_anchors)
self.__threading_call__(self._list_profiles)
def _list_trust_anchors(self, regional_client):
logger.info("RolesAnywhere - Listing Trust Anchors...")
@@ -52,6 +54,53 @@ class RolesAnywhere(AWSService):
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _list_profiles(self, regional_client):
"""List and cache IAM Roles Anywhere profiles for one AWS Region.
Args:
regional_client: Roles Anywhere client for the audited Region.
"""
logger.info("RolesAnywhere - Listing Profiles...")
try:
paginator = regional_client.get_paginator("list_profiles")
for page in paginator.paginate():
for profile in page.get("profiles", []):
arn = profile.get("profileArn", "")
if not arn:
continue
if self.audit_resources and not is_resource_filtered(
arn, self.audit_resources
):
continue
tags = []
try:
tags = regional_client.list_tags_for_resource(
resourceArn=arn
).get("tags", [])
except Exception as error:
logger.warning(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
self.profiles[arn] = Profile(
arn=arn,
id=profile.get("profileId", ""),
name=profile.get("name", ""),
region=regional_client.region,
enabled=profile.get("enabled", False),
role_arns=profile.get("roleArns", []) or [],
session_policy=profile.get("sessionPolicy", "") or "",
managed_policy_arns=profile.get("managedPolicyArns", []) or [],
duration_seconds=profile.get("durationSeconds", 0) or 0,
accept_role_session_name=profile.get(
"acceptRoleSessionName", False
),
tags=tags,
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
class TrustAnchor(BaseModel):
arn: str
@@ -62,3 +111,19 @@ class TrustAnchor(BaseModel):
source_type: str = ""
acm_pca_arn: str = ""
tags: List[Dict[str, str]] = Field(default_factory=list)
class Profile(BaseModel):
"""Represent an IAM Roles Anywhere profile."""
arn: str
id: str
name: str
region: str
enabled: bool = False
role_arns: List[str] = Field(default_factory=list)
session_policy: str = ""
managed_policy_arns: List[str] = Field(default_factory=list)
duration_seconds: int = 0
accept_role_session_name: bool = False
tags: List[Dict[str, str]] = Field(default_factory=list)
@@ -17,6 +17,8 @@ class IAM(GCPService):
self.service_accounts = []
self._get_service_accounts()
self._get_service_accounts_keys()
self.workload_identity_pool_providers = []
self._get_workload_identity_pool_providers()
def _get_service_accounts(self):
for project_id in self.project_ids:
@@ -87,6 +89,94 @@ class IAM(GCPService):
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _get_workload_identity_pool_providers(self):
for project_id in self.project_ids:
try:
pools_request = (
self.client.projects()
.locations()
.workloadIdentityPools()
.list(parent=f"projects/{project_id}/locations/global")
)
while pools_request is not None:
pools_response = pools_request.execute(
num_retries=DEFAULT_RETRY_ATTEMPTS
)
for pool in pools_response.get("workloadIdentityPools", []):
self._get_providers_for_pool(project_id, pool)
pools_request = (
self.client.projects()
.locations()
.workloadIdentityPools()
.list_next(
previous_request=pools_request,
previous_response=pools_response,
)
)
except Exception as error:
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _get_providers_for_pool(self, project_id, pool):
try:
pool_name = pool.get("name", "")
pool_id = pool_name.split("/")[-1]
# A provider can remain ACTIVE while its parent pool is disabled or
# soft-deleted; a disabled pool cannot vend credentials, so the
# pool's effective availability must travel with the provider.
pool_disabled = (
pool.get("disabled", False) or pool.get("state", "ACTIVE") != "ACTIVE"
)
request = (
self.client.projects()
.locations()
.workloadIdentityPools()
.providers()
.list(parent=pool_name)
)
while request is not None:
response = request.execute(num_retries=DEFAULT_RETRY_ATTEMPTS)
for provider in response.get("workloadIdentityPoolProviders", []):
provider_type = next(
(
key
for key in ("oidc", "aws", "saml", "x509")
if key in provider
),
"",
)
self.workload_identity_pool_providers.append(
WorkloadIdentityPoolProvider(
name=provider.get("name", ""),
id=provider.get("name", "").split("/")[-1],
pool_id=pool_id,
pool_disabled=pool_disabled,
project_id=project_id,
state=provider.get("state", ""),
disabled=provider.get("disabled", False),
attribute_condition=provider.get("attributeCondition", ""),
attribute_mapping=provider.get("attributeMapping", {})
or {},
provider_type=provider_type,
issuer_uri=(provider.get("oidc", {}) or {}).get(
"issuerUri", ""
),
display_name=provider.get("displayName", ""),
)
)
request = (
self.client.projects()
.locations()
.workloadIdentityPools()
.providers()
.list_next(previous_request=request, previous_response=response)
)
except Exception as error:
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
class Key(BaseModel):
name: str
@@ -106,6 +196,25 @@ class ServiceAccount(BaseModel):
disabled: bool = False
class WorkloadIdentityPoolProvider(BaseModel):
"""Represent a GCP Workload Identity Federation pool provider."""
name: str
id: str
pool_id: str
# True when the parent pool is disabled or not ACTIVE; such a pool cannot
# vend credentials regardless of the provider's own state.
pool_disabled: bool = False
project_id: str
state: str = ""
disabled: bool = False
attribute_condition: str = ""
attribute_mapping: dict = {}
provider_type: str = ""
issuer_uri: str = ""
display_name: str = ""
class AccessApproval(GCPService):
def __init__(self, provider: GcpProvider):
super().__init__(__class__.__name__, provider)
@@ -0,0 +1,38 @@
{
"Provider": "gcp",
"CheckID": "iam_workload_identity_pool_provider_attribute_condition",
"CheckTitle": "Workload Identity Federation providers trusting a multi-tenant issuer enforce an attribute condition",
"CheckType": [],
"ServiceName": "iam",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "high",
"ResourceType": "iam.googleapis.com/WorkloadIdentityPoolProvider",
"Description": "**Workload Identity Federation providers** define an `attributeCondition` (CEL) restricting which external identities may impersonate Google Cloud principals. When a provider trusts a **multi-tenant issuer** (GitHub Actions, GitLab.com and other shared issuers), omitting the condition trusts every identity that issuer can mint. Providers trusting a dedicated single-tenant issuer are not flagged.",
"Risk": "A provider trusting a multi-tenant issuer without an attribute condition accepts any external identity from that issuer - for example any GitHub repository when the issuer is GitHub Actions. An attacker controlling any tenant on that platform can authenticate through the provider and exchange tokens for federated credentials. Because no key is stored, this access survives credential rotation.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://cloud.google.com/iam/docs/workload-identity-federation",
"https://cloud.google.com/iam/docs/workload-identity-federation#mapping",
"https://cloud.google.com/iam/docs/best-practices-for-using-workload-identity-federation"
],
"Remediation": {
"Code": {
"CLI": "gcloud iam workload-identity-pools providers update-<oidc|aws|saml|x509> <PROVIDER_ID> --location=global --workload-identity-pool=<POOL_ID> --attribute-condition=\"<CEL_condition>\" # use the update subcommand matching the provider type",
"NativeIaC": "",
"Other": "1. In the Google Cloud console, go to IAM & Admin > Workload Identity Federation\n2. Open the affected pool and provider\n3. Set an attribute condition (CEL) restricting the allowed external identities, using an assertion that fits the provider type: OIDC by subject or claim (assertion.sub), SAML by NameID or attribute (assertion.subject or assertion.attributes[...]), AWS by account/role (assertion.account or assertion.arn), X.509 by certificate subject\n4. For OIDC providers, also restrict allowedAudiences to your own audience\n5. Review providers regularly for entries you did not create",
"Terraform": "```hcl\nresource \"google_iam_workload_identity_pool_provider\" \"example\" {\n workload_identity_pool_id = \"my-pool\"\n workload_identity_pool_provider_id = \"my-provider\"\n attribute_condition = \"assertion.repository_owner == 'my-org'\" # FIX: restrict trusted identities\n\n # Declare exactly one of oidc {}, aws {}, saml {}, or x509 {} to match the provider type, e.g.:\n oidc {\n issuer_uri = \"https://token.actions.githubusercontent.com\"\n allowed_audiences = [\"https://my-audience.example.com\"]\n }\n}\n```"
},
"Recommendation": {
"Text": "Set an attribute condition on any active Workload Identity Federation provider that trusts a multi-tenant issuer so only the intended external identities can impersonate Google Cloud principals, and pair OIDC providers with a restricted audience. Providers trusting a dedicated single-tenant issuer should still add a condition as defense-in-depth. Audit pools and providers regularly for unexpected entries.",
"Url": "https://hub.prowler.com/check/iam_workload_identity_pool_provider_attribute_condition"
}
},
"Categories": [
"identity-access",
"trust-boundaries"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "Only active providers that trust a known multi-tenant issuer (GitHub Actions, GitLab.com, Google, HCP Terraform) are failed when they omit an attribute condition; providers trusting a dedicated single-tenant issuer, AWS/SAML/X.509 providers, and disabled or non-ACTIVE providers are reported as PASS. This check verifies that an attribute condition is present; it does not evaluate whether the condition's expression is sufficiently restrictive."
}
@@ -0,0 +1,107 @@
from urllib.parse import urlparse
from prowler.lib.check.models import Check, Check_Report_GCP
from prowler.providers.gcp.services.iam.iam_client import iam_client
# OIDC issuers whose tokens are minted for many independent tenants (any GitHub
# repository, any GitLab project, any Google account, ...). A provider that
# trusts one of these without an ``attributeCondition`` accepts identities
# outside the operator's control, so omitting the condition genuinely expands
# trust. A dedicated, single-tenant issuer only vends tokens to the operator's
# own workloads, so an attribute condition there is defense-in-depth rather than
# a requirement (see Google's guidance for GitHub and other shared issuers).
MULTI_TENANT_OIDC_ISSUER_HOSTS = {
"token.actions.githubusercontent.com", # GitHub Actions (any repository)
"gitlab.com", # GitLab.com SaaS (any project)
"accounts.google.com", # any Google account
"app.terraform.io", # HCP Terraform (any organization)
}
def _is_multi_tenant_issuer(issuer_uri: str) -> bool:
"""Return True when the OIDC issuer is a known multi-tenant/shared issuer."""
if not issuer_uri:
return False
# hostname lowercases and strips port/userinfo (gitlab.com:443, user@host);
# fall back to the raw string for bare hosts without a scheme.
host = urlparse(issuer_uri).hostname or issuer_uri.lower()
return host in MULTI_TENANT_OIDC_ISSUER_HOSTS
class iam_workload_identity_pool_provider_attribute_condition(Check):
"""Ensure WIF providers trusting a multi-tenant issuer enforce an attribute condition.
A workload identity pool provider that trusts a multi-tenant issuer (GitHub
Actions, GitLab.com, ...) without an ``attributeCondition`` accepts every
external identity that issuer can mint. An attacker controlling any tenant on
that platform can then authenticate through the provider and exchange tokens
for federated credentials, surviving credential rotation. Providers that
enforce an attribute condition, that trust a dedicated single-tenant issuer,
that are not OIDC-based, that are disabled/inactive, or whose parent pool is
disabled are reported as PASS.
"""
def execute(self) -> list[Check_Report_GCP]:
"""Evaluate the attribute condition of each Workload Identity provider.
Returns:
list[Check_Report_GCP]: One report per workload identity pool
provider. FAIL for active providers that trust a multi-tenant issuer
without an attribute condition; PASS for providers that enforce one,
trust a dedicated issuer, are not OIDC-based, or are
disabled/inactive.
"""
findings = []
for provider in iam_client.workload_identity_pool_providers:
report = Check_Report_GCP(
metadata=self.metadata(),
resource=provider,
resource_id=provider.name,
resource_name=provider.display_name or provider.id,
location="global",
)
if provider.pool_disabled:
report.status = "PASS"
report.status_extended = (
f"Workload Identity Federation provider {provider.id} belongs "
f"to the disabled pool {provider.pool_id}, which cannot vend "
"credentials."
)
elif provider.disabled or provider.state != "ACTIVE":
report.status = "PASS"
report.status_extended = (
f"Workload Identity Federation provider {provider.id} in pool "
f"{provider.pool_id} is not active and cannot vend credentials."
)
elif provider.attribute_condition:
report.status = "PASS"
report.status_extended = (
f"Workload Identity Federation provider {provider.id} in pool "
f"{provider.pool_id} enforces an attribute condition."
)
elif _is_multi_tenant_issuer(provider.issuer_uri):
report.status = "FAIL"
report.status_extended = (
f"Workload Identity Federation provider {provider.id} in pool "
f"{provider.pool_id} trusts the multi-tenant issuer "
f"{provider.issuer_uri} without an attribute condition, so any "
"identity from that issuer can authenticate through this provider."
)
elif provider.provider_type != "oidc":
report.status = "PASS"
report.status_extended = (
f"Workload Identity Federation provider {provider.id} in pool "
f"{provider.pool_id} is not an OIDC provider trusting a "
"multi-tenant issuer; an attribute condition is recommended as "
"defense-in-depth but not required."
)
else:
report.status = "PASS"
report.status_extended = (
f"Workload Identity Federation provider {provider.id} in pool "
f"{provider.pool_id} trusts a dedicated issuer; an attribute "
"condition is recommended as defense-in-depth but not required."
)
findings.append(report)
return findings
@@ -0,0 +1,762 @@
from types import SimpleNamespace
from unittest import mock
from prowler.providers.aws.services.rolesanywhere.rolesanywhere_service import Profile
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
PROFILE_ID = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
PROFILE_NAME = "workload-profile"
PROFILE_ARN = f"arn:aws:rolesanywhere:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:profile/{PROFILE_ID}"
ADMIN_ROLE_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/admin-role"
READONLY_ROLE_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/workload-role"
UNKNOWN_ROLE_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/cross-account-role"
CUSTOM_ADMIN_ROLE_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/custom-admin-role"
INLINE_ADMIN_ROLE_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/inline-admin-role"
NAME_COLLISION_ROLE_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/name-collision-role"
UNRESOLVED_POLICY_ROLE_ARN = (
f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/unresolved-policy-role"
)
ADMIN_UNRESOLVED_POLICY_ROLE_ARN = (
f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/admin-unresolved-policy-role"
)
INVALID_POLICY_ROLE_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/invalid-policy-role"
DENY_OVERRIDE_ROLE_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/deny-override-role"
CONDITIONAL_DENY_ROLE_ARN = (
f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/conditional-deny-role"
)
CONDITIONAL_ADMIN_ROLE_ARN = (
f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/conditional-admin-role"
)
BOUNDED_ADMIN_ROLE_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/bounded-admin-role"
UNRESOLVED_BOUNDARY_ROLE_ARN = (
f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/unresolved-boundary-role"
)
ADMIN_BOUNDED_ROLE_ARN = (
f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/admin-bounded-admin-role"
)
AWS_ADMIN_POLICY_ARN = "arn:aws:iam::aws:policy/AdministratorAccess"
CUSTOMER_ADMIN_NAMED_POLICY_ARN = (
f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:policy/AdministratorAccess"
)
CUSTOM_ADMIN_POLICY_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:policy/custom-admin"
MANAGED_POLICY_ARN = "arn:aws:iam::aws:policy/ReadOnlyAccess"
CUSTOMER_FULL_ACCESS_POLICY_ARN = (
f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:policy/full-access-session"
)
BOUNDARY_POLICY_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:policy/scoped-boundary"
UNRESOLVED_BOUNDARY_POLICY_ARN = (
f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:policy/unresolved-boundary"
)
UNRESOLVED_SESSION_POLICY_ARN = (
"arn:aws:iam::aws:policy/job-function/SupportUser" # not in iam_client.policies
)
SESSION_POLICY = (
'{"Version":"2012-10-17","Statement":[{"Effect":"Allow",'
'"Action":["s3:GetObject"],"Resource":["*"]}]}'
)
FULL_ACCESS_SESSION_POLICY = (
'{"Version":"2012-10-17","Statement":[{"Effect":"Allow",'
'"Action":"*","Resource":"*"}]}'
)
FULL_ACCESS_DOCUMENT = {
"Version": "2012-10-17",
"Statement": [{"Effect": "Allow", "Action": "*", "Resource": "*"}],
}
READONLY_DOCUMENT = {
"Version": "2012-10-17",
"Statement": [{"Effect": "Allow", "Action": "s3:Get*", "Resource": "*"}],
}
DENY_ALL_DOCUMENT = {
"Version": "2012-10-17",
"Statement": [{"Effect": "Deny", "Action": "*", "Resource": "*"}],
}
CONDITIONAL_ADMIN_DOCUMENT = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "*",
"Resource": "*",
"Condition": {"Bool": {"aws:MultiFactorAuthPresent": "true"}},
}
],
}
CONDITIONAL_DENY_DOCUMENT = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Deny",
"Action": "*",
"Resource": "*",
"Condition": {"StringNotEquals": {"aws:PrincipalTag/team": "security"}},
}
],
}
def _profile(
*,
enabled: bool = True,
session_policy: str = "",
managed_policy_arns=None,
role_arns=None,
):
return Profile(
arn=PROFILE_ARN,
id=PROFILE_ID,
name=PROFILE_NAME,
region=AWS_REGION_US_EAST_1,
enabled=enabled,
role_arns=role_arns if role_arns is not None else [READONLY_ROLE_ARN],
session_policy=session_policy,
managed_policy_arns=managed_policy_arns or [],
)
def _role(arn, attached_policies=None, inline_policies=None, permissions_boundary=None):
return SimpleNamespace(
arn=arn,
attached_policies=attached_policies or [],
inline_policies=inline_policies or [],
permissions_boundary=permissions_boundary,
)
def _iam_client():
"""IAM client stub mirroring iam_service models: roles with attached/inline
policies and a policies dict keyed by ARN (inline keyed {role_arn}:policy/{name}).
"""
roles = [
_role(
ADMIN_ROLE_ARN,
attached_policies=[
{"PolicyName": "AdministratorAccess", "PolicyArn": AWS_ADMIN_POLICY_ARN}
],
),
_role(
READONLY_ROLE_ARN,
attached_policies=[
{"PolicyName": "ReadOnlyAccess", "PolicyArn": MANAGED_POLICY_ARN}
],
),
_role(
CUSTOM_ADMIN_ROLE_ARN,
attached_policies=[
{"PolicyName": "custom-admin", "PolicyArn": CUSTOM_ADMIN_POLICY_ARN}
],
),
_role(INLINE_ADMIN_ROLE_ARN, inline_policies=["inline-admin"]),
_role(
NAME_COLLISION_ROLE_ARN,
attached_policies=[
{
"PolicyName": "AdministratorAccess",
"PolicyArn": CUSTOMER_ADMIN_NAMED_POLICY_ARN,
}
],
),
_role(
UNRESOLVED_POLICY_ROLE_ARN,
attached_policies=[
{
"PolicyName": "unresolved",
"PolicyArn": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:policy/unresolved",
}
],
),
# Proven admin attached policy combined with an unresolved one: the
# unresolved document could contain a deny, so the outcome is unknown.
_role(
ADMIN_UNRESOLVED_POLICY_ROLE_ARN,
attached_policies=[
{
"PolicyName": "AdministratorAccess",
"PolicyArn": AWS_ADMIN_POLICY_ARN,
},
{
"PolicyName": "unresolved",
"PolicyArn": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:policy/unresolved",
},
],
),
# Attached policy resolves to a malformed (non-dict) document.
_role(
INVALID_POLICY_ROLE_ARN,
attached_policies=[
{
"PolicyName": "invalid",
"PolicyArn": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:policy/invalid",
}
],
),
# Allow *:* in the attached policy negated by an unconditional Deny *:*
# in an inline policy: not effectively administrative.
_role(
DENY_OVERRIDE_ROLE_ARN,
attached_policies=[
{"PolicyName": "custom-admin", "PolicyArn": CUSTOM_ADMIN_POLICY_ARN}
],
inline_policies=["deny-all"],
),
# Allow *:* in the attached policy plus a Condition-guarded Deny: the
# deny may or may not apply, so the outcome is unknown.
_role(
CONDITIONAL_DENY_ROLE_ARN,
attached_policies=[
{"PolicyName": "custom-admin", "PolicyArn": CUSTOM_ADMIN_POLICY_ARN}
],
inline_policies=["conditional-deny"],
),
# Allow *:* guarded by a Condition: not statically provable as admin.
_role(
CONDITIONAL_ADMIN_ROLE_ARN,
attached_policies=[
{
"PolicyName": "conditional-admin",
"PolicyArn": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:policy/conditional-admin",
}
],
),
# Administrative identity policies constrained by a restrictive
# permissions boundary: not effectively administrative.
_role(
BOUNDED_ADMIN_ROLE_ARN,
attached_policies=[
{"PolicyName": "AdministratorAccess", "PolicyArn": AWS_ADMIN_POLICY_ARN}
],
permissions_boundary={
"PermissionsBoundaryType": "Policy",
"PermissionsBoundaryArn": BOUNDARY_POLICY_ARN,
},
),
# Boundary present but its document is not in the IAM inventory:
# restrictions cannot be evaluated, so the role is not classified admin.
_role(
UNRESOLVED_BOUNDARY_ROLE_ARN,
attached_policies=[
{"PolicyName": "AdministratorAccess", "PolicyArn": AWS_ADMIN_POLICY_ARN}
],
permissions_boundary={
"PermissionsBoundaryType": "Policy",
"PermissionsBoundaryArn": UNRESOLVED_BOUNDARY_POLICY_ARN,
},
),
# AdministratorAccess as the boundary does not restrict anything.
_role(
ADMIN_BOUNDED_ROLE_ARN,
attached_policies=[
{"PolicyName": "AdministratorAccess", "PolicyArn": AWS_ADMIN_POLICY_ARN}
],
permissions_boundary={
"PermissionsBoundaryType": "Policy",
"PermissionsBoundaryArn": AWS_ADMIN_POLICY_ARN,
},
),
]
policies = {
AWS_ADMIN_POLICY_ARN: SimpleNamespace(document=FULL_ACCESS_DOCUMENT),
CUSTOM_ADMIN_POLICY_ARN: SimpleNamespace(document=FULL_ACCESS_DOCUMENT),
MANAGED_POLICY_ARN: SimpleNamespace(document=READONLY_DOCUMENT),
# Customer-managed policy that merely shares the AdministratorAccess name.
CUSTOMER_ADMIN_NAMED_POLICY_ARN: SimpleNamespace(document=READONLY_DOCUMENT),
f"{INLINE_ADMIN_ROLE_ARN}:policy/inline-admin": SimpleNamespace(
document=FULL_ACCESS_DOCUMENT
),
f"{DENY_OVERRIDE_ROLE_ARN}:policy/deny-all": SimpleNamespace(
document=DENY_ALL_DOCUMENT
),
f"{CONDITIONAL_DENY_ROLE_ARN}:policy/conditional-deny": SimpleNamespace(
document=CONDITIONAL_DENY_DOCUMENT
),
f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:policy/invalid": SimpleNamespace(
document="invalid"
),
f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:policy/conditional-admin": SimpleNamespace(
document=CONDITIONAL_ADMIN_DOCUMENT
),
BOUNDARY_POLICY_ARN: SimpleNamespace(document=READONLY_DOCUMENT),
# Customer-managed session policy whose document grants *:*.
CUSTOMER_FULL_ACCESS_POLICY_ARN: SimpleNamespace(document=FULL_ACCESS_DOCUMENT),
}
iam = mock.MagicMock()
iam.roles = roles
iam.policies = policies
return iam
def _build_client(profiles):
ra_client = mock.MagicMock()
ra_client.profiles = profiles
return ra_client
def _patched(ra_client):
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
check_module = "prowler.providers.aws.services.rolesanywhere.rolesanywhere_profile_restricts_session_permissions.rolesanywhere_profile_restricts_session_permissions"
return [
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
),
mock.patch(f"{check_module}.rolesanywhere_client", new=ra_client),
mock.patch(f"{check_module}.iam_client", new=_iam_client()),
]
def _enter(patches):
from contextlib import ExitStack
stack = ExitStack()
for p in patches:
stack.enter_context(p)
return stack
def _run():
from prowler.providers.aws.services.rolesanywhere.rolesanywhere_profile_restricts_session_permissions.rolesanywhere_profile_restricts_session_permissions import (
rolesanywhere_profile_restricts_session_permissions,
)
return rolesanywhere_profile_restricts_session_permissions().execute()
class Test_rolesanywhere_profile_restricts_session_permissions:
def test_no_profiles(self):
with _enter(_patched(_build_client({}))):
assert len(_run()) == 0
def test_unscoped_profile_with_admin_role_fails(self):
with _enter(
_patched(_build_client({PROFILE_ARN: _profile(role_arns=[ADMIN_ROLE_ARN])}))
):
result = _run()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].resource_id == PROFILE_ID
assert result[0].resource_arn == PROFILE_ARN
assert result[0].region == AWS_REGION_US_EAST_1
assert ADMIN_ROLE_ARN in result[0].status_extended
assert "administrative" in result[0].status_extended
def test_unscoped_profile_with_custom_admin_policy_fails(self):
with _enter(
_patched(
_build_client(
{PROFILE_ARN: _profile(role_arns=[CUSTOM_ADMIN_ROLE_ARN])}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "FAIL"
assert CUSTOM_ADMIN_ROLE_ARN in result[0].status_extended
def test_unscoped_profile_with_inline_admin_policy_fails(self):
with _enter(
_patched(
_build_client(
{PROFILE_ARN: _profile(role_arns=[INLINE_ADMIN_ROLE_ARN])}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "FAIL"
assert INLINE_ADMIN_ROLE_ARN in result[0].status_extended
def test_mixed_roles_fail_lists_only_admin_role(self):
with _enter(
_patched(
_build_client(
{
PROFILE_ARN: _profile(
role_arns=[READONLY_ROLE_ARN, ADMIN_ROLE_ARN]
)
}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "FAIL"
assert ADMIN_ROLE_ARN in result[0].status_extended
assert READONLY_ROLE_ARN not in result[0].status_extended
def test_customer_policy_named_administratoraccess_passes(self):
# Name collision: customer-managed policy called AdministratorAccess
# whose document is read-only must not flag the role as administrative.
with _enter(
_patched(
_build_client(
{PROFILE_ARN: _profile(role_arns=[NAME_COLLISION_ROLE_ARN])}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "PASS"
def test_unresolved_attached_policy_is_manual(self):
# Attached policy ARN missing from iam_client.policies: a missing
# document is unknown, not proof that the role is unprivileged.
with _enter(
_patched(
_build_client(
{PROFILE_ARN: _profile(role_arns=[UNRESOLVED_POLICY_ROLE_ARN])}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "MANUAL"
assert "could not be evaluated" in result[0].status_extended
def test_invalid_policy_document_is_manual(self):
# A malformed policy document cannot prove anything about the role.
with _enter(
_patched(
_build_client(
{PROFILE_ARN: _profile(role_arns=[INVALID_POLICY_ROLE_ARN])}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "MANUAL"
def test_allow_all_with_unresolved_policy_is_manual(self):
# Proven admin policy plus an unresolved one: the unresolved document
# could contain a deny, so the classification is unknown.
with _enter(
_patched(
_build_client(
{
PROFILE_ARN: _profile(
role_arns=[ADMIN_UNRESOLVED_POLICY_ROLE_ARN]
)
}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "MANUAL"
def test_unscoped_profile_with_least_privilege_role_passes(self):
with _enter(
_patched(
_build_client({PROFILE_ARN: _profile(role_arns=[READONLY_ROLE_ARN])})
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "PASS"
assert "defense-in-depth" in result[0].status_extended
def test_unscoped_profile_with_unknown_role_is_manual(self):
# A referenced role missing from the IAM inventory is unknown, not
# proof that no administrative role exists.
with _enter(
_patched(
_build_client({PROFILE_ARN: _profile(role_arns=[UNKNOWN_ROLE_ARN])})
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "MANUAL"
assert "could not be evaluated" in result[0].status_extended
def test_unscoped_profile_without_roles_passes(self):
with _enter(_patched(_build_client({PROFILE_ARN: _profile(role_arns=[])}))):
result = _run()
assert len(result) == 1
assert result[0].status == "PASS"
def test_profile_with_session_policy_passes(self):
with _enter(
_patched(
_build_client(
{
PROFILE_ARN: _profile(
session_policy=SESSION_POLICY, role_arns=[ADMIN_ROLE_ARN]
)
}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "PASS"
assert "session policy" in result[0].status_extended
def test_full_access_session_policy_does_not_scope(self):
# A sessionPolicy granting *:* does not restrict anything.
with _enter(
_patched(
_build_client(
{
PROFILE_ARN: _profile(
session_policy=FULL_ACCESS_SESSION_POLICY,
role_arns=[ADMIN_ROLE_ARN],
)
}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "FAIL"
def test_profile_with_managed_policies_passes(self):
with _enter(
_patched(
_build_client(
{
PROFILE_ARN: _profile(
managed_policy_arns=[MANAGED_POLICY_ARN],
role_arns=[ADMIN_ROLE_ARN],
)
}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "PASS"
def test_admin_managed_session_policy_does_not_scope(self):
# AdministratorAccess as the managed session policy restricts nothing.
with _enter(
_patched(
_build_client(
{
PROFILE_ARN: _profile(
managed_policy_arns=[AWS_ADMIN_POLICY_ARN],
role_arns=[ADMIN_ROLE_ARN],
)
}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "FAIL"
def test_restrictive_inline_with_admin_managed_policy_fails(self):
# The session-policy set is evaluated as a union: AdministratorAccess as
# a managed session policy makes the boundary unrestricted even though
# the inline session policy is restrictive.
with _enter(
_patched(
_build_client(
{
PROFILE_ARN: _profile(
session_policy=SESSION_POLICY,
managed_policy_arns=[AWS_ADMIN_POLICY_ARN],
role_arns=[ADMIN_ROLE_ARN],
)
}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "FAIL"
def test_full_access_inline_with_restrictive_managed_policy_fails(self):
# Conversely, a *:* inline session policy leaves the union unrestricted
# regardless of a restrictive managed session policy.
with _enter(
_patched(
_build_client(
{
PROFILE_ARN: _profile(
session_policy=FULL_ACCESS_SESSION_POLICY,
managed_policy_arns=[MANAGED_POLICY_ARN],
role_arns=[ADMIN_ROLE_ARN],
)
}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "FAIL"
def test_restrictive_inline_and_restrictive_managed_policy_passes(self):
with _enter(
_patched(
_build_client(
{
PROFILE_ARN: _profile(
session_policy=SESSION_POLICY,
managed_policy_arns=[MANAGED_POLICY_ARN],
role_arns=[ADMIN_ROLE_ARN],
)
}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "PASS"
def test_customer_managed_full_access_session_policy_fails(self):
# A customer-managed session policy whose document grants *:* must be
# resolved through iam_client.policies and treated as unscoped.
with _enter(
_patched(
_build_client(
{
PROFILE_ARN: _profile(
managed_policy_arns=[CUSTOMER_FULL_ACCESS_POLICY_ARN],
role_arns=[ADMIN_ROLE_ARN],
)
}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "FAIL"
def test_unresolved_managed_session_policy_is_manual(self):
# A managed session policy whose document was not collected does not
# prove that the session is restricted: the outcome is unknown.
with _enter(
_patched(
_build_client(
{
PROFILE_ARN: _profile(
managed_policy_arns=[UNRESOLVED_SESSION_POLICY_ARN],
role_arns=[ADMIN_ROLE_ARN],
)
}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "MANUAL"
assert "session scoping could not be evaluated" in result[0].status_extended
def test_invalid_inline_session_policy_is_manual(self):
# An inline session policy that fails to parse does not prove that the
# session is restricted: the outcome is unknown.
with _enter(
_patched(
_build_client(
{
PROFILE_ARN: _profile(
session_policy="not-json", role_arns=[ADMIN_ROLE_ARN]
)
}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "MANUAL"
def test_allow_all_with_cross_policy_deny_all_passes(self):
# Allow *:* in an attached policy plus an unconditional Deny *:* in an
# inline policy: the merged evaluation must not classify the role admin.
with _enter(
_patched(
_build_client(
{PROFILE_ARN: _profile(role_arns=[DENY_OVERRIDE_ROLE_ARN])}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "PASS"
def test_conditional_admin_allow_is_manual(self):
# An Allow *:* guarded by a Condition is not statically provable in
# either direction: the classification is unknown.
with _enter(
_patched(
_build_client(
{PROFILE_ARN: _profile(role_arns=[CONDITIONAL_ADMIN_ROLE_ARN])}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "MANUAL"
def test_allow_all_with_conditional_deny_is_manual(self):
# Unconditional Allow *:* plus a Condition-guarded Deny: the deny may
# or may not negate the grant, so the classification is unknown.
with _enter(
_patched(
_build_client(
{PROFILE_ARN: _profile(role_arns=[CONDITIONAL_DENY_ROLE_ARN])}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "MANUAL"
def test_admin_role_with_restrictive_boundary_passes(self):
# Admin identity policies intersected with a read-only permissions
# boundary are not effectively administrative.
with _enter(
_patched(
_build_client(
{PROFILE_ARN: _profile(role_arns=[BOUNDED_ADMIN_ROLE_ARN])}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "PASS"
def test_admin_role_with_unresolved_boundary_is_manual(self):
# When the boundary document cannot be resolved the restrictions are
# unknown: neither administrative nor safe can be proven.
with _enter(
_patched(
_build_client(
{PROFILE_ARN: _profile(role_arns=[UNRESOLVED_BOUNDARY_ROLE_ARN])}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "MANUAL"
def test_admin_role_with_admin_boundary_fails(self):
# An AdministratorAccess boundary restricts nothing: still admin.
with _enter(
_patched(
_build_client(
{PROFILE_ARN: _profile(role_arns=[ADMIN_BOUNDED_ROLE_ARN])}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "FAIL"
assert ADMIN_BOUNDED_ROLE_ARN in result[0].status_extended
def test_disabled_profile_passes(self):
with _enter(
_patched(
_build_client(
{PROFILE_ARN: _profile(enabled=False, role_arns=[ADMIN_ROLE_ARN])}
)
)
):
result = _run()
assert len(result) == 1
assert result[0].status == "PASS"
assert "disabled" in result[0].status_extended
@@ -4,6 +4,7 @@ import botocore
from moto import mock_aws
from prowler.providers.aws.services.rolesanywhere.rolesanywhere_service import (
Profile,
RolesAnywhere,
TrustAnchor,
)
@@ -16,6 +17,9 @@ from tests.providers.aws.utils import (
TA_ID = "11111111-2222-3333-4444-555555555555"
TA_ARN = f"arn:aws:rolesanywhere:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:trust-anchor/{TA_ID}"
PCA_ARN = f"arn:aws:acm-pca:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:certificate-authority/abc"
PROFILE_ID = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
PROFILE_ARN = f"arn:aws:rolesanywhere:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:profile/{PROFILE_ID}"
ROLE_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/workload-role"
make_api_call = botocore.client.BaseClient._make_api_call
@@ -36,6 +40,21 @@ def mock_make_api_call(self, operation_name, kwarg):
}
]
}
if operation_name == "ListProfiles":
return {
"profiles": [
{
"profileArn": PROFILE_ARN,
"profileId": PROFILE_ID,
"name": "workload-profile",
"enabled": True,
"roleArns": [ROLE_ARN],
"sessionPolicy": '{"Version":"2012-10-17","Statement":[]}',
"durationSeconds": 3600,
"acceptRoleSessionName": True,
}
]
}
if operation_name == "ListTagsForResource":
return {"tags": [{"key": "Environment", "value": "test"}]}
return make_api_call(self, operation_name, kwarg)
@@ -78,6 +97,25 @@ class Test_RolesAnywhere_Service:
assert ta.region == AWS_REGION_US_EAST_1
assert ta.tags == [{"key": "Environment", "value": "test"}]
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
@mock_aws
def test_list_profiles(self):
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
rolesanywhere = RolesAnywhere(aws_provider)
assert len(rolesanywhere.profiles) == 1
profile = rolesanywhere.profiles[PROFILE_ARN]
assert isinstance(profile, Profile)
assert profile.id == PROFILE_ID
assert profile.name == "workload-profile"
assert profile.enabled is True
assert profile.role_arns == [ROLE_ARN]
assert profile.session_policy == '{"Version":"2012-10-17","Statement":[]}'
assert profile.managed_policy_arns == []
assert profile.duration_seconds == 3600
assert profile.accept_role_session_name is True
assert profile.region == AWS_REGION_US_EAST_1
assert profile.tags == [{"key": "Environment", "value": "test"}]
@patch(
"botocore.client.BaseClient._make_api_call", new=mock_make_api_call_tags_failure
)
+13
View File
@@ -490,6 +490,19 @@ def mock_api_projects_calls(client: MagicMock):
}
client.projects().serviceAccounts().list_next.return_value = None
# Workload Identity Federation pools/providers: return empty pages and stop
# pagination so the discovery while-loops in the IAM service terminate.
client.projects().locations().workloadIdentityPools().list().execute.return_value = {
"workloadIdentityPools": []
}
client.projects().locations().workloadIdentityPools().list_next.return_value = None
client.projects().locations().workloadIdentityPools().providers().list().execute.return_value = {
"workloadIdentityPoolProviders": []
}
client.projects().locations().workloadIdentityPools().providers().list_next.return_value = (
None
)
def mock_list_service_accounts_keys(name):
return_value = MagicMock()
if (
@@ -0,0 +1,148 @@
from unittest import mock
from tests.providers.gcp.gcp_fixtures import (
GCP_PROJECT_ID,
GCP_US_CENTER1_LOCATION,
set_mocked_gcp_provider,
)
CHECK_MODULE = "prowler.providers.gcp.services.iam.iam_workload_identity_pool_provider_attribute_condition.iam_workload_identity_pool_provider_attribute_condition"
def _run(provider_kwargs):
iam_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(f"{CHECK_MODULE}.iam_client", new=iam_client),
):
from prowler.providers.gcp.services.iam.iam_service import (
WorkloadIdentityPoolProvider,
)
from prowler.providers.gcp.services.iam.iam_workload_identity_pool_provider_attribute_condition.iam_workload_identity_pool_provider_attribute_condition import (
iam_workload_identity_pool_provider_attribute_condition,
)
providers = []
for kwargs in provider_kwargs:
provider_id = kwargs.get("provider_id", "my-provider")
providers.append(
WorkloadIdentityPoolProvider(
name=(
f"projects/{GCP_PROJECT_ID}/locations/global/"
f"workloadIdentityPools/my-pool/providers/{provider_id}"
),
id=provider_id,
pool_id="my-pool",
pool_disabled=kwargs.get("pool_disabled", False),
project_id=GCP_PROJECT_ID,
state=kwargs.get("state", "ACTIVE"),
disabled=kwargs.get("disabled", False),
attribute_condition=kwargs.get("attribute_condition", ""),
provider_type=kwargs.get("provider_type", "oidc"),
issuer_uri=kwargs.get(
"issuer_uri",
"https://token.actions.githubusercontent.com",
),
display_name="My Provider",
)
)
iam_client.project_ids = [GCP_PROJECT_ID]
iam_client.region = GCP_US_CENTER1_LOCATION
iam_client.workload_identity_pool_providers = providers
return iam_workload_identity_pool_provider_attribute_condition().execute()
class Test_iam_workload_identity_pool_provider_attribute_condition:
def test_no_providers(self):
assert len(_run([])) == 0
def test_multi_tenant_issuer_without_attribute_condition_fails(self):
result = _run(
[
{
"attribute_condition": "",
"issuer_uri": "https://token.actions.githubusercontent.com",
}
]
)
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].resource_id.endswith("my-provider")
assert result[0].location == "global"
assert "multi-tenant issuer" in result[0].status_extended
def test_dedicated_issuer_without_attribute_condition_passes(self):
result = _run(
[
{
"attribute_condition": "",
"issuer_uri": "https://oidc.eks.eu-west-1.amazonaws.com/id/ABC123",
}
]
)
assert len(result) == 1
assert result[0].status == "PASS"
assert "dedicated issuer" in result[0].status_extended
def test_non_oidc_provider_without_attribute_condition_passes(self):
result = _run(
[{"attribute_condition": "", "provider_type": "aws", "issuer_uri": ""}]
)
assert len(result) == 1
assert result[0].status == "PASS"
assert "not an OIDC provider" in result[0].status_extended
def test_multi_tenant_issuer_with_port_and_uppercase_fails(self):
result = _run(
[{"attribute_condition": "", "issuer_uri": "https://GitLab.com:443"}]
)
assert len(result) == 1
assert result[0].status == "FAIL"
def test_multi_tenant_issuer_bare_host_fails(self):
result = _run(
[
{
"attribute_condition": "",
"issuer_uri": "token.actions.githubusercontent.com",
}
]
)
assert len(result) == 1
assert result[0].status == "FAIL"
def test_multi_tenant_issuer_with_attribute_condition_passes(self):
result = _run(
[
{
"attribute_condition": "assertion.repository_owner == 'acme'",
"issuer_uri": "https://token.actions.githubusercontent.com",
}
]
)
assert len(result) == 1
assert result[0].status == "PASS"
assert "enforces an attribute condition" in result[0].status_extended
def test_disabled_provider_passes(self):
result = _run([{"disabled": True}])
assert len(result) == 1
assert result[0].status == "PASS"
assert "not active" in result[0].status_extended
def test_non_active_provider_passes(self):
result = _run([{"state": "DELETED"}])
assert len(result) == 1
assert result[0].status == "PASS"
def test_active_provider_in_disabled_pool_passes(self):
# The provider itself is ACTIVE and unconditioned on a multi-tenant
# issuer, but its parent pool is disabled and cannot vend credentials.
result = _run([{"pool_disabled": True}])
assert len(result) == 1
assert result[0].status == "PASS"
assert "disabled pool" in result[0].status_extended
@@ -0,0 +1,215 @@
from unittest.mock import MagicMock, patch
from tests.providers.gcp.gcp_fixtures import (
GCP_PROJECT_ID,
mock_is_api_active,
set_mocked_gcp_provider,
)
PROJECT_A = GCP_PROJECT_ID
PROJECT_B = "test-project-b"
def _pool_name(project_id, pool_id="my-pool"):
return f"projects/{project_id}/locations/global/workloadIdentityPools/{pool_id}"
def _provider_payload(pool_name, provider_id="my-provider"):
return {
"name": f"{pool_name}/providers/{provider_id}",
"state": "ACTIVE",
"disabled": False,
"attributeMapping": {"google.subject": "assertion.sub"},
"oidc": {"issuerUri": "https://token.actions.githubusercontent.com"},
"displayName": "gh",
}
def _empty_service_accounts(client):
"""Stub the service-account calls used by the rest of the IAM __init__."""
sa = client.projects.return_value.serviceAccounts.return_value
sa.list.return_value.execute.return_value = {"accounts": []}
sa.list_next.return_value = None
def _wif_client(_GCPService, _service, _api_version, _credentials):
"""Discovery client stub returning one pool with one provider."""
client = MagicMock()
pool_name = _pool_name(GCP_PROJECT_ID)
pools = (
client.projects.return_value.locations.return_value.workloadIdentityPools.return_value
)
pools.list.return_value.execute.return_value = {
"workloadIdentityPools": [{"name": pool_name, "state": "ACTIVE"}]
}
pools.list_next.return_value = None
providers = pools.providers.return_value
providers.list.return_value.execute.return_value = {
"workloadIdentityPoolProviders": [_provider_payload(pool_name)]
}
providers.list_next.return_value = None
_empty_service_accounts(client)
return client
def _disabled_pool_client(_GCPService, _service, _api_version, _credentials):
"""Discovery client stub: a disabled pool containing an ACTIVE provider."""
client = MagicMock()
pool_name = _pool_name(GCP_PROJECT_ID)
pools = (
client.projects.return_value.locations.return_value.workloadIdentityPools.return_value
)
pools.list.return_value.execute.return_value = {
"workloadIdentityPools": [
{"name": pool_name, "state": "ACTIVE", "disabled": True}
]
}
pools.list_next.return_value = None
providers = pools.providers.return_value
providers.list.return_value.execute.return_value = {
"workloadIdentityPoolProviders": [_provider_payload(pool_name)]
}
providers.list_next.return_value = None
_empty_service_accounts(client)
return client
def _pool_list_failure_client(_GCPService, _service, _api_version, _credentials):
"""Pool listing fails for PROJECT_A but succeeds for PROJECT_B."""
client = MagicMock()
pool_name_b = _pool_name(PROJECT_B)
pools = (
client.projects.return_value.locations.return_value.workloadIdentityPools.return_value
)
def pools_list(parent):
request = MagicMock()
if f"projects/{PROJECT_A}/" in parent:
request.execute.side_effect = Exception("permission denied listing pools")
else:
request.execute.return_value = {
"workloadIdentityPools": [{"name": pool_name_b, "state": "ACTIVE"}]
}
return request
pools.list.side_effect = pools_list
pools.list_next.return_value = None
providers = pools.providers.return_value
providers.list.return_value.execute.return_value = {
"workloadIdentityPoolProviders": [_provider_payload(pool_name_b)]
}
providers.list_next.return_value = None
_empty_service_accounts(client)
return client
def _provider_list_failure_client(_GCPService, _service, _api_version, _credentials):
"""Provider listing fails for pool-1 but succeeds for pool-2 in one project."""
client = MagicMock()
pool_1 = _pool_name(GCP_PROJECT_ID, "pool-1")
pool_2 = _pool_name(GCP_PROJECT_ID, "pool-2")
pools = (
client.projects.return_value.locations.return_value.workloadIdentityPools.return_value
)
pools.list.return_value.execute.return_value = {
"workloadIdentityPools": [
{"name": pool_1, "state": "ACTIVE"},
{"name": pool_2, "state": "ACTIVE"},
]
}
pools.list_next.return_value = None
providers = pools.providers.return_value
def providers_list(parent):
request = MagicMock()
if parent == pool_1:
request.execute.side_effect = Exception(
"permission denied listing providers"
)
else:
request.execute.return_value = {
"workloadIdentityPoolProviders": [
_provider_payload(pool_2, provider_id="provider-2")
]
}
return request
providers.list.side_effect = providers_list
providers.list_next.return_value = None
_empty_service_accounts(client)
return client
def _run_service(client_factory, project_ids):
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(project_ids=project_ids),
),
patch(
"prowler.providers.gcp.lib.service.service.GCPService.__is_api_active__",
new=mock_is_api_active,
),
patch(
"prowler.providers.gcp.lib.service.service.GCPService.__generate_client__",
new=client_factory,
),
):
from prowler.providers.gcp.services.iam.iam_service import IAM
return IAM(set_mocked_gcp_provider(project_ids=project_ids))
class TestIAMWorkloadIdentityService:
def test_get_workload_identity_pool_providers(self):
iam = _run_service(_wif_client, [GCP_PROJECT_ID])
assert len(iam.workload_identity_pool_providers) == 1
provider = iam.workload_identity_pool_providers[0]
assert provider.id == "my-provider"
assert provider.pool_id == "my-pool"
assert provider.project_id == GCP_PROJECT_ID
assert provider.state == "ACTIVE"
assert provider.disabled is False
assert provider.pool_disabled is False
assert provider.attribute_condition == ""
assert provider.provider_type == "oidc"
assert provider.issuer_uri == "https://token.actions.githubusercontent.com"
def test_disabled_pool_state_propagates_to_provider(self):
iam = _run_service(_disabled_pool_client, [GCP_PROJECT_ID])
# The provider is ACTIVE, but its parent pool is disabled: the pool's
# effective state must travel with the provider record.
assert len(iam.workload_identity_pool_providers) == 1
provider = iam.workload_identity_pool_providers[0]
assert provider.state == "ACTIVE"
assert provider.disabled is False
assert provider.pool_disabled is True
def test_pool_list_failure_does_not_block_other_projects(self):
iam = _run_service(_pool_list_failure_client, [PROJECT_A, PROJECT_B])
# PROJECT_A's pool listing failed, but PROJECT_B is still processed.
assert len(iam.workload_identity_pool_providers) == 1
assert iam.workload_identity_pool_providers[0].project_id == PROJECT_B
def test_provider_list_failure_only_skips_that_pool(self):
iam = _run_service(_provider_list_failure_client, [GCP_PROJECT_ID])
# pool-1's provider listing failed, but pool-2's provider is still found.
assert len(iam.workload_identity_pool_providers) == 1
assert iam.workload_identity_pool_providers[0].pool_id == "pool-2"
assert iam.workload_identity_pool_providers[0].id == "provider-2"