Add rpc_helper

rpc_helper/README.rst

@@ -0,0 +1 @@
+wait for the bot ;)

rpc_helper/__init__.py

@@ -0,0 +1 @@
+from .hooks import post_load_hook

rpc_helper/__manifest__.py

@@ -0,0 +1,15 @@
+# Copyright 2022 Camptocamp SA
+# @author: Simone Orsi <simone.orsi@camptocamp.com>
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
+
+{
+    "name": "Disable RPC",
+    "summary": """Helpers for disabling RPC calls""",
+    "version": "14.0.1.0.0",
+    "development_status": "Alpha",
+    "license": "LGPL-3",
+    "website": "https://github.com/OCA/server-tools",
+    "author": "Camptocamp, Odoo Community Association (OCA)",
+    "maintainers": ["simahawk"],
+    "post_load": "post_load_hook",
+}

rpc_helper/decorator.py

@@ -0,0 +1,19 @@
+# Copyright 2022 Camptocamp SA
+# @author: Simone Orsi <simone.orsi@camptocamp.com>
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
+
+
+def disable_rpc(*config):
+    """Decorate classes to disable RPC calls.
+
+    Possible values:
+
+    * none, block all methods
+    * *("$method_name1", "$method_name2"), blocks calls to specific methods
+    """
+
+    def _decorator(target):
+        target._disable_rpc = ("all",) if len(config) == 0 else config
+        return target
+
+    return _decorator

rpc_helper/hooks.py

@@ -0,0 +1,22 @@
+# Copyright 2022 Camptocamp SA
+# @author: Simone Orsi <simone.orsi@camptocamp.com>
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
+
+import logging
+
+from odoo.service import model
+
+from .patch import protected__execute_cr
+
+_logger = logging.getLogger(__name__)
+
+
+def patch__model_execute_cr():
+    """Patch rpc model handler."""
+    protected__execute_cr._orig__execute_cr = model.execute_cr
+    model.execute_cr = protected__execute_cr
+    _logger.info("PATCHED odoo.service.model.execute")
+
+
+def post_load_hook():
+    patch__model_execute_cr()

rpc_helper/patch.py

@@ -0,0 +1,26 @@
+# Copyright 2022 Camptocamp SA
+# @author: Simone Orsi <simone.orsi@camptocamp.com>
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
+
+import odoo
+from odoo.exceptions import UserError
+from odoo.tools.translate import _
+
+
+def protected__execute_cr(cr, uid, obj, method, *args, **kw):
+    # Same as original func in odoo.service.model.execute_cr
+    odoo.api.Environment.reset()  # clean cache etc if we retry the same transaction
+    recs = odoo.api.Environment(cr, uid, {}).get(obj)
+    if recs is None:
+        raise UserError(_("Object %s doesn't exist", obj))
+    # custom code starts here
+    if not _rpc_allowed(recs, method):
+        raise UserError(_("RPC call on %s is not allowed", obj))
+    return protected__execute_cr._orig__execute_cr(cr, uid, obj, method, *args, **kw)
+
+
+def _rpc_allowed(recordset, method):
+    config = getattr(recordset, "_disable_rpc", None)
+    if config is None:
+        return True
+    return "all" not in config and method not in config

rpc_helper/readme/CONTRIBUTORS.rst

@@ -0,0 +1 @@
+* Simone Orsi <simone.orsi@camptocamp.com>

rpc_helper/readme/DESCRIPTION.rst

@@ -0,0 +1 @@
+Provide helpers to authorize RPC calls.

rpc_helper/readme/USAGE.rst

@@ -0,0 +1,15 @@
+Decorate an Odoo model class like this::
+
+    from odoo.addons.rpc_helper.decorator import disable_rpc
+
+    @disable_rpc()
+    class AverageModel(models.Model):
+        _inherit = "avg.model"
+
+This will disable ALL calls.
+
+To selectively disable only some methods::
+
+    @disable_rpc("create", "write", "any_method")
+    class AverageModel(models.Model):
+        _inherit = "avg.model"

rpc_helper/rpc_test_example.py

@@ -0,0 +1,15 @@
+"""Basic example script you can use to test your own models for real.
+"""
+from xmlrpc import client
+
+HOST = "127.0.0.1"
+PORT = 8069
+DB_NAME = "ododdb"
+
+url = "http://%s:%d/xmlrpc/2/" % (HOST, PORT)
+xmlrpc_common = client.ServerProxy(url + "common")
+xmlrpc_db = client.ServerProxy(url + "db")
+xmlrpc_object = client.ServerProxy(url + "object")
+
+uid = xmlrpc_common.login(DB_NAME, "admin", "admin")
+res = xmlrpc_object.execute(DB_NAME, uid, "admin", "res.partner", "search", [])

rpc_helper/tests/__init__.py

@@ -0,0 +1,2 @@
+from . import test_xmlrpc
+from . import test_decorator

rpc_helper/tests/test_decorator.py

@@ -0,0 +1,34 @@
+# Copyright 2022 Camptocamp SA
+# @author: Simone Orsi <simone.orsi@camptocamp.com>
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
+
+
+import unittest
+
+from ..decorator import disable_rpc
+
+
+@disable_rpc()
+class All:
+    pass
+
+
+@disable_rpc("create")
+class One:
+    pass
+
+
+@disable_rpc("create", "write")
+class Multi:
+    pass
+
+
+class TestDecorator(unittest.TestCase):
+    def test_all(self):
+        self.assertEqual(All._disable_rpc, ("all",))
+
+    def test_one(self):
+        self.assertEqual(One._disable_rpc, ("create",))
+
+    def test_multi(self):
+        self.assertEqual(Multi._disable_rpc, ("create", "write"))

rpc_helper/tests/test_xmlrpc.py

@@ -0,0 +1,53 @@
+# Copyright 2022 Camptocamp SA
+# @author: Simone Orsi <simone.orsi@camptocamp.com>
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
+
+
+import xmlrpc
+
+from odoo.tests import common
+
+
+@common.tagged("post_install", "-at_install")
+class TestXMLRPC(common.HttpSavepointCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.admin_uid = cls.env.ref("base.user_admin").id
+
+    def _set_disable(self, val):
+        type(self.env["res.partner"])._disable_rpc = val
+
+    def tearDown(self):
+        klass = type(self.env["res.partner"])
+        if hasattr(klass, "_disable_rpc"):
+            delattr(klass, "_disable_rpc")
+        super().tearDown()
+
+    def _rpc_call(self, method, vals=None):
+        o = self.xmlrpc_object
+        db_name = common.get_db_name()
+        return o.execute(
+            db_name, self.admin_uid, "admin", "res.partner", method, vals or []
+        )
+
+    def test_xmlrpc_search_normal(self):
+        res = self._rpc_call("search")
+        self.assertTrue(isinstance(res, list))
+
+    def test_xmlrpc_all_blocked(self):
+        self._set_disable(("all",))
+        msg = "RPC call on res.partner is not allowed"
+        with self.assertRaisesRegex(xmlrpc.client.Fault, msg):
+            self._rpc_call("search")
+
+        with self.assertRaisesRegex(xmlrpc.client.Fault, msg):
+            self._rpc_call("create", vals=[{"name": "Foo"}])
+
+    def test_xmlrpc_can_search_create_blocked(self):
+        self._set_disable(("create",))
+        self._rpc_call("search")
+
+        msg = "RPC call on res.partner is not allowed"
+        with self.assertRaisesRegex(xmlrpc.client.Fault, msg):
+            self._rpc_call("create", vals=[{"name": "Foo"}])