import assert from "assert" ; import { Blob } from "buffer" ; import type EventEmitter from "events" ; import http from "http" ; import { ReadableByteStreamController , ReadableStream , ReadableStreamDefaultReader , UnderlyingByteSource , } from "stream/web" ; import { URL } from "url" ; import { Log , PluginContext , assertInRequest , getRequestContext , nonCircularClone , waitForOpenInputGate , waitForOpenOutputGate , } from "@miniflare/shared" ; import type { WebSocket } from "@miniflare/web-sockets" ; import type { BusboyHeaders } from "busboy" ; import { Colorize , blue , bold , green , grey , red , yellow } from "kleur/colors" ; import { splitCookiesString } from "set-cookie-parser" ; import { Request as BaseRequest , RequestInfo as BaseRequestInfo , RequestInit as BaseRequestInit , Response as BaseResponse , ResponseInit as BaseResponseInit , BodyInit , Dispatcher , File , FormData , Headers , RequestCache , RequestCredentials , RequestDestination , RequestMode , RequestRedirect , ResponseRedirectStatus , ResponseType , getGlobalDispatcher , } from "undici" ; import type { fetch as baseFetchType } from "undici" ; import { IncomingRequestCfProperties , RequestInitCfProperties } from "./cf" ; import { bufferSourceToArray , buildNotBufferSourceError , isBufferSource , } from "./helpers" ; import { kContentLength } from "./streams" ; const baseFetch = require ( "undici/lib/fetch" ) ; declare class HeadersList extends Array { delete ( name ) ; } const HeadersListImpl = require ( "undici/lib/fetch/headers.js" ) . HeadersList ; const fetchSymbols = require ( "undici/lib/fetch/symbols.js" ) ; export type { BodyInit , HeadersInit , RequestCache , RequestCredentials , RequestDestination , RequestMode , RequestRedirect , ResponseType , ResponseRedirectStatus , } from "undici" ; export { Headers } from "undici" ; const inspect = Symbol . for ( "nodejs.util.inspect.custom" ) ; const nonEnumerable = Object . create ( null ) ; nonEnumerable . enumerable = false ; function makeEnumerable < T > ( prototype , instance , keys ) { for ( const key of keys ) { const descriptor = Object . getOwnPropertyDescriptor ( prototype , key ) ! ; descriptor . enumerable = true ; Object . defineProperty ( instance , key , descriptor ) ; } } Headers . prototype . getAll = function ( key ) : string [ ] { if ( key . toLowerCase ( ) !== "set-cookie" ) { throw new TypeError ( 'getAll() can only be used with the header name "Set-Cookie".' ) ; } const value = this . get ( "set-cookie" ) ; return value ? splitCookiesString ( value ) : [ ] ; } ; internal export function _headersFromIncomingRequest ( req ) { const headers = new Headers ( ) ; for ( const [ name , values ] of Object . entries ( req . headers ) ) { if ( name === "transfer-encoding" || name === "connection" || name === "keep-alive" || name === "expect" ) { continue ; } if ( Array . isArray ( values ) ) { for ( const value of values ) headers . append ( name , value ) ; } else if ( values !== undefined ) { headers . append ( name , values ) ; } } return headers ; } internal export const _kInner = Symbol ( "kInner" ) ; const kInputGated = Symbol ( "kInputGated" ) ; const kFormDataFiles = Symbol ( "kFormDataFiles" ) ; const kCloned = Symbol ( "kCloned" ) ; internal export function _isByteStream ( stream ) { for ( const symbol of Object . getOwnPropertySymbols ( stream ) ) { if ( symbol . description === "kState" ) { const controller = stream [ symbol ] . controller ; return controller instanceof ReadableByteStreamController ; } } return false ; } const enumerableBodyKeys = [ "body" , "bodyUsed" , "headers" ] ; export class Body < Inner extends BaseRequest | BaseResponse > { internal [ _kInner ] : Inner ; [ kInputGated ] = false ; [ kFormDataFiles ] = true ; [ kCloned ] = false ; #bodyStream ? : ReadableStream < Uint8Array > ; constructor ( inner ) { inner . headers [ fetchSymbols . kGuard ] = "none" ; this [ _kInner ] = inner ; makeEnumerable ( Body . prototype , this , enumerableBodyKeys ) ; Object . defineProperty ( this , _kInner , nonEnumerable ) ; Object . defineProperty ( this , kInputGated , nonEnumerable ) ; Object . defineProperty ( this , kFormDataFiles , nonEnumerable ) ; } [ inspect ] ( ) { return this [ _kInner ] ; } get headers ( ) : Headers { return this [ _kInner ] . headers ; } get body ( ) : ReadableStream < Uint8Array > | null { const body = this [ _kInner ] . body ; if ( body === null ) return body ; if ( this . #bodyStream ) return this . #bodyStream ; assert ( body instanceof ReadableStream ) ; if ( ! this [ kInputGated ] && _isByteStream ( body ) ) { return ( this . #bodyStream = body ) ; } let reader ; const source = { type : "bytes" , pull : async ( controller ) => { if ( reader === undefined ) reader = body . getReader ( ) ; let { done , value } = await reader . read ( ) ; while ( ! done && isBufferSource ( value ) && value . byteLength === 0 ) { ( { done , value } = await reader . read ( ) ) ; } if ( this [ kInputGated ] ) await waitForOpenInputGate ( ) ; if ( isBufferSource ( value ) ) { if ( value . byteLength ) { let array = bufferSourceToArray ( value ) ; if ( this [ kCloned ] ) array = array . slice ( ) ; controller . enqueue ( array ) ; } } else if ( value ) { return controller . error ( new TypeError ( buildNotBufferSourceError ( value ) ) ) ; } if ( done ) controller . close ( ) ; } , cancel : ( reason ) => reader . cancel ( reason ) , } ; return ( this . #bodyStream = new ReadableStream ( source ) ) ; } get bodyUsed ( ) : boolean { return this [ _kInner ] . bodyUsed ; } async arrayBuffer ( ) { const body = await this [ _kInner ] . arrayBuffer ( ) ; if ( this [ kInputGated ] ) await waitForOpenInputGate ( ) ; return body ; } async blob ( ) { const body = await this [ _kInner ] . blob ( ) ; if ( this [ kInputGated ] ) await waitForOpenInputGate ( ) ; return body ; } async formData ( ) { const headers = { } ; for ( const [ key , value ] of this . headers ) headers [ key . toLowerCase ( ) ] = value ; if ( headers [ "content-type" ] === undefined ) { throw new TypeError ( "Parsing a Body as FormData requires a Content-Type header." ) ; } const formData = new FormData ( ) ; await new Promise < void > ( async ( resolve ) => { const Busboy = require ( "busboy" ) ; const busboy = new Busboy ( { headers : headers as BusboyHeaders } ) ; busboy . on ( "field" , ( name , value ) => { formData . append ( name , value ) ; } ) ; busboy . on ( "file" , ( name , value , filename , encoding , type ) => { const base64 = encoding . toLowerCase ( ) === "base64" ; const chunks = [ ] ; let totalLength = 0 ; value . on ( "data" , ( chunk ) => { if ( base64 ) chunk = Buffer . from ( chunk . toString ( ) , "base64" ) ; chunks . push ( chunk ) ; totalLength += chunk . byteLength ; } ) ; value . on ( "end" , ( ) => { if ( this [ kFormDataFiles ] ) { const file = new File ( chunks , filename , { type } ) ; formData . append ( name , file ) ; } else { const text = Buffer . concat ( chunks , totalLength ) . toString ( ) ; formData . append ( name , text ) ; } } ) ; } ) ; busboy . on ( "finish" , resolve ) ; const body = this [ _kInner ] . body ; if ( body !== null ) for await ( const chunk of body ) busboy . write ( chunk ) ; busboy . end ( ) ; } ) ; if ( this [ kInputGated ] ) await waitForOpenInputGate ( ) ; return formData ; } async json < T > ( ) { const body = await this [ _kInner ] . json ( ) ; if ( this [ kInputGated ] ) await waitForOpenInputGate ( ) ; return body as T ; } async text ( ) { const body = await this [ _kInner ] . text ( ) ; if ( this [ kInputGated ] ) await waitForOpenInputGate ( ) ; return body ; } } export function withInputGating < Inner extends Body < BaseRequest | BaseResponse > > ( body ) { body [ kInputGated ] = true ; return body ; } export function withStringFormDataFiles < Inner extends Body < BaseRequest | BaseResponse > > ( body ) { body [ kFormDataFiles ] = false ; return body ; } export type RequestInfo = BaseRequestInfo | Request ; export interface RequestInit extends BaseRequestInit { readonly cf ? : IncomingRequestCfProperties | RequestInitCfProperties ; } const enumerableRequestKeys = [ "cf" , "signal" , "redirect" , "url" , "method" , ] ; export class Request extends Body < BaseRequest > { #cf ? : IncomingRequestCfProperties | RequestInitCfProperties ; constructor ( input , init ? ) { const cf = input instanceof Request ? input . #cf : init ?. cf ; if ( input instanceof BaseRequest && ! init ) { super ( input ) ; } else { if ( input instanceof Request ) input = input [ _kInner ] ; super ( new BaseRequest ( input , init ) ) ; } this . #cf = cf ? nonCircularClone ( cf ) : undefined ; const contentLength = ( init ?. body as any ) ?. [ kContentLength ] ; if ( contentLength !== undefined ) { this . headers . set ( "content-length" , contentLength . toString ( ) ) ; } makeEnumerable ( Request . prototype , this , enumerableRequestKeys ) ; } clone ( ) { const innerClone = this [ _kInner ] . clone ( ) ; const clone = new Request ( innerClone ) ; clone [ kInputGated ] = this [ kInputGated ] ; clone [ kFormDataFiles ] = this [ kFormDataFiles ] ; clone . headers [ fetchSymbols . kGuard ] = this . headers [ fetchSymbols . kGuard ] ; clone . #cf = this . cf ? nonCircularClone ( this . cf ) : undefined ; this [ kCloned ] = true ; clone [ kCloned ] = true ; return clone ; } get cf ( ) : IncomingRequestCfProperties | RequestInitCfProperties | undefined { return this . #cf ; } get cache ( ) : RequestCache { return this [ _kInner ] . cache ; } get credentials ( ) : RequestCredentials { return this [ _kInner ] . credentials ; } get destination ( ) : RequestDestination { return this [ _kInner ] . destination ; } get integrity ( ) : string { return this [ _kInner ] . integrity ; } get method ( ) : string { return this [ _kInner ] . method ; } get mode ( ) : RequestMode { return this [ _kInner ] . mode ; } get redirect ( ) : RequestRedirect { return this [ _kInner ] . redirect ; } get referrerPolicy ( ) : string { return this [ _kInner ] . referrerPolicy ; } get url ( ) : string { return this [ _kInner ] . url ; } get keepalive ( ) : boolean { return this [ _kInner ] . keepalive ; } get signal ( ) : AbortSignal { return this [ _kInner ] . signal ; } } export function withImmutableHeaders ( req ) { req . headers [ fetchSymbols . kGuard ] = "immutable" ; return req ; } export interface ResponseInit extends BaseResponseInit { readonly encodeBody ? : "auto" | "manual" ; readonly webSocket ? : WebSocket ; } const kWaitUntil = Symbol ( "kWaitUntil" ) ; const nullBodyStatus = [ 204 , 205 , 304 ] ; const enumerableResponseKeys = [ "encodeBody" , "webSocket" , "url" , "redirected" , "ok" , "statusText" , "status" , ] ; export class Response < WaitUntil extends any [ ] = unknown [ ] > extends Body < BaseResponse > { static redirect ( url , status = 302 ) { const res = BaseResponse . redirect ( url , status ) ; return new Response ( res . body , res ) ; } #encodeBody : "auto" | "manual" ; #status ? : number ; readonly #webSocket ? : WebSocket ; [ kWaitUntil ] ? : Promise < WaitUntil > ; constructor ( body ? , init ? ) { let encodeBody ; let status ; let webSocket ; if ( init instanceof BaseResponse && body === init . body ) { super ( init ) ; } else { if ( init instanceof Response ) { encodeBody = init . #encodeBody ; status = init . #status ; webSocket = init . #webSocket ; init = init [ _kInner ] ; } else if ( ! ( init instanceof BaseResponse ) && init ) { encodeBody = init . encodeBody ; if ( init . webSocket ) { if ( init . status !== 101 ) { throw new RangeError ( "Responses with a WebSocket must have status code 101." ) ; } status = init . status ; webSocket = init . webSocket ; init = { ... init , status : 200 } ; } if ( nullBodyStatus . includes ( init . status ) && body === "" ) body = null ; } super ( new BaseResponse ( body , init ) ) ; } encodeBody ??= "auto" ; if ( encodeBody !== "auto" && encodeBody !== "manual" ) { throw new TypeError ( `encodeBody: unexpected value: ${ encodeBody }` ) ; } this . #encodeBody = encodeBody ; this . #status = status ; this . #webSocket = webSocket ; const contentLength = ( body as any ) ?. [ kContentLength ] ; if ( contentLength !== undefined ) { this . headers . set ( "content-length" , contentLength . toString ( ) ) ; } makeEnumerable ( Response . prototype , this , enumerableResponseKeys ) ; Object . defineProperty ( this , kWaitUntil , nonEnumerable ) ; } clone ( ) { if ( this . #webSocket ) { throw new TypeError ( "Cannot clone a response to a WebSocket handshake." ) ; } const innerClone = this [ _kInner ] . clone ( ) ; const clone = new Response ( innerClone . body , innerClone ) ; clone [ kInputGated ] = this [ kInputGated ] ; clone [ kFormDataFiles ] = this [ kFormDataFiles ] ; clone . #encodeBody = this . #encodeBody ; clone . #status = this . #status ; clone [ kWaitUntil ] = this [ kWaitUntil ] ; this [ kCloned ] = true ; clone [ kCloned ] = true ; return clone ; } get encodeBody ( ) : "auto" | "manual" { return this . #encodeBody ; } get webSocket ( ) : WebSocket | undefined { return this . #webSocket ; } waitUntil ( ) { return this [ kWaitUntil ] ?? Promise . resolve ( [ ] as unknown as WaitUntil ) ; } get status ( ) : number { return this . #status ?? this [ _kInner ] . status ; } get ok ( ) : boolean { return this [ _kInner ] . ok ; } get statusText ( ) : string { return this [ _kInner ] . statusText ; } get type ( ) : ResponseType { throw new Error ( "Failed to get the 'type' property on 'Response': the property is not implemented." ) ; } get url ( ) : string { return this [ _kInner ] . url ; } get redirected ( ) : boolean { return this [ _kInner ] . redirected ; } } export function withWaitUntil < WaitUntil extends any [ ] > ( res , waitUntil ) { const resWaitUntil = res instanceof Response ? ( res as Response < WaitUntil > ) : new Response ( res . body , res ) ; resWaitUntil [ kWaitUntil ] = waitUntil ; return resWaitUntil ; } internal export function _getURLList ( res ) { return res [ fetchSymbols . kState ] ?. urlList ; } internal export function _getBodyLength ( res ) { if ( res instanceof Response ) res = res [ _kInner ] ; return res [ fetchSymbols . kState ] ?. body ?. length ?? undefined ; } internal export const _kLoopHeader = "MF-Loop" ; const kDefaultHeadersToRemove = [ "accept" , "accept-language" , "sec-fetch-mode" , "user-agent" , ] ; class MiniflareDispatcher extends Dispatcher { constructor ( private readonly inner , private readonly removeHeaders , options ? ) { super ( options ) ; } dispatch ( options , handler ) { const headers = options . headers ; if ( headers ) { assert ( headers instanceof HeadersListImpl ) ; for ( const header of this . removeHeaders ) headers . delete ( header ) ; } return this . inner . dispatch ( options , handler ) ; } close ( ... args ) { return this . inner . close ( ... args ) ; } destroy ( ... args ) { return this . inner . destroy ( ... args ) ; } } export async function fetch ( input , init ? ) { const ctx = getRequestContext ( ) ; ctx ?. incrementSubrequests ( ) ; await waitForOpenOutputGate ( ) ; if ( input instanceof Request ) input = input [ _kInner ] ; const req = new BaseRequest ( input , init ) ; req . headers [ fetchSymbols . kGuard ] = "none" ; req . headers . delete ( "host" ) ; req . headers . delete ( "cf-connecting-ip" ) ; req . headers . set ( _kLoopHeader , String ( ctx ?. requestDepth ?? 1 ) ) ; const removeHeaders = [ ] ; for ( const header of kDefaultHeadersToRemove ) { if ( ! req . headers . has ( header ) ) removeHeaders . push ( header ) ; } const dispatcher = new MiniflareDispatcher ( getGlobalDispatcher ( ) , removeHeaders ) ; const baseRes = await baseFetch . call ( dispatcher , req ) ; if ( baseRes . redirected && ctx ) { const urlList = _getURLList ( baseRes ) ; if ( urlList ) ctx . incrementSubrequests ( urlList . length - 1 ) ; } let res ; if ( baseRes . type === "opaqueredirect" ) { const internalResponse = baseRes [ fetchSymbols . kState ] . internalResponse ; const headersList = internalResponse . headersList ; assert ( headersList . length % 2 === 0 ) ; const headers = new Headers ( ) ; for ( let i = 0 ; i < headersList . length ; i += 2 ) { headers . append ( headersList [ i ] , headersList [ i + 1 ] ) ; } res = new Response ( null , { status : internalResponse . status , statusText : internalResponse . statusText , headers , } ) ; } else { res = new Response ( baseRes . body , baseRes ) ; } await waitForOpenInputGate ( ) ; return withInputGating ( res ) ; } internal export function _urlFromRequestInput ( input ) { if ( input instanceof URL ) return input ; if ( input instanceof Request || input instanceof BaseRequest ) { return new URL ( input . url ) ; } return new URL ( input ) ; } internal export function _buildUnknownProtocolWarning ( url ) { let warning = "Worker passed an invalid URL to fetch(). URLs passed to fetch() " + "must begin with either 'http:' or 'https:', not '" + url . protocol + "'. Due to a historical bug, any other protocol used here will " + "be treated the same as 'http:'. We plan to correct this bug in " + "the future, so please update your Worker to use 'http:' or " + "'https:' for all fetch() URLs." ; if ( url . protocol === "ws:" || url . protocol === "wss:" ) { warning += " Note that fetch() treats WebSockets as a special kind of HTTP " + "request, therefore WebSockets should use 'http:'/'https:', not " + "'ws:'/'wss:'." ; } return warning ; } export function createCompatFetch ( { log , compat , globalAsyncIO , } , inner = fetch ) { const refusesUnknown = compat . isEnabled ( "fetch_refuses_unknown_protocols" ) ; const formDataFiles = compat . isEnabled ( "formdata_parser_supports_files" ) ; return async ( input , init ) => { if ( ! globalAsyncIO ) assertInRequest ( ) ; const url = _urlFromRequestInput ( input ) ; if ( url . protocol !== "http:" && url . protocol !== "https:" ) { if ( refusesUnknown ) { throw new TypeError ( `Fetch API cannot load: ${ url . toString ( ) }` ) ; } else { log . warn ( _buildUnknownProtocolWarning ( url ) ) ; if ( init ) { init = new Request ( input , init ) ; } else if ( input instanceof BaseRequest ) { init = new Request ( input ) ; } else if ( input instanceof Request ) { init = input ; } input = url . toString ( ) . replace ( url . protocol , "http:" ) ; } } let res = await inner ( input , init ) ; if ( ! formDataFiles ) res = withStringFormDataFiles ( res ) ; return res ; } ; } export type HRTime = [ seconds : number , nanoseconds : number ] ; function millisFromHRTime ( [ seconds , nanoseconds ] ) { return `${ ( ( seconds * 1e9 + nanoseconds ) / 1e6 ) . toFixed ( 2 ) }ms` ; } function colourFromHTTPStatus ( status ) { if ( 200 <= status && status < 300 ) return green ; if ( 400 <= status && status < 500 ) return yellow ; if ( 500 <= status ) return red ; return blue ; } export async function logResponse ( log , { start , method , url , status , waitUntil , } ) { const responseTime = millisFromHRTime ( process . hrtime ( start ) ) ; let waitUntilResponse ; try { waitUntilResponse = await waitUntil ; } catch ( e ) { waitUntilResponse = [ "" ] ; log . error ( e ) ; } const waitUntilTime = millisFromHRTime ( process . hrtime ( start ) ) ; log . log ( [ `${ bold ( method ) } ${ url } ` , status ? colourFromHTTPStatus ( status ) ( `${ bold ( status ) } ${ http . STATUS_CODES [ status ] } ` ) : "" , grey ( `(${ responseTime }` ) , waitUntilResponse ?. length ? grey ( `, waitUntil: ${ waitUntilTime }` ) : "" , grey ( ")" ) , ] . join ( "" ) ) ; }