From 7a33a523711bd5d8f86ab5f154a6e137aa2d4c77 Mon Sep 17 00:00:00 2001 From: "Thomas Lefebvre (thle)" Date: Wed, 16 Aug 2023 15:39:00 +0200 Subject: [PATCH] [FIX] google_calendar,microsoft_calendar: use ReadonlyDict To avoid developers to update the dict of `_events` in their overrides, to not alter by mistake the default behavior. Using a frozendict will force them to create a copy of the dict. closes odoo/odoo#123261 Signed-off-by: Arnaud Joset (arj) --- .../google_account/models/google_service.py | 6 +++- addons/google_calendar/tests/__init__.py | 1 + .../tests/test_google_event.py | 15 ++++++++ addons/google_calendar/utils/google_event.py | 9 ++--- .../models/microsoft_service.py | 7 +++- .../tests/test_microsoft_event.py | 9 +++++ .../utils/microsoft_event.py | 8 +++-- odoo/addons/base/tests/test_misc.py | 11 ++++++ odoo/tools/misc.py | 34 +++++++++++++++++++ 9 files changed, 91 insertions(+), 9 deletions(-) create mode 100644 addons/google_calendar/tests/test_google_event.py diff --git a/addons/google_account/models/google_service.py b/addons/google_account/models/google_service.py index ab3e582dd7b..40766bba5a0 100644 --- a/addons/google_account/models/google_service.py +++ b/addons/google_account/models/google_service.py @@ -91,7 +91,7 @@ class GoogleService(models.AbstractModel): raise self.env['res.config.settings'].get_config_warning(error_msg) @api.model - def _do_request(self, uri, params=None, headers=None, method='POST', preuri="https://www.googleapis.com", timeout=TIMEOUT): + def _do_request(self, uri, params=None, headers=None, method='POST', preuri=GOOGLE_API_BASE_URL, timeout=TIMEOUT): """ Execute the request to Google API. Return a tuple ('HTTP_CODE', 'HTTP_RESPONSE') :param uri : the url to contact :param params : dict or already encoded parameters for the request to make @@ -104,6 +104,10 @@ class GoogleService(models.AbstractModel): if headers is None: headers = {} + assert urls.url_parse(preuri + uri).host in [ + urls.url_parse(url).host for url in (GOOGLE_TOKEN_ENDPOINT, GOOGLE_API_BASE_URL) + ] + # Remove client_secret key from logs if isinstance(params, str): _log_params = json.loads(params) or {} diff --git a/addons/google_calendar/tests/__init__.py b/addons/google_calendar/tests/__init__.py index 465c59b9b24..f0d3acabc0e 100644 --- a/addons/google_calendar/tests/__init__.py +++ b/addons/google_calendar/tests/__init__.py @@ -1,6 +1,7 @@ # -*- coding: utf-8 -*- # Part of Odoo. See LICENSE file for full copyright and licensing details. +from . import test_google_event from . import test_sync_common from . import test_sync_google2odoo from . import test_sync_odoo2google diff --git a/addons/google_calendar/tests/test_google_event.py b/addons/google_calendar/tests/test_google_event.py new file mode 100644 index 00000000000..c9b8a6969f6 --- /dev/null +++ b/addons/google_calendar/tests/test_google_event.py @@ -0,0 +1,15 @@ +# -*- coding: utf-8 -*- +# Part of Odoo. See LICENSE file for full copyright and licensing details. +from odoo.tests.common import BaseCase +from odoo.addons.google_calendar.utils.google_calendar import GoogleEvent + + +class TestGoogleEvent(BaseCase): + def test_google_event_readonly(self): + event = GoogleEvent() + with self.assertRaises(TypeError): + event._events['foo'] = 'bar' + with self.assertRaises(AttributeError): + event._events.update({'foo': 'bar'}) + with self.assertRaises(TypeError): + dict.update(event._events, {'foo': 'bar'}) diff --git a/addons/google_calendar/utils/google_event.py b/addons/google_calendar/utils/google_event.py index 4a006868072..44e68c53ddc 100644 --- a/addons/google_calendar/utils/google_event.py +++ b/addons/google_calendar/utils/google_event.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- # Part of Odoo. See LICENSE file for full copyright and licensing details. -from odoo.tools import email_normalize +from odoo.tools import email_normalize, ReadonlyDict import logging from typing import Iterator, Mapping from collections import abc @@ -24,14 +24,15 @@ class GoogleEvent(abc.Set): """ def __init__(self, iterable=()): - self._events = {} + _events = {} for item in iterable: if isinstance(item, self.__class__): - self._events[item.id] = item._events[item.id] + _events[item.id] = item._events[item.id] elif isinstance(item, Mapping): - self._events[item.get('id')] = item + _events[item.get('id')] = item else: raise ValueError("Only %s or iterable of dict are supported" % self.__class__.__name__) + self._events = ReadonlyDict(_events) def __iter__(self) -> Iterator['GoogleEvent']: return iter(GoogleEvent([vals]) for vals in self._events.values()) diff --git a/addons/microsoft_account/models/microsoft_service.py b/addons/microsoft_account/models/microsoft_service.py index 68d17d667b7..484ed312f29 100644 --- a/addons/microsoft_account/models/microsoft_service.py +++ b/addons/microsoft_account/models/microsoft_service.py @@ -16,6 +16,7 @@ TIMEOUT = 20 DEFAULT_MICROSOFT_AUTH_ENDPOINT = 'https://login.microsoftonline.com/common/oauth2/v2.0/authorize' DEFAULT_MICROSOFT_TOKEN_ENDPOINT = 'https://login.microsoftonline.com/common/oauth2/v2.0/token' +DEFAULT_MICROSOFT_GRAPH_ENDPOINT = 'https://graph.microsoft.com' RESOURCE_NOT_FOUND_STATUSES = (204, 404) @@ -124,7 +125,7 @@ class MicrosoftService(models.AbstractModel): raise self.env['res.config.settings'].get_config_warning(error_msg) @api.model - def _do_request(self, uri, params=None, headers=None, method='POST', preuri="https://graph.microsoft.com", timeout=TIMEOUT): + def _do_request(self, uri, params=None, headers=None, method='POST', preuri=DEFAULT_MICROSOFT_GRAPH_ENDPOINT, timeout=TIMEOUT): """ Execute the request to Microsoft API. Return a tuple ('HTTP_CODE', 'HTTP_RESPONSE') :param uri : the url to contact :param params : dict or already encoded parameters for the request to make @@ -137,6 +138,10 @@ class MicrosoftService(models.AbstractModel): if headers is None: headers = {} + assert urls.url_parse(preuri + uri).host in [ + urls.url_parse(url).host for url in (DEFAULT_MICROSOFT_TOKEN_ENDPOINT, DEFAULT_MICROSOFT_GRAPH_ENDPOINT) + ] + _logger.debug("Uri: %s - Type : %s - Headers: %s - Params : %s !" % (uri, method, headers, params)) ask_time = fields.Datetime.now() diff --git a/addons/microsoft_calendar/tests/test_microsoft_event.py b/addons/microsoft_calendar/tests/test_microsoft_event.py index 88a8885c80b..15f0b616cec 100644 --- a/addons/microsoft_calendar/tests/test_microsoft_event.py +++ b/addons/microsoft_calendar/tests/test_microsoft_event.py @@ -302,3 +302,12 @@ class TestMicrosoftEvent(TestCommon): self.assertNotIn(self.simple_event, synced_events) self.assertIn(self.simple_event, not_synced_events) + + def test_microsoft_event_readonly(self): + event = MicrosoftEvent() + with self.assertRaises(TypeError): + event._events['foo'] = 'bar' + with self.assertRaises(AttributeError): + event._events.update({'foo': 'bar'}) + with self.assertRaises(TypeError): + dict.update(event._events, {'foo': 'bar'}) diff --git a/addons/microsoft_calendar/utils/microsoft_event.py b/addons/microsoft_calendar/utils/microsoft_event.py index a040e8d23c1..06ece8befdb 100644 --- a/addons/microsoft_calendar/utils/microsoft_event.py +++ b/addons/microsoft_calendar/utils/microsoft_event.py @@ -3,6 +3,7 @@ from odoo.api import model from typing import Iterator, Mapping from collections import abc +from odoo.tools import ReadonlyDict from odoo.addons.microsoft_calendar.utils.event_id_storage import combine_ids @@ -17,14 +18,15 @@ class MicrosoftEvent(abc.Set): """ def __init__(self, iterable=()): - self._events = {} + _events = {} for item in iterable: if isinstance(item, self.__class__): - self._events[item.id] = item._events[item.id] + _events[item.id] = item._events[item.id] elif isinstance(item, Mapping): - self._events[item.get('id')] = item + _events[item.get('id')] = item else: raise ValueError("Only %s or iterable of dict are supported" % self.__class__.__name__) + self._events = ReadonlyDict(_events) def __iter__(self) -> Iterator['MicrosoftEvent']: return iter(MicrosoftEvent([vals]) for vals in self._events.values()) diff --git a/odoo/addons/base/tests/test_misc.py b/odoo/addons/base/tests/test_misc.py index fe1674b5cea..91be1561eb6 100644 --- a/odoo/addons/base/tests/test_misc.py +++ b/odoo/addons/base/tests/test_misc.py @@ -496,3 +496,14 @@ class TestAddonsFileAccess(BaseCase): self.assertCannotRead(__file__, ValueError, filter_ext=('.png',)) # file doesnt exist but has wrong extension self.assertCannotRead(__file__.replace('.py', '.foo'), ValueError, filter_ext=('.png',)) + + +class TestDictTools(BaseCase): + def test_readonly_dict(self): + d = misc.ReadonlyDict({'foo': 'bar'}) + with self.assertRaises(TypeError): + d['baz'] = 'xyz' + with self.assertRaises(AttributeError): + d.update({'baz': 'xyz'}) + with self.assertRaises(TypeError): + dict.update(d, {'baz': 'xyz'}) diff --git a/odoo/tools/misc.py b/odoo/tools/misc.py index 8968a77cb80..c0c4b189b84 100644 --- a/odoo/tools/misc.py +++ b/odoo/tools/misc.py @@ -1656,6 +1656,40 @@ pickle.dumps = pickle_.dumps pickle.HIGHEST_PROTOCOL = pickle_.HIGHEST_PROTOCOL +class ReadonlyDict(Mapping): + """Helper for an unmodifiable dictionary, not even updatable using `dict.update`. + + This is similar to a `frozendict`, with one drawback and one advantage: + + - `dict.update` works for a `frozendict` but not for a `ReadonlyDict`. + - `json.dumps` works for a `frozendict` by default but not for a `ReadonlyDict`. + + This comes from the fact `frozendict` inherits from `dict` + while `ReadonlyDict` inherits from `collections.abc.Mapping`. + + So, depending on your needs, + whether you absolutely must prevent the dictionary from being updated (e.g., for security reasons) + or you require it to be supported by `json.dumps`, you can choose either option. + + E.g. + data = ReadonlyDict({'foo': 'bar'}) + data['baz'] = 'xyz' # raises exception + data.update({'baz', 'xyz'}) # raises exception + dict.update(data, {'baz': 'xyz'}) # raises exception + """ + def __init__(self, data): + self.__data = dict(data) + + def __getitem__(self, key): + return self.__data[key] + + def __len__(self): + return len(self.__data) + + def __iter__(self): + return iter(self.__data) + + class DotDict(dict): """Helper for dot.notation access to dictionary attributes