lms.util.net
Utilities for network and HTTP.
1""" 2Utilities for network and HTTP. 3""" 4 5import re 6import typing 7import urllib.parse 8 9import bs4 10import edq.net.exchange 11import edq.util.json 12import requests 13 14import lms.model.constants 15 16CANVAS_CLEAN_REMOVE_CONTENT_KEYS: typing.List[str] = [ 17 'created_at', 18 'ics', 19 'last_activity_at', 20 'lti_context_id', 21 'preview_url', 22 'secure_params', 23 'total_activity_time', 24 'updated_at', 25 'url', 26 'uuid', 27] 28""" Keys to remove from Canvas content. """ 29 30BLACKBOARD_CLEAN_REMOVE_CONTENT_KEYS: typing.List[str] = [ 31 'created', 32 'modified', 33] 34""" Keys to remove from Blackboard content. """ 35 36BLACKBOARD_CLEAN_REMOVE_HEADERS: typing.Set[str] = { 37 'access-control-allow-origin', 38 'content-encoding', 39 'content-language', 40 'expires', 41 'last-modified', 42 'p3p', 43 'pragma', 44 'strict-transport-security', 45 'transfer-encoding', 46 'vary', 47 'x-blackboard-xsrf', 48} 49""" Keys to remove from Blackboard headers. """ 50 51MOODLE_CLEAN_REMOVE_HEADERS: typing.Set[str] = { 52 'accept-ranges', 53 'content-encoding', 54 'content-language', 55 'content-script-type', 56 'content-style-type', 57 'expires', 58 'keep-alive', 59 'last-modified', 60 'pragma', 61 'vary', 62} 63""" Keys to remove from Moodle headers. """ 64 65MOODLE_FINALIZE_REMOVE_PARAMS: typing.Set[str] = { 66 'logintoken', 67} 68""" Keys to remove from Moodle headers. """ 69 70STANDARDIZED_TIMESTAMP: str = '123456789' 71STANDARDIZED_SESSION_KEY: str = 'abcABC123' 72STANDARDIZED_RANDOM_STRING: str = 'abc123' 73 74def clean_lms_response(response: requests.Response, body: str) -> str: 75 """ 76 A ResponseModifierFunction that attempt to identify 77 if the requests comes from a Learning Management System (LMS), 78 and clean the response accordingly. 79 """ 80 81 # Check the standard LMS Toolkit backend header. 82 backend_type = response.headers.get(lms.model.constants.HEADER_KEY_BACKEND, '').lower() 83 84 if (backend_type == lms.model.constants.BACKEND_TYPE_CANVAS): 85 return clean_canvas_response(response, body) 86 87 if (backend_type == lms.model.constants.BACKEND_TYPE_MOODLE): 88 return clean_moodle_response(response, body) 89 90 # Try looking inside the header keys. 91 for key in response.headers: 92 key = key.lower().strip() 93 94 if ('blackboard' in key): 95 return clean_blackboard_response(response, body) 96 97 if ('canvas' in key): 98 return clean_canvas_response(response, body) 99 100 if ('moodle' in key): 101 return clean_moodle_response(response, body) 102 103 return body 104 105def clean_blackboard_response(response: requests.Response, body: str) -> str: 106 """ 107 See clean_lms_response(), but specifically for the Blackboard LMS. 108 This function will: 109 - Call _clean_base_response(). 110 - Remove specific headers. 111 """ 112 113 body = _clean_base_response(response, body) 114 115 # Work on both request and response headers. 116 for headers in [response.headers, response.request.headers]: 117 for key in list(headers.keys()): # type: ignore[attr-defined] 118 if (key.strip().lower() in BLACKBOARD_CLEAN_REMOVE_HEADERS): 119 headers.pop(key, None) # type: ignore[attr-defined] 120 121 # Most blackboard responses are JSON. 122 try: 123 data = edq.util.json.loads(body, strict = True) 124 except Exception: 125 # Response is not JSON. 126 return body 127 128 # Remove any content keys. 129 _recursive_remove_keys(data, set(BLACKBOARD_CLEAN_REMOVE_CONTENT_KEYS)) 130 131 # Convert body back to a string. 132 body = edq.util.json.dumps(data) 133 134 return body 135 136def clean_canvas_response(response: requests.Response, body: str) -> str: 137 """ 138 See clean_lms_response(), but specifically for the Canvas LMS. 139 This function will: 140 - Call _clean_base_response(). 141 - Remove content keys: [last_activity_at, total_activity_time] 142 """ 143 144 body = _clean_base_response(response, body) 145 146 # Most canvas responses are JSON. 147 try: 148 data = edq.util.json.loads(body, strict = True) 149 except Exception: 150 # Response is not JSON. 151 return body 152 153 # Remove any content keys. 154 _recursive_remove_keys(data, set(CANVAS_CLEAN_REMOVE_CONTENT_KEYS)) 155 156 # Remove special fields. 157 158 if ('submissions/update_grades' in str(response.request.url)): 159 data.pop('id', None) 160 161 # Convert body back to a string. 162 body = edq.util.json.dumps(data) 163 164 return body 165 166def clean_moodle_response(response: requests.Response, body: str) -> str: 167 """ 168 See clean_lms_response(), but specifically for the Moodle LMS. 169 This function will: 170 - Call _clean_base_response(). 171 """ 172 173 body = _clean_base_response(response, body) 174 175 # Standardize timestamp. 176 current_timestamp_match = re.search(r"boost/theme/(\d{10})/favicon", body) 177 if (current_timestamp_match is not None): 178 body = body.replace(current_timestamp_match.group(1), STANDARDIZED_TIMESTAMP) 179 180 # Standardize session key. 181 session_key_match = re.search(r'"sesskey":"([^"]+)"', body) 182 if (session_key_match is not None): 183 body = body.replace(session_key_match.group(1), STANDARDIZED_SESSION_KEY) 184 185 # Standardize "random" string. 186 random_string_match = re.search(r"'random([a-z0-9]+)'", body) 187 if (random_string_match is not None): 188 body = body.replace(random_string_match.group(1), STANDARDIZED_RANDOM_STRING) 189 190 # Standardize logintoken. 191 logintoken_match = re.search(r'name="logintoken" value="(\w+)"', body) 192 if (logintoken_match is not None): 193 body = body.replace(logintoken_match.group(1), STANDARDIZED_SESSION_KEY) 194 195 # Work on both request and response headers. 196 for headers in [response.headers, response.request.headers]: 197 for key in list(headers.keys()): # type: ignore[attr-defined] 198 if (key.strip().lower() in MOODLE_CLEAN_REMOVE_HEADERS): 199 headers.pop(key, None) # type: ignore[attr-defined] 200 201 # Endpoint-Specific Tasks 202 203 # Remove extra data from the course participants response. 204 if (re.search(r'/user/index\.php\?id=(\d+)', response.url.strip())): 205 document = bs4.BeautifulSoup(body, 'html.parser') 206 207 decompose_selectors = ['tr.emptyrow', 'div[data-status="Active"]'] 208 for selector in decompose_selectors: 209 elements = document.select(selector) 210 for element in elements: 211 element.decompose() 212 213 a_tags = document.select('a') 214 for a_tag in a_tags: 215 # Remove extra attributes by keeping only select attributes and replacing the existing attribute dict. 216 a_tag.attrs = {attr: a_tag.attrs[attr] for attr in ['data-column'] if (attr in a_tag.attrs)} 217 218 spans = document.select('tbody tr td span') 219 for span in spans: 220 # Remove all attributes. 221 span.attrs.clear() 222 223 body = str(document.select('table#participants')) 224 225 # Remove Chunking 226 response.headers.pop('transfer-encoding', None) 227 228 return body 229 230def finalize_moodle_exchange(exchange: edq.net.exchange.HTTPExchange) -> edq.net.exchange.HTTPExchange: 231 """ Finalize Moodle exchanges. """ 232 233 for param in MOODLE_FINALIZE_REMOVE_PARAMS: 234 exchange.parameters.pop(param, None) 235 236 return exchange 237 238def _clean_base_response(response: requests.Response, body: str, 239 keep_headers: typing.Union[typing.List[str], None] = None) -> str: 240 """ 241 Do response cleaning that is common amongst all backend types. 242 This function will: 243 - Remove X- headers. 244 """ 245 246 # Index requests are generally for identification, and we use headers. 247 path = urllib.parse.urlparse(response.request.url).path.strip() 248 if (path in ['', '/']): 249 body = '' 250 251 for key in list(response.headers.keys()): 252 key = key.strip().lower() 253 if ((keep_headers is not None) and (key in keep_headers)): 254 continue 255 256 if (key.startswith('x-')): 257 response.headers.pop(key, None) 258 259 return body 260 261def _recursive_remove_keys(data: typing.Any, remove_keys: typing.Set[str]) -> None: 262 """ 263 Recursively descend through the given and remove any instance to the given key from any dictionaries. 264 The data should only be simple types (POD, dicts, lists, tuples). 265 """ 266 267 if (isinstance(data, (list, tuple))): 268 for item in data: 269 _recursive_remove_keys(item, remove_keys) 270 elif (isinstance(data, dict)): 271 for key in list(data.keys()): 272 if (key in remove_keys): 273 del data[key] 274 else: 275 _recursive_remove_keys(data[key], remove_keys) 276 277def parse_cookies( 278 text_cookies: typing.Union[str, None], 279 strip_key_prefix: bool = True, 280 ) -> typing.Dict[str, typing.Any]: 281 """ Parse cookies out of a text string. """ 282 283 cookies: typing.Dict[str, typing.Any] = {} 284 285 if (text_cookies is None): 286 return cookies 287 288 text_cookies = text_cookies.strip() 289 if (len(text_cookies) == 0): 290 return cookies 291 292 for cookie in text_cookies.split('; '): 293 parts = cookie.split('=', maxsplit = 1) 294 295 key = parts[0].lower() 296 297 if (strip_key_prefix): 298 key = key.split(', ')[-1] 299 300 if (len(parts) == 1): 301 cookies[key] = True 302 else: 303 cookies[key] = parts[1] 304 305 return cookies
CANVAS_CLEAN_REMOVE_CONTENT_KEYS: List[str] =
['created_at', 'ics', 'last_activity_at', 'lti_context_id', 'preview_url', 'secure_params', 'total_activity_time', 'updated_at', 'url', 'uuid']
Keys to remove from Canvas content.
BLACKBOARD_CLEAN_REMOVE_CONTENT_KEYS: List[str] =
['created', 'modified']
Keys to remove from Blackboard content.
BLACKBOARD_CLEAN_REMOVE_HEADERS: Set[str] =
{'content-encoding', 'content-language', 'vary', 'expires', 'x-blackboard-xsrf', 'strict-transport-security', 'pragma', 'transfer-encoding', 'last-modified', 'access-control-allow-origin', 'p3p'}
Keys to remove from Blackboard headers.
MOODLE_CLEAN_REMOVE_HEADERS: Set[str] =
{'content-encoding', 'content-language', 'vary', 'expires', 'keep-alive', 'content-script-type', 'content-style-type', 'pragma', 'last-modified', 'accept-ranges'}
Keys to remove from Moodle headers.
MOODLE_FINALIZE_REMOVE_PARAMS: Set[str] =
{'logintoken'}
Keys to remove from Moodle headers.
STANDARDIZED_TIMESTAMP: str =
'123456789'
STANDARDIZED_SESSION_KEY: str =
'abcABC123'
STANDARDIZED_RANDOM_STRING: str =
'abc123'
def
clean_lms_response(response: requests.models.Response, body: str) -> str:
75def clean_lms_response(response: requests.Response, body: str) -> str: 76 """ 77 A ResponseModifierFunction that attempt to identify 78 if the requests comes from a Learning Management System (LMS), 79 and clean the response accordingly. 80 """ 81 82 # Check the standard LMS Toolkit backend header. 83 backend_type = response.headers.get(lms.model.constants.HEADER_KEY_BACKEND, '').lower() 84 85 if (backend_type == lms.model.constants.BACKEND_TYPE_CANVAS): 86 return clean_canvas_response(response, body) 87 88 if (backend_type == lms.model.constants.BACKEND_TYPE_MOODLE): 89 return clean_moodle_response(response, body) 90 91 # Try looking inside the header keys. 92 for key in response.headers: 93 key = key.lower().strip() 94 95 if ('blackboard' in key): 96 return clean_blackboard_response(response, body) 97 98 if ('canvas' in key): 99 return clean_canvas_response(response, body) 100 101 if ('moodle' in key): 102 return clean_moodle_response(response, body) 103 104 return body
A ResponseModifierFunction that attempt to identify if the requests comes from a Learning Management System (LMS), and clean the response accordingly.
def
clean_blackboard_response(response: requests.models.Response, body: str) -> str:
106def clean_blackboard_response(response: requests.Response, body: str) -> str: 107 """ 108 See clean_lms_response(), but specifically for the Blackboard LMS. 109 This function will: 110 - Call _clean_base_response(). 111 - Remove specific headers. 112 """ 113 114 body = _clean_base_response(response, body) 115 116 # Work on both request and response headers. 117 for headers in [response.headers, response.request.headers]: 118 for key in list(headers.keys()): # type: ignore[attr-defined] 119 if (key.strip().lower() in BLACKBOARD_CLEAN_REMOVE_HEADERS): 120 headers.pop(key, None) # type: ignore[attr-defined] 121 122 # Most blackboard responses are JSON. 123 try: 124 data = edq.util.json.loads(body, strict = True) 125 except Exception: 126 # Response is not JSON. 127 return body 128 129 # Remove any content keys. 130 _recursive_remove_keys(data, set(BLACKBOARD_CLEAN_REMOVE_CONTENT_KEYS)) 131 132 # Convert body back to a string. 133 body = edq.util.json.dumps(data) 134 135 return body
See clean_lms_response(), but specifically for the Blackboard LMS. This function will:
- Call _clean_base_response().
- Remove specific headers.
def
clean_canvas_response(response: requests.models.Response, body: str) -> str:
137def clean_canvas_response(response: requests.Response, body: str) -> str: 138 """ 139 See clean_lms_response(), but specifically for the Canvas LMS. 140 This function will: 141 - Call _clean_base_response(). 142 - Remove content keys: [last_activity_at, total_activity_time] 143 """ 144 145 body = _clean_base_response(response, body) 146 147 # Most canvas responses are JSON. 148 try: 149 data = edq.util.json.loads(body, strict = True) 150 except Exception: 151 # Response is not JSON. 152 return body 153 154 # Remove any content keys. 155 _recursive_remove_keys(data, set(CANVAS_CLEAN_REMOVE_CONTENT_KEYS)) 156 157 # Remove special fields. 158 159 if ('submissions/update_grades' in str(response.request.url)): 160 data.pop('id', None) 161 162 # Convert body back to a string. 163 body = edq.util.json.dumps(data) 164 165 return body
See clean_lms_response(), but specifically for the Canvas LMS. This function will:
- Call _clean_base_response().
- Remove content keys: [last_activity_at, total_activity_time]
def
clean_moodle_response(response: requests.models.Response, body: str) -> str:
167def clean_moodle_response(response: requests.Response, body: str) -> str: 168 """ 169 See clean_lms_response(), but specifically for the Moodle LMS. 170 This function will: 171 - Call _clean_base_response(). 172 """ 173 174 body = _clean_base_response(response, body) 175 176 # Standardize timestamp. 177 current_timestamp_match = re.search(r"boost/theme/(\d{10})/favicon", body) 178 if (current_timestamp_match is not None): 179 body = body.replace(current_timestamp_match.group(1), STANDARDIZED_TIMESTAMP) 180 181 # Standardize session key. 182 session_key_match = re.search(r'"sesskey":"([^"]+)"', body) 183 if (session_key_match is not None): 184 body = body.replace(session_key_match.group(1), STANDARDIZED_SESSION_KEY) 185 186 # Standardize "random" string. 187 random_string_match = re.search(r"'random([a-z0-9]+)'", body) 188 if (random_string_match is not None): 189 body = body.replace(random_string_match.group(1), STANDARDIZED_RANDOM_STRING) 190 191 # Standardize logintoken. 192 logintoken_match = re.search(r'name="logintoken" value="(\w+)"', body) 193 if (logintoken_match is not None): 194 body = body.replace(logintoken_match.group(1), STANDARDIZED_SESSION_KEY) 195 196 # Work on both request and response headers. 197 for headers in [response.headers, response.request.headers]: 198 for key in list(headers.keys()): # type: ignore[attr-defined] 199 if (key.strip().lower() in MOODLE_CLEAN_REMOVE_HEADERS): 200 headers.pop(key, None) # type: ignore[attr-defined] 201 202 # Endpoint-Specific Tasks 203 204 # Remove extra data from the course participants response. 205 if (re.search(r'/user/index\.php\?id=(\d+)', response.url.strip())): 206 document = bs4.BeautifulSoup(body, 'html.parser') 207 208 decompose_selectors = ['tr.emptyrow', 'div[data-status="Active"]'] 209 for selector in decompose_selectors: 210 elements = document.select(selector) 211 for element in elements: 212 element.decompose() 213 214 a_tags = document.select('a') 215 for a_tag in a_tags: 216 # Remove extra attributes by keeping only select attributes and replacing the existing attribute dict. 217 a_tag.attrs = {attr: a_tag.attrs[attr] for attr in ['data-column'] if (attr in a_tag.attrs)} 218 219 spans = document.select('tbody tr td span') 220 for span in spans: 221 # Remove all attributes. 222 span.attrs.clear() 223 224 body = str(document.select('table#participants')) 225 226 # Remove Chunking 227 response.headers.pop('transfer-encoding', None) 228 229 return body
See clean_lms_response(), but specifically for the Moodle LMS. This function will:
- Call _clean_base_response().
def
finalize_moodle_exchange(exchange: edq.net.exchange.HTTPExchange) -> edq.net.exchange.HTTPExchange:
231def finalize_moodle_exchange(exchange: edq.net.exchange.HTTPExchange) -> edq.net.exchange.HTTPExchange: 232 """ Finalize Moodle exchanges. """ 233 234 for param in MOODLE_FINALIZE_REMOVE_PARAMS: 235 exchange.parameters.pop(param, None) 236 237 return exchange
Finalize Moodle exchanges.