aws / aws/agentcore-cli

Error in doc: https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/gateway-prerequisites-permissions.html

Open
#1,241 0 comments 0 reactions 0 assignees View on GitHub
bug documentation
Dominant language
TypeScript
Stars
283
Forks
95
Avg merge
1d 2h
Merged PRs (30d)
183

Description

### Description

What the docs say:
From the [Gateway permissions doc](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/gateway-prerequisites-permissions.html):
"If you plan to include a gateway target tool definition from an Amazon S3 URI, you'll need to include permissions for the gateway service role to access the bucket."

What is actually happening:
The caller's identity reads S3, NOT the Gateway service role.

This is an error in the doc.

Practical impact: grant `s3:GetObject` on the **deployment role / Terraform role**; no need to add it to the Gateway service role.

### Steps to Reproduce

Q3_GW_ROLE_NAME = "agentcore-q3-test-gw-role"
Q3_GW_NAME = "agentcore-q3-test-gw"

trust = {
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": {"Service": "bedrock-agentcore.amazonaws.com"},
"Action": "sts:AssumeRole",
}],
}
# Deliberately *no* s3:GetObject — only what the Gateway needs to fetch
# OAuth credentials and to write logs. If the service role were the S3
# reader, target creation against the OpenAPI spec would fail.
gw_policy = {
"Version": "2012-10-17",
"Statement": [
{"Effect": "Allow",
"Action": ["bedrock-agentcore:*", "secretsmanager:GetSecretValue",
"logs:CreateLogGroup", "logs:CreateLogStream",
"logs:PutLogEvents"],
"Resource": "*"},
],
}
try:
iam.create_role(RoleName=Q3_GW_ROLE_NAME,
AssumeRolePolicyDocument=json.dumps(trust))
except iam.exceptions.EntityAlreadyExistsException:
pass
iam.put_role_policy(RoleName=Q3_GW_ROLE_NAME,
PolicyName="gw-inline",
PolicyDocument=json.dumps(gw_policy))
Q3_GW_ROLE_ARN = iam.get_role(RoleName=Q3_GW_ROLE_NAME)["Role"]["Arn"]
print("Gateway service role:", Q3_GW_ROLE_ARN)
print("(intentionally has NO s3:GetObject)")

# Wait briefly for IAM eventual consistency before we hand the role to
# create_gateway — STS AssumeRole on a freshly created role can return
# AccessDenied for a few seconds.
time.sleep(10)

# Drop any prior test gateway with the same name.
# Both delete_gateway_target and delete_gateway are async — list_*
# stops returning the resource as soon as the delete is accepted, but
# create_gateway against the same name will still hit ConflictException
# until the backend's internal consistency check finishes. Poll the
# authoritative get_* APIs until ResourceNotFoundException.
def _drop_q3_gateway(gw_id: str, wait: int = 300) -> None:
"""Delete a gateway and wait until both its targets and the gateway
itself return ResourceNotFoundException — the only reliable signal
that name reservation has been released for re-create."""
tgt_ids: list[str] = []
try:
for tp in agentcore_ctl.get_paginator("list_gateway_targets").paginate(gatewayIdentifier=gw_id):
for t in tp.get("items", []):
tgt_ids.append(t["targetId"])
except Exception as e:
print(" warn list_gateway_targets:", e)
for tid in tgt_ids:
try:
agentcore_ctl.delete_gateway_target(gatewayIdentifier=gw_id, targetId=tid)
print(f" requested delete of target {tid}")
except Exception as e:
print(f" warn delete_gateway_target {tid}: {e}")
for tid in tgt_ids:
elapsed = 0
while elapsed < wait:
try:
agentcore_ctl.get_gateway_target(gatewayIdentifier=gw_id, targetId=tid)
except agentcore_ctl.exceptions.ResourceNotFoundException:
print(f" target {tid} fully deleted")
break
time.sleep(5)
elapsed += 5
else:
raise TimeoutError(f"target {tid} still present after {wait}s")
try:
agentcore_ctl.delete_gateway(gatewayIdentifier=gw_id)
except Exception as e:
print(" warn delete_gateway:", e)
return
elapsed = 0
while elapsed < wait:
try:
agentcore_ctl.get_gateway(gatewayIdentifier=gw_id)
except agentcore_ctl.exceptions.ResourceNotFoundException:
print(f" gateway {gw_id} fully deleted")
return
time.sleep(5)
elapsed += 5
raise TimeoutError(f"gateway {gw_id} still present after {wait}s")

