Description
What the docs say:
From the Gateway permissions doc:
"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
Description
What the docs say:
From the Gateway permissions doc:
"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:GetObjecton 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.
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