import * as _ from 'lodash' ; import * as stream from 'stream' ; import * as net from 'net' ; import * as http from 'http' ; import * as http2 from 'http2' ; import { rawHeadersToHeaders } from '../../util/http' ; /**
 * A manager that intercepts BuildKit sessions, when given the corresponding sockets after the initial
 * upgrade, inject given config settings, and exposes methods to intercept build output streams elsewhere
 * to filter that injected config back out of build output.
 *
 * Under the hood, BuildKit uses the /session endpoint to upgrade the Docker CLI client into an HTTP/2 gRPC
 * server, so that the Docker host can query it to load files on demand, rather than taking the whole context
 * blindly every time. This class handles the traffic within that session, and exposes the data required
 * about that interception to handle other related logic
 *
 */ export class BuildkitSessionManager { connectBuildSession ( sessionUuid , clientSocket , dockerSocket , clientHead , dockerHead ) { if ( clientHead . length !== 0 ) { throw new Error ( "Unexpected CLI head data in BuildKit session upgrade" ) ; } if ( dockerHead . length !== 0 ) dockerSocket . unshift ( dockerHead ) ; const clientConnection = http2 . connect ( 'http://localhost' , { createConnection : ( ) => clientSocket } ) ; const server = http2 . createServer ( ) ; const wrappedDockerSocket = new SocketWrapper ( dockerSocket ) ; server . on ( 'session' , ( dockerSession ) => { clientConnection . on ( 'close' , ( ) => dockerSession . destroy ( ) ) ; } ) ; server . emit ( 'connection' , wrappedDockerSocket ) ; server . on ( 'stream' , ( dockerStream , _h , flags , rawRequestHeaders ) => { const requestHeaders = rawHeadersToHeaders ( rawRequestHeaders ) ; const clientStream = clientConnection . request ( requestHeaders , { endStream : parseHttp2HeaderFlags ( flags ) . endStream , waitForTrailers : true } ) ; dockerStream . pipe ( clientStream ) ; clientStream . pipe ( dockerStream ) ; clientStream . on ( 'response' , ( responseHeaders , flags ) => { dockerStream . respond ( responseHeaders , { waitForTrailers : true , endStream : parseHttp2HeaderFlags ( flags ) . endStream } ) ; } ) ; clientStream . on ( 'headers' , ( extraHeaders , flags ) => { dockerStream . respond ( extraHeaders , { waitForTrailers : true , endStream : parseHttp2HeaderFlags ( flags ) . endStream } ) ; } ) ; let pendingTrailers = { } ; clientStream . on ( 'trailers' , ( trailers ) => { pendingTrailers = { ... pendingTrailers , ... trailers } ; } ) ; dockerStream . once ( 'wantTrailers' , ( ) => { if ( ! _ . isEmpty ( pendingTrailers ) ) { dockerStream . sendTrailers ( pendingTrailers ) ; } else { dockerStream . close ( ) ; } } ) ; } ) ; } } const parseHttp2HeaderFlags = ( flags ) => ( { endStream : Boolean ( flags & 0x1 ) , endHeaders : Boolean ( flags & 0x4 ) , padded : Boolean ( flags & 0x8 ) , priority : Boolean ( flags & 0x20 ) } ) ; class SocketWrapper extends stream . Duplex { constructor ( private socket , options ? ) { super ( options ) ; } _write ( chunk , encoding , cb ) { if ( ! Buffer . isBuffer ( chunk ) ) { chunk = Buffer . from ( chunk , encoding ) ; } return this . socket . write ( chunk , cb ) ; } _read ( size ) { const data = this . socket . read ( size ) ; if ( data && data . length ) this . push ( data ) ; else this . socket . once ( 'data' , ( d ) => { this . push ( d ) ; } ) ; } }