mig5/bouquin-sqlcipher4
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Allow authorizer callback to be cleared

Browse source
This commit is contained in:
Charles Leifer 2021-08-04 14:10:22 -05:00
parent 6247967ad2
commit 18551b8707
2 changed files with 16 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

16
src/connection.c
View file

@ -1225,7 +1225,6 @@ static PyObject* pysqlite_connection_set_authorizer(pysqlite_Connection* self, P
PyObject* authorizer_cb; PyObject* authorizer_cb;
static char *kwlist[] = { "authorizer_callback", NULL }; static char *kwlist[] = { "authorizer_callback", NULL };
int rc;
if (!pysqlite_check_thread(self) || !pysqlite_check_connection(self)) { if (!pysqlite_check_thread(self) || !pysqlite_check_connection(self)) {
return NULL; return NULL;
@ -1236,14 +1235,21 @@ static PyObject* pysqlite_connection_set_authorizer(pysqlite_Connection* self, P
return NULL; return NULL;
} }
rc = sqlite3_set_authorizer(self->db, _authorizer_callback, (void*)authorizer_cb); int rc;
if (authorizer_cb == Py_None) {
rc = sqlite3_set_authorizer(self->db, NULL, NULL);
Py_XSETREF(self->function_pinboard_authorizer_cb, NULL);
}
else {
Py_INCREF(authorizer_cb);
Py_XSETREF(self->function_pinboard_authorizer_cb, authorizer_cb);
rc = sqlite3_set_authorizer(self->db, _authorizer_callback, (void*)authorizer_cb);
}
if (rc != SQLITE_OK) { if (rc != SQLITE_OK) {
PyErr_SetString(pysqlite_OperationalError, "Error setting authorizer callback"); PyErr_SetString(pysqlite_OperationalError, "Error setting authorizer callback");
Py_XSETREF(self->function_pinboard_authorizer_cb, NULL); Py_XSETREF(self->function_pinboard_authorizer_cb, NULL);
return NULL; return NULL;
} else {
Py_INCREF(authorizer_cb);
Py_XSETREF(self->function_pinboard_authorizer_cb, authorizer_cb);
} }
Py_RETURN_NONE; Py_RETURN_NONE;
} }

5
test/userfunctions.py
View file

@ -482,6 +482,11 @@ class AuthorizerTests(unittest.TestCase):
self.con.execute("select c2 from t1") self.con.execute("select c2 from t1")
self.assertIn('prohibited', str(cm.exception)) self.assertIn('prohibited', str(cm.exception))
def test_clear_authorizer(self):
self.con.set_authorizer(None)
self.con.execute('select * from t2')
self.con.execute('select c2 from t1')
class AuthorizerRaiseExceptionTests(AuthorizerTests): class AuthorizerRaiseExceptionTests(AuthorizerTests):
@staticmethod @staticmethod
def authorizer_cb(action, arg1, arg2, dbname, source): def authorizer_cb(action, arg1, arg2, dbname, source):