for page in agentcore_ctl.get_paginator("list_gateways").paginate():
for gw in page.get("items", []):
if gw.get("name") == Q3_GW_NAME:
print(f"Found existing gateway '{Q3_GW_NAME}' ({gw['gatewayId']}); deleting before re-create")
_drop_q3_gateway(gw["gatewayId"])

# Use the SERVICE app for inbound auth on this Gateway — we only need a
# valid JWT to do tools/list later (data-plane re-fetch test). The
# allowedClients reflects only the service app since this isn't part of
# the Q4 multi-client test.
gw = agentcore_ctl.create_gateway(
name=Q3_GW_NAME,
roleArn=Q3_GW_ROLE_ARN,
protocolType="MCP",
authorizerType="CUSTOM_JWT",
authorizerConfiguration={
"customJWTAuthorizer": {
"discoveryUrl": OKTA["discovery_url"],
"allowedClients": [OKTA["service_client_id"]],
"allowedAudience": [OKTA["audience"]],
}
},
description="Q3 validation gateway",
)
Q3_GW_ID = gw["gatewayId"]
Q3_GW_URL = gw["gatewayUrl"]
print("Gateway:", Q3_GW_ID, "url:", Q3_GW_URL)

# create_gateway is also async. Poll until READY before any subsequent
# cell tries to attach a target — otherwise create_gateway_target races
# the gateway initialization and fails with a generic backend error.
deadline = time.time() + 180
while time.time() < deadline:
info = agentcore_ctl.get_gateway(gatewayIdentifier=Q3_GW_ID)
st = info.get("status")
if st == "READY":
print("Gateway READY.")
break
if st == "FAILED":
raise RuntimeError(f"gateway {Q3_GW_ID} entered FAILED: {info.get('statusReasons')}")
time.sleep(5)
else:
raise TimeoutError(f"gateway {Q3_GW_ID} not READY within timeout")

def _create_openapi_target(name: str, key: str) -> dict:
"""Wrapper around create_gateway_target with the OpenAPI-on-S3 schema.

Pulls the spec from ``s3://{Q3_BUCKET}/{key}``. The OpenAPI target
needs an outbound credential provider — we use the dummy
apiKeyCredentialProvider created in A2 (httpbin ignores the key).
The actual outbound call to the target API is irrelevant here:
the test surface is whether ``CreateGatewayTarget`` can read the
OpenAPI spec out of the private S3 bucket.
"""
return agentcore_ctl.create_gateway_target(
gatewayIdentifier=Q3_GW_ID,
name=name,
description=f"Q3 OpenAPI target for s3://{Q3_BUCKET}/{key}",
targetConfiguration={
"mcp": {
"openApiSchema": {
"s3": {"uri": f"s3://{Q3_BUCKET}/{key}"},
}
}
},
credentialProviderConfigurations=[
{
"credentialProviderType": "API_KEY",
"credentialProvider": {
"apiKeyCredentialProvider": {
"providerArn": Q3_APIKEY_PROVIDER_ARN,
"credentialLocation": "HEADER",
"credentialParameterName": "X-API-Key",
}
},
}
],
)

print("→ Test 1: create target while only the *caller* can read S3")
try:
t1 = _create_openapi_target("q3-test1-caller-can-read", Q3_KEY_T1)
Q3_TARGET_ID_1 = t1["targetId"]
print("✅ SUCCESS — targetId:", Q3_TARGET_ID_1)
print(" → Strongly suggests the CALLER identity reads S3 (Q3b confirmed)")
except ClientError as e:
print(f"❌ FAILED: {e.response['Error']['Code']}: {e.response['Error']['Message']}")
Q3_TARGET_ID_1 = None

# Trim sts:AssumedRole// back to iam::role//
# for use in IAM principal matching. The current caller is likely an
# assumed-role; bucket policy aws:PrincipalArn matches the *role* ARN,
# not the session ARN.
def _principal_arn(arn: str) -> str:
if ":assumed-role/" in arn:
# arn:aws:sts::ACCT:assumed-role/RoleName/Session -> arn:aws:iam::ACCT:role/RoleName
parts = arn.split(":")
role_name = parts[5].split("/")[1]
return f"arn:aws:iam::{ACCOUNT_ID}:role/{role_name}"
return arn

CALLER_PRINCIPAL = _principal_arn(CALLER_ARN)
print("Caller principal (for bucket policy):", CALLER_PRINCIPAL)

