[15.0][IMP] webservice: add support for oauth2 authentication

requirements.txt

@@ -0,0 +1,3 @@
+# generated from manifests external_dependencies
+oauthlib
+requests-oauthlib

test-requirements.txt

@@ -0,0 +1 @@
+responses

webservice/README.rst

@@ -7,7 +7,7 @@ WebService
    !! This file is generated by oca-gen-addon-readme !!
    !! changes will be overwritten.                   !!
    !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
-   !! source digest: sha256:b21de3647819aeba7178e146f697f7d79b8cf865eaf19d5cf45f3bdd0bb5802f
+   !! source digest: sha256:a554d8b0213fbcf9d0a3516b0eb65705733d42498e3c009f6105dea07ae6da67
    !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
 
 .. |badge1| image:: https://img.shields.io/badge/maturity-Production%2FStable-green.png
@@ -58,6 +58,7 @@ Contributors
 ~~~~~~~~~~~~
 
 * Enric Tobella <[email protected]>
+* Alexandre Fayolle <[email protected]>
 
 Maintainers
 ~~~~~~~~~~~

webservice/__init__.py

@@ -1,2 +1,3 @@
 from . import components
 from . import models
+from . import controllers

webservice/__manifest__.py

@@ -15,6 +15,7 @@
     "author": "Creu Blanca, Camptocamp, Odoo Community Association (OCA)",
     "website": "https://github.com/OCA/web-api",
     "depends": ["component", "server_environment"],
