import assert from "assert" ; import http from "http" ; import https from "https" ; import { Transform , Writable } from "stream" ; import { ReadableStream } from "stream/web" ; import { URL } from "url" ; import zlib from "zlib" ; import { CorePluginSignatures , MiniflareCore , Request , Response , _getBodyLength , _headersFromIncomingRequest , logResponse , } from "@miniflare/core" ; import { prefixError , randomHex } from "@miniflare/shared" ; import { coupleWebSocket } from "@miniflare/web-sockets" ; import { BodyInit , Headers } from "undici" ; import { getAccessibleHosts } from "./helpers" ; import { HTTPPlugin , RequestMeta } from "./plugin" ; export * from "./helpers" ; export * from "./plugin" ; export const DEFAULT_PORT = 8787 ; export type HTTPPluginSignatures = CorePluginSignatures & { HTTPPlugin : typeof HTTPPlugin ; } ; const liveReloadScript = `<script defer type="application/javascript">
(function () {
  // Miniflare Live Reload
  var url = new URL("/cdn-cgi/mf/reload", location.origin);
  url.protocol = url.protocol.replace("http", "ws");
  function reload() { location.reload(); }
  function connect(reconnected) {
    var ws = new WebSocket(url);
    if (reconnected) ws.onopen = reload;
    ws.onclose = function(e) {
      e.code === 1012 ? reload() : e.code === 1000 || e.code === 1001 || setTimeout(connect, 1000, true);
    }
  }
  connect();
})();
</script>` ; const liveReloadScriptLength = Buffer . byteLength ( liveReloadScript ) ; export async function convertNodeRequest ( req , meta ? ) { const protocol = req . socket . encrypted ? "https" : "http" ; const origin = `${ protocol }://${ req . headers . host ?? "localhost" }` ; const url = new URL ( req . url ?? "" , origin ) ; let body = null ; if ( req . method !== "GET" && req . method !== "HEAD" ) { let iterator ; body = new ReadableStream ( { type : "bytes" , start ( ) { iterator = req [ Symbol . asyncIterator ] ( ) ; } , async pull ( controller ) { const { done , value } = await iterator . next ( ) ; if ( done ) { queueMicrotask ( ( ) => controller . close ( ) ) ; } else { const buffer = Buffer . isBuffer ( value ) ? value : Buffer . from ( value ) ; controller . enqueue ( new Uint8Array ( buffer ) ) ; } } , async cancel ( ) { await iterator . return ?. ( ) ; } , } ) ; } const proto = meta ?. forwardedProto ?? "https" ; let ip = meta ?. realIp ?? req . socket . remoteAddress ?? "" ; if ( ip === "::1" ) ip = "127.0.0.1" ; if ( ip . startsWith ( "::ffff:" ) ) ip = ip . substring ( "::ffff:" . length ) ; req . headers [ "x-forwarded-proto" ] ??= proto ; req . headers [ "x-real-ip" ] ??= ip ; req . headers [ "cf-connecting-ip" ] ??= ip ; req . headers [ "cf-ipcountry" ] ??= meta ?. cf ?. country ?? "US" ; req . headers [ "cf-ray" ] ??= randomHex ( 16 ) ; req . headers [ "cf-visitor" ] ??= `{"scheme":"${ proto }"}` ; req . headers [ "host" ] = url . host ; const headers = _headersFromIncomingRequest ( req ) ; const request = new Request ( url , { method : req . method , headers , body , cf : meta ?. cf , redirect : "manual" , } ) ; return { request , url } ; } export type RequestListener = ( req : http . IncomingMessage , res ? : http . ServerResponse ) => Promise < Response | undefined > ; export function createRequestListener < Plugins extends HTTPPluginSignatures > ( mf ) { return async ( req , res ) => { const { HTTPPlugin } = await mf . getPlugins ( ) ; const start = process . hrtime ( ) ; const { request , url } = await convertNodeRequest ( req , await HTTPPlugin . getRequestMeta ( req ) ) ; let response ; let waitUntil ; let status = 500 ; const pathname = url . pathname . replace ( /\/$/ , "" ) ; if ( pathname . startsWith ( "/cdn-cgi/" ) ) { if ( pathname === "/cdn-cgi/mf/scheduled" ) { req . method = "SCHD" ; const time = url . searchParams . get ( "time" ) ; const cron = url . searchParams . get ( "cron" ) ; waitUntil = mf . dispatchScheduled ( time ? parseInt ( time ) : undefined , cron ?? undefined ) ; status = 200 ; } else { status = 404 ; } res ?. writeHead ( status , { "Content-Type" : "text/plain; charset=UTF-8" } ) ; res ?. end ( ) ; } else { try { response = await mf . dispatchFetch ( request ) ; waitUntil = response . waitUntil ( ) ; status = response . status ; const headers = { } ; for ( let [ key , value ] of response . headers ) { key = key . toLowerCase ( ) ; if ( key === "set-cookie" ) { headers [ "set-cookie" ] = response . headers . getAll ( "set-cookie" ) ; } else { headers [ key ] = value ; } } const contentLengthHeader = response . headers . get ( "Content-Length" ) ; const contentLength = _getBodyLength ( response ) ?? ( contentLengthHeader === null ? null : parseInt ( contentLengthHeader ) ) ; if ( contentLength !== null ) headers [ "content-length" ] = contentLength ; const encoders = [ ] ; if ( headers [ "content-encoding" ] && response . encodeBody === "auto" ) { delete headers [ "content-length" ] ; const codings = headers [ "content-encoding" ] . toString ( ) . toLowerCase ( ) . split ( "," ) . map ( ( x ) => x . trim ( ) ) ; for ( const coding of codings ) { if ( /(x-)?gzip/ . test ( coding ) ) { encoders . push ( zlib . createGzip ( ) ) ; } else if ( /(x-)?deflate/ . test ( coding ) ) { encoders . push ( zlib . createDeflate ( ) ) ; } else if ( coding === "br" ) { encoders . push ( zlib . createBrotliCompress ( ) ) ; } else { mf . log . warn ( `Unknown encoding \"${ coding }\", sending plain response...` ) ; delete headers [ "content-encoding" ] ; encoders . length = 0 ; break ; } } } const liveReloadEnabled = HTTPPlugin . liveReload && response . encodeBody === "auto" && response . headers . get ( "content-type" ) ?. toLowerCase ( ) . includes ( "text/html" ) ; if ( liveReloadEnabled && contentLength !== null ) { if ( ! isNaN ( contentLength ) ) { headers [ "content-length" ] = contentLength + liveReloadScriptLength ; } } res ?. writeHead ( status , headers ) ; if ( res ) { let initialStream = res ; for ( let i = encoders . length - 1 ; i >= 0 ; i -- ) { encoders [ i ] . pipe ( initialStream ) ; initialStream = encoders [ i ] ; } if ( response . body ) { for await ( const chunk of response . body ) { if ( chunk ) initialStream . write ( chunk ) ; } if ( liveReloadEnabled ) { initialStream . write ( liveReloadScript ) ; } } initialStream . end ( ) ; } } catch ( e ) { const accept = req . headers . accept ?. toLowerCase ( ) ?? "" ; if ( accept . includes ( "text/html" ) || accept . includes ( "*/*" ) || accept . includes ( "text/*" ) ) { const Youch = require ( "youch" ) ; const youch = new Youch ( e , req ) ; youch . addLink ( ( ) => { const links = [ '<a href="https://developers.cloudflare.com/workers/" target="_blank" style="text-decoration:none">📚 Workers Docs</a>' , '<a href="https://discord.gg/cloudflaredev" target="_blank" style="text-decoration:none">💬 Workers Discord</a>' , '<a href="https://miniflare.dev" target="_blank" style="text-decoration:none">🔥 Miniflare Docs</a>' , ] ; if ( HTTPPlugin . liveReload ) links . push ( liveReloadScript ) ; return links . join ( "" ) ; } ) ; const errorHtml = await youch . toHTML ( ) ; res ?. writeHead ( 500 , { "Content-Type" : "text/html; charset=UTF-8" } ) ; res ?. end ( errorHtml , "utf8" ) ; } else { res ?. writeHead ( 500 , { "Content-Type" : "text/plain; charset=UTF-8" } ) ; res ?. end ( e . stack , "utf8" ) ; } mf . log . error ( prefixError ( `${ req . method } ${ req . url }` , e ) ) ; } } assert ( req . method && req . url ) ; await logResponse ( mf . log , { start , method : req . method , url : req . url , status , waitUntil , } ) ; return response ; } ; } const restrictedWebSocketUpgradeHeaders = [ "upgrade" , "connection" , "sec-websocket-accept" , ] ; export async function createServer < Plugins extends HTTPPluginSignatures > ( mf , options ? ) { const plugins = await mf . getPlugins ( ) ; const listener = createRequestListener ( mf ) ; let server ; if ( plugins . HTTPPlugin . httpsEnabled ) { const httpsOptions = plugins . HTTPPlugin . httpsOptions ; assert ( httpsOptions ) ; server = https . createServer ( { ... httpsOptions , ... options } , listener ) ; } else { server = http . createServer ( options ?? { } , listener ) ; } const { WebSocketServer } = require ( "ws" ) ; const webSocketServer = new WebSocketServer ( { noServer : true } ) ; const liveReloadServer = new WebSocketServer ( { noServer : true } ) ; const extraHeaders = new WeakMap < http . IncomingMessage , Headers > ( ) ; webSocketServer . on ( "headers" , ( headers , req ) => { const extra = extraHeaders . get ( req ) ; extraHeaders . delete ( req ) ; if ( extra ) { for ( const [ key , value ] of extra ) { if ( ! restrictedWebSocketUpgradeHeaders . includes ( key . toLowerCase ( ) ) ) { headers . push ( `${ key }: ${ value }` ) ; } } } } ) ; server . on ( "upgrade" , async ( request , socket , head ) => { const { pathname } = new URL ( request . url ?? "" , "http://localhost" ) ; if ( pathname === "/cdn-cgi/mf/reload" ) { liveReloadServer . handleUpgrade ( request , socket as any , head , ( ws ) => { liveReloadServer . emit ( "connection" , ws , request ) ; } ) ; } else { const response = await listener ( request ) ; const webSocket = response ?. webSocket ; if ( response ?. status !== 101 || ! webSocket ) { socket . write ( "HTTP/1.1 500 Internal Server Error\r\n\r\n" ) ; socket . destroy ( ) ; mf . log . error ( new TypeError ( "Web Socket request did not return status 101 Switching Protocols response with Web Socket" ) ) ; return ; } extraHeaders . set ( request , response . headers ) ; webSocketServer . handleUpgrade ( request , socket as any , head , ( ws ) => { void coupleWebSocket ( ws , webSocket ) ; webSocketServer . emit ( "connection" , ws , request ) ; } ) ; } } ) ; const reloadListener = ( ) => { for ( const ws of liveReloadServer . clients ) { ws . close ( 1012 , "Service Restart" ) ; } for ( const ws of webSocketServer . clients ) { ws . close ( 1012 , "Service Restart" ) ; } } ; mf . addEventListener ( "reload" , reloadListener ) ; server . on ( "close" , ( ) => mf . removeEventListener ( "reload" , reloadListener ) ) ; return server ; } export async function startServer < Plugins extends HTTPPluginSignatures > ( mf , options ? ) { const server = await createServer ( mf , options ) ; const plugins = await mf . getPlugins ( ) ; const { httpsEnabled , host , port = DEFAULT_PORT } = plugins . HTTPPlugin ; return new Promise ( ( resolve ) => { server . listen ( port , host , ( ) => { const log = mf . log ; const protocol = httpsEnabled ? "https" : "http" ; const accessibleHosts = host ? [ host ] : getAccessibleHosts ( true ) ; log . info ( `Listening on ${ host ?? "" }:${ port }` ) ; for ( const accessibleHost of accessibleHosts ) { log . info ( `- ${ protocol }://${ accessibleHost }:${ port }` ) ; } resolve ( server ) ; } ) ; } ) ; }