NoNameItem / django-magnificent-messages / 49

Build # 49

Build Type

cron

travis-ci

Committed by artem.vasin

Commit Message

fix: updated migrations

Run Details

129 of 146 branches covered (88.36%)

Branch coverage included in aggregate %.

4 of 4 new or added lines in 1 file covered. (100.0%)

1 existing line in 1 file now uncovered.

833 of 850 relevant lines covered (98.0%)

19.6 hits per line

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

93.1

/django_magnificent_messages/storage/notification_storage/cookie.py

import json

from django import VERSION
from django.conf import settings

from .base import BaseNotificationStorage
from ..base import Message
from django.http import SimpleCookie
from django.utils.crypto import constant_time_compare, salted_hmac
from django.utils.safestring import SafeData, mark_safe


class NotificationsEncoder(json.JSONEncoder):
    """
    Compactly serialize instances of the ``Message`` class as JSON.
    """
    message_key = '__json_notification'

    def default(self, obj):
        if isinstance(obj, Message):
            # Using 0/1 here instead of False/True to produce more compact json
            is_safe_data = 1 if isinstance(obj.text, SafeData) else 0
            message = [self.message_key, is_safe_data, obj.level, obj.text, obj.subject, obj.extra]
            return message
        return super().default(obj)


class NotificationDecoder(json.JSONDecoder):
    """
    Decode JSON that includes serialized ``Message`` instances.
    """

    def process_notifications(self, obj):
        if isinstance(obj, list) and obj:
            if obj[0] == NotificationsEncoder.message_key:
                if obj[1]:
                    obj[3] = mark_safe(obj[3])
                return Message(*obj[2:])
            return [self.process_notifications(item) for item in obj]
        if isinstance(obj, dict):
            return {key: self.process_notifications(value)
                    for key, value in obj.items()}
        return obj

    def decode(self, s, **kwargs):
        decoded = super().decode(s, **kwargs)
        return self.process_notifications(decoded)


class CookieStorage(BaseNotificationStorage):
    """
    Store messages in a cookie.
    """
    cookie_name = 'notifications'
    # uwsgi's default configuration enforces a maximum size of 4kb for all the
    # HTTP headers. In order to leave some room for other cookies and headers,
    # restrict the session cookie to 1/2 of 4kb. See #18781.
    max_cookie_size = 2048
    not_finished = '__notificationsnotfinished__'

    def _get(self, *args, **kwargs):
        """
        Retrieve a list of notifications from the notifications cookie. If the
        not_finished sentinel value is found at the end of the notification list,
        remove it and return a result indicating that not all notifications were
        retrieved by this notification storage.
        """
        data = self.request.COOKIES.get(self.cookie_name)
        notifications = self._decode(data)
        all_retrieved = not (notifications and notifications[-1] == self.not_finished)
        if notifications and not all_retrieved:
            # remove the sentinel value
            notifications.pop()
        return notifications, all_retrieved

    def _update_cookie(self, encoded_data, response):
        """
        Either set the cookie with the encoded data if there is any data to
        store, or delete the cookie.
        """
        if encoded_data:
            if VERSION[:2] == (2, 0):
                response.set_cookie(
                    self.cookie_name, encoded_data,
                    domain=settings.SESSION_COOKIE_DOMAIN,
                    secure=settings.SESSION_COOKIE_SECURE or None,
                    httponly=settings.SESSION_COOKIE_HTTPONLY or None,
                )
            else:
                response.set_cookie(
                    self.cookie_name, encoded_data,
                    domain=settings.SESSION_COOKIE_DOMAIN,
                    secure=settings.SESSION_COOKIE_SECURE or None,
                    httponly=settings.SESSION_COOKIE_HTTPONLY or None,
                    samesite=settings.SESSION_COOKIE_SAMESITE,
                )
        else:
            response.delete_cookie(self.cookie_name, domain=settings.SESSION_COOKIE_DOMAIN)

    def _store(self, notifications, response, remove_oldest=True, *args, **kwargs):
        """
        Store the notifications to a cookie and return a list of any notifications which
        could not be stored.

        If the encoded data is larger than ``max_cookie_size``, remove
        notifications until the data fits (these are the notifications which are
        returned), and notifications_add the not_finished sentinel value to indicate as much.
        """
        unstored_notifications = []
        encoded_data = self._encode(notifications)
        if self.max_cookie_size:
            # data is going to be stored eventually by SimpleCookie, which
            # adds its own overhead, which we must account for.
            cookie = SimpleCookie()  # create outside the loop

            def stored_length(val):
                return len(cookie.value_encode(val)[1])

            while encoded_data and stored_length(encoded_data) > self.max_cookie_size:
                if remove_oldest:
                    unstored_notifications.append(notifications.pop(0))
                else:
                    unstored_notifications.insert(0, notifications.pop())
                encoded_data = self._encode(notifications + [self.not_finished],
                                            encode_empty=bool(unstored_notifications))
        self._update_cookie(encoded_data, response)
        return unstored_notifications

    @staticmethod
    def _hash(value):
        """
        Create an HMAC/SHA1 hash based on the value and the project setting's
        SECRET_KEY, modified to make it unique for the present purpose.
        """
        key_salt = 'django_magnificent_messages'
        return salted_hmac(key_salt, value).hexdigest()

    def _encode(self, notifications, encode_empty=False):
        """
        Return an encoded version of the notifications list which can be stored as
        plain text.

        Since the data will be retrieved from the client-side, the encoded data
        also contains a hash to ensure that the data was not tampered with.
        """
        if notifications or encode_empty:
            encoder = NotificationsEncoder(separators=(',', ':'))
            value = encoder.encode(notifications)
            return '%s$%s' % (self._hash(value), value)

    def _decode(self, data):
        """
        Safely decode an encoded text stream back into a list of notifications.

        If the encoded text stream contained an invalid hash or was in an
        invalid format, return None.
        """
        if not data:
            return None
        bits = data.split('$', 1)
        if len(bits) == 2:
            _hash, value = bits
            if constant_time_compare(_hash, self._hash(value)):
                try:
                    # If we get here (and the JSON decode works), everything is
                    # good. In any other case, drop back and return None.
                    return json.loads(value, cls=NotificationDecoder)
                except json.JSONDecodeError:
                    pass
        # Mark the data as used (so it gets removed) since something was wrong
        # with the data.
        self.used = True
        return None

1	import json	20✔
2
3	from django import VERSION	20✔
4	from django.conf import settings	20✔
5
6	from .base import BaseNotificationStorage	20✔
7	from ..base import Message	20✔
8	from django.http import SimpleCookie	20✔
9	from django.utils.crypto import constant_time_compare, salted_hmac	20✔
10	from django.utils.safestring import SafeData, mark_safe	20✔
11
12
13	class NotificationsEncoder(json.JSONEncoder):	20✔
14	"""
15	Compactly serialize instances of the ``Message`` class as JSON.
16	"""
17	message_key = '__json_notification'	20✔
18
19	def default(self, obj):	20✔
20	if isinstance(obj, Message):	20!
21	# Using 0/1 here instead of False/True to produce more compact json
22	is_safe_data = 1 if isinstance(obj.text, SafeData) else 0	20✔
23	message = [self.message_key, is_safe_data, obj.level, obj.text, obj.subject, obj.extra]	20✔
24	return message	20✔
25	return super().default(obj)	×
26
27
28	class NotificationDecoder(json.JSONDecoder):	20✔
29	"""
30	Decode JSON that includes serialized ``Message`` instances.
31	"""
32
33	def process_notifications(self, obj):	20✔
34	if isinstance(obj, list) and obj:	20✔
35	if obj[0] == NotificationsEncoder.message_key:	20✔
36	if obj[1]:	20✔
37	obj[3] = mark_safe(obj[3])	20✔
38	return Message(*obj[2:])	20✔
39	return [self.process_notifications(item) for item in obj]	20✔
40	if isinstance(obj, dict):	20✔
41	return {key: self.process_notifications(value)	20✔
42	for key, value in obj.items()}
43	return obj	20✔
44
45	def decode(self, s, **kwargs):	20✔
46	decoded = super().decode(s, **kwargs)	20✔
47	return self.process_notifications(decoded)	20✔
48
49
50	class CookieStorage(BaseNotificationStorage):	20✔
51	"""
52	Store messages in a cookie.
53	"""
54	cookie_name = 'notifications'	20✔
55	# uwsgi's default configuration enforces a maximum size of 4kb for all the
56	# HTTP headers. In order to leave some room for other cookies and headers,
57	# restrict the session cookie to 1/2 of 4kb. See #18781.
58	max_cookie_size = 2048	20✔
59	not_finished = '__notificationsnotfinished__'	20✔
60
61	def _get(self, args, *kwargs):	20✔
62	"""
63	Retrieve a list of notifications from the notifications cookie. If the
64	not_finished sentinel value is found at the end of the notification list,
65	remove it and return a result indicating that not all notifications were
66	retrieved by this notification storage.
67	"""
68	data = self.request.COOKIES.get(self.cookie_name)	20✔
69	notifications = self._decode(data)	20✔
70	all_retrieved = not (notifications and notifications[-1] == self.not_finished)	20✔
71	if notifications and not all_retrieved:	20✔
72	# remove the sentinel value
73	notifications.pop()	20✔
74	return notifications, all_retrieved	20✔
75
76	def _update_cookie(self, encoded_data, response):	20✔
77	"""
78	Either set the cookie with the encoded data if there is any data to
79	store, or delete the cookie.
80	"""
81	if encoded_data:	20✔
82	if VERSION[:2] == (2, 0):	20!
UNCOV 83	response.set_cookie(	×
84	self.cookie_name, encoded_data,
85	domain=settings.SESSION_COOKIE_DOMAIN,
86	secure=settings.SESSION_COOKIE_SECURE or None,
87	httponly=settings.SESSION_COOKIE_HTTPONLY or None,
88	)
89	else:
90	response.set_cookie(	20✔
91	self.cookie_name, encoded_data,
92	domain=settings.SESSION_COOKIE_DOMAIN,
93	secure=settings.SESSION_COOKIE_SECURE or None,
94	httponly=settings.SESSION_COOKIE_HTTPONLY or None,
95	samesite=settings.SESSION_COOKIE_SAMESITE,
96	)
97	else:
98	response.delete_cookie(self.cookie_name, domain=settings.SESSION_COOKIE_DOMAIN)	20✔
99
100	def _store(self, notifications, response, remove_oldest=True, args, *kwargs):	20✔
101	"""
102	Store the notifications to a cookie and return a list of any notifications which
103	could not be stored.
104
105	If the encoded data is larger than ``max_cookie_size``, remove
106	notifications until the data fits (these are the notifications which are
107	returned), and notifications_add the not_finished sentinel value to indicate as much.
108	"""
109	unstored_notifications = []	20✔
110	encoded_data = self._encode(notifications)	20✔
111	if self.max_cookie_size:	20!
112	# data is going to be stored eventually by SimpleCookie, which
113	# adds its own overhead, which we must account for.
114	cookie = SimpleCookie() # create outside the loop	20✔
115
116	def stored_length(val):	20✔
117	return len(cookie.value_encode(val)[1])	20✔
118
119	while encoded_data and stored_length(encoded_data) > self.max_cookie_size:	20✔
120	if remove_oldest:	20✔
121	unstored_notifications.append(notifications.pop(0))	20✔
122	else:
123	unstored_notifications.insert(0, notifications.pop())	20✔
124	encoded_data = self._encode(notifications + [self.not_finished],	20✔
125	encode_empty=bool(unstored_notifications))
126	self._update_cookie(encoded_data, response)	20✔
127	return unstored_notifications	20✔
128
129	@staticmethod	20✔
130	def _hash(value):	8✔
131	"""
132	Create an HMAC/SHA1 hash based on the value and the project setting's
133	SECRET_KEY, modified to make it unique for the present purpose.
134	"""
135	key_salt = 'django_magnificent_messages'	20✔
136	return salted_hmac(key_salt, value).hexdigest()	20✔
137
138	def _encode(self, notifications, encode_empty=False):	20✔
139	"""
140	Return an encoded version of the notifications list which can be stored as
141	plain text.
142
143	Since the data will be retrieved from the client-side, the encoded data
144	also contains a hash to ensure that the data was not tampered with.
145	"""
146	if notifications or encode_empty:	20✔
147	encoder = NotificationsEncoder(separators=(',', ':'))	20✔
148	value = encoder.encode(notifications)	20✔
149	return '%s$%s' % (self._hash(value), value)	20✔
150
151	def _decode(self, data):	20✔
152	"""
153	Safely decode an encoded text stream back into a list of notifications.
154
155	If the encoded text stream contained an invalid hash or was in an
156	invalid format, return None.
157	"""
158	if not data:	20✔
159	return None	20✔
160	bits = data.split('$', 1)	20✔
161	if len(bits) == 2:	20!
162	_hash, value = bits	20✔
163	if constant_time_compare(_hash, self._hash(value)):	20✔
164	try:	20✔
165	# If we get here (and the JSON decode works), everything is
166	# good. In any other case, drop back and return None.
167	return json.loads(value, cls=NotificationDecoder)	20✔
168	except json.JSONDecodeError:	×
169	pass	×
170	# Mark the data as used (so it gets removed) since something was wrong
171	# with the data.
172	self.used = True	20✔
173	return None	20✔

NoNameItem / django-magnificent-messages / 49

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