Don't leak objects out of access policies when used in a computed glo…

…bal (#6926)

The caching for our computed globals failed to account for whether
access policies are in effect, which could allow leaking of objects
that should not have been visible. Fix this by augmenting the cache
key.

A downside of this fix is that sometimes now we will generate two
identical computations of a computed global. This is probably fine for
now, since this is an important bug to fix, and 2 is still a lot
better than the N that it was without the cache.

edb/edgeql/compiler/context.py

@@ -151,7 +151,10 @@ class Environment:
     path_scope: irast.ScopeTreeNode
     """Overrall expression path scope tree."""
 
-    schema_view_cache: Dict[s_types.Type, tuple[s_types.Type, irast.Set]]
+    schema_view_cache: Dict[
+        tuple[s_types.Type, object],
+        tuple[s_types.Type, irast.Set],
+    ]
     """Type cache used by schema-level views."""
 
     query_parameters: Dict[str, irast.Param]
@@ -764,6 +767,12 @@ def maybe_create_anchor(
         else:
             return ir
 
+    # Return an additional key for any compilation caches that may
+    # vary based on "security contexts" such as whether we are in an
+    # access policy.
+    def get_security_context(self) -> object:
+        return bool(self.suppress_rewrites)
+
 
 class CompilerContext(compiler.CompilerContext[ContextLevel]):
     ContextLevelClass = ContextLevel

edb/edgeql/compiler/stmtctx.py

@@ -746,7 +746,15 @@ def declare_view(
 def _declare_view_from_schema(
         viewcls: s_types.Type, *,
         ctx: context.ContextLevel) -> tuple[s_types.Type, irast.Set]:
-    e = ctx.env.schema_view_cache.get(viewcls)
+    # We need to include "security context" things (currently just
+    # access policy state) in the cache key, here.
+    #
+    # FIXME: Could we do better? Sometimes we might compute a single
+    # global twice now, as a result of this. It should be possible to
+    # make some decisions based on whether the alias actually does
+    # touch any access policies...
+    key = viewcls, ctx.get_security_context()
+    e = ctx.env.schema_view_cache.get(key)
     if e is not None:
         return e
 
@@ -771,7 +779,7 @@ def _declare_view_from_schema(
         vs = subctx.aliased_views[viewcls_name]
         assert vs is not None
         vc = setgen.get_set_type(vs, ctx=ctx)
-        ctx.env.schema_view_cache[viewcls] = vc, view_set
+        ctx.env.schema_view_cache[key] = vc, view_set
 
     return vc, view_set
 

tests/test_edgeql_policies.py

@@ -1330,3 +1330,23 @@ async def test_edgeql_policies_global_01(self):
             }
             ''' + clan_and_global
         )
+
+    async def test_edgeql_policies_global_02(self):
+        await self.con.execute('''
+            create type T {
+                create access policy ok allow all;
+                create access policy no deny select;
+            };
+            insert T;
+            create global foo := (select T limit 1);
+            create type S {
+                create access policy ok allow all using (exists global foo)
+            };
+        ''')
+
+        await self.assert_query_result(
+            r'''
+            select { s := S, foo := global foo };
+            ''',
+            [{"s": [], "foo": None}]
+        )