Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 31 additions & 31 deletions dms/models/directory.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@
import base64
import logging
import os
from ast import literal_eval
from collections import defaultdict
from typing import Literal # noqa # pylint: disable=unused-import

Expand Down Expand Up @@ -88,7 +87,7 @@ def _default_parent_id(self):
if context.get("active_model") == self._name and context.get("active_id"):
return context["active_id"]
else:
return False
return context.get("dms_parent_id")

group_ids = fields.Many2many(
comodel_name="dms.access.group",
Expand Down Expand Up @@ -380,8 +379,8 @@ def _search_panel_directory(self, **kwargs):
@api.model
def _search_starred(self, operator, operand):
if operator in ("=", "in") and operand:
return [("user_star_ids", "in", [self.env.uid])]
return [("user_star_ids", "not in", [self.env.uid])]
return Domain([("user_star_ids", "in", [self.env.uid])])
return Domain([("user_star_ids", "not in", [self.env.uid])])

@api.depends("name", "parent_id.complete_name")
def _compute_complete_name(self):
Expand Down Expand Up @@ -490,8 +489,12 @@ def _compute_parent_id(self):
if record.is_root_directory:
record.parent_id = None
else:
# HACK: Not needed in v14 due to odoo/odoo#64359
record.parent_id = record.parent_id
record.parent_id = (
record.parent_id
or record._origin.parent_id
or self.env.context.get("default_parent_id")
or self.env.context.get("dms_parent_id")
)

@api.depends("is_root_directory", "parent_id")
def _compute_root_id(self):
Expand Down Expand Up @@ -753,37 +756,34 @@ def _search_panel_domain_image(
def action_dms_directories_all_directory(self):
self.ensure_one()
action = self.env["ir.actions.act_window"]._for_xml_id(
"dms.action_dms_directory"
)
domain = Domain.AND(
[
literal_eval(action["domain"].strip()),
[("parent_id", "child_of", self.id)],
]
"dms.action_dms_directories_all_directory"
)
action["display_name"] = self.name
action["domain"] = domain
action["context"] = dict(
self.env.context,
default_parent_id=self.id,
searchpanel_default_parent_id=self.id,
)
action["domain"] = [
("parent_id", "child_of", self.id),
("is_hidden", "=", False),
("id", "!=", self.id),
]
action["context"] = {
"default_parent_id": self.id,
"dms_parent_id": self.id,
"searchpanel_default_parent_id": self.id,
}
return action

def action_dms_files_all_directory(self):
self.ensure_one()
action = self.env["ir.actions.act_window"]._for_xml_id("dms.action_dms_file")
domain = Domain.AND(
[
literal_eval(action["domain"].strip()),
[("directory_id", "child_of", self.id)],
]
action = self.env["ir.actions.act_window"]._for_xml_id(
"dms.action_dms_files_all_directory"
)
action["display_name"] = self.name
action["domain"] = domain
action["context"] = dict(
self.env.context,
default_directory_id=self.id,
searchpanel_default_directory_id=self.id,
)
action["domain"] = [
("directory_id", "child_of", self.id),
("is_hidden", "=", False),
]
action["context"] = {
"default_directory_id": self.id,
"dms_directory_id": self.id,
"searchpanel_default_directory_id": self.id,
}
return action
3 changes: 3 additions & 0 deletions dms/models/dms_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ class DMSFile(models.Model):
required=True,
index="btree",
tracking=True, # Leave log if "moved" to another directory
default=lambda self: self.env.context.get("dms_directory_id"),
)
root_directory_id = fields.Many2one(related="directory_id.root_directory_id")
# Override acording to defined in AbstractDmsMixin
Expand Down Expand Up @@ -568,6 +569,8 @@ def _create_model_attachment(self, vals):
directory_id = self.env.context.get("active_id")
elif self.env.context.get("default_directory_id"):
directory_id = self.env.context.get("default_directory_id")
elif self.env.context.get("dms_directory_id"):
directory_id = self.env.context.get("dms_directory_id")
directory = self.env["dms.directory"].browse(directory_id)
if (
directory.res_model
Expand Down
142 changes: 67 additions & 75 deletions dms/models/dms_security_mixin.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,10 @@
# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).


import functools
from logging import getLogger

from odoo import SUPERUSER_ID, api, fields, models
from odoo import api, fields, models
from odoo.exceptions import AccessError
from odoo.fields import Domain
from odoo.tools import SQL
Expand Down Expand Up @@ -197,34 +198,25 @@ def _get_domain_by_access_groups(self, operation):
return result

@api.model
def _get_permission_domain(self, operator, value, operation):
"""Abstract logic for searching computed permission fields."""
_self = self
# HACK ir.rule domains are evaluated in superuser mode while env.uid
# stays the acting user, so `su` together with a non-root uid means we
# are resolving the `permission_<op> = user.id` rule on that user's
# behalf. The Domain engine coerces that sentinel to this Boolean
# field's type before we get here, so we rely on env.uid (used by
# _get_access_groups_query) rather than the value to build the domain.
if self.env.su and self.env.uid != SUPERUSER_ID:
_self = self.sudo(False)
value = bool(value)
# Tricky one, to know if you want to search
# positive or negative access
positive = (operator not in Domain.NEGATIVE_OPERATORS) == bool(value)
if _self.env.su:
# You're SUPERUSER_ID
return Domain.TRUE if positive else Domain.FALSE

result = Domain.OR(
def _get_dms_access_domain(self, operation):
"""Domain matching the records the current user may access for
``operation`` through DMS access groups or inheritance."""
return Domain.OR(
[
_self._get_domain_by_access_groups(operation),
_self._get_domain_by_inheritance(operation),
self._get_domain_by_access_groups(operation),
self._get_domain_by_inheritance(operation),
]
)
if not positive:
result = ~Domain(result)
return result

@api.model
def _get_permission_domain(self, operator, value, operation):
"""Search implementation of the computed ``permission_<op>`` fields,
used by field domains (e.g. ``directory_id``'s create filter)."""
positive = (operator not in Domain.NEGATIVE_OPERATORS) == bool(value)
if self.env.su:
return Domain.TRUE if positive else Domain.FALSE
result = self._get_dms_access_domain(operation)
return result if positive else ~result

@api.model
def _search_permission_create(self, operator, value):
Expand All @@ -242,49 +234,57 @@ def _search_permission_unlink(self, operator, value):
def _search_permission_write(self, operator, value):
return self._get_permission_domain(operator, value, "write")

def filtered_domain(self, domain):
"""This method is needed to inhibit the behavior when called from the
_check_access() method with sudo() https://github.com/odoo/odoo/blob/fc737a147b9aefbd6ae5d111835ce3f4f7b4240a/odoo/models.py#L4465.
It would cause the error that multiple records are not accessed to be
displayed.
The _filtered_access() method is also overwritten to prevent this sudo()
specific behavior and to be able to access only the appropriate records.
"""
if self.env.su:
return self
return super().filtered_domain(domain)

def _filtered_access_no_recursion(self, operation: str):
"""This method is just the same as _filtered_access
but it can not be called withoud super due to
recursion error.
"""
if self and not self.env.su and (result := self._check_access(operation)):
return self - result[0]
return self
def _search(
self,
domain,
offset=0,
limit=None,
order=None,
*,
bypass_access=False,
**kwargs,
):
"""Restrict searches to the records the current user may read through
DMS access groups or inheritance (mirrors ``mail.message._search``)."""
if self.env.su or bypass_access:
return super()._search(
domain, offset, limit, order, bypass_access=bypass_access, **kwargs
)
domain = Domain.AND([Domain(domain), self._get_dms_access_domain("read")])
return super()._search(domain, offset, limit, order, **kwargs)

def _filtered_access(self, operation):
# Only kept to not break inheritance; see next comment
result = super()._filtered_access(operation)
# HACK Always fall back to applying rules by SQL.
# Upstream `_filtered_access()` doesn't use computed fields
# search methods. Thus, it will take the `[('permission_{operation}',
# '=', user.id)]` rule literally. Obviously that will always fail
# because `self[f"permission_{operation}"]` will always be a `bool`,
# while `user.id` will always be an `int`.
result |= self._filtered_access_no_recursion(operation)
def _check_access(self, operation):
"""Add the DMS access-group / inheritance restriction to the
record-level access check (mirrors ``mail.message._check_access``)."""
result = super()._check_access(operation)
if self.env.su or not any(self._ids):
return result
records = self - result[0] if result else self
forbidden = records._get_forbidden_dms_access(operation)
if forbidden:
if result:
return result[0] + forbidden, result[1]
Rule = self.env["ir.rule"]
return forbidden, functools.partial(
Rule._make_access_error, operation, forbidden
)
return result

def _check_access_dms_record(self, operation: str) -> tuple | None:
"""Specific method "similar" to _check_access() but with a different
behavior: check if you do not really have access to any of the records
in to avoid performing the corresponding create/write/unlink action."""
if any(self._ids) and not self.env.su:
Rule = self.env["ir.rule"]
domain = Rule._compute_domain(self._name, operation)
items = self.with_context(active_test=False).search(domain)
if any(x_id not in items.ids for x_id in self.ids):
raise Rule._make_access_error(operation, (self - items))
def _get_forbidden_dms_access(self, operation):
"""Return the subset of ``self`` the current user cannot access for
``operation`` under the DMS access-group / inheritance rules. The
domain carries an SQL sub-query, so it is resolved by search (not by
``filtered_domain``), bypassing access to evaluate exactly this domain."""
domain = Domain.AND(
[
Domain([("id", "in", self.ids)]),
self._get_dms_access_domain(operation),
]
)
allowed = self.browse(
self.with_context(active_test=False)._search(domain, bypass_access=True)
)
return self - allowed

@api.model_create_multi
def create(self, vals_list):
Expand All @@ -296,13 +296,5 @@ def create(self, vals_list):
res.flush_recordset()
# Go back to the original sudo state and check we really had creation permission
res = res.sudo(self.env.su)
res._check_access_dms_record("create")
res.check_access("create")
return res

def write(self, vals):
self._check_access_dms_record("write")
return super().write(vals)

def unlink(self):
self._check_access_dms_record("unlink")
return super().unlink()
3 changes: 2 additions & 1 deletion dms/models/storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

from odoo import api, fields, models
from odoo.exceptions import AccessError
from odoo.fields import Domain

_logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -88,7 +89,7 @@ class Storage(models.Model):

def _search_model(self, operator, value):
allowed_items = self.env["ir.model"].sudo().search([("model", operator, value)])
return [("model_ids", "in", allowed_items.ids)]
return Domain([("model_ids", "in", allowed_items.ids)])

@api.onchange("save_type")
def _onchange_save_type(self):
Expand Down
81 changes: 0 additions & 81 deletions dms/security/security.xml
Original file line number Diff line number Diff line change
Expand Up @@ -111,85 +111,4 @@
<field name="perm_unlink" eval="1" />
<field name="domain_force">[('is_hidden', '=', True)]</field>
</record>
<!-- These rules leverage computed permission management -->
<record id="rule_directory_computed_create" model="ir.rule">
<field name="name">Apply computed create permissions.</field>
<field name="model_id" ref="model_dms_directory" />
<field name="global" eval="True" />
<field name="perm_read" eval="0" />
<field name="perm_create" eval="1" />
<field name="perm_write" eval="0" />
<field name="perm_unlink" eval="0" />
<field name="domain_force">[('permission_create', '=', user.id)]</field>
</record>
<record id="rule_directory_computed_read" model="ir.rule">
<field name="name">Apply computed read permissions.</field>
<field name="model_id" ref="model_dms_directory" />
<field name="global" eval="True" />
<field name="perm_read" eval="1" />
<field name="perm_create" eval="0" />
<field name="perm_write" eval="0" />
<field name="perm_unlink" eval="0" />
<field name="domain_force">[('permission_read', '=', user.id)]</field>
</record>
<record id="rule_directory_computed_unlink" model="ir.rule">
<field name="name">Apply computed unlink permissions.</field>
<field name="model_id" ref="model_dms_directory" />
<field name="global" eval="True" />
<field name="perm_read" eval="0" />
<field name="perm_create" eval="0" />
<field name="perm_write" eval="0" />
<field name="perm_unlink" eval="1" />
<field name="domain_force">[('permission_unlink', '=', user.id)]</field>
</record>
<record id="rule_directory_computed_write" model="ir.rule">
<field name="name">Apply computed write permissions.</field>
<field name="model_id" ref="model_dms_directory" />
<field name="global" eval="True" />
<field name="perm_read" eval="0" />
<field name="perm_create" eval="0" />
<field name="perm_write" eval="1" />
<field name="perm_unlink" eval="0" />
<field name="domain_force">[('permission_write', '=', user.id)]</field>
</record>
<record id="rule_file_computed_create" model="ir.rule">
<field name="name">Apply computed create permissions.</field>
<field name="model_id" ref="model_dms_file" />
<field name="global" eval="True" />
<field name="perm_read" eval="0" />
<field name="perm_create" eval="1" />
<field name="perm_write" eval="0" />
<field name="perm_unlink" eval="0" />
<field name="domain_force">[('permission_create', '=', user.id)]</field>
</record>
<record id="rule_file_computed_read" model="ir.rule">
<field name="name">Apply computed read permissions.</field>
<field name="model_id" ref="model_dms_file" />
<field name="global" eval="True" />
<field name="perm_read" eval="1" />
<field name="perm_create" eval="0" />
<field name="perm_write" eval="0" />
<field name="perm_unlink" eval="0" />
<field name="domain_force">[('permission_read', '=', user.id)]</field>
</record>
<record id="rule_file_computed_unlink" model="ir.rule">
<field name="name">Apply computed unlink permissions.</field>
<field name="model_id" ref="model_dms_file" />
<field name="global" eval="True" />
<field name="perm_read" eval="0" />
<field name="perm_create" eval="0" />
<field name="perm_write" eval="0" />
<field name="perm_unlink" eval="1" />
<field name="domain_force">[('permission_unlink', '=', user.id)]</field>
</record>
<record id="rule_file_computed_write" model="ir.rule">
<field name="name">Apply computed write permissions.</field>
<field name="model_id" ref="model_dms_file" />
<field name="global" eval="True" />
<field name="perm_read" eval="0" />
<field name="perm_create" eval="0" />
<field name="perm_write" eval="1" />
<field name="perm_unlink" eval="0" />
<field name="domain_force">[('permission_write', '=', user.id)]</field>
</record>
</odoo>
Loading
Loading