[14.0][PORT] backport fastapi_auth_jwt, fastapi_auth_jwt_demo from 16.0

.github/workflows/test.yml

@@ -50,7 +50,7 @@ jobs:
         ports:
           - 5432:5432
     env:
-      EXCLUDE: extendable,extendable_fastapi,fastapi
+      EXCLUDE: extendable,extendable_fastapi,fastapi,fastapi_auth_jwt,fastapi_auth_jwt_demo
     steps:
       - uses: actions/checkout@v2
         with:
@@ -92,7 +92,7 @@ jobs:
         ports:
           - 5432:5432
     env:
-      INCLUDE: extendable,extendable_fastapi,fastapi
+      INCLUDE: extendable,extendable_fastapi,fastapi,fastapi_auth_jwt,fastapi_auth_jwt_demo
     steps:
       - uses: actions/checkout@v2
         with:

fastapi/tests/test_fastapi.py

@@ -1,13 +1,10 @@
 # Copyright 2022 ACSONE SA/NV
 # License LGPL-3.0 or later (http://www.gnu.org/licenses/LGPL).
 
-import os
-import unittest
+from odoo.tests.common import HttpSavepointCase, tagged
 
-from odoo.tests.common import HttpSavepointCase
 
-
-@unittest.skipIf(os.getenv("SKIP_HTTP_CASE"), "EndpointHttpCase skipped")
+@tagged("post_install", "-at_install")
 class FastAPIHttpCase(HttpSavepointCase):
     @classmethod
     def setUpClass(cls):
@@ -33,6 +30,12 @@ def test_call(self):
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response.content, b'{"Hello":"World"}')
 
+    def test_post(self):
+        route = "/fastapi_demo/post_demo"
+        response = self.url_open(route, data={"foo": "bar"})
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.content, b'{"foo":"bar"}')
+
     def test_lang(self):
         self._assert_expected_lang("fr,en;q=0.7,en-GB;q=0.3", b'"fr_BE"')
         self._assert_expected_lang("en,fr;q=0.7,en-GB;q=0.3", b'"en_US"')

fastapi_auth_jwt/__manifest__.py

@@ -0,0 +1,19 @@
+# Copyright 2023 ACSONE SA/NV
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
+
+{
+    "name": "FastAPI Auth JWT support",
+    "summary": """
+        JWT bearer token authentication for FastAPI.""",
+    "version": "14.0.1.0.0",
+    "license": "LGPL-3",
+    "author": "ACSONE SA/NV,Odoo Community Association (OCA)",
+    "maintainers": ["sbidoul"],
+    "website": "https://github.com/OCA/rest-framework",
+    "depends": [
+        "fastapi",
+        "auth_jwt",
+    ],
+    "data": [],
+    "demo": [],
+}

fastapi_auth_jwt/dependencies.py

