damien
/
muk_base
mirror of https://github.com/muk-it/muk_base
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

publish muk_security - 12.0

pull/9/head
MuK IT GmbH 6 years ago
parent
da761b78f6
commit
5eb4eabde8
8 changed files with 234 additions and 225 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      muk_security/__manifest__.py
  2. 5
      muk_security/doc/changelog.rst
  3. 2
      muk_security/models/__init__.py
  4. 48
      muk_security/models/base.py
  5. 321
      muk_security/models/mixins_access_groups.py
  6. 65
      muk_security/models/mixins_access_rights.py
  7. 8
      muk_security/models/mixins_locking.py
  8. 8
      muk_security/models/res_users.py

2
muk_security/__manifest__.py
View File

@ -20,7 +20,7 @@
{
"name": "MuK Security",
"summary": """Security Features""",
"version": "12.0.1.1.2",
"version": "12.0.1.2.0",
"category": "Extra Tools",
"license": "AGPL-3",
"website": "http://www.mukit.at",

5
muk_security/doc/changelog.rst
View File

@ -1,3 +1,8 @@
`1.2.0`
-------
- Restructuring
`1.1.0`
-------

2
muk_security/models/__init__.py
View File

@ -19,7 +19,7 @@
from . import base
from . import mixins_locking
from . import mixins_access
from . import mixins_access_rights
from . import mixins_access_groups
from . import access_groups
from . import ir_model_access

48
muk_security/models/base.py
View File

@ -29,12 +29,58 @@ class Base(models.AbstractModel):
_inherit = 'base'
#----------------------------------------------------------
# Helper
#----------------------------------------------------------
def _filter_access_rules(self, operation):
if isinstance(self.env.uid, NoSecurityUid):
return self
return super(Base, self)._filter_access_rules(operation)
@api.multi
def _filter_access(self, operation):
if self.check_access_rights('read', False):
return self._filter_access_rules(operation)
return self.env[self._name]
@api.multi
def _filter_access_ids(self, operation):
return self._filter_access_rules(operation).ids
@api.model
def _apply_ir_rules(self, query, mode='read'):
if isinstance(self.env.uid, NoSecurityUid):
return None
return super(Base, self)._apply_ir_rules(query, mode=mode)
#----------------------------------------------------------
# Function
#----------------------------------------------------------
@api.model
def suspend_security(self, user=None):
return self.sudo(user=NoSecurityUid(user or self.env.uid))
@api.multi
def check_access_rule(self, operation):
if isinstance(self.env.uid, NoSecurityUid):
return None
return super(Base, self).check_access_rule(operation)
@api.model
def check_field_access_rights(self, operation, fields):
if isinstance(self.env.uid, NoSecurityUid):
return fields or list(self._fields)
return super(Base, self).check_field_access_rights(operation, fields)
return super(Base, self).check_field_access_rights(operation, fields)
@api.multi
def check_access(self, operation, raise_exception=False):
try:
access_right = self.check_access_rights(operation, raise_exception)
access_rule = self.check_access_rule(operation) is None
return access_right and access_rule
except AccessError:
if raise_exception:
raise
return False

321
muk_security/models/mixins_access_groups.py
View File

@ -19,9 +19,9 @@
import logging
from odoo import _, SUPERUSER_ID
from odoo import models, api, fields
from odoo import _, models, api, fields, SUPERUSER_ID
from odoo.exceptions import AccessError
from odoo.osv import expression
from odoo.addons.muk_security.tools.security import NoSecurityUid
@ -31,22 +31,19 @@ class AccessGroupsModel(models.AbstractModel):
_name = 'muk_security.mixins.access_groups'
_description = "Group Access Mixin"
_inherit = 'muk_security.mixins.access'
# Set it to True to enforced security even if no group has been set
_strict_security = False
_inherit = 'muk_security.mixins.access_rights'
# If set the group fields are restricted by the access group
_field_groups = None
_access_groups_fields = None
# If set to True the model is extended by fields to suspend the checks
_active_suspend = False
# If set the suspend fields are restricted by the access group
_suspend_groups = None
# Set it to True to enforced security even if no group has been set
_access_groups_strict = False
# Set it to True to let the non strict mode check for existing groups per mode
_access_groups_mode = False
#----------------------------------------------------------
# Datebase
# Datebase
#----------------------------------------------------------
@api.model
@ -55,31 +52,6 @@ class AccessGroupsModel(models.AbstractModel):
def add(name, field):
if name not in self._fields:
self._add_field(name, field)
if self._active_suspend:
add('suspend_security_read', fields.Boolean(
_module=self._module,
string="Suspend Security for Read",
automatic=True,
default=False,
groups=self._suspend_groups))
add('suspend_security_create', fields.Boolean(
_module=self._module,
string="Suspend Security for Create",
automatic=True,
default=False,
groups=self._suspend_groups))
add('suspend_security_write', fields.Boolean(
_module=self._module,
string="Suspend Security for Write",
automatic=True,
default=False,
groups=self._suspend_groups))
add('suspend_security_unlink', fields.Boolean(
_module=self._module,
string="Suspend Security for Unlink",
automatic=True,
default=False,
groups=self._suspend_groups))
add('groups', fields.Many2many(
_module=self._module,
comodel_name='muk_security.access_groups',
@ -88,7 +60,7 @@ class AccessGroupsModel(models.AbstractModel):
column2='gid',
string="Groups",
automatic=True,
groups=self._field_groups))
groups=self._access_groups_fields))
add('complete_groups', fields.Many2many(
_module=self._module,
comodel_name='muk_security.access_groups',
@ -99,177 +71,160 @@ class AccessGroupsModel(models.AbstractModel):
compute='_compute_groups',
store=True,
automatic=True,
groups=self._field_groups))
groups=self._access_groups_fields))
#----------------------------------------------------------
# Function
# Helper
#----------------------------------------------------------
@api.model
def _get_no_access_ids(self):
if not self._strict_security:
sql = '''
SELECT id
FROM %s a
WHERE NOT EXISTS (
SELECT *
FROM %s_complete_groups_rel r
WHERE r.aid = a.id
);
''' % (self._table, self._table)
self.env.cr.execute(sql)
fetch = self.env.cr.fetchall()
return len(fetch) > 0 and list(map(lambda x: x[0], fetch)) or []
else:
return []
@api.model
def _get_suspended_access_ids(self, operation):
if self._active_suspend:
sql = '''
SELECT id
FROM %s a
WHERE a.suspend_security_%s = true
''' % (self._table, operation)
self.env.cr.execute(sql)
fetch = self.env.cr.fetchall()
return len(fetch) > 0 and list(map(lambda x: x[0], fetch)) or []
else:
return []
@api.model
def _get_access_ids(self):
model = self._name.split(".")[-1]
sql = '''
SELECT r.aid
FROM %s_complete_groups_rel r
JOIN muk_security_access_groups g ON r.gid = g.id
JOIN muk_security_access_groups_users_rel u ON r.gid = u.gid
WHERE u.uid = %s AND g.perm_read = true
''' % (self._table, self.env.user.id)
self.env.cr.execute(sql)
fetch = self.env.cr.fetchall()
access_ids = len(fetch) > 0 and list(map(lambda x: x[0], fetch)) or []
return access_ids
@api.model
def _get_ids_without_security(self, operation):
no_access_ids = self._get_no_access_ids()
suspended_access_ids = self._get_suspended_access_ids(operation)
return list(set(no_access_ids).union(suspended_access_ids))
@api.model
def _get_complete_access_ids(self, operation):
access_ids = self._get_access_ids()
no_access_ids = self._get_no_access_ids()
suspended_access_ids = self._get_suspended_access_ids(operation)
return list(set(access_ids).union(no_access_ids, suspended_access_ids))
@api.multi
def _eval_access_skip(self, operation):
if isinstance(self.env.uid, NoSecurityUid):
return True
return False
def _filter_access(self, operation):
records = super(AccessGroupsModel, self)._filter_access(operation)
return records.filter_access_groups(operation)
@api.multi
def check_access_groups(self, operation):
if self.env.user.id == SUPERUSER_ID or self._eval_access_skip(operation):
@api.model
def _apply_access_groups(self, query, mode='read'):
if self.env.user.id == SUPERUSER_ID or isinstance(self.env.uid, NoSecurityUid):
return None
filter_ids = self._get_ids_without_security(operation)
for record in self.filtered(lambda rec: rec.id not in filter_ids):
sql = '''
SELECT perm_%s
FROM %s_complete_groups_rel r
where_clause = '''
"{table}"."id" IN (
SELECT r.aid
JOIN {table}_complete_groups_rel r
JOIN muk_security_access_groups g ON r.gid = g.id
JOIN muk_security_access_groups_users_rel u ON r.gid = u.gid
WHERE r.aid = %s AND u.uid = %s
''' % (operation, self._table, record.id, self.env.user.id)
self.env.cr.execute(sql)
fetch = self.env.cr.fetchall()
if not any(list(map(lambda x: x[0], fetch))):
raise AccessError(_("This operation is forbidden!"))
WHERE (u.uid = %s AND g.perm_{mode} = true)
)
'''.format(table=self._table, mode=mode)
if not self._access_groups_strict:
or_clause = '''
OR NOT EXISTS (
SELECT 1
FROM {table}_complete_groups_rel sr
JOIN muk_security_access_groups sg ON sr.gid = sg.id
WHERE sr.aid = "{table}"."id"
)
'''.format(table=self._table)
if self._access_groups_mode:
or_clause += 'AND sg.perm_{mode} = true'.format(mode=mode)
where_clause += or_clause
query.where_clause += [where_clause]
query.where_clause_params += [self.env.user.id]
@api.model
def _apply_ir_rules(self, query, mode='read'):
super(AccessGroupsModel, self)._apply_ir_rules(query, mode=mode)
self._apply_access_groups(query, mode=mode)
#----------------------------------------------------------
# Function
#----------------------------------------------------------
@api.multi
def check_access(self, operation, raise_exception=False):
res = super(BaseModelAccessGroups, self).check_access(operation, raise_exception)
res = super(AccessGroupsModel, self).check_access(operation, raise_exception)
try:
access_groups = self.check_access_groups(operation) == None
return res and access_groups
return res and self.check_access_groups(operation) == None
except AccessError:
if raise_exception:
raise
return False
#----------------------------------------------------------
# Read
#----------------------------------------------------------
@api.multi
def _after_read(self, result, *largs, **kwargs):
result = super(BaseModelAccessGroups, self)._after_read(result)
if self.env.user.id == SUPERUSER_ID or self._eval_access_skip("read"):
return result
access_ids = self._get_complete_access_ids("read")
result = [result] if not isinstance(result, list) else result
if len(access_ids) > 0:
access_result = []
for record in result:
if record['id'] in access_ids:
access_result.append(record)
return access_result
return []
@api.model
def _after_search(self, result, *largs, **kwargs):
result = super(BaseModelAccessGroups, self)._after_search(result)
if self.env.user.id == SUPERUSER_ID or self._eval_access_skip("read"):
return result
access_ids = self._get_complete_access_ids("read")
if len(access_ids) > 0:
access_result = self.env[self._name]
if isinstance(result, int):
if result in access_ids:
return result
else:
for record in result:
if record.id in access_ids:
access_result += record
return access_result
return self.env[self._name]
@api.model
def _after_name_search(self, result, *largs, **kwargs):
result = super(BaseModelAccessGroups, self)._after_name_search(result)
if self.env.user.id == SUPERUSER_ID or self._eval_access_skip("read"):
return result
access_ids = self._get_complete_access_ids("read")
if len(access_ids) > 0:
access_result = []
for tuple in result:
if tuple[0] in access_ids:
access_result.append(tuple)
return access_result
return []
#----------------------------------------------------------
# Read, View
# Security
#----------------------------------------------------------
@api.depends('groups')
def _compute_groups(self):
for record in self:
record.complete_groups = record.groups
@api.multi
def check_access_groups(self, operation):
if self.env.user.id == SUPERUSER_ID or isinstance(self.env.uid, NoSecurityUid):
return None
sql_query = '''
SELECT perm_{operation}
FROM {table}_complete_groups_rel r
JOIN muk_security_access_groups g ON r.gid = g.id
JOIN muk_security_access_groups_users_rel u ON r.gid = u.gid
WHERE (r.aid = ANY (VALUES {ids}) AND u.uid = %s)
'''.format(
operation=operation,
table=self._table,
ids=', '.join(map(lambda id: '(%s)' % id, self.ids)),
)
if not self._access_groups_strict:
or_clause = '''
OR NOT EXISTS (
SELECT 1
FROM {table}_complete_groups_rel sr
JOIN muk_security_access_groups sg ON sr.gid = sg.id
WHERE sr.aid = ANY (VALUES {ids})
)
'''.format(
table=self._table,
ids=', '.join(map(lambda id: '(%s)' % id, self.ids))
)
if self._access_groups_mode:
or_clause += 'AND sg.perm_{operation} = true'.format(operation=operation)
sql_query += or_clause
self.env.cr.execute(sql_query, [self.env.user.id])
result = self.env.cr.fetchall()
if len(result) < self.ids or any(list(map(lambda val: val[0], result))):
raise AccessError(_(
'The requested operation cannot be completed due to security restrictions. '
'Please contact your system administrator.\n\n(Document type: %s, Operation: %s)'
) % (self._description, operation))
@api.multi
def filter_access_groups(self, operation):
if self.env.user.id == SUPERUSER_ID or isinstance(self.env.uid, NoSecurityUid):
return self
sql_query = '''
SELECT r.aid
FROM {table}_complete_groups_rel r
JOIN muk_security_access_groups g ON r.gid = g.id
JOIN muk_security_access_groups_users_rel u ON r.gid = u.gid
WHERE (r.aid = ANY (VALUES {ids}) AND u.uid = %s AND g.perm_{operation} = true)
'''.format(
table=self._table,
ids=', '.join(map(lambda id: '(%s)' % id, self.ids)),
operation=operation,
)
if not self._access_groups_strict:
or_clause = '''
OR NOT EXISTS (
SELECT 1
FROM {table}_complete_groups_rel sr
JOIN muk_security_access_groups sg ON sr.gid = sg.id
WHERE sr.aid = ANY (VALUES {ids})
)
'''.format(
table=self._table,
ids=', '.join(map(lambda id: '(%s)' % id, self.ids))
)
if self._access_groups_mode:
or_clause += 'AND sg.perm_{operation} = true'.format(operation=operation)
sql_query += or_clause
self.env.cr.execute(sql_query, [self.env.user.id])
result = self.env.cr.fetchall()
return self.browse(list(map(lambda val: val[0], result)))
#----------------------------------------------------------
# Create, Update, Delete
#----------------------------------------------------------
@api.multi
def _before_write(self, vals, *largs, **kwargs):
def _write(self, vals):
self.check_access_groups('write')
return super(BaseModelAccessGroups, self)._before_write(vals, *largs, **kwargs)
return super(AccessGroupsModel, self)._write(vals)
@api.multi
def _before_unlink(self, *largs, **kwargs):
def unlink(self):
self.check_access_groups('unlink')
return super(BaseModelAccessGroups, self)._before_unlink(*largs, **kwargs)
return super(AccessGroupsModel, self).unlink()
#----------------------------------------------------------
# Groups
#----------------------------------------------------------
@api.depends('groups')
def _compute_groups(self):
for record in self:
record.complete_groups = record.groups

65
muk_security/models/mixins_access.py → muk_security/models/mixins_access_rights.py
View File

@ -27,7 +27,7 @@ _logger = logging.getLogger(__name__)
class AccessModel(models.AbstractModel):
_name = 'muk_security.mixins.access'
_name = 'muk_security.mixins.access_rights'
_description = 'Access Mixin'
#----------------------------------------------------------
@ -53,21 +53,6 @@ class AccessModel(models.AbstractModel):
compute='_compute_permissions_unlink',
search='_search_permission_unlink',
string="Delete Access")
#----------------------------------------------------------
# Function
#----------------------------------------------------------
@api.multi
def check_access(self, operation, raise_exception=False):
try:
access_right = self.check_access_rights(operation, raise_exception)
access_rule = self.check_access_rule(operation) is None
return access_right and access_rule
except AccessError:
if raise_exception:
raise
return False
#----------------------------------------------------------
# Search
@ -75,31 +60,27 @@ class AccessModel(models.AbstractModel):
@api.model
def _search_permission_read(self, operator, operand):
records = self.search([]).filtered(lambda r: r.check_access('read') == True)
if operator == '=' and operand:
return [('id', 'in', records.mapped('id'))]
return [('id', 'not in', records.mapped('id'))]
return [('id', 'in', self._filter_access_ids('read'))]
return [('id', 'not in', self._filter_access_ids('read'))]
@api.model
def _search_permission_create(self, operator, operand):
records = self.search([]).filtered(lambda r: r.check_access('create') == True)
if operator == '=' and operand:
return [('id', 'in', records.mapped('id'))]
return [('id', 'not in', records.mapped('id'))]
return [('id', 'in', self._filter_access_ids('create'))]
return [('id', 'not in', self._filter_access_ids('create'))]
@api.model
def _search_permission_write(self, operator, operand):
records = self.search([]).filtered(lambda r: r.check_access('write') == True)
if operator == '=' and operand:
return [('id', 'in', records.mapped('id'))]
return [('id', 'not in', records.mapped('id'))]
return [('id', 'in', self._filter_access_ids('write'))]
return [('id', 'not in', self._filter_access_ids('write'))]
@api.model
def _search_permission_unlink(self, operator, operand):
records = self.search([]).filtered(lambda r: r.check_access('unlink') == True)
if operator == '=' and operand:
return [('id', 'in', records.mapped('id'))]
return [('id', 'not in', records.mapped('id'))]
return [('id', 'in', self._filter_access_ids('unlink'))]
return [('id', 'not in', self._filter_access_ids('unlink'))]
#----------------------------------------------------------
# Read, View
@ -107,20 +88,32 @@ class AccessModel(models.AbstractModel):
@api.multi
def _compute_permissions_read(self):
for record in self:
record.update({'permission_read': record.check_access('read')})
records = self._filter_access('read')
for record in records:
record.update({'permission_read': True})
for record in self - records:
record.update({'permission_read': False})
@api.multi
def _compute_permissions_create(self):
for record in self:
record.update({'permission_create': record.check_access('create')})
records = self._filter_access('create')
for record in records:
record.update({'permission_create': True})
for record in no_records:
record.update({'permission_create': False})
@api.multi
def _compute_permissions_write(self):
for record in self:
record.update({'permission_write': record.check_access('write')})
records = self._filter_access('write')
for record in records:
record.update({'permission_write': True})
for record in no_records:
record.update({'permission_write': False})
@api.multi
def _compute_permissions_unlink(self):
for record in self:
record.update({'permission_unlink': record.check_access('unlink')})
records = self._filter_access('unlink')
for record in records:
record.update({'permission_unlink': True})
for record in no_records:
record.update({'permission_unlink': False})

