# Part of Odoo. See LICENSE file for full copyright and licensing details.

r"""\

Odoo HTTP layer / WSGI application

The main duty of this module is to prepare and dispatch all http

requests to their corresponding controllers: from a raw http request

arriving on the WSGI entrypoint to a :class:`~http.Request`: arriving at

a module controller with a fully setup ORM available.

Application developers mostly know this module thanks to the

:class:`~odoo.http.Controller`: class and its companion the

:func:`~odoo.http.route`: method decorator. Together they are used to

register methods responsible of delivering web content to matching URLS.

Those two are only the tip of the iceberg, below is a call graph that

shows the various processing layers each request passes through before

ending at the @route decorated endpoint. Hopefully, this call graph and

the attached function descriptions will help you understand this module.

Here be dragons:

Application.__call__

if path is like '/<module>/static/<path>':

Request._serve_static

elif not request.db:

Request._serve_nodb

App.nodb_routing_map.match

Dispatcher.pre_dispatch

Dispatcher.dispatch

route_wrapper

endpoint

Dispatcher.post_dispatch

else:

Request._serve_db

env['ir.http']._match

if not match:

model.retrying(Request._serve_ir_http_fallback)

env['ir.http']._serve_fallback

env['ir.http']._post_dispatch

else:

model.retrying(Request._serve_ir_http)

env['ir.http']._authenticate

env['ir.http']._pre_dispatch

Dispatcher.pre_dispatch

Dispatcher.dispatch

env['ir.http']._dispatch

route_wrapper

endpoint

env['ir.http']._post_dispatch

Application.__call__

WSGI entry point, it sanitizes the request, it wraps it in a werkzeug

request and itself in an Odoo http request. The Odoo http request is

exposed at ``http.request`` then it is forwarded to either

``_serve_static``, ``_serve_nodb`` or ``_serve_db`` depending on the

request path and the presence of a database. It is also responsible of

ensuring any error is properly logged and encapsuled in a HTTP error

response.

Request._serve_static

Handle all requests to ``/<module>/static/<asset>`` paths, open the

underlying file on the filesystem and stream it via

:meth:``Request.send_file``

Request._serve_nodb

Handle requests to ``@route(auth='none')`` endpoints when the user is

not connected to a database. It performs limited operations, just

matching the auth='none' endpoint using the request path and then it

delegates to Dispatcher.

Request._serve_db

Handle all requests that are not static when it is possible to connect

to a database. It opens a registry on the database, manage the request

cursor and environment. The function decides whether to use a

read-only or a read/write cursor for its operations:

``check_signaling``, ``match`` and ``serve_fallback`` are called using

the same read-only cursor; ``_serve_ir_http`` is called reusing the

same (but reset) read-only cursor, or a new read/write one.

service.model.retrying

Manage the cursor, the environment and exceptions that occured while

executing the underlying function. They recover from various

exceptions such as serialization errors and writes in read-only

transactions. They catches all other exceptions and attach a http

response to them (e.g. 500 - Internal Server Error)

ir.http._match

Match the controller endpoint that correspond to the request path.

Beware that there is an important override for portal and website

inside of the ``http_routing`` module.

ir.http._serve_fallback

Find alternative ways to serve a request when its path does not match

any controller. The path could be matching an attachment URL, a blog

page, etc.

ir.http._authenticate

Ensure the user on the current environment fulfill the requirement of

``@route(auth=...)``. Using the ORM outside of abstract models is

unsafe prior of calling this function.

ir.http._pre_dispatch/Dispatcher.pre_dispatch

Prepare the system the handle the current request, often used to save

some extra query-string parameters in the session (e.g. ?debug=1)

ir.http._dispatch/Dispatcher.dispatch

Deserialize the HTTP request body into ``request.params`` according to

@route(type=...), call the controller endpoint, serialize its return

value into an HTTP Response object.

ir.http._post_dispatch/Dispatcher.post_dispatch

Post process the response returned by the controller endpoint. Used to

inject various headers such as Content-Security-Policy.

ir.http._handle_error

Not present in the call-graph, is called for un-managed exceptions (SE

or RO) that occured inside of ``Request._retrying``. It returns a http

response that wraps the error that occured.

route_wrapper, closure of the http.route decorator

Sanitize the request parameters, call the route endpoint and

optionally coerce the endpoint result.

endpoint

The @route(...) decorated controller method.

"""

