reporting_origin.py (1793B)
1 """Test reporting origin server used for two reasons: 2 3 1. It is a workaround for lack of preflight support in the test server. 4 2. Stashes requests so they can be inspected by tests. 5 """ 6 7 from wptserve.stash import Stash 8 import json 9 10 REQUESTS = "9250f93f-2c05-4aae-83b9-2817b0e18b4d" 11 12 13 headers = [ 14 b"attribution-reporting-eligible", 15 b"attribution-reporting-support", 16 b"referer", 17 ] 18 19 20 def store_request(request) -> None: 21 obj = { 22 "method": request.method, 23 "url": request.url, 24 } 25 for header in headers: 26 value = request.headers.get(header) 27 if value is not None: 28 obj[str(header, "utf-8")] = str(value, "utf-8") 29 with request.server.stash.lock: 30 requests = request.server.stash.take(REQUESTS) 31 if not requests: 32 requests = [] 33 requests.append(obj) 34 request.server.stash.put(REQUESTS, requests) 35 return None 36 37 38 def get_requests(request) -> str: 39 with request.server.stash.lock: 40 return json.dumps(request.server.stash.take(REQUESTS)) 41 42 43 def main(request, response): 44 """ 45 For most requests, simply returns a 200. Actual source/trigger registration 46 headers are piped using the `pipe` query param. 47 48 If a `clear-stash` param is set, it will clear the stash. 49 """ 50 if request.GET.get(b"clear-stash"): 51 request.server.stash.take(REQUESTS) 52 return 53 54 # We dont want to redirect preflight requests. The cors headers are piped 55 # so we can simply return a 200 and redirect the following request 56 if request.method == "OPTIONS": 57 response.status = 200 58 return 59 60 if request.GET.get(b"get-requests"): 61 return get_requests(request) 62 63 if request.GET.get(b"store-request"): 64 store_request(request) 65 return ""