10388083977

Committed 14 Aug 2024 01:01PM UTC coverage: 88.203% (-0.007%) from 88.21%

Build # 10388083977

Build Type

Pull #545

github

Committed by

web-flow

Commit Message

Merge 0ad8d66a2 into 2a2576628

Pull Request Pull Request #545: Add ability to define different widget layouts for different screen sizes

Run Details

8209 of 9836 branches covered (83.46%)

Branch coverage included in aggregate %.

459 of 517 new or added lines in 18 files covered. (88.78%)

31 existing lines in 8 files now uncovered.

21697 of 24070 relevant lines covered (90.14%)

31.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.86

src/wirecloud/commons/middleware.py

# -*- coding: utf-8 -*-

# Copyright (c) 2008-2016 CoNWeT Lab., Universidad Politécnica de Madrid
# Copyright (c) 2019-2020 Future Internet Consulting and Development Solutions S.L.

# This file is part of Wirecloud.

# Wirecloud is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# Wirecloud is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public License
# along with Wirecloud.  If not, see <http://www.gnu.org/licenses/>.

from importlib import import_module

from django.contrib.auth.middleware import get_user
from django.urls import reverse
from django.utils import translation
from django.utils.deprecation import MiddlewareMixin
from django.utils.functional import SimpleLazyObject
from django.utils.translation import ugettext as _

from wirecloud.commons.exceptions import HttpBadCredentials


class URLMiddleware(MiddlewareMixin):

    _middleware = None
    _path_mapping = None

    def __init__(self, get_response=None):
        self.get_response = get_response
        self._middleware = {}

    def load_middleware(self, group):
        """
        Populate middleware lists from settings.URL_MIDDLEWARE_CLASSES.
        """
        from django.conf import settings
        from django.core import exceptions

        middleware = {
            'process_request': [],
            'process_view': [],
            'process_template_response': [],
            'process_response': [],
            'process_exception': [],
        }
        for middleware_path in settings.URL_MIDDLEWARE_CLASSES[group]:
            try:
                mw_module, mw_classname = middleware_path.rsplit('.', 1)
            except ValueError:
                raise exceptions.ImproperlyConfigured('%s isn\'t a middleware module' % middleware_path)
            try:
                mod = import_module(mw_module)
            except ImportError as e:
                raise exceptions.ImproperlyConfigured('Error importing middleware %s: "%s"' % (mw_module, e))
            try:
                mw_class = getattr(mod, mw_classname)
            except AttributeError:
                raise exceptions.ImproperlyConfigured('Middleware module "%s" does not define a "%s" class' % (mw_module, mw_classname))
            try:
                mw_instance = mw_class()
            except exceptions.MiddlewareNotUsed:
                continue
            except Exception:
                raise exceptions.ImproperlyConfigured('Middleware module "%s" does not define a "%s" class' % (mw_module, mw_classname))

            if hasattr(mw_instance, 'process_request'):
                middleware['process_request'].append(mw_instance.process_request)
            if hasattr(mw_instance, 'process_view'):
                middleware['process_view'].append(mw_instance.process_view)
            if hasattr(mw_instance, 'process_template_response'):
                middleware['process_template_response'].insert(0, mw_instance.process_template_response)
            if hasattr(mw_instance, 'process_response'):
                middleware['process_response'].insert(0, mw_instance.process_response)
            if hasattr(mw_instance, 'process_exception'):
                middleware['process_exception'].insert(0, mw_instance.process_exception)

        # We only assign to this when initialization is complete as it is used
        # as a flag for initialization being complete.
        self._middleware[group] = middleware

    def get_matched_middleware(self, path, middleware_method):

        if self._path_mapping is None:

            from django.conf import settings

            self._path_mapping = {
                '/api/': 'api'
            }

            if 'wirecloud.platform' in settings.INSTALLED_APPS:
                proxy_path = reverse('wirecloud|proxy', kwargs={'protocol': 'a', 'domain': 'a', 'path': ''})[:-len('a/a')]
                self._path_mapping[proxy_path] = 'proxy'

        group = 'default'
        for path_mapping in self._path_mapping:
            if path.startswith(path_mapping):
                group = self._path_mapping[path_mapping]
                break

        if group not in self._middleware:
            self.load_middleware(group)

        return self._middleware[group][middleware_method]

    def process_request(self, request):
        matched_middleware = self.get_matched_middleware(request.path, 'process_request')
        for middleware in matched_middleware:
            response = middleware(request)
            if response:
                return response

    def process_view(self, request, view_func, view_args, view_kwargs):
        matched_middleware = self.get_matched_middleware(request.path, 'process_view')
        for middleware in matched_middleware:
            response = middleware(request, view_func, view_args, view_kwargs)
            if response:
                return response

    def process_template_response(self, request, response):
        matched_middleware = self.get_matched_middleware(request.path, 'process_template_response')
        for middleware in matched_middleware:
            response = middleware(request, response)
        return response

    def process_response(self, request, response):
        matched_middleware = self.get_matched_middleware(request.path, 'process_response')
        for middleware in matched_middleware:
            response = middleware(request, response)
        return response

    def process_exception(self, request, exception):
        matched_middleware = self.get_matched_middleware(request.path, 'process_exception')
        for middleware in matched_middleware:
            response = middleware(request, exception)
            if response:
                return response


