Repository: geldata/gel
Pull request: Report failing access policy
Closes #4095


- refactor clauses.py
- report names of failed access policies

I have tests to do, but would ask for a quick peek if the major changes are ok.


Problem statement:
Add custom error message on object level security constraints
Currently, when a access policy constraint is violated the error looks like this:
`edgedb error: EdgeDBError: access policy violation on insert of default::User`

It does not carry any info about what constraint has been violated (it can have more than 1 insert constraint for eg).

Would be great to be able to set a custom error message like we do on type/field level constraints.

Hints from issue discussion:
There's some design work needed to figure out exactly what error message should be output. 
An access policy is violated if one or two things happens:
1. Every allow rule is false
2. Some reject rule is true

If a reject rule is false, we should output the error message for it. The main question, then, is what to do for allow.
I propose that on errors we include the text of *every* relevant error message, which is to say:
1. For every allow rule if they are all false, otherwise none of them
2. The error message for each reject rule that is true
Then we'll omit mentioning any rule without a message specified.
From an implementation perspective, we'll need to keep the policy expressions separate in the IR for DML, and track the error messages we want. Currently we combine them all together first, here: https://github.com/edgedb/edgedb/blob/ccbff2bd328fd3738619c000729685d5e4bfdf56/edb/edgeql/compiler/policies.py#L121-L177

Commit message: add errmessage to AccessPolicy

Patch to review:
diff --git a/edb/edgeql/compiler/clauses.py b/edb/edgeql/compiler/clauses.py
index 871d003f204..d5dd46eb0b4 100644
--- a/edb/edgeql/compiler/clauses.py
+++ b/edb/edgeql/compiler/clauses.py
@@ -40,12 +40,11 @@
 
 
 def compile_where_clause(
-        ir_stmt: irast.FilteredStmt,
-        where: Optional[qlast.Base], *,
-        ctx: context.ContextLevel) -> None:
+    where: Optional[qlast.Base], *, ctx: context.ContextLevel
+) -> Optional[irast.Set]:
 
     if where is None:
-        return
+        return None
 
     if ctx.partial_path_prefix:
         pathctx.register_set_in_scope(ctx.partial_path_prefix, ctx=ctx)