import odoo.init # import first for core setup

import base64

import collections.abc

import contextlib

import functools

import glob

import hashlib

import hmac

import importlib.metadata

import inspect

import json

import logging

import mimetypes

import os

import re

import threading

import time

import traceback

import warnings

from abc import ABC, abstractmethod

from datetime import datetime, timedelta

from hashlib import sha512

from http import HTTPStatus

from io import BytesIO

from os.path import join as opj

from pathlib import Path

from urllib.parse import urlparse

from zlib import adler32

import babel.core

try:

import geoip2.database

import geoip2.models

import geoip2.errors

except ImportError:

geoip2 = None

try:

import maxminddb

except ImportError:

maxminddb = None

import psycopg2

import werkzeug.datastructures

import werkzeug.exceptions

import werkzeug.local

import werkzeug.routing

import werkzeug.security

import werkzeug.wrappers

import werkzeug.wsgi

from werkzeug.urls import URL, url_parse, url_encode, url_quote

from werkzeug.exceptions import (

default_exceptions as werkzeug_default_exceptions,

HTTPException, NotFound, UnsupportedMediaType, UnprocessableEntity,

InternalServerError

)

try:

from werkzeug.middleware.proxy_fix import ProxyFix as ProxyFix_

ProxyFix = functools.partial(ProxyFix_, x_for=1, x_proto=1, x_host=1)

except ImportError:

from werkzeug.contrib.fixers import ProxyFix

try:

from werkzeug.utils import send_file as _send_file

except ImportError:

from .tools._vendor.send_file import send_file as _send_file

import odoo.addons

from .exceptions import UserError, AccessError, AccessDenied

from .modules import module as module_manager

from .modules.registry import Registry

from .service import security, model as service_model

from .service.server import thread_local

from .tools import (config, consteq, file_path, get_lang, json_default,

parse_version, profiler, unique, exception_to_unicode)

from .tools.facade import Proxy, ProxyAttr, ProxyFunc

from .tools.func import filter_kwargs

from .tools.misc import submap, real_time

from .tools._vendor import sessions

from .tools._vendor.useragents import UserAgent

_logger = logging.getLogger(__name__)

# =========================================================

# Const

# =========================================================

# The validity duration of a preflight response, one day.

CORS_MAX_AGE = 60 * 60 * 24

# The HTTP methods that do not require a CSRF validation.

SAFE_HTTP_METHODS = ('GET', 'HEAD', 'OPTIONS', 'TRACE')

# The default csrf token lifetime, a salt against BREACH, one year

CSRF_TOKEN_SALT = 60 * 60 * 24 * 365

# The default lang to use when the browser doesn't specify it

DEFAULT_LANG = 'en_US'

# The dictionary to initialise a new session with.

def get_default_session():

return {

'context': {}, # 'lang': request.default_lang() # must be set at runtime

'create_time': time.time(),

'db': None,

'debug': '',

'login': None,

'uid': None,

'session_token': None,

'_trace': [],

'create_time': time.time(),

}

DEFAULT_MAX_CONTENT_LENGTH = 128 * 1024 * 1024 # 128MiB

# Two empty objects used when the geolocalization failed. They have the

# sames attributes as real countries/cities except that accessing them

# evaluates to None.

if geoip2:

GEOIP_EMPTY_COUNTRY = geoip2.models.Country({})

GEOIP_EMPTY_CITY = geoip2.models.City({})

