Skip to content
Merged
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
1 change: 1 addition & 0 deletions checklog-odoo.cfg
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
[checklog-odoo]
ignore=
WARNING.* 0 failed, 0 error\(s\).*
WARNING.*DeprecationWarning: PyUnicode_FromUnicode\(NULL, size\) is deprecated; use PyUnicode_New\(\) instead
1 change: 1 addition & 0 deletions storage_file/__manifest__.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,5 +23,6 @@
"data/ir_cron.xml",
"data/storage_backend.xml",
"wizards/swap_backend.xml",
"data/ir_config_parameter.xml",
],
}
11 changes: 11 additions & 0 deletions storage_file/controllers/main.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.

from werkzeug.exceptions import NotFound

from odoo import http
from odoo.exceptions import AccessError
from odoo.http import request


Expand All @@ -13,6 +15,15 @@ def content_common(self, slug_name_with_id, token=None, download=None, **kw):
storage_file = request.env["storage.file"].get_from_slug_name_with_id(
slug_name_with_id
)
if not storage_file.exists():
raise NotFound()
try:
storage_file.check_access("read")
except AccessError as err:
# If you don't have access you should not know
# that the file exists (as anon user).
# You can inspect the traceback to see it's coming from an access error.
raise NotFound() from err
Comment thread
simahawk marked this conversation as resolved.
stream = request.env["ir.binary"]._get_stream_from(
storage_file, field_name="data"
)
Expand Down
7 changes: 7 additions & 0 deletions storage_file/data/ir_config_parameter.xml
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
<?xml version="1.0" encoding="utf-8" ?>
<odoo noupdate="1">
<record id="storage_file_backend" model="ir.config_parameter">
<field name="key">storage.file.backend_id</field>
<field name="value" ref="storage_backend.default_storage_backend" />
</record>
</odoo>
5 changes: 4 additions & 1 deletion storage_file/models/storage_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,10 @@ def _get_base_url_from_param(self):
def _get_url_for_file(self, storage_file, exclude_base_url=False):
"""Return final full URL for given file."""
backend = self.sudo()
if backend.served_by == "odoo":
Comment thread
hparfr marked this conversation as resolved.
# Make sure that no matter if you have a CDN URL or not,
# you can always access the file via Odoo.
force_serve_via_odoo = backend.served_by == "external" and not backend.base_url
if backend.served_by == "odoo" or force_serve_via_odoo:
parts = [
self._get_base_url_from_param() if not exclude_base_url else "/",
"storage.file",
Expand Down
35 changes: 34 additions & 1 deletion storage_file/models/storage_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,11 @@ class StorageFile(models.Model):

name = fields.Char(required=True, index=True)
backend_id = fields.Many2one(
"storage.backend", "Storage", index=True, required=True
"storage.backend",
"Storage",
index=True,
required=True,
default=lambda self: self._get_default_backend_id(),
)
url = fields.Char(compute="_compute_url", help="HTTP accessible path to the file")
url_path = fields.Char(
Expand Down Expand Up @@ -65,6 +69,33 @@ class StorageFile(models.Model):
"res.company", "Company", default=lambda self: self.env.user.company_id.id
)
file_type = fields.Selection([])
is_public = fields.Boolean(
compute="_compute_is_public",
compute_sudo=True,
search="_search_is_public",
# Not stored to avoid massive recomputes when the backend flag changes.
help="Reflects the `is_public` flag of the related backend.",
)

@api.depends("backend_id.is_public")
def _compute_is_public(self):
for rec in self:
rec.is_public = rec.backend_id.is_public

@api.model
def _get_default_backend_id(self):
return self.env["storage.backend"]._get_backend_id_from_param(
self.env, "storage.file.backend_id"
)

def _search_is_public(self, operator, value):
# Look up matching backends with sudo so that users with limited ACL
# on `storage.backend` can still filter their accessible files by
# public flag.
backends = (
self.env["storage.backend"].sudo().search([("is_public", operator, value)])
)
return [("backend_id", "in", backends.ids)]

_sql_constraints = [
(
Expand Down Expand Up @@ -175,6 +206,8 @@ def _get_url(self, exclude_base_url=False):

:param exclude_base_url: skip base_url
"""
if not self.backend_id or not self.relative_path:
return ""
return self.backend_id._get_url_for_file(
self, exclude_base_url=exclude_base_url
)
Expand Down
1 change: 1 addition & 0 deletions storage_file/security/ir.model.access.csv
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink
access_storage_file_edit,storage_file edit,model_storage_file,base.group_system,1,1,1,1
access_storage_file_read_public,storage_file public read,model_storage_file,base.group_user,1,0,0,0
access_storage_file_read_portal,storage_file portal read,model_storage_file,base.group_public,1,0,0,0
access_storage_file_replace,storage_file_replace public,model_storage_file_replace,base.group_user,1,1,1,1
access_storage_file_swap_backend,storage_file_swap_backend admin,model_storage_file_swap_backend,base.group_system,1,1,1,1
11 changes: 11 additions & 0 deletions storage_file/security/storage_file.xml
Original file line number Diff line number Diff line change
Expand Up @@ -13,4 +13,15 @@
<field name="perm_create" eval="False" />
<field name="perm_unlink" eval="False" />
</record>
<!--Internal users can read all files (public and private)-->
<record id="ir_rule_storage_file_internal" model="ir.rule">
<field name="name">Storage file internal read all</field>
<field name="model_id" ref="model_storage_file" />
<field name="groups" eval="[(4, ref('base.group_user'))]" />
<field name="domain_force">[(1, '=', 1)]</field>
<field name="perm_read" eval="True" />
<field name="perm_write" eval="False" />
<field name="perm_create" eval="False" />
<field name="perm_unlink" eval="False" />
</record>
</odoo>
2 changes: 2 additions & 0 deletions storage_file/tests/__init__.py
Original file line number Diff line number Diff line change
@@ -1,2 +1,4 @@
from . import test_storage_file
from . import test_swap_backend
from . import test_is_public
from . import test_controller
156 changes: 156 additions & 0 deletions storage_file/tests/test_controller.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,156 @@
# Copyright 2026 Camptocamp SA
# @author Simone Orsi <simone.orsi@camptocamp.com>
# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).

import base64

from odoo.tests.common import HttpCase, tagged


@tagged("-at_install", "post_install")
class TestStorageFileController(HttpCase):
"""Test the /storage.file/ controller with public/private and odoo/external."""

@classmethod
def setUpClass(cls):
super().setUpClass()
cls.env = cls.env(context=dict(cls.env.context, tracking_disable=True))
cls.data = b"Hello, storage!"
cls.filedata = base64.b64encode(cls.data)
cls.backend_odoo_public = cls.env["storage.backend"].create(
{
"name": "Odoo Public",
"backend_type": "filesystem",
"served_by": "odoo",
"is_public": True,
"filename_strategy": "name_with_id",
}
)
cls.backend_odoo_private = cls.env["storage.backend"].create(
{
"name": "Odoo Private",
"backend_type": "filesystem",
"served_by": "odoo",
"is_public": False,
"filename_strategy": "name_with_id",
}
)
cls.backend_ext_public = cls.env["storage.backend"].create(
{
"name": "Ext Public (no CDN)",
"backend_type": "filesystem",
"served_by": "external",
"base_url": "",
"is_public": True,
"filename_strategy": "name_with_id",
}
)
cls.backend_ext_private = cls.env["storage.backend"].create(
{
"name": "Ext Private (no CDN)",
"backend_type": "filesystem",
"served_by": "external",
"base_url": "",
"is_public": False,
"filename_strategy": "name_with_id",
}
)
cls.file_odoo_public = cls.env["storage.file"].create(
{
"name": "pub-odoo.txt",
"backend_id": cls.backend_odoo_public.id,
"data": cls.filedata,
}
)
cls.file_odoo_private = cls.env["storage.file"].create(
{
"name": "priv-odoo.txt",
"backend_id": cls.backend_odoo_private.id,
"data": cls.filedata,
}
)
cls.file_ext_public = cls.env["storage.file"].create(
{
"name": "pub-ext.txt",
"backend_id": cls.backend_ext_public.id,
"data": cls.filedata,
}
)
cls.file_ext_private = cls.env["storage.file"].create(
{
"name": "priv-ext.txt",
"backend_id": cls.backend_ext_private.id,
"data": cls.filedata,
}
)
cls.internal_user = (
cls.env["res.users"]
.with_context(no_reset_password=True)
.create(
{
"name": "Storage Test User",
"login": "storage_test_user",
"password": "storage_test_user",
"groups_id": [
(4, cls.env.ref("base.group_user").id),
],
}
)
)

def _url_for(self, storage_file):
return f"/storage.file/{storage_file.slug}"

# ---- Public user (anonymous) ----

def test_public_user_odoo_public(self):
"""Public user + public odoo backend -> 200."""
resp = self.url_open(self._url_for(self.file_odoo_public))
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp.content, self.data)

def test_public_user_odoo_private(self):
"""Public user + private odoo backend -> 404."""
resp = self.url_open(self._url_for(self.file_odoo_private))
self.assertEqual(resp.status_code, 404)

def test_public_user_ext_public(self):
"""Public user + public external backend (no CDN) -> 200."""
resp = self.url_open(self._url_for(self.file_ext_public))
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp.content, self.data)

def test_public_user_ext_private(self):
"""Public user + private external backend (no CDN) -> 404."""
resp = self.url_open(self._url_for(self.file_ext_private))
self.assertEqual(resp.status_code, 404)

# ---- Internal (authenticated) user ----

def test_internal_user_odoo_public(self):
"""Internal user + public odoo backend -> 200."""
self.authenticate("storage_test_user", "storage_test_user")
resp = self.url_open(self._url_for(self.file_odoo_public))
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp.content, self.data)

def test_internal_user_odoo_private(self):
"""Internal user + private odoo backend -> 200."""
self.authenticate("storage_test_user", "storage_test_user")
resp = self.url_open(self._url_for(self.file_odoo_private))
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp.content, self.data)

def test_internal_user_ext_public(self):
"""Internal user + public external backend (no CDN) -> 200."""
self.authenticate("storage_test_user", "storage_test_user")
resp = self.url_open(self._url_for(self.file_ext_public))
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp.content, self.data)

def test_internal_user_ext_private(self):
"""Internal user + private external backend (no CDN) -> 200."""
self.authenticate("storage_test_user", "storage_test_user")
resp = self.url_open(self._url_for(self.file_ext_private))
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp.content, self.data)
59 changes: 59 additions & 0 deletions storage_file/tests/test_is_public.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
# Copyright 2026 Camptocamp SA
# @author Simone Orsi <simone.orsi@camptocamp.com>
# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).

from odoo.addons.component.tests.common import TransactionComponentCase


class TestStorageFileIsPublic(TransactionComponentCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.backend = cls.env["storage.backend"].create(
{"name": "Test backend", "backend_type": "filesystem"}
)
cls.storage_file = cls.env["storage.file"].create(
{
"name": "test-public.txt",
"backend_id": cls.backend.id,
"data": b"aGVsbG8=", # "hello" base64
}
)

def test_reflects_backend_flag(self):
self.assertFalse(self.storage_file.is_public)
self.backend.is_public = True
self.assertTrue(self.storage_file.is_public)

def test_search_true(self):
self.backend.is_public = True
result = self.env["storage.file"].search(
[("is_public", "=", True), ("id", "=", self.storage_file.id)]
)
self.assertIn(self.storage_file, result)
Comment thread
simahawk marked this conversation as resolved.

def test_search_false(self):
self.backend.is_public = False
result = self.env["storage.file"].search(
[("is_public", "=", False), ("id", "=", self.storage_file.id)]
)
self.assertIn(self.storage_file, result)
Comment thread
simahawk marked this conversation as resolved.

def test_search_with_two_backends(self):
public_backend = self.env["storage.backend"].create(
{
"name": "Public backend",
"backend_type": "filesystem",
"is_public": True,
}
)
public_file = self.env["storage.file"].create(
{
"name": "public.txt",
"backend_id": public_backend.id,
"data": b"aGVsbG8=",
}
)
result = self.env["storage.file"].search([("is_public", "=", True)])
self.assertIn(public_file, result)
self.assertNotIn(self.storage_file, result)
Loading
Loading