reports.py (2898B)
1 """Methods for the report-shared-storage and report-protected-audience endpoints (including debug endpoints)""" 2 import json 3 from typing import List, Optional, Tuple, Union 4 import urllib.parse 5 6 from wptserve.request import Request 7 from wptserve.stash import Stash 8 from wptserve.utils import isomorphic_decode, isomorphic_encode 9 10 # Arbitrary key used to access the reports in the stash. 11 REPORTS_KEY = "9d285691-4386-45ad-9a79-d2ec29557bfe" 12 13 CLEAR_STASH_AS_BYTES = isomorphic_encode("clear_stash") 14 15 Header = Tuple[str, str] 16 Status = Union[int, Tuple[int, str]] 17 Response = Tuple[Status, List[Header], str] 18 19 def get_request_origin(request: Request) -> str: 20 return "%s://%s" % (request.url_parts.scheme, 21 request.url_parts.netloc) 22 23 def handle_post_request(request: Request) -> Response: 24 """Handles POST request for reports. 25 26 Retrieves the report from the request body and stores the report in the 27 stash. If clear_stash is specified in the query params, clears the stash. 28 """ 29 if request.GET.get(CLEAR_STASH_AS_BYTES): 30 clear_stash(request.server.stash) 31 return 200, [], "Stash successfully cleared." 32 33 store_report(request.server.stash, get_request_origin(request), 34 request.body.decode("utf-8")) 35 return 200, [], "" 36 37 38 def handle_get_request(request: Request) -> Response: 39 """Handles GET request for reports. 40 41 Retrieves and returns all reports from the stash. 42 """ 43 headers = [("Content-Type", "application/json")] 44 reports = take_reports(request.server.stash, get_request_origin(request)) 45 headers.append(("Access-Control-Allow-Origin", "*")) 46 return 200, headers, json.dumps(reports) 47 48 49 def store_report(stash: Stash, origin: str, report: str) -> None: 50 """Stores the report in the stash. Report here is a JSON.""" 51 with stash.lock: 52 reports_dict = stash.take(REPORTS_KEY) 53 if not reports_dict: 54 reports_dict = {} 55 reports = reports_dict.get(origin, []) 56 reports.append(report) 57 reports_dict[origin] = reports 58 stash.put(REPORTS_KEY, reports_dict) 59 return None 60 61 def clear_stash(stash: Stash) -> None: 62 "Clears the stash." 63 stash.take(REPORTS_KEY) 64 return None 65 66 def take_reports(stash: Stash, origin: str) -> List[str]: 67 """Takes all the reports from the stash and returns them.""" 68 with stash.lock: 69 reports_dict = stash.take(REPORTS_KEY) 70 if not reports_dict: 71 reports_dict = {} 72 73 reports = reports_dict.pop(origin, []) 74 stash.put(REPORTS_KEY, reports_dict) 75 return reports 76 77 78 def handle_request(request: Request) -> Response: 79 """Handles request to get or store reports.""" 80 if request.method == "POST": 81 return handle_post_request(request) 82 if request.method == "GET": 83 return handle_get_request(request) 84 85 return (405, "Method Not Allowed"), [("Content-Type", "application/json")], json.dumps({ 86 "code": 405, 87 "message": "Only GET or POST methods are supported." 88 })