8
muk_security/models/mixins_locking.py
View File

@ -26,6 +26,8 @@ from odoo import _, SUPERUSER_ID
from odoo import models, api, fields
from odoo.exceptions import AccessError
from odoo.addons.muk_security.tools.security import NoSecurityUid
_logger = logging.getLogger(__name__)
class LockingModel(models.AbstractModel):
@ -60,11 +62,15 @@ class LockingModel(models.AbstractModel):
@api.multi
def unlock(self):
self.write({'locked_by': None})
@api.model
def _check_lock_user(self, uid):
return uid in (self.env.uid, SUPERUSER_ID) or isinstance(self.env.uid, NoSecurityUid)
@api.multi
def check_lock(self, *largs, **kwargs):
for record in self:
if record.locked_by.exists() and not record.locked_by.id in (self.env.uid, SUPERUSER_ID):
if record.locked_by.exists() and not self._check_lock_user(record.locked_by.id):
raise AccessError(_("The record (%s [%s]) is locked, by an other user.") % (record._description, record.id))
#----------------------------------------------------------

8
muk_security/models/res_users.py
View File

@ -37,7 +37,11 @@ class AccessUser(models.Model):
@classmethod
def _browse(cls, ids, env, prefetch=None):
return super(AccessUser, cls)._browse([
access_ids = [
id if not isinstance(id, NoSecurityUid)
else super(NoSecurityUid, id).__int__()
for id in ids], env, prefetch=prefetch)
for id in ids
]
return super(AccessUser, cls)._browse(
access_ids, env, prefetch=prefetch
)
Write Preview
Loading…
Cancel
Save