pengyuthon/odoo_source
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[FIX] base_import_module: restore <field file="..."> feature

Browse Source 
Revision odoo/odoo@cabb9e7e57 introduced a
regression: This is no longer possible to import a data module
using `<field file="..."/>` in their data file.

This revision targets to restore the feature as expected.
The unit tests added covers the feature, so that regression
no longer happens in the future.

It introduces a new concept of temporary directory `file_open`
can read from.
e.g.
```py
with odoo.tools.file_open_temporary_directory(self.env) as module_dir:
   with zipfile.ZipFile('foo.zip', "r") as z:
      z.extract('foo/__manifest__.py', module_dir)
   with odoo.tools.file_open('foo/__manifest__.py', env=self.env) as f:
      manifest = f.read()
```

Note that `file_open` will be allowed to read from that temporary
directory only if `env` is passed to `file_open`,
and if the `env` is part of the same transaction/request than the `env`
passed to `file_open_temporary_directory`.

This is to avoid having users, whether from other databases,
or even the same database,
trying to access these directories not belonging to them.
e.g. If an admin uploads sensitive data in this temporary directory,
no one than him must be allowed to read from these files, not even
another user from his database.

closes odoo/odoo#126326

X-original-commit: ea5cfe2e9493e4ef5a9b81851af452c8c51e407e
Signed-off-by: Denis Ledoux (dle) <dle@odoo.com>
This commit is contained in:
Denis Ledoux
2023-06-24 14:01:40 +02:00
parent 9fed5ebe8f
commit 2f19dc6095
4 changed files with 122 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
addons/base_import_module/models/ir_module.py
+41 -54
View File
@@ -10,25 +10,16 @@ import zipfile
from collections import defaultdict
from os.path import join as opj
import odoo
from odoo import api, fields, models, _
from odoo.exceptions import UserError
from odoo.modules.module import MANIFEST_NAMES
from odoo.tools import convert_csv_import, convert_sql_import, convert_xml_import, exception_to_unicode
from odoo.tools import file_open, file_open_temporary_directory
_logger = logging.getLogger(__name__)
MAX_FILE_SIZE = 100 * 1024 * 1024 # in megabytes
__import_paths__ = {}
def _file_open(env, path):
path = os.path.normcase(os.path.normpath(path))
import_path = __import_paths__.get(env)
if import_path and path.startswith(import_path):
return open(path, 'rb')
raise FileNotFoundError("File not found in the import paths: " + path)
class IrModule(models.Model):
_inherit = "ir.module.module"
@@ -54,7 +45,7 @@ class IrModule(models.Model):
terp = {}
manifest_path = next((opj(path, name) for name in MANIFEST_NAMES if os.path.exists(opj(path, name))), None)
if manifest_path:
with _file_open(self.env, manifest_path) as f:
with file_open(manifest_path, 'rb', env=self.env) as f:
terp.update(ast.literal_eval(f.read().decode()))
if not terp:
return False
@@ -101,7 +92,7 @@ class IrModule(models.Model):
noupdate = True
pathname = opj(path, filename)
idref = {}
with _file_open(self.env, pathname) as fp:
with file_open(pathname, 'rb', env=self.env) as fp:
if ext == '.csv':
convert_csv_import(self.env, module, pathname, fp.read(), idref, mode, noupdate)
elif ext == '.sql':
@@ -115,7 +106,7 @@ class IrModule(models.Model):
for root, dirs, files in os.walk(path_static):
for static_file in files:
full_path = opj(root, static_file)
with _file_open(self.env, full_path) as fp:
with file_open(full_path, 'rb', env=self.env) as fp:
data = base64.b64encode(fp.read())
url_path = '/{}{}'.format(module, full_path.split(path)[1].replace(os.path.sep, '/'))
if not isinstance(url_path, str):
@@ -196,49 +187,45 @@ class IrModule(models.Model):
if zf.file_size > MAX_FILE_SIZE:
raise UserError(_("File '%s' exceed maximum allowed file size", zf.filename))
with tempfile.TemporaryDirectory() as module_dir:
try:
__import_paths__[self.env] = module_dir
manifest_files = [
file
for file in z.filelist
if file.filename.count('/') == 1
and file.filename.split('/')[1] in MANIFEST_NAMES
]
module_data_files = defaultdict(list)
for manifest in manifest_files:
manifest_path = z.extract(manifest, module_dir)
mod_name = manifest.filename.split('/')[0]
try:
with _file_open(self.env, manifest_path) as f:
terp = ast.literal_eval(f.read().decode())
except Exception:
with file_open_temporary_directory(self.env) as module_dir:
manifest_files = [
file
for file in z.filelist
if file.filename.count('/') == 1
and file.filename.split('/')[1] in MANIFEST_NAMES
]
module_data_files = defaultdict(list)
for manifest in manifest_files:
manifest_path = z.extract(manifest, module_dir)
mod_name = manifest.filename.split('/')[0]
try:
with file_open(manifest_path, 'rb', env=self.env) as f:
terp = ast.literal_eval(f.read().decode())
except Exception:
continue
for filename in terp.get('data', []) + terp.get('init_xml', []) + terp.get('update_xml', []):
if os.path.splitext(filename)[1].lower() not in ('.xml', '.csv', '.sql'):
continue
for filename in terp.get('data', []) + terp.get('init_xml', []) + terp.get('update_xml', []):
if os.path.splitext(filename)[1].lower() not in ('.xml', '.csv', '.sql'):
continue
module_data_files[mod_name].append('%s/%s' % (mod_name, filename))
for file in z.filelist:
filename = file.filename
mod_name = filename.split('/')[0]
is_data_file = filename in module_data_files[mod_name]
is_static = filename.startswith('%s/static' % mod_name)
if is_data_file or is_static:
z.extract(file, module_dir)
module_data_files[mod_name].append('%s/%s' % (mod_name, filename))
for file in z.filelist:
filename = file.filename
mod_name = filename.split('/')[0]
is_data_file = filename in module_data_files[mod_name]
is_static = filename.startswith('%s/static' % mod_name)
if is_data_file or is_static:
z.extract(file, module_dir)
dirs = [d for d in os.listdir(module_dir) if os.path.isdir(opj(module_dir, d))]
for mod_name in dirs:
module_names.append(mod_name)
try:
# assert mod_name.startswith('theme_')
path = opj(module_dir, mod_name)
if self._import_module(mod_name, path, force=force):
success.append(mod_name)
except Exception as e:
_logger.exception('Error while importing module')
errors[mod_name] = exception_to_unicode(e)
finally:
__import_paths__.pop(self.env)
dirs = [d for d in os.listdir(module_dir) if os.path.isdir(opj(module_dir, d))]
for mod_name in dirs:
module_names.append(mod_name)
try:
# assert mod_name.startswith('theme_')
path = opj(module_dir, mod_name)
if self._import_module(mod_name, path, force=force):
success.append(mod_name)
except Exception as e:
_logger.exception('Error while importing module')
errors[mod_name] = exception_to_unicode(e)
r = ["Successfully imported module '%s'" % mod for mod in success]
for mod, error in errors.items():
r.append("Error while importing module '%s'.\n\n %s \n Make sure those modules are installed and try again." % (mod, error))
addons/base_import_module/tests/test_import_module.py
+40
View File
@@ -298,3 +298,43 @@ class TestImportModuleHttp(TestImportModule, odoo.tests.HttpCase):
self.assertEqual(self.url_open('/' + foo_icon_path).content, foo_icon_data)
# Assert icon of module bar, which must be the icon of the base module as none was provided
self.assertEqual(self.env.ref('base.module_bar').icon_image, self.env.ref('base.module_base').icon_image)
def test_import_module_field_file(self):
files = [
('foo/__manifest__.py', b"{'data': ['data.xml']}"),
('foo/data.xml', b"""
<data>
<record id="logo" model="ir.attachment">
<field name="name">Company Logo</field>
<field name="datas" type="base64" file="foo/static/src/img/content/logo.png"/>
<field name="res_model">ir.ui.view</field>
<field name="public" eval="True"/>
</record>
</data>
"""),
('foo/static/src/img/content/logo.png', b"foo_logo"),
]
self.import_zipfile(files)
logo_path, logo_data = files[2]
self.assertEqual(base64.b64decode(self.env.ref('foo.logo').datas), logo_data)
self.assertEqual(self.url_open('/' + logo_path).content, logo_data)
def test_import_module_assets_http(self):
asset_bundle = 'web_assets_backend'
asset_path = '/foo/static/src/js/test.js'
files = [
('foo/__manifest__.py', json.dumps({
'assets': {
asset_bundle: [
asset_path,
]
},
})),
('foo/static/src/js/test.js', b"foo_assets_backend"),
]
self.import_zipfile(files)
asset = self.env.ref('foo.web_assets_backend_/foo/static/src/js/test_js')
self.assertEqual(asset.bundle, asset_bundle)
self.assertEqual(asset.path, asset_path)
asset_data = files[1][1]
self.assertEqual(self.url_open(asset_path).content, asset_data)
odoo/tools/convert.py
+1 -1
View File
@@ -146,7 +146,7 @@ def _eval_xml(self, node, env):
data = node.text
if node.get('file'):
with file_open(node.get('file'), 'rb') as f:
with file_open(node.get('file'), 'rb', env=env) as f:
data = f.read()
if t == 'base64':
odoo/tools/misc.py
+40 -3
View File
@@ -19,6 +19,7 @@ import re
import socket
import subprocess
import sys
import tempfile
import threading
import time
import traceback
@@ -145,7 +146,7 @@ def exec_pg_command_pipe(name, *args):
# File paths
#----------------------------------------------------------
def file_path(file_path, filter_ext=('',)):
def file_path(file_path, filter_ext=('',), env=None):
"""Verify that a file exists under a known `addons_path` directory and return its full path.
Examples::
@@ -156,12 +157,16 @@ def file_path(file_path, filter_ext=('',)):
:param str file_path: absolute file path, or relative path within any `addons_path` directory
:param list[str] filter_ext: optional list of supported extensions (lowercase, with leading dot)
:param env: optional environment, required for a file path within a temporary directory
created using `file_open_temporary_directory()`
:return: the absolute path to the file
:raise FileNotFoundError: if the file is not found under the known `addons_path` directories
:raise ValueError: if the file doesn't have one of the supported extensions (`filter_ext`)
"""
root_path = os.path.abspath(config['root_path'])
addons_paths = odoo.addons.__path__ + [root_path]
if env and hasattr(env.transaction, '__file_open_tmp_paths'):
addons_paths += env.transaction.__file_open_tmp_paths
is_abs = os.path.isabs(file_path)
normalized_path = os.path.normpath(os.path.normcase(file_path))
@@ -183,7 +188,7 @@ def file_path(file_path, filter_ext=('',)):
raise FileNotFoundError("File not found: " + file_path)
def file_open(name, mode="r", filter_ext=None):
def file_open(name, mode="r", filter_ext=None, env=None):
"""Open a file from within the addons_path directories, as an absolute or relative path.
Examples::
@@ -196,11 +201,13 @@ def file_open(name, mode="r", filter_ext=None):
:param name: absolute or relative path to a file located inside an addon
:param mode: file open mode, as for `open()`
:param list[str] filter_ext: optional list of supported extensions (lowercase, with leading dot)
:param env: optional environment, required to open a file within a temporary directory
created using `file_open_temporary_directory()`
:return: file object, as returned by `open()`
:raise FileNotFoundError: if the file is not found under the known `addons_path` directories
:raise ValueError: if the file doesn't have one of the supported extensions (`filter_ext`)
"""
path = file_path(name, filter_ext=filter_ext)
path = file_path(name, filter_ext=filter_ext, env=env)
if os.path.isfile(path):
if 'b' not in mode:
# Force encoding for text mode, as system locale could affect default encoding,
@@ -213,6 +220,36 @@ def file_open(name, mode="r", filter_ext=None):
return open(path, mode)
raise FileNotFoundError("Not a file: " + name)
@contextmanager
def file_open_temporary_directory(env):
"""Create and return a temporary directory added to the directories `file_open` is allowed to read from.
`file_open` will be allowed to open files within the temporary directory
only for environments of the same transaction than `env`.
Meaning, other transactions/requests from other users or even other databases
won't be allowed to open files from this directory.
Examples::
>>> with odoo.tools.file_open_temporary_directory(self.env) as module_dir:
... with zipfile.ZipFile('foo.zip', 'r') as z:
... z.extract('foo/__manifest__.py', module_dir)
... with odoo.tools.file_open('foo/__manifest__.py', env=self.env) as f:
... manifest = f.read()
:param env: environment for which the temporary directory is created.
:return: the absolute path to the created temporary directory
"""
assert not hasattr(env.transaction, '__file_open_tmp_paths'), 'Reentrancy is not implemented for this method'
with tempfile.TemporaryDirectory() as module_dir:
try:
env.transaction.__file_open_tmp_paths = (module_dir,)
yield module_dir
finally:
del env.transaction.__file_open_tmp_paths
#----------------------------------------------------------
# iterables
#----------------------------------------------------------