MISSING_CSRF_WARNING = """\

No CSRF validation token provided for path %r

Odoo URLs are CSRF-protected by default (when accessed with unsafe

HTTP methods). See

https://www.odoo.com/documentation/master/developer/reference/addons/http.html#csrf

for more details.

* if this endpoint is accessed through Odoo via py-QWeb form, embed a CSRF

token in the form, Tokens are available via `request.csrf_token()`

can be provided through a hidden input and must be POST-ed named

`csrf_token` e.g. in your form add:

<input type="hidden" name="csrf_token" t-att-value="request.csrf_token()"/>

* if the form is generated or posted in javascript, the token value is

available as `csrf_token` on `web.core` and as the `csrf_token`

value in the default js-qweb execution context

* if the form is accessed by an external third party (e.g. REST API

endpoint, payment gateway callback) you will need to disable CSRF

protection (and implement your own protection if necessary) by

passing the `csrf=False` parameter to the `route` decorator.

"""

NOT_FOUND_NODB = """\

<!DOCTYPE html>

<title>404 Not Found</title>

<h1>Not Found</h1>

<p>No database is selected and the requested URL was not found in the server-wide controllers.</p>

<p>Please verify the hostname, <a href=/web/login>login</a> and try again.</p>

<!-- Alternatively, use the X-Odoo-Database header. -->

"""

# The @route arguments to propagate from the decorated method to the

# routing rule.

ROUTING_KEYS = {

'defaults', 'subdomain', 'build_only', 'strict_slashes', 'redirect_to',

'alias', 'host', 'methods',

}

if parse_version(importlib.metadata.version('werkzeug')) >= parse_version('2.0.2'):

# Werkzeug 2.0.2 adds the websocket option. If a websocket request

# (ws/wss) is trying to access an HTTP route, a WebsocketMismatch

# exception is raised. On the other hand, Werkzeug 0.16 does not

# support the websocket routing key. In order to bypass this issue,

# let's add the websocket key only when appropriate.

ROUTING_KEYS.add('websocket')

# The default duration of a user session cookie. Inactive sessions are reaped

# server-side as well with a threshold that can be set via an optional

# config parameter `sessions.max_inactivity_seconds` (default: SESSION_LIFETIME)

SESSION_LIFETIME = 60 * 60 * 24 * 7

# The default duration (3h) before a session is rotated, changing the

# session id (also on the cookie) but keeping the same content.

SESSION_ROTATION_INTERVAL = 60 * 60 * 3

# After a session is rotated, the session should be kept for a couple of

# seconds to account for network delay between multiple requests which are

# made at the same time and all use the same old cookie.

SESSION_DELETION_TIMER = 120

# The amount of bytes of the session that will remain static and can be used

# for calculating the csrf token and be stored inside the database.

STORED_SESSION_BYTES = 42

# The cache duration for static content from the filesystem, one week.

STATIC_CACHE = 60 * 60 * 24 * 7

# The cache duration for content where the url uniquely identifies the

# content (usually using a hash), one year.

STATIC_CACHE_LONG = 60 * 60 * 24 * 365

# =========================================================

# Helpers

# =========================================================

class RegistryError(RuntimeError):

pass

class SessionExpiredException(Exception):

http_status = HTTPStatus.FORBIDDEN

def content_disposition(filename, disposition_type='attachment'):

"""

Craft a ``Content-Disposition`` header, see :rfc:`6266`.

:param filename: The name of the file, should that file be saved on

disk by the browser.

:param disposition_type: Tell the browser what to do with the file,

either ``"attachment"`` to save the file on disk,

either ``"inline"`` to display the file.

"""

if disposition_type not in ('attachment', 'inline'):

e = f"Invalid disposition_type: {disposition_type!r}"

raise ValueError(e)

return "{}; filename*=UTF-8''{}".format(

disposition_type,

url_quote(filename, safe='', unsafe='()<>@,;:"/[]?={}\\*\'%') # RFC6266

)

def db_list(force=False, host=None):

"""

Get the list of available databases.

:param bool force: See :func:`~odoo.service.db.list_dbs`:

:param host: The Host used to replace %h and %d in the dbfilters

regexp. Taken from the current request when omitted.

:returns: the list of available databases

:rtype: List[str]

"""

try:

dbs = odoo.service.db.list_dbs(force)

except psycopg2.OperationalError:

return []

return db_filter(dbs, host)

def db_filter(dbs, host=None):

