diff --git a/.well-known/attribution-reporting/report-event-attribution b/.well-known/attribution-reporting/report-event-attribution new file mode 100644 index 000000000000000..8ca40faaff7b727 --- /dev/null +++ b/.well-known/attribution-reporting/report-event-attribution @@ -0,0 +1,84 @@ +"""Endpoint to receive and return event-level attribution reports.""" +import json +from typing import List, Optional, Tuple, TypedDict + +from wptserve.request import Request +from wptserve.stash import Stash +from wptserve.utils import isomorphic_decode, isomorphic_encode + +# Key used to access the reports in the stash. Since stash is unique per URL +# path, we can use the same key for both event-level and aggregatable reports. +STASH_KEY = "4691a2d7fca5430fb0f33b1bd8a9d388" + +Header = Tuple[str, str] +Status = Tuple[int, str] +Response = Tuple[Status, List[Header], str] + +CLEAR_STASH = isomorphic_encode("clear_stash") + + +def handle_post_report(request: Request, headers: List[Header]) -> Response: + """Handles POST request for reports. + + Retrieves the report from the request body and stores the report in the + stash. If clear_stash is specified in the query params, clears the stash. + """ + if request.GET.get(CLEAR_STASH): + clear_stash(request.server.stash) + return (200, "OK"), headers, json.dumps({ + "code": 200, + "message": "Stash successfully cleared.", + }) + store_report(request.server.stash, request.body.decode("utf-8")) + return (201, "OK"), headers, json.dumps({ + "code": 201, + "message": "Report successfully stored." + }) + + +def handle_get_reports(request: Request, headers: List[Header]) -> Response: + """Handles GET request for reports. + + Retrieves and returns all reports from the stash. + """ + reports = take_reports(request.server.stash) + return (200, "OK"), headers, json.dumps({ + "code": 200, + "reports": reports, + }) + + +def store_report(stash: Stash, report: str) -> None: + """Stores the report in the stash. Report here is a JSON.""" + reports = stash.take(STASH_KEY) + if not reports: + reports = [] + reports.append(report) + stash.put(STASH_KEY, reports) + return None + + +def take_reports(stash: Stash) -> List[str]: + """Takes all the reports from the stash and returns them.""" + reports = stash.take(STASH_KEY) + if not reports: + reports = [] + return reports + + +def clear_stash(stash: Stash) -> None: + "Clears the stash." + stash.take(STASH_KEY) + return None + + +def main(request, response): + headers = [("Content-Type", "application/json")] + if request.method == "POST": + return handle_post_report(request, headers) + if request.method == "GET": + return handle_get_reports(request, headers) + return (405, "Method Not Allowed"), headers, json.dumps({ + "code": 405, + "message": "Only GET or POST methods are supported." + })