From 5ab51d8e35046a1a246cf86cecf2ea1e71edde81 Mon Sep 17 00:00:00 2001 From: Andrew Jackson Date: Tue, 13 Oct 2026 23:03:45 -0500 Subject: [PATCH 1/5] fix: add row level security over shared tables Currently any user that has any ability to read the pg_net request or queue tables also has the ability to read every other users data on these tables. This is particularly problematic if users are using secrets (passwords, API keys, etc) in a multi tenant database environment. --- sql/pg_net.sql | 11 ++++++++--- src/core.c | 36 ++++++++++++++++++++++++++++++------ src/core.h | 2 ++ src/worker.c | 5 ++--- 4 files changed, 42 insertions(+), 12 deletions(-) diff --git a/sql/pg_net.sql b/sql/pg_net.sql index b0acd86a..c1313c91 100644 --- a/sql/pg_net.sql +++ b/sql/pg_net.sql @@ -15,8 +15,11 @@ create unlogged table net.http_request_queue( url text not null, headers jsonb, body bytea, - timeout_milliseconds int not null + timeout_milliseconds int not null, + username text not null default current_user ); +alter table net.http_request_queue enable row level security; +create policy net_request_policy on net.http_request_queue using (username OPERATOR(pg_catalog.=) current_user); create or replace function net.check_worker_is_up() returns void as $$ begin @@ -40,9 +43,11 @@ create unlogged table net._http_response( content text, timed_out bool, error_msg text, - created timestamptz not null default now() + created timestamptz not null default now(), + username text not null ); - +alter table net._http_response enable row level security; +create policy net_response_policy on net._http_response using (username OPERATOR(pg_catalog.=) current_user); create index on net._http_response (created); -- Blocks until an http_request is complete diff --git a/src/core.c b/src/core.c index a7f0824a..6e8b6cca 100644 --- a/src/core.c +++ b/src/core.c @@ -99,6 +99,7 @@ void init_curl_handle(CurlHandle *handle, RequestQueueRow row) { } handle->url = TextDatumGetCString(row.url); + handle->username = TextDatumGetCString(row.username); handle->req_body = !row.bodyBin.isnull ? TextDatumGetCString(row.bodyBin.value) : NULL; @@ -219,8 +220,14 @@ uint64 consume_request_queue(const int batch_size) { )\ DELETE FROM net.http_request_queue q\ USING rows WHERE q.id = rows.id\ - RETURNING q.id, q.method, q.url, timeout_milliseconds, array(select key || ': ' || value from jsonb_each_text(q.headers)), q.body", - 1, (Oid[]){INT4OID}); + RETURNING q.id, q.method, q.url, timeout_milliseconds, array(select key || ': ' || value from jsonb_each_text(q.headers)), q.body, q.username", + 1, (Oid[]){INT4OID}); + + if (tmp == NULL) + ereport(ERROR, errmsg("SPI_prepare failed: %s", SPI_result_code_string(SPI_result))); + + del_return_queue_plan = SPI_saveplan(tmp); + if (del_return_queue_plan == NULL) ereport(ERROR, errmsg("SPI_saveplan failed")); } int ret_code = @@ -257,7 +264,10 @@ RequestQueueRow get_request_queue_row(HeapTuple spi_tupval, TupleDesc spi_tupdes NullableDatum bodyBin = {.value = SPI_getbinval(spi_tupval, spi_tupdesc, 6, &tupIsNull), .isnull = tupIsNull}; - return (RequestQueueRow){id, method, url, timeout_milliseconds, headersBin, bodyBin}; + Datum username = SPI_getbinval(spi_tupval, spi_tupdesc, 7, &tupIsNull); + EREPORT_NULL_ATTR(tupIsNull, username); + + return (RequestQueueRow){id, method, url, timeout_milliseconds, headersBin, bodyBin, username}; } static Jsonb *jsonb_headers_from_curl_handle(CURL *ez_handle) { @@ -278,15 +288,23 @@ static Jsonb *jsonb_headers_from_curl_handle(CURL *ez_handle) { return PG_JSONB_OBJECT_FINISH(headers); } -enum { response_nparams = 7 }; // using an enum because const size_t doesn't compile +enum { response_nparams = 8 }; // using an enum because const size_t doesn't compile static void execute_insert_response(Datum *vals, char *nulls) { if (ins_response_plan == NULL) { ins_response_plan = prepare_and_save_plan( "\ - insert into net._http_response(id, status_code, content, headers, content_type, timed_out, error_msg) values ($1, $2, $3, $4, $5, $6, $7)", + insert into net._http_response(id, status_code, content, headers, content_type, timed_out, error_msg, username) values ($1, $2, $3, $4, $5, $6, $7, $8)", response_nparams, - (Oid[response_nparams]){INT8OID, INT4OID, TEXTOID, JSONBOID, TEXTOID, BOOLOID, TEXTOID}); + (Oid[response_nparams]){INT8OID, INT4OID, TEXTOID, JSONBOID, TEXTOID, BOOLOID, TEXTOID, TEXTOID}); + + if (tmp == NULL) + ereport(ERROR, errmsg("SPI_prepare failed: %s", SPI_result_code_string(SPI_result))); + + ins_response_plan = SPI_saveplan(tmp); + if (ins_response_plan == NULL) ereport(ERROR, errmsg("SPI_saveplan failed")); + + SPI_freeplan(tmp); } int ret_code = SPI_execute_plan(ins_response_plan, vals, nulls, false, 0); @@ -307,6 +325,8 @@ void insert_rejected_response(CurlHandle *handle) { nulls[5] = ' '; vals[6] = CStringGetTextDatum(handle->rejected_reason); nulls[6] = ' '; + vals[7] = CStringGetTextDatum(handle->username); + nulls[7] = ' '; execute_insert_response(vals, nulls); } @@ -320,6 +340,9 @@ void insert_response(CurlHandle *handle, CURLcode curl_return_code) { vals[0] = Int64GetDatum(handle->id); nulls[0] = ' '; + vals[7] = CStringGetTextDatum(handle->username); + nulls[7] = ' '; + if (curl_return_code == CURLE_OK) { Jsonb *jsonb_headers = jsonb_headers_from_curl_handle(handle->ez_handle); long res_http_status_code = 0; @@ -346,6 +369,7 @@ void insert_response(CurlHandle *handle, CURLcode curl_return_code) { vals[5] = BoolGetDatum(false); nulls[5] = ' '; + } else { bool timed_out = curl_return_code == CURLE_OPERATION_TIMEDOUT; diff --git a/src/core.h b/src/core.h index d5c449ed..b1db74fd 100644 --- a/src/core.h +++ b/src/core.h @@ -31,6 +31,7 @@ typedef struct { int32 timeout_milliseconds; NullableDatum headersBin; NullableDatum bodyBin; + Datum username; } RequestQueueRow; // The curl easy handle plus additional data, this acts for both the request and @@ -45,6 +46,7 @@ typedef struct { char *method; CURL *ez_handle; char *rejected_reason; // set when the request must not be sent + char *username; // set when the request must not be sent } CurlHandle; uint64 delete_expired_responses(char *ttl, int batch_size); diff --git a/src/worker.c b/src/worker.c index f711c954..09cd950d 100644 --- a/src/worker.c +++ b/src/worker.c @@ -323,15 +323,14 @@ void pg_net_worker(__attribute__((unused)) Datum main_arg) { if (requests_consumed > 0) { CurlHandle *handles = palloc(mul_size(sizeof(CurlHandle), requests_consumed)); - // insert_rejected_response() runs SPI, which resets SPI_tuptable, so keep our own pointer // to the consumed rows SPITupleTable *queue_rows = SPI_tuptable; // initialize curl handles for (size_t j = 0; j < requests_consumed; j++) { - init_curl_handle(&handles[j], - get_request_queue_row(queue_rows->vals[j], queue_rows->tupdesc)); + RequestQueueRow req_queue_row = get_request_queue_row(queue_rows->vals[j], queue_rows->tupdesc); + init_curl_handle(&handles[j], req_queue_row); if (handles[j].rejected_reason) { insert_rejected_response(&handles[j]); From 95eb4d8de9c84ebd71642511fcecf2a0beae8e7f Mon Sep 17 00:00:00 2001 From: Andrew Jackson Date: Tue, 13 Oct 2026 22:59:43 -0500 Subject: [PATCH 2/5] feat: remove breaking changes --- sql/pg_net.sql | 52 +++++++++++++++++++++++-------- src/core.c | 16 +++++----- src/core.h | 4 +-- test/test_rls.py | 80 ++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 129 insertions(+), 23 deletions(-) create mode 100644 test/test_rls.py diff --git a/sql/pg_net.sql b/sql/pg_net.sql index c1313c91..630ed191 100644 --- a/sql/pg_net.sql +++ b/sql/pg_net.sql @@ -16,10 +16,18 @@ create unlogged table net.http_request_queue( headers jsonb, body bytea, timeout_milliseconds int not null, - username text not null default current_user + rolename text not null default 'public' ); alter table net.http_request_queue enable row level security; -create policy net_request_policy on net.http_request_queue using (username OPERATOR(pg_catalog.=) current_user); + +create policy net_request_policy +on net.http_request_queue +using ( + case + when rolename = 'public' then true + else pg_has_role(current_user, rolename, 'member') + end +); create or replace function net.check_worker_is_up() returns void as $$ begin @@ -44,12 +52,21 @@ create unlogged table net._http_response( timed_out bool, error_msg text, created timestamptz not null default now(), - username text not null + rolename text not null ); alter table net._http_response enable row level security; -create policy net_response_policy on net._http_response using (username OPERATOR(pg_catalog.=) current_user); + create index on net._http_response (created); +create policy net_response_policy +on net._http_response +using ( + case + when rolename = 'public' then true + else pg_has_role(current_user, rolename, 'member') + end +); + -- Blocks until an http_request is complete -- API: Private create or replace function net._await_response( @@ -121,7 +138,9 @@ create or replace function net.http_get( -- key/values to be included in request headers headers jsonb default '{}'::jsonb, -- the maximum number of milliseconds the request may take before being cancelled - timeout_milliseconds int default 5000 + timeout_milliseconds int default 5000, + -- the role that should have access to the request/response data + rolename text default 'public' ) -- request_id reference returns bigint @@ -136,12 +155,13 @@ begin from jsonb_each_text(params); -- Add to the request queue - insert into net.http_request_queue(method, url, headers, timeout_milliseconds) + insert into net.http_request_queue(method, url, headers, timeout_milliseconds, rolename) values ( 'GET', net._encode_url_with_params_array(url, params_array), headers, - timeout_milliseconds + timeout_milliseconds, + rolename ) returning id into request_id; @@ -164,7 +184,9 @@ create or replace function net.http_post( -- key/values to be included in request headers headers jsonb default '{"Content-Type": "application/json"}'::jsonb, -- the maximum number of milliseconds the request may take before being cancelled - timeout_milliseconds int DEFAULT 5000 + timeout_milliseconds int DEFAULT 5000, + -- the role that should have access to the request/response data + rolename text default 'public' ) -- request_id reference returns bigint @@ -205,13 +227,14 @@ begin jsonb_each_text(params); -- Add to the request queue - insert into net.http_request_queue(method, url, headers, body, timeout_milliseconds) + insert into net.http_request_queue(method, url, headers, body, timeout_milliseconds, rolename) values ( 'POST', net._encode_url_with_params_array(url, params_array), headers, convert_to(body::text, 'UTF8'), - timeout_milliseconds + timeout_milliseconds, + rolename ) returning id into request_id; @@ -234,7 +257,9 @@ create or replace function net.http_delete( -- the maximum number of milliseconds the request may take before being cancelled timeout_milliseconds int default 5000, -- optional body of the request - body jsonb default NULL + body jsonb default NULL, + -- the role that should have access to the request/response data + rolename text default 'public' ) -- request_id reference returns bigint @@ -249,13 +274,14 @@ begin from jsonb_each_text(params); -- Add to the request queue - insert into net.http_request_queue(method, url, headers, body, timeout_milliseconds) + insert into net.http_request_queue(method, url, headers, body, timeout_milliseconds, rolename) values ( 'DELETE', net._encode_url_with_params_array(url, params_array), headers, convert_to(body::text, 'UTF8'), - timeout_milliseconds + timeout_milliseconds, + rolename ) returning id into request_id; diff --git a/src/core.c b/src/core.c index 6e8b6cca..5eb80d2c 100644 --- a/src/core.c +++ b/src/core.c @@ -99,7 +99,7 @@ void init_curl_handle(CurlHandle *handle, RequestQueueRow row) { } handle->url = TextDatumGetCString(row.url); - handle->username = TextDatumGetCString(row.username); + handle->rolename = TextDatumGetCString(row.rolename); handle->req_body = !row.bodyBin.isnull ? TextDatumGetCString(row.bodyBin.value) : NULL; @@ -220,7 +220,7 @@ uint64 consume_request_queue(const int batch_size) { )\ DELETE FROM net.http_request_queue q\ USING rows WHERE q.id = rows.id\ - RETURNING q.id, q.method, q.url, timeout_milliseconds, array(select key || ': ' || value from jsonb_each_text(q.headers)), q.body, q.username", + RETURNING q.id, q.method, q.url, timeout_milliseconds, array(select key || ': ' || value from jsonb_each_text(q.headers)), q.body, q.rolename", 1, (Oid[]){INT4OID}); if (tmp == NULL) @@ -264,10 +264,10 @@ RequestQueueRow get_request_queue_row(HeapTuple spi_tupval, TupleDesc spi_tupdes NullableDatum bodyBin = {.value = SPI_getbinval(spi_tupval, spi_tupdesc, 6, &tupIsNull), .isnull = tupIsNull}; - Datum username = SPI_getbinval(spi_tupval, spi_tupdesc, 7, &tupIsNull); - EREPORT_NULL_ATTR(tupIsNull, username); + Datum rolename = SPI_getbinval(spi_tupval, spi_tupdesc, 7, &tupIsNull); + EREPORT_NULL_ATTR(tupIsNull, rolename); - return (RequestQueueRow){id, method, url, timeout_milliseconds, headersBin, bodyBin, username}; + return (RequestQueueRow){id, method, url, timeout_milliseconds, headersBin, bodyBin, rolename}; } static Jsonb *jsonb_headers_from_curl_handle(CURL *ez_handle) { @@ -294,7 +294,7 @@ static void execute_insert_response(Datum *vals, char *nulls) { if (ins_response_plan == NULL) { ins_response_plan = prepare_and_save_plan( "\ - insert into net._http_response(id, status_code, content, headers, content_type, timed_out, error_msg, username) values ($1, $2, $3, $4, $5, $6, $7, $8)", + insert into net._http_response(id, status_code, content, headers, content_type, timed_out, error_msg, rolename) values ($1, $2, $3, $4, $5, $6, $7, $8)", response_nparams, (Oid[response_nparams]){INT8OID, INT4OID, TEXTOID, JSONBOID, TEXTOID, BOOLOID, TEXTOID, TEXTOID}); @@ -325,7 +325,7 @@ void insert_rejected_response(CurlHandle *handle) { nulls[5] = ' '; vals[6] = CStringGetTextDatum(handle->rejected_reason); nulls[6] = ' '; - vals[7] = CStringGetTextDatum(handle->username); + vals[7] = CStringGetTextDatum(handle->rolename); nulls[7] = ' '; execute_insert_response(vals, nulls); @@ -340,7 +340,7 @@ void insert_response(CurlHandle *handle, CURLcode curl_return_code) { vals[0] = Int64GetDatum(handle->id); nulls[0] = ' '; - vals[7] = CStringGetTextDatum(handle->username); + vals[7] = CStringGetTextDatum(handle->rolename); nulls[7] = ' '; if (curl_return_code == CURLE_OK) { diff --git a/src/core.h b/src/core.h index b1db74fd..f04bf14b 100644 --- a/src/core.h +++ b/src/core.h @@ -31,7 +31,7 @@ typedef struct { int32 timeout_milliseconds; NullableDatum headersBin; NullableDatum bodyBin; - Datum username; + Datum rolename; } RequestQueueRow; // The curl easy handle plus additional data, this acts for both the request and @@ -46,7 +46,7 @@ typedef struct { char *method; CURL *ez_handle; char *rejected_reason; // set when the request must not be sent - char *username; // set when the request must not be sent + char *rolename; // set when the request must not be sent } CurlHandle; uint64 delete_expired_responses(char *ttl, int batch_size); diff --git a/test/test_rls.py b/test/test_rls.py new file mode 100644 index 00000000..8c154780 --- /dev/null +++ b/test/test_rls.py @@ -0,0 +1,80 @@ +def test_queue_table_insert_default(conn) -> None: + sql = "CREATE USER test_user;" + conn.execute(sql) + + sql = """ + insert into net.http_request_queue(method, url, headers, timeout_milliseconds) + values ( + 'GET', + 'http://localhost:8080', + '{}'::jsonb, + 5000 + ) + returning id; + """ + id = conn.execute(sql).fetchall()[0][0] + conn.commit() + sql = f"SELECT * FROM net.http_request_queue where id = {id};" + response = conn.execute(sql).fetchall() + assert len(response) == 1 + sql = "set role test_user;" + response = conn.execute(sql) + sql = f"SELECT * FROM net.http_request_queue where id = {id};" + response = conn.execute(sql).fetchall() + assert len(response) == 1 + sql = "RESET ROLE;" + response = conn.execute(sql) + sql = "DROP ROLE test_user;" + conn.execute(sql) + conn.commit() + + +def test_queue_table_insert_non_default(conn) -> None: + sql = "CREATE role test_role;" + conn.execute(sql) + sql = "CREATE USER test_user;" + conn.execute(sql) + sql = "GRANT test_role to test_user;" + conn.execute(sql) + + sql = """ + insert into net.http_request_queue(method, url, headers, timeout_milliseconds, rolename) + values ( + 'GET', + 'http://localhost:8080', + '{}'::jsonb, + 5000, + 'test_role' + ) + returning id; + """ + id = conn.execute(sql).fetchall()[0][0] + conn.commit() + sql = f"SELECT * FROM net.http_request_queue where id = {id};" + response = conn.execute(sql).fetchall() + assert len(response) == 1 + sql = f"set role test_user;" + conn.execute(sql) + sql = f"SELECT * FROM net.http_request_queue where id = {id};" + response = conn.execute(sql).fetchall() + assert len(response) == 1 + + sql = "RESET ROLE" + conn.execute(sql) + + sql = "REVOKE test_role FROM test_user;" + conn.execute(sql) + + sql = "set role test_user;" + conn.execute(sql) + + sql = f"SELECT * FROM net.http_request_queue where id = {id};" + response = conn.execute(sql).fetchall() + assert len(response) == 0 + sql = "RESET ROLE;" + conn.execute(sql) + sql = "DROP ROLE test_user;" + conn.execute(sql) + sql = "DROP ROLE test_role;" + conn.execute(sql) + conn.commit() From 5d95f4ab77fe09c73af140a068a7548bfff64719 Mon Sep 17 00:00:00 2001 From: Andrew Jackson Date: Tue, 13 Oct 2026 23:07:20 -0500 Subject: [PATCH 3/5] fix: merge conflict --- src/core.c | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/src/core.c b/src/core.c index 5eb80d2c..94af6c0d 100644 --- a/src/core.c +++ b/src/core.c @@ -222,12 +222,6 @@ uint64 consume_request_queue(const int batch_size) { USING rows WHERE q.id = rows.id\ RETURNING q.id, q.method, q.url, timeout_milliseconds, array(select key || ': ' || value from jsonb_each_text(q.headers)), q.body, q.rolename", 1, (Oid[]){INT4OID}); - - if (tmp == NULL) - ereport(ERROR, errmsg("SPI_prepare failed: %s", SPI_result_code_string(SPI_result))); - - del_return_queue_plan = SPI_saveplan(tmp); - if (del_return_queue_plan == NULL) ereport(ERROR, errmsg("SPI_saveplan failed")); } int ret_code = @@ -297,14 +291,6 @@ static void execute_insert_response(Datum *vals, char *nulls) { insert into net._http_response(id, status_code, content, headers, content_type, timed_out, error_msg, rolename) values ($1, $2, $3, $4, $5, $6, $7, $8)", response_nparams, (Oid[response_nparams]){INT8OID, INT4OID, TEXTOID, JSONBOID, TEXTOID, BOOLOID, TEXTOID, TEXTOID}); - - if (tmp == NULL) - ereport(ERROR, errmsg("SPI_prepare failed: %s", SPI_result_code_string(SPI_result))); - - ins_response_plan = SPI_saveplan(tmp); - if (ins_response_plan == NULL) ereport(ERROR, errmsg("SPI_saveplan failed")); - - SPI_freeplan(tmp); } int ret_code = SPI_execute_plan(ins_response_plan, vals, nulls, false, 0); From c3d5ae33774dcd85142b2aef5b4a173c07fd4ed6 Mon Sep 17 00:00:00 2001 From: Andrew Jackson Date: Tue, 13 Oct 2026 23:12:51 -0500 Subject: [PATCH 4/5] chore: net style --- src/core.c | 13 +++--- src/core.h | 2 +- src/worker.c | 3 +- test/test_engine.py | 2 +- test/test_rls.py | 102 ++++++++++++++++++++++---------------------- 5 files changed, 62 insertions(+), 60 deletions(-) diff --git a/src/core.c b/src/core.c index 94af6c0d..0f0c8f94 100644 --- a/src/core.c +++ b/src/core.c @@ -98,7 +98,7 @@ void init_curl_handle(CurlHandle *handle, RequestQueueRow row) { handle->request_headers = request_headers; } - handle->url = TextDatumGetCString(row.url); + handle->url = TextDatumGetCString(row.url); handle->rolename = TextDatumGetCString(row.rolename); handle->req_body = !row.bodyBin.isnull ? TextDatumGetCString(row.bodyBin.value) : NULL; @@ -221,7 +221,7 @@ uint64 consume_request_queue(const int batch_size) { DELETE FROM net.http_request_queue q\ USING rows WHERE q.id = rows.id\ RETURNING q.id, q.method, q.url, timeout_milliseconds, array(select key || ': ' || value from jsonb_each_text(q.headers)), q.body, q.rolename", - 1, (Oid[]){INT4OID}); + 1, (Oid[]){INT4OID}); } int ret_code = @@ -286,11 +286,12 @@ enum { response_nparams = 8 }; // using an enum because const size_t doesn't com static void execute_insert_response(Datum *vals, char *nulls) { if (ins_response_plan == NULL) { - ins_response_plan = prepare_and_save_plan( - "\ + ins_response_plan = + prepare_and_save_plan("\ insert into net._http_response(id, status_code, content, headers, content_type, timed_out, error_msg, rolename) values ($1, $2, $3, $4, $5, $6, $7, $8)", - response_nparams, - (Oid[response_nparams]){INT8OID, INT4OID, TEXTOID, JSONBOID, TEXTOID, BOOLOID, TEXTOID, TEXTOID}); + response_nparams, + (Oid[response_nparams]){INT8OID, INT4OID, TEXTOID, JSONBOID, TEXTOID, + BOOLOID, TEXTOID, TEXTOID}); } int ret_code = SPI_execute_plan(ins_response_plan, vals, nulls, false, 0); diff --git a/src/core.h b/src/core.h index f04bf14b..32391772 100644 --- a/src/core.h +++ b/src/core.h @@ -46,7 +46,7 @@ typedef struct { char *method; CURL *ez_handle; char *rejected_reason; // set when the request must not be sent - char *rolename; // set when the request must not be sent + char *rolename; // set when the request must not be sent } CurlHandle; uint64 delete_expired_responses(char *ttl, int batch_size); diff --git a/src/worker.c b/src/worker.c index 09cd950d..6277fcda 100644 --- a/src/worker.c +++ b/src/worker.c @@ -329,7 +329,8 @@ void pg_net_worker(__attribute__((unused)) Datum main_arg) { // initialize curl handles for (size_t j = 0; j < requests_consumed; j++) { - RequestQueueRow req_queue_row = get_request_queue_row(queue_rows->vals[j], queue_rows->tupdesc); + RequestQueueRow req_queue_row = + get_request_queue_row(queue_rows->vals[j], queue_rows->tupdesc); init_curl_handle(&handles[j], req_queue_row); if (handles[j].rejected_reason) { diff --git a/test/test_engine.py b/test/test_engine.py index af8f1450..349e19dc 100644 --- a/test/test_engine.py +++ b/test/test_engine.py @@ -2,4 +2,4 @@ def test_connect(conn): """Sanity test verifying connection to postgres works""" conn.execute("select 1") - assert [(1, )] == conn.execute("select 1").fetchall() + assert [(1,)] == conn.execute("select 1").fetchall() diff --git a/test/test_rls.py b/test/test_rls.py index 8c154780..2f63e123 100644 --- a/test/test_rls.py +++ b/test/test_rls.py @@ -1,8 +1,8 @@ def test_queue_table_insert_default(conn) -> None: - sql = "CREATE USER test_user;" - conn.execute(sql) + sql = "CREATE USER test_user;" + conn.execute(sql) - sql = """ + sql = """ insert into net.http_request_queue(method, url, headers, timeout_milliseconds) values ( 'GET', @@ -12,32 +12,32 @@ def test_queue_table_insert_default(conn) -> None: ) returning id; """ - id = conn.execute(sql).fetchall()[0][0] - conn.commit() - sql = f"SELECT * FROM net.http_request_queue where id = {id};" - response = conn.execute(sql).fetchall() - assert len(response) == 1 - sql = "set role test_user;" - response = conn.execute(sql) - sql = f"SELECT * FROM net.http_request_queue where id = {id};" - response = conn.execute(sql).fetchall() - assert len(response) == 1 - sql = "RESET ROLE;" - response = conn.execute(sql) - sql = "DROP ROLE test_user;" - conn.execute(sql) - conn.commit() + id = conn.execute(sql).fetchall()[0][0] + conn.commit() + sql = f"SELECT * FROM net.http_request_queue where id = {id};" + response = conn.execute(sql).fetchall() + assert len(response) == 1 + sql = "set role test_user;" + response = conn.execute(sql) + sql = f"SELECT * FROM net.http_request_queue where id = {id};" + response = conn.execute(sql).fetchall() + assert len(response) == 1 + sql = "RESET ROLE;" + response = conn.execute(sql) + sql = "DROP ROLE test_user;" + conn.execute(sql) + conn.commit() def test_queue_table_insert_non_default(conn) -> None: - sql = "CREATE role test_role;" - conn.execute(sql) - sql = "CREATE USER test_user;" - conn.execute(sql) - sql = "GRANT test_role to test_user;" - conn.execute(sql) + sql = "CREATE role test_role;" + conn.execute(sql) + sql = "CREATE USER test_user;" + conn.execute(sql) + sql = "GRANT test_role to test_user;" + conn.execute(sql) - sql = """ + sql = """ insert into net.http_request_queue(method, url, headers, timeout_milliseconds, rolename) values ( 'GET', @@ -48,33 +48,33 @@ def test_queue_table_insert_non_default(conn) -> None: ) returning id; """ - id = conn.execute(sql).fetchall()[0][0] - conn.commit() - sql = f"SELECT * FROM net.http_request_queue where id = {id};" - response = conn.execute(sql).fetchall() - assert len(response) == 1 - sql = f"set role test_user;" - conn.execute(sql) - sql = f"SELECT * FROM net.http_request_queue where id = {id};" - response = conn.execute(sql).fetchall() - assert len(response) == 1 + id = conn.execute(sql).fetchall()[0][0] + conn.commit() + sql = f"SELECT * FROM net.http_request_queue where id = {id};" + response = conn.execute(sql).fetchall() + assert len(response) == 1 + sql = f"set role test_user;" + conn.execute(sql) + sql = f"SELECT * FROM net.http_request_queue where id = {id};" + response = conn.execute(sql).fetchall() + assert len(response) == 1 - sql = "RESET ROLE" - conn.execute(sql) + sql = "RESET ROLE" + conn.execute(sql) - sql = "REVOKE test_role FROM test_user;" - conn.execute(sql) + sql = "REVOKE test_role FROM test_user;" + conn.execute(sql) - sql = "set role test_user;" - conn.execute(sql) + sql = "set role test_user;" + conn.execute(sql) - sql = f"SELECT * FROM net.http_request_queue where id = {id};" - response = conn.execute(sql).fetchall() - assert len(response) == 0 - sql = "RESET ROLE;" - conn.execute(sql) - sql = "DROP ROLE test_user;" - conn.execute(sql) - sql = "DROP ROLE test_role;" - conn.execute(sql) - conn.commit() + sql = f"SELECT * FROM net.http_request_queue where id = {id};" + response = conn.execute(sql).fetchall() + assert len(response) == 0 + sql = "RESET ROLE;" + conn.execute(sql) + sql = "DROP ROLE test_user;" + conn.execute(sql) + sql = "DROP ROLE test_role;" + conn.execute(sql) + conn.commit() From a2d966ed9c68d9e54dd8f71fdd777be4481a4b8f Mon Sep 17 00:00:00 2001 From: Andrew Jackson Date: Tue, 13 Oct 2026 23:15:49 -0500 Subject: [PATCH 5/5] fix: address issue from steve --- src/worker.c | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/worker.c b/src/worker.c index 6277fcda..b030a89f 100644 --- a/src/worker.c +++ b/src/worker.c @@ -329,9 +329,7 @@ void pg_net_worker(__attribute__((unused)) Datum main_arg) { // initialize curl handles for (size_t j = 0; j < requests_consumed; j++) { - RequestQueueRow req_queue_row = - get_request_queue_row(queue_rows->vals[j], queue_rows->tupdesc); - init_curl_handle(&handles[j], req_queue_row); + init_curl_handle(&handles[j], get_request_queue_row(queue_rows->vals[j], queue_rows->tupdesc)); if (handles[j].rejected_reason) { insert_rejected_response(&handles[j]);