"""

Return the subset of ``dbs`` that match the dbfilter or the dbname

server configuration. In case neither are configured, return ``dbs``

as-is.

:param Iterable[str] dbs: The list of database names to filter.

:param host: The Host used to replace %h and %d in the dbfilters

regexp. Taken from the current request when omitted.

:returns: The original list filtered.

:rtype: List[str]

"""

if config['dbfilter']:

# host

# -----------

# www.example.com:80

# -------

# domain

if host is None:

host = request.httprequest.environ.get('HTTP_HOST', '')

host = host.partition(':')[0]

if host.startswith('www.'):

host = host[4:]

domain = host.partition('.')[0]

dbfilter_re = re.compile(

config["dbfilter"].replace("%h", re.escape(host))

.replace("%d", re.escape(domain)))

return [db for db in dbs if dbfilter_re.match(db)]

if config['db_name']:

# In case --db-filter is not provided and --database is passed, Odoo will

# use the value of --database as a comma separated list of exposed databases.

return sorted(set(config['db_name']).intersection(dbs))

return list(dbs)

def dispatch_rpc(service_name, method, params):

"""

Perform a RPC call.

:param str service_name: either "common", "db" or "object".

:param str method: the method name of the given service to execute

:param Mapping params: the keyword arguments for method call

:return: the return value of the called method

:rtype: Any

"""

rpc_dispatchers = {

'common': odoo.service.common.dispatch,

'db': odoo.service.db.dispatch,

'object': odoo.service.model.dispatch,

}

with borrow_request():

threading.current_thread().uid = None

threading.current_thread().dbname = None

dispatch = rpc_dispatchers[service_name]

return dispatch(method, params)

def get_session_max_inactivity(env):

if not env or env.cr._closed:

return SESSION_LIFETIME

ICP = env['ir.config_parameter'].sudo()

try:

return int(ICP.get_param('sessions.max_inactivity_seconds', SESSION_LIFETIME))

except ValueError:

_logger.warning("Invalid value for 'sessions.max_inactivity_seconds', using default value.")

return SESSION_LIFETIME

def is_cors_preflight(request, endpoint):

return request.httprequest.method == 'OPTIONS' and endpoint.routing.get('cors', False)

def serialize_exception(exception, *, message=None, arguments=None):

name = type(exception).__name__

module = type(exception).__module__

return {

'name': f'{module}.{name}' if module else name,

'message': exception_to_unicode(exception) if message is None else message,

'arguments': exception.args if arguments is None else arguments,

'context': getattr(exception, 'context', {}),

'debug': ''.join(traceback.format_exception(exception)),

}

# =========================================================

# File Streaming

# =========================================================

class Stream:

"""

Send the content of a file, an attachment or a binary field via HTTP

This utility is safe, cache-aware and uses the best available

streaming strategy. Works best with the --x-sendfile cli option.

Create a Stream via one of the constructors: :meth:`~from_path`:, or

:meth:`~from_binary_field`:, generate the corresponding HTTP response

object via :meth:`~get_response`:.

Instantiating a Stream object manually without using one of the

dedicated constructors is discouraged.

"""

type: str = '' # 'data' or 'path' or 'url'

data = None

path = None

url = None

mimetype = None

as_attachment = False

download_name = None

conditional = True

etag = True

last_modified = None

max_age = None

immutable = False

size = None

public = False

def __init__(self, **kwargs):

# Remove class methods from the instances

self.from_path = self.from_attachment = self.from_binary_field = None

self.__dict__.update(kwargs)

@classmethod

def from_path(cls, path, filter_ext=('',), public=False):

"""

Create a :class:`~Stream`: from an addon resource.

:param path: See :func:`~odoo.tools.file_path`

:param filter_ext: See :func:`~odoo.tools.file_path`

:param bool public: Advertise the resource as being cachable by

intermediate proxies, otherwise only let the browser caches

it.

"""

path = file_path(path, filter_ext)

check = adler32(path.encode())

stat = os.stat(path)

return cls(

type='path',

path=path,

mimetype=mimetypes.guess_type(path)[0],

download_name=os.path.basename(path),

etag=f'{int(stat.st_mtime)}-{stat.st_size}-{check}',

last_modified=stat.st_mtime,

size=stat.st_size,

public=public,

)