@@ -57,7 +56,7 @@ def compile_where_clause(
         bool_t = ctx.env.get_track_schema_type(sn.QualName('std', 'bool'))
         ir_set = setgen.scoped_set(ir_expr, typehint=bool_t, ctx=subctx)
 
-    ir_stmt.where = ir_set
+    return ir_set
 
 
 def compile_orderby_clause(
diff --git a/edb/edgeql/compiler/inference/cardinality.py b/edb/edgeql/compiler/inference/cardinality.py
index f91ad315350..52c4e46b3fa 100644
--- a/edb/edgeql/compiler/inference/cardinality.py
+++ b/edb/edgeql/compiler/inference/cardinality.py
@@ -1098,11 +1098,16 @@ def _infer_dml_check_cardinality(
     ctx: inference_context.InfCtx,
 ) -> None:
     pctx = ctx._replace(singletons=ctx.singletons | {ir.result.path_id})
-    for pol in [
-        *ir.read_policy_exprs.values(), *ir.write_policy_exprs.values()
-    ]:
-        pol.cardinality = infer_cardinality(
-            pol.expr, scope_tree=scope_tree, ctx=pctx)
+    for read_pol in ir.read_policies.values():
+        read_pol.cardinality = infer_cardinality(
+            read_pol.expr, scope_tree=scope_tree, ctx=pctx
+        )
+
+    for write_pol in ir.write_policies.values():
+        for p in write_pol.policies:
+            p.cardinality = infer_cardinality(
+                p.expr, scope_tree=scope_tree, ctx=pctx
+            )
 
     if ir.conflict_checks:
         for on_conflict in ir.conflict_checks:
@@ -1258,8 +1263,10 @@ def __infer_insert_stmt(
     # ... except if UNLESS CONFLICT is used
     else:
         return _infer_on_conflict_cardinality(
-            ir.on_conflict, type_has_rewrites=bool(ir.write_policy_exprs),
-            scope_tree=scope_tree, ctx=ctx,
+            ir.on_conflict,
+            type_has_rewrites=bool(ir.write_policies),
+            scope_tree=scope_tree,
+            ctx=ctx,
         )
 
 
diff --git a/edb/edgeql/compiler/inference/multiplicity.py b/edb/edgeql/compiler/inference/multiplicity.py
index 60e85225635..a2640f057a1 100644
--- a/edb/edgeql/compiler/inference/multiplicity.py
+++ b/edb/edgeql/compiler/inference/multiplicity.py
@@ -683,11 +683,12 @@ def _infer_mutating_stmt(
         for clause in ir.conflict_checks:
             _infer_on_conflict_clause(clause, scope_tree=scope_tree, ctx=ctx)
 
-    for policy in ir.write_policy_exprs.values():
-        infer_multiplicity(policy.expr, scope_tree=scope_tree, ctx=ctx)
+    for write_pol in ir.write_policies.values():
+        for policy in write_pol.policies:
+            infer_multiplicity(policy.expr, scope_tree=scope_tree, ctx=ctx)
 
-    for policy in ir.read_policy_exprs.values():
-        infer_multiplicity(policy.expr, scope_tree=scope_tree, ctx=ctx)
+    for read_pol in ir.read_policies.values():
+        infer_multiplicity(read_pol.expr, scope_tree=scope_tree, ctx=ctx)
 
 
 def _infer_on_conflict_clause(
diff --git a/edb/edgeql/compiler/policies.py b/edb/edgeql/compiler/policies.py
index 1a6555ec482..49a03faf2fb 100644
--- a/edb/edgeql/compiler/policies.py
+++ b/edb/edgeql/compiler/policies.py
@@ -39,6 +39,8 @@
 from . import dispatch
 from . import setgen
 
+T = TypeVar("T")
+
 
 def get_access_policies(
     stype: s_objtypes.ObjectType, *, ctx: context.ContextLevel,
@@ -79,7 +81,6 @@ def has_own_policies(
 
 
 def compile_pol(
-    stype: s_objtypes.ObjectType,
     pol: s_policies.AccessPolicy, *,
     ctx: context.ContextLevel,
 ) -> irast.Set:
@@ -99,6 +100,7 @@ def compile_pol(
         expr = expr_field.qlast
     else:
         expr = qlast.BooleanConstant(value='true')
+
     if condition := pol.get_condition(schema):
         expr = qlast.BinOp(op='AND', left=condition.qlast, right=expr)
 
@@ -125,6 +127,8 @@ def get_rewrite_filter(
 ) -> Optional[qlast.Expr]:
     schema = ctx.env.schema
     pols = get_access_policies(stype, ctx=ctx)
+    if not pols:
+        return None
 
     ctx.anchors = ctx.anchors.copy()
 
@@ -133,7 +137,7 @@ def get_rewrite_filter(
         if mode not in pol.get_access_kinds(schema):
             continue
 
-        ir_set = compile_pol(stype, pol, ctx=ctx)
+        ir_set = compile_pol(pol, ctx=ctx)
         expr = ctx.create_anchor(ir_set)
 
         is_allow = pol.get_action(schema) == qltypes.AccessPolicyAction.Allow
@@ -142,9 +146,7 @@ def get_rewrite_filter(
         else:
             deny.append(expr)
 
-    if not pols:
-        filter_expr = None
-    elif allow:
+    if allow:
         filter_expr = astutils.extend_binop(None, *allow, op='OR')
     else:
         filter_expr = qlast.BooleanConstant(value='false')
@@ -161,7 +163,7 @@ def get_rewrite_filter(
     # from bogusly optimizing away the entire type CTE if it can prove
     # it empty (which could then result in assert_exists on links to
     # the type not always firing).
-    if filter_expr and mode == qltypes.AccessKind.Select:
+    if mode == qltypes.AccessKind.Select:
         bogus_check = qlast.BinOp(
             op='?=',
             left=qlast.Path(partial=True, steps=[qlast.Ptr(
@@ -271,8 +273,7 @@ def try_type_rewrite(
                 subctx.partial_path_prefix = base_set
                 subctx.path_scope = subctx.env.path_scope.root.attach_fence()
 
-                clauses.compile_where_clause(
-                    filtered_stmt,
+                filtered_stmt.where = clauses.compile_where_clause(
                     get_rewrite_filter(
                         stype, mode=qltypes.AccessKind.Select, ctx=subctx),
                     ctx=subctx)
@@ -314,36 +315,87 @@ def try_type_rewrite(
     type_rewrites[rw_key] = filtered_set
 
 
-def compile_dml_policy(
+def compile_dml_write_policies(
     stype: s_objtypes.ObjectType,
     result: irast.Set,
     mode: qltypes.AccessKind, *,
     ctx: context.ContextLevel,
-) -> Optional[irast.PolicyExpr]:
-    """Compile a policy filter for a DML statement at a particular type"""
+) -> Optional[irast.WritePolicies]:
+    """Compile policy filters and wrap them into irast.WritePolicies"""
     if not ctx.env.type_rewrites.get((stype, False)):
         return None
 
-    pols = get_access_policies(stype, ctx=ctx)
-    if not pols:
+    with ctx.detached() as _, ctx.newscope(fenced=True) as subctx:
+        # TODO: can we make sure to always avoid generating needless
+        # select filters
+        _prepare_dml_policy_context(stype, result, ctx=subctx)
+
+        schema = subctx.env.schema
+        subctx.anchors = subctx.anchors.copy()
+
+        pols = get_access_policies(stype, ctx=ctx)
+        if not pols:
+            return None
+
+        policies = []
+        for pol in pols:
+            if mode not in pol.get_access_kinds(schema):
+                continue
+
+            ir_set = compile_pol(pol, ctx=subctx)
+
+            action = pol.get_action(schema)
+            name = str(pol.get_shortname(schema))
+
+            policies.append(
+                irast.WritePolicy(
+                    expr=ir_set,
+                    action=action,
+                    name=name,
+                    error_msg=pol.get_errmessage(schema),
+                )
+            )
+
+        return irast.WritePolicies(policies=policies)
+
+
+def compile_dml_read_policies(
+    stype: s_objtypes.ObjectType,
+    result: irast.Set,
+    mode: qltypes.AccessKind,
+    *,
+    ctx: context.ContextLevel,
+) -> Optional[irast.ReadPolicyExpr]:
+    """Compile a policy filter for a DML statement at a particular type"""
+    if not ctx.env.type_rewrites.get((stype, False)):
         return None
 
     with ctx.detached() as _, ctx.newscope(fenced=True) as subctx:
         # TODO: can we make sure to always avoid generating needless
         # select filters
-        skip_subtypes = (stype, False) not in ctx.env.type_rewrites
-        result = setgen.class_set(
-            stype, path_id=result.path_id, skip_subtypes=skip_subtypes,
-            ctx=ctx)
-
-        subctx.anchors[qlast.Subject().name] = result
-        subctx.partial_path_prefix = result
+        _prepare_dml_policy_context(stype, result, ctx=subctx)
 
         condition = get_rewrite_filter(stype, mode=mode, ctx=subctx)
-        assert condition
+        if not condition:
+            return None
 
-        return irast.PolicyExpr(
+        return irast.ReadPolicyExpr(
             expr=setgen.scoped_set(
                 dispatch.compile(condition, ctx=subctx), ctx=subctx
             ),
         )
+
+
+def _prepare_dml_policy_context(
+    stype: s_objtypes.ObjectType,
+    result: irast.Set,
+    *,
+    ctx: context.ContextLevel,
+) -> None:
+    skip_subtypes = (stype, False) not in ctx.env.type_rewrites
+    result = setgen.class_set(
+        stype, path_id=result.path_id, skip_subtypes=skip_subtypes, ctx=ctx
+    )
+
+    ctx.anchors[qlast.Subject().name] = result
+    ctx.partial_path_prefix = result
diff --git a/edb/edgeql/compiler/stmt.py b/edb/edgeql/compiler/stmt.py
index 230bd5fa310..f42ff4ca481 100644
--- a/edb/edgeql/compiler/stmt.py
+++ b/edb/edgeql/compiler/stmt.py
@@ -133,8 +133,7 @@ def compile_SelectQuery(
             forward_rptr=forward_rptr,
             ctx=sctx)
 
-        clauses.compile_where_clause(
-            stmt, expr.where, ctx=sctx)
+        stmt.where = clauses.compile_where_clause(expr.where, ctx=sctx)
 
         stmt.orderby = clauses.compile_orderby_clause(
             expr.orderby, ctx=sctx)
@@ -391,8 +390,7 @@ def compile_InternalGroupQuery(
                 result_alias=expr.result_alias,
                 ctx=bctx)
 
-            clauses.compile_where_clause(
-                stmt, expr.where, ctx=bctx)
+            stmt.where = clauses.compile_where_clause(expr.where, ctx=bctx)
 
             stmt.orderby = clauses.compile_orderby_clause(
                 expr.orderby, ctx=bctx)
@@ -512,10 +510,10 @@ def compile_InsertQuery(
                 ctx=resultctx,
             )
 
-        if pol_condition := policies.compile_dml_policy(
+        if pol_condition := policies.compile_dml_write_policies(
             mat_stype, result, mode=qltypes.AccessKind.Insert, ctx=ctx
         ):
-            stmt.write_policy_exprs[mat_stype.id] = pol_condition
+            stmt.write_policies[mat_stype.id] = pol_condition
 
         result = fini_stmt(stmt, expr, ctx=ictx, parent_ctx=ctx)
 
@@ -600,8 +598,7 @@ def compile_UpdateQuery(
 
         ictx.partial_path_prefix = subject
 
-        clauses.compile_where_clause(
-            stmt, expr.where, ctx=ictx)
+        stmt.where = clauses.compile_where_clause(expr.where, ctx=ictx)
 
         with ictx.new() as bodyctx:
             bodyctx.class_view_overrides = ictx.class_view_overrides.copy()
@@ -634,14 +631,14 @@ def compile_UpdateQuery(
             )
 
         for dtype in schemactx.get_all_concrete(mat_stype, ctx=ctx):
-            if pol_cond := policies.compile_dml_policy(
+            if read_pol := policies.compile_dml_read_policies(
                 dtype, result, mode=qltypes.AccessKind.UpdateRead, ctx=ctx
             ):
-                stmt.read_policy_exprs[dtype.id] = pol_cond
-            if pol_cond := policies.compile_dml_policy(
+                stmt.read_policies[dtype.id] = read_pol
+            if write_pol := policies.compile_dml_write_policies(
                 dtype, result, mode=qltypes.AccessKind.UpdateWrite, ctx=ctx
             ):
-                stmt.write_policy_exprs[dtype.id] = pol_cond
+                stmt.write_policies[dtype.id] = write_pol
 
         stmt.conflict_checks = conflicts.compile_inheritance_conflict_checks(
             stmt, mat_stype, ctx=ictx)
@@ -754,10 +751,10 @@ def compile_DeleteQuery(
             )
 
         for dtype in schemactx.get_all_concrete(mat_stype, ctx=ctx):
-            if pol_cond := policies.compile_dml_policy(
+            if pol_cond := policies.compile_dml_read_policies(
                 dtype, result, mode=qltypes.AccessKind.Delete, ctx=ctx
             ):
-                stmt.read_policy_exprs[dtype.id] = pol_cond
+                stmt.read_policies[dtype.id] = pol_cond
 
         result = fini_stmt(stmt, expr, ctx=ictx, parent_ctx=ctx)
 
diff --git a/edb/ir/ast.py b/edb/ir/ast.py
index fad57bb9272..07a273c1184 100644
--- a/edb/ir/ast.py
+++ b/edb/ir/ast.py
@@ -1033,11 +1033,13 @@ class MutatingStmt(Stmt):
     # for.
     conflict_checks: typing.Optional[typing.List[OnConflictClause]] = None
     # Access policy checks that we should raise errors on
-    write_policy_exprs: typing.Dict[
-        uuid.UUID, PolicyExpr] = ast.field(factory=dict)
+    write_policies: typing.Dict[uuid.UUID, WritePolicies] = ast.field(
+        factory=dict
+    )
     # Access policy checks that we should filter on
-    read_policy_exprs: typing.Dict[
-        uuid.UUID, PolicyExpr] = ast.field(factory=dict)
+    read_policies: typing.Dict[uuid.UUID, ReadPolicyExpr] = ast.field(
+        factory=dict
+    )
 
     @property
     def material_type(self) -> TypeRef:
@@ -1048,8 +1050,21 @@ def material_type(self) -> TypeRef:
         raise NotImplementedError
 
 
-class PolicyExpr(Base):
+class ReadPolicyExpr(Base):
+    expr: Set
+    cardinality: qltypes.Cardinality = qltypes.Cardinality.UNKNOWN
+
+
+class WritePolicies(Base):
+    policies: typing.List[WritePolicy]
+
+
+class WritePolicy(Base):
     expr: Set
+    action: qltypes.AccessPolicyAction
+    name: str
+    error_msg: typing.Optional[str]
+
     cardinality: qltypes.Cardinality = qltypes.Cardinality.UNKNOWN
 
 
diff --git a/edb/lib/schema.edgeql b/edb/lib/schema.edgeql
index e1bd503bfd4..2a1355d449b 100644
--- a/edb/lib/schema.edgeql
+++ b/edb/lib/schema.edgeql
@@ -409,6 +409,7 @@ ALTER TYPE schema::AccessPolicy {
   CREATE PROPERTY condition -> std::str;
   CREATE REQUIRED PROPERTY action -> schema::AccessPolicyAction;
   CREATE PROPERTY expr -> std::str;
+  CREATE PROPERTY errmessage -> std::str;
 };
 
 
diff --git a/edb/pgsql/compiler/astutils.py b/edb/pgsql/compiler/astutils.py
index 15af6760b91..d09fb459565 100644
--- a/edb/pgsql/compiler/astutils.py
+++ b/edb/pgsql/compiler/astutils.py
@@ -198,6 +198,81 @@ def extend_binop(
     return result
 
 
+def extend_concat(
+    expr: str | pgast.BaseExpr, *exprs: str | pgast.BaseExpr
+) -> pgast.BaseExpr:
+    return extend_binop(
+        pgast.StringConstant(val=expr) if isinstance(expr, str) else expr,
+        *[
+            pgast.StringConstant(val=e) if isinstance(e, str) else e
+            for e in exprs
+        ],
+        op='||',
+    )
+
+
+def extend_select_op(
+    stmt: Optional[pgast.SelectStmt],
+    *stmts: pgast.SelectStmt,
+    op: str = 'UNION',
+) -> Optional[pgast.SelectStmt]:
+    stmt_list = list(stmts)
+    result: pgast.SelectStmt
+
+    if stmt is None:
+        if len(stmt_list) == 0:
+            return None
+        result = stmt_list.pop(0)
+    else:
+        result = stmt
+
+    for s in stmt_list:
+        if s is not None and s is not result:
+            result = pgast.SelectStmt(larg=result, op=op, rarg=s)
+
+    return result
+
+
+def conditional_string_agg(
+    pairs: Sequence[Tuple[pgast.BaseExpr, pgast.BaseExpr]],
+) -> Optional[pgast.BaseExpr]:
+
+    selects = [
+        pgast.SelectStmt(
+            target_list=[pgast.ResTarget(val=value)],
+            where_clause=cond,
+        )
+        for value, cond in pairs
+    ]
+    union = extend_select_op(None, *selects)
+
+    if not union:
+        return None
+
+    return pgast.SubLink(
+        type=pgast.SubLinkType.EXPR,
+        expr=pgast.SelectStmt(
+            target_list=[
+                pgast.ResTarget(
+                    val=pgast.FuncCall(
+                        name=('string_agg',),
+                        args=[
+                            pgast.ColumnRef(name=('error_msg',)),
+                            pgast.StringConstant(val=', '),
+                        ],
+                    )
+                )
+            ],
+            from_clause=[
+                pgast.RangeSubselect(
+                    subquery=union,
+                    alias=pgast.Alias(aliasname='t', colnames=['error_msg']),
+                )
+            ],
+        ),
+    )
+
+
 def new_unop(op: str, expr: pgast.BaseExpr) -> pgast.Expr:
     return pgast.Expr(
         kind=pgast.ExprKind.OP,
diff --git a/edb/pgsql/compiler/dml.py b/edb/pgsql/compiler/dml.py
index 616adcd2080..d5cd6844fa9 100644
--- a/edb/pgsql/compiler/dml.py
+++ b/edb/pgsql/compiler/dml.py
@@ -286,9 +286,7 @@ def gen_dml_cte(
                 range_rvar, target_ir_set.path_id, env=ctx.env)
         )
         # Do any read-side filtering
-        if (
-            (pol_expr := ir_stmt.read_policy_exprs.get(typeref.id))
-        ):
+        if pol_expr := ir_stmt.read_policies.get(typeref.id):
             with ctx.newrel() as sctx:
                 pathctx.put_path_value_rvar(
                     sctx.rel, target_path_id, relation, env=ctx.env)
@@ -767,7 +765,7 @@ def dynamic_get_path(
 
     dml_cte = contents_cte if not needs_insert_on_conflict else insert_cte
 
-    pol_expr = ir_stmt.write_policy_exprs.get(typeref.id)
+    pol_expr = ir_stmt.write_policies.get(typeref.id)
     pol_ctx = None
     if pol_expr:
         with ctx.new() as pol_ctx:
@@ -821,12 +819,13 @@ def dynamic_get_path(
 
 
 def compile_policy_check(
-        dml_cte: pgast.CommonTableExpr,
-        ir_stmt: irast.MutatingStmt,
-        policy_expr: irast.PolicyExpr,
-        typeref: irast.TypeRef,
-        *,
-        ctx: context.CompilerContextLevel) -> pgast.CommonTableExpr:
+    dml_cte: pgast.CommonTableExpr,
+    ir_stmt: irast.MutatingStmt,
+    access_policies: irast.WritePolicies,
+    typeref: irast.TypeRef,
+    *,
+    ctx: context.CompilerContextLevel,
+) -> pgast.CommonTableExpr:
     subject_id = ir_stmt.subject.path_id
 
     with ctx.newrel() as ictx:
@@ -839,33 +838,99 @@ def compile_policy_check(
         dml_rvar = relctx.rvar_for_rel(dml_cte, ctx=ctx)
         relctx.include_rvar(ictx.rel, dml_rvar, path_id=subject_id, ctx=ictx)
 
-        cond_ref = clauses.compile_filter_clause(
-            policy_expr.expr, policy_expr.cardinality, ctx=ictx)
+        # split and compile
+        allow, deny = [], []
+        for policy in access_policies.policies:
+            cond_ref = clauses.compile_filter_clause(
+                policy.expr, policy.cardinality, ctx=ictx
+            )
+
+            if policy.action == qltypes.AccessPolicyAction.Allow:
+                allow.append((policy, cond_ref))
+            else:
+                deny.append((policy, cond_ref))
+
+        def raise_if(
+            a: pgast.BaseExpr, b: bool, msg: pgast.BaseExpr
+        ) -> pgast.BaseExpr:
+            return pgast.FuncCall(
+                name=('edgedb', 'raise_on_null'),
+                args=[
+                    pgast.FuncCall(
+                        name=('nullif',),
+                        args=[a, pgast.BooleanConstant(val=str(b))],
+                    ),
+                    pgast.StringConstant(val='insufficient_privilege'),
+                    pgast.NamedFuncArg(
+                        name='msg',
+                        val=msg,
+                    ),
+                    pgast.NamedFuncArg(
+                        name='table',
+                        val=pgast.StringConstant(val=str(typeref.id)),
+                    ),
+                ],
+            )
 
-        op = 'insert' if isinstance(ir_stmt, irast.InsertStmt) else 'update'
+        if isinstance(ir_stmt, irast.InsertStmt):
+            op = 'insert'
+        else:
+            op = 'update'
         msg = f'access policy violation on {op} of {typeref.name_hint}'
-        maybe_raise = pgast.FuncCall(
-            name=('edgedb', 'raise_on_null'),
-            args=[
-                pgast.FuncCall(
-                    name=('nullif',),
-                    args=[cond_ref, pgast.BooleanConstant(val='false')],
-                ),
-                pgast.StringConstant(val='insufficient_privilege'),
-                pgast.NamedFuncArg(
-                    name='msg',
-                    val=pgast.StringConstant(val=msg),
-                ),
-                pgast.NamedFuncArg(
-                    name='table',
-                    val=pgast.StringConstant(val=str(typeref.id)),
-                ),
-            ],
-        )
+
+        # allow
+        if allow:
+            allow_conds = (cond for _, cond in allow)
+            allow_expr = astutils.extend_binop(None, *allow_conds, op='OR')
+        else:
+            allow_expr = pgast.BooleanConstant(val='false')
+
+        allow_msgs = (pol.error_msg for pol, _ in allow if pol.error_msg)
+        if allow_msgs:
+            hint = 'none of these allow policies match: '
+            hint += ', '.join(allow_msgs)
+            allow_msg = f'{msg} ({hint})'
+        else:
+            allow_msg = f'{msg} (no allow policies)'
+        allow_msg_expr = pgast.StringConstant(val=allow_msg)
+
         ictx.rel.target_list.append(
-            pgast.ResTarget(name='error', val=maybe_raise)
+            pgast.ResTarget(
+                name=f'error_allow',
+                val=raise_if(allow_expr, False, msg=allow_msg_expr),
+            )
         )
 
+        # deny
+        if deny:
+            deny_conds = (cond for _, cond in deny)
+            deny_expr = astutils.extend_binop(None, *deny_conds, op='OR')
+
+            deny_messages = [
+                (
+                    pgast.StringConstant(
+                        val=f'denied by policy {pol.error_msg}'
+                    ),
+                    cond,
+                )
+                for pol, cond in deny
+                if pol.error_msg
+            ]
+            deny_message = astutils.conditional_string_agg(deny_messages)
+            if deny_message:
+                deny_message = astutils.extend_concat(
+                    msg + ' (', deny_message, ')'
+                )
+            else:
+                deny_message = pgast.StringConstant(val=f'{msg} (denied)')
+
+            ictx.rel.target_list.append(
+                pgast.ResTarget(
+                    name=f'error_deny',
+                    val=raise_if(deny_expr, True, msg=deny_message),
+                )
+            )
+
         policy_cte = pgast.CommonTableExpr(
             query=ictx.rel,
             name=ctx.env.aliases.get('policy'),
@@ -1427,7 +1492,7 @@ def dynamic_get_path(
 
         update_cte.query = update_stmt
 
-    pol_expr = ir_stmt.write_policy_exprs.get(typeref.id)
+    pol_expr = ir_stmt.write_policies.get(typeref.id)
     pol_ctx = None
     if pol_expr:
         with ctx.new() as pol_ctx:
diff --git a/edb/schema/policies.py b/edb/schema/policies.py
index 4e1e16326b3..98e4e02b39d 100644
--- a/edb/schema/policies.py
+++ b/edb/schema/policies.py
@@ -85,6 +85,10 @@ class AccessPolicy(
         compcoef=None,
         inheritable=False)
 
+    errmessage = so.SchemaField(
+        str, default=None, compcoef=0.971, allow_ddl_set=True
+    )
+
     @classmethod
     def get_schema_class_displayname(cls) -> str:
         return 'access policy'
diff --git a/tests/test_edgeql_policies.py b/tests/test_edgeql_policies.py
index 197d559d479..89b2141b8c3 100644
--- a/tests/test_edgeql_policies.py
+++ b/tests/test_edgeql_policies.py
@@ -864,3 +864,34 @@ async def test_edgeql_policies_internal_shape_01(self):
             ''',
             [],
         )
+
+    async def test_edgeql_policies_messages(self):
+        await self.con.execute(
+            '''
+            create type NoAllows {
+                create access policy allow_select
+                    allow select;
+            };
+            create type TwoAllows {
+                create required property val -> str;
+                create access policy allow_insert_of_a
+                    allow insert using (.val = 'a')
+                    { errmessage := 'you can insert a' };
+                create access policy allow_insert_of_b
+                    allow insert using (.val = 'b')
+                    { errmessage := 'you can insert b' };
+            };            
+        '''
+        )
+
+        await self.con.execute("insert TwoAllows { val := 'a' };")
+
+        async with self.assertRaisesRegexTx(
+                edgedb.InvalidValueError,
+                r"no allow policies"):
+            await self.con.query('insert NoAllows')
+
+        async with self.assertRaisesRegexTx(
+                edgedb.InvalidValueError,
+                r"none of these allow policies match: "):
+            await self.con.query("insert TwoAllows { val := 'c' }")


Code review comments:
diff_hunk: @@ -839,33 +838,99 @@ def compile_policy_check(
         dml_rvar = relctx.rvar_for_rel(dml_cte, ctx=ctx)
         relctx.include_rvar(ictx.rel, dml_rvar, path_id=subject_id, ctx=ictx)
 
-        cond_ref = clauses.compile_filter_clause(
-            policy_expr.expr, policy_expr.cardinality, ctx=ictx)
+        # split and compile
+        allow, deny = [], []
+        for policy in access_policies.policies:
+            cond_ref = clauses.compile_filter_clause(
+                policy.expr, policy.cardinality, ctx=ictx
+            )
+
+            if policy.action == qltypes.AccessPolicyAction.Allow:
+                allow.append((policy, cond_ref))
+            else:
+                deny.append((policy, cond_ref))
+
+        def raise_if(
+            a: pgast.BaseExpr, b: bool, msg: pgast.BaseExpr
+        ) -> pgast.BaseExpr:
+            return pgast.FuncCall(
+                name=('edgedb', 'raise_on_null'),
+                args=[
+                    pgast.FuncCall(
+                        name=('nullif',),
+                        args=[a, pgast.BooleanConstant(val=str(b))],
+                    ),
+                    pgast.StringConstant(val='insufficient_privilege'),
+                    pgast.NamedFuncArg(
+                        name='msg',
+                        val=msg,
+                    ),
+                    pgast.NamedFuncArg(
+                        name='table',
+                        val=pgast.StringConstant(val=str(typeref.id)),
+                    ),
+                ],
+            )
 
-        op = 'insert' if isinstance(ir_stmt, irast.InsertStmt) else 'update'
+        if isinstance(ir_stmt, irast.InsertStmt):
+            op = 'insert'
+        else:
+            op = 'update'
         msg = f'access policy violation on {op} of {typeref.name_hint}'
-        maybe_raise = pgast.FuncCall(
-            name=('edgedb', 'raise_on_null'),
-            args=[
-                pgast.FuncCall(
-                    name=('nullif',),
-                    args=[cond_ref, pgast.BooleanConstant(val='false')],
-                ),
-                pgast.StringConstant(val='insufficient_privilege'),
-                pgast.NamedFuncArg(
-                    name='msg',
-                    val=pgast.StringConstant(val=msg),
-                ),
-                pgast.NamedFuncArg(
-                    name='table',
-                    val=pgast.StringConstant(val=str(typeref.id)),
-                ),
-            ],
-        )
+
+        # allow
+        if allow:
+            allow_conds = (cond for _, cond in allow)
+            allow_expr = astutils.extend_binop(None, *allow_conds, op='OR')
+        else:
+            allow_expr = pgast.BooleanConstant(val='false')
+
+        allow_msgs = (pol.error_msg for pol, _ in allow if pol.error_msg)
+        if allow_msgs:
+            hint = 'none of these allow policies match: '
+            hint += ', '.join(allow_msgs)
+            allow_msg = f'{msg} ({hint})'
+        else:
+            allow_msg = f'{msg} (no allow policies)'
path: edb/pgsql/compiler/dml.py
text: @author:
I agree, given that this is security it should not be verbose when denying access.