OZ-819: Skip logout confirmation

auth_oidc/controllers/main.py

@@ -6,10 +6,15 @@
 import hashlib
 import logging
 import secrets
+from urllib.parse import parse_qs, urljoin, urlparse
 
 from werkzeug.urls import url_decode, url_encode
 
+from odoo import http
+from odoo.http import request
+
 from odoo.addons.auth_oauth.controllers.main import OAuthLogin
+from odoo.addons.web.controllers.session import Session
 
 _logger = logging.getLogger(__name__)
 
@@ -48,3 +53,30 @@ def list_providers(self):
                     provider["auth_endpoint"], url_encode(params)
                 )
         return providers
+
+
+class OpenIDLogout(Session):
+    @http.route("/web/session/logout", type="http", auth="none")
+    def logout(self, redirect="/web/login"):
+        # https://openid.net/specs/openid-connect-rpinitiated-1_0.html#RPLogout
+        user = request.env["res.users"].sudo().browse(request.session.uid)
+        if user.oauth_provider_id.id:
+            provider = (
+                request.env["auth.oauth.provider"]
+                .sudo()
+                .browse(user.oauth_provider_id.id)
+            )
+            if provider.end_session_endpoint:
+                redirect_url = urljoin(request.httprequest.url_root, redirect)
+                components = urlparse(provider.end_session_endpoint)
+                params = parse_qs(components.query)
+                params["client_id"] = provider.client_id
+                params["post_logout_redirect_uri"] = redirect_url
+                if provider.skip_logout_confirmation and user.oauth_id_token:
+                    params["id_token_hint"] = user.oauth_id_token
+                logout_url = components._replace(query=url_encode(params)).geturl()
+                request.session.logout(keep_db=True)
+                return request.redirect(logout_url, local=False)
+        # User has no account with any provider
+        # or no logout URL is configured for the provider
+        return super().logout(redirect=redirect)

auth_oidc/models/auth_oauth_provider.py

@@ -46,6 +46,18 @@ class AuthOauthProvider(models.Model):
         string="Token URL", help="Required for OpenID Connect authorization code flow."
     )
     jwks_uri = fields.Char(string="JWKS URL", help="Required for OpenID Connect.")
+    end_session_endpoint = fields.Char(
+        string="End Session URL",
+        help="If set, the user is logged out in the authorization provider upon logout "
+        "in the client, should be the value of end_session_endpoint specified by "
+        "the authorization provider.",
+    )
+    skip_logout_confirmation = fields.Boolean(
+        default=False,
+        string="Skip Logout Confirmation",
+        help="If set to true, the logout confirmation is skipped in the "
+        "authorization provider.",
+    )
 
     @tools.ormcache("self.jwks_uri", "kid")
     def _get_keys(self, kid):

auth_oidc/models/res_users.py

@@ -6,7 +6,7 @@
 
 import requests
 
-from odoo import api, models
+from odoo import api, fields, models
 from odoo.exceptions import AccessDenied
 from odoo.http import request
 
@@ -16,6 +16,8 @@
 class ResUsers(models.Model):
     _inherit = "res.users"
 
+    oauth_id_token = fields.Char(string="OAuth Id Token", readonly=True, copy=False)
+
     def _auth_oauth_get_tokens_implicit_flow(self, oauth_provider, params):
         # https://openid.net/specs/openid-connect-core-1_0.html#ImplicitAuthResponse
         return params.get("access_token"), params.get("id_token")
@@ -75,7 +77,12 @@ def auth_oauth(self, provider, params):
             raise AccessDenied()
         # retrieve and sign in user
         params["access_token"] = access_token
+        params["id_token"] = id_token
         login = self._auth_oauth_signin(provider, validation, params)
+        oauth_user = self.search(
+            [("login", "=", login), ("oauth_access_token", "=", access_token)]
+        )
+        oauth_user.write({"oauth_id_token": params["id_token"]})
         if not login:
             raise AccessDenied()
         # return user credentials

auth_oidc/tests/__init__.py

@@ -1 +1,2 @@
 from . import test_auth_oidc_auth_code
+from . import test_auth_oidc_logout

auth_oidc/tests/test_auth_oidc_logout.py