@classmethod

def from_binary_field(cls, record, field_name):

""" Create a :class:`~Stream`: from a binary field. """

data_b64 = record[field_name]

data = base64.b64decode(data_b64) if data_b64 else b''

return cls(

type='data',

data=data,

etag=request.env['ir.attachment']._compute_checksum(data),

last_modified=record.write_date if record._log_access else None,

size=len(data),

public=record.env.user._is_public() # good enough

)

def read(self):

""" Get the stream content as bytes. """

if self.type == 'url':

raise ValueError("Cannot read an URL")

if self.type == 'data':

return self.data

with open(self.path, 'rb') as file:

return file.read()

def get_response(

self,

as_attachment=None,

immutable=None,

content_security_policy="default-src 'none'",

**send_file_kwargs

):

"""

Create the corresponding :class:`~Response` for the current stream.

:param bool|None as_attachment: Indicate to the browser that it

should offer to save the file instead of displaying it.

:param bool|None immutable: Add the ``immutable`` directive to

the ``Cache-Control`` response header, allowing intermediary

proxies to aggressively cache the response. This option also

set the ``max-age`` directive to 1 year.

:param str|None content_security_policy: Optional value for the

``Content-Security-Policy`` (CSP) header. This header is

used by browsers to allow/restrict the downloaded resource

to itself perform new http requests. By default CSP is set

to ``"default-scr 'none'"`` which restrict all requests.

:param send_file_kwargs: Other keyword arguments to send to

:func:`odoo.tools._vendor.send_file.send_file` instead of

the stream sensitive values. Discouraged.

"""

assert self.type in ('url', 'data', 'path'), "Invalid type: {self.type!r}, should be 'url', 'data' or 'path'."

assert getattr(self, self.type) is not None, "There is nothing to stream, missing {self.type!r} attribute."

if self.type == 'url':

if self.max_age is not None:

res = request.redirect(self.url, code=302, local=False)

res.headers['Cache-Control'] = f'max-age={self.max_age}'

return res

return request.redirect(self.url, code=301, local=False)

if as_attachment is None:

as_attachment = self.as_attachment

if immutable is None:

immutable = self.immutable

send_file_kwargs = {

'mimetype': self.mimetype,

'as_attachment': as_attachment,

'download_name': self.download_name,

'conditional': self.conditional,

'etag': self.etag,

'last_modified': self.last_modified,

'max_age': STATIC_CACHE_LONG if immutable else self.max_age,

'environ': request.httprequest.environ,

'response_class': Response,

**send_file_kwargs,

}

if self.type == 'data':

res = _send_file(BytesIO(self.data), **send_file_kwargs)

else: # self.type == 'path'

send_file_kwargs['use_x_sendfile'] = False

if config['x_sendfile']:

with contextlib.suppress(ValueError): # outside of the filestore

fspath = Path(self.path).relative_to(opj(config['data_dir'], 'filestore'))

x_accel_redirect = f'/web/filestore/{fspath}'

send_file_kwargs['use_x_sendfile'] = True

res = _send_file(self.path, **send_file_kwargs)

if 'X-Sendfile' in res.headers:

res.headers['X-Accel-Redirect'] = x_accel_redirect

# In case of X-Sendfile/X-Accel-Redirect, the body is empty,

# yet werkzeug gives the length of the file. This makes

# NGINX wait for content that'll never arrive.

res.headers['Content-Length'] = '0'

res.headers['X-Content-Type-Options'] = 'nosniff'

if content_security_policy: # see also Application.set_csp()

res.headers['Content-Security-Policy'] = content_security_policy

if self.public:

if (res.cache_control.max_age or 0) > 0:

res.cache_control.public = True

else:

res.cache_control.pop('public', '')

res.cache_control.private = True

if immutable:

res.cache_control['immutable'] = None # None sets the directive

return res

# =========================================================

# Controller and routes

# =========================================================

class Controller:

"""

Class mixin that provide module controllers the ability to serve

content over http and to be extended in child modules.

Each class :ref:`inheriting <python:tut-inheritance>` from

:class:`~odoo.http.Controller` can use the :func:`~odoo.http.route`:

decorator to route matching incoming web requests to decorated

methods.

Like models, controllers can be extended by other modules. The

extension mechanism is different because controllers can work in a

database-free environment and therefore cannot use

:class:~odoo.api.Registry:.

To *override* a controller, :ref:`inherit <python:tut-inheritance>`

from its class, override relevant methods and re-expose them with

:func:`~odoo.http.route`:. Please note that the decorators of all

methods are combined, if the overriding method’s decorator has no

argument all previous ones will be kept, any provided argument will

override previously defined ones.

.. code-block:

class GreetingController(odoo.http.Controller):

@route('/greet', type='http', auth='public')

def greeting(self):

return 'Hello'

class UserGreetingController(GreetingController):

@route(auth='user') # override auth, keep path and type

def greeting(self):

return super().handler()

"""

children_classes = collections.defaultdict(list) # indexed by module

@classmethod

def __init_subclass__(cls):

super().__init_subclass__()

if Controller in cls.__bases__:

path = cls.__module__.split('.')

module = path[2] if path[:2] == ['odoo', 'addons'] else ''

Controller.children_classes[module].append(cls)

@property

def env(self):

return request.env if request else None

def route(route=None, **routing):

"""

Decorate a controller method in order to route incoming requests

matching the given URL and options to the decorated method.

.. warning::

It is mandatory to re-decorate any method that is overridden in

controller extensions but the arguments can be omitted. See

:class:`~odoo.http.Controller` for more details.

:param Union[str, Iterable[str]] route: The paths that the decorated

method is serving. Incoming HTTP request paths matching this

route will be routed to this decorated method. See `werkzeug

routing documentation <http://werkzeug.pocoo.org/docs/routing/>`_

for the format of route expressions.

:param str type: The type of request, either ``'jsonrpc'`` or

``'http'``. It describes where to find the request parameters

and how to serialize the response.

:param str auth: The authentication method, one of the following:

* ``'user'``: The user must be authenticated and the current

request will be executed using the rights of the user.

* ``'bearer'``: The user is authenticated using an "Authorization"

request header, using the Bearer scheme with an API token.

The request will be executed with the permissions of the

corresponding user. If the header is missing, the request

must belong to an authentication session, as for the "user"

authentication method.

* ``'public'``: The user may or may not be authenticated. If he

isn't, the current request will be executed using the shared

Public user.

* ``'none'``: The method is always active, even if there is no

database. Mainly used by the framework and authentication

modules. The request code will not have any facilities to

access the current user.

:param Iterable[str] methods: A list of http methods (verbs) this

route applies to. If not specified, all methods are allowed.

:param str cors: The Access-Control-Allow-Origin cors directive value.

:param bool csrf: Whether CSRF protection should be enabled for the

route. Enabled by default for ``'http'``-type requests, disabled

by default for ``'jsonrpc'``-type requests.

:param Union[bool, Callable[[registry, request], bool]] readonly:

Whether this endpoint should open a cursor on a read-only

replica instead of (by default) the primary read/write database.

:param Callable[[Exception], Response] handle_params_access_error:

Implement a custom behavior if an error occurred when retrieving

the record from the URL parameters (access error or missing error).

:param str captcha: The action name of the captcha. When set the

request will be validated against a captcha implementation. Upon

failing these requests will return a UserError.

:param bool save_session: Whether it should set a session_id cookie

on the http response and save dirty session on disk. ``False``

by default for ``auth='bearer'``. ``True`` by default otherwise.

"""

def decorator(endpoint):

fname = f"<function {endpoint.__module__}.{endpoint.__name__}>"

# Sanitize the routing

if routing.get('type') == 'json':

warnings.warn(

"Since 19.0, @route(type='json') is a deprecated alias to @route(type='jsonrpc')",

DeprecationWarning,

stacklevel=3,

)

routing['type'] = 'jsonrpc'

assert routing.get('type', 'http') in _dispatchers.keys(), \

f"@route(type={routing['type']!r}) is not one of {_dispatchers.keys()}"

if route:

routing['routes'] = [route] if isinstance(route, str) else route