class LocaleMiddleware(MiddlewareMixin):
    """
    Parse a request and decide what translation object to install in the
    current thread context. This allows pages to be dynamically translated to
    the language the user desires (if the language is available, of course).
    """

    def process_request(self, request):
        if 'lang' in request.GET and translation.check_for_language(request.GET['lang']):
            language = request.GET['lang']
        elif 'lang' in request.COOKIES and translation.check_for_language(request.COOKIES['lang']):
            language = request.COOKIES['lang']
        else:
            language = translation.get_language_from_request(request, check_path=False)
        translation.activate(language)
        request.LANGUAGE_CODE = translation.get_language()

    def process_response(self, request, response):
        language = translation.get_language()
        response.setdefault('Content-Language', language)
        return response


def get_api_user(request):

    from wirecloud.platform.plugins import get_api_auth_backends

    parts = request.META['HTTP_AUTHORIZATION'].split(' ', 1)
    if len(parts) != 2:
        raise HttpBadCredentials(_('Bad credentials'))

    (auth_type, token) = parts
    backends = get_api_auth_backends()
    try:
        return backends[auth_type](auth_type, token)
    except HttpBadCredentials:
        raise
    except Exception:
        raise HttpBadCredentials(_('Bad credentials'), '%(auth_type)s realm="WireCloud", error="invalid_token", error_description="bad credentials"' % {"auth_type": auth_type})


class AuthenticationMiddleware(object):

    def process_request(self, request):

        if 'HTTP_AUTHORIZATION' in request.META:
            request.user = SimpleLazyObject(lambda: get_api_user(request))
        else:
            request.user = SimpleLazyObject(lambda: get_user(request))

