Pin conformance proxies to configured origins

scripts/conformance/crates-registry-source.mjs

@@ -0,0 +1,13 @@
+const OFFICIAL_CRATES_IO_REGISTRY_SOURCES = new Set([
+  'registry+https://github.com/rust-lang/crates.io-index',
+  'registry+https://index.crates.io',
+  'registry+https://index.crates.io/',
+  'registry+sparse+https://index.crates.io',
+  'registry+sparse+https://index.crates.io/',
+  'sparse+https://index.crates.io',
+  'sparse+https://index.crates.io/',
+]);
+
+export function isOfficialCratesIoRegistrySource(source) {
+  return typeof source === 'string' && OFFICIAL_CRATES_IO_REGISTRY_SOURCES.has(source);
+}

scripts/conformance/heartbeat-shared-relay.mjs

@@ -3,6 +3,8 @@ import http from 'node:http';
 import path from 'node:path';
 import process from 'node:process';
 import { pathToFileURL } from 'node:url';
+import { brotliDecompressSync, gunzipSync, inflateSync } from 'node:zlib';
+import { proxyHttpRequest, proxyTargetUrl } from './proxy-target.mjs';
 
 const MAXIMUM_REQUEST_BYTES = 1024 * 1024;
 const MAXIMUM_RESPONSE_BYTES = 16 * 1024 * 1024;