wrong = routing.pop('method', None)

if wrong is not None:

_logger.warning("%s defined with invalid routing parameter 'method', assuming 'methods'", fname)

routing['methods'] = wrong

if routing.get('auth') == 'bearer':

routing.setdefault('save_session', False) # stateless

@functools.wraps(endpoint)

def route_wrapper(self, *args, **params):

params_ok = filter_kwargs(endpoint, params)

params_ko = set(params) - set(params_ok)

if params_ko:

_logger.warning("%s called ignoring args %s", fname, params_ko)

result = endpoint(self, *args, **params_ok)

if routing['type'] == 'http': # _generate_routing_rules() ensures type is set

return Response.load(result)

return result

route_wrapper.original_routing = routing

route_wrapper.original_endpoint = endpoint

return route_wrapper

return decorator

def _generate_routing_rules(modules, nodb_only, converters=None):

"""

Two-fold algorithm used to (1) determine which method in the

controller inheritance tree should bind to what URL with respect to

the list of installed modules and (2) merge the various @route

arguments of said method with the @route arguments of the method it

overrides.

"""

def is_valid(cls):

""" Determine if the class is defined in an addon. """

path = cls.__module__.split('.')

return path[:2] == ['odoo', 'addons'] and path[2] in modules

def get_leaf_classes(cls):

"""

Find the classes that have no child and that have ``cls`` as

ancestor.

"""

result = []

for subcls in cls.__subclasses__():

if is_valid(subcls):

result.extend(get_leaf_classes(subcls))

if not result and is_valid(cls):

result.append(cls)

return result

def build_controllers():

"""

Create dummy controllers that inherit only from the controllers

defined at the given ``modules`` (often system wide modules or

installed modules). Modules in this context are Odoo addons.

"""

# Controllers defined outside of odoo addons are outside of the

# controller inheritance/extension mechanism.

yield from (ctrl() for ctrl in Controller.children_classes.get('', []))

# Controllers defined inside of odoo addons can be extended in

# other installed addons. Rebuild the class inheritance here.

highest_controllers = []

for module in modules:

highest_controllers.extend(Controller.children_classes.get(module, []))

for top_ctrl in highest_controllers:

leaf_controllers = list(unique(get_leaf_classes(top_ctrl)))

name = top_ctrl.__name__

if leaf_controllers != [top_ctrl]:

name += ' (extended by %s)' % ', '.join(

bot_ctrl.__name__

for bot_ctrl in leaf_controllers

if bot_ctrl is not top_ctrl

)

Ctrl = type(name, tuple(reversed(leaf_controllers)), {})

yield Ctrl()

for ctrl in build_controllers():

for method_name, method in inspect.getmembers(ctrl, inspect.ismethod):

# Skip this method if it is not @route decorated anywhere in

# the hierarchy

def is_method_a_route(cls):

return getattr(getattr(cls, method_name, None), 'original_routing', None) is not None

if not any(map(is_method_a_route, type(ctrl).mro())):

continue

merged_routing = {

# 'type': 'http', # set below

'auth': 'user',

'methods': None,

'routes': [],

}

for cls in unique(reversed(type(ctrl).mro()[:-2])): # ancestors first

if method_name not in cls.__dict__:

continue

submethod = getattr(cls, method_name)

if not hasattr(submethod, 'original_routing'):

_logger.warning("The endpoint %s is not decorated by @route(), decorating it myself.", f'{cls.__module__}.{cls.__name__}.{method_name}')

submethod = route()(submethod)

_check_and_complete_route_definition(cls, submethod, merged_routing)

merged_routing.update(submethod.original_routing)

if not merged_routing['routes']:

_logger.warning("%s is a controller endpoint without any route, skipping.", f'{cls.__module__}.{cls.__name__}.{method_name}')

continue

if nodb_only and merged_routing['auth'] != "none":

continue

for url in merged_routing['routes']:

# duplicates the function (partial) with a copy of the

# original __dict__ (update_wrapper) to keep a reference

# to `original_routing` and `original_endpoint`, assign

# the merged routing ONLY on the duplicated function to

