Add first set of DBSC web platform tests

thefrog-gh · chromium-wpt-export-bot · commit 3f8c984dcb82 · 2025-02-28T14:23:16.000-08:00
The two test files added are: - create-session.https.html - not-secure-connection.html ".py" files are either server endpoints or helper files used by server endpoints. __init__.py is required to use `import importlib`, which is required to import modules under directories with a hyphen (like device-bound- session-credentials). Bug: 353767385 Change-Id: I3178c0d0d3f0b08ca1f77c01cdd40c4c9028f3f0 Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/6309272 Reviewed-by: Weizhong Xia <weizhong@google.com> Commit-Queue: thefrog <thefrog@chromium.org> Reviewed-by: Daniel Rubery <drubery@chromium.org> Cr-Commit-Position: refs/heads/main@{#1426552}
diff --git a/device-bound-session-credentials/__init__.py b/device-bound-session-credentials/__init__.py
diff --git a/device-bound-session-credentials/clear_server_state_and_end_sessions.py b/device-bound-session-credentials/clear_server_state_and_end_sessions.py
@@ -0,0 +1,6 @@
+import importlib
+session_provider = importlib.import_module('device-bound-session-credentials.session_provider')
+
+def main(request, response):
+    session_provider.clear_server_state()
+    return (200, [("Clear-Site-Data", '"cookies"')], "")
diff --git a/device-bound-session-credentials/create-session.https.html b/device-bound-session-credentials/create-session.https.html
@@ -0,0 +1,33 @@
+<!DOCTYPE html>
+<meta charset="utf-8">
+<title>DBSC session-creating tests</title>
+<script src="/resources/testharness.js"></script>
+<script src="/resources/testharnessreport.js"></script>
+<script src="/common/get-host-info.sub.js"></script>
+<script src="helper.js" type="module"></script>
+
+<script type="module">
+  import { expireCookie, documentHasCookie, waitForCookie, addCookieAndServerCleanup } from "./helper.js";
+
+  promise_test(async t => {
+    const expectedCookieAndValue = "auth_cookie=abcdef0123";
+    const expectedCookieAndAttributes = `${expectedCookieAndValue};Domain=${get_host_info().ORIGINAL_HOST};Path=/device-bound-session-credentials`;
+    addCookieAndServerCleanup(t, expectedCookieAndAttributes);
+
+    // Prompt starting a session, and wait until registration completes.
+    const login_response = await fetch('login.py');
+    assert_equals(login_response.status, 200);
+    assert_true(await waitForCookie(expectedCookieAndValue));
+
+    // Confirm that a request has the cookie set.
+    const auth_response = await fetch('verify_authenticated.py');
+    assert_equals(auth_response.status, 200);
+
+    // Confirm that expiring the cookie still leads to a request with the cookie set (refresh occurs).
+    expireCookie(expectedCookieAndAttributes);
+    assert_false(documentHasCookie(expectedCookieAndValue));
+    const auth_response_after_expiry = await fetch('verify_authenticated.py');
+    assert_equals(auth_response_after_expiry.status, 200);
+    assert_true(documentHasCookie(expectedCookieAndValue));
+  }, "An established session can refresh a cookie");
+</script>
diff --git a/device-bound-session-credentials/helper.js b/device-bound-session-credentials/helper.js
@@ -0,0 +1,33 @@
+export function documentHasCookie(cookieAndValue) {
+  return document.cookie.split(';').some(item => item.includes(cookieAndValue));
+}
+
+export function waitForCookie(cookieAndValue) {
+  const startTime = Date.now();
+  return new Promise(resolve => {
+    const interval = setInterval(() => {
+      if (documentHasCookie(cookieAndValue)) {
+        clearInterval(interval);
+        resolve(true);
+      }
+      if (Date.now() - startTime >= 1000) {
+        clearInterval(interval);
+        resolve(false);
+      }
+    }, 100);
+  });
+}
+
+export function expireCookie(cookieAndAttributes) {
+  document.cookie =
+      cookieAndAttributes + '; expires=Thu, 01 Jan 1970 00:00:00 UTC;';
+}
+
+export function addCookieAndServerCleanup(test, cookieAndAttributes) {
+  // Clean up any set cookies once the test completes.
+  test.add_cleanup(async () => {
+    const response = await fetch('clear_server_state_and_end_sessions.py');
+    assert_equals(response.status, 200);
+    expireCookie(cookieAndAttributes);
+  });
+}
diff --git a/device-bound-session-credentials/jwt_helper.py b/device-bound-session-credentials/jwt_helper.py
@@ -0,0 +1,76 @@
+import json
+import base64
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import rsa, padding
+
+# This method decodes the JWT and verifies the signature. If a key is provided,
+# that will be used for signature verification. Otherwise, the key sent within
+# the JWT payload will be used instead.
+# This returns a tuple of (decoded_header, decoded_payload, verify_succeeded).
+def decode_jwt(token, key=None):
+    try:
+        # Decode the header and payload.
+        header, payload, signature = token.split('.')
+        decoded_header = decode_base64_json(header)
+        decoded_payload = decode_base64_json(payload)
+
+        # If decoding failed, return nothing.
+        if not decoded_header or not decoded_payload:
+            return None, None, False
+
+        # If there is a key passed in (for refresh), use that for checking the signature below.
+        # Otherwise (for registration), use the key sent within the JWT to check the signature.
+        if key == None:
+            key = decoded_payload.get('key')
+        public_key = serialization.load_pem_public_key(jwk_to_pem(key))
+        # Verifying the signature will throw an exception if it fails.
+        verify_rs256_signature(header, payload, signature, public_key)
+        return decoded_header, decoded_payload, True
+    except Exception:
+        return None, None, False
+
+def jwk_to_pem(jwk_data):
+    jwk = json.loads(jwk_data) if isinstance(jwk_data, str) else jwk_data
+    key_type = jwk.get("kty")
+
+    if key_type != "RSA":
+        raise ValueError(f"Unsupported key type: {key_type}")
+
+    n = int.from_bytes(decode_base64url(jwk["n"]), 'big')
+    e = int.from_bytes(decode_base64url(jwk["e"]), 'big')
+    public_key = rsa.RSAPublicNumbers(e, n).public_key()
+    pem_public_key = public_key.public_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PublicFormat.SubjectPublicKeyInfo
+    )
+    return pem_public_key
+
+def verify_rs256_signature(encoded_header, encoded_payload, signature, public_key):
+    message = (encoded_header + '.' + encoded_payload).encode('utf-8')
+    signature_bytes = decode_base64(signature)
+    # This will throw an exception if verification fails.
+    public_key.verify(
+        signature_bytes,
+        message,
+        padding.PKCS1v15(),
+        hashes.SHA256()
+    )
+
+def add_base64_padding(encoded_data):
+    remainder = len(encoded_data) % 4
+    if remainder > 0:
+        encoded_data += '=' * (4 - remainder)
+    return encoded_data
+
+def decode_base64url(encoded_data):
+    encoded_data = add_base64_padding(encoded_data)
+    encoded_data = encoded_data.replace("-", "+").replace("_", "/")
+    return base64.b64decode(encoded_data)
+
+def decode_base64(encoded_data):
+    encoded_data = add_base64_padding(encoded_data)
+    return base64.urlsafe_b64decode(encoded_data)
+
+def decode_base64_json(encoded_data):
+    return json.loads(decode_base64(encoded_data))
diff --git a/device-bound-session-credentials/login.py b/device-bound-session-credentials/login.py
@@ -0,0 +1,3 @@
+def main(request, response):
+    headers = [('Sec-Session-Registration', '(RS256);challenge="login_challenge_value";path="/device-bound-session-credentials/start_session.py"')]
+    return (200, headers, "")
diff --git a/device-bound-session-credentials/not-secure-connection.html b/device-bound-session-credentials/not-secure-connection.html
@@ -0,0 +1,23 @@
+<!DOCTYPE html>
+<meta charset="utf-8">
+<title>No DBSC if connection is HTTP</title>
+<script src="/resources/testharness.js"></script>
+<script src="/resources/testharnessreport.js"></script>
+<script src="/common/get-host-info.sub.js"></script>
+<script src="helper.js" type="module"></script>
+
+<script type="module">
+  import { expireCookie, waitForCookie, addCookieAndServerCleanup } from "./helper.js";
+
+  promise_test(async t => {
+    const expectedCookieAndValue = "auth_cookie=abcdef0123";
+    const expectedCookieAndAttributes = `${expectedCookieAndValue};Domain=${get_host_info().ORIGINAL_HOST};Path=/device-bound-session-credentials`;
+    addCookieAndServerCleanup(t, expectedCookieAndAttributes);
+
+    // Prompt starting a session, and wait until registration completes.
+    const login_response = await fetch('login.py');
+    assert_equals(login_response.status, 200);
+    // For HTTP, this call will time out, because the cookie is never set.
+    assert_false(await waitForCookie(expectedCookieAndValue));
+  }, "Try to establish a session over HTTP");
+</script>
diff --git a/device-bound-session-credentials/refresh_session.py b/device-bound-session-credentials/refresh_session.py
@@ -0,0 +1,23 @@
+import importlib
+jwt_helper = importlib.import_module('device-bound-session-credentials.jwt_helper')
+session_provider = importlib.import_module('device-bound-session-credentials.session_provider')
+
+def main(request, response):
+    session_id_header = request.headers.get("Sec-Session-Id")
+    if session_id_header == None:
+        return (400, response.headers, "")
+    session_id = session_id_header.decode('utf-8')
+    session_key = session_provider.get_session_key(session_id)
+    if session_key == None:
+        return (400, response.headers, "")
+
+    challenge = "refresh_challenge_value"
+    if request.headers.get("Sec-Session-Response") == None:
+        return (401, [('Sec-Session-Challenge', '"' + challenge + '";id="' + session_id + '"')], "")
+
+    jwt_header, jwt_payload, verified = jwt_helper.decode_jwt(request.headers.get("Sec-Session-Response").decode('utf-8'), session_key)
+
+    if not verified or jwt_payload.get("jti") != challenge:
+        return (400, response.headers, "")
+
+    return session_provider.get_session_instructions_response(session_id, request)
diff --git a/device-bound-session-credentials/session_provider.py b/device-bound-session-credentials/session_provider.py
@@ -0,0 +1,46 @@
+import json
+
+session_to_key_map = {}
+
+def create_new_session():
+    session_id = str(len(session_to_key_map))
+    session_to_key_map[session_id] = None
+    return session_id
+
+def set_session_key(session_id, key):
+    if session_id not in session_to_key_map:
+        return False
+    session_to_key_map[session_id] = key
+    return True
+
+def get_session_key(session_id):
+    return session_to_key_map.get(session_id)
+
+def clear_server_state():
+    global session_to_key_map
+    session_to_key_map = {}
+
+def get_session_instructions_response(session_id, request):
+    refresh_url = "/device-bound-session-credentials/refresh_session.py"
+
+    response_body = {
+        "session_identifier": session_id,
+        "refresh_url": refresh_url,
+        "scope": {
+            "include_site": True,
+            "scope_specification" : [
+                { "type": "exclude", "domain": request.url_parts.hostname, "path": "/device-bound-session-credentials/clear_server_state_and_end_sessions.py" },
+            ]
+        },
+        "credentials": [{
+            "type": "cookie",
+            "name": "auth_cookie",
+            "attributes": "Domain=" + request.url_parts.hostname + "; Path=/device-bound-session-credentials"
+        }]
+    }
+    headers = [
+        ("Content-Type", "application/json"),
+        ("Cache-Control", "no-store"),
+        ("Set-Cookie", "auth_cookie=abcdef0123; Domain=" + request.url_parts.hostname + "; Path=/device-bound-session-credentials")
+    ]
+    return (200, headers, json.dumps(response_body))
diff --git a/device-bound-session-credentials/start_session.py b/device-bound-session-credentials/start_session.py
@@ -0,0 +1,13 @@
+import importlib
+jwt_helper = importlib.import_module('device-bound-session-credentials.jwt_helper')
+session_provider = importlib.import_module('device-bound-session-credentials.session_provider')
+
+def main(request, response):
+    jwt_header, jwt_payload, verified = jwt_helper.decode_jwt(request.headers.get("Sec-Session-Response").decode('utf-8'))
+    session_id = session_provider.create_new_session()
+    session_provider.set_session_key(session_id, jwt_payload.get('key'))
+
+    if not verified or jwt_payload.get("jti") != "login_challenge_value":
+        return (400, response.headers, "")
+
+    return session_provider.get_session_instructions_response(session_id, request)
diff --git a/device-bound-session-credentials/verify_authenticated.py b/device-bound-session-credentials/verify_authenticated.py
@@ -0,0 +1,5 @@
+def main(request, response):
+    cookie = request.cookies.get(b'auth_cookie')
+    if cookie == None or cookie.value != b'abcdef0123':
+        return (401, response.headers, "")
+    return (200, response.headers, "")

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+def main(request, response):`
	`2`	`+ headers = [('Sec-Session-Registration', '(RS256);challenge="login_challenge_value";path="/device-bound-session-credentials/start_session.py"')]`
	`3`	`+ return (200, headers, "")`