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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
|
import json
test_to_session_manager_mapping = {}
def initialize_test():
test_id = str(len(test_to_session_manager_mapping))
test_to_session_manager_mapping[test_id] = SessionManager()
return test_id
def find_for_request(request):
test_id = request.cookies.get(b'test_id').value.decode('utf-8')
manager = test_to_session_manager_mapping.get(test_id)
if manager == None:
raise Exception(f"Could not find manager for test_id: {test_id}")
return manager
class CookieDetail:
def __init__(self, name_and_value = None, attributes = None):
self.name_and_value = name_and_value
self.attributes = attributes
def get_name_and_value(self):
if self.name_and_value is None:
return "auth_cookie=abcdef0123"
return self.name_and_value
def get_attributes(self, request):
if self.attributes is None:
return f"Domain={request.url_parts.hostname}; Path=/device-bound-session-credentials"
return self.attributes
class SessionManager:
def __init__(self):
self.session_to_key_map = {}
self.should_refresh_end_session = False
self.authorization_value = None
self.scope_origin = None
self.registration_sends_challenge_before_instructions = False
self.registration_sends_challenge_with_instructions = False
self.cookie_details = None
self.session_to_cookie_details_map = {}
self.session_to_early_challenge_map = {}
self.has_called_refresh = False
self.scope_specification_items = []
self.refresh_sends_challenge = True
self.refresh_url = "/device-bound-session-credentials/refresh_session.py"
self.include_site = True
self.refresh_endpoint_unavailable = False
self.response_session_id_override = None
self.allowed_refresh_initiators = ["*"]
self.provider_session_id = None
self.provider_url = None
self.provider_key = None
self.use_empty_response = False
self.registration_extra_cookies = []
self.has_custom_query_param = False
def next_session_id(self):
return len(self.session_to_key_map)
def create_new_session(self):
session_id = self.next_session_id()
self.session_to_key_map[session_id] = None
return session_id
def set_session_key(self, session_id, key):
if session_id not in self.session_to_key_map:
return False
self.session_to_key_map[session_id] = key
return True
def get_session_key(self, session_id):
return self.session_to_key_map.get(session_id)
def get_session_ids(self):
return list(self.session_to_key_map.keys())
def configure_state_for_test(self, configuration):
should_refresh_end_session = configuration.get("shouldRefreshEndSession")
if should_refresh_end_session is not None:
self.should_refresh_end_session = should_refresh_end_session
authorization_value = configuration.get("authorizationValue")
if authorization_value is not None:
self.authorization_value = authorization_value
scope_origin = configuration.get("scopeOrigin")
if scope_origin is not None:
self.scope_origin = scope_origin
registration_sends_challenge_before_instructions = configuration.get("registrationSendsChallengeBeforeInstructions")
if registration_sends_challenge_before_instructions is not None:
self.registration_sends_challenge_before_instructions = registration_sends_challenge_before_instructions
registration_sends_challenge_with_instructions = configuration.get("registrationSendsChallengeWithInstructions")
if registration_sends_challenge_with_instructions is not None:
self.registration_sends_challenge_with_instructions = registration_sends_challenge_with_instructions
cookie_details = configuration.get("cookieDetails")
if cookie_details is not None:
self.cookie_details = []
for detail in cookie_details:
self.cookie_details.append(CookieDetail(detail.get("nameAndValue"), detail.get("attributes")))
next_sessions_cookie_details = configuration.get("cookieDetailsForNextRegisteredSessions")
if next_sessions_cookie_details is not None:
next_session_id = self.next_session_id()
for session in next_sessions_cookie_details:
self.session_to_cookie_details_map[next_session_id] = []
for detail in session:
self.session_to_cookie_details_map[next_session_id].append(CookieDetail(detail.get("nameAndValue"), detail.get("attributes")))
next_session_id += 1
next_session_early_challenge = configuration.get("earlyChallengeForNextRegisteredSession")
if next_session_early_challenge is not None:
self.session_to_early_challenge_map[self.next_session_id()] = next_session_early_challenge
scope_specification_items = configuration.get("scopeSpecificationItems")
if scope_specification_items is not None:
self.scope_specification_items = scope_specification_items
refresh_sends_challenge = configuration.get("refreshSendsChallenge")
if refresh_sends_challenge is not None:
self.refresh_sends_challenge = refresh_sends_challenge
refresh_url = configuration.get("refreshUrl")
if refresh_url is not None:
self.refresh_url = refresh_url
include_site = configuration.get("includeSite")
if include_site is not None:
self.include_site = include_site
refresh_endpoint_unavailable = configuration.get("refreshEndpointUnavailable")
if refresh_endpoint_unavailable is not None:
self.refresh_endpoint_unavailable = refresh_endpoint_unavailable
response_session_id_override = configuration.get("responseSessionIdOverride")
if response_session_id_override is not None:
self.response_session_id_override = response_session_id_override
allowed_refresh_initiators = configuration.get("allowedRefreshInitiators")
if allowed_refresh_initiators is not None:
self.allowed_refresh_initiators = allowed_refresh_initiators
provider_session_id = configuration.get("providerSessionId")
if provider_session_id is not None:
self.provider_session_id = provider_session_id
provider_url = configuration.get("providerUrl")
if provider_url is not None:
self.provider_url = provider_url
provider_key = configuration.get("providerKey")
if provider_key is not None:
self.provider_key = provider_key
use_empty_response = configuration.get("useEmptyResponse")
if use_empty_response is not None:
self.use_empty_response = use_empty_response
registration_extra_cookies = configuration.get("registrationExtraCookies")
if registration_extra_cookies is not None:
self.registration_extra_cookies = []
for detail in registration_extra_cookies:
self.registration_extra_cookies.append(CookieDetail(detail.get("nameAndValue"), detail.get("attributes")))
has_custom_query_param = configuration.get("hasCustomQueryParam")
if has_custom_query_param is not None:
self.has_custom_query_param = has_custom_query_param
def get_should_refresh_end_session(self):
return self.should_refresh_end_session
def get_authorization_value(self):
return self.authorization_value
def get_registration_sends_challenge_before_instructions(self):
return self.registration_sends_challenge_before_instructions
def reset_registration_sends_challenge_before_instructions(self):
self.registration_sends_challenge_before_instructions = False
def get_registration_sends_challenge_with_instructions(self):
return self.registration_sends_challenge_with_instructions
def reset_registration_sends_challenge_with_instructions(self):
self.registration_sends_challenge_with_instructions = False
def get_refresh_sends_challenge(self):
return self.refresh_sends_challenge
def set_has_called_refresh(self, has_called_refresh):
self.has_called_refresh = has_called_refresh
def get_has_custom_query_param(self):
return self.has_custom_query_param
def pull_server_state(self):
return {
"hasCalledRefresh": self.has_called_refresh
}
def get_cookie_details(self, session_id):
# Try to use the session-specific override first.
if self.session_to_cookie_details_map.get(session_id) is not None:
return self.session_to_cookie_details_map[session_id]
# If there isn't any, use the general override.
if self.cookie_details is not None:
return self.cookie_details
return [CookieDetail()]
def get_early_challenge(self, session_id):
return self.session_to_early_challenge_map.get(session_id)
def get_refresh_url(self):
if not self.has_custom_query_param:
return self.refresh_url
return self.refresh_url + "?refreshQueryParam=456"
def get_sessions_instructions_response_credentials(self, session_id, request):
return list(map(lambda cookie_detail: {
"type": "cookie",
"name": cookie_detail.get_name_and_value().split("=")[0],
"attributes": cookie_detail.get_attributes(request)
}, self.get_cookie_details(session_id)))
def get_set_cookie_headers(self, cookies, request):
header_values = list(map(
lambda cookie_detail: f"{cookie_detail.get_name_and_value()}; {cookie_detail.get_attributes(request)}",
cookies
))
return [("Set-Cookie", header_value) for header_value in header_values]
def get_session_instructions_response(self, session_id, request):
response_session_id = session_id
if self.response_session_id_override is not None:
response_session_id = self.response_session_id_override
scope_origin = ""
if self.scope_origin is not None:
scope_origin = self.scope_origin
response_body = {
"session_identifier": str(response_session_id),
"refresh_url": self.get_refresh_url(),
"scope": {
"origin": scope_origin,
"include_site": self.include_site,
"scope_specification" : self.scope_specification_items + [
{ "type": "exclude", "domain": request.url_parts.hostname, "path": "/device-bound-session-credentials/request_early_challenge.py" },
{ "type": "exclude", "domain": request.url_parts.hostname, "path": "/device-bound-session-credentials/end_session_via_clear_site_data.py" },
{ "type": "exclude", "domain": request.url_parts.hostname, "path": "/device-bound-session-credentials/pull_server_state.py" },
{ "type": "exclude", "domain": request.url_parts.hostname, "path": "/device-bound-session-credentials/set_cookie.py" },
]
},
"credentials": self.get_sessions_instructions_response_credentials(session_id, request),
"allowed_refresh_initiators": self.allowed_refresh_initiators,
}
headers = self.get_set_cookie_headers(self.get_cookie_details(session_id), request) + [
("Content-Type", "application/json"),
("Cache-Control", "no-store")
]
response_body = "" if self.use_empty_response else json.dumps(response_body)
return (200, headers, response_body)
def get_refresh_endpoint_unavailable(self):
return self.refresh_endpoint_unavailable
def get_provider_session_id(self):
return self.provider_session_id
def get_provider_url(self):
return self.provider_url
def get_provider_key(self):
return self.provider_key
|