1	# -- coding: utf-8 --
2
3	# Copyright (c) 2008-2016 CoNWeT Lab., Universidad Politécnica de Madrid
4	# Copyright (c) 2019-2020 Future Internet Consulting and Development Solutions S.L.
5
6	# This file is part of Wirecloud.
7
8	# Wirecloud is free software: you can redistribute it and/or modify
9	# it under the terms of the GNU Affero General Public License as published by
10	# the Free Software Foundation, either version 3 of the License, or
11	# (at your option) any later version.
12
13	# Wirecloud is distributed in the hope that it will be useful,
14	# but WITHOUT ANY WARRANTY; without even the implied warranty of
15	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16	# GNU Affero General Public License for more details.
17
18	# You should have received a copy of the GNU Affero General Public License
19	# along with Wirecloud. If not, see <http://www.gnu.org/licenses/>.
20
21	from importlib import import_module	15✔
22
23	from django.contrib.auth.middleware import get_user	15✔
24	from django.urls import reverse	15✔
25	from django.utils import translation	15✔
26	from django.utils.deprecation import MiddlewareMixin	15✔
27	from django.utils.functional import SimpleLazyObject	15✔
28	from django.utils.translation import ugettext as _	15✔
29
30	from wirecloud.commons.exceptions import HttpBadCredentials	15✔
31
32
33	class URLMiddleware(MiddlewareMixin):	15✔
34
35	_middleware = None	15✔
36	_path_mapping = None	15✔
37
38	def __init__(self, get_response=None):	15✔
39	self.get_response = get_response	15✔
40	self._middleware = {}	15✔
41
42	def load_middleware(self, group):	15✔
43	"""
44	Populate middleware lists from settings.URL_MIDDLEWARE_CLASSES.
45	"""
46	from django.conf import settings	15✔
47	from django.core import exceptions	15✔
48
49	middleware = {	15✔
50	'process_request': [],
51	'process_view': [],
52	'process_template_response': [],
53	'process_response': [],
54	'process_exception': [],
55	}
56	for middleware_path in settings.URL_MIDDLEWARE_CLASSES[group]:	15✔
57	try:	15✔
58	mw_module, mw_classname = middleware_path.rsplit('.', 1)	15✔
59	except ValueError:	12✔
60	raise exceptions.ImproperlyConfigured('%s isn\'t a middleware module' % middleware_path)	12✔
61	try:	15✔
62	mod = import_module(mw_module)	15✔
63	except ImportError as e:	12✔
64	raise exceptions.ImproperlyConfigured('Error importing middleware %s: "%s"' % (mw_module, e))	12✔
65	try:	15✔
66	mw_class = getattr(mod, mw_classname)	15✔
67	except AttributeError:	12✔
68	raise exceptions.ImproperlyConfigured('Middleware module "%s" does not define a "%s" class' % (mw_module, mw_classname))	12✔
69	try:	15✔
70	mw_instance = mw_class()	15✔
71	except exceptions.MiddlewareNotUsed:	12✔
72	continue	12✔
73	except Exception:	12✔
74	raise exceptions.ImproperlyConfigured('Middleware module "%s" does not define a "%s" class' % (mw_module, mw_classname))	12✔
75
76	if hasattr(mw_instance, 'process_request'):	15✔
77	middleware['process_request'].append(mw_instance.process_request)	15✔
78	if hasattr(mw_instance, 'process_view'):	15✔
79	middleware['process_view'].append(mw_instance.process_view)	12✔
80	if hasattr(mw_instance, 'process_template_response'):	15✔
81	middleware['process_template_response'].insert(0, mw_instance.process_template_response)	12✔
82	if hasattr(mw_instance, 'process_response'):	15✔
83	middleware['process_response'].insert(0, mw_instance.process_response)	15✔
84	if hasattr(mw_instance, 'process_exception'):	15✔
85	middleware['process_exception'].insert(0, mw_instance.process_exception)	12✔
86
87	# We only assign to this when initialization is complete as it is used
88	# as a flag for initialization being complete.
89	self._middleware[group] = middleware	15✔
90
91	def get_matched_middleware(self, path, middleware_method):	15✔
92
93	if self._path_mapping is None:	15✔
94
95	from django.conf import settings	15✔
96
97	self._path_mapping = {	15✔
98	'/api/': 'api'
99	}
100
101	if 'wirecloud.platform' in settings.INSTALLED_APPS:	15✔
102	proxy_path = reverse('wirecloud\|proxy', kwargs={'protocol': 'a', 'domain': 'a', 'path': ''})[:-len('a/a')]	15✔
103	self._path_mapping[proxy_path] = 'proxy'	15✔
104
105	group = 'default'	15✔
106	for path_mapping in self._path_mapping:	15✔
107	if path.startswith(path_mapping):	15✔
108	group = self._path_mapping[path_mapping]	15✔
109	break	15✔
110
111	if group not in self._middleware:	15✔
112	self.load_middleware(group)	15✔
113
114	return self._middleware[group][middleware_method]	15✔
115
116	def process_request(self, request):	15✔
117	matched_middleware = self.get_matched_middleware(request.path, 'process_request')	15✔
118	for middleware in matched_middleware:	15✔
119	response = middleware(request)	15✔
120	if response:	15✔
121	return response	12✔
122
123	def process_view(self, request, view_func, view_args, view_kwargs):	15✔
124	matched_middleware = self.get_matched_middleware(request.path, 'process_view')	15✔
125	for middleware in matched_middleware:	15✔
126	response = middleware(request, view_func, view_args, view_kwargs)	12✔
127	if response:	12✔
128	return response	12✔
129
130	def process_template_response(self, request, response):	15✔
131	matched_middleware = self.get_matched_middleware(request.path, 'process_template_response')	13✔
132	for middleware in matched_middleware:	13✔
133	response = middleware(request, response)	12✔
134	return response	13✔
135
136	def process_response(self, request, response):	15✔
137	matched_middleware = self.get_matched_middleware(request.path, 'process_response')	15✔
138	for middleware in matched_middleware:	15✔
139	response = middleware(request, response)	15✔
140	return response	15✔
141
142	def process_exception(self, request, exception):	15✔
143	matched_middleware = self.get_matched_middleware(request.path, 'process_exception')	13✔
144	for middleware in matched_middleware:	13✔
145	response = middleware(request, exception)	12✔
146	if response:	12✔
147	return response	12✔
148
149
150	class LocaleMiddleware(MiddlewareMixin):	15✔
151	"""
152	Parse a request and decide what translation object to install in the
153	current thread context. This allows pages to be dynamically translated to
154	the language the user desires (if the language is available, of course).
155	"""
156
157	def process_request(self, request):	15✔
158	if 'lang' in request.GET and translation.check_for_language(request.GET['lang']):	15✔
159	language = request.GET['lang']	13✔
160	elif 'lang' in request.COOKIES and translation.check_for_language(request.COOKIES['lang']):	15!
UNCOV 161	language = request.COOKIES['lang']	×
162	else:
163	language = translation.get_language_from_request(request, check_path=False)	15✔
164	translation.activate(language)	15✔
165	request.LANGUAGE_CODE = translation.get_language()	15✔
166
167	def process_response(self, request, response):	15✔
168	language = translation.get_language()	15✔
169	response.setdefault('Content-Language', language)	15✔
170	return response	15✔
171
172
173	def get_api_user(request):	15✔
174
175	from wirecloud.platform.plugins import get_api_auth_backends	12✔
176
177	parts = request.META['HTTP_AUTHORIZATION'].split(' ', 1)	12✔
178	if len(parts) != 2:	12✔
179	raise HttpBadCredentials(_('Bad credentials'))	12✔
180
181	(auth_type, token) = parts	12✔
182	backends = get_api_auth_backends()	12✔
183	try:	12✔
184	return backends[auth_type](auth_type, token)	12✔
185	except HttpBadCredentials:	12✔
186	raise	12✔
187	except Exception:	12✔
188	raise HttpBadCredentials(_('Bad credentials'), '%(auth_type)s realm="WireCloud", error="invalid_token", error_description="bad credentials"' % {"auth_type": auth_type})	12✔
189
190
191	class AuthenticationMiddleware(object):	15✔
192
193	def process_request(self, request):	15✔
194
195	if 'HTTP_AUTHORIZATION' in request.META:	15✔
196	request.user = SimpleLazyObject(lambda: get_api_user(request))	12✔
197	else:
198	request.user = SimpleLazyObject(lambda: get_user(request))	15✔

Wirecloud / wirecloud / 10388083977

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous