1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
|
"""Internal utilities for generating cache keys that are used for request matching
.. automodsumm:: requests_cache.cache_keys
:functions-only:
:nosignatures:
"""
from __future__ import annotations
import json
from hashlib import blake2b
from logging import getLogger
from typing import (
TYPE_CHECKING,
Any,
Dict,
Iterable,
List,
Mapping,
MutableMapping,
Optional,
Union,
)
from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
from requests import Request, Session
from requests.models import CaseInsensitiveDict
from url_normalize import url_normalize
from ._utils import decode, encode
__all__ = [
'create_key',
'normalize_body',
'normalize_headers',
'normalize_request',
'normalize_url',
]
if TYPE_CHECKING:
from .models import AnyPreparedRequest, AnyRequest, CachedResponse
# Maximum JSON request body size that will be normalized
MAX_NORM_BODY_SIZE = 10 * 1024 * 1024
ParamList = Optional[Iterable[str]]
RequestContent = Union[Mapping, str, bytes]
logger = getLogger(__name__)
def create_key(
request: AnyRequest,
ignored_parameters: ParamList = None,
match_headers: Union[ParamList, bool] = False,
serializer: Any = None,
**request_kwargs,
) -> str:
"""Create a normalized cache key based on a request object
Args:
request: Request object to generate a cache key from
ignored_parameters: Request paramters, headers, and/or JSON body params to exclude
match_headers: Match only the specified headers, or ``True`` to match all headers
request_kwargs: Additional keyword arguments for :py:func:`~requests.request`
"""
# Normalize and gather all relevant request info to match against
request = normalize_request(request, ignored_parameters)
key_parts = [
request.method or '',
request.url,
request.body or '',
request_kwargs.get('verify', True),
*get_matched_headers(request.headers, match_headers),
str(serializer),
]
# Generate a hash based on this info
key = blake2b(digest_size=8)
for part in key_parts:
key.update(encode(part))
return key.hexdigest()
def get_matched_headers(
headers: CaseInsensitiveDict, match_headers: Union[ParamList, bool]
) -> List[str]:
"""Get only the headers we should match against as a list of ``k=v`` strings, given an optional
include list.
"""
if not match_headers:
return []
if match_headers is True:
match_headers = headers
return [
f'{k.lower()}={headers[k]}'
for k in sorted(match_headers, key=lambda x: x.lower())
if k in headers
]
def normalize_request(
request: AnyRequest, ignored_parameters: ParamList = None
) -> AnyPreparedRequest:
"""Normalize and remove ignored parameters from request URL, body, and headers.
This is used for both:
* Increasing cache hits by generating more precise cache keys
* Redacting potentially sensitive info from cached requests
Args:
request: Request object to normalize
ignored_parameters: Request paramters, headers, and/or JSON body params to exclude
"""
if isinstance(request, Request):
norm_request: AnyPreparedRequest = Session().prepare_request(request)
else:
norm_request = request.copy()
norm_request.method = (norm_request.method or '').upper()
norm_request.url = normalize_url(norm_request.url or '', ignored_parameters)
norm_request.headers = normalize_headers(norm_request.headers, ignored_parameters)
norm_request.body = normalize_body(norm_request, ignored_parameters)
return norm_request
def normalize_headers(
headers: MutableMapping[str, str], ignored_parameters: ParamList = None
) -> CaseInsensitiveDict:
"""Sort and filter request headers, and normalize minor variations in multi-value headers"""
if ignored_parameters:
headers = filter_sort_dict(headers, ignored_parameters)
for k, v in headers.items():
if ',' in v:
values = [v.strip() for v in v.lower().split(',') if v.strip()]
headers[k] = ', '.join(sorted(values))
return CaseInsensitiveDict(headers)
def normalize_url(url: str, ignored_parameters: ParamList) -> str:
"""Normalize and filter a URL. This includes request parameters, IDN domains, scheme, host,
port, etc.
"""
# Strip query params from URL, sort and filter, and reassemble into a complete URL
url_tokens = urlparse(url)
url = urlunparse(
(
url_tokens.scheme,
url_tokens.netloc,
url_tokens.path,
url_tokens.params,
normalize_params(url_tokens.query, ignored_parameters),
url_tokens.fragment,
)
)
return url_normalize(url)
def normalize_body(request: AnyPreparedRequest, ignored_parameters: ParamList) -> bytes:
"""Normalize and filter a request body if possible, depending on Content-Type"""
original_body = request.body or b''
content_type = request.headers.get('Content-Type')
# Filter and sort params if possible
filtered_body: Union[str, bytes] = original_body
if content_type == 'application/json':
filtered_body = normalize_json_body(original_body, ignored_parameters)
elif content_type == 'application/x-www-form-urlencoded':
filtered_body = normalize_params(original_body, ignored_parameters)
return encode(filtered_body)
def normalize_json_body(
original_body: Union[str, bytes], ignored_parameters: ParamList
) -> Union[str, bytes]:
"""Normalize and filter a request body with serialized JSON data"""
if len(original_body) <= 2 or len(original_body) > MAX_NORM_BODY_SIZE:
return original_body
try:
body = json.loads(decode(original_body))
body = filter_sort_json(body, ignored_parameters)
return json.dumps(body)
# If it's invalid JSON, then don't mess with it
except (AttributeError, TypeError, ValueError):
logger.debug('Invalid JSON body')
return original_body
def normalize_params(value: Union[str, bytes], ignored_parameters: ParamList) -> str:
"""Normalize and filter urlencoded params from either a URL or request body with form data"""
query_str = decode(value)
params = dict(parse_qsl(query_str))
# parse_qsl doesn't handle key-only params, so add those here
key_only_params = [k for k in query_str.split('&') if k and '=' not in k]
params.update({k: '' for k in key_only_params})
params = filter_sort_dict(params, ignored_parameters)
return urlencode(params)
def redact_response(response: CachedResponse, ignored_parameters: ParamList) -> CachedResponse:
"""Redact any ignored parameters (potentially containing sensitive info) from a cached request"""
if ignored_parameters:
response.url = normalize_url(response.url, ignored_parameters)
response.request = normalize_request(response.request, ignored_parameters) # type: ignore
return response
def filter_sort_json(data: Union[List, Mapping], ignored_parameters: ParamList):
if isinstance(data, Mapping):
return filter_sort_dict(data, ignored_parameters)
else:
return filter_sort_list(data, ignored_parameters)
def filter_sort_dict(data: Mapping[str, str], ignored_parameters: ParamList) -> Dict[str, str]:
if not ignored_parameters:
return dict(sorted(data.items()))
return {k: v for k, v in sorted(data.items()) if k not in set(ignored_parameters)}
def filter_sort_list(data: List, ignored_parameters: ParamList) -> List:
if not ignored_parameters:
return sorted(data)
return [k for k in sorted(data) if k not in set(ignored_parameters)]
|