feat(backends): adds grant on table and sequences to roles

Author: weynelucas

db_adapter/db/backends/base/operations.py

@@ -3,25 +3,32 @@
 from django.db.utils import ProgrammingError
 from django.utils.functional import cached_property
 
-from db_adapter.name_builders import ObjectNameBuilder
+from db_adapter.settings import db_settings
 from db_adapter.utils import enforce_model, enforce_model_fields
 
 
 class DatabaseOperations:
+ # Overrideable SQL statements
  sql_create_sequence = None
  sql_create_trigger = None
+ sql_grant = 'GRANT %(privileges)s ON %(name)s TO %(role)s'
 
- name_builder_class = ObjectNameBuilder
+ # Setting variables
+ role_name = db_settings.DEFAULT_ROLE_NAME
+ name_builder_class = db_settings.DEFAULT_NAME_BUILDER_CLASS
+ default_object_privileges = db_settings.DEFAULT_OBJECT_PRIVILEGES
 
  def autoinc_sql(self, table, column):
  if not self.sql_create_sequence and not self.sql_create_trigger:
  return None
 
  model, field = self._enforce_model_field_instances(table, column)
+ sequence_name = self._get_sequence_name(model, field)
+ trigger_name = self._get_trigger_name(model, field)
 
  args = {
- 'sq_name': self._get_sequence_name(model, field),
- 'tr_name': self._get_trigger_name(model, field),
+ 'sq_name': sequence_name,
+ 'tr_name': trigger_name,
  'tbl_name': self.quote_name(table),
  'col_name': self.quote_name(column),
  }
@@ -33,16 +40,30 @@ def autoinc_sql(self, table, column):
  pass
 
  try:
- sequence_sql = self.sql_create_sequence % args
+ sequence_sql = self._get_sequence_sql(sequence_name, args)
  trigger_sql = self.sql_create_trigger % args
- return sequence_sql, trigger_sql
+ return [*sequence_sql, trigger_sql]
  except KeyError as err:
  if 'sq_max_value' in err.args:
  raise ProgrammingError(
  'Cannot retrieve the range of the column type bound to the '
  'field %s' % field.name
  )
 
+ def control_sql(self, name, privileges=None):
+ if not self.role_name:
+ return None
+
+ privileges = (
+ self.default_object_privileges if privileges is None else privileges
+ )
+
+ return self.sql_grant % dict(
+ name=self.quote_name(name),
+ privileges=', '.join(privileges),
+ role=self.quote_name(self.role_name),
+ )
+
  @cached_property
  def name_builder(self):
  return self.name_builder_class()
@@ -63,3 +84,12 @@ def _get_trigger_name(self, table, column=''):
  model, field = self._enforce_model_field_instances(table, column)
  name = self.name_builder.process_name(model, [field], type='trigger')
  return self.quote_name(name)
+
+ def _get_sequence_sql(self, name, args):
+ sequence_sql = [self.sql_create_sequence % args]
+
+ grant_sql = self.control_sql(name, privileges=['SELECT'])
+ if grant_sql:
+ sequence_sql.append(grant_sql)
+
+ return sequence_sql

db_adapter/db/backends/base/schema.py

