Allura/allura/config/middleware.py - allura - Git at Google

 #       Licensed to the Apache Software Foundation (ASF) under one
 #       or more contributor license agreements.  See the NOTICE file
 #       distributed with this work for additional information
 #       regarding copyright ownership.  The ASF licenses this file
 #       to you under the Apache License, Version 2.0 (the
 #       "License"); you may not use this file except in compliance
 #       with the License.  You may obtain a copy of the License at
 #
 #         http://www.apache.org/licenses/LICENSE-2.0
 #
 #       Unless required by applicable law or agreed to in writing,
 #       software distributed under the License is distributed on an
 #       "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 #       KIND, either express or implied.  See the License for the
 #       specific language governing permissions and limitations
 #       under the License.

 """core app and WSGI middleware initialization.  New TurboGears2 apps name it application.py"""

 import ast
 import importlib
 import mimetypes
 import pickle
 import re
 import warnings

 import six
 import tg
 import pkg_resources
 from tg import config
 from paste.deploy.converters import asbool, aslist, asint
 from tg.support.registry import RegistryManager
 from beaker.middleware import SessionMiddleware
 from beaker.util import PickleSerializer
 from paste.exceptions.errormiddleware import ErrorMiddleware
 from werkzeug.debug import DebuggedApplication

 import activitystream
 import ew
 import formencode
 import ming
 from ming.odm.middleware import MingMiddleware
 from beaker_session_jwt import JWTCookieSession

 # Must apply patches before other Allura imports to ensure all the patches are effective.
 # This file gets imported from paste/deploy/loadwsgi.py pretty early in the app execution
 # ruff: noqa: E402
 from allura.lib import patches
 patches.apply()
 try:
     import newrelic
 except ImportError:
     pass
 else:
     patches.newrelic()

 from allura.config.app_cfg import AlluraJinjaRenderer
 from allura.config.app_cfg import ForgeConfig
 from allura.lib.custom_middleware import AlluraTimerMiddleware
 from allura.lib.custom_middleware import SSLMiddleware
 from allura.lib.custom_middleware import StaticFilesMiddleware
 from allura.lib.custom_middleware import CSRFMiddleware
 from allura.lib.custom_middleware import CORSMiddleware
 from allura.lib.custom_middleware import LoginRedirectMiddleware
 from allura.lib.custom_middleware import RememberLoginMiddleware
 from allura.lib.custom_middleware import SetRequestHostFromConfig
 from allura.lib.custom_middleware import MingTaskSessionSetupMiddleware
 from allura.lib.custom_middleware import ContentSecurityPolicyMiddleware
 from allura.lib.custom_middleware import SetHeadersMiddleware
 from allura.lib.custom_middleware import StatusCodeRedirect
 from allura.lib import helpers as h
 from allura.lib.utils import configure_ming

 __all__ = ['make_app']


 # this is webapp entry point from setup.py
 def make_app(global_conf: dict, **app_conf):
     override_root_module_name = app_conf.get('override_root', None)
     if override_root_module_name:
         # get an actual instance of it, like BasetestProjectRootController or TaskController
         className = override_root_module_name.title().replace('_', '') + 'Controller'
         module = importlib.import_module(f'allura.controllers.{override_root_module_name}')
         rootClass = getattr(module, className)
         root = rootClass()
     else:
         root = None  # will default to RootController in root.py
     return _make_core_app(root, global_conf, **app_conf)


 class BeakerPickleSerializerWithLatin1(PickleSerializer):
     def loads(self, data_string):
         # need latin1 to decode py2 timestamps in py  https://docs.python.org/3/library/pickle.html#pickle.Unpickler
         return pickle.loads(data_string, encoding='latin1')  # noqa: S301


 def _make_core_app(root, global_conf: dict, **app_conf):
     """
     Set allura up with the settings found in the PasteDeploy configuration
     file used.

     :param root: The controller module containing the TG root
     :param global_conf: The global settings for allura (those
         defined under the ``[DEFAULT]`` section).
     :type global_conf: dict
     :return: The allura application with all the relevant middleware
         loaded.

     This is the PasteDeploy factory for the allura application.

     ``app_conf`` contains all the application-specific settings (those defined
     under ``[app:main]``.
     """
     # Run all the initialization code here
     mimetypes.init([pkg_resources.resource_filename('allura', 'etc/mime.types')] + mimetypes.knownfiles)

     # Configure MongoDB
     configure_ming(app_conf)

     # Configure ActivityStream
     if asbool(app_conf.get('activitystream.recording.enabled', False)):
         activitystream.configure(**h.convert_bools(app_conf, prefix='activitystream.'))

     # Configure EW variable provider
     ew.render.TemplateEngine.register_variable_provider(get_tg_vars)

     # Set FormEncode language to english, as we don't support any other locales
     formencode.api.set_stdtranslation(domain='FormEncode', languages=['en'])

     # Create base app
     base_config = ForgeConfig(root)
     app = base_config.make_wsgi_app(global_conf, app_conf, wrap_app=None)

     for mw_ep in h.iter_entry_points('allura.middleware'):
         Middleware = mw_ep.load()
         if getattr(Middleware, 'when', 'inner') == 'inner':
             app = Middleware(app, config)
     # CSP headers
     app = ContentSecurityPolicyMiddleware(app, config)
     # broswer permissions policy
     app = SetHeadersMiddleware(app, config)
     # Required for sessions
     with warnings.catch_warnings():
         # the session_class= arg triggers this warning but is the only way it works, so suppress warning
         warnings.filterwarnings('ignore',
                                 re.escape('Session options should start with session. instead of session_.'),
                                 DeprecationWarning)
         app = SessionMiddleware(app, config,
                                 original_format_data_serializer=BeakerPickleSerializerWithLatin1(),
                                 session_class=JWTCookieSession)
     # Handle "Remember me" functionality
     app = RememberLoginMiddleware(app, config)
     # Redirect 401 to the login page
     app = LoginRedirectMiddleware(app)
     # Add instrumentation
     app = AlluraTimerMiddleware(app, app_conf)
     # Clear cookies when the CSRF field isn't posted
     if not app_conf.get('disable_csrf_protection'):
         app = CSRFMiddleware(app, '_session_id')
     if asbool(config.get('cors.enabled', False)):
         # Handle CORS requests
         allowed_methods = aslist(config.get('cors.methods'))
         allowed_headers = aslist(config.get('cors.headers'))
         cache_duration = asint(config.get('cors.cache_duration', 0))
         app = CORSMiddleware(app, allowed_methods, allowed_headers, cache_duration)
     # Setup the allura SOPs
     app = allura_globals_middleware(app)
     # Ensure http and https used per config
     if config.get('override_root') != 'task':
         app = SSLMiddleware(app, app_conf.get('no_redirect.pattern'),
                             app_conf.get('force_ssl.pattern'))
     # Setup resource manager, widget context SOP
     app = ew.WidgetMiddleware(
         app,
         compress=True,
         use_cache=not asbool(global_conf['debug']),
         script_name=app_conf.get('ew.script_name', '/_ew_resources/'),
         url_base=app_conf.get('ew.url_base', '/_ew_resources/'),
         extra_headers=ast.literal_eval(app_conf.get('ew.extra_headers', '[]')),
         cache_max_age=asint(app_conf.get('ew.cache_header_seconds', 60*60*24*365)),

         # settings to pass through to jinja Environment for EW core widgets
         # these are for the easywidgets' own [easy_widgets.engines] entry point
         # (the Allura [easy_widgets.engines] entry point is named "jinja" (not jinja2) but it doesn't need
         #  any settings since it is a class that uses the same jinja env as the rest of allura)
         **{
             'jinja2.auto_reload': asbool(config['auto_reload_templates']),
             'jinja2.bytecode_cache': AlluraJinjaRenderer._setup_bytecode_cache(),
             'jinja2.cache_size': asint(config.get('jinja_cache_size', -1)),
         }
     )
     # Handle static files (by tool)
     app = StaticFilesMiddleware(app, app_conf.get('static.script_name'))
     # Handle setup and flushing of Ming ORM sessions
     app = MingTaskSessionSetupMiddleware(app)
     app = MingMiddleware(app)
     # Set up the registry for stacked object proxies (SOPs).
     app = RegistryManager(app,
                           # streaming=True causes cleanup problems when StatusCodeRedirect does an extra request
                           streaming=False,
                           preserve_exceptions=asbool(config['debug']),  # allow inspecting them when debugging errors
                           )

     # "task" wsgi would get a 2nd request to /error/document if we used this middleware
     if config.get('override_root') not in ('task', 'basetest_project_root'):
         if asbool(config['debug']):
             # Converts exceptions to HTTP errors, shows traceback in debug mode
             app = DebuggedApplication(app, evalex=True)
         else:
             app = ErrorMiddleware(app, config)

         app = SetRequestHostFromConfig(app, config)

         # Redirect some status codes to /error/document
         handle_status_codes = [403, 404, 410]
         if asbool(config['debug']):
             app = StatusCodeRedirect(app, handle_status_codes)
         else:
             app = StatusCodeRedirect(app, handle_status_codes + [500])

     for mw_ep in h.iter_entry_points('allura.middleware'):
         Middleware = mw_ep.load()
         if getattr(Middleware, 'when', 'inner') == 'outer':
             app = Middleware(app, config)

     return app


 def allura_globals_middleware(app):
     def AlluraGlobalsMiddleware(environ, start_response):
         import allura.lib.security
         import allura.lib.app_globals
         registry = environ['paste.registry']
         registry.register(allura.credentials,
                           allura.lib.security.Credentials())
         return app(environ, start_response)
     return AlluraGlobalsMiddleware


 def get_tg_vars(context):
     import tg
     from allura.lib import helpers as h
     from urllib.parse import quote, quote_plus
     context.setdefault('g', tg.app_globals)
     context.setdefault('c', tg.tmpl_context)
     context.setdefault('h', h)
     context.setdefault('request', tg.request)
     context.setdefault('response', tg.response)
     context.setdefault('url', tg.url)
     context.setdefault('tg', dict(
         config=tg.config,
         flash_obj=tg.flash,
         quote=quote,
         quote_plus=quote_plus,
         url=tg.url))
	# Licensed to the Apache Software Foundation (ASF) under one
	# or more contributor license agreements. See the NOTICE file
	# distributed with this work for additional information
	# regarding copyright ownership. The ASF licenses this file
	# to you under the Apache License, Version 2.0 (the
	# "License"); you may not use this file except in compliance
	# with the License. You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing,
	# software distributed under the License is distributed on an
	# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	# KIND, either express or implied. See the License for the
	# specific language governing permissions and limitations
	# under the License.

	"""core app and WSGI middleware initialization. New TurboGears2 apps name it application.py"""

	import ast
	import importlib
	import mimetypes
	import pickle
	import re
	import warnings

	import six
	import tg
	import pkg_resources
	from tg import config
	from paste.deploy.converters import asbool, aslist, asint
	from tg.support.registry import RegistryManager
	from beaker.middleware import SessionMiddleware
	from beaker.util import PickleSerializer
	from paste.exceptions.errormiddleware import ErrorMiddleware
	from werkzeug.debug import DebuggedApplication

	import activitystream
	import ew
	import formencode
	import ming
	from ming.odm.middleware import MingMiddleware
	from beaker_session_jwt import JWTCookieSession

	# Must apply patches before other Allura imports to ensure all the patches are effective.
	# This file gets imported from paste/deploy/loadwsgi.py pretty early in the app execution
	# ruff: noqa: E402
	from allura.lib import patches
	patches.apply()
	try:
	import newrelic
	except ImportError:
	pass
	else:
	patches.newrelic()

	from allura.config.app_cfg import AlluraJinjaRenderer
	from allura.config.app_cfg import ForgeConfig
	from allura.lib.custom_middleware import AlluraTimerMiddleware
	from allura.lib.custom_middleware import SSLMiddleware
	from allura.lib.custom_middleware import StaticFilesMiddleware
	from allura.lib.custom_middleware import CSRFMiddleware
	from allura.lib.custom_middleware import CORSMiddleware
	from allura.lib.custom_middleware import LoginRedirectMiddleware
	from allura.lib.custom_middleware import RememberLoginMiddleware
	from allura.lib.custom_middleware import SetRequestHostFromConfig
	from allura.lib.custom_middleware import MingTaskSessionSetupMiddleware
	from allura.lib.custom_middleware import ContentSecurityPolicyMiddleware
	from allura.lib.custom_middleware import SetHeadersMiddleware
	from allura.lib.custom_middleware import StatusCodeRedirect
	from allura.lib import helpers as h
	from allura.lib.utils import configure_ming

	__all__ = ['make_app']


	# this is webapp entry point from setup.py
	def make_app(global_conf: dict, **app_conf):
	override_root_module_name = app_conf.get('override_root', None)
	if override_root_module_name:
	# get an actual instance of it, like BasetestProjectRootController or TaskController
	className = override_root_module_name.title().replace('_', '') + 'Controller'
	module = importlib.import_module(f'allura.controllers.{override_root_module_name}')
	rootClass = getattr(module, className)
	root = rootClass()
	else:
	root = None # will default to RootController in root.py
	return _make_core_app(root, global_conf, **app_conf)


	class BeakerPickleSerializerWithLatin1(PickleSerializer):
	def loads(self, data_string):
	# need latin1 to decode py2 timestamps in py https://docs.python.org/3/library/pickle.html#pickle.Unpickler
	return pickle.loads(data_string, encoding='latin1') # noqa: S301


	def _make_core_app(root, global_conf: dict, **app_conf):
	"""
	Set allura up with the settings found in the PasteDeploy configuration
	file used.

	:param root: The controller module containing the TG root
	:param global_conf: The global settings for allura (those
	defined under the ``[DEFAULT]`` section).
	:type global_conf: dict
	:return: The allura application with all the relevant middleware
	loaded.

	This is the PasteDeploy factory for the allura application.

	``app_conf`` contains all the application-specific settings (those defined
	under ``[app:main]``.
	"""
	# Run all the initialization code here
	mimetypes.init([pkg_resources.resource_filename('allura', 'etc/mime.types')] + mimetypes.knownfiles)

	# Configure MongoDB
	configure_ming(app_conf)

	# Configure ActivityStream
	if asbool(app_conf.get('activitystream.recording.enabled', False)):
	activitystream.configure(**h.convert_bools(app_conf, prefix='activitystream.'))

	# Configure EW variable provider
	ew.render.TemplateEngine.register_variable_provider(get_tg_vars)

	# Set FormEncode language to english, as we don't support any other locales
	formencode.api.set_stdtranslation(domain='FormEncode', languages=['en'])

	# Create base app
	base_config = ForgeConfig(root)
	app = base_config.make_wsgi_app(global_conf, app_conf, wrap_app=None)

	for mw_ep in h.iter_entry_points('allura.middleware'):
	Middleware = mw_ep.load()
	if getattr(Middleware, 'when', 'inner') == 'inner':
	app = Middleware(app, config)
	# CSP headers
	app = ContentSecurityPolicyMiddleware(app, config)
	# broswer permissions policy
	app = SetHeadersMiddleware(app, config)
	# Required for sessions
	with warnings.catch_warnings():
	# the session_class= arg triggers this warning but is the only way it works, so suppress warning
	warnings.filterwarnings('ignore',
	re.escape('Session options should start with session. instead of session_.'),
	DeprecationWarning)
	app = SessionMiddleware(app, config,
	original_format_data_serializer=BeakerPickleSerializerWithLatin1(),
	session_class=JWTCookieSession)
	# Handle "Remember me" functionality
	app = RememberLoginMiddleware(app, config)
	# Redirect 401 to the login page
	app = LoginRedirectMiddleware(app)
	# Add instrumentation
	app = AlluraTimerMiddleware(app, app_conf)
	# Clear cookies when the CSRF field isn't posted
	if not app_conf.get('disable_csrf_protection'):
	app = CSRFMiddleware(app, '_session_id')
	if asbool(config.get('cors.enabled', False)):
	# Handle CORS requests
	allowed_methods = aslist(config.get('cors.methods'))
	allowed_headers = aslist(config.get('cors.headers'))
	cache_duration = asint(config.get('cors.cache_duration', 0))
	app = CORSMiddleware(app, allowed_methods, allowed_headers, cache_duration)
	# Setup the allura SOPs
	app = allura_globals_middleware(app)
	# Ensure http and https used per config
	if config.get('override_root') != 'task':
	app = SSLMiddleware(app, app_conf.get('no_redirect.pattern'),
	app_conf.get('force_ssl.pattern'))
	# Setup resource manager, widget context SOP
	app = ew.WidgetMiddleware(
	app,
	compress=True,
	use_cache=not asbool(global_conf['debug']),
	script_name=app_conf.get('ew.script_name', '/_ew_resources/'),
	url_base=app_conf.get('ew.url_base', '/_ew_resources/'),
	extra_headers=ast.literal_eval(app_conf.get('ew.extra_headers', '[]')),
	cache_max_age=asint(app_conf.get('ew.cache_header_seconds', 606024*365)),

	# settings to pass through to jinja Environment for EW core widgets
	# these are for the easywidgets' own [easy_widgets.engines] entry point
	# (the Allura [easy_widgets.engines] entry point is named "jinja" (not jinja2) but it doesn't need
	# any settings since it is a class that uses the same jinja env as the rest of allura)
	**{
	'jinja2.auto_reload': asbool(config['auto_reload_templates']),
	'jinja2.bytecode_cache': AlluraJinjaRenderer._setup_bytecode_cache(),
	'jinja2.cache_size': asint(config.get('jinja_cache_size', -1)),
	}
	)
	# Handle static files (by tool)
	app = StaticFilesMiddleware(app, app_conf.get('static.script_name'))
	# Handle setup and flushing of Ming ORM sessions
	app = MingTaskSessionSetupMiddleware(app)
	app = MingMiddleware(app)
	# Set up the registry for stacked object proxies (SOPs).
	app = RegistryManager(app,
	# streaming=True causes cleanup problems when StatusCodeRedirect does an extra request
	streaming=False,
	preserve_exceptions=asbool(config['debug']), # allow inspecting them when debugging errors
	)

	# "task" wsgi would get a 2nd request to /error/document if we used this middleware
	if config.get('override_root') not in ('task', 'basetest_project_root'):
	if asbool(config['debug']):
	# Converts exceptions to HTTP errors, shows traceback in debug mode
	app = DebuggedApplication(app, evalex=True)
	else:
	app = ErrorMiddleware(app, config)

	app = SetRequestHostFromConfig(app, config)

	# Redirect some status codes to /error/document
	handle_status_codes = [403, 404, 410]
	if asbool(config['debug']):
	app = StatusCodeRedirect(app, handle_status_codes)
	else:
	app = StatusCodeRedirect(app, handle_status_codes + [500])

	for mw_ep in h.iter_entry_points('allura.middleware'):
	Middleware = mw_ep.load()
	if getattr(Middleware, 'when', 'inner') == 'outer':
	app = Middleware(app, config)

	return app


	def allura_globals_middleware(app):
	def AlluraGlobalsMiddleware(environ, start_response):
	import allura.lib.security
	import allura.lib.app_globals
	registry = environ['paste.registry']
	registry.register(allura.credentials,
	allura.lib.security.Credentials())
	return app(environ, start_response)
	return AlluraGlobalsMiddleware


	def get_tg_vars(context):
	import tg
	from allura.lib import helpers as h
	from urllib.parse import quote, quote_plus
	context.setdefault('g', tg.app_globals)
	context.setdefault('c', tg.tmpl_context)
	context.setdefault('h', h)
	context.setdefault('request', tg.request)
	context.setdefault('response', tg.response)
	context.setdefault('url', tg.url)
	context.setdefault('tg', dict(
	config=tg.config,
	flash_obj=tg.flash,
	quote=quote,
	quote_plus=quote_plus,
	url=tg.url))