1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
import json
from urllib.parse import unquote_plus
# Script to generate trusted bidding signals. The responses depends on the
# keys and interestGroupNames - some result in entire response failures, others
# affect only their own value. Keys are preferentially used over
# interestGroupName, since keys are composible, but some tests need to cover
# there being no keys.
def main(request, response):
hostname = None
keys = None
interestGroupNames = None
# Manually parse query params. Can't use request.GET because it unescapes as well as splitting,
# and commas mean very different things from escaped commas.
for param in request.url_parts.query.split("&"):
pair = param.split("=", 1)
if len(pair) != 2:
return fail(response, "Bad query parameter: " + param)
# Browsers should escape query params consistently.
if "%20" in pair[1]:
return fail(response, "Query parameter should escape using '+': " + param)
# Hostname can't be empty. The empty string can be a key or interest group name, though.
if pair[0] == "hostname" and hostname == None and len(pair[1]) > 0:
hostname = pair[1]
continue
if pair[0] == "keys" and keys == None:
keys = list(map(unquote_plus, pair[1].split(",")))
continue
if pair[0] == "interestGroupNames" and interestGroupNames == None:
interestGroupNames = list(map(unquote_plus, pair[1].split(",")))
continue
return fail(response, "Unexpected query parameter: " + param)
# "interestGroupNames" and "hostname" are mandatory.
if not hostname:
return fail(response, "hostname missing")
if not interestGroupNames:
return fail(response, "interestGroupNames missing")
response.status = (200, b"OK")
# The JSON representation of this is used as the response body. This does
# not currently include a "perInterestGroupData" object.
responseBody = {"keys": {}}
# Set when certain special keys are observed, used in place of the JSON
# representation of `responseBody`, when set.
body = None
contentType = "application/json"
xAllowFledge = "true"
dataVersion = None
if keys:
for key in keys:
value = "default value"
if key == "close-connection":
# Close connection without writing anything, to simulate a
# network error. The write call is needed to avoid writing the
# default headers.
response.writer.write("")
response.close_connection = True
return
elif key.startswith("replace-body:"):
# Replace entire response body. Continue to run through other
# keys, to allow them to modify request headers.
body = key.split(':', 1)[1]
elif key.startswith("data-version:"):
dataVersion = key.split(':', 1)[1]
elif key == "http-error":
response.status = (404, b"Not found")
elif key == "no-content-type":
contentType = None
elif key == "wrong-content-type":
contentType = 'text/plain'
elif key == "wrongContentType":
contentType = 'text/plain'
elif key == "bad-allow-fledge":
xAllowFledge = "sometimes"
elif key == "fledge-not-allowed":
xAllowFledge = "false"
elif key == "no-allow-fledge":
xAllowFledge = None
elif key == "no-value":
continue
elif key == "wrong-value":
responseBody["keys"]["another-value"] = "another-value"
continue
elif key == "null-value":
value = None
elif key == "num-value":
value = 1
elif key == "string-value":
value = "1"
elif key == "array-value":
value = [1, "foo", None]
elif key == "object-value":
value = {"a":"b", "c":["d"]}
elif key == "interest-group-names":
value = json.dumps(interestGroupNames)
elif key == "hostname":
value = request.GET.first(b"hostname", b"not-found").decode("ASCII")
responseBody["keys"][key] = value
if "data-version" in interestGroupNames:
dataVersion = "4"
if contentType:
response.headers.set("Content-Type", contentType)
if xAllowFledge:
response.headers.set("X-Allow-FLEDGE", xAllowFledge)
if dataVersion:
response.headers.set("Data-Version", dataVersion)
response.headers.set("X-fledge-bidding-signals-format-version", "2")
if body != None:
return body
return json.dumps(responseBody)
def fail(response, body):
response.status = (400, "Bad Request")
response.headers.set(b"Content-Type", b"text/plain")
return body
|
