From 345ab7da43c1e1f192c49838e7048f7f5f1e7bc2 Mon Sep 17 00:00:00 2001 From: "rbelson@amazon.com" Date: Sun, 23 Aug 2026 18:53:41 -0400 Subject: [PATCH 1/2] fix(gateway/fgac): filter semantic search results by scope + add search tests The FGAC tutorial documents three access-control patterns, including pattern 2: 'Semantic search with FGAC (RESPONSE interceptor) - filter search results so users only see tools they have access to'. But the RESPONSE interceptor only filtered tools/list shapes (result.tools / result.structuredContent.tools), and the demo (invoke.py) plus the README Test Cases table exercised only tools/call and tools/list - no semantic-search coverage, despite the README promising it. Changes: - RESPONSE interceptor Lambda (fgac-interceptors-stack.yaml): also filter the semantic-search response, whichever shape the gateway emits - result.tools, result.structuredContent.tools, and a JSON string in result.content[*].text carrying {"tools": [...]}. Non-JSON text content passes through untouched. So x_amz_bedrock_agentcore_search results are now scope-filtered like tools/list. - invoke.py: add Test 8 (search with getOrder scope -> only getOrder) and Test 9 (search with full scope -> multiple tools), plus an extract_search_tools() helper tolerant of all three response shapes. - README: add rows 8 and 9 to the Test Cases table so the documented tests match the demo and the promised pattern 2. --- .../fine-grain-access-control/README.md | 2 + .../fgac-interceptors-stack.yaml | 55 +++++++++++---- .../fine-grain-access-control/invoke.py | 67 ++++++++++++++++++- 3 files changed, 110 insertions(+), 14 deletions(-) diff --git a/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/04-advanced-concepts/fine-grain-access-control/README.md b/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/04-advanced-concepts/fine-grain-access-control/README.md index f91b0eeac..f65f50e78 100644 --- a/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/04-advanced-concepts/fine-grain-access-control/README.md +++ b/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/04-advanced-concepts/fine-grain-access-control/README.md @@ -219,6 +219,8 @@ uv run python scripts/fine-grain-access-control/invoke.py | 5 | `fgac/fgac-mcp-target` (full) | all tools | ALLOW all | | 6 | `fgac/fgac-mcp-target:getOrder` | `tools/list` | Only `getOrder` visible | | 7 | `fgac/fgac-mcp-target` (full) | `tools/list` | All 4 tools visible | +| 8 | `fgac/fgac-mcp-target:getOrder` | semantic search | Only `getOrder` in results | +| 9 | `fgac/fgac-mcp-target` (full) | semantic search | All relevant tools in results | ## Cleanup diff --git a/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/gatewaylabproject/cloudformation/fine-grain-access-control/fgac-interceptors-stack.yaml b/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/gatewaylabproject/cloudformation/fine-grain-access-control/fgac-interceptors-stack.yaml index 3492d0147..295bf6a35 100644 --- a/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/gatewaylabproject/cloudformation/fine-grain-access-control/fgac-interceptors-stack.yaml +++ b/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/gatewaylabproject/cloudformation/fine-grain-access-control/fgac-interceptors-stack.yaml @@ -209,21 +209,50 @@ Resources: scopes = claims.get("scope", "") result = body.get("result", {}) - tools = result.get("tools", []) - if not tools: - tools = result.get("structuredContent", {}).get("tools", []) + # tools/list responses expose the tools directly on the result + # (or under structuredContent). Semantic search + # (x_amz_bedrock_agentcore_search) returns its matches the same + # way, but some gateway versions also embed the tool list as a + # JSON string in result.content[0].text. Filter every shape we + # find so users only ever see tools their scopes allow — + # whether they discovered them via tools/list or search. + filtered_body = body.copy() + + # Shape 1: result.tools (plain tools/list) + tools = result.get("tools", []) if tools: - filtered_tools = filter_tools_by_scope(tools, scopes) - print(f"[FGAC] Filtered {len(tools)} -> {len(filtered_tools)} tools") - - filtered_body = body.copy() - if "result" in filtered_body: - if "structuredContent" in filtered_body["result"]: - filtered_body["result"]["structuredContent"]["tools"] = filtered_tools - else: - filtered_body["result"]["tools"] = filtered_tools - body = filtered_body + filtered = filter_tools_by_scope(tools, scopes) + print(f"[FGAC] Filtered result.tools {len(tools)} -> {len(filtered)}") + filtered_body["result"]["tools"] = filtered + + # Shape 2: result.structuredContent.tools (tools/list + search) + sc_tools = result.get("structuredContent", {}).get("tools", []) + if sc_tools: + filtered = filter_tools_by_scope(sc_tools, scopes) + print(f"[FGAC] Filtered structuredContent.tools {len(sc_tools)} -> {len(filtered)}") + filtered_body["result"]["structuredContent"]["tools"] = filtered + + # Shape 3: result.content[*].text is a JSON string carrying + # {"tools": [...]} — the semantic-search payload shape. Parse, + # filter, and re-serialize; leave non-JSON text untouched. + content = result.get("content", []) + if isinstance(content, list): + for item in content: + if not isinstance(item, dict) or item.get("type") != "text": + continue + text = item.get("text", "") + try: + payload = json.loads(text) + except (ValueError, TypeError): + continue + if isinstance(payload, dict) and isinstance(payload.get("tools"), list): + before = len(payload["tools"]) + payload["tools"] = filter_tools_by_scope(payload["tools"], scopes) + print(f"[FGAC] Filtered search content.tools {before} -> {len(payload['tools'])}") + item["text"] = json.dumps(payload) + + body = filtered_body except Exception as e: print(f"[FGAC] Response filter error: {e}") diff --git a/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/gatewaylabproject/scripts/fine-grain-access-control/invoke.py b/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/gatewaylabproject/scripts/fine-grain-access-control/invoke.py index bcafbffaf..980e0b43f 100644 --- a/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/gatewaylabproject/scripts/fine-grain-access-control/invoke.py +++ b/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/gatewaylabproject/scripts/fine-grain-access-control/invoke.py @@ -20,6 +20,29 @@ from gateway_mcp_client import GatewayMCPClient +def extract_search_tools(result): + """Return the tool list from a semantic-search response, tolerating the + shapes the gateway can return: result.structuredContent.tools, result.tools, + or a JSON string in result.content[*].text carrying {"tools": [...]}.""" + import json + + res = result.get("result", {}) if isinstance(result, dict) else {} + tools = res.get("structuredContent", {}).get("tools") + if tools: + return tools + if res.get("tools"): + return res["tools"] + for item in res.get("content", []) or []: + if isinstance(item, dict) and item.get("type") == "text": + try: + payload = json.loads(item.get("text", "")) + except (ValueError, TypeError): + continue + if isinstance(payload, dict) and isinstance(payload.get("tools"), list): + return payload["tools"] + return [] + + def load_env(): env_path = os.path.join(os.path.dirname(__file__), ".env") if os.path.exists(env_path): @@ -177,11 +200,53 @@ def main(): else: print(f" FAIL: Expected 4 tools, got {len(tool_names)}") + # --- Test 8: semantic search with limited scope → filtered --- + print("\n" + "=" * 60) + print("Test 8: semantic search with getOrder scope (SHOULD SHOW ONLY getOrder)") + print("=" * 60) + + scope = "fgac/fgac-mcp-target:getOrder" + token = get_token(token_endpoint, fgac_client_id, fgac_client_secret, scope) + mcp = GatewayMCPClient(gateway_url, lambda: token, protocol_version="2025-11-25") + + result = mcp.call_tool( + "x_amz_bedrock_agentcore_search", + {"query": "tools to look up or read an order"}, + ) + search_tools = extract_search_tools(result) + tool_names = [t["name"] for t in search_tools if "___" in t.get("name", "")] + print(f" Tools returned: {tool_names}") + if tool_names and all("getOrder" in n for n in tool_names): + print(" PASS: Only getOrder-scoped tools returned by search") + else: + print(f" FAIL: Expected only getOrder, got {tool_names}") + + # --- Test 9: semantic search with full scope → all relevant --- + print("\n" + "=" * 60) + print("Test 9: semantic search with full access scope (SHOULD SHOW ALL RELEVANT)") + print("=" * 60) + + scope = "fgac/fgac-mcp-target" + token = get_token(token_endpoint, fgac_client_id, fgac_client_secret, scope) + mcp = GatewayMCPClient(gateway_url, lambda: token, protocol_version="2025-11-25") + + result = mcp.call_tool( + "x_amz_bedrock_agentcore_search", + {"query": "tools to manage an order"}, + ) + search_tools = extract_search_tools(result) + tool_names = [t["name"] for t in search_tools if "___" in t.get("name", "")] + print(f" Tools returned: {tool_names}") + if len(tool_names) > 1: + print(f" PASS: Full scope surfaces multiple tools ({len(tool_names)})") + else: + print(f" FAIL: Expected multiple tools with full scope, got {tool_names}") + print("\n" + "=" * 60) print("Summary") print("=" * 60) print(" REQUEST interceptor: blocks unauthorized tool/call") - print(" RESPONSE interceptor: filters tools/list by scope") + print(" RESPONSE interceptor: filters tools/list AND semantic search by scope") if __name__ == "__main__": From 3bcad71a1da767d5be157628dc19a2bdcc541951 Mon Sep 17 00:00:00 2001 From: "rbelson@amazon.com" Date: Sun, 23 Aug 2026 18:59:28 -0400 Subject: [PATCH 2/2] test(gateway/fgac): add offline unit test for RESPONSE interceptor filtering MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extracts the inline RESPONSE-interceptor Lambda from the CloudFormation template and exercises lambda_handler against synthetic gateway events — no AWS, no live gateway, no network. Asserts scope-based filtering across all three response shapes (result.tools, structuredContent.tools, and the semantic-search content[*].text JSON payload): a limited scope keeps only the authorized tool, a full scope keeps all, non-JSON text is untouched, and a missing token fails safe. 5/5 pass. --- .../test_response_interceptor.py | 160 ++++++++++++++++++ 1 file changed, 160 insertions(+) create mode 100644 01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/gatewaylabproject/scripts/fine-grain-access-control/test_response_interceptor.py diff --git a/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/gatewaylabproject/scripts/fine-grain-access-control/test_response_interceptor.py b/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/gatewaylabproject/scripts/fine-grain-access-control/test_response_interceptor.py new file mode 100644 index 000000000..a4d544263 --- /dev/null +++ b/01-features/07-centralize-and-govern-your-ai-infrastructure/01-gateway/gatewaylabproject/scripts/fine-grain-access-control/test_response_interceptor.py @@ -0,0 +1,160 @@ +"""Offline unit tests for the FGAC RESPONSE interceptor filtering logic. + +Extracts the inline RESPONSE-interceptor Lambda from the CloudFormation +template and exercises its ``lambda_handler`` against synthetic gateway +events — no AWS, no live gateway, no network. Verifies that scope-based +filtering works across every response shape the gateway can emit: + + 1. ``result.tools`` (plain tools/list) + 2. ``result.structuredContent.tools`` (tools/list + search) + 3. ``result.content[*].text`` JSON payload (semantic search) + +Run: + uv run python scripts/fine-grain-access-control/test_response_interceptor.py + # or: python3 scripts/fine-grain-access-control/test_response_interceptor.py +""" + +import base64 +import json +import os +import sys +import types + +REPO_YAML = os.path.join( + os.path.dirname(__file__), + "..", + "..", + "cloudformation", + "fine-grain-access-control", + "fgac-interceptors-stack.yaml", +) + + +def _extract_zipfile_blocks(path): + """Pull each ``ZipFile: |`` literal block out of the CFN template by + indentation (avoids a hard PyYAML dependency).""" + lines = open(path).read().splitlines() + blocks, i = [], 0 + while i < len(lines): + if lines[i].strip() == "ZipFile: |": + code_indent = (len(lines[i]) - len(lines[i].lstrip())) + 2 + i += 1 + body = [] + while i < len(lines): + ln = lines[i] + if ln.strip() == "": + body.append("") + i += 1 + continue + if (len(ln) - len(ln.lstrip())) < code_indent: + break + body.append(ln[code_indent:]) + i += 1 + blocks.append("\n".join(body)) + else: + i += 1 + return blocks + + +def _load_response_interceptor(): + blocks = _extract_zipfile_blocks(REPO_YAML) + assert len(blocks) == 2, f"expected 2 inline Lambdas, found {len(blocks)}" + os.environ["GATEWAY_TARGET_NAME"] = "fgac-mcp-target" + mod = types.ModuleType("fgac_response_interceptor") + exec(compile(blocks[1], "response_interceptor.py", "exec"), mod.__dict__) + return mod + + +def _make_jwt(scope): + header = base64.urlsafe_b64encode(b'{"alg":"none"}').decode().rstrip("=") + payload = ( + base64.urlsafe_b64encode(json.dumps({"scope": scope}).encode()) + .decode() + .rstrip("=") + ) + return f"{header}.{payload}.sig" + + +TOOLS = [ + {"name": "fgac-mcp-target___getOrder", "description": "get"}, + {"name": "fgac-mcp-target___updateOrder", "description": "upd"}, + {"name": "fgac-mcp-target___cancelOrder", "description": "cxl"}, + {"name": "fgac-mcp-target___deleteOrder", "description": "del"}, +] + +LIMITED = _make_jwt("fgac/fgac-mcp-target:getOrder") +FULL = _make_jwt("fgac/fgac-mcp-target") + + +def _event(result_body, jwt): + return { + "mcp": { + "gatewayRequest": {"headers": {"Authorization": f"Bearer {jwt}"}}, + "gatewayResponse": { + "headers": {}, + "body": {"jsonrpc": "2.0", "id": 1, "result": result_body}, + }, + } + } + + +def _out(resp): + return resp["mcp"]["transformedGatewayResponse"]["body"]["result"] + + +def _short(names): + return sorted(n.split("___")[-1] for n in names) + + +def main(): + mod = _load_response_interceptor() + handler = mod.lambda_handler + passed = failed = 0 + + def check(label, got, expect): + nonlocal passed, failed + ok = sorted(got) == sorted(expect) + print((" PASS " if ok else " FAIL ") + label + f" -> {got}" + + ("" if ok else f" (expected {expect})")) + passed += int(ok) + failed += int(not ok) + + # Shape 1: result.tools, limited scope -> only getOrder + r = handler(_event({"tools": [dict(t) for t in TOOLS]}, LIMITED), None) + check("result.tools (limited)", + _short(t["name"] for t in _out(r)["tools"]), ["getOrder"]) + + # Shape 2: structuredContent.tools, limited scope -> only getOrder + r = handler(_event({"structuredContent": {"tools": [dict(t) for t in TOOLS]}}, LIMITED), None) + check("structuredContent.tools (limited)", + _short(t["name"] for t in _out(r)["structuredContent"]["tools"]), ["getOrder"]) + + # Shape 3: semantic-search content[0].text JSON, limited scope -> only getOrder + search_payload = json.dumps({"tools": [dict(t) for t in TOOLS]}) + r = handler(_event({"content": [{"type": "text", "text": search_payload}]}, LIMITED), None) + parsed = json.loads(_out(r)["content"][0]["text"]) + check("search content.text (limited)", + _short(t["name"] for t in parsed["tools"]), ["getOrder"]) + + # Shape 3 with full scope -> all four survive (no over-filtering) + r = handler(_event({"content": [{"type": "text", "text": search_payload}]}, FULL), None) + parsed = json.loads(_out(r)["content"][0]["text"]) + check("search content.text (full)", + _short(t["name"] for t in parsed["tools"]), + ["getOrder", "updateOrder", "cancelOrder", "deleteOrder"]) + + # Non-JSON text content is left untouched + r = handler(_event({"content": [{"type": "text", "text": "human readable, not json"}]}, LIMITED), None) + check("non-JSON text untouched", + [_out(r)["content"][0]["text"]], ["human readable, not json"]) + + # Missing token fails safe (no crash, no filtering exception surfaced) + handler({"mcp": {"gatewayRequest": {"headers": {}}, + "gatewayResponse": {"headers": {}, "body": {"result": {"tools": []}}}}}, None) + + print(f"\nRESULT: {passed} passed, {failed} failed") + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main())