@@ -0,0 +1,254 @@
+# Copyright 2023 ACSONE SA/NV
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
+
+import logging
+import sys
+from typing import Any, Dict, Optional, Tuple, Union
+
+from starlette.status import HTTP_401_UNAUTHORIZED
+
+from odoo.api import Environment
+
+from odoo.addons.auth_jwt.exceptions import (
+    ConfigurationError,
+    Unauthorized,
+    UnauthorizedCompositeJwtError,
+    UnauthorizedMissingAuthorizationHeader,
+    UnauthorizedMissingCookie,
+)
+from odoo.addons.auth_jwt.models.auth_jwt_validator import AuthJwtValidator
+from odoo.addons.base.models.res_partner import Partner
+from odoo.addons.fastapi.dependencies import odoo_env
+
+from fastapi import Depends, HTTPException, Request, Response
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+if sys.version_info >= (3, 9):
+    from typing import Annotated
+else:
+    from typing_extensions import Annotated
+
+_logger = logging.getLogger(__name__)
+
+
+Payload = Dict[str, Any]
+
+
+def _get_auth_jwt_validator(
+    validator_name: Union[str, None],
+    env: Environment,
+) -> AuthJwtValidator:
+    validator = env["auth.jwt.validator"].sudo()._get_validator_by_name(validator_name)
+    assert len(validator) == 1
+    return validator
+
+
+def _request_has_authentication(
+    request: Request,
+    authorization_header: Optional[str],
+    validator: AuthJwtValidator,
+) -> Union[Payload, None]:
+    if authorization_header is not None:
+        return True
+    if not validator.cookie_enabled:
+        # no Authorization header and cookies not enabled
+        return False
+    return request.cookies.get(validator.cookie_name) is not None
+
+
+def _get_jwt_payload(
+    request: Request,
+    authorization_header: Optional[str],
+    validator: AuthJwtValidator,
+) -> Payload:
+    """Obtain and validate the JWT payload from the request authorization header or
+    cookie (if enabled on the validator)."""
+    if authorization_header is not None:
+        return validator._decode(authorization_header)
+    if not validator.cookie_enabled:
+        _logger.info("Missing or malformed authorization header.")
+        raise UnauthorizedMissingAuthorizationHeader()
+    assert validator.cookie_name
+    cookie_token = request.cookies.get(validator.cookie_name)
+    if not cookie_token:
+        _logger.info(
+            "Missing or malformed authorization header, and %s cookie not present.",
+            validator.cookie_name,
+        )
+        raise UnauthorizedMissingCookie()
+    return validator._decode(cookie_token, secret=validator._get_jwt_cookie_secret())
+
+
+def _get_jwt_payload_and_validator(
+    request: Request,
+    response: Response,
+    authorization_header: Optional[str],
+    validator: AuthJwtValidator,
+) -> Tuple[Payload, AuthJwtValidator]:
+    try:
+        payload = None
+        exceptions = {}
+        while validator:
+            try:
+                payload = _get_jwt_payload(request, authorization_header, validator)
+                break
+            except Unauthorized as e:
+                exceptions[validator.name] = e
+                validator = validator.next_validator_id
+
+        if not payload:
+            if len(exceptions) == 1:
+                raise list(exceptions.values())[0]
+            raise UnauthorizedCompositeJwtError(exceptions)
+
+        if validator.cookie_enabled:
+            if not validator.cookie_name:
+                _logger.info("Cookie name not set for validator %s", validator.name)
+                raise ConfigurationError()
+            response.set_cookie(
+                key=validator.cookie_name,
+                value=validator._encode(
+                    payload,
+                    secret=validator._get_jwt_cookie_secret(),
+                    expire=validator.cookie_max_age,
+                ),
+                max_age=validator.cookie_max_age,
+                path=validator.cookie_path or "/",
+                secure=validator.cookie_secure,
+                httponly=True,
+            )
+
+        return payload, validator
+    except Unauthorized as e:
+        raise HTTPException(status_code=HTTP_401_UNAUTHORIZED) from e
+
+
+def auth_jwt_default_validator_name() -> Union[str, None]:
+    return None
+
+
+def auth_jwt_http_header_authorization(
+    credentials: Annotated[
+        Optional[HTTPAuthorizationCredentials],
+        Depends(HTTPBearer(auto_error=False)),
+    ]
+):
+    if credentials is None:
+        return None
+    return credentials.credentials
+
+
+class BaseAuthJwt:  # noqa: B903
+    def __init__(
+        self, validator_name: Optional[str] = None, allow_unauthenticated: bool = False
+    ):
+        self.validator_name = validator_name
+        self.allow_unauthenticated = allow_unauthenticated
+
+
+class AuthJwtPayload(BaseAuthJwt):
+    def __call__(
+        self,
+        request: Request,
+        response: Response,
+        authorization_header: Annotated[
+            Optional[str],
+            Depends(auth_jwt_http_header_authorization),
+        ],
+        default_validator_name: Annotated[
+            Union[str, None],
+            Depends(auth_jwt_default_validator_name),
+        ],
+        env: Annotated[
+            Environment,
+            Depends(odoo_env),
+        ],
+    ) -> Optional[Payload]:
+        validator = _get_auth_jwt_validator(
+            self.validator_name or default_validator_name, env
+        )
+        if self.allow_unauthenticated and not _request_has_authentication(
+            request, authorization_header, validator
+        ):
+            return None
+        return _get_jwt_payload_and_validator(
+            request, response, authorization_header, validator
+        )[0]
+
+
+class AuthJwtPartner(BaseAuthJwt):
+    def __call__(
+        self,
+        request: Request,
+        response: Response,
+        authorization_header: Annotated[
+            Optional[str],
+            Depends(auth_jwt_http_header_authorization),
+        ],
+        default_validator_name: Annotated[
+            Union[str, None],
+            Depends(auth_jwt_default_validator_name),
+        ],
+        env: Annotated[
+            Environment,
+            Depends(odoo_env),
+        ],
+    ) -> Partner:
+        validator = _get_auth_jwt_validator(
+            self.validator_name or default_validator_name, env
+        )
+        if self.allow_unauthenticated and not _request_has_authentication(
+            request, authorization_header, validator
+        ):
+            return env["res.partner"].with_user(env.ref("base.public_user")).browse()
+        payload, validator = _get_jwt_payload_and_validator(
+            request, response, authorization_header, validator
+        )
+        try:
+            uid = validator._get_and_check_uid(payload)
+            partner_id = validator._get_and_check_partner_id(payload)
+        except Unauthorized as e:
+            raise HTTPException(status_code=HTTP_401_UNAUTHORIZED) from e
+        if not partner_id:
+            _logger.info("Could not determine partner from JWT payload.")
+            raise HTTPException(status_code=HTTP_401_UNAUTHORIZED)
+        return env["res.partner"].with_user(uid).browse(partner_id)
+
+
+class AuthJwtOdooEnv(BaseAuthJwt):
+    def __call__(
+        self,
+        request: Request,
+        response: Response,
+        authorization_header: Annotated[
+            Optional[str],
+            Depends(auth_jwt_http_header_authorization),
+        ],
+        default_validator_name: Annotated[
+            Union[str, None],
+            Depends(auth_jwt_default_validator_name),
+        ],
+        env: Annotated[
+            Environment,
+            Depends(odoo_env),
+        ],
+    ) -> Environment:
+        validator = _get_auth_jwt_validator(
+            self.validator_name or default_validator_name, env
+        )
+        payload, validator = _get_jwt_payload_and_validator(
+            request, response, authorization_header, validator
+        )
+        uid = validator._get_and_check_uid(payload)
+        return odoo_env(user=uid)
+
+
+auth_jwt_authenticated_payload = AuthJwtPayload()
+
+auth_jwt_optionally_authenticated_payload = AuthJwtPayload(allow_unauthenticated=True)
+
+auth_jwt_authenticated_partner = AuthJwtPartner()
+
+auth_jwt_optionally_authenticated_partner = AuthJwtPartner(allow_unauthenticated=True)
+
+auth_jwt_authenticated_odoo_env = AuthJwtOdooEnv()