+    "external_dependencies": {"python": ["requests-oauthlib", "oauthlib"]},
     "data": [
         "security/ir.model.access.csv",
         "security/ir_rule.xml",

webservice/components/request_adapter.py

@@ -3,10 +3,18 @@
 # @author Simone Orsi <[email protected]>
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
 
+import json
+import logging
+import time
+
 import requests
+from oauthlib.oauth2 import BackendApplicationClient, WebApplicationClient
+from requests_oauthlib import OAuth2Session
 
 from odoo.addons.component.core import Component
 
+_logger = logging.getLogger(__name__)
+
 
 class BaseRestRequestsAdapter(Component):
     _name = "base.requests"
@@ -65,3 +73,165 @@ def _get_url(self, url=None, url_params=None, **kwargs):
             url = self.collection.url
         url_params = url_params or kwargs
         return url.format(**url_params)
+
+
+class BackendApplicationOAuth2RestRequestsAdapter(Component):
+    _name = "oauth2.requests.backend.application"
+    _webservice_protocol = "http+oauth2-backend_application"
+    _inherit = "base.requests"
+
+    def get_client(self, oauth_params: dict):
+        return BackendApplicationClient(client_id=oauth_params["oauth2_clientid"])
+
+    def __init__(self, *args, **kw):
+        super().__init__(*args, **kw)
+        # cached value to avoid hitting the database each time we need the token
+        self._token = {}
+
+    def _is_token_valid(self, token):
+        """Validate given oauth2 token.
+
+        We consider that a token in valid if it has at least 10% of
+        its valid duration. So if a token has a validity of 1h, we will
+        renew it if we try to use it 6 minutes before its expiration date.
+        """
+        expires_at = token.get("expires_at", 0)
+        expires_in = token.get("expires_in", 3600)  # default to 1h
+        now = time.time()
+        return now <= (expires_at - 0.1 * expires_in)
+
+    @property
+    def token(self):
+        """Return a valid oauth2 token.
+
+        The tokens are stored in the database, and we check if they are still
+        valid, and renew them if needed.
+        """
+        if self._is_token_valid(self._token):
+            return self._token
+        backend = self.collection
+        with backend.env.registry.cursor() as cr:
+            cr.execute(
+                "SELECT oauth2_token FROM webservice_backend "
+                "WHERE id=%s "
+                "FOR NO KEY UPDATE",  # prevent concurrent token fetching
+                (backend.id,),
+            )
+            token_str = cr.fetchone()[0] or "{}"
+            token = json.loads(token_str)
+            if self._is_token_valid(token):
+                self._token = token
+            else:
+                new_token = self._fetch_new_token(old_token=token)
+                cr.execute(
+                    "UPDATE webservice_backend " "SET oauth2_token=%s " "WHERE id=%s",
+                    (json.dumps(new_token), backend.id),
+                )
+                self._token = new_token
+        return self._token
+
+    def _fetch_new_token(self, old_token):
+        # TODO: check if the old token has a refresh_token that can
+        # be used (and use it in that case)
+        oauth_params = self.collection.sudo().read(
+            [
+                "oauth2_clientid",
+                "oauth2_client_secret",
+                "oauth2_token_url",
+                "oauth2_audience",
+                "redirect_url",
+            ]
+        )[0]
+        client = self.get_client(oauth_params)
+        with OAuth2Session(client=client) as session:
+            token = session.fetch_token(
+                token_url=oauth_params["oauth2_token_url"],
+                cliend_id=oauth_params["oauth2_clientid"],
+                client_secret=oauth_params["oauth2_client_secret"],
+                audience=oauth_params.get("oauth2_audience") or "",
+            )
+        return token
+
+    def _request(self, method, url=None, url_params=None, **kwargs):
+        url = self._get_url(url=url, url_params=url_params)
+        new_kwargs = kwargs.copy()
+        new_kwargs.update(
+            {
+                "headers": self._get_headers(**kwargs),
+                "timeout": None,
+            }
+        )
+        client = BackendApplicationClient(client_id=self.collection.oauth2_clientid)
+        with OAuth2Session(client=client, token=self.token) as session:
+            # pylint: disable=E8106
+            request = session.request(method, url, **new_kwargs)
+            request.raise_for_status()
+            return request.content
+
+
+class WebApplicationOAuth2RestRequestsAdapter(Component):
+    _name = "oauth2.requests.web.application"
+    _webservice_protocol = "http+oauth2-web_application"
+    _inherit = "oauth2.requests.backend.application"
+
+    def get_client(self, oauth_params: dict):
+        return WebApplicationClient(
+            client_id=oauth_params["oauth2_clientid"],
+            code=oauth_params.get("oauth2_autorization"),
+            redirect_uri=oauth_params["redirect_url"],
+        )
+
+    def _fetch_token_from_authorization(self, authorization_code):
+
+        oauth_params = self.collection.sudo().read(
+            [
+                "oauth2_clientid",
+                "oauth2_client_secret",
+                "oauth2_token_url",
+                "oauth2_audience",
+                "redirect_url",
+            ]
+        )[0]
+        client = WebApplicationClient(client_id=oauth_params["oauth2_clientid"])
+
+        with OAuth2Session(
+            client=client, redirect_uri=oauth_params.get("redirect_url")
+        ) as session:
+            token = session.fetch_token(
+                oauth_params["oauth2_token_url"],
+                client_secret=oauth_params["oauth2_client_secret"],
+                code=authorization_code,
+                audience=oauth_params.get("oauth2_audience") or "",
+                include_client_id=True,
+            )
+        return token
+
+    def redirect_to_authorize(self, **authorization_url_extra_params):
+        """set the oauth2_state on the backend
+        :return: the webservice authorization url with the proper parameters
+        """
+        # we are normally authenticated at this stage, so no need to sudo()
+        backend = self.collection
+        oauth_params = backend.read(
+            [
+                "oauth2_clientid",
+                "oauth2_token_url",
+                "oauth2_audience",
+                "oauth2_authorization_url",
+                "oauth2_scope",
+                "redirect_url",
+            ]
+        )[0]
+        client = WebApplicationClient(
+            client_id=oauth_params["oauth2_clientid"],
+        )
+
+        with OAuth2Session(
+            client=client,
+            redirect_uri=oauth_params.get("redirect_url"),
+        ) as session:
+            authorization_url, state = session.authorization_url(
+                backend.oauth2_authorization_url, **authorization_url_extra_params
+            )
+            backend.oauth2_state = state
+            return authorization_url

webservice/controllers/__init__.py

@@ -0,0 +1 @@
+from . import oauth2

webservice/controllers/oauth2.py

@@ -0,0 +1,64 @@
+# Copyright 2024 Camptocamp SA
+# @author Alexandre Fayolle <[email protected]>
+# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
+import json
+import logging
+
+from oauthlib.oauth2.rfc6749 import errors
+from werkzeug.urls import url_encode
+
+from odoo import http
+from odoo.http import request
+
+_logger = logging.getLogger(__name__)
+
+
+class OAuth2Controller(http.Controller):
+    @http.route(
+        "/webservice/<int:backend_id>/oauth2/redirect",
+        type="http",
+        auth="public",
+        csrf=False,
+    )
+    def redirect(self, backend_id, **params):
+        backend = request.env["webservice.backend"].browse(backend_id).sudo()
+        if backend.auth_type != "oauth2" or backend.oauth2_flow != "web_application":
+            _logger.error("unexpected backed config for backend %d", backend_id)
+            raise errors.MismatchingRedirectURIError()
+        expected_state = backend.oauth2_state
+        state = params.get("state")
+        if state != expected_state:
+            _logger.error("unexpected state: %s", state)
+            raise errors.MismatchingStateError()
+        code = params.get("code")
+        adapter = (
+            backend._get_adapter()
+        )  # we expect an adapter that supports web_application
+        token = adapter._fetch_token_from_authorization(code)
+        backend.write(
+            {
+                "oauth2_token": json.dumps(token),
+                "oauth2_state": False,
+            }
+        )
+        # after saving the token, redirect to the backend form view
+        cids = request.httprequest.cookies.get("cids", "")
+        if cids:
+            cids = [int(cid) for cid in cids.split(",")]
+        else:
+            cids = []
+        record_action = backend.get_access_action()
+        url_params = {
+            "model": backend._name,
+            "id": backend.id,
+            "active_id": backend.id,
+            "action": record_action.get("id"),
+        }
+        view_id = backend.get_formview_id()
+        if view_id:
+            url_params["view_id"] = view_id
+
+        if cids:
+            url_params["cids"] = ",".join([str(cid) for cid in cids])
+        url = "/web?#%s" % url_encode(url_params)
+        return request.redirect(url)