60 private const SENSITIVE_HEADERS =
'/(^|-|_)(authorization|auth|password|cookie)($|-|_)/';
87 protected $userAgent =
'wikimedia/multi-http-client v1.1';
95 private const TIMEOUT_ACCURACY_FACTOR = 0.1;
122 if ( isset( $options[
'caBundlePath'] ) ) {
123 $this->caBundlePath = $options[
'caBundlePath'];
124 if ( !file_exists( $this->caBundlePath ) ) {
125 throw new InvalidArgumentException(
"Cannot find CA bundle: " . $this->caBundlePath );
129 'connTimeout',
'maxConnTimeout',
'reqTimeout',
'maxReqTimeout',
130 'usePipelining',
'maxConnsPerHost',
'proxy',
'userAgent',
'logger',
131 'localProxy',
'localVirtualHosts',
'headers',
'telemetry'
133 foreach ( $opts as $key ) {
134 if ( isset( $options[$key] ) ) {
135 $this->$key = $options[$key];
138 $this->logger ??=
new NullLogger;
164 public function run( array $req, array $opts = [] ) {
165 return $this->
runMulti( [ $req ], $opts )[0][
'response'];
199 public function runMulti( array $reqs, array $opts = [] ) {
200 $this->normalizeRequests( $reqs );
203 if ( $this->maxConnTimeout && $opts[
'connTimeout'] > $this->maxConnTimeout ) {
206 if ( $this->maxReqTimeout && $opts[
'reqTimeout'] > $this->maxReqTimeout ) {
211 switch ( $opts[
'httpVersion'] ??
null ) {
213 $opts[
'httpVersion'] = CURL_HTTP_VERSION_1_0;
216 $opts[
'httpVersion'] = CURL_HTTP_VERSION_1_1;
220 $opts[
'httpVersion'] = CURL_HTTP_VERSION_2_0;
223 $opts[
'httpVersion'] = CURL_HTTP_VERSION_NONE;
225 return $this->runMultiCurl( $reqs, $opts );
227 # TODO: Add handling for httpVersion option
228 return $this->runMultiHttp( $reqs, $opts );
240 return extension_loaded(
'curl' ) && function_exists(
'curl_multi_init' );
260 private function runMultiCurl( array $reqs, array $opts ) {
263 $selectTimeout = $this->getSelectTimeout( $opts );
267 foreach ( $reqs as $index => &$req ) {
269 curl_multi_add_handle( $chm, $handles[$index] );
279 $mrc = curl_multi_exec( $chm, $active );
280 $info = curl_multi_info_read( $chm );
281 if ( $info !==
false ) {
285 $infos[(int)$info[
'handle']] = $info;
287 }
while ( $mrc == CURLM_CALL_MULTI_PERFORM );
289 if ( $active > 0 && $mrc == CURLM_OK && curl_multi_select( $chm, $selectTimeout ) == -1 ) {
293 }
while ( $active > 0 && $mrc == CURLM_OK );
296 foreach ( $reqs as $index => &$req ) {
297 $ch = $handles[$index];
298 curl_multi_remove_handle( $chm, $ch );
300 if ( isset( $infos[(
int)$ch] ) ) {
301 $info = $infos[(int)$ch];
302 $errno = $info[
'result'];
303 if ( $errno !== 0 ) {
304 $req[
'response'][
'error'] =
"(curl error: $errno)";
305 if ( function_exists(
'curl_strerror' ) ) {
306 $req[
'response'][
'error'] .=
" " . curl_strerror( $errno );
308 $this->logger->warning(
"Error fetching URL \"{$req['url']}\": " .
309 $req[
'response'][
'error'] );
311 $this->logger->debug(
312 "HTTP complete: {method} {url} code={response_code} size={size} " .
313 "total={total_time} connect={connect_time}",
315 'method' => $req[
'method'],
316 'url' => $req[
'url'],
317 'response_code' => $req[
'response'][
'code'],
318 'size' => curl_getinfo( $ch, CURLINFO_SIZE_DOWNLOAD ),
319 'total_time' => $this->getCurlTime(
320 $ch, CURLINFO_TOTAL_TIME,
'CURLINFO_TOTAL_TIME_T'
322 'connect_time' => $this->getCurlTime(
323 $ch, CURLINFO_CONNECT_TIME,
'CURLINFO_CONNECT_TIME_T'
329 $req[
'response'][
'error'] =
"(curl error: no status set)";
333 $req[
'response'][0] = $req[
'response'][
'code'];
334 $req[
'response'][1] = $req[
'response'][
'reason'];
335 $req[
'response'][2] = $req[
'response'][
'headers'];
336 $req[
'response'][3] = $req[
'response'][
'body'];
337 $req[
'response'][4] = $req[
'response'][
'error'];
340 if ( isset( $req[
'_closeHandle'] ) ) {
341 fclose( $req[
'_closeHandle'] );
342 unset( $req[
'_closeHandle'] );
365 curl_setopt( $ch, CURLOPT_PROXY, $req[
'proxy'] ?? $this->proxy );
366 curl_setopt( $ch, CURLOPT_CONNECTTIMEOUT_MS, intval( $opts[
'connTimeout'] * 1e3 ) );
367 curl_setopt( $ch, CURLOPT_TIMEOUT_MS, intval( $opts[
'reqTimeout'] * 1e3 ) );
368 curl_setopt( $ch, CURLOPT_FOLLOWLOCATION, 1 );
369 curl_setopt( $ch, CURLOPT_MAXREDIRS, 4 );
370 curl_setopt( $ch, CURLOPT_HEADER, 0 );
371 if ( $this->caBundlePath !==
null ) {
372 curl_setopt( $ch, CURLOPT_SSL_VERIFYPEER,
true );
373 curl_setopt( $ch, CURLOPT_CAINFO, $this->caBundlePath );
375 curl_setopt( $ch, CURLOPT_RETURNTRANSFER, 1 );
378 $query = http_build_query( $req[
'query'],
'',
'&', PHP_QUERY_RFC3986 );
379 if ( $query !=
'' ) {
380 $url .= strpos( $req[
'url'],
'?' ) ===
false ?
"?$query" :
"&$query";
382 curl_setopt( $ch, CURLOPT_URL,
$url );
383 curl_setopt( $ch, CURLOPT_CUSTOMREQUEST, $req[
'method'] );
384 curl_setopt( $ch, CURLOPT_NOBODY, ( $req[
'method'] ===
'HEAD' ) );
385 curl_setopt( $ch, CURLOPT_HTTP_VERSION, $opts[
'httpVersion'] ?? CURL_HTTP_VERSION_NONE );
387 if ( $req[
'method'] ===
'PUT' ) {
388 curl_setopt( $ch, CURLOPT_PUT, 1 );
390 if ( is_resource( $req[
'body'] ) ) {
391 curl_setopt( $ch, CURLOPT_INFILE, $req[
'body'] );
392 if ( isset( $req[
'headers'][
'content-length'] ) ) {
393 curl_setopt( $ch, CURLOPT_INFILESIZE, $req[
'headers'][
'content-length'] );
394 } elseif ( isset( $req[
'headers'][
'transfer-encoding'] ) &&
395 $req[
'headers'][
'transfer-encoding'] ===
'chunks'
397 curl_setopt( $ch, CURLOPT_UPLOAD,
true );
399 throw new InvalidArgumentException(
"Missing 'Content-Length' or 'Transfer-Encoding' header." );
401 } elseif ( $req[
'body'] !==
'' ) {
402 $fp = fopen(
"php://temp",
"wb+" );
403 fwrite( $fp, $req[
'body'], strlen( $req[
'body'] ) );
405 curl_setopt( $ch, CURLOPT_INFILE, $fp );
406 curl_setopt( $ch, CURLOPT_INFILESIZE, strlen( $req[
'body'] ) );
407 $req[
'_closeHandle'] = $fp;
409 curl_setopt( $ch, CURLOPT_INFILESIZE, 0 );
411 curl_setopt( $ch, CURLOPT_READFUNCTION,
412 static function ( $ch, $fd, $length ) {
413 return (
string)fread( $fd, $length );
416 } elseif ( $req[
'method'] ===
'POST' ) {
417 curl_setopt( $ch, CURLOPT_POST, 1 );
418 curl_setopt( $ch, CURLOPT_POSTFIELDS, $req[
'body'] );
421 if ( is_resource( $req[
'body'] ) || $req[
'body'] !==
'' ) {
422 throw new InvalidArgumentException(
"HTTP body specified for a non PUT/POST request." );
424 $req[
'headers'][
'content-length'] = 0;
427 if ( !isset( $req[
'headers'][
'user-agent'] ) ) {
432 foreach ( $req[
'headers'] as $name => $value ) {
433 if ( strpos( $name,
':' ) !==
false ) {
434 throw new InvalidArgumentException(
"Header name must not contain colon-space." );
436 $headers[] = $name .
': ' . trim( $value );
438 curl_setopt( $ch, CURLOPT_HTTPHEADER,
$headers );
440 curl_setopt( $ch, CURLOPT_HEADERFUNCTION,
441 static function ( $ch,
$header ) use ( &$req ) {
442 if ( !empty( $req[
'flags'][
'relayResponseHeaders'] ) && trim(
$header ) !==
'' ) {
447 if ( preg_match(
"/^(HTTP\/(?:1\.[01]|2)) (\d{3}) (.*)/",
$header,
$matches ) ) {
448 $req[
'response'][
'code'] = (int)
$matches[2];
449 $req[
'response'][
'reason'] = trim(
$matches[3] );
452 $req[
'response'][
'headers'] = [];
455 if ( strpos(
$header,
":" ) ===
false ) {
458 [ $name, $value ] = explode(
":",
$header, 2 );
459 $name = strtolower( $name );
460 $value = trim( $value );
461 if ( isset( $req[
'response'][
'headers'][$name] ) ) {
462 $req[
'response'][
'headers'][$name] .=
', ' . $value;
464 $req[
'response'][
'headers'][$name] = $value;
471 $hasOutputStream = isset( $req[
'stream'] );
472 curl_setopt( $ch, CURLOPT_WRITEFUNCTION,
473 static function ( $ch, $data ) use ( &$req, $hasOutputStream ) {
474 if ( $hasOutputStream ) {
476 return fwrite( $req[
'stream'], $data );
479 $req[
'response'][
'body'] .= $data;
481 return strlen( $data );
497 $cmh = curl_multi_init();
500 curl_multi_setopt(
$cmh, CURLMOPT_MAXCONNECTS, (
int)$this->maxConnsPerHost );
504 $curlVersion = curl_version()[
'version'];
507 if ( version_compare( $curlVersion,
'7.30.0',
'>=' ) ) {
510 curl_multi_setopt( $this->cmh, CURLMOPT_MAX_HOST_CONNECTIONS, (
int)$maxHostConns );
513 if ( $opts[
'usePipelining'] ?? $this->usePipelining ) {
514 if ( version_compare( $curlVersion,
'7.43',
'<' ) ) {
517 } elseif ( version_compare( $curlVersion,
'7.62',
'<' ) ) {
519 $pipelining = CURLPIPE_HTTP1 | CURLPIPE_MULTIPLEX;
522 $pipelining = CURLPIPE_MULTIPLEX;
526 @curl_multi_setopt( $this->cmh, CURLMOPT_PIPELINING, $pipelining );
542 private function getCurlTime( $ch, $oldOption, $newConstName ): string {
543 if ( defined( $newConstName ) ) {
544 return sprintf(
"%.6F", curl_getinfo( $ch, constant( $newConstName ) ) / 1e6 );
546 return (
string)curl_getinfo( $ch, $oldOption );
565 private function runMultiHttp( array $reqs, array $opts = [] ) {
567 'timeout' => $opts[
'reqTimeout'] ?? $this->reqTimeout,
568 'connectTimeout' => $opts[
'connTimeout'] ?? $this->connTimeout,
569 'logger' => $this->logger,
570 'caInfo' => $this->caBundlePath,
572 foreach ( $reqs as &$req ) {
573 $reqOptions = $httpOptions + [
574 'method' => $req[
'method'],
575 'proxy' => $req[
'proxy'] ?? $this->proxy,
576 'userAgent' => $req[
'headers'][
'user-agent'] ?? $this->userAgent,
577 'postData' => $req[
'body'],
581 $query = http_build_query( $req[
'query'],
'',
'&', PHP_QUERY_RFC3986 );
582 if ( $query !=
'' ) {
583 $url .= strpos( $req[
'url'],
'?' ) ===
false ?
"?$query" :
"&$query";
586 $httpRequest = MediaWikiServices::getInstance()->getHttpRequestFactory()->create(
587 $url, $reqOptions, __METHOD__ );
588 $httpRequest->setLogger( $this->logger );
589 foreach ( $req[
'headers'] as
$header => $value ) {
590 $httpRequest->setHeader(
$header, $value );
592 $sv = $httpRequest->execute()->getStatusValue();
594 $respHeaders = array_map(
595 static function ( $v ) {
596 return implode(
', ', $v );
598 $httpRequest->getResponseHeaders() );
601 'code' => $httpRequest->getStatus(),
603 'headers' => $respHeaders,
604 'body' => $httpRequest->getContent(),
608 if ( !$sv->isOK() ) {
609 $svErrors = $sv->getErrors();
610 if ( isset( $svErrors[0] ) ) {
611 $req[
'response'][
'error'] = $svErrors[0][
'message'];
614 if ( isset( $svErrors[0][
'params'][0] ) ) {
615 if ( is_numeric( $svErrors[0][
'params'][0] ) ) {
616 if ( isset( $svErrors[0][
'params'][1] ) ) {
618 $req[
'response'][
'reason'] = $svErrors[0][
'params'][1];
621 $req[
'response'][
'reason'] = $svErrors[0][
'params'][0];
627 $req[
'response'][0] = $req[
'response'][
'code'];
628 $req[
'response'][1] = $req[
'response'][
'reason'];
629 $req[
'response'][2] = $req[
'response'][
'headers'];
630 $req[
'response'][3] = $req[
'response'][
'body'];
631 $req[
'response'][4] = $req[
'response'][
'error'];
642 private function normalizeHeaders( array $headers ): array {
644 foreach ( $headers as $name => $value ) {
645 $normalized[strtolower( $name )] = $value;
655 private function normalizeRequests( array &$reqs ) {
656 foreach ( $reqs as &$req ) {
664 if ( isset( $req[0] ) ) {
665 $req[
'method'] = $req[0];
668 if ( isset( $req[1] ) ) {
669 $req[
'url'] = $req[1];
672 if ( !isset( $req[
'method'] ) ) {
673 throw new InvalidArgumentException(
"Request has no 'method' field set." );
674 } elseif ( !isset( $req[
'url'] ) ) {
675 throw new InvalidArgumentException(
"Request has no 'url' field set." );
677 if ( $this->localProxy !==
false && $this->isLocalURL( $req[
'url'] ) ) {
678 $this->useReverseProxy( $req, $this->localProxy );
680 $req[
'query'] ??= [];
681 $req[
'headers'] = $this->normalizeHeaders(
684 $this->telemetry ? $this->telemetry->getRequestHeaders() : [],
685 $req[
'headers'] ?? []
689 if ( !isset( $req[
'body'] ) ) {
691 $req[
'headers'][
'content-length'] = 0;
694 $logHeaders = $req[
'headers'];
695 foreach ( $logHeaders as
$header => $value ) {
696 if ( preg_match( self::SENSITIVE_HEADERS,
$header ) === 1 ) {
697 $logHeaders[
$header] =
'[redacted]';
700 $this->logger->debug(
"HTTP start: {method} {url}",
702 'method' => $req[
'method'],
703 'url' => $req[
'url'],
704 'headers' => $logHeaders,
707 $req[
'flags'] ??= [];
711 private function useReverseProxy( array &$req, $proxy ) {
712 $parsedProxy = parse_url( $proxy );
713 if ( $parsedProxy ===
false ) {
714 throw new InvalidArgumentException(
"Invalid reverseProxy configured: $proxy" );
716 $parsedUrl = parse_url( $req[
'url'] );
717 if ( $parsedUrl ===
false ) {
718 throw new InvalidArgumentException(
"Invalid url specified: {$req['url']}" );
722 $req[
'headers'][
'Host'] = $parsedUrl[
'host'];
725 $parsedUrl[
'scheme'] = $parsedProxy[
'scheme'];
727 $parsedUrl[
'host'] = $parsedProxy[
'host'];
728 if ( isset( $parsedProxy[
'port'] ) ) {
729 $parsedUrl[
'port'] = $parsedProxy[
'port'];
731 unset( $parsedUrl[
'port'] );
733 $req[
'url'] = self::assembleUrl( $parsedUrl );
736 $req[
'proxy'] =
false;
749 private static function assembleUrl( array $urlParts ): string {
750 $result = isset( $urlParts[
'scheme'] ) ? $urlParts[
'scheme'] .
'://' :
'';
752 if ( isset( $urlParts[
'host'] ) ) {
753 if ( isset( $urlParts[
'user'] ) ) {
754 $result .= $urlParts[
'user'];
755 if ( isset( $urlParts[
'pass'] ) ) {
756 $result .=
':' . $urlParts[
'pass'];
761 $result .= $urlParts[
'host'];
763 if ( isset( $urlParts[
'port'] ) ) {
764 $result .=
':' . $urlParts[
'port'];
768 if ( isset( $urlParts[
'path'] ) ) {
769 $result .= $urlParts[
'path'];
772 if ( isset( $urlParts[
'query'] ) && $urlParts[
'query'] !==
'' ) {
773 $result .=
'?' . $urlParts[
'query'];
776 if ( isset( $urlParts[
'fragment'] ) ) {
777 $result .=
'#' . $urlParts[
'fragment'];
790 private function isLocalURL(
$url ) {
791 if ( !$this->localVirtualHosts ) {
798 if ( preg_match(
'!^https?://([\w.-]+)[/:].*$!',
$url,
$matches ) ) {
801 $domainParts = explode(
'.', $host );
803 $domainParts = array_reverse( $domainParts );
806 $countParts = count( $domainParts );
807 for ( $i = 0; $i < $countParts; $i++ ) {
808 $domainPart = $domainParts[$i];
810 $domain = $domainPart;
812 $domain = $domainPart .
'.' . $domain;
815 if ( in_array( $domain, $this->localVirtualHosts ) ) {
830 private function getSelectTimeout( $opts ) {
831 $connTimeout = $opts[
'connTimeout'] ?? $this->connTimeout;
832 $reqTimeout = $opts[
'reqTimeout'] ?? $this->reqTimeout;
833 $timeouts = array_filter( [ $connTimeout, $reqTimeout ] );
834 if ( count( $timeouts ) === 0 ) {
838 $selectTimeout = min( $timeouts ) * self::TIMEOUT_ACCURACY_FACTOR;
840 if ( $selectTimeout < 10e-6 ) {
841 $selectTimeout = 10e-6;
843 return $selectTimeout;
850 $this->logger = $logger;
855 curl_multi_close( $this->cmh );