@@ -0,0 +1,191 @@
+# Copyright 2021 ACSONE SA/NV <https://acsone.eu>
+# License: AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
+
+import contextlib
+import logging
+from unittest.mock import Mock
+from urllib.parse import parse_qsl, urljoin, urlparse
+
+from werkzeug.urls import url_encode
+
+import odoo
+from odoo.tests import common
+from odoo.tools.misc import DotDict
+
+from odoo.addons.website.tools import MockRequest
+
+from ..controllers.main import OpenIDLogout
+
+BASE_URL = "http://localhost:%s" % odoo.tools.config["http_port"]
+CLIENT_ID = "auth_oidc-test"
+LOGIN_PATH = "/web/login"
+OIDC_BASE_LOGOUT_URL = "http://keycloak"
+OIDC_LOGOUT_PATH = "/logout"
+
+logger = logging.getLogger(__name__)
+
+
+@contextlib.contextmanager
+def create_request(env, user_id, user_logout_func):
+    with MockRequest(env) as request:
+        request.httprequest.url_root = BASE_URL + "/"
+        request.session = DotDict(uid=user_id, logout=user_logout_func)
+        yield request
+
+
+class TestOpenIDLogout(common.HttpCase):
+    @staticmethod
+    def mock_logout_user(keep_db):
+        logger.info("Logging out user in Odoo")
+
+    def setUp(self):
+        super().setUp()
+        # search our test provider and bind the demo user to it
+        self.provider = self.env["auth.oauth.provider"].search(
+            [("client_id", "=", CLIENT_ID)]
+        )
+        self.assertEqual(len(self.provider), 1)
+
+    def _prepare_login_test_user(self, provider_id):
+        user = self.env.ref("base.user_demo")
+        user.write({"oauth_provider_id": provider_id, "oauth_uid": user.login})
+        return user
+
+    def _set_test_oidc_logout_url(self, end_session_endpoint):
+        self.provider.write({"end_session_endpoint": end_session_endpoint})
+
+    def test_skip_oidc_logout_for_user(self):
+        """Test that oidc logout is skipped if user is not associated to a provider"""
+        user = self._prepare_login_test_user(None)
+        with create_request(self.env, user.id, self.mock_logout_user):
+            resp = OpenIDLogout().logout()
+            self.assertEqual(LOGIN_PATH, resp.location)
+
+    def test_skip_oidc_logout_for_all_users(self):
+        """
+        Test that oidc logout is skipped for all users if provider has no logout url
+        """
+        self.assertFalse(self.provider.end_session_endpoint)
+        user = self._prepare_login_test_user(self.provider)
+        with create_request(self.env, user.id, self.mock_logout_user):
+            resp = OpenIDLogout().logout()
+            self.assertEqual(LOGIN_PATH, resp.location)
+
+    def test_oidc_logout(self):
+        """Test that oidc logout"""
+        self._set_test_oidc_logout_url(urljoin(OIDC_BASE_LOGOUT_URL, OIDC_LOGOUT_PATH))
+        user = self._prepare_login_test_user(self.provider)
+        mock_session = Mock()
+        with MockRequest(self.env) as request:
+            request.httprequest.url_root = BASE_URL + "/"
+            request.session = mock_session
+            mock_session.uid = user.id
+            resp = OpenIDLogout().logout()
+            self.assertTrue(resp.location.startswith(OIDC_BASE_LOGOUT_URL))
+            actual_components = urlparse(resp.location)
+            self.assertEqual(OIDC_LOGOUT_PATH, actual_components.path)
+            actual_params = dict(parse_qsl(actual_components.query))
+            self.assertEqual(CLIENT_ID, actual_params["client_id"])
+            self.assertEqual(
+                urljoin(BASE_URL, LOGIN_PATH), actual_params["post_logout_redirect_uri"]
+            )
+            mock_session.logout.assert_called_once_with(keep_db=True)
+
+    def test_oidc_logout_with_params(self):
+        """Test that params both in the logout and redirect urls are preserved"""
+        logout_url_params = {"param_1": 1, "param_2": 2}
+        oidc_logout_path = f"{OIDC_LOGOUT_PATH}?{url_encode(logout_url_params)}"
+        logout_url = urljoin(OIDC_BASE_LOGOUT_URL, oidc_logout_path)
+        self._set_test_oidc_logout_url(logout_url)
+        user = self._prepare_login_test_user(self.provider)
+        with create_request(self.env, user.id, self.mock_logout_user):
+            redirect_path = "{}?{}".format(
+                LOGIN_PATH, url_encode({"param_3": 3, "param_4": 4})
+            )
+            params = {}
+            params.update(logout_url_params)
+            params["client_id"] = CLIENT_ID
+            params["post_logout_redirect_uri"] = urljoin(BASE_URL, redirect_path)
+            resp = OpenIDLogout().logout(redirect_path)
+            self.assertTrue(resp.location.startswith(OIDC_BASE_LOGOUT_URL))
+            actual_components = urlparse(resp.location)
+            self.assertEqual(OIDC_LOGOUT_PATH, actual_components.path)
+            actual_params = dict(parse_qsl(actual_components.query))
+            self.assertEqual(CLIENT_ID, actual_params["client_id"])
+            self.assertEqual("1", actual_params["param_1"])
+            self.assertEqual("2", actual_params["param_2"])
+            post_logout_url = actual_params["post_logout_redirect_uri"]
+            self.assertTrue(post_logout_url.startswith(BASE_URL))
+            post_logout_components = urlparse(post_logout_url)
+            self.assertEqual(LOGIN_PATH, post_logout_components.path)
+            post_logout_params = dict(parse_qsl(post_logout_components.query))
+            self.assertEqual("3", post_logout_params["param_3"])
+            self.assertEqual("4", post_logout_params["param_4"])
+
+    def test_oidc_logout_with_absolute_redirect_url(self):
+        """Test that oidc logout allows an absolute redirect url"""
+        self._set_test_oidc_logout_url(urljoin(OIDC_BASE_LOGOUT_URL, OIDC_LOGOUT_PATH))
+        user = self._prepare_login_test_user(self.provider)
+        with create_request(self.env, user.id, self.mock_logout_user):
+            resp = OpenIDLogout().logout(BASE_URL)
+            self.assertTrue(resp.location.startswith(OIDC_BASE_LOGOUT_URL))
+            actual_components = urlparse(resp.location)
+            self.assertEqual(OIDC_LOGOUT_PATH, actual_components.path)
+            actual_params = dict(parse_qsl(actual_components.query))
+            self.assertEqual(CLIENT_ID, actual_params["client_id"])
+            self.assertEqual(BASE_URL, actual_params["post_logout_redirect_uri"])
+
+    def test_oidc_logout_skip_confirmation(self):
+        """Test that oidc logout skips confirmation"""
+        id_token = "test-id-token"
+        self._set_test_oidc_logout_url(urljoin(OIDC_BASE_LOGOUT_URL, OIDC_LOGOUT_PATH))
+        self.provider.write({"skip_logout_confirmation": True})
+        user = self._prepare_login_test_user(self.provider)
+        user.write({"oauth_id_token": id_token})
+        with create_request(self.env, user.id, self.mock_logout_user):
+            resp = OpenIDLogout().logout()
+            self.assertTrue(resp.location.startswith(OIDC_BASE_LOGOUT_URL))
+            actual_components = urlparse(resp.location)
+            self.assertEqual(OIDC_LOGOUT_PATH, actual_components.path)
+            actual_params = dict(parse_qsl(actual_components.query))
+            self.assertEqual(CLIENT_ID, actual_params["client_id"])
+            self.assertEqual(id_token, actual_params["id_token_hint"])
+            self.assertEqual(
+                urljoin(BASE_URL, LOGIN_PATH), actual_params["post_logout_redirect_uri"]
+            )
+
+    def test_oidc_logout_not_skip_confirmation_if_no_id_token(self):
+        """Test that oidc logout does not skip confirmation if user has no oauth_id_token"""
+        self._set_test_oidc_logout_url(urljoin(OIDC_BASE_LOGOUT_URL, OIDC_LOGOUT_PATH))
+        self.provider.write({"skip_logout_confirmation": True})
+        user = self._prepare_login_test_user(self.provider)
+        with create_request(self.env, user.id, self.mock_logout_user):
+            resp = OpenIDLogout().logout()
+            self.assertTrue(resp.location.startswith(OIDC_BASE_LOGOUT_URL))
+            actual_components = urlparse(resp.location)
+            self.assertEqual(OIDC_LOGOUT_PATH, actual_components.path)
+            actual_params = dict(parse_qsl(actual_components.query))
+            self.assertEqual(CLIENT_ID, actual_params["client_id"])
+            self.assertIsNone(actual_params.get("id_token_hint"))
+            self.assertEqual(
+                urljoin(BASE_URL, LOGIN_PATH), actual_params["post_logout_redirect_uri"]
+            )
+
+    def test_oidc_logout_not_skip_confirmation_if_not_enabled(self):
+        """Test that oidc logout skips confirmation"""
+        id_token = "test-id-token"
+        self._set_test_oidc_logout_url(urljoin(OIDC_BASE_LOGOUT_URL, OIDC_LOGOUT_PATH))
+        self.provider.write({"skip_logout_confirmation": False})
+        user = self._prepare_login_test_user(self.provider)
+        user.write({"oauth_id_token": id_token})
+        with create_request(self.env, user.id, self.mock_logout_user):
+            resp = OpenIDLogout().logout()
+            self.assertTrue(resp.location.startswith(OIDC_BASE_LOGOUT_URL))
+            actual_components = urlparse(resp.location)
+            self.assertEqual(OIDC_LOGOUT_PATH, actual_components.path)
+            actual_params = dict(parse_qsl(actual_components.query))
+            self.assertEqual(CLIENT_ID, actual_params["client_id"])
+            self.assertIsNone(actual_params.get("id_token_hint"))
+            self.assertEqual(
+                urljoin(BASE_URL, LOGIN_PATH), actual_params["post_logout_redirect_uri"]
+            )

auth_oidc/views/auth_oauth_provider.xml

@@ -18,6 +18,8 @@
             <field name="validation_endpoint" position="after">
                 <field name="token_endpoint" />
                 <field name="jwks_uri" />
+                <field name="end_session_endpoint" />
+                <field name="skip_logout_confirmation" />
             </field>
         </field>
     </record>