diff --git a/addons/web/controllers/database.py b/addons/web/controllers/database.py index a6ac8db8c8e..3be41f4f06c 100644 --- a/addons/web/controllers/database.py +++ b/addons/web/controllers/database.py @@ -140,7 +140,7 @@ class Database(http.Controller): error = "Database backup error: %s" % (str(e) or repr(e)) return self._render_template(error=error) - @http.route('/web/database/restore', type='http', auth="none", methods=['POST'], csrf=False) + @http.route('/web/database/restore', type='http', auth="none", methods=['POST'], csrf=False, max_content_length=None) def restore(self, master_pwd, backup_file, name, copy=False, neutralize_database=False): insecure = odoo.tools.config.verify_admin_password('admin') if insecure and master_pwd: diff --git a/addons/web/tests/test_db_manager.py b/addons/web/tests/test_db_manager.py index 9af81f4f892..4327c7c51e6 100644 --- a/addons/web/tests/test_db_manager.py +++ b/addons/web/tests/test_db_manager.py @@ -4,6 +4,7 @@ import operator import re import secrets +from io import BytesIO from unittest.mock import patch import requests @@ -68,6 +69,14 @@ class TestDatabaseOperations(BaseCase): def assertDbs(self, dbs): self.assertEqual(self.list_dbs_filtered() - self.base_databases, set(dbs)) + def url_open_drop(self, dbname): + res = self.session.post(self.url('/web/database/drop'), data={ + 'master_pwd': self.password, + 'name': dbname, + }, allow_redirects=False) + res.raise_for_status() + return res + def test_database_creation(self): # check verify_admin_password patch self.assertTrue(odoo.tools.config.verify_admin_password(self.password)) @@ -88,10 +97,7 @@ class TestDatabaseOperations(BaseCase): self.assertDbs([test_db_name]) # delete the created database - res = self.session.post(self.url('/web/database/drop'), data={ - 'master_pwd': self.password, - 'name': test_db_name, - }, allow_redirects=False) + res = self.url_open_drop(test_db_name) self.assertEqual(res.status_code, 303) self.assertIn('/web/database/manager', res.headers['Location']) self.assertDbs([]) @@ -110,10 +116,70 @@ class TestDatabaseOperations(BaseCase): self.assertDbs([test_db_name]) # delete the created database - res = self.session.post(self.url('/web/database/drop'), data={ - 'master_pwd': self.password, - 'name': test_db_name, - }, allow_redirects=False) - self.assertEqual(res.status_code, 303) + res = self.url_open_drop(test_db_name) self.assertIn('/web/database/manager', res.headers['Location']) self.assertDbs([]) + + def test_database_restore(self): + test_db_name = self.db_name + '-test-database-restore' + self.assertNotIn(test_db_name, self.list_dbs_filtered()) + + # backup the current database inside a temporary zip file + res = self.session.post( + self.url('/web/database/backup'), + data={ + 'master_pwd': self.password, + 'name': self.db_name, + }, + allow_redirects=False, + stream=True, + ) + res.raise_for_status() + datetime_pattern = r'\d\d\d\d-\d\d-\d\d_\d\d-\d\d-\d\d' + self.assertRegex( + res.headers.get('Content-Disposition'), + fr"attachment; filename\*=UTF-8''{self.db_name}_{datetime_pattern}\.zip" + ) + backup_file = BytesIO() + backup_file.write(res.content) + self.assertGreater(backup_file.tell(), 0, "The backup seems corrupted") + + # upload the backup under a new name (create a duplicate) + with self.subTest(DEFAULT_MAX_CONTENT_LENGTH=None), \ + patch.object(odoo.http, 'DEFAULT_MAX_CONTENT_LENGTH', None): + backup_file.seek(0) + self.session.post( + self.url('/web/database/restore'), + data={ + 'master_pwd': self.password, + 'name': test_db_name, + 'copy': True, + }, + files={ + 'backup_file': backup_file, + }, + allow_redirects=False + ).raise_for_status() + self.assertDbs([test_db_name]) + self.url_open_drop(test_db_name) + + # upload the backup again, this time simulating that the file is + # too large under the default size limit, the default size limit + # shouldn't apply to /web/database URLs + with self.subTest(DEFAULT_MAX_CONTENT_LENGTH=1024), \ + patch.object(odoo.http, 'DEFAULT_MAX_CONTENT_LENGTH', 1024): + backup_file.seek(0) + self.session.post( + self.url('/web/database/restore'), + data={ + 'master_pwd': self.password, + 'name': test_db_name, + 'copy': True, + }, + files={ + 'backup_file': backup_file, + }, + allow_redirects=False + ).raise_for_status() + self.assertDbs([test_db_name]) + self.url_open_drop(test_db_name) diff --git a/odoo/addons/base/models/ir_http.py b/odoo/addons/base/models/ir_http.py index 48f938ab1f3..5f748c5c5f4 100644 --- a/odoo/addons/base/models/ir_http.py +++ b/odoo/addons/base/models/ir_http.py @@ -172,6 +172,21 @@ class IrHttp(models.AbstractModel): @classmethod def _pre_dispatch(cls, rule, args): + ICP = request.env['ir.config_parameter'].with_user(SUPERUSER_ID) + + # Change the default database-wide 128MiB upload limit on the + # ICP value. Do it before calling http's generic pre_dispatch + # so that the per-route limit @route(..., max_content_length=x) + # takes over. + try: + key = 'web.max_file_upload_size' + if value := (ICP.get_param(key, None) is not None): + request.httprequest.max_content_length = int(value) + except ValueError: # better not crash on ALL requests + _logger.error("invalid %s: %r, using %s instead", + key, value, request.httprequest.max_content_length, + ) + request.dispatcher.pre_dispatch(rule, args) # Replace uid placeholder by the current request.env.uid diff --git a/odoo/http.py b/odoo/http.py index fe742ba5553..a2967cd41fe 100644 --- a/odoo/http.py +++ b/odoo/http.py @@ -1492,18 +1492,6 @@ class Request: :returns: The merged key-value pairs. :rtype: dict """ - if self.env: - ICP = self.env['ir.config_parameter'].sudo() - try: - key = 'web.max_file_upload_size' - self.httprequest.max_content_length = int(ICP.get_param( - key, DEFAULT_MAX_CONTENT_LENGTH - )) - except ValueError: # better not crash on ALL requests - _logger.error("invalid %s: %r, use %s instead", - key, ICP.get_param(key), self.httprequest.max_content_length, - ) - params = { **self.httprequest.args, **self.httprequest.form, @@ -1814,6 +1802,9 @@ class Dispatcher(ABC): 'Origin, X-Requested-With, Content-Type, Accept, Authorization') werkzeug.exceptions.abort(Response(status=204)) + if 'max_content_length' in routing: + self.request.httprequest.max_content_length = routing['max_content_length'] + @abstractmethod def dispatch(self, endpoint, args): """