Add WPT for query params for registration + refresh endpoints

thefrog-gh · chromium-wpt-export-bot · commit 965e56dfd37b · 2025-10-03T10:04:49.000-07:00
This also fixes start_session.py to convert response.headers (type: ResponseHeaders) into a list before concatenating with another list. Change-Id: Ie759e2f24869ead60aa872e6989182ed98cc665c Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/7003076 Reviewed-by: Daniel Rubery <drubery@chromium.org> Commit-Queue: thefrog <thefrog@chromium.org> Cr-Commit-Position: refs/heads/main@{#1524818}
diff --git a/device-bound-session-credentials/refresh_session.py b/device-bound-session-credentials/refresh_session.py
@@ -1,5 +1,6 @@
 import importlib
 import json
+from urllib.parse import parse_qs
 jwt_helper = importlib.import_module('device-bound-session-credentials.jwt_helper')
 session_manager = importlib.import_module('device-bound-session-credentials.session_manager')
 
@@ -23,6 +24,9 @@ def main(request, response):
         }
         return (200, response.headers, json.dumps(response_body))
 
+    if test_session_manager.get_has_custom_query_param() and 'refreshQueryParam' not in parse_qs(request.url_parts.query):
+        return (400, response.headers, "")
+
     session_key = test_session_manager.get_session_key(session_id)
     if session_key == None:
         return (400, response.headers, "")
diff --git a/device-bound-session-credentials/requests-have-query-params.https.html b/device-bound-session-credentials/requests-have-query-params.https.html
@@ -0,0 +1,43 @@
+<!DOCTYPE html>
+<meta charset="utf-8">
+<script src="/resources/testharness.js"></script>
+<script src="/resources/testharnessreport.js"></script>
+<script src="helper.js" type="module"></script>
+
+<script type="module">
+  import { configureServer, expireCookie, documentHasCookie, waitForCookie, addCookieAndSessionCleanup, setupShardedServerState, postJson } from "./helper.js";
+
+  promise_test(async t => {
+    await setupShardedServerState();
+    const expectedCookieAndValue = "auth_cookie=abcdef0123";
+    const expectedCookieAndAttributes = `${expectedCookieAndValue};Domain=${location.hostname};Path=/device-bound-session-credentials`;
+    addCookieAndSessionCleanup(t);
+
+    // Configure server to:
+    // 1. Check that registration has the query param specified below.
+    // 2. Set a refresh endpoint query param in the session instructions and
+    //    verify that refresh has that query param.
+    await configureServer({ hasCustomQueryParam: true });
+
+    // Prompt starting a session, and wait until registration completes. Pass
+    // through the query param to registration.
+    const registrationUrl = `start_session.py?registrationQueryParam=123`;
+    const loginResponse = await postJson('login.py', { registrationUrl });
+    assert_equals(loginResponse.status, 200);
+    await waitForCookie(expectedCookieAndValue, /*expectCookie=*/true);
+
+    // Confirm that a request has the cookie set.
+    const authResponse = await fetch('verify_authenticated.py');
+    assert_equals(authResponse.status, 200);
+
+    // Trigger refresh by expiring the cookie.
+    expireCookie(expectedCookieAndAttributes);
+    assert_false(documentHasCookie(expectedCookieAndValue));
+    // The server refresh will fail if the refresh endpoint query param is not
+    // present during refresh.
+    const authResponseAfterExpiry = await fetch('verify_authenticated.py');
+    assert_equals(authResponseAfterExpiry.status, 200);
+    assert_true(documentHasCookie(expectedCookieAndValue));
+  }, "Registration and refresh endpoints can contain query params");
+</script>
+
diff --git a/device-bound-session-credentials/session_manager.py b/device-bound-session-credentials/session_manager.py
@@ -53,6 +53,7 @@ def __init__(self):
         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)
@@ -164,6 +165,10 @@ def configure_state_for_test(self, configuration):
             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
 
@@ -188,6 +193,9 @@ def get_refresh_sends_challenge(self):
     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
@@ -205,6 +213,11 @@ def get_cookie_details(self, session_id):
     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",
@@ -230,7 +243,7 @@ def get_session_instructions_response(self, session_id, request):
 
         response_body = {
             "session_identifier": str(response_session_id),
-            "refresh_url": self.refresh_url,
+            "refresh_url": self.get_refresh_url(),
             "scope": {
                 "origin": scope_origin,
                 "include_site": self.include_site,
diff --git a/device-bound-session-credentials/start_session.py b/device-bound-session-credentials/start_session.py
@@ -1,4 +1,5 @@
 import importlib
+from urllib.parse import parse_qs
 jwt_helper = importlib.import_module('device-bound-session-credentials.jwt_helper')
 session_manager = importlib.import_module('device-bound-session-credentials.session_manager')
 
@@ -15,13 +16,16 @@ def main(request, response):
     test_session_manager.set_session_key(session_id, jwt_header.get('jwk'))
 
     if not verified or jwt_payload.get("jti") != "login_challenge_value":
-        return (400, response.headers + extra_cookie_headers, "")
+        return (400, list(response.headers) + extra_cookie_headers, "")
 
     if jwt_payload.get("authorization") != test_session_manager.get_authorization_value():
-        return (400, response.headers + extra_cookie_headers, "")
+        return (400, list(response.headers) + extra_cookie_headers, "")
 
     if jwt_payload.get("sub") is not None:
-        return (400, response.headers + extra_cookie_headers, "")
+        return (400, list(response.headers) + extra_cookie_headers, "")
+
+    if test_session_manager.get_has_custom_query_param() and 'registrationQueryParam' not in parse_qs(request.url_parts.query):
+        return (400, list(response.headers) + extra_cookie_headers, "")
 
     (code, headers, body) = test_session_manager.get_session_instructions_response(session_id, request)
     headers += extra_cookie_headers
