fix(core): prevent false-positive RoleValidationError under condition…#6029

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
raza-khan0108 wants to merge 1 commit into
base:master
Choose a base branch
from
Open

fix(core): prevent false-positive RoleValidationError under condition… #6029

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
123 changes: 105 additions & 18 deletions sagemaker-core/src/sagemaker/core/helper/iam_role_resolver.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@

import json
import logging
import os
import time
from typing import List, Optional, Set, Tuple, Union
from urllib.parse import unquote
Expand DownExpand Up@@ -303,13 +304,45 @@ def _resolve_caller_role_arn(
# ---------------------------------------------------------------------------
# Read-only permission / trust validation
# ---------------------------------------------------------------------------
def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> List[str]:
"""Return the subset of ``actions`` that ``role_arn`` is NOT allowed to perform.
def _is_unverifiable_denial(result: dict) -> bool:
"""Return True if an evaluation result's non-allowed decision is unverifiable.

Because iam:SimulatePrincipalPolicy does not evaluate condition-based SCPs or
condition-based permissions boundaries, an `implicitDeny` caused by (or
masked by) `AllowedByOrganizations == false` or
`AllowedByPermissionsBoundary == false` cannot be definitively verified by
the client-side policy simulator.
"""
if result.get("EvalDecision") == "explicitDeny":
return False
org_detail = result.get("OrganizationsDecisionDetail", {})
org_allowed = org_detail.get("AllowedByOrganizations") if isinstance(org_detail, dict) else None
pb_detail = result.get("PermissionsBoundaryDecisionDetail", {})
pb_allowed = pb_detail.get("AllowedByPermissionsBoundary") if isinstance(pb_detail, dict) else None

# Handle boolean False, string "false", "False", etc.
if org_allowed in (False, "false", "False") or pb_allowed in (False, "false", "False"):
return True
return False


def _simulate_denied_actions(
iam_client, role_arn: str, actions: List[str]
) -> Tuple[List[str], List[str]]:
"""Return lists of (denied_actions, unverifiable_actions) for ``role_arn``.

Wraps the paginated ``iam:SimulatePrincipalPolicy`` call so both the role-
validation path and the HyperPod caller-side check share one implementation.
An empty list means every action is allowed. The pagination matters: a
truncated first page must not produce a false "all allowed" verdict.
The pagination matters: a truncated first page must not produce a false
"all allowed" verdict.

Returns:
(denied_actions, unverifiable_actions):
denied_actions: actions definitively denied (explicitDeny, or
implicitDeny without org/permissions-boundary conditions).
unverifiable_actions: actions returning implicitDeny with
AllowedByOrganizations=False or AllowedByPermissionsBoundary=False,
which the simulator cannot evaluate due to conditions.

Raises ClientError on failures the caller must interpret (e.g. AccessDenied
when the principal can't self-simulate, NoSuchEntity when the role is gone).
Expand All@@ -319,11 +352,16 @@ def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> L
for page in paginator.paginate(PolicySourceArn=role_arn, ActionNames=actions):
evaluation_results.extend(page.get("EvaluationResults", []))

return [
result["EvalActionName"]
for result in evaluation_results
if result["EvalDecision"] != "allowed"
]
denied = []
unverifiable = []
for result in evaluation_results:
if result.get("EvalDecision") != "allowed":
action_name = result["EvalActionName"]
if _is_unverifiable_denial(result):
unverifiable.append(action_name)
else:
denied.append(action_name)
return denied, unverifiable


def _evaluate_permissions(
Expand All@@ -339,21 +377,32 @@ def _evaluate_permissions(
verdict True — all gated actions are allowed (denied_actions empty).
verdict False — at least one gated action is denied (denied_actions lists them).
verdict None — could not be determined, e.g. the caller lacks
iam:SimulatePrincipalPolicy (denied_actions empty).
iam:SimulatePrincipalPolicy or conditional SCPs prevent
evaluation (denied_actions empty).
"""
required_actions = _get_smoke_test_actions(role_type)
if not required_actions:
return True, []

try:
denied = _simulate_denied_actions(iam_client, role_arn, required_actions)
denied, unverifiable = _simulate_denied_actions(iam_client, role_arn, required_actions)
if denied:
logger.info(
"Role '%s' is missing permissions for: %s",
role_arn,
", ".join(denied[:5]) + ("..." if len(denied) > 5 else ""),
)
return False, denied
if unverifiable:
logger.info(
"Cannot definitively verify permissions for role '%s' due to "
"Organizations SCPs or permissions boundaries (%s returned implicitDeny "
"with AllowedByOrganizations/AllowedByPermissionsBoundary=false); "
"permission verdict unknown.",
role_arn,
", ".join(unverifiable[:5]) + ("..." if len(unverifiable) > 5 else ""),
)
return None, []
return True, []

except ClientError as e:
Expand DownExpand Up@@ -518,28 +567,35 @@ def _build_validation_error_message(


def resolve_and_validate_role(
provided_role: Optional[str],
role_type: str,
provided_role: Optional[str] = None,
role_type: str = "training",
sagemaker_session=None,
validate_role: bool = True,
*,
role_arn: Optional[str] = None,
) -> str:
"""Resolve the role to use and validate it (read-only; does not mutate IAM).

Resolution:
1. ``provided_role`` given → resolve it to an ARN (must exist).
1. ``provided_role`` (or ``role_arn``) given → resolve it to an ARN (must exist).
2. Otherwise → resolve the caller's own identity role.

The resolved role is then VALIDATED (read-only, via iam:SimulatePrincipalPolicy
+ trust inspection):
+ trust inspection) unless ``validate_role=False`` or environment variable
``SAGEMAKER_VALIDATE_ROLE=false`` is set:
* permissions allowed AND trusted → return the ARN.
* a required permission is definitively denied → raise RoleValidationError.
* the trust policy definitively excludes the service → raise RoleValidationError.
* permissions cannot be verified (caller lacks iam:SimulatePrincipalPolicy,
the common Studio/notebook case) → return the ARN with a WARNING.
or conditional SCPs/permissions boundaries prevent evaluation) → return
the ARN with a WARNING.

Args:
provided_role: User-supplied role name or ARN. If set, used directly.
role_type: One of ROLE_TYPES.
sagemaker_session: SageMaker session (used to get the boto session).
validate_role: If False, skips client-side permission/trust validation.
role_arn: Keyword alias for ``provided_role``.

Returns:
IAM role ARN.
Expand All@@ -552,6 +608,7 @@ def resolve_and_validate_role(
if role_type not in ROLE_TYPES:
raise ValueError(f"Invalid role_type '{role_type}'. Must be one of: {ROLE_TYPES}")

provided_role = provided_role or role_arn
boto_session = _get_boto_session(sagemaker_session)
iam_client = boto_session.client("iam")

Expand All@@ -567,6 +624,10 @@ def resolve_and_validate_role(
if not role_arn:
raise RoleValidationError(_build_validation_error_message(None, role_type))

if not validate_role or os.getenv("SAGEMAKER_VALIDATE_ROLE", "true").lower() in ("false", "0", "no"):
logger.info("Skipping IAM role validation for '%s' (%s).", role_arn, role_type)
return role_arn

# Permission check (definitive denial blocks; unverifiable warns).
verdict, denied = _evaluate_permissions(iam_client, role_arn, role_type)
if verdict is False:
Expand All@@ -584,7 +645,8 @@ def resolve_and_validate_role(
if verdict is None:
logger.warning(
"Could not verify permissions for role '%s' (caller lacks "
"iam:SimulatePrincipalPolicy). Proceeding with it. If the operation "
"iam:SimulatePrincipalPolicy or conditional SCPs/permissions boundaries "
"prevent evaluation). Proceeding with it. If the operation "
"later fails with an access-denied error, ensure the role has the "
"required permissions for '%s' (see "
"IamRoleResolver().get_required_actions('%s')) or create a dedicated "
Expand DownExpand Up@@ -643,7 +705,7 @@ def verify_hyperpod_connect_permissions(
return None

try:
denied = _simulate_denied_actions(
denied, unverifiable = _simulate_denied_actions(
iam_client, caller_role_arn, list(HYPERPOD_CLI_CONNECT_ACTIONS)
)
except ClientError as e:
Expand DownExpand Up@@ -671,6 +733,15 @@ def verify_hyperpod_connect_permissions(
)
return False

if unverifiable:
logger.info(
"Cannot definitively verify HyperPod connect permissions for '%s' due to "
"Organizations SCPs or permissions boundaries; the HyperPod CLI will "
"validate access at submit time.",
caller_role_arn,
)
return None

logger.info(
"Caller '%s' has the HyperPod CLI connect permissions.", caller_role_arn
)
Expand DownExpand Up@@ -732,6 +803,22 @@ def __init__(self, sagemaker_session=None):
self._sts_client = self._boto_session.client("sts")

# -- public API ---------------------------------------------------------
def resolve_and_validate_role(
self,
provided_role: Optional[str] = None,
role_type: str = "training",
*,
role_arn: Optional[str] = None,
validate_role: bool = True,
) -> str:
"""Resolve and validate an IAM role (read-only; does not mutate IAM)."""
return resolve_and_validate_role(
provided_role=provided_role or role_arn,
role_type=role_type,
sagemaker_session=self._sagemaker_session,
validate_role=validate_role,
)

def get_required_actions(self, role_type: str) -> List[str]:
"""Return the IAM actions a role of ``role_type`` needs (read-only preview)."""
self._validate_role_type(role_type)
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Add copy buttons to all 
 blocks\n(function() {\n function addCopyButtons() {\n document.querySelectorAll('pre code').forEach(function(codeBlock) {\n if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;\n codeBlock.parentElement.setAttribute('data-copy-added', 'true');\n \n var btn = document.createElement('button');\n btn.textContent = 'Copy';\n btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';\n btn.onmouseover = function() { this.style.opacity = '1'; };\n btn.onmouseout = function() { this.style.opacity = '0.7'; };\n btn.onclick = function() {\n navigator.clipboard.writeText(codeBlock.textContent).then(function() {\n btn.textContent = 'Copied!';\n setTimeout(function() { btn.textContent = 'Copy'; }, 1500);\n });\n };\n codeBlock.parentElement.style.position = 'relative';\n codeBlock.parentElement.appendChild(btn);\n });\n }\n \n addCopyButtons();\n \n // Re-run on dynamic content\n var observer = new MutationObserver(addCopyButtons);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Add Copy Buttons to Code Blocks");
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '

Skip to content
fix(core): prevent false-positive RoleValidationError under condition…#6029

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

raza-khan0108
wants to merge
1
commit into

base:master
Choose a base branch

from

 Open


fix(core): prevent false-positive RoleValidationError under condition…
#6029
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


123 changes: 105 additions & 18 deletions
sagemaker-core/src/sagemaker/core/helper/iam_role_resolver.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@

import json
import logging
import os
import time
from typing import List, Optional, Set, Tuple, Union
from urllib.parse import unquote
Expand DownExpand Up@@ -303,13 +304,45 @@ def _resolve_caller_role_arn(
# ---------------------------------------------------------------------------
# Read-only permission / trust validation
# ---------------------------------------------------------------------------
def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> List[str]:
 """Return the subset of ``actions`` that ``role_arn`` is NOT allowed to perform.
def _is_unverifiable_denial(result: dict) -> bool:
 """Return True if an evaluation result's non-allowed decision is unverifiable.

 Because iam:SimulatePrincipalPolicy does not evaluate condition-based SCPs or
 condition-based permissions boundaries, an `implicitDeny` caused by (or
 masked by) `AllowedByOrganizations == false` or
 `AllowedByPermissionsBoundary == false` cannot be definitively verified by
 the client-side policy simulator.
 """
 if result.get("EvalDecision") == "explicitDeny":
 return False
 org_detail = result.get("OrganizationsDecisionDetail", {})
 org_allowed = org_detail.get("AllowedByOrganizations") if isinstance(org_detail, dict) else None
 pb_detail = result.get("PermissionsBoundaryDecisionDetail", {})
 pb_allowed = pb_detail.get("AllowedByPermissionsBoundary") if isinstance(pb_detail, dict) else None

 # Handle boolean False, string "false", "False", etc.
 if org_allowed in (False, "false", "False") or pb_allowed in (False, "false", "False"):
 return True
 return False


def _simulate_denied_actions(
 iam_client, role_arn: str, actions: List[str]
) -> Tuple[List[str], List[str]]:
 """Return lists of (denied_actions, unverifiable_actions) for ``role_arn``.

 Wraps the paginated ``iam:SimulatePrincipalPolicy`` call so both the role-
 validation path and the HyperPod caller-side check share one implementation.
 An empty list means every action is allowed. The pagination matters: a
 truncated first page must not produce a false "all allowed" verdict.
 The pagination matters: a truncated first page must not produce a false
 "all allowed" verdict.

 Returns:
 (denied_actions, unverifiable_actions):
 denied_actions: actions definitively denied (explicitDeny, or
 implicitDeny without org/permissions-boundary conditions).
 unverifiable_actions: actions returning implicitDeny with
 AllowedByOrganizations=False or AllowedByPermissionsBoundary=False,
 which the simulator cannot evaluate due to conditions.

 Raises ClientError on failures the caller must interpret (e.g. AccessDenied
 when the principal can't self-simulate, NoSuchEntity when the role is gone).
Expand All@@ -319,11 +352,16 @@ def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> L
 for page in paginator.paginate(PolicySourceArn=role_arn, ActionNames=actions):
 evaluation_results.extend(page.get("EvaluationResults", []))

 return [
 result["EvalActionName"]
 for result in evaluation_results
 if result["EvalDecision"] != "allowed"
 ]
 denied = []
 unverifiable = []
 for result in evaluation_results:
 if result.get("EvalDecision") != "allowed":
 action_name = result["EvalActionName"]
 if _is_unverifiable_denial(result):
 unverifiable.append(action_name)
 else:
 denied.append(action_name)
 return denied, unverifiable


def _evaluate_permissions(
Expand All@@ -339,21 +377,32 @@ def _evaluate_permissions(
 verdict True — all gated actions are allowed (denied_actions empty).
 verdict False — at least one gated action is denied (denied_actions lists them).
 verdict None — could not be determined, e.g. the caller lacks
 iam:SimulatePrincipalPolicy (denied_actions empty).
 iam:SimulatePrincipalPolicy or conditional SCPs prevent
 evaluation (denied_actions empty).
 """
 required_actions = _get_smoke_test_actions(role_type)
 if not required_actions:
 return True, []

 try:
 denied = _simulate_denied_actions(iam_client, role_arn, required_actions)
 denied, unverifiable = _simulate_denied_actions(iam_client, role_arn, required_actions)
 if denied:
 logger.info(
 "Role '%s' is missing permissions for: %s",
 role_arn,
 ", ".join(denied[:5]) + ("..." if len(denied) > 5 else ""),
 )
 return False, denied
 if unverifiable:
 logger.info(
 "Cannot definitively verify permissions for role '%s' due to "
 "Organizations SCPs or permissions boundaries (%s returned implicitDeny "
 "with AllowedByOrganizations/AllowedByPermissionsBoundary=false); "
 "permission verdict unknown.",
 role_arn,
 ", ".join(unverifiable[:5]) + ("..." if len(unverifiable) > 5 else ""),
 )
 return None, []
 return True, []

 except ClientError as e:
Expand DownExpand Up@@ -518,28 +567,35 @@ def _build_validation_error_message(


def resolve_and_validate_role(
 provided_role: Optional[str],
 role_type: str,
 provided_role: Optional[str] = None,
 role_type: str = "training",
 sagemaker_session=None,
 validate_role: bool = True,
 *,
 role_arn: Optional[str] = None,
) -> str:
 """Resolve the role to use and validate it (read-only; does not mutate IAM).

 Resolution:
 1. ``provided_role`` given → resolve it to an ARN (must exist).
 1. ``provided_role`` (or ``role_arn``) given → resolve it to an ARN (must exist).
 2. Otherwise → resolve the caller's own identity role.

 The resolved role is then VALIDATED (read-only, via iam:SimulatePrincipalPolicy
 + trust inspection):
 + trust inspection) unless ``validate_role=False`` or environment variable
 ``SAGEMAKER_VALIDATE_ROLE=false`` is set:
 * permissions allowed AND trusted → return the ARN.
 * a required permission is definitively denied → raise RoleValidationError.
 * the trust policy definitively excludes the service → raise RoleValidationError.
 * permissions cannot be verified (caller lacks iam:SimulatePrincipalPolicy,
 the common Studio/notebook case) → return the ARN with a WARNING.
 or conditional SCPs/permissions boundaries prevent evaluation) → return
 the ARN with a WARNING.

 Args:
 provided_role: User-supplied role name or ARN. If set, used directly.
 role_type: One of ROLE_TYPES.
 sagemaker_session: SageMaker session (used to get the boto session).
 validate_role: If False, skips client-side permission/trust validation.
 role_arn: Keyword alias for ``provided_role``.

 Returns:
 IAM role ARN.
Expand All@@ -552,6 +608,7 @@ def resolve_and_validate_role(
 if role_type not in ROLE_TYPES:
 raise ValueError(f"Invalid role_type '{role_type}'. Must be one of: {ROLE_TYPES}")

 provided_role = provided_role or role_arn
 boto_session = _get_boto_session(sagemaker_session)
 iam_client = boto_session.client("iam")

Expand All@@ -567,6 +624,10 @@ def resolve_and_validate_role(
 if not role_arn:
 raise RoleValidationError(_build_validation_error_message(None, role_type))

 if not validate_role or os.getenv("SAGEMAKER_VALIDATE_ROLE", "true").lower() in ("false", "0", "no"):
 logger.info("Skipping IAM role validation for '%s' (%s).", role_arn, role_type)
 return role_arn

 # Permission check (definitive denial blocks; unverifiable warns).
 verdict, denied = _evaluate_permissions(iam_client, role_arn, role_type)
 if verdict is False:
Expand All@@ -584,7 +645,8 @@ def resolve_and_validate_role(
 if verdict is None:
 logger.warning(
 "Could not verify permissions for role '%s' (caller lacks "
 "iam:SimulatePrincipalPolicy). Proceeding with it. If the operation "
 "iam:SimulatePrincipalPolicy or conditional SCPs/permissions boundaries "
 "prevent evaluation). Proceeding with it. If the operation "
 "later fails with an access-denied error, ensure the role has the "
 "required permissions for '%s' (see "
 "IamRoleResolver().get_required_actions('%s')) or create a dedicated "
Expand DownExpand Up@@ -643,7 +705,7 @@ def verify_hyperpod_connect_permissions(
 return None

 try:
 denied = _simulate_denied_actions(
 denied, unverifiable = _simulate_denied_actions(
 iam_client, caller_role_arn, list(HYPERPOD_CLI_CONNECT_ACTIONS)
 )
 except ClientError as e:
Expand DownExpand Up@@ -671,6 +733,15 @@ def verify_hyperpod_connect_permissions(
 )
 return False

 if unverifiable:
 logger.info(
 "Cannot definitively verify HyperPod connect permissions for '%s' due to "
 "Organizations SCPs or permissions boundaries; the HyperPod CLI will "
 "validate access at submit time.",
 caller_role_arn,
 )
 return None

 logger.info(
 "Caller '%s' has the HyperPod CLI connect permissions.", caller_role_arn
 )
Expand DownExpand Up@@ -732,6 +803,22 @@ def __init__(self, sagemaker_session=None):
 self._sts_client = self._boto_session.client("sts")

 # -- public API ---------------------------------------------------------
 def resolve_and_validate_role(
 self,
 provided_role: Optional[str] = None,
 role_type: str = "training",
 *,
 role_arn: Optional[str] = None,
 validate_role: bool = True,
 ) -> str:
 """Resolve and validate an IAM role (read-only; does not mutate IAM)."""
 return resolve_and_validate_role(
 provided_role=provided_role or role_arn,
 role_type=role_type,
 sagemaker_session=self._sagemaker_session,
 validate_role=validate_role,
 )

 def get_required_actions(self, role_type: str) -> List[str]:
 """Return the IAM actions a role of ``role_type`` needs (read-only preview)."""
 self._validate_role_type(role_type)
Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Force GitHub README to respect dark mode\n(function() {\n var style = document.createElement('style');\n style.textContent = '\n .markdown-body {\n color-scheme: dark light;\n }\n .markdown-body pre { background: #161b22 !important; }\n .markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }\n .markdown-body table th, .markdown-body table td { border-color: #30363d !important; }\n .markdown-body img { background: #0d1117; }\n .markdown-body blockquote { border-left-color: #8b949e; }\n .markdown-body hr { border-color: #30363d; }\n ';\n document.head.appendChild(style);\n})();", "GitHub Dark Mode README Fix");
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
fix(core): prevent false-positive RoleValidationError under condition…#6029

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

raza-khan0108
wants to merge
1
commit into

base:master
Choose a base branch

from

 Open


fix(core): prevent false-positive RoleValidationError under condition…
#6029
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


123 changes: 105 additions & 18 deletions
sagemaker-core/src/sagemaker/core/helper/iam_role_resolver.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@

import json
import logging
import os
import time
from typing import List, Optional, Set, Tuple, Union
from urllib.parse import unquote
Expand DownExpand Up@@ -303,13 +304,45 @@ def _resolve_caller_role_arn(
# ---------------------------------------------------------------------------
# Read-only permission / trust validation
# ---------------------------------------------------------------------------
def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> List[str]:
 """Return the subset of ``actions`` that ``role_arn`` is NOT allowed to perform.
def _is_unverifiable_denial(result: dict) -> bool:
 """Return True if an evaluation result's non-allowed decision is unverifiable.

 Because iam:SimulatePrincipalPolicy does not evaluate condition-based SCPs or
 condition-based permissions boundaries, an `implicitDeny` caused by (or
 masked by) `AllowedByOrganizations == false` or
 `AllowedByPermissionsBoundary == false` cannot be definitively verified by
 the client-side policy simulator.
 """
 if result.get("EvalDecision") == "explicitDeny":
 return False
 org_detail = result.get("OrganizationsDecisionDetail", {})
 org_allowed = org_detail.get("AllowedByOrganizations") if isinstance(org_detail, dict) else None
 pb_detail = result.get("PermissionsBoundaryDecisionDetail", {})
 pb_allowed = pb_detail.get("AllowedByPermissionsBoundary") if isinstance(pb_detail, dict) else None

 # Handle boolean False, string "false", "False", etc.
 if org_allowed in (False, "false", "False") or pb_allowed in (False, "false", "False"):
 return True
 return False


def _simulate_denied_actions(
 iam_client, role_arn: str, actions: List[str]
) -> Tuple[List[str], List[str]]:
 """Return lists of (denied_actions, unverifiable_actions) for ``role_arn``.

 Wraps the paginated ``iam:SimulatePrincipalPolicy`` call so both the role-
 validation path and the HyperPod caller-side check share one implementation.
 An empty list means every action is allowed. The pagination matters: a
 truncated first page must not produce a false "all allowed" verdict.
 The pagination matters: a truncated first page must not produce a false
 "all allowed" verdict.

 Returns:
 (denied_actions, unverifiable_actions):
 denied_actions: actions definitively denied (explicitDeny, or
 implicitDeny without org/permissions-boundary conditions).
 unverifiable_actions: actions returning implicitDeny with
 AllowedByOrganizations=False or AllowedByPermissionsBoundary=False,
 which the simulator cannot evaluate due to conditions.

 Raises ClientError on failures the caller must interpret (e.g. AccessDenied
 when the principal can't self-simulate, NoSuchEntity when the role is gone).
Expand All@@ -319,11 +352,16 @@ def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> L
 for page in paginator.paginate(PolicySourceArn=role_arn, ActionNames=actions):
 evaluation_results.extend(page.get("EvaluationResults", []))

 return [
 result["EvalActionName"]
 for result in evaluation_results
 if result["EvalDecision"] != "allowed"
 ]
 denied = []
 unverifiable = []
 for result in evaluation_results:
 if result.get("EvalDecision") != "allowed":
 action_name = result["EvalActionName"]
 if _is_unverifiable_denial(result):
 unverifiable.append(action_name)
 else:
 denied.append(action_name)
 return denied, unverifiable


def _evaluate_permissions(
Expand All@@ -339,21 +377,32 @@ def _evaluate_permissions(
 verdict True — all gated actions are allowed (denied_actions empty).
 verdict False — at least one gated action is denied (denied_actions lists them).
 verdict None — could not be determined, e.g. the caller lacks
 iam:SimulatePrincipalPolicy (denied_actions empty).
 iam:SimulatePrincipalPolicy or conditional SCPs prevent
 evaluation (denied_actions empty).
 """
 required_actions = _get_smoke_test_actions(role_type)
 if not required_actions:
 return True, []

 try:
 denied = _simulate_denied_actions(iam_client, role_arn, required_actions)
 denied, unverifiable = _simulate_denied_actions(iam_client, role_arn, required_actions)
 if denied:
 logger.info(
 "Role '%s' is missing permissions for: %s",
 role_arn,
 ", ".join(denied[:5]) + ("..." if len(denied) > 5 else ""),
 )
 return False, denied
 if unverifiable:
 logger.info(
 "Cannot definitively verify permissions for role '%s' due to "
 "Organizations SCPs or permissions boundaries (%s returned implicitDeny "
 "with AllowedByOrganizations/AllowedByPermissionsBoundary=false); "
 "permission verdict unknown.",
 role_arn,
 ", ".join(unverifiable[:5]) + ("..." if len(unverifiable) > 5 else ""),
 )
 return None, []
 return True, []

 except ClientError as e:
Expand DownExpand Up@@ -518,28 +567,35 @@ def _build_validation_error_message(


def resolve_and_validate_role(
 provided_role: Optional[str],
 role_type: str,
 provided_role: Optional[str] = None,
 role_type: str = "training",
 sagemaker_session=None,
 validate_role: bool = True,
 *,
 role_arn: Optional[str] = None,
) -> str:
 """Resolve the role to use and validate it (read-only; does not mutate IAM).

 Resolution:
 1. ``provided_role`` given → resolve it to an ARN (must exist).
 1. ``provided_role`` (or ``role_arn``) given → resolve it to an ARN (must exist).
 2. Otherwise → resolve the caller's own identity role.

 The resolved role is then VALIDATED (read-only, via iam:SimulatePrincipalPolicy
 + trust inspection):
 + trust inspection) unless ``validate_role=False`` or environment variable
 ``SAGEMAKER_VALIDATE_ROLE=false`` is set:
 * permissions allowed AND trusted → return the ARN.
 * a required permission is definitively denied → raise RoleValidationError.
 * the trust policy definitively excludes the service → raise RoleValidationError.
 * permissions cannot be verified (caller lacks iam:SimulatePrincipalPolicy,
 the common Studio/notebook case) → return the ARN with a WARNING.
 or conditional SCPs/permissions boundaries prevent evaluation) → return
 the ARN with a WARNING.

 Args:
 provided_role: User-supplied role name or ARN. If set, used directly.
 role_type: One of ROLE_TYPES.
 sagemaker_session: SageMaker session (used to get the boto session).
 validate_role: If False, skips client-side permission/trust validation.
 role_arn: Keyword alias for ``provided_role``.

 Returns:
 IAM role ARN.
Expand All@@ -552,6 +608,7 @@ def resolve_and_validate_role(
 if role_type not in ROLE_TYPES:
 raise ValueError(f"Invalid role_type '{role_type}'. Must be one of: {ROLE_TYPES}")

 provided_role = provided_role or role_arn
 boto_session = _get_boto_session(sagemaker_session)
 iam_client = boto_session.client("iam")

Expand All@@ -567,6 +624,10 @@ def resolve_and_validate_role(
 if not role_arn:
 raise RoleValidationError(_build_validation_error_message(None, role_type))

 if not validate_role or os.getenv("SAGEMAKER_VALIDATE_ROLE", "true").lower() in ("false", "0", "no"):
 logger.info("Skipping IAM role validation for '%s' (%s).", role_arn, role_type)
 return role_arn

 # Permission check (definitive denial blocks; unverifiable warns).
 verdict, denied = _evaluate_permissions(iam_client, role_arn, role_type)
 if verdict is False:
Expand All@@ -584,7 +645,8 @@ def resolve_and_validate_role(
 if verdict is None:
 logger.warning(
 "Could not verify permissions for role '%s' (caller lacks "
 "iam:SimulatePrincipalPolicy). Proceeding with it. If the operation "
 "iam:SimulatePrincipalPolicy or conditional SCPs/permissions boundaries "
 "prevent evaluation). Proceeding with it. If the operation "
 "later fails with an access-denied error, ensure the role has the "
 "required permissions for '%s' (see "
 "IamRoleResolver().get_required_actions('%s')) or create a dedicated "
Expand DownExpand Up@@ -643,7 +705,7 @@ def verify_hyperpod_connect_permissions(
 return None

 try:
 denied = _simulate_denied_actions(
 denied, unverifiable = _simulate_denied_actions(
 iam_client, caller_role_arn, list(HYPERPOD_CLI_CONNECT_ACTIONS)
 )
 except ClientError as e:
Expand DownExpand Up@@ -671,6 +733,15 @@ def verify_hyperpod_connect_permissions(
 )
 return False

 if unverifiable:
 logger.info(
 "Cannot definitively verify HyperPod connect permissions for '%s' due to "
 "Organizations SCPs or permissions boundaries; the HyperPod CLI will "
 "validate access at submit time.",
 caller_role_arn,
 )
 return None

 logger.info(
 "Caller '%s' has the HyperPod CLI connect permissions.", caller_role_arn
 )
Expand DownExpand Up@@ -732,6 +803,22 @@ def __init__(self, sagemaker_session=None):
 self._sts_client = self._boto_session.client("sts")

 # -- public API ---------------------------------------------------------
 def resolve_and_validate_role(
 self,
 provided_role: Optional[str] = None,
 role_type: str = "training",
 *,
 role_arn: Optional[str] = None,
 validate_role: bool = True,
 ) -> str:
 """Resolve and validate an IAM role (read-only; does not mutate IAM)."""
 return resolve_and_validate_role(
 provided_role=provided_role or role_arn,
 role_type=role_type,
 sagemaker_session=self._sagemaker_session,
 validate_role=validate_role,
 )

 def get_required_actions(self, role_type: str) -> List[str]:
 """Return the IAM actions a role of ``role_type`` needs (read-only preview)."""
 self._validate_role_type(role_type)
Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Highlight search terms from Google/DuckDuckGo/Bing referrer\n(function() {\n var ref = document.referrer;\n var terms = [];\n \n if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {\n var url = new URL(ref);\n var q = url.searchParams.get('q') || url.searchParams.get('p');\n if (q) {\n terms = q.split(/\\s+/).filter(function(t) { return t.length > 2; });\n }\n }\n \n if (terms.length === 0) return;\n \n var style = document.createElement('style');\n style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';\n document.head.appendChild(style);\n \n function highlight(node) {\n if (node.nodeType === 3) { // text node\n var text = node.textContent;\n var found = false;\n terms.forEach(function(term) {\n var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\\]\\\\]/g, '\\\\') + ')', 'gi');\n if (regex.test(text)) {\n found = true;\n var frag = document.createDocumentFragment();\n var parts = text.split(regex);\n parts.forEach(function(part, i) {\n if (i % 2 === 0) {\n frag.appendChild(document.createTextNode(part));\n } else {\n var span = document.createElement('span');\n span.className = 'userscript-highlight';\n span.textContent = part;\n frag.appendChild(span);\n }\n });\n node.parentNode.replaceChild(frag, node);\n }\n });\n } else if (node.nodeType === 1 && node.childNodes) { // element\n var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];\n if (!skipTags.includes(node.tagName)) {\n Array.from(node.childNodes).forEach(highlight);\n }\n }\n }\n \n highlight(document.body);\n \n // Re-highlight on dynamic content\n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1 || node.nodeType === 3) highlight(node);\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Highlight Search Terms");
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
fix(core): prevent false-positive RoleValidationError under condition…#6029

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

raza-khan0108
wants to merge
1
commit into

base:master
Choose a base branch

from

 Open


fix(core): prevent false-positive RoleValidationError under condition…
#6029
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


123 changes: 105 additions & 18 deletions
sagemaker-core/src/sagemaker/core/helper/iam_role_resolver.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@

import json
import logging
import os
import time
from typing import List, Optional, Set, Tuple, Union
from urllib.parse import unquote
Expand DownExpand Up@@ -303,13 +304,45 @@ def _resolve_caller_role_arn(
# ---------------------------------------------------------------------------
# Read-only permission / trust validation
# ---------------------------------------------------------------------------
def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> List[str]:
 """Return the subset of ``actions`` that ``role_arn`` is NOT allowed to perform.
def _is_unverifiable_denial(result: dict) -> bool:
 """Return True if an evaluation result's non-allowed decision is unverifiable.

 Because iam:SimulatePrincipalPolicy does not evaluate condition-based SCPs or
 condition-based permissions boundaries, an `implicitDeny` caused by (or
 masked by) `AllowedByOrganizations == false` or
 `AllowedByPermissionsBoundary == false` cannot be definitively verified by
 the client-side policy simulator.
 """
 if result.get("EvalDecision") == "explicitDeny":
 return False
 org_detail = result.get("OrganizationsDecisionDetail", {})
 org_allowed = org_detail.get("AllowedByOrganizations") if isinstance(org_detail, dict) else None
 pb_detail = result.get("PermissionsBoundaryDecisionDetail", {})
 pb_allowed = pb_detail.get("AllowedByPermissionsBoundary") if isinstance(pb_detail, dict) else None

 # Handle boolean False, string "false", "False", etc.
 if org_allowed in (False, "false", "False") or pb_allowed in (False, "false", "False"):
 return True
 return False


def _simulate_denied_actions(
 iam_client, role_arn: str, actions: List[str]
) -> Tuple[List[str], List[str]]:
 """Return lists of (denied_actions, unverifiable_actions) for ``role_arn``.

 Wraps the paginated ``iam:SimulatePrincipalPolicy`` call so both the role-
 validation path and the HyperPod caller-side check share one implementation.
 An empty list means every action is allowed. The pagination matters: a
 truncated first page must not produce a false "all allowed" verdict.
 The pagination matters: a truncated first page must not produce a false
 "all allowed" verdict.

 Returns:
 (denied_actions, unverifiable_actions):
 denied_actions: actions definitively denied (explicitDeny, or
 implicitDeny without org/permissions-boundary conditions).
 unverifiable_actions: actions returning implicitDeny with
 AllowedByOrganizations=False or AllowedByPermissionsBoundary=False,
 which the simulator cannot evaluate due to conditions.

 Raises ClientError on failures the caller must interpret (e.g. AccessDenied
 when the principal can't self-simulate, NoSuchEntity when the role is gone).
Expand All@@ -319,11 +352,16 @@ def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> L
 for page in paginator.paginate(PolicySourceArn=role_arn, ActionNames=actions):
 evaluation_results.extend(page.get("EvaluationResults", []))

 return [
 result["EvalActionName"]
 for result in evaluation_results
 if result["EvalDecision"] != "allowed"
 ]
 denied = []
 unverifiable = []
 for result in evaluation_results:
 if result.get("EvalDecision") != "allowed":
 action_name = result["EvalActionName"]
 if _is_unverifiable_denial(result):
 unverifiable.append(action_name)
 else:
 denied.append(action_name)
 return denied, unverifiable


def _evaluate_permissions(
Expand All@@ -339,21 +377,32 @@ def _evaluate_permissions(
 verdict True — all gated actions are allowed (denied_actions empty).
 verdict False — at least one gated action is denied (denied_actions lists them).
 verdict None — could not be determined, e.g. the caller lacks
 iam:SimulatePrincipalPolicy (denied_actions empty).
 iam:SimulatePrincipalPolicy or conditional SCPs prevent
 evaluation (denied_actions empty).
 """
 required_actions = _get_smoke_test_actions(role_type)
 if not required_actions:
 return True, []

 try:
 denied = _simulate_denied_actions(iam_client, role_arn, required_actions)
 denied, unverifiable = _simulate_denied_actions(iam_client, role_arn, required_actions)
 if denied:
 logger.info(
 "Role '%s' is missing permissions for: %s",
 role_arn,
 ", ".join(denied[:5]) + ("..." if len(denied) > 5 else ""),
 )
 return False, denied
 if unverifiable:
 logger.info(
 "Cannot definitively verify permissions for role '%s' due to "
 "Organizations SCPs or permissions boundaries (%s returned implicitDeny "
 "with AllowedByOrganizations/AllowedByPermissionsBoundary=false); "
 "permission verdict unknown.",
 role_arn,
 ", ".join(unverifiable[:5]) + ("..." if len(unverifiable) > 5 else ""),
 )
 return None, []
 return True, []

 except ClientError as e:
Expand DownExpand Up@@ -518,28 +567,35 @@ def _build_validation_error_message(


def resolve_and_validate_role(
 provided_role: Optional[str],
 role_type: str,
 provided_role: Optional[str] = None,
 role_type: str = "training",
 sagemaker_session=None,
 validate_role: bool = True,
 *,
 role_arn: Optional[str] = None,
) -> str:
 """Resolve the role to use and validate it (read-only; does not mutate IAM).

 Resolution:
 1. ``provided_role`` given → resolve it to an ARN (must exist).
 1. ``provided_role`` (or ``role_arn``) given → resolve it to an ARN (must exist).
 2. Otherwise → resolve the caller's own identity role.

 The resolved role is then VALIDATED (read-only, via iam:SimulatePrincipalPolicy
 + trust inspection):
 + trust inspection) unless ``validate_role=False`` or environment variable
 ``SAGEMAKER_VALIDATE_ROLE=false`` is set:
 * permissions allowed AND trusted → return the ARN.
 * a required permission is definitively denied → raise RoleValidationError.
 * the trust policy definitively excludes the service → raise RoleValidationError.
 * permissions cannot be verified (caller lacks iam:SimulatePrincipalPolicy,
 the common Studio/notebook case) → return the ARN with a WARNING.
 or conditional SCPs/permissions boundaries prevent evaluation) → return
 the ARN with a WARNING.

 Args:
 provided_role: User-supplied role name or ARN. If set, used directly.
 role_type: One of ROLE_TYPES.
 sagemaker_session: SageMaker session (used to get the boto session).
 validate_role: If False, skips client-side permission/trust validation.
 role_arn: Keyword alias for ``provided_role``.

 Returns:
 IAM role ARN.
Expand All@@ -552,6 +608,7 @@ def resolve_and_validate_role(
 if role_type not in ROLE_TYPES:
 raise ValueError(f"Invalid role_type '{role_type}'. Must be one of: {ROLE_TYPES}")

 provided_role = provided_role or role_arn
 boto_session = _get_boto_session(sagemaker_session)
 iam_client = boto_session.client("iam")

Expand All@@ -567,6 +624,10 @@ def resolve_and_validate_role(
 if not role_arn:
 raise RoleValidationError(_build_validation_error_message(None, role_type))

 if not validate_role or os.getenv("SAGEMAKER_VALIDATE_ROLE", "true").lower() in ("false", "0", "no"):
 logger.info("Skipping IAM role validation for '%s' (%s).", role_arn, role_type)
 return role_arn

 # Permission check (definitive denial blocks; unverifiable warns).
 verdict, denied = _evaluate_permissions(iam_client, role_arn, role_type)
 if verdict is False:
Expand All@@ -584,7 +645,8 @@ def resolve_and_validate_role(
 if verdict is None:
 logger.warning(
 "Could not verify permissions for role '%s' (caller lacks "
 "iam:SimulatePrincipalPolicy). Proceeding with it. If the operation "
 "iam:SimulatePrincipalPolicy or conditional SCPs/permissions boundaries "
 "prevent evaluation). Proceeding with it. If the operation "
 "later fails with an access-denied error, ensure the role has the "
 "required permissions for '%s' (see "
 "IamRoleResolver().get_required_actions('%s')) or create a dedicated "
Expand DownExpand Up@@ -643,7 +705,7 @@ def verify_hyperpod_connect_permissions(
 return None

 try:
 denied = _simulate_denied_actions(
 denied, unverifiable = _simulate_denied_actions(
 iam_client, caller_role_arn, list(HYPERPOD_CLI_CONNECT_ACTIONS)
 )
 except ClientError as e:
Expand DownExpand Up@@ -671,6 +733,15 @@ def verify_hyperpod_connect_permissions(
 )
 return False

 if unverifiable:
 logger.info(
 "Cannot definitively verify HyperPod connect permissions for '%s' due to "
 "Organizations SCPs or permissions boundaries; the HyperPod CLI will "
 "validate access at submit time.",
 caller_role_arn,
 )
 return None

 logger.info(
 "Caller '%s' has the HyperPod CLI connect permissions.", caller_role_arn
 )
Expand DownExpand Up@@ -732,6 +803,22 @@ def __init__(self, sagemaker_session=None):
 self._sts_client = self._boto_session.client("sts")

 # -- public API ---------------------------------------------------------
 def resolve_and_validate_role(
 self,
 provided_role: Optional[str] = None,
 role_type: str = "training",
 *,
 role_arn: Optional[str] = None,
 validate_role: bool = True,
 ) -> str:
 """Resolve and validate an IAM role (read-only; does not mutate IAM)."""
 return resolve_and_validate_role(
 provided_role=provided_role or role_arn,
 role_type=role_type,
 sagemaker_session=self._sagemaker_session,
 validate_role=validate_role,
 )

 def get_required_actions(self, role_type: str) -> List[str]:
 """Return the IAM actions a role of ``role_type`` needs (read-only preview)."""
 self._validate_role_type(role_type)
Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Strip utm_, fbclid, gclid, etc. from all links on page\n(function() {\n var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',\n 'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',\n 'ref', 'ref_src', 'source', 'medium', 'campaign'];\n \n function cleanUrl(url) {\n try {\n var u = new URL(url, window.location.origin);\n var changed = false;\n trackingParams.forEach(function(p) {\n if (u.searchParams.has(p)) {\n u.searchParams.delete(p);\n changed = true;\n }\n });\n return changed ? u.toString() : url;\n } catch (e) {\n return url;\n }\n }\n \n function cleanLinks() {\n document.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n \n cleanLinks();\n \n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1) {\n if (node.tagName === 'A') cleanLinks();\n node.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Remove Tracking Parameters from Links");
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '

Skip to content
fix(core): prevent false-positive RoleValidationError under condition…#6029

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

raza-khan0108
wants to merge
1
commit into

base:master
Choose a base branch

from

 Open


fix(core): prevent false-positive RoleValidationError under condition…
#6029
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


123 changes: 105 additions & 18 deletions
sagemaker-core/src/sagemaker/core/helper/iam_role_resolver.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@

import json
import logging
import os
import time
from typing import List, Optional, Set, Tuple, Union
from urllib.parse import unquote
Expand DownExpand Up@@ -303,13 +304,45 @@ def _resolve_caller_role_arn(
# ---------------------------------------------------------------------------
# Read-only permission / trust validation
# ---------------------------------------------------------------------------
def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> List[str]:
 """Return the subset of ``actions`` that ``role_arn`` is NOT allowed to perform.
def _is_unverifiable_denial(result: dict) -> bool:
 """Return True if an evaluation result's non-allowed decision is unverifiable.

 Because iam:SimulatePrincipalPolicy does not evaluate condition-based SCPs or
 condition-based permissions boundaries, an `implicitDeny` caused by (or
 masked by) `AllowedByOrganizations == false` or
 `AllowedByPermissionsBoundary == false` cannot be definitively verified by
 the client-side policy simulator.
 """
 if result.get("EvalDecision") == "explicitDeny":
 return False
 org_detail = result.get("OrganizationsDecisionDetail", {})
 org_allowed = org_detail.get("AllowedByOrganizations") if isinstance(org_detail, dict) else None
 pb_detail = result.get("PermissionsBoundaryDecisionDetail", {})
 pb_allowed = pb_detail.get("AllowedByPermissionsBoundary") if isinstance(pb_detail, dict) else None

 # Handle boolean False, string "false", "False", etc.
 if org_allowed in (False, "false", "False") or pb_allowed in (False, "false", "False"):
 return True
 return False


def _simulate_denied_actions(
 iam_client, role_arn: str, actions: List[str]
) -> Tuple[List[str], List[str]]:
 """Return lists of (denied_actions, unverifiable_actions) for ``role_arn``.

 Wraps the paginated ``iam:SimulatePrincipalPolicy`` call so both the role-
 validation path and the HyperPod caller-side check share one implementation.
 An empty list means every action is allowed. The pagination matters: a
 truncated first page must not produce a false "all allowed" verdict.
 The pagination matters: a truncated first page must not produce a false
 "all allowed" verdict.

 Returns:
 (denied_actions, unverifiable_actions):
 denied_actions: actions definitively denied (explicitDeny, or
 implicitDeny without org/permissions-boundary conditions).
 unverifiable_actions: actions returning implicitDeny with
 AllowedByOrganizations=False or AllowedByPermissionsBoundary=False,
 which the simulator cannot evaluate due to conditions.

 Raises ClientError on failures the caller must interpret (e.g. AccessDenied
 when the principal can't self-simulate, NoSuchEntity when the role is gone).
Expand All@@ -319,11 +352,16 @@ def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> L
 for page in paginator.paginate(PolicySourceArn=role_arn, ActionNames=actions):
 evaluation_results.extend(page.get("EvaluationResults", []))

 return [
 result["EvalActionName"]
 for result in evaluation_results
 if result["EvalDecision"] != "allowed"
 ]
 denied = []
 unverifiable = []
 for result in evaluation_results:
 if result.get("EvalDecision") != "allowed":
 action_name = result["EvalActionName"]
 if _is_unverifiable_denial(result):
 unverifiable.append(action_name)
 else:
 denied.append(action_name)
 return denied, unverifiable


def _evaluate_permissions(
Expand All@@ -339,21 +377,32 @@ def _evaluate_permissions(
 verdict True — all gated actions are allowed (denied_actions empty).
 verdict False — at least one gated action is denied (denied_actions lists them).
 verdict None — could not be determined, e.g. the caller lacks
 iam:SimulatePrincipalPolicy (denied_actions empty).
 iam:SimulatePrincipalPolicy or conditional SCPs prevent
 evaluation (denied_actions empty).
 """
 required_actions = _get_smoke_test_actions(role_type)
 if not required_actions:
 return True, []

 try:
 denied = _simulate_denied_actions(iam_client, role_arn, required_actions)
 denied, unverifiable = _simulate_denied_actions(iam_client, role_arn, required_actions)
 if denied:
 logger.info(
 "Role '%s' is missing permissions for: %s",
 role_arn,
 ", ".join(denied[:5]) + ("..." if len(denied) > 5 else ""),
 )
 return False, denied
 if unverifiable:
 logger.info(
 "Cannot definitively verify permissions for role '%s' due to "
 "Organizations SCPs or permissions boundaries (%s returned implicitDeny "
 "with AllowedByOrganizations/AllowedByPermissionsBoundary=false); "
 "permission verdict unknown.",
 role_arn,
 ", ".join(unverifiable[:5]) + ("..." if len(unverifiable) > 5 else ""),
 )
 return None, []
 return True, []

 except ClientError as e:
Expand DownExpand Up@@ -518,28 +567,35 @@ def _build_validation_error_message(


def resolve_and_validate_role(
 provided_role: Optional[str],
 role_type: str,
 provided_role: Optional[str] = None,
 role_type: str = "training",
 sagemaker_session=None,
 validate_role: bool = True,
 *,
 role_arn: Optional[str] = None,
) -> str:
 """Resolve the role to use and validate it (read-only; does not mutate IAM).

 Resolution:
 1. ``provided_role`` given → resolve it to an ARN (must exist).
 1. ``provided_role`` (or ``role_arn``) given → resolve it to an ARN (must exist).
 2. Otherwise → resolve the caller's own identity role.

 The resolved role is then VALIDATED (read-only, via iam:SimulatePrincipalPolicy
 + trust inspection):
 + trust inspection) unless ``validate_role=False`` or environment variable
 ``SAGEMAKER_VALIDATE_ROLE=false`` is set:
 * permissions allowed AND trusted → return the ARN.
 * a required permission is definitively denied → raise RoleValidationError.
 * the trust policy definitively excludes the service → raise RoleValidationError.
 * permissions cannot be verified (caller lacks iam:SimulatePrincipalPolicy,
 the common Studio/notebook case) → return the ARN with a WARNING.
 or conditional SCPs/permissions boundaries prevent evaluation) → return
 the ARN with a WARNING.

 Args:
 provided_role: User-supplied role name or ARN. If set, used directly.
 role_type: One of ROLE_TYPES.
 sagemaker_session: SageMaker session (used to get the boto session).
 validate_role: If False, skips client-side permission/trust validation.
 role_arn: Keyword alias for ``provided_role``.

 Returns:
 IAM role ARN.
Expand All@@ -552,6 +608,7 @@ def resolve_and_validate_role(
 if role_type not in ROLE_TYPES:
 raise ValueError(f"Invalid role_type '{role_type}'. Must be one of: {ROLE_TYPES}")

 provided_role = provided_role or role_arn
 boto_session = _get_boto_session(sagemaker_session)
 iam_client = boto_session.client("iam")

Expand All@@ -567,6 +624,10 @@ def resolve_and_validate_role(
 if not role_arn:
 raise RoleValidationError(_build_validation_error_message(None, role_type))

 if not validate_role or os.getenv("SAGEMAKER_VALIDATE_ROLE", "true").lower() in ("false", "0", "no"):
 logger.info("Skipping IAM role validation for '%s' (%s).", role_arn, role_type)
 return role_arn

 # Permission check (definitive denial blocks; unverifiable warns).
 verdict, denied = _evaluate_permissions(iam_client, role_arn, role_type)
 if verdict is False:
Expand All@@ -584,7 +645,8 @@ def resolve_and_validate_role(
 if verdict is None:
 logger.warning(
 "Could not verify permissions for role '%s' (caller lacks "
 "iam:SimulatePrincipalPolicy). Proceeding with it. If the operation "
 "iam:SimulatePrincipalPolicy or conditional SCPs/permissions boundaries "
 "prevent evaluation). Proceeding with it. If the operation "
 "later fails with an access-denied error, ensure the role has the "
 "required permissions for '%s' (see "
 "IamRoleResolver().get_required_actions('%s')) or create a dedicated "
Expand DownExpand Up@@ -643,7 +705,7 @@ def verify_hyperpod_connect_permissions(
 return None

 try:
 denied = _simulate_denied_actions(
 denied, unverifiable = _simulate_denied_actions(
 iam_client, caller_role_arn, list(HYPERPOD_CLI_CONNECT_ACTIONS)
 )
 except ClientError as e:
Expand DownExpand Up@@ -671,6 +733,15 @@ def verify_hyperpod_connect_permissions(
 )
 return False

 if unverifiable:
 logger.info(
 "Cannot definitively verify HyperPod connect permissions for '%s' due to "
 "Organizations SCPs or permissions boundaries; the HyperPod CLI will "
 "validate access at submit time.",
 caller_role_arn,
 )
 return None

 logger.info(
 "Caller '%s' has the HyperPod CLI connect permissions.", caller_role_arn
 )
Expand DownExpand Up@@ -732,6 +803,22 @@ def __init__(self, sagemaker_session=None):
 self._sts_client = self._boto_session.client("sts")

 # -- public API ---------------------------------------------------------
 def resolve_and_validate_role(
 self,
 provided_role: Optional[str] = None,
 role_type: str = "training",
 *,
 role_arn: Optional[str] = None,
 validate_role: bool = True,
 ) -> str:
 """Resolve and validate an IAM role (read-only; does not mutate IAM)."""
 return resolve_and_validate_role(
 provided_role=provided_role or role_arn,
 role_type=role_type,
 sagemaker_session=self._sagemaker_session,
 validate_role=validate_role,
 )

 def get_required_actions(self, role_type: str) -> List[str]:
 """Return the IAM actions a role of ``role_type`` needs (read-only preview)."""
 self._validate_role_type(role_type)
Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Auto-enable theater mode on YouTube\n(function() {\n function tryTheater() {\n var btn = document.querySelector('button[aria-label=\"Theater mode\"], ytd-player #player button[title=\"Theater mode\"]');\n if (btn && !btn.classList.contains('activated')) {\n btn.click();\n }\n }\n \n // Try immediately\n tryTheater();\n \n // Try after navigation (SPA)\n var lastUrl = location.href;\n setInterval(function() {\n if (location.href !== lastUrl) {\n lastUrl = location.href;\n setTimeout(tryTheater, 500);\n }\n }, 1000);\n \n // Also try on player load\n var observer = new MutationObserver(tryTheater);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "YouTube Theater Mode Default");
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
fix(core): prevent false-positive RoleValidationError under condition…#6029

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

raza-khan0108
wants to merge
1
commit into

base:master
Choose a base branch

from

 Open


fix(core): prevent false-positive RoleValidationError under condition…
#6029
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


123 changes: 105 additions & 18 deletions
sagemaker-core/src/sagemaker/core/helper/iam_role_resolver.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@

import json
import logging
import os
import time
from typing import List, Optional, Set, Tuple, Union
from urllib.parse import unquote
Expand DownExpand Up@@ -303,13 +304,45 @@ def _resolve_caller_role_arn(
# ---------------------------------------------------------------------------
# Read-only permission / trust validation
# ---------------------------------------------------------------------------
def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> List[str]:
 """Return the subset of ``actions`` that ``role_arn`` is NOT allowed to perform.
def _is_unverifiable_denial(result: dict) -> bool:
 """Return True if an evaluation result's non-allowed decision is unverifiable.

 Because iam:SimulatePrincipalPolicy does not evaluate condition-based SCPs or
 condition-based permissions boundaries, an `implicitDeny` caused by (or
 masked by) `AllowedByOrganizations == false` or
 `AllowedByPermissionsBoundary == false` cannot be definitively verified by
 the client-side policy simulator.
 """
 if result.get("EvalDecision") == "explicitDeny":
 return False
 org_detail = result.get("OrganizationsDecisionDetail", {})
 org_allowed = org_detail.get("AllowedByOrganizations") if isinstance(org_detail, dict) else None
 pb_detail = result.get("PermissionsBoundaryDecisionDetail", {})
 pb_allowed = pb_detail.get("AllowedByPermissionsBoundary") if isinstance(pb_detail, dict) else None

 # Handle boolean False, string "false", "False", etc.
 if org_allowed in (False, "false", "False") or pb_allowed in (False, "false", "False"):
 return True
 return False


def _simulate_denied_actions(
 iam_client, role_arn: str, actions: List[str]
) -> Tuple[List[str], List[str]]:
 """Return lists of (denied_actions, unverifiable_actions) for ``role_arn``.

 Wraps the paginated ``iam:SimulatePrincipalPolicy`` call so both the role-
 validation path and the HyperPod caller-side check share one implementation.
 An empty list means every action is allowed. The pagination matters: a
 truncated first page must not produce a false "all allowed" verdict.
 The pagination matters: a truncated first page must not produce a false
 "all allowed" verdict.

 Returns:
 (denied_actions, unverifiable_actions):
 denied_actions: actions definitively denied (explicitDeny, or
 implicitDeny without org/permissions-boundary conditions).
 unverifiable_actions: actions returning implicitDeny with
 AllowedByOrganizations=False or AllowedByPermissionsBoundary=False,
 which the simulator cannot evaluate due to conditions.

 Raises ClientError on failures the caller must interpret (e.g. AccessDenied
 when the principal can't self-simulate, NoSuchEntity when the role is gone).
Expand All@@ -319,11 +352,16 @@ def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> L
 for page in paginator.paginate(PolicySourceArn=role_arn, ActionNames=actions):
 evaluation_results.extend(page.get("EvaluationResults", []))

 return [
 result["EvalActionName"]
 for result in evaluation_results
 if result["EvalDecision"] != "allowed"
 ]
 denied = []
 unverifiable = []
 for result in evaluation_results:
 if result.get("EvalDecision") != "allowed":
 action_name = result["EvalActionName"]
 if _is_unverifiable_denial(result):
 unverifiable.append(action_name)
 else:
 denied.append(action_name)
 return denied, unverifiable


def _evaluate_permissions(
Expand All@@ -339,21 +377,32 @@ def _evaluate_permissions(
 verdict True — all gated actions are allowed (denied_actions empty).
 verdict False — at least one gated action is denied (denied_actions lists them).
 verdict None — could not be determined, e.g. the caller lacks
 iam:SimulatePrincipalPolicy (denied_actions empty).
 iam:SimulatePrincipalPolicy or conditional SCPs prevent
 evaluation (denied_actions empty).
 """
 required_actions = _get_smoke_test_actions(role_type)
 if not required_actions:
 return True, []

 try:
 denied = _simulate_denied_actions(iam_client, role_arn, required_actions)
 denied, unverifiable = _simulate_denied_actions(iam_client, role_arn, required_actions)
 if denied:
 logger.info(
 "Role '%s' is missing permissions for: %s",
 role_arn,
 ", ".join(denied[:5]) + ("..." if len(denied) > 5 else ""),
 )
 return False, denied
 if unverifiable:
 logger.info(
 "Cannot definitively verify permissions for role '%s' due to "
 "Organizations SCPs or permissions boundaries (%s returned implicitDeny "
 "with AllowedByOrganizations/AllowedByPermissionsBoundary=false); "
 "permission verdict unknown.",
 role_arn,
 ", ".join(unverifiable[:5]) + ("..." if len(unverifiable) > 5 else ""),
 )
 return None, []
 return True, []

 except ClientError as e:
Expand DownExpand Up@@ -518,28 +567,35 @@ def _build_validation_error_message(


def resolve_and_validate_role(
 provided_role: Optional[str],
 role_type: str,
 provided_role: Optional[str] = None,
 role_type: str = "training",
 sagemaker_session=None,
 validate_role: bool = True,
 *,
 role_arn: Optional[str] = None,
) -> str:
 """Resolve the role to use and validate it (read-only; does not mutate IAM).

 Resolution:
 1. ``provided_role`` given → resolve it to an ARN (must exist).
 1. ``provided_role`` (or ``role_arn``) given → resolve it to an ARN (must exist).
 2. Otherwise → resolve the caller's own identity role.

 The resolved role is then VALIDATED (read-only, via iam:SimulatePrincipalPolicy
 + trust inspection):
 + trust inspection) unless ``validate_role=False`` or environment variable
 ``SAGEMAKER_VALIDATE_ROLE=false`` is set:
 * permissions allowed AND trusted → return the ARN.
 * a required permission is definitively denied → raise RoleValidationError.
 * the trust policy definitively excludes the service → raise RoleValidationError.
 * permissions cannot be verified (caller lacks iam:SimulatePrincipalPolicy,
 the common Studio/notebook case) → return the ARN with a WARNING.
 or conditional SCPs/permissions boundaries prevent evaluation) → return
 the ARN with a WARNING.

 Args:
 provided_role: User-supplied role name or ARN. If set, used directly.
 role_type: One of ROLE_TYPES.
 sagemaker_session: SageMaker session (used to get the boto session).
 validate_role: If False, skips client-side permission/trust validation.
 role_arn: Keyword alias for ``provided_role``.

 Returns:
 IAM role ARN.
Expand All@@ -552,6 +608,7 @@ def resolve_and_validate_role(
 if role_type not in ROLE_TYPES:
 raise ValueError(f"Invalid role_type '{role_type}'. Must be one of: {ROLE_TYPES}")

 provided_role = provided_role or role_arn
 boto_session = _get_boto_session(sagemaker_session)
 iam_client = boto_session.client("iam")

Expand All@@ -567,6 +624,10 @@ def resolve_and_validate_role(
 if not role_arn:
 raise RoleValidationError(_build_validation_error_message(None, role_type))

 if not validate_role or os.getenv("SAGEMAKER_VALIDATE_ROLE", "true").lower() in ("false", "0", "no"):
 logger.info("Skipping IAM role validation for '%s' (%s).", role_arn, role_type)
 return role_arn

 # Permission check (definitive denial blocks; unverifiable warns).
 verdict, denied = _evaluate_permissions(iam_client, role_arn, role_type)
 if verdict is False:
Expand All@@ -584,7 +645,8 @@ def resolve_and_validate_role(
 if verdict is None:
 logger.warning(
 "Could not verify permissions for role '%s' (caller lacks "
 "iam:SimulatePrincipalPolicy). Proceeding with it. If the operation "
 "iam:SimulatePrincipalPolicy or conditional SCPs/permissions boundaries "
 "prevent evaluation). Proceeding with it. If the operation "
 "later fails with an access-denied error, ensure the role has the "
 "required permissions for '%s' (see "
 "IamRoleResolver().get_required_actions('%s')) or create a dedicated "
Expand DownExpand Up@@ -643,7 +705,7 @@ def verify_hyperpod_connect_permissions(
 return None

 try:
 denied = _simulate_denied_actions(
 denied, unverifiable = _simulate_denied_actions(
 iam_client, caller_role_arn, list(HYPERPOD_CLI_CONNECT_ACTIONS)
 )
 except ClientError as e:
Expand DownExpand Up@@ -671,6 +733,15 @@ def verify_hyperpod_connect_permissions(
 )
 return False

 if unverifiable:
 logger.info(
 "Cannot definitively verify HyperPod connect permissions for '%s' due to "
 "Organizations SCPs or permissions boundaries; the HyperPod CLI will "
 "validate access at submit time.",
 caller_role_arn,
 )
 return None

 logger.info(
 "Caller '%s' has the HyperPod CLI connect permissions.", caller_role_arn
 )
Expand DownExpand Up@@ -732,6 +803,22 @@ def __init__(self, sagemaker_session=None):
 self._sts_client = self._boto_session.client("sts")

 # -- public API ---------------------------------------------------------
 def resolve_and_validate_role(
 self,
 provided_role: Optional[str] = None,
 role_type: str = "training",
 *,
 role_arn: Optional[str] = None,
 validate_role: bool = True,
 ) -> str:
 """Resolve and validate an IAM role (read-only; does not mutate IAM)."""
 return resolve_and_validate_role(
 provided_role=provided_role or role_arn,
 role_type=role_type,
 sagemaker_session=self._sagemaker_session,
 validate_role=validate_role,
 )

 def get_required_actions(self, role_type: str) -> List[str]:
 """Return the IAM actions a role of ``role_type`` needs (read-only preview)."""
 self._validate_role_type(role_type)
Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Remove or un-stick sticky/fixed headers that block content\n(function() {\n function unstick() {\n document.querySelectorAll('header, nav, [role=\"banner\"], .header, .navbar, .sticky, .fixed-top, [style*=\"position: fixed\"], [style*=\"position:sticky\"]').forEach(function(el) {\n if (el.style.position === 'fixed' || el.style.position === 'sticky' || \n getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {\n el.style.position = 'static';\n el.style.top = 'auto';\n el.style.zIndex = 'auto';\n }\n });\n }\n \n unstick();\n \n var observer = new MutationObserver(unstick);\n observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });\n})();", "Kill Sticky Headers");
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
fix(core): prevent false-positive RoleValidationError under condition…#6029

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

raza-khan0108
wants to merge
1
commit into

base:master
Choose a base branch

from

 Open


fix(core): prevent false-positive RoleValidationError under condition…
#6029
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


123 changes: 105 additions & 18 deletions
sagemaker-core/src/sagemaker/core/helper/iam_role_resolver.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@

import json
import logging
import os
import time
from typing import List, Optional, Set, Tuple, Union
from urllib.parse import unquote
Expand DownExpand Up@@ -303,13 +304,45 @@ def _resolve_caller_role_arn(
# ---------------------------------------------------------------------------
# Read-only permission / trust validation
# ---------------------------------------------------------------------------
def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> List[str]:
 """Return the subset of ``actions`` that ``role_arn`` is NOT allowed to perform.
def _is_unverifiable_denial(result: dict) -> bool:
 """Return True if an evaluation result's non-allowed decision is unverifiable.

 Because iam:SimulatePrincipalPolicy does not evaluate condition-based SCPs or
 condition-based permissions boundaries, an `implicitDeny` caused by (or
 masked by) `AllowedByOrganizations == false` or
 `AllowedByPermissionsBoundary == false` cannot be definitively verified by
 the client-side policy simulator.
 """
 if result.get("EvalDecision") == "explicitDeny":
 return False
 org_detail = result.get("OrganizationsDecisionDetail", {})
 org_allowed = org_detail.get("AllowedByOrganizations") if isinstance(org_detail, dict) else None
 pb_detail = result.get("PermissionsBoundaryDecisionDetail", {})
 pb_allowed = pb_detail.get("AllowedByPermissionsBoundary") if isinstance(pb_detail, dict) else None

 # Handle boolean False, string "false", "False", etc.
 if org_allowed in (False, "false", "False") or pb_allowed in (False, "false", "False"):
 return True
 return False


def _simulate_denied_actions(
 iam_client, role_arn: str, actions: List[str]
) -> Tuple[List[str], List[str]]:
 """Return lists of (denied_actions, unverifiable_actions) for ``role_arn``.

 Wraps the paginated ``iam:SimulatePrincipalPolicy`` call so both the role-
 validation path and the HyperPod caller-side check share one implementation.
 An empty list means every action is allowed. The pagination matters: a
 truncated first page must not produce a false "all allowed" verdict.
 The pagination matters: a truncated first page must not produce a false
 "all allowed" verdict.

 Returns:
 (denied_actions, unverifiable_actions):
 denied_actions: actions definitively denied (explicitDeny, or
 implicitDeny without org/permissions-boundary conditions).
 unverifiable_actions: actions returning implicitDeny with
 AllowedByOrganizations=False or AllowedByPermissionsBoundary=False,
 which the simulator cannot evaluate due to conditions.

 Raises ClientError on failures the caller must interpret (e.g. AccessDenied
 when the principal can't self-simulate, NoSuchEntity when the role is gone).
Expand All@@ -319,11 +352,16 @@ def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> L
 for page in paginator.paginate(PolicySourceArn=role_arn, ActionNames=actions):
 evaluation_results.extend(page.get("EvaluationResults", []))

 return [
 result["EvalActionName"]
 for result in evaluation_results
 if result["EvalDecision"] != "allowed"
 ]
 denied = []
 unverifiable = []
 for result in evaluation_results:
 if result.get("EvalDecision") != "allowed":
 action_name = result["EvalActionName"]
 if _is_unverifiable_denial(result):
 unverifiable.append(action_name)
 else:
 denied.append(action_name)
 return denied, unverifiable


def _evaluate_permissions(
Expand All@@ -339,21 +377,32 @@ def _evaluate_permissions(
 verdict True — all gated actions are allowed (denied_actions empty).
 verdict False — at least one gated action is denied (denied_actions lists them).
 verdict None — could not be determined, e.g. the caller lacks
 iam:SimulatePrincipalPolicy (denied_actions empty).
 iam:SimulatePrincipalPolicy or conditional SCPs prevent
 evaluation (denied_actions empty).
 """
 required_actions = _get_smoke_test_actions(role_type)
 if not required_actions:
 return True, []

 try:
 denied = _simulate_denied_actions(iam_client, role_arn, required_actions)
 denied, unverifiable = _simulate_denied_actions(iam_client, role_arn, required_actions)
 if denied:
 logger.info(
 "Role '%s' is missing permissions for: %s",
 role_arn,
 ", ".join(denied[:5]) + ("..." if len(denied) > 5 else ""),
 )
 return False, denied
 if unverifiable:
 logger.info(
 "Cannot definitively verify permissions for role '%s' due to "
 "Organizations SCPs or permissions boundaries (%s returned implicitDeny "
 "with AllowedByOrganizations/AllowedByPermissionsBoundary=false); "
 "permission verdict unknown.",
 role_arn,
 ", ".join(unverifiable[:5]) + ("..." if len(unverifiable) > 5 else ""),
 )
 return None, []
 return True, []

 except ClientError as e:
Expand DownExpand Up@@ -518,28 +567,35 @@ def _build_validation_error_message(


def resolve_and_validate_role(
 provided_role: Optional[str],
 role_type: str,
 provided_role: Optional[str] = None,
 role_type: str = "training",
 sagemaker_session=None,
 validate_role: bool = True,
 *,
 role_arn: Optional[str] = None,
) -> str:
 """Resolve the role to use and validate it (read-only; does not mutate IAM).

 Resolution:
 1. ``provided_role`` given → resolve it to an ARN (must exist).
 1. ``provided_role`` (or ``role_arn``) given → resolve it to an ARN (must exist).
 2. Otherwise → resolve the caller's own identity role.

 The resolved role is then VALIDATED (read-only, via iam:SimulatePrincipalPolicy
 + trust inspection):
 + trust inspection) unless ``validate_role=False`` or environment variable
 ``SAGEMAKER_VALIDATE_ROLE=false`` is set:
 * permissions allowed AND trusted → return the ARN.
 * a required permission is definitively denied → raise RoleValidationError.
 * the trust policy definitively excludes the service → raise RoleValidationError.
 * permissions cannot be verified (caller lacks iam:SimulatePrincipalPolicy,
 the common Studio/notebook case) → return the ARN with a WARNING.
 or conditional SCPs/permissions boundaries prevent evaluation) → return
 the ARN with a WARNING.

 Args:
 provided_role: User-supplied role name or ARN. If set, used directly.
 role_type: One of ROLE_TYPES.
 sagemaker_session: SageMaker session (used to get the boto session).
 validate_role: If False, skips client-side permission/trust validation.
 role_arn: Keyword alias for ``provided_role``.

 Returns:
 IAM role ARN.
Expand All@@ -552,6 +608,7 @@ def resolve_and_validate_role(
 if role_type not in ROLE_TYPES:
 raise ValueError(f"Invalid role_type '{role_type}'. Must be one of: {ROLE_TYPES}")

 provided_role = provided_role or role_arn
 boto_session = _get_boto_session(sagemaker_session)
 iam_client = boto_session.client("iam")

Expand All@@ -567,6 +624,10 @@ def resolve_and_validate_role(
 if not role_arn:
 raise RoleValidationError(_build_validation_error_message(None, role_type))

 if not validate_role or os.getenv("SAGEMAKER_VALIDATE_ROLE", "true").lower() in ("false", "0", "no"):
 logger.info("Skipping IAM role validation for '%s' (%s).", role_arn, role_type)
 return role_arn

 # Permission check (definitive denial blocks; unverifiable warns).
 verdict, denied = _evaluate_permissions(iam_client, role_arn, role_type)
 if verdict is False:
Expand All@@ -584,7 +645,8 @@ def resolve_and_validate_role(
 if verdict is None:
 logger.warning(
 "Could not verify permissions for role '%s' (caller lacks "
 "iam:SimulatePrincipalPolicy). Proceeding with it. If the operation "
 "iam:SimulatePrincipalPolicy or conditional SCPs/permissions boundaries "
 "prevent evaluation). Proceeding with it. If the operation "
 "later fails with an access-denied error, ensure the role has the "
 "required permissions for '%s' (see "
 "IamRoleResolver().get_required_actions('%s')) or create a dedicated "
Expand DownExpand Up@@ -643,7 +705,7 @@ def verify_hyperpod_connect_permissions(
 return None

 try:
 denied = _simulate_denied_actions(
 denied, unverifiable = _simulate_denied_actions(
 iam_client, caller_role_arn, list(HYPERPOD_CLI_CONNECT_ACTIONS)
 )
 except ClientError as e:
Expand DownExpand Up@@ -671,6 +733,15 @@ def verify_hyperpod_connect_permissions(
 )
 return False

 if unverifiable:
 logger.info(
 "Cannot definitively verify HyperPod connect permissions for '%s' due to "
 "Organizations SCPs or permissions boundaries; the HyperPod CLI will "
 "validate access at submit time.",
 caller_role_arn,
 )
 return None

 logger.info(
 "Caller '%s' has the HyperPod CLI connect permissions.", caller_role_arn
 )
Expand DownExpand Up@@ -732,6 +803,22 @@ def __init__(self, sagemaker_session=None):
 self._sts_client = self._boto_session.client("sts")

 # -- public API ---------------------------------------------------------
 def resolve_and_validate_role(
 self,
 provided_role: Optional[str] = None,
 role_type: str = "training",
 *,
 role_arn: Optional[str] = None,
 validate_role: bool = True,
 ) -> str:
 """Resolve and validate an IAM role (read-only; does not mutate IAM)."""
 return resolve_and_validate_role(
 provided_role=provided_role or role_arn,
 role_type=role_type,
 sagemaker_session=self._sagemaker_session,
 validate_role=validate_role,
 )

 def get_required_actions(self, role_type: str) -> List[str]:
 """Return the IAM actions a role of ``role_type`` needs (read-only preview)."""
 self._validate_role_type(role_type)
Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Universal Dark Mode - works on any site\n(function() {\n var enabled = true;\n \n function applyDarkMode() {\n if (!enabled) return;\n \n // Create style element if it doesn't exist\n var style = document.getElementById('universal-dark-mode-style');\n if (!style) {\n style = document.createElement('style');\n style.id = 'universal-dark-mode-style';\n document.head.appendChild(style);\n }\n \n // Dark mode CSS - inverts colors but preserves images/video\n style.textContent = '\n /* Invert everything except media */\n html {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #1a1a2e !important;\n }\n \n /* Restore images, videos, iframes, canvas */\n img, video, iframe, canvas, svg, picture, [style*=\"background-image\"] {\n filter: invert(1) hue-rotate(180deg) !important;\n }\n \n /* Preserve specific elements that should not be inverted */\n .no-dark-mode, .no-dark-mode *,\n [data-theme=\"light\"], [data-theme=\"light\"],\n .ace_editor, .ace_editor *,\n .CodeMirror, .CodeMirror *,\n .monaco-editor, .monaco-editor *,\n .markdown-body pre, .markdown-body pre *,\n .highlight, .highlight *,\n pre code, pre code * {\n filter: none !important;\n }\n \n /* Fix common UI elements */\n .modal, .popup, .dropdown-menu, .tooltip, .popover {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #2d2d44 !important;\n border-color: #444 !important;\n }\n \n /* Scrollbars */\n ::-webkit-scrollbar { background: #1a1a2e !important; }\n ::-webkit-scrollbar-thumb { background: #444 !important; }\n ::-webkit-scrollbar-thumb:hover { background: #555 !important; }\n \n /* Selection */\n ::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ';\n }\n \n function removeDarkMode() {\n var style = document.getElementById('universal-dark-mode-style');\n if (style) style.remove();\n }\n \n // Toggle with Alt+Shift+D\n document.addEventListener('keydown', function(e) {\n if (e.altKey && e.shiftKey && e.key === 'D') {\n e.preventDefault();\n enabled = !enabled;\n if (enabled) {\n applyDarkMode();\n console.log('[Universal Dark Mode] Enabled');\n } else {\n removeDarkMode();\n console.log('[Universal Dark Mode] Disabled');\n }\n }\n });\n \n // Apply on load\n applyDarkMode();\n \n // Re-apply on dynamic content\n var observer = new MutationObserver(function(mutations) {\n if (enabled && !document.getElementById('universal-dark-mode-style')) {\n applyDarkMode();\n }\n });\n observer.observe(document.head, { childList: true });\n \n console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');\n})();", "Universal Dark Mode");
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();

Skip to content
fix(core): prevent false-positive RoleValidationError under condition…#6029

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Open

raza-khan0108
wants to merge
1
commit into

base:master
Choose a base branch

from

 Open


fix(core): prevent false-positive RoleValidationError under condition…
#6029
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


1 commit


Select commit


File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


123 changes: 105 additions & 18 deletions
sagemaker-core/src/sagemaker/core/helper/iam_role_resolver.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@

import json
import logging
import os
import time
from typing import List, Optional, Set, Tuple, Union
from urllib.parse import unquote
Expand DownExpand Up@@ -303,13 +304,45 @@ def _resolve_caller_role_arn(
# ---------------------------------------------------------------------------
# Read-only permission / trust validation
# ---------------------------------------------------------------------------
def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> List[str]:
 """Return the subset of ``actions`` that ``role_arn`` is NOT allowed to perform.
def _is_unverifiable_denial(result: dict) -> bool:
 """Return True if an evaluation result's non-allowed decision is unverifiable.

 Because iam:SimulatePrincipalPolicy does not evaluate condition-based SCPs or
 condition-based permissions boundaries, an `implicitDeny` caused by (or
 masked by) `AllowedByOrganizations == false` or
 `AllowedByPermissionsBoundary == false` cannot be definitively verified by
 the client-side policy simulator.
 """
 if result.get("EvalDecision") == "explicitDeny":
 return False
 org_detail = result.get("OrganizationsDecisionDetail", {})
 org_allowed = org_detail.get("AllowedByOrganizations") if isinstance(org_detail, dict) else None
 pb_detail = result.get("PermissionsBoundaryDecisionDetail", {})
 pb_allowed = pb_detail.get("AllowedByPermissionsBoundary") if isinstance(pb_detail, dict) else None

 # Handle boolean False, string "false", "False", etc.
 if org_allowed in (False, "false", "False") or pb_allowed in (False, "false", "False"):
 return True
 return False


def _simulate_denied_actions(
 iam_client, role_arn: str, actions: List[str]
) -> Tuple[List[str], List[str]]:
 """Return lists of (denied_actions, unverifiable_actions) for ``role_arn``.

 Wraps the paginated ``iam:SimulatePrincipalPolicy`` call so both the role-
 validation path and the HyperPod caller-side check share one implementation.
 An empty list means every action is allowed. The pagination matters: a
 truncated first page must not produce a false "all allowed" verdict.
 The pagination matters: a truncated first page must not produce a false
 "all allowed" verdict.

 Returns:
 (denied_actions, unverifiable_actions):
 denied_actions: actions definitively denied (explicitDeny, or
 implicitDeny without org/permissions-boundary conditions).
 unverifiable_actions: actions returning implicitDeny with
 AllowedByOrganizations=False or AllowedByPermissionsBoundary=False,
 which the simulator cannot evaluate due to conditions.

 Raises ClientError on failures the caller must interpret (e.g. AccessDenied
 when the principal can't self-simulate, NoSuchEntity when the role is gone).
Expand All@@ -319,11 +352,16 @@ def _simulate_denied_actions(iam_client, role_arn: str, actions: List[str]) -> L
 for page in paginator.paginate(PolicySourceArn=role_arn, ActionNames=actions):
 evaluation_results.extend(page.get("EvaluationResults", []))

 return [
 result["EvalActionName"]
 for result in evaluation_results
 if result["EvalDecision"] != "allowed"
 ]
 denied = []
 unverifiable = []
 for result in evaluation_results:
 if result.get("EvalDecision") != "allowed":
 action_name = result["EvalActionName"]
 if _is_unverifiable_denial(result):
 unverifiable.append(action_name)
 else:
 denied.append(action_name)
 return denied, unverifiable


def _evaluate_permissions(
Expand All@@ -339,21 +377,32 @@ def _evaluate_permissions(
 verdict True — all gated actions are allowed (denied_actions empty).
 verdict False — at least one gated action is denied (denied_actions lists them).
 verdict None — could not be determined, e.g. the caller lacks
 iam:SimulatePrincipalPolicy (denied_actions empty).
 iam:SimulatePrincipalPolicy or conditional SCPs prevent
 evaluation (denied_actions empty).
 """
 required_actions = _get_smoke_test_actions(role_type)
 if not required_actions:
 return True, []

 try:
 denied = _simulate_denied_actions(iam_client, role_arn, required_actions)
 denied, unverifiable = _simulate_denied_actions(iam_client, role_arn, required_actions)
 if denied:
 logger.info(
 "Role '%s' is missing permissions for: %s",
 role_arn,
 ", ".join(denied[:5]) + ("..." if len(denied) > 5 else ""),
 )
 return False, denied
 if unverifiable:
 logger.info(
 "Cannot definitively verify permissions for role '%s' due to "
 "Organizations SCPs or permissions boundaries (%s returned implicitDeny "
 "with AllowedByOrganizations/AllowedByPermissionsBoundary=false); "
 "permission verdict unknown.",
 role_arn,
 ", ".join(unverifiable[:5]) + ("..." if len(unverifiable) > 5 else ""),
 )
 return None, []
 return True, []

 except ClientError as e:
Expand DownExpand Up@@ -518,28 +567,35 @@ def _build_validation_error_message(


def resolve_and_validate_role(
 provided_role: Optional[str],
 role_type: str,
 provided_role: Optional[str] = None,
 role_type: str = "training",
 sagemaker_session=None,
 validate_role: bool = True,
 *,
 role_arn: Optional[str] = None,
) -> str:
 """Resolve the role to use and validate it (read-only; does not mutate IAM).

 Resolution:
 1. ``provided_role`` given → resolve it to an ARN (must exist).
 1. ``provided_role`` (or ``role_arn``) given → resolve it to an ARN (must exist).
 2. Otherwise → resolve the caller's own identity role.

 The resolved role is then VALIDATED (read-only, via iam:SimulatePrincipalPolicy
 + trust inspection):
 + trust inspection) unless ``validate_role=False`` or environment variable
 ``SAGEMAKER_VALIDATE_ROLE=false`` is set:
 * permissions allowed AND trusted → return the ARN.
 * a required permission is definitively denied → raise RoleValidationError.
 * the trust policy definitively excludes the service → raise RoleValidationError.
 * permissions cannot be verified (caller lacks iam:SimulatePrincipalPolicy,
 the common Studio/notebook case) → return the ARN with a WARNING.
 or conditional SCPs/permissions boundaries prevent evaluation) → return
 the ARN with a WARNING.

 Args:
 provided_role: User-supplied role name or ARN. If set, used directly.
 role_type: One of ROLE_TYPES.
 sagemaker_session: SageMaker session (used to get the boto session).
 validate_role: If False, skips client-side permission/trust validation.
 role_arn: Keyword alias for ``provided_role``.

 Returns:
 IAM role ARN.
Expand All@@ -552,6 +608,7 @@ def resolve_and_validate_role(
 if role_type not in ROLE_TYPES:
 raise ValueError(f"Invalid role_type '{role_type}'. Must be one of: {ROLE_TYPES}")

 provided_role = provided_role or role_arn
 boto_session = _get_boto_session(sagemaker_session)
 iam_client = boto_session.client("iam")

Expand All@@ -567,6 +624,10 @@ def resolve_and_validate_role(
 if not role_arn:
 raise RoleValidationError(_build_validation_error_message(None, role_type))

 if not validate_role or os.getenv("SAGEMAKER_VALIDATE_ROLE", "true").lower() in ("false", "0", "no"):
 logger.info("Skipping IAM role validation for '%s' (%s).", role_arn, role_type)
 return role_arn

 # Permission check (definitive denial blocks; unverifiable warns).
 verdict, denied = _evaluate_permissions(iam_client, role_arn, role_type)
 if verdict is False:
Expand All@@ -584,7 +645,8 @@ def resolve_and_validate_role(
 if verdict is None:
 logger.warning(
 "Could not verify permissions for role '%s' (caller lacks "
 "iam:SimulatePrincipalPolicy). Proceeding with it. If the operation "
 "iam:SimulatePrincipalPolicy or conditional SCPs/permissions boundaries "
 "prevent evaluation). Proceeding with it. If the operation "
 "later fails with an access-denied error, ensure the role has the "
 "required permissions for '%s' (see "
 "IamRoleResolver().get_required_actions('%s')) or create a dedicated "
Expand DownExpand Up@@ -643,7 +705,7 @@ def verify_hyperpod_connect_permissions(
 return None

 try:
 denied = _simulate_denied_actions(
 denied, unverifiable = _simulate_denied_actions(
 iam_client, caller_role_arn, list(HYPERPOD_CLI_CONNECT_ACTIONS)
 )
 except ClientError as e:
Expand DownExpand Up@@ -671,6 +733,15 @@ def verify_hyperpod_connect_permissions(
 )
 return False

 if unverifiable:
 logger.info(
 "Cannot definitively verify HyperPod connect permissions for '%s' due to "
 "Organizations SCPs or permissions boundaries; the HyperPod CLI will "
 "validate access at submit time.",
 caller_role_arn,
 )
 return None

 logger.info(
 "Caller '%s' has the HyperPod CLI connect permissions.", caller_role_arn
 )
Expand DownExpand Up@@ -732,6 +803,22 @@ def __init__(self, sagemaker_session=None):
 self._sts_client = self._boto_session.client("sts")

 # -- public API ---------------------------------------------------------
 def resolve_and_validate_role(
 self,
 provided_role: Optional[str] = None,
 role_type: str = "training",
 *,
 role_arn: Optional[str] = None,
 validate_role: bool = True,
 ) -> str:
 """Resolve and validate an IAM role (read-only; does not mutate IAM)."""
 return resolve_and_validate_role(
 provided_role=provided_role or role_arn,
 role_type=role_type,
 sagemaker_session=self._sagemaker_session,
 validate_role=validate_role,
 )

 def get_required_actions(self, role_type: str) -> List[str]:
 """Return the IAM actions a role of ``role_type`` needs (read-only preview)."""
 self._validate_role_type(role_type)
Expand Down
Loading
Loading