lms.util.net

  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.hash
 12import edq.util.json
 13import requests
 14
 15import lms.model.constants
 16
 17CANVAS_CLEAN_REMOVE_CONTENT_KEYS: typing.List[str] = [
 18    'created_at',
 19    'ics',
 20    'last_activity_at',
 21    'lti_context_id',
 22    'preview_url',
 23    'secure_params',
 24    'total_activity_time',
 25    'updated_at',
 26    'url',
 27    'uuid',
 28]
 29""" Keys to remove from Canvas content. """
 30
 31BLACKBOARD_CLEAN_REMOVE_CONTENT_KEYS: typing.List[str] = [
 32    'created',
 33    'modified',
 34]
 35""" Keys to remove from Blackboard content. """
 36
 37BLACKBOARD_CLEAN_REMOVE_HEADERS: typing.Set[str] = {
 38    'access-control-allow-origin',
 39    'content-encoding',
 40    'content-language',
 41    'expires',
 42    'last-modified',
 43    'p3p',
 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    'vary',
 61}
 62""" Keys to remove from Moodle headers. """
 63
 64MOODLE_FINALIZE_REMOVE_PARAMS: typing.Set[str] = {
 65    'logintoken',
 66}
 67""" Keys to remove from Moodle headers. """
 68
 69MOODLE_HTML_CLEAN: typing.Dict[str, typing.Dict[str, typing.Any]] = {
 70    r'/login/index\.php': {
 71        'delete_elements': ['script', 'footer'],
 72        'remove_all_attrs': ['body', 'div'],
 73        'hoist_elements': {
 74            'div#page-wrapper': 'input[name="logintoken"]',
 75        },
 76    },
 77    r'/user/index\.php\?id=(\d+)': {
 78        'delete_elements': [
 79            'tr.emptyrow',
 80            'div[data-status="Active"]',
 81            'i',
 82            'th.a',
 83        ],
 84        'attrs_to_keep': {
 85            'a': ['data-column'],
 86            'th': ['class'],
 87            'td': ['class'],
 88        },
 89        'remove_all_attrs': ['tr'],
 90        'final_selector': 'table#participants',
 91    },
 92    r'/user/profile.php': {
 93        'filter_elements_by_descendant': {
 94            'div.card-body': ('h3', 'Course details'),
 95        },
 96        'hoist_elements': {
 97            'div.card-body ul': 'div.card-body ul li ul',
 98        },
 99        'final_selector': 'div.card-body',
100    },
101}
102""" A mapping of Moodle URL patterns to clean_html() kwargs. """
103
104STANDARDIZED_TIMESTAMP: str = '123456789'
105STANDARDIZED_SESSION_KEY: str = 'abcABC123'
106STANDARDIZED_RANDOM_STRING: str = 'abc123'
107
108def clean_lms_response(response: requests.Response, body: str) -> str:
109    """
110    A ResponseModifierFunction that attempt to identify
111    if the requests comes from a Learning Management System (LMS),
112    and clean the response accordingly.
113    """
114
115    # Check the standard LMS Toolkit backend header.
116    raw_backend_type = response.headers.get(lms.model.constants.HEADER_KEY_BACKEND, '').lower()
117
118    if (raw_backend_type == lms.model.constants.BackendType.CANVAS.value):
119        return clean_canvas_response(response, body)
120
121    if (raw_backend_type == lms.model.constants.BackendType.MOODLE.value):
122        return clean_moodle_response(response, body)
123
124    # Try looking inside the header keys.
125    for key in response.headers:
126        key = key.lower().strip()
127
128        if ('blackboard' in key):
129            return clean_blackboard_response(response, body)
130
131        if ('canvas' in key):
132            return clean_canvas_response(response, body)
133
134        if ('moodle' in key):
135            return clean_moodle_response(response, body)
136
137    return body
138
139def clean_blackboard_response(response: requests.Response, body: str) -> str:
140    """
141    See clean_lms_response(), but specifically for the Blackboard LMS.
142    This function will:
143     - Call _clean_base_response().
144     - Remove specific headers.
145    """
146
147    body = _clean_base_response(response, body)
148
149    # Work on both request and response headers.
150    for headers in [response.headers, response.request.headers]:
151        for key in list(headers.keys()):  # type: ignore[attr-defined]
152            if (key.strip().lower() in BLACKBOARD_CLEAN_REMOVE_HEADERS):
153                headers.pop(key, None)  # type: ignore[attr-defined]
154
155    # Most blackboard responses are JSON.
156    try:
157        data = edq.util.json.loads(body, strict = True)
158    except Exception:
159        # Response is not JSON.
160        return body
161
162    # Remove any content keys.
163    _recursive_remove_keys(data, set(BLACKBOARD_CLEAN_REMOVE_CONTENT_KEYS))
164
165    # Convert body back to a string.
166    body = edq.util.json.dumps(data)
167
168    return body
169
170def clean_canvas_response(response: requests.Response, body: str) -> str:
171    """
172    See clean_lms_response(), but specifically for the Canvas LMS.
173    This function will:
174     - Call _clean_base_response().
175     - Remove content keys: [last_activity_at, total_activity_time]
176    """
177
178    body = _clean_base_response(response, body)
179    url = str(response.request.url)
180
181    if ('/files_api' in url):
182        # Replace the scheme + netloc (host + port) with a slug that can be replaced when testing.
183        parts = urllib.parse.urlsplit(response.headers['location'])
184        parts = parts._replace(netloc = lms.model.constants.SERVER_SLUG, scheme = '')
185        response.headers['location'] = parts.geturl().replace(f"//{lms.model.constants.SERVER_SLUG}", lms.model.constants.SERVER_SLUG)
186
187    # Most canvas responses are JSON.
188    try:
189        data = edq.util.json.loads(body, strict = True)
190    except Exception:
191        # Response is not JSON.
192        return body
193
194    # Remove any content keys.
195    _recursive_remove_keys(data, set(CANVAS_CLEAN_REMOVE_CONTENT_KEYS))
196
197    # Handle endpoint-specific cases.
198    if ('submissions/update_grades' in url):
199        data.pop('id', None)
200    elif (re.search(r'api/v1/courses/\w+/files', url) is not None):
201        # Replace the scheme + netloc (host + port) with a slug that can be replaced when testing.
202        parts = urllib.parse.urlsplit(data['upload_url'])
203        parts = parts._replace(netloc = lms.model.constants.SERVER_SLUG, scheme = '')
204        data['upload_url'] = parts.geturl().replace(f"//{lms.model.constants.SERVER_SLUG}", lms.model.constants.SERVER_SLUG)
205
206    # Convert body back to a string.
207    body = edq.util.json.dumps(data)
208
209    return body
210
211def finalize_canvas_exchange(exchange: edq.net.exchange.HTTPExchange) -> edq.net.exchange.HTTPExchange:
212    """ Finalize Canvas exchanges. """
213
214    if (re.search(r'^api/v1/courses/\w+/files$', exchange.url_path) is not None):
215        # Clean out random data for the file upload.
216        data = edq.util.json.loads(str(exchange.response_body))
217
218        body_data = {
219            'upload_params': {
220                'Filename': data['upload_params']['Filename'],
221            },
222            'upload_url': data['upload_url'],
223        }
224
225        exchange.response_body = edq.util.json.dumps(body_data)
226    elif (exchange.url_path == 'files_api'):
227        # File upload calls contain several random pieces of information generated from the server.
228        # Drop information we don't need in testing and make the rest consistent.
229
230        filename = exchange.parameters['Filename']
231        exchange.parameters = {'filename': filename}
232        exchange.files = []
233
234        location = exchange.response_headers.get('location', None)
235        if (location is not None):
236            location = re.sub(r'uuid=.*$', f"uuid={edq.util.hash.sha256_hex(filename)}", str(location))
237
238        exchange.response_headers['location'] = location
239    elif (re.search(r'^api/v1/files/\w+/create_success$', exchange.url_path) is not None):
240        # Change the uuid parameter to match the one set in files_api (which redirects to this URL).
241        data = edq.util.json.loads(str(exchange.response_body))
242        exchange.parameters['uuid'] = edq.util.hash.sha256_hex(data['display_name'])
243        exchange.response_body = edq.util.json.dumps({'id': data['id']})
244
245    return exchange
246
247def clean_moodle_response(response: requests.Response, body: str) -> str:
248    """
249    See clean_lms_response(), but specifically for the Moodle LMS.
250    This function will:
251     - Call _clean_base_response().
252    """
253
254    body = _clean_base_response(response, body)
255
256    # Standardize timestamp.
257    current_timestamp_match = re.search(r"boost/theme/(\d{10})/favicon", body)
258    if (current_timestamp_match is not None):
259        body = body.replace(current_timestamp_match.group(1), STANDARDIZED_TIMESTAMP)
260
261    # Standardize session key.
262    session_key_match = re.search(r'"sesskey":"([^"]+)"', body)
263    if (session_key_match is not None):
264        body = body.replace(session_key_match.group(1), STANDARDIZED_SESSION_KEY)
265
266    # Standardize "random" string.
267    random_string_match = re.search(r"'random([a-z0-9]+)'", body)
268    if (random_string_match is not None):
269        body = body.replace(random_string_match.group(1), STANDARDIZED_RANDOM_STRING)
270
271    # Standardize logintoken.
272    logintoken_match = re.search(r'name="logintoken" value="(\w+)"', body)
273    if (logintoken_match is not None):
274        body = body.replace(logintoken_match.group(1), STANDARDIZED_SESSION_KEY)
275
276    # Standardize last access to course.
277    last_access_match = re.search(r'(\d+) secs', body)
278    if (last_access_match is not None):
279        body = body.replace(last_access_match.group(0), f'{STANDARDIZED_TIMESTAMP} secs')
280
281    # Work on both request and response headers.
282    for headers in [response.headers, response.request.headers]:
283        for key in list(headers.keys()):  # type: ignore[attr-defined]
284            if (key.strip().lower() in MOODLE_CLEAN_REMOVE_HEADERS):
285                headers.pop(key, None)  # type: ignore[attr-defined]
286
287    # Remove Chunking
288    response.headers.pop('transfer-encoding', None)
289
290    # Endpoint-Specific Tasks
291
292    # Clean HTML responses.
293    for (pattern, clean_params) in MOODLE_HTML_CLEAN.items():
294        if (re.search(pattern, response.url.strip())):
295            body = clean_html(body, **clean_params)
296
297    return body
298
299def clean_html(
300        html: str,
301        filter_elements_by_descendant: typing.Union[typing.Dict[str, typing.Tuple[str, str]], None] = None,
302        delete_elements: typing.Union[typing.List[str], None] = None,
303        attrs_to_keep: typing.Union[typing.Dict[str, typing.List[str]], None] = None,
304        classes_to_keep: typing.Union[typing.Dict[str, typing.List[str]], None] = None,
305        remove_all_attrs: typing.Union[typing.List[str], None] = None,
306        hoist_elements: typing.Union[typing.Dict[str, str], None] = None,
307        replacements: typing.Union[typing.List[typing.Tuple[str, str]], None] = None,
308        final_selector: str = 'body',
309        ) -> str:
310    """
311    General purpose HTML cleaning function.
312
313    filter_elements_by_descendant: { selector: (descendant_selector, text), ... }
314    Removes all elements matching a selector if the selected element's specific descendant does not exist or does not have the specified text.
315    For example:
316    ```
317    >>> body = '<div><h2>Hello World!</h2></div><div><h2>Hello Universe!</h2></div><div><p>No h2 Element.</p></div>'
318    >>> clean_html(body, filter_elements_by_descendant = {'div': ('h2', 'Hello World!')})
319    <div><h2>Hello World!</h2></div>
320    ```
321
322    delete_elements: [ selector, ... ]
323    Deletes any matching elements.
324
325    attrs_to_keep: { selector: [attribute, ...], ... }
326    Keeps only the listed attributes of elements matching the selector.
327
328    classes_to_keep: { selector: [class, ...], ... }
329    Keeps only listed classes of elements matching the selector.
330
331    remove_all_attrs: [ selector, ... ]
332    Removes all attributes of matching elements.
333
334    hoist_elements: { parent_selector: child_selector, ... }
335    For each pair of selectors, the parent element is replaced by the first matching child within the parent.
336    No replacement occurs if both matches are not found.
337
338    replacements: [ (pattern, replacement), ... ]
339    Performs a replacement for all regex matches.
340    """
341
342    if (filter_elements_by_descendant is None):
343        filter_elements_by_descendant = {}
344
345    if (delete_elements is None):
346        delete_elements = []
347
348    if (attrs_to_keep is None):
349        attrs_to_keep = {}
350
351    if (classes_to_keep is None):
352        classes_to_keep = {}
353
354    if (remove_all_attrs is None):
355        remove_all_attrs = []
356
357    if (hoist_elements is None):
358        hoist_elements = {}
359
360    if (replacements is None):
361        replacements = [(r'\n', '')]
362
363    document = bs4.BeautifulSoup(html, 'html.parser')
364
365    # Filter Elements by Descendant
366    for (selector, (descendant_selector, text)) in filter_elements_by_descendant.items():
367        for element in document.select(selector):
368            descendant = element.select_one(descendant_selector)
369            if ((descendant is None) or (descendant.get_text() != text)):
370                element.decompose()
371
372    # Delete Elements
373    for selector in delete_elements:
374        for element in document.select(selector):
375            element.decompose()
376
377    # Keep Attributes
378    for (element_selector, attrs) in attrs_to_keep.items():
379        for element in document.select(element_selector):
380            # Remove extra attributes by keeping only select attributes and replacing the existing attribute dict.
381            element.attrs = {attr: element.attrs[attr] for attr in attrs if (attr in element.attrs)}
382
383    # Keep Classes
384    for (element_selector, keep_classes) in classes_to_keep.items():
385        for element in document.select(element_selector):
386            classes = element.get('class')
387            if ((classes is None) or (len(classes) == 0)):
388                continue
389
390            # Keep only classes listed for this selector.
391            kept = [keep_class for keep_class in classes if (keep_class in keep_classes)]
392
393            element['class'] = kept  # type: ignore[assignment]
394
395    # Remove All Attributes
396    for selector in remove_all_attrs:
397        for element in document.select(selector):
398            element.attrs.clear()
399
400    # Element Hoisting
401    for (parent_selector, child_selector) in hoist_elements.items():
402        for parent in document.select(parent_selector):
403            child = parent.select_one(child_selector)  # type: ignore[assignment]
404            if (child is None):
405                continue
406
407            parent.replace_with(child.extract())
408
409    # Replacements
410    document_string = str(document.select(final_selector))
411    for (pattern, replacement) in replacements:
412        document_string = re.sub(pattern, replacement, document_string)
413
414    return document_string
415
416def finalize_moodle_exchange(exchange: edq.net.exchange.HTTPExchange) -> edq.net.exchange.HTTPExchange:
417    """ Finalize Moodle exchanges. """
418
419    for param in MOODLE_FINALIZE_REMOVE_PARAMS:
420        exchange.parameters.pop(param, None)
421
422    return exchange
423
424def _clean_base_response(response: requests.Response, body: str,
425        keep_headers: typing.Union[typing.List[str], None] = None) -> str:
426    """
427    Do response cleaning that is common amongst all backend types.
428    This function will:
429     - Remove X- headers.
430    """
431
432    # Index requests are generally for identification, and we use headers.
433    path = urllib.parse.urlparse(response.request.url).path.strip()
434    if (path in ['', '/']):
435        body = ''
436
437    for key in list(response.headers.keys()):
438        key = key.strip().lower()
439        if ((keep_headers is not None) and (key in keep_headers)):
440            continue
441
442        if (key.startswith('x-')):
443            response.headers.pop(key, None)
444
445    return body
446
447def _recursive_remove_keys(data: typing.Any, remove_keys: typing.Set[str]) -> None:
448    """
449    Recursively descend through the given and remove any instance to the given key from any dictionaries.
450    The data should only be simple types (POD, dicts, lists, tuples).
451    """
452
453    if (isinstance(data, (list, tuple))):
454        for item in data:
455            _recursive_remove_keys(item, remove_keys)
456    elif (isinstance(data, dict)):
457        for key in list(data.keys()):
458            if (key in remove_keys):
459                del data[key]
460            else:
461                _recursive_remove_keys(data[key], remove_keys)
462
463def parse_cookies(
464        text_cookies: typing.Union[str, None],
465        strip_key_prefix: bool = True,
466        ) -> typing.Dict[str, typing.Any]:
467    """ Parse cookies out of a text string. """
468
469    cookies: typing.Dict[str, typing.Any] = {}
470
471    if (text_cookies is None):
472        return cookies
473
474    text_cookies = text_cookies.strip()
475    if (len(text_cookies) == 0):
476        return cookies
477
478    for cookie in text_cookies.split('; '):
479        parts = cookie.split('=', maxsplit = 1)
480
481        key = parts[0].lower()
482
483        if (strip_key_prefix):
484            key = key.split(', ')[-1]
485
486        if (len(parts) == 1):
487            cookies[key] = True
488        else:
489            cookies[key] = parts[1]
490
491    return cookies