@@ -17,14 +17,14 @@ class DatabaseSchemaEditor:
  renaming, index fiddling, and so on.
  """
 
- # Deferred column SQL order
  deferred_sql_order = (
  'PRIMARY KEY',
  'UNIQUE',
  'FOREIGN KEY',
  'CHECK',
  'INDEX',
  'COMMENT',
+ 'CONTROL',
  'SEQUENCE',
  'TRIGGER',
  )
@@ -47,6 +47,7 @@ class DatabaseSchemaEditor:
  '_idx': 'index',
  }
 
+ # Setting variables
  name_builder_class = db_settings.DEFAULT_NAME_BUILDER_CLASS
 
  def __init__(self, *args, **kwargs):
@@ -144,8 +145,8 @@ def column_sql(
  model._meta.db_table, field.column
  )
  if autoinc_sql:
- sequence_sql, trigger_sql = autoinc_sql
- self.deferred_column_sql['SEQUENCE'].append(sequence_sql)
+ *sequence_sql, trigger_sql = autoinc_sql
+ self.deferred_column_sql['SEQUENCE'].extend(sequence_sql)
  self.deferred_column_sql['TRIGGER'].append(trigger_sql)
 
  # Comment columns for fields with help_text
@@ -191,6 +192,11 @@ def table_sql(self, model: Model) -> Tuple[str, list]:
  # _remake_table needs it)
  self.deferred_table_sql['INDEX'].extend(self._model_indexes_sql(model))
 
+ # Grant/Revoke object privileges
+ control_sql = self.connection.ops.control_sql(model._meta.db_table)
+ if control_sql:
+ self.deferred_table_sql['CONTROL'].append(control_sql)
+
  return sql, params
 
  def create_model(self, model: Model):

db_adapter/settings.py

@@ -36,8 +36,17 @@
  'DEFAULT_UNIQUE_NAME': '{table}_{columns}_uniq',
  'DEFAULT_CHECK_NAME': '{table}_{columns}{qualifier}_check',
 
+ # Grant options
+ 'DEFAULT_ROLE_NAME': '',
+
  # Base policies
  'DEFAULT_NAME_BUILDER_CLASS': 'db_adapter.name_builders.ObjectNameBuilder',
+ 'DEFAULT_OBJECT_PRIVILEGES': [
+ 'SELECT',
+ 'INSERT',
+ 'UPDATE',
+ 'DELETE',
+ ]
 }
 # fmt: on
 

tests/connection.py

@@ -20,7 +20,7 @@ def quote_name(self, name: str) -> str:
  return name
 
 
-class TestDatabaseOperationsWithSqls(TestDatabaseOperations):
+class TestDatabaseOperationsAutoincSql(TestDatabaseOperations):
  integer_field_ranges = {
  'SmallIntegerField': (-99999999999, 99999999999),
  'IntegerField': (-99999999999, 99999999999),
@@ -52,7 +52,11 @@ class TestDatabaseOperationsWithSqls(TestDatabaseOperations):
 '''
 
 
-class TestDatabaseOperationsOptionalRange(TestDatabaseOperationsWithSqls):
+class TestDatabaseOperationsControlSql(TestDatabaseOperationsAutoincSql):
+ role_name = 'rl_tests'
+
+
+class TestDatabaseOperationsOptionalRange(TestDatabaseOperationsAutoincSql):
  sql_create_sequence = '''
 DECLARE
  i INTEGER;
@@ -71,7 +75,7 @@ class TestDatabaseSchemaEditor(DatabaseSchemaEditor, BaseDatabaseSchemaEditor):
 
 
 class TestDatabaseWrapper(DatabaseWrapper):
- ops_class = TestDatabaseOperationsWithSqls
+ ops_class = TestDatabaseOperationsAutoincSql
  SchemaEditorClass = TestDatabaseSchemaEditor
 
  data_types = {
@@ -119,4 +123,9 @@ class TestDatabaseWrapper(DatabaseWrapper):
  }
 
 
-test_connection = TestDatabaseWrapper(settings_dict={}, alias=DEFAULT_DB_ALIAS)
+class TestDatabaseWrapperControlSql(TestDatabaseWrapper):
+ ops_class = TestDatabaseOperationsControlSql
+
+
+test_connection = TestDatabaseWrapper({}, DEFAULT_DB_ALIAS)
+test_control_connection = TestDatabaseWrapperControlSql({}, DEFAULT_DB_ALIAS)

tests/test_backends/test_base/test_operations.py

@@ -3,21 +3,45 @@
 
 from tests.connection import (
  TestDatabaseOperations,
+ TestDatabaseOperationsAutoincSql,
+ TestDatabaseOperationsControlSql,
  TestDatabaseOperationsOptionalRange,
- TestDatabaseOperationsWithSqls,
  test_connection,
+ test_control_connection,
 )
 
 
+class SqlControlTests(TestCase):
+ def test_control_sql(self):
+ ops = TestDatabaseOperationsAutoincSql(test_connection)
+ control_ops = TestDatabaseOperationsControlSql(test_control_connection)
+
+ sql_none = ops.control_sql('tbl_article')
+ sql_default_privileges = control_ops.control_sql('tbl_article')
+ sql_custom_privileges = control_ops.control_sql(
+ 'tbl_article_sq', privileges=['SELECT']
+ )
+
+ self.assertIsNone(sql_none)
+ self.assertEqual(
+ sql_default_privileges,
+ 'GRANT SELECT, INSERT, UPDATE, DELETE ON tbl_article TO rl_tests',
+ )
+ self.assertEqual(
+ sql_custom_privileges,
+ 'GRANT SELECT ON tbl_article_sq TO rl_tests',
+ )
+
+
 class SqlAutoincTests(TestCase):
  def test_autoinc_sql_without_overwritten_sqls(self):
- ops = TestDatabaseOperations(connection=test_connection)
+ ops = TestDatabaseOperations(test_connection)
  autoinc_sql = ops.autoinc_sql('tbl_article', 'article_id')
 
  self.assertIsNone(autoinc_sql)
 
  def test_autoinc_sql_with_overwritten_sqls(self):
- ops = TestDatabaseOperationsWithSqls(connection=test_connection)
+ ops = TestDatabaseOperationsAutoincSql(test_connection)
  autoinc_sql = ops.autoinc_sql('tbl_article', 'article_id')
 
  self.assertEqual(len(autoinc_sql), 2)
@@ -61,11 +85,11 @@ def test_autoinc_sql_for_required_integer_field_range(self):
  )
 
  with self.assertRaises(ProgrammingError, msg=msg):
- ops = TestDatabaseOperationsWithSqls(connection=test_connection)
+ ops = TestDatabaseOperationsAutoincSql(test_connection)
  ops.autoinc_sql('tbl_article', 'active')
 
  def test_autoinc_sql_for_optional_integer_field_range(self):
- ops = TestDatabaseOperationsOptionalRange(connection=test_connection)
+ ops = TestDatabaseOperationsOptionalRange(test_connection)
  autoinc_sql = ops.autoinc_sql('tbl_article', 'active')
 
  self.assertEqual(len(autoinc_sql), 2)
@@ -84,3 +108,14 @@ def test_autoinc_sql_for_optional_integer_field_range(self):
  END IF;
 END''',
  )
+
+ def test_autoinc_sql_grant_sequence(self):
+ ops = TestDatabaseOperationsControlSql(test_control_connection)
+ autoinc_sql = ops.autoinc_sql('tbl_article', 'article_id')
+
+ self.assertEqual(len(autoinc_sql), 3)
+
+ _, grant_sequence_sql, _ = autoinc_sql
+ self.assertEqual(
+ grant_sequence_sql, 'GRANT SELECT ON tbl_article_sq TO rl_tests'
+ )