# ensure method's immutability.

endpoint = functools.partial(method)

functools.update_wrapper(endpoint, method)

endpoint.routing = merged_routing

yield (url, endpoint)

def _check_and_complete_route_definition(controller_cls, submethod, merged_routing):

"""Verify and complete the route definition.

* Ensure 'type' is defined on each method's own routing.

* Ensure overrides don't change the routing type or the read/write mode

:param submethod: route method

:param dict merged_routing: accumulated routing values

"""

default_type = submethod.original_routing.get('type', 'http')

routing_type = merged_routing.setdefault('type', default_type)

if submethod.original_routing.get('type') not in (None, routing_type):

_logger.warning(

"The endpoint %s changes the route type, using the original type: %r.",

f'{controller_cls.__module__}.{controller_cls.__name__}.{submethod.__name__}',

routing_type)

submethod.original_routing['type'] = routing_type

default_auth = submethod.original_routing.get('auth', merged_routing['auth'])

default_mode = submethod.original_routing.get('readonly', default_auth == 'none')

parent_readonly = merged_routing.setdefault('readonly', default_mode)

child_readonly = submethod.original_routing.get('readonly')

if child_readonly not in (None, parent_readonly) and not callable(child_readonly):

_logger.warning(

"The endpoint %s made the route %s altough its parent was defined as %s. Setting the route read/write.",

f'{controller_cls.__module__}.{controller_cls.__name__}.{submethod.__name__}',

'readonly' if child_readonly else 'read/write',

'readonly' if parent_readonly else 'read/write',

)

submethod.original_routing['readonly'] = False

# =========================================================

# Session

# =========================================================

_base64_urlsafe_re = re.compile(r'^[A-Za-z0-9_-]{84}$')

_session_identifier_re = re.compile(r'^[A-Za-z0-9_-]{%s}$' % STORED_SESSION_BYTES)

class FilesystemSessionStore(sessions.FilesystemSessionStore):

""" Place where to load and save session objects. """

def get_session_filename(self, sid):

# scatter sessions across 4096 (64^2) directories

if not self.is_valid_key(sid):

raise ValueError(f'Invalid session id {sid!r}')

sha_dir = sid[:2]

dirname = os.path.join(self.path, sha_dir)

session_path = os.path.join(dirname, sid)

return session_path

def save(self, session):

session_path = self.get_session_filename(session.sid)

dirname = os.path.dirname(session_path)

if not os.path.isdir(dirname):

with contextlib.suppress(OSError):

os.mkdir(dirname, 0o0755)

super().save(session)

def delete_old_sessions(self, session):

if 'gc_previous_sessions' in session:

if session['create_time'] + SESSION_DELETION_TIMER < time.time():

self.delete_from_identifiers([session.sid[:STORED_SESSION_BYTES]])

del session['gc_previous_sessions']

self.save(session)

def get(self, sid):

# retro compatibility

old_path = super().get_session_filename(sid)

session_path = self.get_session_filename(sid)

if os.path.isfile(old_path) and not os.path.isfile(session_path):

dirname = os.path.dirname(session_path)

if not os.path.isdir(dirname):

with contextlib.suppress(OSError):

os.mkdir(dirname, 0o0755)

with contextlib.suppress(OSError):

os.rename(old_path, session_path)

session = super().get(sid)

return session

def rotate(self, session, env, soft=False):

# With a soft rotation, things like the CSRF token will still work. It's used for rotating

# the session in a way that half the bytes remain to identify the user and the other half

# to authenticate the user. Meanwhile with a hard rotation the entire session id is changed,

# which is useful in cases such as logging the user out.

if soft:

# Multiple network requests can occur at the same time, all using the old session.

# We don't want to create a new session for each request, it's better to reference the one already made.

static = session.sid[:STORED_SESSION_BYTES]

recent_session = self.get(session.sid)

if 'next_sid' in recent_session:

# A new session has already been saved on disk by a concurrent request,

# the _save_session is going to simply use session.sid to set a new cookie.

session.sid = recent_session['next_sid']

return

next_sid = static + self.generate_key()[STORED_SESSION_BYTES:]