Check INSERT privileges in graph generators

Makefile

@@ -215,6 +215,7 @@ REGRESS = scan \
           index \
           analyze \
           graph_generation \
+          graph_generation_security \
           name_validation \
           jsonb_operators \
           list_comprehension \

regress/expected/graph_generation_security.out

@@ -0,0 +1,129 @@
+/* Graph generators must respect table INSERT privileges and RLS. */
+LOAD 'age';
+SET search_path = ag_catalog, public;
+SELECT create_graph('gen_security');
+NOTICE:  graph "gen_security" has been created
+ create_graph 
+--------------
+
+(1 row)
+
+SELECT * FROM cypher('gen_security',
+    $$CREATE (:Node {key:'root'})-[:LINK]->(:Node {key:'target'})$$)
+    AS (v agtype);
+ v 
+---
+(0 rows)
+
+CREATE ROLE gen_reader;
+GRANT USAGE ON SCHEMA ag_catalog, gen_security TO gen_reader;
+GRANT USAGE ON ALL SEQUENCES IN SCHEMA gen_security TO gen_reader;
+SET ROLE gen_reader;
+SELECT create_complete_graph('gen_security', 2, 'LINK', 'Node');
+ERROR:  permission denied for table Node
+SELECT age_create_barbell_graph('gen_security', 3, 0, 'Node', NULL,
+                                'LINK', NULL);
+ERROR:  permission denied for table Node
+RESET ROLE;
+SELECT count(*) FROM gen_security."Node";
+ count 
+-------
+     2
+(1 row)
+
+SELECT count(*) FROM gen_security."LINK";
+ count 
+-------
+     1
+(1 row)
+
+/* A missing edge INSERT grant must fail before any vertex is written. */
+GRANT INSERT ON gen_security."Node" TO gen_reader;
+SET ROLE gen_reader;
+SELECT create_complete_graph('gen_security', 2, 'LINK', 'Node');
+ERROR:  permission denied for table LINK
+RESET ROLE;
+SELECT count(*) FROM gen_security."Node";
+ count 
+-------
+     2
+(1 row)
+
+SELECT count(*) FROM gen_security."LINK";
+ count 
+-------
+     1
+(1 row)
+
+/* Column-level INSERT grants on every written column are sufficient. */
+CREATE ROLE gen_columns;
+GRANT USAGE ON SCHEMA ag_catalog, gen_security TO gen_columns;
+GRANT USAGE ON ALL SEQUENCES IN SCHEMA gen_security TO gen_columns;
+GRANT INSERT (id, properties) ON gen_security."Node" TO gen_columns;
+GRANT INSERT (id, start_id, end_id, properties)
+    ON gen_security."LINK" TO gen_columns;
+SET ROLE gen_columns;
+SELECT create_complete_graph('gen_security', 2, 'LINK', 'Node');
+ create_complete_graph 
+-----------------------
+
+(1 row)
+
+RESET ROLE;
+SELECT count(*) FROM gen_security."Node";
+ count 
+-------
+     4
+(1 row)
+
+SELECT count(*) FROM gen_security."LINK";
+ count 
+-------
+     2
+(1 row)
+
+/* Direct tuple insertion cannot apply the graph tables' INSERT policies. */
+GRANT INSERT ON ALL TABLES IN SCHEMA gen_security TO gen_reader;
+ALTER TABLE gen_security."_ag_label_vertex" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE gen_security."_ag_label_vertex" FORCE ROW LEVEL SECURITY;
+ALTER TABLE gen_security."Node" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE gen_security."Node" FORCE ROW LEVEL SECURITY;
+ALTER TABLE gen_security."_ag_label_edge" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE gen_security."_ag_label_edge" FORCE ROW LEVEL SECURITY;
+ALTER TABLE gen_security."LINK" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE gen_security."LINK" FORCE ROW LEVEL SECURITY;
+SET ROLE gen_reader;
+SELECT create_complete_graph('gen_security', 2, 'LINK', 'Node');
+ERROR:  graph generation is not supported with row-level security
+HINT:  Use Cypher CREATE instead.
+SELECT age_create_barbell_graph('gen_security', 3, 0, 'Node', NULL,
+                                'LINK', NULL);
+ERROR:  graph generation is not supported with row-level security
+HINT:  Use Cypher CREATE instead.
+RESET ROLE;
+SELECT count(*) FROM gen_security."Node";
+ count 
+-------
+     4
+(1 row)
+
+SELECT count(*) FROM gen_security."LINK";
+ count 
+-------
+     2
+(1 row)
+
+SELECT drop_graph('gen_security', true);
+NOTICE:  drop cascades to 4 other objects
+DETAIL:  drop cascades to table gen_security._ag_label_vertex
+drop cascades to table gen_security._ag_label_edge
+drop cascades to table gen_security."Node"
+drop cascades to table gen_security."LINK"
+NOTICE:  graph "gen_security" has been dropped
+ drop_graph 
+------------
+
+(1 row)
+
+DROP OWNED BY gen_reader, gen_columns;
+DROP ROLE gen_reader, gen_columns;

