return {

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

'db': None,

'debug': '',

'login': None,

'uid': None,

'session_token': None,

'_trace': [],

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.

"""

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 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.

exposed_dbs = {db.strip() for db in config['db_name'].split(',')}

return sorted(exposed_dbs.intersection(dbs))

"""

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]

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.")

name = type(exception).__name__

module = type(exception).__module__

return {

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

'debug': traceback.format_exc(),

'message': exception_to_unicode(exception),

'arguments': exception.args,

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

"""

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

"""

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 ''

"""

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 ``'json'`` 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 ``'json'``-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).

"""

def decorator(endpoint):

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

# Sanitize the routing

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

if route:

routing['routes'] = route if isinstance(route, list) 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

@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

"""

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

"""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',

)

""" 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 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)

return super().get(sid)

def rotate(self, session, env):

self.delete(session)

session.sid = self.generate_key()

if session.uid and env:

session.session_token = security.compute_session_token(session, env)

session.should_rotate = False

self.save(session)

def vacuum(self, max_lifetime=SESSION_LIFETIME):

threshold = time.time() - max_lifetime

for fname in glob.iglob(os.path.join(root.session_store.path, '*', '*')):

path = os.path.join(root.session_store.path, fname)

with contextlib.suppress(OSError):

if os.path.getmtime(path) < threshold:

os.unlink(path)

def generate_key(self, salt=None):

# The generated key is case sensitive (base64) and the length is 84 chars.

# In the worst-case scenario, i.e. in an insensitive filesystem (NTFS for example)

# taking into account the proportion of characters in the pool and a length

# of 42 (stored part in the database), the entropy for the base64 generated key

# is 217.875 bits which is better than the 160 bits entropy of a hexadecimal key

# with a length of 40 (method ``generate_key`` of ``SessionStore``).

# The risk of collision is negligible in practice.

# Formulas:

# - L: length of generated word

# - p_char: probability of obtaining the character in the pool

# - n: size of the pool

# - k: number of generated word

# Entropy = - L * sum(p_char * log2(p_char))

# Collision ~= (1 - exp((-k * (k - 1)) / (2 * (n**L))))

key = str(time.time()).encode() + os.urandom(64)

hash_key = sha512(key).digest()[:-1] # prevent base64 padding

return base64.urlsafe_b64encode(hash_key).decode('utf-8')

def is_valid_key(self, key):

return _base64_urlsafe_re.match(key) is not None

def delete_from_identifiers(self, identifiers):

files_to_unlink = []

for identifier in identifiers:

# Avoid to remove a session if less than 42 chars.

# This prevent malicious user to delete sessions from a different

# database by specifying a ``res.device.log`` with only 2 characters.

if len(identifier) < 42:

continue

normalized_path = os.path.normpath(os.path.join(self.path, identifier[:2], identifier + '*'))

if normalized_path.startswith(self.path):

files_to_unlink.extend(glob.glob(normalized_path))

for fn in files_to_unlink:

with contextlib.suppress(OSError):

""" Structure containing data persisted across requests. """

__slots__ = ('can_save', '_Session__data', 'is_dirty', 'is_new',

'should_rotate', 'sid')

def __init__(self, data, sid, new=False):

self.can_save = True

self.__data = {}

self.update(data)

self.is_dirty = False

self.is_new = new

self.should_rotate = False

self.sid = sid

#

# MutableMapping implementation with DocDict-like extension

#

def __getitem__(self, item):

return self.__data[item]

def __setitem__(self, item, value):

value = json.loads(json.dumps(value))

if item not in self.__data or self.__data[item] != value:

self.is_dirty = True

self.__data[item] = value

def __delitem__(self, item):

del self.__data[item]

self.is_dirty = True

def __len__(self):

return len(self.__data)