16.0 fastapi auth jwt lmi

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": "16.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,235 @@
+# Copyright 2023 ACSONE SA/NV
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
+
+import logging
+from typing import Annotated, 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
+
+_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_credentials: Optional[HTTPAuthorizationCredentials],
+    validator: AuthJwtValidator,
+) -> Union[Payload, None]:
+    if authorization_credentials 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[HTTPAuthorizationCredentials],
+    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.credentials)
+    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 authorization cookie %s.", 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[HTTPAuthorizationCredentials],
+    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
+
+
+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[HTTPAuthorizationCredentials],
+            Depends(HTTPBearer(auto_error=False)),
+        ],
+        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[HTTPAuthorizationCredentials],
+            Depends(HTTPBearer(auto_error=False)),
+        ],
+        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
+        )
+        uid = validator._get_and_check_uid(payload)
+        partner_id = validator._get_and_check_partner_id(payload)
+        if not partner_id:
+            if self.allow_unauthenticated:
+                return (
+                    env["res.partner"].with_user(env.ref("base.public_user")).browse()
+                )
+            _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[HTTPAuthorizationCredentials],
+            Depends(HTTPBearer(auto_error=False)),
+        ],
+        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,48 @@
+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.

fastapi_auth_jwt_demo/__init__.py

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

fastapi_auth_jwt_demo/__manifest__.py

@@ -0,0 +1,16 @@
+# Copyright 2023 ACSONE SA/NV
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
+
+{
+    "name": "FastAPI Auth JWT Test",
+    "summary": """
+        Test/demo module for fastapi_auth_jwt.""",
+    "version": "16.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", "auth_jwt_demo"],
+    "data": [],
+    "demo": ["demo/fastapi_endpoint.xml"],
+}

fastapi_auth_jwt_demo/demo/fastapi_endpoint.xml

@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<odoo>
+    <record model="fastapi.endpoint" id="fastapi_endpoint_auth_jwt_demo">
+        <field name="name">Auth JWT Fastapi Demo Endpoint</field>
+        <field name="app">auth_jwt_demo</field>
+        <field name="root_path">/fastapi_auth_jwt_demo</field>
+        <field name="user_id" ref="fastapi.my_demo_app_user" />
+    </record>
+</odoo>

fastapi_auth_jwt_demo/models/__init__.py

@@ -0,0 +1 @@
+from .fastapi_endpoint import FastapiEndpoint, auth_jwt_demo_api_router

fastapi_auth_jwt_demo/models/fastapi_endpoint.py

@@ -0,0 +1,24 @@
+# Copyright 2023 ACSONE SA/NV
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
+
+from odoo import api, fields, models
+
+from ..routers.auth_jwt_demo_api import router as auth_jwt_demo_api_router
+
+APP_NAME = "auth_jwt_demo"
+
+
+class FastapiEndpoint(models.Model):
+
+    _inherit = "fastapi.endpoint"
+
+    app: str = fields.Selection(
+        selection_add=[(APP_NAME, "Auth JWT Demo Endpoint")],
+        ondelete={APP_NAME: "cascade"},
+    )
+
+    @api.model
+    def _get_fastapi_routers(self):
+        if self.app == APP_NAME:
+            return [auth_jwt_demo_api_router]
+        return super()._get_fastapi_routers()

fastapi_auth_jwt_demo/readme/DESCRIPTION.rst

@@ -0,0 +1,4 @@
+Tests and demo routes for ``fastapi_auth_jwt``.
+
+The tests and routes are almost identical to those in ``auth_jwt_demo``, and
+the JWT validators used are those from ``auth_jwt_demo``.

fastapi_auth_jwt_demo/routers/__init__.py

@@ -0,0 +1 @@
+from .auth_jwt_demo_api import router