deny_caller_policy = {
"Version": "2012-10-17",
"Statement": [
{
"Sid": "DenyCallerOnly",
"Effect": "Deny",
"Principal": {"AWS": CALLER_PRINCIPAL},
"Action": "s3:GetObject",
"Resource": f"arn:aws:s3:::{Q3_BUCKET}/*",
},
],
}
s3.put_bucket_policy(Bucket=Q3_BUCKET, Policy=json.dumps(deny_caller_policy))
print("Applied bucket policy: DENY GetObject for caller principal")
# Allow propagation
time.sleep(10)

print("→ Test 2: create target while caller is explicitly DENIED")
try:
t2 = _create_openapi_target("q3-test2-caller-denied", Q3_KEY_T2)
Q3_TARGET_ID_2 = t2["targetId"]
print(f"❌ UNEXPECTED SUCCESS — targetId: {Q3_TARGET_ID_2}")
print(" → If S3 was actually read on this call, the GW SERVICE ROLE")
print(" must be the reader. Or AgentCore is using a privileged path.")
except ClientError as e:
print(f"✅ FAILED AS EXPECTED: {e.response['Error']['Code']}: "
f"{e.response['Error']['Message']}")
print(" → Confirms the CALLER identity reads S3")
Q3_TARGET_ID_2 = None

only_gw_role_policy = {
"Version": "2012-10-17",
"Statement": [
{
"Sid": "AllowGatewayServiceRoleOnly",
"Effect": "Allow",
"Principal": {"AWS": Q3_GW_ROLE_ARN},
"Action": "s3:GetObject",
"Resource": f"arn:aws:s3:::{Q3_BUCKET}/*",
},
{
"Sid": "DenyEveryoneElse",
"Effect": "Deny",
"NotPrincipal": {"AWS": [Q3_GW_ROLE_ARN]},
"Action": "s3:GetObject",
"Resource": f"arn:aws:s3:::{Q3_BUCKET}/*",
},
],
}
s3.put_bucket_policy(Bucket=Q3_BUCKET, Policy=json.dumps(only_gw_role_policy))
print("Applied bucket policy: only the Gateway service role may GetObject")
time.sleep(10)

# But wait — the GW role itself has no s3:GetObject in its inline policy.
# Bucket-policy Allow alone is insufficient for an IAM principal whose
# identity policy doesn't grant the action. To make this test ONLY about
# "is the GW role the reader", we *also* grant s3:GetObject on the role.
iam.put_role_policy(
RoleName=Q3_GW_ROLE_NAME,
PolicyName="q3-test3-temp-s3-allow",
PolicyDocument=json.dumps({
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow", "Action": "s3:GetObject",
"Resource": f"arn:aws:s3:::{Q3_BUCKET}/*",
}],
}),
)
print("Granted s3:GetObject on the Gateway service role (inline).")
time.sleep(10)

print("→ Test 3: create target while ONLY the GW service role can read S3")
try:
t3 = _create_openapi_target("q3-test3-only-role-can-read", Q3_KEY_T3)
Q3_TARGET_ID_3 = t3["targetId"]
print(f"⚠ SUCCESS — targetId: {Q3_TARGET_ID_3}")
print(" → This would mean the GW SERVICE ROLE is the reader.")
print(" (Doc-aligned interpretation, contradicts Test 2.)")
except ClientError as e:
print(f"✅ FAILED AS EXPECTED: {e.response['Error']['Code']}: "
f"{e.response['Error']['Message']}")
print(" → Confirms the GW service role is NOT used to read S3.")
print(" Combined with Test 2, this DEFINITIVELY proves the CALLER reads S3.")
Q3_TARGET_ID_3 = None

# Restore: clear bucket policy + remove the temp role policy so subsequent
# tests don't carry over.
s3.delete_bucket_policy(Bucket=Q3_BUCKET)
iam.delete_role_policy(RoleName=Q3_GW_ROLE_NAME, PolicyName="q3-test3-temp-s3-allow")
print("Reverted bucket policy + temp role policy.")

### Expected Behavior

"If you plan to include a gateway target tool definition from an Amazon S3 URI, you'll need to include permissions for the gateway service role to access the bucket."

### Actual Behavior

The caller's identity reads S3, NOT the Gateway service role.

### CLI Version

_No response_

### Operating System

macOS

### Additional Context

N/A

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.