[MIG] 13.0 migration

galicea_base/__init__.py

@@ -1,2 +1 @@
-# -*- coding: utf-8 -*-
 

galicea_base/__manifest__.py

@@ -1,23 +1,14 @@
-# -*- coding: utf-8 -*-
 {
-    'name': "Base menu for Odoo Galicea Ecosystem",
-
-    'summary': """
+    "name": "Base menu for Odoo Galicea Ecosystem",
+    "summary": """
         Menu only
      """,
-
-    'author': "Jurek Wawro",
-    'maintainer': "Galicea",
-    'website': "http://galicea.pl",
-
-    'category': 'Technical Settings',
-    'version': '12.0.1.0',
-
-    'depends': ['web',],
-
-    'data': [
-        'views/base_menu.xml',
-    ],
-
-    'installable': True
+    "author": "Jurek Wawro",
+    "maintainer": "Galicea",
+    "website": "http://galicea.pl",
+    "category": "Technical Settings",
+    "version": "13.0.1.0",
+    "depends": ["web",],
+    "data": ["views/base_menu.xml",],
+    "installable": True,
 }

galicea_base/views/base_menu.xml

@@ -2,5 +2,4 @@
 <odoo>
   <menuitem name="Galicea" id="galicea_admin_menu"
     parent="base.menu_administration" groups="base.group_erp_manager" />
-
 </odoo>

galicea_environment_checkup/__init__.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 from . import models
 
 from .environment_checkup.custom import custom_check

galicea_environment_checkup/__manifest__.py

@@ -1,32 +1,19 @@
-# -*- coding: utf-8 -*-
 {
-    'name': "Galicea Environment Check-up",
-
-    'summary': """
+    "name": "Galicea Environment Check-up",
+    "summary": """
         Programmatically validate environment, including internal and external
         dependencies""",
-
-    'author': "Maciej Wawro",
-    'maintainer': "Galicea",
-    'website': "http://galicea.pl",
-
-    'category': 'Technical Settings',
-    'version': '12.0.1.0',
-
-    'depends': ['web','galicea_base',],
-
-    'data': [
-        'views/data.xml',
-        'views/views.xml',
-        'views/environment_checks.xml'
+    "author": "Maciej Wawro",
+    "maintainer": "Galicea",
+    "website": "http://galicea.pl",
+    "category": "Technical Settings",
+    "version": "13.0.1.0",
+    "depends": ["web", "galicea_base",],
+    "data": ["views/data.xml", "views/views.xml", "views/environment_checks.xml"],
+    "qweb": ["static/src/xml/templates.xml"],
+    "images": [
+        "static/description/images/custom_screenshot.png",
+        "static/description/images/dependencies_screenshot.png",
     ],
-
-    'qweb': ['static/src/xml/templates.xml'],
-
-    'images': [
-        'static/description/images/custom_screenshot.png',
-        'static/description/images/dependencies_screenshot.png'
-    ],
-
-    'installable': True
+    "installable": True,
 }

galicea_environment_checkup/controllers/__init__.py

@@ -1,3 +1 @@
-# -*- coding: utf-8 -*-
-
 from . import dashboard

galicea_environment_checkup/controllers/dashboard.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 from odoo import http
 from odoo.exceptions import AccessError
 from odoo.http import request
@@ -7,20 +5,17 @@ from odoo.http import request
 from ..environment_checkup.runtime import all_installed_checks, display_data
 from ..environment_checkup.core import CheckResult
 
+
 class Dashboard(http.Controller):
-    @http.route('/galicea_environment_checkup/data', type='json', auth='user')
+    @http.route("/galicea_environment_checkup/data", type="json", auth="user")
     def data(self, request, **kw):
-        if not request.env.user.has_group('base.group_erp_manager'):
+        if not request.env.user.has_group("base.group_erp_manager"):
             raise AccessError("Access Denied")
 
         checks = all_installed_checks(request.env)
         response = display_data(request.env, checks)
 
-        priority = {
-            CheckResult.FAIL: 0,
-            CheckResult.WARNING: 1,
-            CheckResult.SUCCESS: 2
-        }
-        response.sort(key=lambda res: (priority[res['result']], res['module']))
+        priority = {CheckResult.FAIL: 0, CheckResult.WARNING: 1, CheckResult.SUCCESS: 2}
+        response.sort(key=lambda res: (priority[res["result"]], res["module"]))
 
         return response

galicea_environment_checkup/environment_checkup/__init__.py

@@ -1,4 +1,2 @@
-# -*- coding: utf-8 -*-
-
 from . import core
 from . import custom

galicea_environment_checkup/environment_checkup/core.py

@@ -1,37 +1,42 @@
-# -*- coding: utf-8 -*-
-
 import logging
+
 _logger = logging.getLogger(__name__)
 
+
 class CheckResult(object):
-    SUCCESS = 'success'
-    WARNING = 'warning'
-    FAIL = 'fail'
+    SUCCESS = "success"
+    WARNING = "warning"
+    FAIL = "fail"
 
-    def __init__(self, result, message, details = None):
+    def __init__(self, result, message, details=None):
         super(CheckResult, self).__init__()
 
         self.result = result
         self.message = message
         self.details = details
 
+
 class CheckSuccess(CheckResult):
     def __init__(self, message, **kwargs):
         super(CheckSuccess, self).__init__(CheckResult.SUCCESS, message, **kwargs)
 
+
 class CheckIssue(CheckResult, Exception):
     def __init__(self, result, message, **kwargs):
         Exception.__init__(self, message)
         CheckResult.__init__(self, result, message, **kwargs)
 
+
 class CheckFail(CheckIssue):
     def __init__(self, message, **kwargs):
         super(CheckFail, self).__init__(CheckResult.FAIL, message, **kwargs)
 
+
 class CheckWarning(CheckIssue):
     def __init__(self, message, **kwargs):
         super(CheckWarning, self).__init__(CheckResult.WARNING, message, **kwargs)
 
+
 class Check(object):
     def __init__(self, module):
         self.module = module
@@ -43,7 +48,7 @@ class Check(object):
             return issue
         except Exception as ex:
             _logger.exception(ex)
-            return CheckFail('Check failed when processing: {}'.format(ex))
+            return CheckFail("Check failed when processing: {}".format(ex))
 
     def _run(self, env):
-        raise NotImplementedError('Should be overriden by the subclass')
+        raise NotImplementedError("Should be overriden by the subclass")

galicea_environment_checkup/environment_checkup/custom.py

@@ -1,11 +1,10 @@
-# -*- coding: utf-8 -*-
-
 import collections
 
 from .core import Check
 
 custom_checks_per_module = collections.defaultdict(list)
 
+
 class CustomCheck(Check):
     def __init__(self, module, func):
         super(CustomCheck, self).__init__(module)
@@ -14,17 +13,17 @@ class CustomCheck(Check):
     def _run(self, env):
         return self.func(env)
 
+
 def custom_check(func):
     try:
-        module = func.__module__.split('.')[2]
+        module = func.__module__.split(".")[2]
     except IndexError:
-        module = ''
+        module = ""
 
-    custom_checks_per_module[module].append(
-        CustomCheck(module=module, func=func)
-    )
+    custom_checks_per_module[module].append(CustomCheck(module=module, func=func))
 
     return func
 
+
 def get_checks_for_module(module_name):
     return custom_checks_per_module[module_name]

galicea_environment_checkup/environment_checkup/dependencies.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 import subprocess
 import re
 import cgi
@@ -8,6 +6,7 @@ from odoo.tools import which
 
 from .core import Check, CheckSuccess, CheckWarning, CheckFail
 
+
 class DependencyCheck(Check):
     dependency_type = None
 
@@ -16,94 +15,106 @@ class DependencyCheck(Check):
         self.dependency = dependency
 
     def _dependency_installed(self, env, name):
-        raise NotImplementedError('Should be overriden by the subclass')
+        raise NotImplementedError("Should be overriden by the subclass")
 
     def _installed_version(self, env, name):
-        raise NotImplementedError('Should be overriden by the subclass')
+        raise NotImplementedError("Should be overriden by the subclass")
 
     def _details(self):
-        if 'install' in self.dependency:
-            return 'Install command: <pre>{}</pre>'.format(self.dependency['install'])
+        if "install" in self.dependency:
+            return "Install command: <pre>{}</pre>".format(self.dependency["install"])
         return None
 
     def __has_required_version(self, installed_version, version_expression):
-        version_operator = '='
-        version = self.dependency['version']
-        if version[:1] in ['=', '~', '^']:
+        version_operator = "="
+        version = self.dependency["version"]
+        if version[:1] in ["=", "~", "^"]:
             version_operator = version[:1]
             version = version[1:]
-        elif version[:2] in ['>=']:
+        elif version[:2] in [">="]:
             version_operator = version[:2]
             version = version[2:]
 
         # Py3 : map -> list(map
         #  https://stackoverflow.com/questions/33717314/attributeerror-map-obejct-has-no-attribute-index-python-3
         try:
-            parsed_version = list(map(int, version.split('.')))
+            parsed_version = list(map(int, version.split(".")))
         except ValueError:
             raise CheckFail(
-                'Invalid version expression',
-                details = """
+                "Invalid version expression",
+                details="""
 Allowed expressions are <pre>=x.y.z</pre>, <pre>&gt;=x.y.z</pre>, <pre>^x.z.y</pre>,
-<pre>~x.y.z. Got <pre>{}</pre>""".format(cgi.escape(self.dependency['version']))
+<pre>~x.y.z. Got <pre>{}</pre>""".format(
+                    cgi.escape(self.dependency["version"])
+                ),
             )
-        parsed_installed_version = list(map(int, installed_version.split('.'))) 
+        parsed_installed_version = list(map(int, installed_version.split(".")))
 
-        parsed_version.extend(0 for _ in range(len(parsed_installed_version) - len(parsed_version)))
-        parsed_installed_version.extend(0 for _ in range(len(parsed_version) - len(parsed_installed_version)))
+        parsed_version.extend(
+            0 for _ in range(len(parsed_installed_version) - len(parsed_version))
+        )
+        parsed_installed_version.extend(
+            0 for _ in range(len(parsed_version) - len(parsed_installed_version))
+        )
 
-        if version_operator == '^':
+        if version_operator == "^":
             if parsed_installed_version[:1] != parsed_version[:1]:
                 return False
-            version_operator = '>='
-        elif version_operator == '~':
+            version_operator = ">="
+        elif version_operator == "~":
             if parsed_installed_version[:2] != parsed_version[:2]:
                 return False
-            version_operator = '>='
+            version_operator = ">="
 
-        if version_operator == '>=':
+        if version_operator == ">=":
             return tuple(parsed_installed_version) >= tuple(parsed_version)
-        elif version_operator == '=':
+        elif version_operator == "=":
             return tuple(parsed_installed_version) == tuple(parsed_version)
 
         assert False
 
     def _run(self, env):
-        name = self.dependency['name']
+        name = self.dependency["name"]
         if not self._dependency_installed(env, name):
             raise CheckFail(
-                'Required {} - {} - is not installed.'.format(self.dependency_type, name),
-                details=self._details()
+                "Required {} - {} - is not installed.".format(
+                    self.dependency_type, name
+                ),
+                details=self._details(),
             )
-        if 'version' in self.dependency:
-            version_expression = self.dependency['version']
+        if "version" in self.dependency:
+            version_expression = self.dependency["version"]
             installed_version = self._installed_version(env, name)
             if not self.__has_required_version(installed_version, version_expression):
                 raise CheckWarning(
-                    'Required {} - {} - has version {}, but {} is needed.'.format(
+                    "Required {} - {} - has version {}, but {} is needed.".format(
                         self.dependency_type,
                         name,
                         installed_version,
-                        version_expression
+                        version_expression,
                     ),
-                    details=self._details()
+                    details=self._details(),
                 )
         return CheckSuccess(
-            'Required {} - {} - is installed.'.format(self.dependency_type, name),
-            details=self._details()
+            "Required {} - {} - is installed.".format(self.dependency_type, name),
+            details=self._details(),
         )
 
+
 class InternalDependencyCheck(DependencyCheck):
-    dependency_type = 'Odoo module'
+    dependency_type = "Odoo module"
 
     def _dependency_installed(self, env, name):
         return name in env.registry._init_modules
 
     def _installed_version(self, env, name):
-        return env['ir.module.module'].sudo().search([('name', '=', name)]).latest_version
+        return (
+            env["ir.module.module"].sudo().search([("name", "=", name)]).latest_version
+        )
+
 
 class PythonDependencyCheck(DependencyCheck):
-    dependency_type = 'Python module'
+    dependency_type = "Python module"
 
     def _dependency_installed(self, env, name):
         try:
@@ -117,12 +128,13 @@ class PythonDependencyCheck(DependencyCheck):
             return __import__(name).__version__
         except AttributeError:
             raise CheckWarning(
-                'Could not detect version of the Python module: {}.'.format(name),
-                details=self._details()
+                "Could not detect version of the Python module: {}.".format(name),
+                details=self._details(),
             )
 
+
 class ExternalDependencyCheck(DependencyCheck):
-    dependency_type = 'system executable'
+    dependency_type = "system executable"
 
     def _dependency_installed(self, env, name):
         try:
@@ -134,36 +146,40 @@ class ExternalDependencyCheck(DependencyCheck):
     def _installed_version(self, env, name):
         try:
             exe = which(name)
-            out = str(subprocess.check_output([exe, '--version'])) # Py3 str()
-            match = re.search('[\d.]+', out)
+            out = str(subprocess.check_output([exe, "--version"]))  # Py3 str()
+            match = re.search("[\d.]+", out)
             if not match:
                 raise CheckWarning(
-                    'Unable to detect version for executable {}'.format(name),
-                    details="Command {} --version returned <pre>{}</pre>".format(exe, out)
+                    "Unable to detect version for executable {}".format(name),
+                    details="Command {} --version returned <pre>{}</pre>".format(
+                        exe, out
+                    ),
                 )
             return match.group(0)
         except subprocess.CalledProcessError as e:
             raise CheckWarning(
-                'Unable to detect version for executable {}: {}'.format(name, e),
-                details=self._details()
+                "Unable to detect version for executable {}: {}".format(name, e),
+                details=self._details(),
             )
 
+
 def get_checks_for_module(module_name):
     result = []
 
     manifest = load_information_from_description_file(module_name)
-    manifest_checks = manifest.get('environment_checkup') or {}
-    dependencies = manifest_checks.get('dependencies') or {}
+    manifest_checks = manifest.get("environment_checkup") or {}
+    dependencies = manifest_checks.get("dependencies") or {}
 
-    for dependency in dependencies.get('python') or []:
+    for dependency in dependencies.get("python") or []:
         result.append(PythonDependencyCheck(module_name, dependency))
-    for dependency in dependencies.get('external') or []:
+    for dependency in dependencies.get("external") or []:
         result.append(ExternalDependencyCheck(module_name, dependency))
-    for dependency in dependencies.get('internal') or []:
+    for dependency in dependencies.get("internal") or []:
         result.append(InternalDependencyCheck(module_name, dependency))
 
     return result
 
+
 def get_checks_for_module_recursive(module):
     class ModuleDFS(object):
         def __init__(self):

galicea_environment_checkup/environment_checkup/runtime.py

@@ -1,7 +1,6 @@
-# -*- coding: utf-8 -*-
-
 from . import custom, dependencies
 
+
 def all_installed_checks(env):
     result = []
     installed_modules = env.registry._init_modules
@@ -10,15 +9,18 @@ def all_installed_checks(env):
         result += dependencies.get_checks_for_module(module_name)
     return result
 
+
 def display_data(env, checks):
     response = []
     for check in checks:
         result = check.run(env)
-        response.append({
-            'module': check.module,
-            'message': result.message,
-            'details': result.details,
-            'result': result.result
-        })
+        response.append(
+            {
+                "module": check.module,
+                "message": result.message,
+                "details": result.details,
+                "result": result.result,
+            }
+        )
 
     return response

galicea_environment_checkup/models/ext_module.py

@@ -1,17 +1,14 @@
-# -*- coding: utf-8 -*-
-
 import json
 from odoo import api, fields, models
 
 from ..environment_checkup import dependencies
 from ..environment_checkup.runtime import display_data
 
+
 class Module(models.Model):
-    _inherit = 'ir.module.module'
+    _inherit = "ir.module.module"
 
-    dependency_checks = fields.Text(
-        compute='_compute_dependency_checks'
-    )
+    dependency_checks = fields.Text(compute="_compute_dependency_checks")
 
     @api.one
     def _compute_dependency_checks(self):

galicea_openapi/__init__.py

@@ -1,5 +1,2 @@
-# -*- coding: utf-8 -*-
-
-#from . import models
 from . import controllers
 

galicea_openapi/__manifest__.py

@@ -1,33 +1,23 @@
-# -*- coding: utf-8 -*-
 {
-    'name': "openapi",
-
-    'summary': """
+    "name": "openapi",
+    "summary": """
         Odoo Opnapi
         UWAGA! Obecnie dekorator apiroute ma ograniczoną funkcjonalność.
         M.in. tylko jeden URL
         controllers/api.py zawiera przykład wykorzystania -
         pod adresem /oapi/api zwraca dokumentację w JSON 
         """,
-
-    'description': """
+    "description": """
         
         """,
-
-    'author': 'Jerzy Wawro',
-    'maintainer': "Galicea",
-    'website': "http://www.galicea.pl",
-    'category': 'Tools',
-    'version': '12.0.0.1',
-
-    'depends': [
-    ],
-    'external_dependencies': {
-      'python': [ 'fastapi', 'pydantic', 'starlette' ]
-    },
-    'data': [
-    ],
-    'application': True,
-    'installable': True,
-    
+    "author": "Jerzy Wawro",
+    "maintainer": "Galicea",
+    "website": "http://www.galicea.pl",
+    "category": "Tools",
+    "version": "13.0.0.1",
+    "depends": [],
+    "external_dependencies": {"python": ["fastapi", "pydantic", "starlette"]},
+    "data": [],
+    "application": True,
+    "installable": True,
 }

galicea_openapi/controllers/__init__.py

@@ -1,4 +1,2 @@
-# -*- coding: utf-8 -*-
-
 from . import api
 

galicea_openapi/controllers/api.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 import json
 
 from fastapi.openapi.docs import get_swagger_ui_html
@@ -9,48 +7,38 @@ from ..openapi import apiroute
 from ..openapi import oapi
 
 
-
-
-
 class OpenApiTest(http.Controller):
-
-
-    @http.route(['/oapi/tst1',], type='http', auth="user", website=True)
+    @http.route(["/oapi/tst1",], type="http", auth="user", website=True)
     def tst1(self, **kw):
         return "tst1"
 
-    @oapi.get('/oapi/tst2')
-    @http.route(['/oapi/tst2',], type='http', auth="user", website=True)
+    @oapi.get("/oapi/tst2")
+    @http.route(["/oapi/tst2",], type="http", auth="user", website=True)
     def tst2(self):
-        return 'ok test2'
+        return "ok test2"
 
-    @oapi.api_route('/oapi/tst3')
-    @http.route(['/oapi/tst3',], type='http', auth="user", website=True)
+    @oapi.api_route("/oapi/tst3")
+    @http.route(["/oapi/tst3",], type="http", auth="user", website=True)
     def tst3(self, par1="abc"):
         return par1
 
-
-    @oapi.api_route('/oapi/tst4')
-    @http.route(['/oapi/tst4', ], type='http', auth="user", website=True)
-    def tst4(self,par1="444"):
+    @oapi.api_route("/oapi/tst4")
+    @http.route(["/oapi/tst4",], type="http", auth="user", website=True)
+    def tst4(self, par1="444"):
         return par1
 
-
-    @apiroute('/oapi/tst5')
+    @apiroute("/oapi/tst5")
     def tst5(self, par1="555"):
         return par1
 
-    @http.route(['/oapi/api',], type='http', auth="user", website=True)
+    @http.route(["/oapi/api",], type="http", auth="user", website=True)
     def api(self, **kw):
         return json.dumps(oapi.openapi())
-#        wynik możesz skopiować do https://editor.swagger.io/
 
-    @http.route(['/oapi/docs',], type='http', auth="user", website=True)
+    #        wynik możesz skopiować do https://editor.swagger.io/
+
+    @http.route(["/oapi/docs",], type="http", auth="user", website=True)
     def api_UI(self, **kw):
-        response = get_swagger_ui_html(openapi_url = '/oapi/api', title = 'tytuł')
+        response = get_swagger_ui_html(openapi_url="/oapi/api", title="tytuł")
         return response.body
 
-
-
-
-

galicea_openapi/models/__init__.py

@@ -1,3 +0,0 @@
-# -*- coding: utf-8 -*-
-
-

galicea_openapi/openapi.py

@@ -1,12 +1,8 @@
-# -*- coding: utf-8 -*-
-
-
 import functools
 from fastapi import FastAPI
 from fastapi.openapi.utils import get_openapi
 
 
-
 def custom_openapi():
     if oapi.openapi_schema:
         return oapi.openapi_schema
@@ -22,19 +18,22 @@ def custom_openapi():
     oapi.openapi_schema = openapi_schema
     return oapi.openapi_schema
 
+
 oapi = FastAPI()
 oapi.openapi = custom_openapi
 
+
 def apiroute(route=None, **kw):
 
     routing = kw.copy()
+
     def apidecorator(f):
         if route:
             if isinstance(route, list):
                 routes = route
             else:
                 routes = [route]
-            routing['routes'] = routes
+            routing["routes"] = routes
 
         @functools.wraps(f)
         def response_wrap(*args, **kw):

galicea_openid_connect/__init__.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 from . import controllers
 from . import models
 from . import system_checks

galicea_openid_connect/__manifest__.py

@@ -1,51 +1,37 @@
-# -*- coding: utf-8 -*-
 {
-    'name': "Galicea OpenID Connect Provider",
-
-    'summary': """OpenID Connect Provider and OAuth2 resource server""",
-
-    'author': "Maciej Wawro",
-    'maintainer': "Galicea",
-    'website': "http://galicea.pl",
-
-    'category': 'Technical Settings',
-    'version': '12.0.0.0',
-
-    'depends': ['web', 'galicea_environment_checkup', 'galicea_base' ],
-
-    'external_dependencies': {
-        'python': ['jwcrypto', 'cryptography']
-    },
-
-    'data': [
-        'security/security.xml',
-        'security/ir.model.access.csv',
-#        'security/init.yml',
-        'security/init.xml',
-        'views/views.xml',
-        'views/templates.xml'
+    "name": "Galicea OpenID Connect Provider",
+    "summary": """OpenID Connect Provider and OAuth2 resource server""",
+    "author": "Maciej Wawro",
+    "maintainer": "Galicea",
+    "website": "http://galicea.pl",
+    "category": "Technical Settings",
+    "version": "13.0.0.0",
+    "depends": ["web", "galicea_environment_checkup", "galicea_base"],
+    "external_dependencies": {"python": ["jwcrypto", "cryptography"]},
+    "data": [
+        "security/security.xml",
+        "security/ir.model.access.csv",
+        #        'security/init.yml',
+        "security/init.xml",
+        "views/views.xml",
+        "views/templates.xml",
     ],
-
-    'environment_checkup': {
-        'dependencies': {
-            'python': [
+    "environment_checkup": {
+        "dependencies": {
+            "python": [
+                {"name": "jwcrypto", "install": "pip install 'jwcrypto==0.5.0'"},
                 {
-                    'name': 'jwcrypto',
-                    'install': "pip install 'jwcrypto==0.5.0'"
+                    "name": "cryptography",
+                    "version": ">=2.3",
+                    "install": "pip install 'cryptography>=2.3'",
                 },
-                {
-                    'name': 'cryptography',
-                    'version': '>=2.3',
-                    'install': "pip install 'cryptography>=2.3'"
-                }
             ]
         }
     },
-
-    'images': [
-        'static/description/images/master_screenshot.png',
-        'static/description/images/client_screenshot.png',
-        'static/description/images/login_screenshot.png',
-        'static/description/images/error_screenshot.png'
-    ]
+    "images": [
+        "static/description/images/master_screenshot.png",
+        "static/description/images/client_screenshot.png",
+        "static/description/images/login_screenshot.png",
+        "static/description/images/error_screenshot.png",
+    ],
 }

galicea_openid_connect/api.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 import json
 import logging
 from functools import wraps
@@ -9,92 +7,97 @@ import werkzeug
 
 _logger = logging.getLogger(__name__)
 
+
 class ApiException(Exception):
-    INVALID_REQUEST = 'invalid_request'
+    INVALID_REQUEST = "invalid_request"
 
     def __init__(self, message, code=None):
         super(Exception, self).__init__(message)
         self.code = code if code else self.INVALID_REQUEST
 
-def resource(path, method, auth='user'):
-    assert auth in ['user', 'client']
+
+def resource(path, method, auth="user"):
+    assert auth in ["user", "client"]
 
     def endpoint_decorator(func):
-        @http.route(path, auth='public', type='http', methods=[method, 'OPTIONS'], csrf=False)
+        @http.route(
+            path, auth="public", type="http", methods=[method, "OPTIONS"], csrf=False
+        )
         @wraps(func)
         def func_wrapper(self, req, **query):
             cors_headers = {
-                'Access-Control-Allow-Origin': '*',
-                'Access-Control-Allow-Headers': 'Origin, X-Requested-With, Content-Type, Accept, X-Debug-Mode, Authorization',
-                'Access-Control-Max-Age': 60 * 60 * 24,
+                "Access-Control-Allow-Origin": "*",
+                "Access-Control-Allow-Headers": "Origin, X-Requested-With, Content-Type, Accept, X-Debug-Mode, Authorization",
+                "Access-Control-Max-Age": 60 * 60 * 24,
             }
-            if req.httprequest.method == 'OPTIONS':
-                return http.Response(
-                    status=200,
-                    headers=cors_headers
-                )
+            if req.httprequest.method == "OPTIONS":
+                return http.Response(status=200, headers=cors_headers)
 
             try:
                 access_token = None
-                if 'Authorization' in req.httprequest.headers:
-                    authorization_header = req.httprequest.headers['Authorization']
-                    if authorization_header[:7] == 'Bearer ':
-                        access_token = authorization_header.split(' ', 1)[1]
+                if "Authorization" in req.httprequest.headers:
+                    authorization_header = req.httprequest.headers["Authorization"]
+                    if authorization_header[:7] == "Bearer ":
+                        access_token = authorization_header.split(" ", 1)[1]
                 if access_token is None:
-                    access_token = query.get('access_token')
+                    access_token = query.get("access_token")
                 if not access_token:
                     raise ApiException(
-                        'access_token param is missing',
-                        'invalid_request',
+                        "access_token param is missing", "invalid_request",
                     )
-                if auth == 'user':
-                    token = req.env['galicea_openid_connect.access_token'].sudo().search(
-                        [('token', '=', access_token)]
+                if auth == "user":
+                    token = (
+                        req.env["galicea_openid_connect.access_token"]
+                        .sudo()
+                        .search([("token", "=", access_token)])
                     )
                     if not token:
                         raise ApiException(
-                            'access_token is invalid',
-                            'invalid_request',
+                            "access_token is invalid", "invalid_request",
                         )
                     req.uid = token.user_id.id
-                elif auth == 'client':
-                    token = req.env['galicea_openid_connect.client_access_token'].sudo().search(
-                        [('token', '=', access_token)]
+                elif auth == "client":
+                    token = (
+                        req.env["galicea_openid_connect.client_access_token"]
+                        .sudo()
+                        .search([("token", "=", access_token)])
                     )
                     if not token:
                         raise ApiException(
-                            'access_token is invalid',
-                            'invalid_request',
+                            "access_token is invalid", "invalid_request",
                         )
                     req.uid = token.client_id.system_user_id.id
 
                 ctx = req.context.copy()
-                ctx.update({'client_id': token.client_id.id})
+                ctx.update({"client_id": token.client_id.id})
                 req.context = ctx
 
                 response = func(self, req, **query)
                 return werkzeug.Response(
-                    response=json.dumps(response),
-                    headers=cors_headers,
-                    status=200
+                    response=json.dumps(response), headers=cors_headers, status=200
                 )
             except ApiException as e:
                 error_message = "error: {0}".format(e)
                 return werkzeug.Response(
-                    response=json.dumps({'error': e.code, 'error_message': error_message}),
+                    response=json.dumps(
+                        {"error": e.code, "error_message": error_message}
+                    ),
                     status=400,
-                    headers=cors_headers
+                    headers=cors_headers,
                 )
             except:
-                _logger.exception('Unexpected exception while processing API request')
+                _logger.exception("Unexpected exception while processing API request")
                 return werkzeug.Response(
-                    response=json.dumps({
-                        'error': 'server_error',
-                        'error_message': 'Unexpected server error',
-                    }),
+                    response=json.dumps(
+                        {
+                            "error": "server_error",
+                            "error_message": "Unexpected server error",
+                        }
+                    ),
                     headers=cors_headers,
-                    status=500
+                    status=500,
                 )
 
         return func_wrapper
+
     return endpoint_decorator

galicea_openid_connect/controllers/__init__.py

@@ -1,4 +1,2 @@
-# -*- coding: utf-8 -*-
-
 from . import ext_web_login
 from . import main

galicea_openid_connect/controllers/ext_web_login.py

@@ -1,17 +1,15 @@
-# -*- coding: utf-8 -*-
-
 import time
 
 from odoo import http
 from odoo.addons import web
 
-class Home(web.controllers.main.Home):
 
-    @http.route('/web/login', type='http', auth="none")
+class Home(web.controllers.main.Home):
+    @http.route("/web/login", type="http", auth="none")
     def web_login(self, redirect=None, **kw):
         result = super(Home, self).web_login(redirect, **kw)
-        if result.is_qweb and 'force_auth_and_redirect' in kw:
-            result.qcontext['redirect'] = kw['force_auth_and_redirect']
-        if http.request.params.get('login_success'):
-            http.request.session['auth_time'] = int(time.time())
+        if result.is_qweb and "force_auth_and_redirect" in kw:
+            result.qcontext["redirect"] = kw["force_auth_and_redirect"]
+        if http.request.params.get("login_success"):
+            http.request.session["auth_time"] = int(time.time())
         return result

galicea_openid_connect/controllers/main.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 import json
 import time
 import os
@@ -8,7 +6,7 @@ import base64
 from odoo import http
 import werkzeug
 
-from .. api import resource
+from ..api import resource
 
 try:
     from jwcrypto import jwk, jwt
@@ -17,137 +15,144 @@ try:
 except ImportError:
     pass
 
+
 def jwk_from_json(json_key):
     key = jwk.JWK()
     key.import_key(**json.loads(json_key))
     return key
 
+
 def jwt_encode(claims, key):
     token = jwt.JWT(
-        header={'alg': key._params['alg'], 'kid': key._params['kid']},
-        claims=claims
+        header={"alg": key._params["alg"], "kid": key._params["kid"]}, claims=claims
     )
     token.make_signed_token(key)
     return token.serialize()
 
+
 def jwt_decode(serialized, key):
     token = jwt.JWT(jwt=serialized, key=key)
     return json.loads(token.claims)
 
-RESPONSE_TYPES_SUPPORTED = [
-    'code',
-    'token',
-    'id_token token',
-    'id_token'
-]
+
+RESPONSE_TYPES_SUPPORTED = ["code", "token", "id_token token", "id_token"]
+
 
 class OAuthException(Exception):
-    INVALID_REQUEST = 'invalid_request'
-    INVALID_CLIENT = 'invalid_client'
-    UNSUPPORTED_RESPONSE_TYPE = 'unsupported_response_type'
-    INVALID_GRANT = 'invalid_grant'
-    UNSUPPORTED_GRANT_TYPE = 'unsupported_grant_type'
+    INVALID_REQUEST = "invalid_request"
+    INVALID_CLIENT = "invalid_client"
+    UNSUPPORTED_RESPONSE_TYPE = "unsupported_response_type"
+    INVALID_GRANT = "invalid_grant"
+    UNSUPPORTED_GRANT_TYPE = "unsupported_grant_type"
 
     def __init__(self, message, type):
         super(Exception, self).__init__(message)
         self.type = type
 
+
 class Main(http.Controller):
     def __get_authorization_code_jwk(self, req):
-        return jwk_from_json(req.env['ir.config_parameter'].sudo().get_param(
-            'galicea_openid_connect.authorization_code_jwk'
-        ))
+        return jwk_from_json(
+            req.env["ir.config_parameter"]
+            .sudo()
+            .get_param("galicea_openid_connect.authorization_code_jwk")
+        )
 
     def __get_id_token_jwk(self, req):
-        return jwk_from_json(req.env['ir.config_parameter'].sudo().get_param(
-            'galicea_openid_connect.id_token_jwk'
-        ))
+        return jwk_from_json(
+            req.env["ir.config_parameter"]
+            .sudo()
+            .get_param("galicea_openid_connect.id_token_jwk")
+        )
 
     def __validate_client(self, req, **query):
-        if 'client_id' not in query:
+        if "client_id" not in query:
             raise OAuthException(
-                'client_id param is missing',
-                OAuthException.INVALID_CLIENT,
+                "client_id param is missing", OAuthException.INVALID_CLIENT,
             )
-        client_id = query['client_id']
-        client = req.env['galicea_openid_connect.client'].sudo().search(
-            [('client_id', '=', client_id)]
+        client_id = query["client_id"]
+        client = (
+            req.env["galicea_openid_connect.client"]
+            .sudo()
+            .search([("client_id", "=", client_id)])
         )
         if not client:
             raise OAuthException(
-                'client_id param is invalid',
-                OAuthException.INVALID_CLIENT,
+                "client_id param is invalid", OAuthException.INVALID_CLIENT,
             )
         return client
 
     def __validate_redirect_uri(self, client, req, **query):
-        if 'redirect_uri' not in query:
+        if "redirect_uri" not in query:
             raise OAuthException(
-                'redirect_uri param is missing',
-                OAuthException.INVALID_GRANT,
+                "redirect_uri param is missing", OAuthException.INVALID_GRANT,
             )
 
-        redirect_uri = query['redirect_uri']
+        redirect_uri = query["redirect_uri"]
         if client.auth_redirect_uri != redirect_uri:
             raise OAuthException(
-                'redirect_uri param doesn\'t match the pre-configured redirect URI',
+                "redirect_uri param doesn't match the pre-configured redirect URI",
                 OAuthException.INVALID_GRANT,
             )
 
         return redirect_uri
 
     def __validate_client_secret(self, client, req, **query):
-        if 'client_secret' not in query or query['client_secret'] != client.secret:
+        if "client_secret" not in query or query["client_secret"] != client.secret:
             raise OAuthException(
-                'client_secret param is not valid',
-                OAuthException.INVALID_CLIENT,
+                "client_secret param is not valid", OAuthException.INVALID_CLIENT,
             )
 
-    @http.route('/.well-known/openid-configuration', auth='public', type='http')
+    @http.route("/.well-known/openid-configuration", auth="public", type="http")
     def metadata(self, req, **query):
         base_url = http.request.httprequest.host_url
         data = {
-            'issuer': base_url,
-            'authorization_endpoint': base_url + 'oauth/authorize',
-            'token_endpoint': base_url + 'oauth/token',
-            'userinfo_endpoint': base_url + 'oauth/userinfo',
-            'jwks_uri': base_url + 'oauth/jwks',
-            'scopes_supported': ['openid'],
-            'response_types_supported': RESPONSE_TYPES_SUPPORTED,
-            'grant_types_supported': ['authorization_code', 'implicit', 'password', 'client_credentials'],
-            'subject_types_supported': ['public'],
-            'id_token_signing_alg_values_supported': ['RS256'],
-            'token_endpoint_auth_methods_supported': ['client_secret_post']
+            "issuer": base_url,
+            "authorization_endpoint": base_url + "oauth/authorize",
+            "token_endpoint": base_url + "oauth/token",
+            "userinfo_endpoint": base_url + "oauth/userinfo",
+            "jwks_uri": base_url + "oauth/jwks",
+            "scopes_supported": ["openid"],
+            "response_types_supported": RESPONSE_TYPES_SUPPORTED,
+            "grant_types_supported": [
+                "authorization_code",
+                "implicit",
+                "password",
+                "client_credentials",
+            ],
+            "subject_types_supported": ["public"],
+            "id_token_signing_alg_values_supported": ["RS256"],
+            "token_endpoint_auth_methods_supported": ["client_secret_post"],
         }
         return json.dumps(data)
 
-    @http.route('/oauth/jwks', auth='public', type='http')
+    @http.route("/oauth/jwks", auth="public", type="http")
     def jwks(self, req, **query):
         keyset = jwk.JWKSet()
         keyset.add(self.__get_id_token_jwk(req))
         return keyset.export(private_keys=False)
 
-    @resource('/oauth/userinfo', method='GET')
+    @resource("/oauth/userinfo", method="GET")
     def userinfo(self, req, **query):
         user = req.env.user
         values = {
-            'sub': str(user.id),
+            "sub": str(user.id),
             # Needed in case the client is another Odoo instance
-            'user_id': str(user.id),
-            'name': user.name,
+            "user_id": str(user.id),
+            "name": user.name,
         }
         if user.email:
-            values['email'] = user.email
+            values["email"] = user.email
         return values
 
-    @resource('/oauth/clientinfo', method='GET', auth='client')
+    @resource("/oauth/clientinfo", method="GET", auth="client")
     def clientinfo(self, req, **query):
-        client = req.env['galicea_openid_connect.client'].browse(req.context['client_id'])
-        return {
-            'name': client.name
-        }
+        client = req.env["galicea_openid_connect.client"].browse(
+            req.context["client_id"]
+        )
+        return {"name": client.name}
 
-    @http.route('/oauth/authorize', auth='public', type='http', csrf=False)
+    @http.route("/oauth/authorize", auth="public", type="http", csrf=False)
     def authorize(self, req, **query):
         # First, validate client_id and redirect_uri params.
         try:
@@ -156,131 +161,152 @@ class Main(http.Controller):
         except OAuthException as e:
             # If those are not valid, we must not redirect back to the client
             # - instead, we display a message to the user
-            return req.render('galicea_openid_connect.error', {'exception': e})
+            return req.render("galicea_openid_connect.error", {"exception": e})
 
-        scopes = query['scope'].split(' ') if query.get('scope') else []
-        is_openid_request = 'openid' in scopes
+        scopes = query["scope"].split(" ") if query.get("scope") else []
+        is_openid_request = "openid" in scopes
 
         # state, if present, is just mirrored back to the client
         response_params = {}
-        if 'state' in query:
-            response_params['state'] = query['state']
+        if "state" in query:
+            response_params["state"] = query["state"]
 
-        response_mode = query.get('response_mode')
+        response_mode = query.get("response_mode")
         try:
-            if response_mode and response_mode not in ['query', 'fragment']:
+            if response_mode and response_mode not in ["query", "fragment"]:
                 response_mode = None
                 raise OAuthException(
-                    'The only supported response_modes are \'query\' and \'fragment\'',
-                    OAuthException.INVALID_REQUEST
+                    "The only supported response_modes are 'query' and 'fragment'",
+                    OAuthException.INVALID_REQUEST,
                 )
 
-            if 'response_type' not in query:
+            if "response_type" not in query:
                 raise OAuthException(
-                    'response_type param is missing',
-                    OAuthException.INVALID_REQUEST,
+                    "response_type param is missing", OAuthException.INVALID_REQUEST,
                 )
-            response_type = query['response_type']
+            response_type = query["response_type"]
             if response_type not in RESPONSE_TYPES_SUPPORTED:
                 raise OAuthException(
-                    'The only supported response_types are: {}'.format(', '.join(RESPONSE_TYPES_SUPPORTED)),
+                    "The only supported response_types are: {}".format(
+                        ", ".join(RESPONSE_TYPES_SUPPORTED)
+                    ),
                     OAuthException.UNSUPPORTED_RESPONSE_TYPE,
                 )
         except OAuthException as e:
-            response_params['error'] = e.type
-            response_params['error_description'] = e.message
-            return self.__redirect(redirect_uri, response_params, response_mode or 'query')
+            response_params["error"] = e.type
+            response_params["error_description"] = e.message
+            return self.__redirect(
+                redirect_uri, response_params, response_mode or "query"
+            )
 
         if not response_mode:
-            response_mode = 'query' if response_type == 'code' else 'fragment'
+            response_mode = "query" if response_type == "code" else "fragment"
 
         user = req.env.user
         # In case user is not logged in, we redirect to the login page and come back
-        needs_login = user.login == 'public'
+        needs_login = user.login == "public"
         # Also if they didn't authenticate recently enough
-        if 'max_age' in query and http.request.session.get('auth_time', 0) + int(query['max_age']) < time.time():
+        if (
+            "max_age" in query
+            and http.request.session.get("auth_time", 0) + int(query["max_age"])
+            < time.time()
+        ):
             needs_login = True
         if needs_login:
             params = {
-              'force_auth_and_redirect': '/oauth/authorize?{}'.format(werkzeug.url_encode(query))
+                "force_auth_and_redirect": "/oauth/authorize?{}".format(
+                    werkzeug.url_encode(query)
+                )
             }
-            return self.__redirect('/web/login', params, 'query')
+            return self.__redirect("/web/login", params, "query")
 
         response_types = response_type.split()
 
         extra_claims = {
-            'sid': http.request.httprequest.session.sid,
+            "sid": http.request.httprequest.session.sid,
         }
-        if 'nonce' in query:
-            extra_claims['nonce'] = query['nonce']
+        if "nonce" in query:
+            extra_claims["nonce"] = query["nonce"]
 
-        if 'code' in response_types:
+        if "code" in response_types:
             # Generate code that can be used by the client server to retrieve
             # the token. It's set to be valid for 60 seconds only.
             # TODO: The spec says the code should be single-use. We're not enforcing
             # that here.
             payload = {
-                'redirect_uri': redirect_uri,
-                'client_id': client.client_id,
-                'user_id': user.id,
-                'scopes': scopes,
-                'exp': int(time.time()) + 60
+                "redirect_uri": redirect_uri,
+                "client_id": client.client_id,
+                "user_id": user.id,
+                "scopes": scopes,
+                "exp": int(time.time()) + 60,
             }
             payload.update(extra_claims)
             key = self.__get_authorization_code_jwk(req)
-            response_params['code'] = jwt_encode(payload, key)
-        if 'token' in response_types:
-            access_token = req.env['galicea_openid_connect.access_token'].sudo().retrieve_or_create(
-                user.id,
-                client.id
-            ).token
-            response_params['access_token'] = access_token
-            response_params['token_type'] = 'bearer'
+            response_params["code"] = jwt_encode(payload, key)
+        if "token" in response_types:
+            access_token = (
+                req.env["galicea_openid_connect.access_token"]
+                .sudo()
+                .retrieve_or_create(user.id, client.id)
+                .token
+            )
+            response_params["access_token"] = access_token
+            response_params["token_type"] = "bearer"
 
             digest = hashes.Hash(hashes.SHA256(), backend=default_backend())
-            digest.update(access_token.encode('ascii'))
+            digest.update(access_token.encode("ascii"))
             at_hash = digest.finalize()
-            extra_claims['at_hash'] = base64.urlsafe_b64encode(at_hash[:16]).strip('=')
-        if 'id_token' in response_types:
-            response_params['id_token'] = self.__create_id_token(req, user.id, client, extra_claims)
+            extra_claims["at_hash"] = base64.urlsafe_b64encode(at_hash[:16]).strip("=")
+        if "id_token" in response_types:
+            response_params["id_token"] = self.__create_id_token(
+                req, user.id, client, extra_claims
+            )
 
         return self.__redirect(redirect_uri, response_params, response_mode)
 
-    @http.route('/oauth/token', auth='public', type='http', methods=['POST', 'OPTIONS'], csrf=False)
+    @http.route(
+        "/oauth/token",
+        auth="public",
+        type="http",
+        methods=["POST", "OPTIONS"],
+        csrf=False,
+    )
     def token(self, req, **query):
         cors_headers = {
-            'Access-Control-Allow-Origin': '*',
-            'Access-Control-Allow-Headers': 'Origin, X-Requested-With, Content-Type, Accept, X-Debug-Mode, Authorization',
-            'Access-Control-Max-Age': 60 * 60 * 24,
+            "Access-Control-Allow-Origin": "*",
+            "Access-Control-Allow-Headers": "Origin, X-Requested-With, Content-Type, Accept, X-Debug-Mode, Authorization",
+            "Access-Control-Max-Age": 60 * 60 * 24,
         }
-        if req.httprequest.method == 'OPTIONS':
-            return http.Response(
-                status=200,
-                headers=cors_headers
-            )
+        if req.httprequest.method == "OPTIONS":
+            return http.Response(status=200, headers=cors_headers)
 
         try:
-            if 'grant_type' not in query:
+            if "grant_type" not in query:
                 raise OAuthException(
-                    'grant_type param is missing',
-                    OAuthException.INVALID_REQUEST,
+                    "grant_type param is missing", OAuthException.INVALID_REQUEST,
                 )
-            if query['grant_type'] == 'authorization_code':
-                return json.dumps(self.__handle_grant_type_authorization_code(req, **query))
-            elif query['grant_type'] == 'client_credentials':
-                return json.dumps(self.__handle_grant_type_client_credentials(req, **query))
-            elif query['grant_type'] == 'password':
+            if query["grant_type"] == "authorization_code":
+                return json.dumps(
+                    self.__handle_grant_type_authorization_code(req, **query)
+                )
+            elif query["grant_type"] == "client_credentials":
+                return json.dumps(
+                    self.__handle_grant_type_client_credentials(req, **query)
+                )
+            elif query["grant_type"] == "password":
                 return werkzeug.Response(
-                    response=json.dumps(self.__handle_grant_type_password(req, **query)),
-                    headers=cors_headers
+                    response=json.dumps(
+                        self.__handle_grant_type_password(req, **query)
+                    ),
+                    headers=cors_headers,
                 )
             else:
                 raise OAuthException(
-                    'Unsupported grant_type param: \'{}\''.format(query['grant_type']),
+                    "Unsupported grant_type param: '{}'".format(query["grant_type"]),
                     OAuthException.UNSUPPORTED_GRANT_TYPE,
                 )
         except OAuthException as e:
-            body = json.dumps({'error': e.type, 'error_description': e.message})
+            body = json.dumps({"error": e.type, "error_description": e.message})
             return werkzeug.Response(response=body, status=400, headers=cors_headers)
 
     def __handle_grant_type_authorization_code(self, req, **query):
@@ -288,124 +314,116 @@ class Main(http.Controller):
         redirect_uri = self.__validate_redirect_uri(client, req, **query)
         self.__validate_client_secret(client, req, **query)
 
-        if 'code' not in query:
+        if "code" not in query:
             raise OAuthException(
-                'code param is missing',
-                OAuthException.INVALID_GRANT,
+                "code param is missing", OAuthException.INVALID_GRANT,
             )
         try:
-            payload = jwt_decode(query['code'], self.__get_authorization_code_jwk(req))
+            payload = jwt_decode(query["code"], self.__get_authorization_code_jwk(req))
         except jwt.JWTExpired:
             raise OAuthException(
-                'Code expired',
-                OAuthException.INVALID_GRANT,
+                "Code expired", OAuthException.INVALID_GRANT,
             )
         except ValueError:
             raise OAuthException(
-                'code malformed',
-                OAuthException.INVALID_GRANT,
+                "code malformed", OAuthException.INVALID_GRANT,
             )
-        if payload['client_id'] != client.client_id:
+        if payload["client_id"] != client.client_id:
             raise OAuthException(
-                'client_id doesn\'t match the authorization request',
+                "client_id doesn't match the authorization request",
                 OAuthException.INVALID_GRANT,
             )
-        if payload['redirect_uri'] != redirect_uri:
+        if payload["redirect_uri"] != redirect_uri:
             raise OAuthException(
-                'redirect_uri doesn\'t match the authorization request',
+                "redirect_uri doesn't match the authorization request",
                 OAuthException.INVALID_GRANT,
             )
 
         # Retrieve/generate access token. We currently only store one per user/client
-        token = req.env['galicea_openid_connect.access_token'].sudo().retrieve_or_create(
-            payload['user_id'],
-            client.id
+        token = (
+            req.env["galicea_openid_connect.access_token"]
+            .sudo()
+            .retrieve_or_create(payload["user_id"], client.id)
         )
-        response = {
-            'access_token': token.token,
-            'token_type': 'bearer'
+        response = {"access_token": token.token, "token_type": "bearer"}
+        if "openid" in payload["scopes"]:
+            extra_claims = {
+                name: payload[name] for name in payload if name in ["sid", "nonce"]
             }
-        if 'openid' in payload['scopes']:
-            extra_claims = { name: payload[name] for name in payload if name in ['sid', 'nonce'] }
-            response['id_token'] = self.__create_id_token(req, payload['user_id'], client, extra_claims)
+            response["id_token"] = self.__create_id_token(
+                req, payload["user_id"], client, extra_claims
+            )
         return response
 
     def __handle_grant_type_password(self, req, **query):
         client = self.__validate_client(req, **query)
         if not client.allow_password_grant:
             raise OAuthException(
-                'This client is not allowed to perform password flow',
-                OAuthException.UNSUPPORTED_GRANT_TYPE
+                "This client is not allowed to perform password flow",
+                OAuthException.UNSUPPORTED_GRANT_TYPE,
             )
 
-        for param in ['username', 'password']:
+        for param in ["username", "password"]:
             if param not in query:
                 raise OAuthException(
-                    '{} is required'.format(param),
-                    OAuthException.INVALID_REQUEST
+                    "{} is required".format(param), OAuthException.INVALID_REQUEST
                 )
-        user_id = req.env['res.users'].authenticate(
-            req.env.cr.dbname,
-            query['username'],
-            query['password'],
-            None
+        user_id = req.env["res.users"].authenticate(
+            req.env.cr.dbname, query["username"], query["password"], None
         )
         if not user_id:
             raise OAuthException(
-                'Invalid username or password',
-                OAuthException.INVALID_REQUEST
+                "Invalid username or password", OAuthException.INVALID_REQUEST
             )
 
-        scopes = query['scope'].split(' ') if query.get('scope') else []
+        scopes = query["scope"].split(" ") if query.get("scope") else []
         # Retrieve/generate access token. We currently only store one per user/client
-        token = req.env['galicea_openid_connect.access_token'].sudo().retrieve_or_create(
-            user_id,
-            client.id
+        token = (
+            req.env["galicea_openid_connect.access_token"]
+            .sudo()
+            .retrieve_or_create(user_id, client.id)
         )
-        response = {
-            'access_token': token.token,
-            'token_type': 'bearer'
-        }
-        if 'openid' in scopes:
-            response['id_token'] = self.__create_id_token(req, user_id, client, {})
+        response = {"access_token": token.token, "token_type": "bearer"}
+        if "openid" in scopes:
+            response["id_token"] = self.__create_id_token(req, user_id, client, {})
         return response
 
     def __handle_grant_type_client_credentials(self, req, **query):
         client = self.__validate_client(req, **query)
         self.__validate_client_secret(client, req, **query)
-        token = req.env['galicea_openid_connect.client_access_token'].sudo().retrieve_or_create(client.id)
-        return {
-            'access_token': token.token,
-            'token_type': 'bearer'
-        }
+        token = (
+            req.env["galicea_openid_connect.client_access_token"]
+            .sudo()
+            .retrieve_or_create(client.id)
+        )
+        return {"access_token": token.token, "token_type": "bearer"}
 
     def __create_id_token(self, req, user_id, client, extra_claims):
         claims = {
-            'iss': http.request.httprequest.host_url,
-            'sub': str(user_id),
-            'aud': client.client_id,
-            'iat': int(time.time()),
-            'exp': int(time.time()) + 15 * 60
+            "iss": http.request.httprequest.host_url,
+            "sub": str(user_id),
+            "aud": client.client_id,
+            "iat": int(time.time()),
+            "exp": int(time.time()) + 15 * 60,
         }
-        auth_time = extra_claims.get('sid') and http.root.session_store.get(extra_claims['sid']).get('auth_time')
+        auth_time = extra_claims.get("sid") and http.root.session_store.get(
+            extra_claims["sid"]
+        ).get("auth_time")
         if auth_time:
-            claims['auth_time'] = auth_time
-        if 'nonce' in extra_claims:
-            claims['nonce'] = extra_claims['nonce']
-        if 'at_hash' in extra_claims:
-            claims['at_hash'] = extra_claims['at_hash']
+            claims["auth_time"] = auth_time
+        if "nonce" in extra_claims:
+            claims["nonce"] = extra_claims["nonce"]
+        if "at_hash" in extra_claims:
+            claims["at_hash"] = extra_claims["at_hash"]
 
         key = self.__get_id_token_jwk(req)
         return jwt_encode(claims, key)
 
     def __redirect(self, url, params, response_mode):
-        location = '{}{}{}'.format(
-            url,
-            '?' if response_mode == 'query' else '#',
-            werkzeug.url_encode(params)
+        location = "{}{}{}".format(
+            url, "?" if response_mode == "query" else "#", werkzeug.url_encode(params)
         )
         return werkzeug.Response(
-            headers={'Location': location},
-            response=None,
-            status=302,
+            headers={"Location": location}, response=None, status=302,
         )
+

galicea_openid_connect/models/__init__.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 from . import client
 from . import access_token
 from . import config_parameter

galicea_openid_connect/models/access_token.py

@@ -1,10 +1,9 @@
-# -*- coding: utf-8 -*-
-
 from odoo import models, fields, api
 from .. import random_tokens
 
+
 class AccessTokenBase(models.AbstractModel):
-    _name = 'galicea_openid_connect.access_token_base'
+    _name = "galicea_openid_connect.access_token_base"
 
     token = fields.Char(
         readonly=True,
@@ -13,52 +12,43 @@ class AccessTokenBase(models.AbstractModel):
         index=True,
     )
     client_id = fields.Many2one(
-        'galicea_openid_connect.client',
+        "galicea_openid_connect.client",
         readonly=True,
         index=True,
         required=True,
-        ondelete='cascade'
+        ondelete="cascade",
     )
 
+
 class AccessToken(models.Model):
-    _inherit = 'galicea_openid_connect.access_token_base'
-    _name = 'galicea_openid_connect.access_token'
-    _description = 'Acccess token representing user-client pair'
+    _inherit = "galicea_openid_connect.access_token_base"
+    _name = "galicea_openid_connect.access_token"
+    _description = "Acccess token representing user-client pair"
 
     user_id = fields.Many2one(
-        'res.users',
-        required=True,
-        readonly=True,
-        index=True,
-        ondelete='cascade'
+        "res.users", required=True, readonly=True, index=True, ondelete="cascade"
     )
 
     @api.model
     def retrieve_or_create(self, user_id, client_id):
         existing_tokens = self.search(
-            [
-                ('user_id', '=', user_id),
-                ('client_id', '=', client_id),
-            ]
+            [("user_id", "=", user_id), ("client_id", "=", client_id),]
         )
         if existing_tokens:
             return existing_tokens[0]
         else:
-            return self.create({'user_id': user_id, 'client_id': client_id})
+            return self.create({"user_id": user_id, "client_id": client_id})
+
 
 class ClientAccessToken(models.Model):
-    _inherit = 'galicea_openid_connect.access_token_base'
-    _name = 'galicea_openid_connect.client_access_token'
-    _description = 'Access token representing client credentials'
+    _inherit = "galicea_openid_connect.access_token_base"
+    _name = "galicea_openid_connect.client_access_token"
+    _description = "Access token representing client credentials"
 
     @api.model
     def retrieve_or_create(self, client_id):
-        existing_tokens = self.search(
-            [
-                ('client_id', '=', client_id),
-            ]
-        )
+        existing_tokens = self.search([("client_id", "=", client_id),])
         if existing_tokens:
             return existing_tokens[0]
         else:
-            return self.create({'client_id': client_id})
+            return self.create({"client_id": client_id})

galicea_openid_connect/models/client.py

@@ -1,64 +1,66 @@
-# -*- coding: utf-8 -*-
-
 from odoo import models, fields, api
 from .. import random_tokens
 
+
 class Client(models.Model):
-    _name = 'galicea_openid_connect.client'
-    _description = 'OpenID Connect client'
+    _name = "galicea_openid_connect.client"
+    _description = "OpenID Connect client"
 
     name = fields.Char(required=True)
-    auth_redirect_uri = fields.Char('Redirect URI for user login')
+    auth_redirect_uri = fields.Char("Redirect URI for user login")
     client_id = fields.Char(
-        string='Client ID',
+        string="Client ID",
         required=True,
         readonly=True,
         index=True,
         default=lambda _: random_tokens.lower_case(16),
     )
     secret = fields.Char(
-        string='Client secret',
+        string="Client secret",
         required=True,
         readonly=True,
         default=lambda _: random_tokens.alpha_numeric(32),
-        groups='galicea_openid_connect.group_admin'
+        groups="galicea_openid_connect.group_admin",
     )
     system_user_id = fields.Many2one(
-        'res.users',
-        'Artificial user representing the client in client credentials requests',
+        "res.users",
+        "Artificial user representing the client in client credentials requests",
         readonly=True,
         required=True,
-        ondelete='restrict'
+        ondelete="restrict",
     )
     allow_password_grant = fields.Boolean(
-        string='Allow OAuth2 password grant',
-        default=False,
+        string="Allow OAuth2 password grant", default=False,
     )
 
     @api.model
     def __system_user_name(self, client_name):
-        return '{} - API system user'.format(client_name)
+        return "{} - API system user".format(client_name)
 
     @api.model
     def create(self, values):
-        if 'name' in values:
-            system_user = self.env['res.users'].create({
-                'name': self.__system_user_name(values['name']),
-                'login': random_tokens.lower_case(8),
-                'groups_id': [(4, self.env.ref('galicea_openid_connect.group_system_user').id)]
-            })
+        if "name" in values:
+            system_user = self.env["res.users"].create(
+                {
+                    "name": self.__system_user_name(values["name"]),
+                    "login": random_tokens.lower_case(8),
+                    "groups_id": [
+                        (4, self.env.ref("galicea_openid_connect.group_system_user").id)
+                    ],
+                }
+            )
             # Do not include in the "Pending invitations" list
             system_user.sudo(system_user.id)._update_last_login()
-            values['system_user_id'] = system_user.id
+            values["system_user_id"] = system_user.id
         return super(Client, self).create(values)
 
     @api.multi
     def write(selfs, values):
         super(Client, selfs).write(values)
-        if 'name' in values:
-            selfs.mapped(lambda client: client.system_user_id).write({
-                'name': selfs.__system_user_name(values['name'])
-            })
+        if "name" in values:
+            selfs.mapped(lambda client: client.system_user_id).write(
+                {"name": selfs.__system_user_name(values["name"])}
+            )
         return True
 
     @api.multi

galicea_openid_connect/models/config_parameter.py

@@ -1,28 +1,40 @@
-# -*- coding: utf-8 -*-
-
 from odoo import models, fields, api
 from .. import random_tokens
+
 try:
     from jwcrypto import jwk
 except ImportError:
     pass
 
+
 class ConfigParameter(models.Model):
-  _inherit = 'ir.config_parameter'
+    _inherit = "ir.config_parameter"
 
     @api.model
     def openid_init_keys(self):
         keys = {
-        'galicea_openid_connect.authorization_code_jwk': lambda: \
-            jwk.JWK.generate(kty='oct', size=256, kid=random_tokens.alpha_numeric(16), use='sig', alg='HS256').export(),
-        'galicea_openid_connect.id_token_jwk': lambda: \
-            jwk.JWK.generate(kty='RSA', size=2054, kid=random_tokens.alpha_numeric(16), use='sig', alg='RS256').export()
+            "galicea_openid_connect.authorization_code_jwk": lambda: jwk.JWK.generate(
+                kty="oct",
+                size=256,
+                kid=random_tokens.alpha_numeric(16),
+                use="sig",
+                alg="HS256",
+            ).export(),
+            "galicea_openid_connect.id_token_jwk": lambda: jwk.JWK.generate(
+                kty="RSA",
+                size=2054,
+                kid=random_tokens.alpha_numeric(16),
+                use="sig",
+                alg="RS256",
+            ).export(),
         }
 
         for key, gen in iter(keys.items()):
-        if not self.search([('key', '=', key)]):
-            self.create({
-                'key': key,
-                'value': gen(),
-                'group_ids': [(4, self.env.ref('base.group_erp_manager').id)]
-            })
+            if not self.search([("key", "=", key)]):
+                self.create(
+                    {
+                        "key": key,
+                        "value": gen(),
+                        "group_ids": [(4, self.env.ref("base.group_erp_manager").id)],
+                    }
+                )

galicea_openid_connect/random_tokens.py

@@ -1,14 +1,15 @@
-# -*- coding: utf-8 -*-
-
 from random import SystemRandom
 
+
 def random_token(length, byte_filter):
-    allowed_bytes = ''.join(c for c in map(chr, range(128)) if byte_filter(c))
+    allowed_bytes = "".join(c for c in map(chr, range(128)) if byte_filter(c))
     random = SystemRandom()
-    return ''.join([random.choice(allowed_bytes) for _ in range(length)])
+    return "".join([random.choice(allowed_bytes) for _ in range(length)])
+
 
 def alpha_numeric(length):
     return random_token(length, str.isalnum)
 
+
 def lower_case(length):
     return random_token(length, str.islower)

galicea_openid_connect/security/__init__.py

@@ -1,23 +1,37 @@
-# -*- coding: utf-8 -*-
-
 from .. import random_tokens
+
 try:
     from jwcrypto import jwk
 except ImportError:
     pass
 
+
 def init_keys(IrConfigParameter):
     keys = {
-        'galicea_openid_connect.authorization_code_jwk': lambda: \
-            jwk.JWK.generate(kty='oct', size=256, kid=random_tokens.alpha_numeric(16), use='sig', alg='HS256').export(),
-        'galicea_openid_connect.id_token_jwk': lambda: \
-            jwk.JWK.generate(kty='RSA', size=2054, kid=random_tokens.alpha_numeric(16), use='sig', alg='RS256').export()
+        "galicea_openid_connect.authorization_code_jwk": lambda: jwk.JWK.generate(
+            kty="oct",
+            size=256,
+            kid=random_tokens.alpha_numeric(16),
+            use="sig",
+            alg="HS256",
+        ).export(),
+        "galicea_openid_connect.id_token_jwk": lambda: jwk.JWK.generate(
+            kty="RSA",
+            size=2054,
+            kid=random_tokens.alpha_numeric(16),
+            use="sig",
+            alg="RS256",
+        ).export(),
     }
 
     for key, gen in keys.iteritems():
-        if not IrConfigParameter.search([('key', '=', key)]):
-            IrConfigParameter.create({
-                'key': key,
-                'value': gen(),
-                'group_ids': [(4, IrConfigParameter.env.ref('base.group_erp_manager').id)]
-            })
+        if not IrConfigParameter.search([("key", "=", key)]):
+            IrConfigParameter.create(
+                {
+                    "key": key,
+                    "value": gen(),
+                    "group_ids": [
+                        (4, IrConfigParameter.env.ref("base.group_erp_manager").id)
+                    ],
+                }
+            )

galicea_openid_connect/system_checks.py

@@ -1,24 +1,24 @@
-# -*- coding: utf-8 -*-
-
-from odoo.addons.galicea_environment_checkup import \
-        custom_check, CheckWarning, CheckSuccess, CheckFail
+from odoo.addons.galicea_environment_checkup import (
+    custom_check,
+    CheckWarning,
+    CheckSuccess,
+    CheckFail,
+)
 
 from odoo import http
 
+
 @custom_check
 def check_single_db(env):
     if not http.request:
-        raise CheckWarning('Could not detect DB settings.')
+        raise CheckWarning("Could not detect DB settings.")
 
     dbs = http.db_list(True, http.request.httprequest)
     if len(dbs) == 1:
-        return CheckSuccess('Odoo runs in a single-DB mode.')
+        return CheckSuccess("Odoo runs in a single-DB mode.")
 
     details = (
-        '<p>Odoo runs in a multi-DB mode, which will cause API request routing to fail.</p>'
-        '<p>Run Odoo with <tt>--dbfilter</tt> or <tt>--database</tt> flag.</p>'
-    )
-    return CheckFail(
-        'Odoo runs in a multi-DB mode.',
-        details=details
+        "<p>Odoo runs in a multi-DB mode, which will cause API request routing to fail.</p>"
+        "<p>Run Odoo with <tt>--dbfilter</tt> or <tt>--database</tt> flag.</p>"
     )
+    return CheckFail("Odoo runs in a multi-DB mode.", details=details)

galicea_toolset/__manifest__.py

@@ -1,20 +1,13 @@
 # -*- coding: utf-8 -*-
 {
-    'name': "galicea toolset",
-
-    'summary': """
+    "name": "galicea toolset",
+    "summary": """
         A couple of small convenience widgets and functions""",
-
-    'author': "Maciej Wawro",
-    'maintainer': "Galicea",
-    'website': "http://www.galicea.pl",
-
-    'category': 'Technical Settings',
-    'version': '12.0.0.1',
-
-    'depends': ['base'],
-
-    'data': [
-        'views/data.xml'
-    ],
+    "author": "Maciej Wawro",
+    "maintainer": "Galicea",
+    "website": "http://www.galicea.pl",
+    "category": "Technical Settings",
+    "version": "13.0.0.1",
+    "depends": ["base"],
+    "data": ["views/data.xml"],
 }

galicea_toolset/utils.py

@@ -1,7 +1,6 @@
-# -*- coding: utf-8 -*-
-
 from odoo import http
 
+
 def get_base_url(env):
     """
     Better host name detection
@@ -13,4 +12,4 @@ def get_base_url(env):
     else:
         # Jeśli nie jesteśmy wewnątrz zapytania HTTP, zwróć domenę ostatnio użytą
         # przez admina do zalogowania
-        return env['ir.config_parameter'].get_param('web.base.url') + '/'
+        return env["ir.config_parameter"].get_param("web.base.url") + "/"