regress/sql/graph_generation_security.sql

@@ -0,0 +1,65 @@
+/* Graph generators must respect table INSERT privileges and RLS. */
+LOAD 'age';
+SET search_path = ag_catalog, public;
+
+SELECT create_graph('gen_security');
+SELECT * FROM cypher('gen_security',
+    $$CREATE (:Node {key:'root'})-[:LINK]->(:Node {key:'target'})$$)
+    AS (v agtype);
+
+CREATE ROLE gen_reader;
+GRANT USAGE ON SCHEMA ag_catalog, gen_security TO gen_reader;
+GRANT USAGE ON ALL SEQUENCES IN SCHEMA gen_security TO gen_reader;
+
+SET ROLE gen_reader;
+SELECT create_complete_graph('gen_security', 2, 'LINK', 'Node');
+SELECT age_create_barbell_graph('gen_security', 3, 0, 'Node', NULL,
+                                'LINK', NULL);
+RESET ROLE;
+SELECT count(*) FROM gen_security."Node";
+SELECT count(*) FROM gen_security."LINK";
+
+/* A missing edge INSERT grant must fail before any vertex is written. */
+GRANT INSERT ON gen_security."Node" TO gen_reader;
+SET ROLE gen_reader;
+SELECT create_complete_graph('gen_security', 2, 'LINK', 'Node');
+RESET ROLE;
+SELECT count(*) FROM gen_security."Node";
+SELECT count(*) FROM gen_security."LINK";
+
+/* Column-level INSERT grants on every written column are sufficient. */
+CREATE ROLE gen_columns;
+GRANT USAGE ON SCHEMA ag_catalog, gen_security TO gen_columns;
+GRANT USAGE ON ALL SEQUENCES IN SCHEMA gen_security TO gen_columns;
+GRANT INSERT (id, properties) ON gen_security."Node" TO gen_columns;
+GRANT INSERT (id, start_id, end_id, properties)
+    ON gen_security."LINK" TO gen_columns;
+
+SET ROLE gen_columns;
+SELECT create_complete_graph('gen_security', 2, 'LINK', 'Node');
+RESET ROLE;
+SELECT count(*) FROM gen_security."Node";
+SELECT count(*) FROM gen_security."LINK";
+
+/* Direct tuple insertion cannot apply the graph tables' INSERT policies. */
+GRANT INSERT ON ALL TABLES IN SCHEMA gen_security TO gen_reader;
+ALTER TABLE gen_security."_ag_label_vertex" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE gen_security."_ag_label_vertex" FORCE ROW LEVEL SECURITY;
+ALTER TABLE gen_security."Node" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE gen_security."Node" FORCE ROW LEVEL SECURITY;
+ALTER TABLE gen_security."_ag_label_edge" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE gen_security."_ag_label_edge" FORCE ROW LEVEL SECURITY;
+ALTER TABLE gen_security."LINK" ENABLE ROW LEVEL SECURITY;
+ALTER TABLE gen_security."LINK" FORCE ROW LEVEL SECURITY;
+
+SET ROLE gen_reader;
+SELECT create_complete_graph('gen_security', 2, 'LINK', 'Node');
+SELECT age_create_barbell_graph('gen_security', 3, 0, 'Node', NULL,
+                                'LINK', NULL);
+RESET ROLE;
+SELECT count(*) FROM gen_security."Node";
+SELECT count(*) FROM gen_security."LINK";
+
+SELECT drop_graph('gen_security', true);
+DROP OWNED BY gen_reader, gen_columns;
+DROP ROLE gen_reader, gen_columns;

