m

addons/mail/models/ir_attachment.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 # Part of Odoo. See LICENSE file for full copyright and licensing details.
 
-from odoo import api, fields, models
+from odoo import api, models
 
 
 class IrAttachment(models.Model):

addons/mail/models/mail_thread.py

@@ -109,7 +109,7 @@ class MailThread(models.AbstractModel):
     message_has_error_counter = fields.Integer(
         'Number of error', compute='_compute_message_has_error',
         help="Number of messages with delivery error")
-    message_attachment_count = fields.Integer('Attachment Count', compute='_compute_message_attachment_count')
+    message_attachment_count = fields.Integer('Attachment Count', compute='_compute_message_attachment_count', groups="base.group_user")
     message_main_attachment_id = fields.Many2one(string="Main Attachment", comodel_name='ir.attachment', index=True, copy=False)
 
     @api.one

addons/test_mail/tests/test_odoobot.py

@@ -61,7 +61,12 @@ def test_onboarding_flow(self):
             answer=("attachment",)
         )
         kwargs['body'] = ''
-        kwargs['attachment_ids'] = [1]
+        attachment = self.env['ir.attachment'].sudo(self.user_employee).create({
+            'datas': 'bWlncmF0aW9uIHRlc3Q=',
+            'datas_fname': 'catalog.doc',
+            'name': 'picture_of_your_dog.doc',
+        })
+        kwargs['attachment_ids'] = [attachment.id]
         last_message = self.assertNextMessage(
             channel.message_post(**kwargs),
             sender=self.odoobot,

addons/website_sale_digital/models/product.py

@@ -41,11 +41,11 @@ class Product(models.Model):
 
     @api.multi
     def _compute_attachment_count(self):
-        IrAttachment = self.env['ir.attachment']
         for product in self:
-            prod_tmpl_attach_count = IrAttachment.search_count([('res_model', '=', 'product.template'), ('res_id', 'in', product.product_tmpl_id.ids), ('product_downloadable', '=', True)])
-            prod_attach_count = IrAttachment.search_count([('res_model', '=', 'product.product'), ('res_id', 'in', product.ids), ('product_downloadable', '=', True)])
-            product.attachment_count = prod_tmpl_attach_count + prod_attach_count
+            product.attachment_count = self.env['ir.attachment'].search_count([
+                '|',
+                ('res_model', '=', 'product.template'), ('res_id', '=', product.product_tmpl_id.id), ('product_downloadable', '=', True),
+                ('res_model', '=', 'product.product'), ('res_id', '=', product.id), ('product_downloadable', '=', True)])
 
     @api.multi
     def action_open_attachments(self):

odoo/addons/base/models/ir_attachment.py

@@ -330,8 +330,10 @@ def check(self, mode, values=None):
         model_ids = defaultdict(set)            # {model_name: set(ids)}
         require_employee = False
         if self:
-            self._cr.execute('SELECT res_model, res_id, create_uid, public FROM ir_attachment WHERE id IN %s', [tuple(self.ids)])
-            for res_model, res_id, create_uid, public in self._cr.fetchall():
+            self._cr.execute('SELECT res_model, res_id, create_uid, public, res_field FROM ir_attachment WHERE id IN %s', [tuple(self.ids)])
+            for res_model, res_id, create_uid, public, res_field in self._cr.fetchall():
+                if self.env.user.id != SUPERUSER_ID and not self.env.user._is_system() and res_field:
+                    raise AccessError(_("Sorry, you are not allowed to access this document."))
                 if public and mode == 'read':
                     continue
                 if not (res_model and res_id):
@@ -367,14 +369,24 @@ def check(self, mode, values=None):
             if not (self.env.user._is_admin() or self.env.user.has_group('base.group_user')):
                 raise AccessError(_("Sorry, you are not allowed to access this document."))
 
+    def _read_group_allowed_fields(self):
+        return ['type', 'company_id', 'res_id', 'create_date', 'create_uid', 'res_name', 'name', 'mimetype', 'id', 'url', 'datas_fname', 'res_field', 'res_model']
+
     @api.model
     def read_group(self, domain, fields, groupby, offset=0, limit=None, orderby=False, lazy=True):
         """Override read_group to add res_field=False in domain if not present."""
+        if not fields:
+            raise AccessError(_("Sorry, you must provide fields to read on attachments"))
+        if any('(' in field for field in fields + groupby):
+            raise AccessError(_("Sorry, the syntax 'name:agg(field)' is not available for attachments"))
         if not any(item[0] in ('id', 'res_field') for item in domain):
             domain.insert(0, ('res_field', '=', False))
-        return super(IrAttachment, self).read_group(domain, fields, groupby,
-                                                    offset=offset, limit=limit,
-                                                    orderby=orderby, lazy=lazy)
+        groupby = [groupby] if isinstance(groupby, str) else groupby
+        allowed_fields = self._read_group_allowed_fields()
+        fields_set = set(field.split(':')[0] for field in fields + groupby)
+        if not self.env.user._is_system() and (not fields or fields_set.difference(allowed_fields)):
+            raise AccessError(_("Sorry, you are not allowed to access these fields on attachments."))
+        return super().read_group(domain, fields, groupby, offset=offset, limit=limit, orderby=orderby, lazy=lazy)
 
     @api.model
     def _search(self, args, offset=0, limit=None, order=None, count=False, access_rights_uid=None):
@@ -386,7 +398,7 @@ def _search(self, args, offset=0, limit=None, order=None, count=False, access_ri
         ids = super(IrAttachment, self)._search(args, offset=offset, limit=limit, order=order,
                                                 count=False, access_rights_uid=access_rights_uid)
 
-        if self._uid == SUPERUSER_ID:
+        if self.env.user._is_system():
             # rules do not apply for the superuser
             return len(ids) if count else ids
 
@@ -404,12 +416,19 @@ def _search(self, args, offset=0, limit=None, order=None, count=False, access_ri
         # Use pure SQL rather than read() as it is about 50% faster for large dbs (100k+ docs),
         # and the permissions are checked in super() and below anyway.
         model_attachments = defaultdict(lambda: defaultdict(set))   # {res_model: {res_id: set(ids)}}
-        self._cr.execute("""SELECT id, res_model, res_id, public FROM ir_attachment WHERE id IN %s""", [tuple(ids)])
+        binary_fields_attachments = set()
+        self._cr.execute("""SELECT id, res_model, res_id, public, res_field FROM ir_attachment WHERE id IN %s""", [tuple(ids)])
         for row in self._cr.dictfetchall():
             if not row['res_model'] or row['public']:
                 continue
             # model_attachments = {res_model: {res_id: set(ids)}}
             model_attachments[row['res_model']][row['res_id']].add(row['id'])
+            # Should not retrieve binary fields attachments
+            if row['res_field']:
+                binary_fields_attachments.add(row['id'])
+
+        if binary_fields_attachments:
+            ids.difference_update(binary_fields_attachments)
 
         # To avoid multiple queries for each attachment found, checks are
         # performed in batch as much as possible.

odoo/addons/base/models/ir_http.py

@@ -260,11 +260,19 @@ def _get_record_and_check(self, xmlid=None, model=None, id=None, field='datas',
         if not record or not record.exists() or field not in record:
             return None, 404
 
-        # access token grant access
-        if model == 'ir.attachment' and access_token:
-            record = record.sudo()
-            if not consteq(record.access_token or '', access_token):
+        if model == 'ir.attachment':
+            record_sudo = record.sudo()
+            if access_token and not consteq(record_sudo.access_token or '', access_token):
                 return None, 403
+            elif (access_token and consteq(record_sudo.access_token or '', access_token)):
+                record = record_sudo
+            elif record_sudo.public:
+                record = record_sudo
+            elif self.env.user.has_group('base.group_portal'):
+                # Check the read access on the record linked to the attachment
+                # eg: Allow to download an attachment on a task from /my/task/task_id 
+                record.check('read')
+                record = record_sudo
 
         # check read access
         try:
@@ -316,7 +324,7 @@ def _binary_record_content(
 
         field_def = record._fields[field]
         if field_def.type == 'binary' and field_def.attachment:
-            field_attachment = self.env['ir.attachment'].search_read(domain=[('res_model', '=', model), ('res_id', '=', record.id), ('res_field', '=', field)], fields=['datas', 'mimetype', 'checksum'], limit=1)
+            field_attachment = self.env['ir.attachment'].sudo().search_read(domain=[('res_model', '=', model), ('res_id', '=', record.id), ('res_field', '=', field)], fields=['datas', 'mimetype', 'checksum'], limit=1)
             if field_attachment:
                 mimetype = field_attachment[0]['mimetype']
                 content = field_attachment[0]['datas']

odoo/addons/base/security/ir.model.access.csv

@@ -1,7 +1,6 @@
 "id","name","model_id:id","group_id:id","perm_read","perm_write","perm_create","perm_unlink"
-"access_ir_attachment_all","ir_attachment all","model_ir_attachment",,1,0,0,0
 "access_ir_attachment_group_user","ir_attachment group_user","model_ir_attachment","group_user",1,1,1,1
-"access_ir_attachment_portal","ir.attachment.portal","model_ir_attachment","group_portal",1,0,1,0
+"access_ir_attachment_group_portal_public","ir_attachment group_portal_public","model_ir_attachment",,0,0,0,0
 "access_ir_cron_group_cron","ir_cron group_cron","model_ir_cron","group_system",1,1,1,1
 "access_ir_exports_group_system","ir_exports group_system","model_ir_exports","base.group_user",1,1,1,1
 "access_ir_exports_line_group_system","ir_exports_line group_system","model_ir_exports_line","base.group_user",1,1,1,1

odoo/models.py

@@ -3189,7 +3189,7 @@ def unlink(self):
                 cr.execute(query, (self._name, sub_ids))
                 attachments = Attachment.browse([row[0] for row in cr.fetchall()])
                 if attachments:
-                    attachments.unlink()
+                    attachments.sudo().unlink()
 
                 if any(field.translate for field in self._fields.values()):
                     # For the same reason, remove the relevant records in ir_translation