fastapi_auth_jwt/readme/DESCRIPTION.rst

@@ -0,0 +1,2 @@
+This module provides ``FastAPI`` ``Depends`` to allow authentication with `auth_jwt
+<https://github.com/OCA/server-auth/tree/16.0/auth_jwt>`_.

fastapi_auth_jwt/readme/USAGE.rst

@@ -0,0 +1,56 @@
+The following FastAPI dependencies  are provided and importable from
+``odoo.addons.fastapi_auth_jwt.dependencies``:
+
+``def auth_jwt_authenticated_payload() -> Payload``
+
+  Return the authenticated JWT payload. Raise a 401 (unauthorized) if absent or invalid.
+
+``def auth_jwt_optionally_authenticated_payload() -> Payload | None``
+
+  Return the authenticated JWT payload, or ``None`` if the ``Authorization`` header and
+  cookie are absent. Raise a 401 (unauthorized) if present and invalid.
+
+``def auth_jwt_authenticated_partner() -> Partner``
+
+  Obtain the authenticated partner corresponding to the provided JWT token, according to
+  the partner strategy defined on the ``auth_jwt`` validator. Raise a 401 (unauthorized)
+  if the partner could not be determined for any reason.
+
+  This is function suitable and intended to override
+  ``odoo.addons.fastapi.dependencies.authenticated_partner_impl``.
+
+  The partner record returned by this function is bound to an environment that uses the
+  Odoo user obtained from the user strategy defined on the ``auth_jwt`` validator. When
+  used ``authenticated_partner_impl`` this in turn ensures that
+  ``odoo.addons.fastapi.dependencies.authenticated_partner_env`` is also bound to the
+  correct Odoo user.
+
+``def auth_jwt_optionally_authenticated_partner() -> Partner``
+
+  Same as ``auth_jwt_partner`` except it returns an empty recordset bound to the
+  ``public`` user if the ``Authorization`` header and cookie are absent, or if the JWT
+  validator could not find the partner and declares that the partner is not required.
+
+``def auth_jwt_authenticated_odoo_env() -> Environment``
+
+  Return an Odoo environment using the the Odoo user obtained from the user strategy
+  defined on the ``auth_jwt`` validator, if the request could be authenticated using a
+  JWT validator. Raise a 401 (unauthorized) otherwise.
+
+  This is function suitable and intended to override
+  ``odoo.addons.fastapi.dependencies.authenticated_odoo_env_impl``.
+
+``def auth_jwt_default_validator_name() -> str | None``
+
+  Return the name of the default JWT validator to use.
+
+  The default implementation returns ``None`` meaning only one active JWT validator is
+  allowed. This dependency is meant to be overridden.
+
+``def auth_jwt_http_header_authorization() -> str | None``
+
+  By default, return the credentials part of the ``Authorization`` header, or ``None``
+  if absent. This dependency is meant to be overridden, in particular with
+  ``fastapi.security.OAuth2AuthorizationCodeBearer`` to let swagger handle OAuth2
+  authorization (such override is only necessary for comfort when using the swagger
+  interface).

fastapi_auth_jwt_demo/__init__.py

@@ -0,0 +1,2 @@
+from . import models
+from . import routers