[IMP] webservice: allow web application flow

Author: gurneyalex

requirements.txt

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

test-requirements.txt

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

webservice/__init__.py

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

webservice/__manifest__.py

@@ -15,7 +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"]},
+    "external_dependencies": {"python": ["requests-oauthlib", "oauthlib"]},
     "data": [
         "security/ir.model.access.csv",
         "views/webservice_backend.xml",

webservice/components/request_adapter.py

@@ -4,14 +4,17 @@
 # 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
+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"
@@ -72,11 +75,14 @@ def _get_url(self, url=None, url_params=None, **kwargs):
         return url.format(**url_params)
 
 
-class OAuth2RestRequestsAdapter(Component):
-    _name = "oauth2.requests"
-    _webservice_protocol = "http+oauth2"
+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
@@ -133,9 +139,10 @@ def _fetch_new_token(self, old_token):
                 "oauth2_client_secret",
                 "oauth2_token_url",
                 "oauth2_audience",
+                "redirect_url",
             ]
         )[0]
-        client = BackendApplicationClient(client_id=oauth_params["oauth2_clientid"])
+        client = self.get_client(oauth_params)
         with OAuth2Session(client=client) as session:
             token = session.fetch_token(
                 token_url=oauth_params["oauth2_token_url"],
@@ -160,3 +167,71 @@ def _request(self, method, url=None, url_params=None, **kwargs):
             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)

webservice/models/webservice_backend.py

@@ -3,10 +3,12 @@
 # @author Simone Orsi <[email protected]>
 # @author Alexandre Fayolle <[email protected]>
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
-
+import logging
 
 from odoo import _, api, exceptions, fields, models
 
+_logger = logging.getLogger(__name__)
+
 
 class WebserviceBackend(models.Model):
 
@@ -23,22 +25,41 @@ class WebserviceBackend(models.Model):
             ("none", "Public"),
             ("user_pwd", "Username & password"),
             ("api_key", "API Key"),
-            ("oauth2", "OAuth2 Backend Application Flow (Client Credentials Grant)"),
+            ("oauth2", "OAuth2"),
         ],
         required=True,
     )
     username = fields.Char(auth_type="user_pwd")
     password = fields.Char(auth_type="user_pwd")
     api_key = fields.Char(string="API Key", auth_type="api_key")
     api_key_header = fields.Char(string="API Key header", auth_type="api_key")
+    oauth2_flow = fields.Selection(
+        [
+            ("backend_application", "Backend Application (Client Credentials Grant)"),
+            ("web_application", "Web Application (Authorization Code Grant)"),
+        ],
+        readonly=False,
+        store=True,
+        compute="_compute_oauth2_flow",
+    )
     oauth2_clientid = fields.Char(string="Client ID", auth_type="oauth2")
     oauth2_client_secret = fields.Char(string="Client Secret", auth_type="oauth2")
     oauth2_token_url = fields.Char(string="Token URL", auth_type="oauth2")
+    oauth2_authorization_url = fields.Char(string="Authorization URL")
     oauth2_audience = fields.Char(
         string="Audience"
         # no auth_type because not required
     )
+    oauth2_scope = fields.Char(help="scope of the the authorization")
     oauth2_token = fields.Char(help="the OAuth2 token (serialized JSON)")
+    redirect_url = fields.Char(
+        compute="_compute_redirect_url",
+        help="The redirect URL to be used as part of the OAuth2 authorisation flow",
+    )
+    oauth2_state = fields.Char(
+        help="random key generated when authorization flow starts "
+        "to ensure that no CSRF attack happen"
+    )
     content_type = fields.Selection(
         [
             ("application/json", "JSON"),
@@ -82,7 +103,10 @@ def _valid_field_parameter(self, field, name):
         return name in extra_params or super()._valid_field_parameter(field, name)
 
     def call(self, method, *args, **kwargs):
-        return getattr(self._get_adapter(), method)(*args, **kwargs)
+        _logger.debug("backend %s: call %s %s %s", self.name, method, args, kwargs)
+        response = getattr(self._get_adapter(), method)(*args, **kwargs)
+        _logger.debug("backend %s: response: \n%s", self.name, response)
+        return response
 
     def _get_adapter(self):
         with self.work_on(self._name) as work:
@@ -94,9 +118,42 @@ def _get_adapter(self):
     def _get_adapter_protocol(self):
         protocol = self.protocol
         if self.auth_type.startswith("oauth2"):
-            protocol += f"+{self.auth_type}"
+            protocol += f"+{self.auth_type}-{self.oauth2_flow}"
         return protocol
 
+    @api.depends("auth_type")
+    def _compute_oauth2_flow(self):
+        for rec in self:
+            if rec.auth_type != "oauth2":
+                rec.oauth2_flow = False
+
+    @api.depends("auth_type", "oauth2_flow")
+    def _compute_redirect_url(self):
+        get_param = self.env["ir.config_parameter"].sudo().get_param
+        base_url = get_param("web.base.url")
+        if base_url.startswith("http://"):
+            _logger.warning(
+                "web.base.url is configured in http. Oauth2 requires using https"
+            )
+            base_url = base_url[len("http://") :]
+        if not base_url.startswith("https://"):
+            base_url = f"https://{base_url}"
+        for rec in self:
+            if rec.auth_type == "oauth2" and rec.oauth2_flow == "web_application":
+                rec.redirect_url = f"{base_url}/webservice/{rec.id}/oauth2/redirect"
+            else:
+                rec.redirect_url = False
+
+    def button_authorize(self):
+        _logger.info("Button OAuth2 Authorize")
+        authorize_url = self._get_adapter().redirect_to_authorize()
+        _logger.info("Redirecting to %s", authorize_url)
+        return {
+            "type": "ir.actions.act_url",
+            "url": authorize_url,
+            "target": "self",
+        }
+
     @property
     def _server_env_fields(self):
         base_fields = super()._server_env_fields
@@ -109,8 +166,11 @@ def _server_env_fields(self):
             "api_key": {},
             "api_key_header": {},
             "content_type": {},
+            "oauth2_flow": {},
+            "oauth2_scope": {},
             "oauth2_clientid": {},
             "oauth2_client_secret": {},
+            "oauth2_authorization_url": {},
             "oauth2_token_url": {},
             "oauth2_audience": {},
         }