fix(gateway/fgac): filter semantic search results by scope + add search tests

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
 

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}")

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__":

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())