Skip to content
1 change: 1 addition & 0 deletions storage_file/__manifest__.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,5 +22,6 @@
"security/storage_file.xml",
"data/ir_cron.xml",
"data/storage_backend.xml",
"data/ir_config_parameter.xml",
],
}
8 changes: 8 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,12 @@ 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:
raise NotFound() from err
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why hidding AccessError ?

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>
4 changes: 3 additions & 1 deletion storage_file/models/storage_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,9 @@ 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":
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

backend.servedy_by is external but the file is served by odoo

It's confusing

May be add a new served_by value like "force_odoo_for_external_backend" or whatever :)

if backend.served_by == "odoo" or (
backend.served_by == "external" and not backend.base_url
):
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 @@ -162,6 +193,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,4 +1,5 @@
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
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 +1,3 @@
from . import test_storage_file
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)
60 changes: 60 additions & 0 deletions storage_file/tests/test_is_public.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,60 @@
# 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.storage_file.invalidate_recordset(["is_public"])
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This cache invalidation should not be necessary. Does this test still succeed if this line is removed?

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)
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nitpicking:

Suggested change
self.assertIn(self.storage_file, result)
self.assertEqual(self.storage_file, result)


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)
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nitpicking:

Suggested change
self.assertIn(self.storage_file, result)
self.assertEqual(self.storage_file, result)


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)
32 changes: 32 additions & 0 deletions storage_file/tests/test_storage_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from urllib import parse

from odoo.exceptions import AccessError, UserError
from odoo.tests import Form

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

Expand Down Expand Up @@ -110,6 +111,17 @@ def test_url(self):
stfile.url, f"https://foo.com/baz/test-of-my_file-{stfile.id}.txt"
)

def test_url_external_no_base_url_falls_back_to_odoo(self):
"""External backend w/o base_url uses the internal Odoo route."""
stfile = self._create_storage_file()
params = self.env["ir.config_parameter"].sudo()
base_url = params.get_param("web.base.url")
stfile.backend_id.update({"served_by": "external", "base_url": ""})
self.assertEqual(
stfile.url,
f"{base_url}/storage.file/test-of-my_file-{stfile.id}.txt",
)

def test_url_without_base_url(self):
stfile = self._create_storage_file()
# served by odoo
Expand Down Expand Up @@ -306,3 +318,23 @@ def test_empty(self):
# get_url is called on new records
empty = self.env["storage.file"].new({})._get_url()
self.assertEqual(empty, "")

def test_default_backend_id_on_form(self):
"""Form pre-fills the default backend for storage.file."""
form = Form(self.env["storage.file"])
self.assertEqual(form.backend_id, self.backend)

def test_default_backend_id_from_param(self):
"""storage.file.backend_id param overrides the form default."""
other_backend = self.env["storage.backend"].create(
{
"name": "Other",
"backend_type": "filesystem",
"filename_strategy": "name_with_id",
}
)
self.env["ir.config_parameter"].sudo().set_param(
"storage.file.backend_id", str(other_backend.id)
)
form = Form(self.env["storage.file"])
self.assertEqual(form.backend_id, other_backend)
Loading
Loading