src/backend/utils/graph_generation.c

@@ -20,8 +20,42 @@
 #include "postgres.h"
 
 #include "access/genam.h"
+#include "catalog/pg_namespace.h"
+#include "miscadmin.h"
 #include "commands/graph_commands.h"
+#include "utils/acl.h"
 #include "utils/load/age_load.h"
+#include "utils/rls.h"
+
+/* The generators insert tuples directly, without the executor's ACL/RLS checks. */
+static void check_generation_insert(Oid relid, int columns)
+{
+    if (pg_class_aclcheck(relid, GetUserId(), ACL_INSERT) != ACLCHECK_OK)
+    {
+        int attnum;
+
+        /* INSERT on every written column is also sufficient in PostgreSQL. */
+        for (attnum = 1; attnum <= columns; attnum++)
+        {
+            AclResult aclresult = pg_attribute_aclcheck(relid, attnum,
+                                                         GetUserId(), ACL_INSERT);
+
+            if (aclresult != ACLCHECK_OK)
+            {
+                aclcheck_error(aclresult, OBJECT_TABLE, get_rel_name(relid));
+            }
+        }
+    }
+
+    /* These direct writes cannot evaluate INSERT WITH CHECK policies. */
+    if (check_enable_rls(relid, InvalidOid, true) == RLS_ENABLED)
+    {
+        ereport(ERROR,
+                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+                 errmsg("graph generation is not supported with row-level security"),
+                 errhint("Use Cypher CREATE instead.")));
+    }
+}
 
 
 int64 get_nextval_internal(graph_cache_data* graph_cache,
@@ -169,6 +203,19 @@ Datum create_complete_graph(PG_FUNCTION_ARGS)
     vtx_seq_id = get_relname_relid(vtx_seq_name_str, nsp_id);
     edge_seq_id = get_relname_relid(edge_seq_name_str, nsp_id);
 
+    /* Check both labels before consuming sequence values or inserting rows. */
+    {
+        AclResult aclresult = object_aclcheck(NamespaceRelationId, nsp_id,
+                                              GetUserId(), ACL_USAGE);
+
+        if (aclresult != ACLCHECK_OK)
+        {
+            aclcheck_error(aclresult, OBJECT_SCHEMA, graph_name_str);
+        }
+    }
+    check_generation_insert(get_label_relation(vtx_name_str, graph_oid), 2);
+    check_generation_insert(get_label_relation(edge_name_str, graph_oid), 4);
+
     props = create_empty_agtype();
 
     /* Creating vertices*/

tools/graph-generation-security/README.md

@@ -0,0 +1,15 @@
+# Graph generator permission reproducer
+
+Run this only against a disposable PostgreSQL database with AGE installed. The
+test creates and removes synthetic graphs and login roles. The generated logins
+need trust authentication for a separate connection.
+
+```sh
+python3 -m pip install 'psycopg[binary]'
+python3 reproduce.py --host 127.0.0.1 --dbname age_security_audit
+```
+
+Each case compares a denied SQL `INSERT` with a graph generator call and checks
+the committed row counts from a new connection. `bypass: true` indicates the
+generator wrote rows that the login could not insert with SQL. The RLS cases
+grant table `INSERT` but force RLS with no `INSERT` policy.

tools/graph-generation-security/reproduce.py

@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+"""Reproduce graph generator INSERT/RLS bypasses in a disposable AGE database."""
+
+import argparse
+import json
+import uuid
+
+import psycopg
+
+
+def outcome(conn, sql):
+    try:
+        conn.execute(sql)
+        return {"sqlstate": "00000"}
+    except psycopg.Error as exc:
+        return {"sqlstate": exc.sqlstate, "message": exc.diag.message_primary}
+
+
+def counts(conn, graph):
+    return {
+        label: conn.execute(f'SELECT count(*) FROM {graph}."{label}"').fetchone()[0]
+        for label in ("Node", "LINK")
+    }
+
+
+def run_case(connection, function, rls):
+    suffix = uuid.uuid4().hex[:10]
+    graph = f"generator_repro_{suffix}"
+    role = f"generator_reader_{suffix}"
+    report = {"function": function, "rls": rls}
+
+    with psycopg.connect(user=connection["admin_user"], host=connection["host"],
+                         dbname=connection["dbname"], autocommit=True) as admin:
+        admin.execute("SET search_path=ag_catalog,public")
+        admin.execute(f"CREATE ROLE {role} LOGIN NOSUPERUSER NOBYPASSRLS NOINHERIT")
+        try:
+            admin.execute(f"SELECT ag_catalog.create_graph('{graph}')")
+            admin.execute(f"SELECT * FROM ag_catalog.cypher('{graph}', "
+                          "$cy$CREATE (:Node {key:'root'})-[:LINK]->"
+                          "(:Node {key:'target'})$cy$) AS (v agtype)")
+            admin.execute(f"GRANT USAGE ON SCHEMA ag_catalog,{graph} TO {role}")
+            admin.execute(f"GRANT USAGE ON ALL SEQUENCES IN SCHEMA {graph} TO {role}")
+            if rls:
+                admin.execute(f"GRANT INSERT ON ALL TABLES IN SCHEMA {graph} TO {role}")
+                for label in ("_ag_label_vertex", "Node", "_ag_label_edge", "LINK"):
+                    admin.execute(f'ALTER TABLE {graph}."{label}" ENABLE ROW LEVEL SECURITY')
+                    admin.execute(f'ALTER TABLE {graph}."{label}" FORCE ROW LEVEL SECURITY')
+            report["before"] = counts(admin, graph)
+
+            with psycopg.connect(user=role, host=connection["host"],
+                                 dbname=connection["dbname"], autocommit=True) as reader:
+                reader.execute("SET search_path=ag_catalog,public")
+                reader.execute("SET statement_timeout='2s'")
+                report["identity"] = reader.execute(
+                    "SELECT rolsuper,rolbypassrls FROM pg_roles WHERE rolname=current_user"
+                ).fetchone()
+                report["table_insert"] = {
+                    label: reader.execute(
+                        f"SELECT has_table_privilege(current_user,'{graph}.\"{label}\"','INSERT')"
+                    ).fetchone()[0]
+                    for label in ("Node", "LINK")
+                }
+                report["direct_insert"] = outcome(
+                    reader, f'INSERT INTO {graph}."Node" DEFAULT VALUES')
+                if function == "create_complete_graph":
+                    call = (f"SELECT ag_catalog.create_complete_graph('{graph}', "
+                            "2, 'LINK', 'Node')")
+                else:
+                    call = (f"SELECT ag_catalog.age_create_barbell_graph('{graph}', "
+                            "3, 0, 'Node', NULL, 'LINK', NULL)")
+                report["generator"] = outcome(reader, call)
+
+            # A separate connection checks committed rows after the caller exits.
+            with psycopg.connect(user=connection["admin_user"],
+                                 host=connection["host"],
+                                 dbname=connection["dbname"], autocommit=True) as verify:
+                report["after"] = counts(verify, graph)
+            report["bypass"] = (report["direct_insert"]["sqlstate"] == "42501"
+                                and report["generator"]["sqlstate"] == "00000"
+                                and report["after"] != report["before"])
+        finally:
+            admin.execute(f"SELECT ag_catalog.drop_graph('{graph}', true)")
+            admin.execute(f"DROP OWNED BY {role}")
+            admin.execute(f"DROP ROLE {role}")
+    return report
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--host", required=True)
+    parser.add_argument("--dbname", required=True)
+    parser.add_argument("--admin-user", default="postgres")
+    args = parser.parse_args()
+    connection = vars(args)
+    results = [
+        run_case(connection, function, rls)
+        for rls in (False, True)
+        for function in ("create_complete_graph", "age_create_barbell_graph")
+    ]
+    print(json.dumps(results, indent=2))
+
+
+if __name__ == "__main__":
+    main()