@@ -252,44 +254,73 @@ async function requestBody(request) {
 async function responseBody(response) {
   const chunks = [];
   let bytes = 0;
-  for await (const chunk of response.body ?? []) {
+  for await (const chunk of response) {
     const buffer = Buffer.from(chunk);
     bytes += buffer.length;
     if (bytes > MAXIMUM_RESPONSE_BYTES) {
       throw new Error(`relay response exceeded ${MAXIMUM_RESPONSE_BYTES} bytes`);
     }
     chunks.push(buffer);
   }
-  return Buffer.concat(chunks);
+  const encoded = Buffer.concat(chunks);
+  const encoding = String(response.headers['content-encoding'] ?? 'identity').toLowerCase();
+  const options = { maxOutputLength: MAXIMUM_RESPONSE_BYTES };
+  const decoders = {
+    identity: (value) => value,
+    gzip: (value) => gunzipSync(value, options),
+    deflate: (value) => inflateSync(value, options),
+    br: (value) => brotliDecompressSync(value, options),
+  };
+  if (!(encoding in decoders)) throw new Error(`unsupported relay response encoding: ${encoding}`);
+  const decoded = decoders[encoding](encoded);
+  if (decoded.length > MAXIMUM_RESPONSE_BYTES) {
+    throw new Error(`relay response exceeded ${MAXIMUM_RESPONSE_BYTES} bytes`);
+  }
+  return decoded;
+}
+
+async function directHttpRequest({ requestTarget, allowedOrigin, method, headers, body }) {
+  const response = await new Promise((resolve, reject) => {
+    const request = proxyHttpRequest(requestTarget, allowedOrigin, { method, headers }, resolve);
+    request.setTimeout(REQUEST_TIMEOUT_MS, () => {
+      request.destroy(new Error(`relay request timed out after ${REQUEST_TIMEOUT_MS}ms`));
+    });
+    request.once('error', reject);
+    request.end(body.length > 0 || !['GET', 'HEAD'].includes(method) ? body : undefined);
+  });
+  return {
+    status: response.statusCode ?? 502,
+    contentType: response.headers['content-type'] ?? '',
+    body: await responseBody(response),
+  };
 }
 
 export async function directRelayRequest({
   targetUrl,
+  allowedOrigin,
   method,
   headers,
   body,
-  fetchImpl = globalThis.fetch,
+  requestImpl = directHttpRequest,
 }) {
-  if (typeof fetchImpl !== 'function') throw new Error('fetchImpl must be a function');
-  const forwardedHeaders = new Headers();
+  if (typeof requestImpl !== 'function') throw new Error('requestImpl must be a function');
+  const target = new URL(targetUrl);
+  if (target.origin !== new URL(allowedOrigin).origin) {
+    throw new Error('relay request changed the configured upstream origin');
+  }
+  const forwardedHeaders = {};
   for (const [name, value] of Object.entries(headers)) {
     if (EXCLUDED_HEADERS.has(name) || value === undefined) continue;
-    const values = Array.isArray(value) ? value : [value];
-    for (const entry of values) forwardedHeaders.append(name, String(entry));
+    forwardedHeaders[name] = Array.isArray(value) ? value.map(String) : String(value);
   }
 
-  const response = await fetchImpl(targetUrl, {
+  return requestImpl({
+    requestTarget: `${target.pathname}${target.search}`,
+    allowedOrigin,
     method,
     headers: forwardedHeaders,
-    body: body.length > 0 || !['GET', 'HEAD'].includes(method) ? body : undefined,
-    redirect: 'manual',
-    signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
+    body,
   });
-  return {
-    status: response.status,
-    contentType: response.headers.get('content-type') ?? '',
-    body: await responseBody(response),
-  };
 }
 
 export function createHeartbeatRelayServer({
@@ -305,10 +336,10 @@ export function createHeartbeatRelayServer({
   return http.createServer(async (request, response) => {
     try {
       const body = await requestBody(request);
-      const requestedUrl = new URL(request.url ?? '/', 'http://heartbeat-relay.invalid');
-      const targetUrl = new URL(`${requestedUrl.pathname}${requestedUrl.search}`, target);
+      const targetUrl = proxyTargetUrl(request.url, target);
       const forwarded = await executeRequest({
         targetUrl: targetUrl.toString(),
+        allowedOrigin: target.origin,
         method: request.method ?? 'GET',
         headers: request.headers,
         body,

scripts/conformance/proxy-target.mjs

@@ -0,0 +1,53 @@
+import http from 'node:http';
+import https from 'node:https';
+
+const REQUEST_TARGET_BASE = 'http://request-target.invalid';
+const DESTINATION_OPTIONS = ['protocol', 'host', 'hostname', 'port', 'socketPath', 'href', 'origin', 'path'];
+
+function proxyUpstream(configuredOrigin) {
+  const upstream = configuredOrigin instanceof URL
+    ? configuredOrigin
+    : new URL(configuredOrigin);
+  if (!['http:', 'https:'].includes(upstream.protocol)
+      || upstream.username
+      || upstream.password
+      || upstream.pathname !== '/'
+      || upstream.search
+      || upstream.hash) {
+    throw new Error('proxy upstream must be an HTTP(S) origin without credentials');
+  }
+
+  return upstream;
+}
+
+export function proxyTargetPath(requestTarget) {
+  const incoming = new URL(requestTarget ?? '/', REQUEST_TARGET_BASE);
+  return {
+    path: `${incoming.pathname}${incoming.search}`,
+    pathname: incoming.pathname,
+  };
+}
+
+export function proxyTargetUrl(requestTarget, configuredOrigin) {
+  const upstream = proxyUpstream(configuredOrigin);
+  const target = new URL(proxyTargetPath(requestTarget).path, upstream);
+  if (target.origin !== upstream.origin) {
+    throw new Error('proxy request target changed the configured upstream origin');
+  }
+
+  return target;
+}
+
+export function proxyHttpRequest(requestTarget, configuredOrigin, options = {}, callback) {
+  const upstream = proxyUpstream(configuredOrigin);
+  for (const name of DESTINATION_OPTIONS) {
+    if (Object.hasOwn(options, name)) {
+      throw new Error(`proxy request options must not override ${name}`);
+    }
+  }
+  const transport = upstream.protocol === 'https:' ? https : http;
+  return transport.request(upstream.origin, {
+    ...options,
+    path: proxyTargetPath(requestTarget).path,
+  }, callback);
+}

scripts/conformance/server-parity/published-sdk-patch/observe-http.mjs

@@ -2,6 +2,7 @@
 import http from 'node:http';
 import {appendFileSync} from 'node:fs';
 import {gunzipSync, inflateSync, brotliDecompressSync} from 'node:zlib';
+import {proxyHttpRequest, proxyTargetPath} from '../../proxy-target.mjs';
 
 const [target, journal, observationMode, gateMode, gateWorkflowId, ...extra] = process.argv.slice(2);
 if (observationMode !== undefined && observationMode !== '--forwarding-state') throw new Error('Unknown observation mode');
@@ -29,7 +30,7 @@ const server = http.createServer((request, response) => {
   const record = fields => {
     if (recorded) return;
     recorded = true;
-    appendFileSync(journal, JSON.stringify({method: request.method, path: new URL(request.url, upstream).pathname,
+    appendFileSync(journal, JSON.stringify({method: request.method, path: proxyTargetPath(request.url).pathname,
       ...(captureNamespace ? {namespace: request.headers['x-namespace'] ?? null} : {}),
       request_body: Buffer.concat(chunks).toString('utf8'), status: 0, response_encoding: 'identity',
       response_retry_after: null, response_bytes_base64: '', response_body: '', client_cancelled: false,
@@ -58,13 +59,13 @@ const server = http.createServer((request, response) => {
       if (recorded) return;
       const arrivalLine = arrivalJournal ? JSON.stringify({schema: 'durable-workflow.server-parity-http-request-arrival/v1',
         request_observation_id: requestObservationId, method: request.method,
-        path: new URL(request.url, upstream).pathname, request_body: body.toString('utf8'),
+        path: proxyTargetPath(request.url).pathname, request_body: body.toString('utf8'),
         request_body_complete: true, upstream_request_started: true, response_observed: false})+'\n' : null;
       const arrivalSize = arrivalLine === null ? 0 : Buffer.byteLength(arrivalLine);
       if (arrivalLine && (arrivalCount >= 512 || arrivalBytes + arrivalSize > 4 * 1024 * 1024)) {
         throw new Error('Request-arrival journal exceeds its bounded fixture budget');
       }
-      forwarded = http.request(new URL(request.url, upstream), {
+      forwarded = proxyHttpRequest(request.url, upstream, {
         method: request.method,
         headers: {...request.headers, host: upstream.host},
       }, incoming => {
@@ -82,7 +83,7 @@ const server = http.createServer((request, response) => {
             response_bytes_base64: bytes.toString('base64'), response_body: decoded});
           response.writeHead(incoming.statusCode, incoming.headers);
           response.end(bytes);
-          if (!startObserved && request.method === 'POST' && new URL(request.url, upstream).pathname === '/api/workflows'
+          if (!startObserved && request.method === 'POST' && proxyTargetPath(request.url).pathname === '/api/workflows'
             && incoming.statusCode === 201) {
             let started, receipt;
             try { started = JSON.parse(body.toString('utf8')); receipt = JSON.parse(decoded); } catch { return; }
@@ -107,7 +108,7 @@ const server = http.createServer((request, response) => {
     // Keep the real worker request intact until this fixture's accepted start
     // is fully observed. Other cohorts retain their original forwarding behavior.
     if (!startObserved && request.method === 'POST'
-      && new URL(request.url, upstream).pathname === '/api/worker/workflow-tasks/poll') waitingPolls.add(forwardRequest);
+      && proxyTargetPath(request.url).pathname === '/api/worker/workflow-tasks/poll') waitingPolls.add(forwardRequest);
     else forwardRequest();
   });
 });

scripts/conformance/server-parity/query-authority/observer.mjs

@@ -4,6 +4,7 @@ import http from 'node:http';
 import {appendFileSync} from 'node:fs';
 import {createHash} from 'node:crypto';
 import {gunzipSync, inflateSync, brotliDecompressSync} from 'node:zlib';
+import {proxyHttpRequest, proxyTargetPath} from '../../proxy-target.mjs';
 
 const [target, mode, journal] = process.argv.slice(2);
 if (!['timely', 'late-poll', 'late-handler-async', 'late-handler-sync'].includes(mode)) throw Error('Unknown case');
@@ -34,7 +35,7 @@ function exchange(method, path, headers, body = Buffer.alloc(0)) {
       };
       reject(error);
     };
-    const request = http.request(new URL(path, upstream), {method, headers: {...headers, host: upstream.host}}, incoming => {
+    const request = proxyHttpRequest(path, upstream, {method, headers: {...headers, host: upstream.host}}, incoming => {
       response = incoming;
       const chunks = [];
       incoming.on('data', chunk => {
@@ -66,7 +67,7 @@ function json(reply) {
 }
 const sha = bytes => createHash('sha256').update(bytes).digest('hex');
 const server = http.createServer(async (request, response) => {
-  const path = new URL(request.url, upstream).pathname;
+  const path = proxyTargetPath(request.url).pathname;
   let upstreamStarted = false;
   const forward = (...arguments_) => {
     upstreamStarted = true;

scripts/conformance/skew-published-artifacts.mjs

@@ -5,6 +5,8 @@ import path from 'node:path';
 import process from 'node:process';
 import { spawn } from 'node:child_process';
 import { fileURLToPath } from 'node:url';
+import { brotliDecompressSync, gunzipSync, inflateSync } from 'node:zlib';
+import { proxyHttpRequest, proxyTargetUrl } from './proxy-target.mjs';
 import { isExactPythonRelease, isExactSemverRelease } from './version-identities.mjs';
 
 const RESULT_SCHEMA = 'durable-workflow.v2.skew-refusal-matrix.result';
@@ -3356,6 +3358,7 @@ function normalizeArtifactHeaders(headers) {
 
 async function withRecordingProxy({ targetUrl, artifactProxyHost = null }, callback) {
   const captures = [];
+  const targetOrigin = new URL(targetUrl);
   const server = http.createServer(async (request, response) => {
     const chunks = [];
     request.on('data', (chunk) => chunks.push(chunk));
@@ -3366,7 +3369,7 @@ async function withRecordingProxy({ targetUrl, artifactProxyHost = null }, callb
       delete headers.connection;
       delete headers['content-length'];
 
-      const requestUrl = new URL(request.url ?? '/', targetUrl);
+      const requestUrl = proxyTargetUrl(request.url, targetOrigin);
       const capture = {
         id: `proxy-${captures.length + 1}`,
         request: {
@@ -3383,20 +3386,29 @@ async function withRecordingProxy({ targetUrl, artifactProxyHost = null }, callb
       };
 
       try {
-        const upstream = await fetch(requestUrl, {
-          method: request.method,
-          headers,
-          body: requestBody.length > 0 ? requestBody : undefined,
+        const upstream = await new Promise((resolve, reject) => {
+          const forwarded = proxyHttpRequest(request.url, targetOrigin, {
+            method: request.method,
+            headers,
+          }, resolve);
+          forwarded.once('error', reject);
+          forwarded.end(requestBody.length > 0 ? requestBody : undefined);
         });
-        const text = await upstream.text();
+        const responseChunks = [];
+        for await (const chunk of upstream) responseChunks.push(Buffer.from(chunk));
+        const encoded = Buffer.concat(responseChunks);
+        const encoding = String(upstream.headers['content-encoding'] ?? 'identity').toLowerCase();
+        const decoders = { identity: (value) => value, gzip: gunzipSync, deflate: inflateSync, br: brotliDecompressSync };
+        if (!(encoding in decoders)) throw new Error(`Unsupported proxy response encoding: ${encoding}`);
+        const text = decoders[encoding](encoded).toString('utf8');
         const parsed = parseJson(text);
-        const responseHeaders = Object.fromEntries(upstream.headers.entries());
+        const responseHeaders = normalizeArtifactHeaders(upstream.headers);
         capture.response = {
-          status: upstream.status,
+          status: upstream.statusCode ?? 502,
           headers: responseHeaders,
           body: parsed ?? text,
         };
-        response.statusCode = upstream.status;
+        response.statusCode = upstream.statusCode ?? 502;
         for (const [key, value] of Object.entries(responseHeaders)) {
           if (['content-encoding', 'content-length', 'transfer-encoding', 'connection'].includes(key.toLowerCase())) {
             continue;

scripts/conformance/workflow-lifecycle-published-artifacts.mjs

@@ -1,5 +1,6 @@
 import fs from 'node:fs';
 import path from 'node:path';
+import { isOfficialCratesIoRegistrySource } from './crates-registry-source.mjs';
 import { isExactPythonRelease, isExactSemverRelease } from './version-identities.mjs';
 
 const RESULT_DIR = mustEnv('RESULT_DIR');
@@ -1169,7 +1170,7 @@ function rustArtifactMismatch(outputs) {
     || stringValue(provenance.package) !== 'durable-workflow'
     || stringValue(provenance.requested_version) !== sdkVersion
     || stringValue(provenance.installed_version) !== sdkVersion
-    || !stringValue(provenance.registry_source).includes('crates.io')
+    || !isOfficialCratesIoRegistrySource(stringValue(provenance.registry_source))
     || !/^[0-9a-f]{64}$/.test(stringValue(provenance.registry_checksum_sha256));
 }
 
@@ -2283,7 +2284,7 @@ function validateRustSdkLifecycleSurface(outputs) {
   if (provenance.package !== 'durable-workflow'
       || provenance.requested_version !== outputs.artifact_version
       || provenance.installed_version !== outputs.artifact_version
-      || !normalizedText(provenance.registry_source).includes('crates.io')
+      || !isOfficialCratesIoRegistrySource(normalizedText(provenance.registry_source))
       || !/^[0-9a-f]{64}$/.test(stringValue(provenance.registry_checksum_sha256))) {
     failures.push('install_provenance must prove the exact crates.io durable-workflow package and checksum');
   }
@@ -2292,7 +2293,7 @@ function validateRustSdkLifecycleSurface(outputs) {
       || payload.envelope_contract !== 'durable-workflow-published-envelope'
       || payload.apache_avro_package !== 'apache-avro'
       || !truthyFlag(payload.official_crates_io_provenance)
-      || !normalizedText(payload.apache_avro_registry_source).includes('crates.io')
+      || !isOfficialCratesIoRegistrySource(normalizedText(payload.apache_avro_registry_source))
       || !/^[0-9a-f]{64}$/.test(stringValue(payload.apache_avro_registry_checksum_sha256))) {
     failures.push('payload_contract must prove the official apache-avro crate and published Avro envelope');
   }
@@ -2434,7 +2435,7 @@ function normalizeSuppliedFindings(entry, scenario, status, classification, fall
 
 function generatedFinding(scenario, status, classification, summary) {
   return {
-    finding_id: `workflow-lifecycle-${scenario.id.replace(/_/g, '-')}-${classification.replace(/-/g, '-')}`,
+    finding_id: `workflow-lifecycle-${scenario.id.replace(/_/g, '-')}-${classification}`,
     finding_type: findingTypeFor(classification, status),
     classification,
     scenario_id: scenario.id,
@@ -2614,7 +2615,7 @@ function normalizeScenario(scenario, entry, policy) {
 
 function findingForPolicy(index, classification, summary) {
   return {
-    finding_id: `workflow-lifecycle-${classification.replace(/-/g, '-')}-${index + 1}`,
+    finding_id: `workflow-lifecycle-${classification}-${index + 1}`,
     finding_type: findingTypeFor(classification, classification === 'runner-gap' ? 'runner_blocked' : 'fail'),
     classification,
     scenario_id: 'artifact_policy',

scripts/conformance/workflow-lifecycle-rust-published-artifacts.mjs

@@ -2,6 +2,7 @@ import crypto from 'node:crypto';
 import fs from 'node:fs';
 import path from 'node:path';
 import { spawnSync } from 'node:child_process';
+import { isOfficialCratesIoRegistrySource } from './crates-registry-source.mjs';
 import { isExactSemverRelease } from './version-identities.mjs';
 
 const RESULT_DIR = required('RESULT_DIR');
@@ -148,7 +149,7 @@ function provenance(lock, name, version = '') {
   const resolvedVersion = block.match(/version = "([^"]+)"/)?.[1] || '';
   const source = block.match(/source = "([^"]+)"/)?.[1] || '';
   const checksum = block.match(/checksum = "([0-9a-f]{64})"/)?.[1] || '';
-  if (!source.includes('crates.io') || !checksum) {
+  if (!isOfficialCratesIoRegistrySource(source) || !checksum) {
     throw new Error(`${name} did not resolve from crates.io with a registry checksum`);
   }
   return { package: name, resolved_version: resolvedVersion, registry_source: source, registry_checksum_sha256: checksum };