WIP: WPT tests for security features (CSP, mixed content, fetch metadata) applied to speculation rules resource fetch

Change-Id: I3e5de39191241ebee592c8e9ea85e38f462e1d7c

Author: jeremyroman

common/security-features/resources/common.sub.js

@@ -826,6 +826,37 @@ function requestViaWebSocket(url) {
     });
 }
 
+/**
+ * Requests speculation rules via the header in a new window.
+ * @param {string} url The URL of the resource to request.
+ * @return {Promise} The promise for success/error events.
+ */
+async function requestViaSpeculationRulesHeader(url) {
+  console.error(url);
+  let windowURL = `/common/blank.html?pipe=header(Speculation-Rules,"${encodeURIComponent(url)}")`;
+  console.error(windowURL);
+  let w = window.open(windowURL);
+  try {
+    // This is similar to t.step_wait, but t is not available here.
+    let continuePolling = true;
+    let status = 'blocked';
+    const params = new URLSearchParams(new URL(url).search);
+    const pollURL = `/common/security-features/subresource/xhr.py?action=peek&path=${encodeURIComponent(params.get('path'))}&key=${encodeURIComponent(params.get('key'))}`;
+    step_timeout(() => continuePolling = false, 3000);
+    do {
+      // This sort of polling logic might be more generally useful to other
+      // kinds of request that don't necessarily always receive a "done" event
+      // of some kind. But at the moment that would be a more radical refactor.
+      await new Promise(resolve => step_timeout(resolve, 200));
+      let json = await fetch(pollURL).then(r => r.json());
+      status = json.status;
+    } while (status !== 'allowed' && continuePolling);
+    return status;
+  } finally {
+    w.close();
+  }
+}
+
 /**
   @typedef SubresourceType
   @type {string}
@@ -942,6 +973,11 @@ const subresourceMap = {
     path: "/stash_responder",
     invoker: requestViaWebSocket,
   },
+
+  "speculationrules": {
+    path: "/common/security-features/subresource/speculationrules.py",
+    invoker: requestViaSpeculationRulesHeader,
+  },
 };
 for (const workletType of ['animation', 'audio', 'layout', 'paint']) {
   subresourceMap[`worklet-${workletType}`] = {

common/security-features/subresource/speculationrules.py

@@ -0,0 +1,13 @@
+import importlib
+subresource = importlib.import_module("common.security-features.subresource.subresource")
+
+def generate_payload(server_data):
+    return subresource.get_template(u"speculationrules.json.template") % server_data
+
+def main(request, response):
+    import sys
+    print('got speculation rules request', file=sys.stderr)
+    subresource.respond(request,
+                        response,
+                        payload_generator = generate_payload,
+                        content_type = b"application/speculationrules+json")

common/security-features/subresource/subresource.py

@@ -139,6 +139,14 @@ def preprocess_stash_action(request, response):
         else:
             status = u"blocked"
         response_data = json.dumps({u"status": status, u"result": value})
+    elif action == b"peek":
+        value = stash.take(key=key, path=path)
+        stash.put(key=key, value=value, path=path)
+        if value is None:
+            status = u"allowed"
+        else:
+            status = u"blocked"
+        response_data = json.dumps({u"status": status, u"result": value})
     else:
         return False
 

common/security-features/subresource/template/speculationrules.json.template

@@ -0,0 +1 @@
+{}

common/security-features/tools/generate.py

@@ -321,7 +321,7 @@ def generate_test_source_files(spec_directory, test_helper_filenames,
     # Choose a debug/release template depending on the target.
     html_template = "test.%s.html.template" % target
 
-    artifact_order = test_expansion_schema.keys()
+    artifact_order = list(test_expansion_schema.keys())
     artifact_order.remove('expansion')
 
     excluded_selection_pattern = ''

common/security-features/tools/spec.src.json

@@ -111,6 +111,7 @@
         "sharedworker-import",
         "sharedworker-import-data",
         "sharedworker-module",
+        "speculationrules",
         "video-tag",
         "worker-classic",
         "worker-import",
@@ -156,6 +157,31 @@
       ],
       "origin": "*",
       "expectation": "*"
+    },
+    {
+      // Only test speculation rules in the top document.
+      "expansion": "*",
+      "source_scheme": "*",
+      "source_context_list": [
+        "srcdoc-inherit",
+        "srcdoc",
+        "iframe",
+        "iframe-blank-inherit",
+        "worker-classic",
+        "worker-classic-data",
+        "worker-module",
+        "worker-module-data",
+        "sharedworker-classic",
+        "sharedworker-classic-data",
+        "sharedworker-module",
+        "sharedworker-module-data"
+      ],
+      "delivery_type": "*",
+      "delivery_value": "*",
+      "redirection": "*",
+      "subresource": "speculationrules",
+      "origin": "*",
+      "expectation": "*"
     }
   ],
   "source_context_schema": {
@@ -513,6 +539,7 @@
       "sharedworker-import",
       "sharedworker-import-data",
       "sharedworker-module",
+      "speculationrules",
       "video-tag",
       "websocket",
       "worker-classic",

common/security-features/tools/spec_validator.py

@@ -5,7 +5,7 @@
 
 def assert_non_empty_string(obj, field):
     assert field in obj, 'Missing field "%s"' % field
-    assert isinstance(obj[field], basestring), \
+    assert isinstance(obj[field], str), \
         'Field "%s" must be a string' % field
     assert len(obj[field]) > 0, 'Field "%s" must not be empty' % field
 
@@ -34,7 +34,7 @@ def assert_value_from(obj, field, items):
 
 
 def assert_atom_or_list_items_from(obj, field, items):
-    if isinstance(obj[field], basestring) or isinstance(
+    if isinstance(obj[field], str) or isinstance(
             obj[field], int) or obj[field] is None:
         assert_value_from(obj, field, items)
         return
@@ -126,6 +126,8 @@ def validate(spec_json, details):
         "worklet-layout-import", "worklet-paint-import",
         "worklet-animation-import-data", "worklet-audio-import-data",
         "worklet-layout-import-data", "worklet-paint-import-data"
+    ] + [
+        "speculationrules"
     ]
 
     # Validate each single spec.
@@ -236,7 +238,7 @@ def assert_valid_spec_json(spec_json):
     try:
         validate(spec_json, error_details)
     except AssertionError as err:
-        print('ERROR:', err.message)
+        print('ERROR:', err)
         print(json.dumps(error_details, indent=4))
         sys.exit(1)