61 private const SENSITIVE_HEADERS =
'/(^|-|_)(authorization|auth|password|cookie)($|-|_)/';
88 protected $userAgent =
'wikimedia/multi-http-client v1.1';
96 private const TIMEOUT_ACCURACY_FACTOR = 0.1;
123 if ( isset( $options[
'caBundlePath'] ) ) {
124 $this->caBundlePath = $options[
'caBundlePath'];
125 if ( !file_exists( $this->caBundlePath ) ) {
126 throw new InvalidArgumentException(
"Cannot find CA bundle: " . $this->caBundlePath );
130 'connTimeout',
'maxConnTimeout',
'reqTimeout',
'maxReqTimeout',
131 'usePipelining',
'maxConnsPerHost',
'proxy',
'userAgent',
'logger',
132 'localProxy',
'localVirtualHosts',
'headers',
'telemetry'
134 foreach ( $opts as $key ) {
135 if ( isset( $options[$key] ) ) {
136 $this->$key = $options[$key];
139 $this->logger ??=
new NullLogger;
166 public function run( array $req, array $opts = [],
string $caller = __METHOD__ ) {
167 return $this->
runMulti( [ $req ], $opts, $caller )[0][
'response'];
202 public function runMulti( array $reqs, array $opts = [],
string $caller = __METHOD__ ) {
203 $this->normalizeRequests( $reqs );
206 if ( $this->maxConnTimeout && $opts[
'connTimeout'] > $this->maxConnTimeout ) {
209 if ( $this->maxReqTimeout && $opts[
'reqTimeout'] > $this->maxReqTimeout ) {
214 switch ( $opts[
'httpVersion'] ??
null ) {
216 $opts[
'httpVersion'] = CURL_HTTP_VERSION_1_0;
219 $opts[
'httpVersion'] = CURL_HTTP_VERSION_1_1;
223 $opts[
'httpVersion'] = CURL_HTTP_VERSION_2_0;
226 $opts[
'httpVersion'] = CURL_HTTP_VERSION_NONE;
228 return $this->runMultiCurl( $reqs, $opts, $caller );
230 # TODO: Add handling for httpVersion option
231 return $this->runMultiHttp( $reqs, $opts );
243 return extension_loaded(
'curl' ) && function_exists(
'curl_multi_init' );
264 private function runMultiCurl( array $reqs, array $opts,
string $caller = __METHOD__ ) {
267 $selectTimeout = $this->getSelectTimeout( $opts );
271 foreach ( $reqs as $index => &$req ) {
273 curl_multi_add_handle( $chm, $handles[$index] );
282 $mrc = curl_multi_exec( $chm, $active );
284 if ( $mrc !== CURLM_OK ) {
285 $error = curl_multi_strerror( $mrc );
286 $this->logger->error(
'curl_multi_exec() failed: {error}', [
288 'exception' =>
new RuntimeException(),
295 if ( $active > 0 && curl_multi_select( $chm, $selectTimeout ) === -1 ) {
296 $errno = curl_multi_errno( $chm );
297 $error = curl_multi_strerror( $errno );
298 $this->logger->error(
'curl_multi_select() failed: {error}', [
300 'exception' =>
new RuntimeException(),
304 }
while ( $active > 0 );
306 $queuedMessages =
null;
308 $info = curl_multi_info_read( $chm, $queuedMessages );
309 if ( $info !==
false && $info[
'msg'] === CURLMSG_DONE ) {
313 $infos[(int)$info[
'handle']] = $info;
315 }
while ( $queuedMessages > 0 );
318 foreach ( $reqs as $index => &$req ) {
319 $ch = $handles[$index];
320 curl_multi_remove_handle( $chm, $ch );
322 if ( isset( $infos[(
int)$ch] ) ) {
323 $info = $infos[(int)$ch];
324 $errno = $info[
'result'];
325 if ( $errno !== 0 ) {
326 $req[
'response'][
'error'] =
"(curl error: $errno)";
327 if ( function_exists(
'curl_strerror' ) ) {
328 $req[
'response'][
'error'] .=
" " . curl_strerror( $errno );
330 $this->logger->error(
'Error fetching URL "{url}": {error}', [
331 'url' => $req[
'url'],
332 'error' => $req[
'response'][
'error'],
333 'exception' =>
new RuntimeException(),
337 $this->logger->debug(
338 "HTTP complete: {method} {url} code={response_code} size={size} " .
339 "total={total_time} connect={connect_time}",
341 'method' => $req[
'method'],
342 'url' => $req[
'url'],
343 'response_code' => $req[
'response'][
'code'],
344 'size' => curl_getinfo( $ch, CURLINFO_SIZE_DOWNLOAD ),
345 'total_time' => $this->getCurlTime(
346 $ch, CURLINFO_TOTAL_TIME,
'CURLINFO_TOTAL_TIME_T'
348 'connect_time' => $this->getCurlTime(
349 $ch, CURLINFO_CONNECT_TIME,
'CURLINFO_CONNECT_TIME_T'
355 $req[
'response'][
'error'] =
"(curl error: no status set)";
359 $req[
'response'][0] = $req[
'response'][
'code'];
360 $req[
'response'][1] = $req[
'response'][
'reason'];
361 $req[
'response'][2] = $req[
'response'][
'headers'];
362 $req[
'response'][3] = $req[
'response'][
'body'];
363 $req[
'response'][4] = $req[
'response'][
'error'];
366 if ( isset( $req[
'_closeHandle'] ) ) {
367 fclose( $req[
'_closeHandle'] );
368 unset( $req[
'_closeHandle'] );
391 curl_setopt( $ch, CURLOPT_PROXY, $req[
'proxy'] ?? $this->proxy );
392 curl_setopt( $ch, CURLOPT_CONNECTTIMEOUT_MS, intval( $opts[
'connTimeout'] * 1e3 ) );
393 curl_setopt( $ch, CURLOPT_TIMEOUT_MS, intval( $opts[
'reqTimeout'] * 1e3 ) );
394 curl_setopt( $ch, CURLOPT_FOLLOWLOCATION, 1 );
395 curl_setopt( $ch, CURLOPT_MAXREDIRS, 4 );
396 curl_setopt( $ch, CURLOPT_HEADER, 0 );
397 if ( $this->caBundlePath !==
null ) {
398 curl_setopt( $ch, CURLOPT_SSL_VERIFYPEER,
true );
399 curl_setopt( $ch, CURLOPT_CAINFO, $this->caBundlePath );
401 curl_setopt( $ch, CURLOPT_RETURNTRANSFER, 1 );
404 $query = http_build_query( $req[
'query'],
'',
'&', PHP_QUERY_RFC3986 );
405 if ( $query !=
'' ) {
406 $url .= strpos( $req[
'url'],
'?' ) ===
false ?
"?$query" :
"&$query";
408 curl_setopt( $ch, CURLOPT_URL,
$url );
409 curl_setopt( $ch, CURLOPT_CUSTOMREQUEST, $req[
'method'] );
410 curl_setopt( $ch, CURLOPT_NOBODY, ( $req[
'method'] ===
'HEAD' ) );
411 curl_setopt( $ch, CURLOPT_HTTP_VERSION, $opts[
'httpVersion'] ?? CURL_HTTP_VERSION_NONE );
413 if ( $req[
'method'] ===
'PUT' ) {
414 curl_setopt( $ch, CURLOPT_PUT, 1 );
416 if ( is_resource( $req[
'body'] ) ) {
417 curl_setopt( $ch, CURLOPT_INFILE, $req[
'body'] );
418 if ( isset( $req[
'headers'][
'content-length'] ) ) {
419 curl_setopt( $ch, CURLOPT_INFILESIZE, $req[
'headers'][
'content-length'] );
420 } elseif ( isset( $req[
'headers'][
'transfer-encoding'] ) &&
421 $req[
'headers'][
'transfer-encoding'] ===
'chunks'
423 curl_setopt( $ch, CURLOPT_UPLOAD,
true );
425 throw new InvalidArgumentException(
"Missing 'Content-Length' or 'Transfer-Encoding' header." );
427 } elseif ( $req[
'body'] !==
'' ) {
428 $fp = fopen(
"php://temp",
"wb+" );
429 fwrite( $fp, $req[
'body'], strlen( $req[
'body'] ) );
431 curl_setopt( $ch, CURLOPT_INFILE, $fp );
432 curl_setopt( $ch, CURLOPT_INFILESIZE, strlen( $req[
'body'] ) );
433 $req[
'_closeHandle'] = $fp;
435 curl_setopt( $ch, CURLOPT_INFILESIZE, 0 );
437 curl_setopt( $ch, CURLOPT_READFUNCTION,
438 static function ( $ch, $fd, $length ) {
439 return (
string)fread( $fd, $length );
442 } elseif ( $req[
'method'] ===
'POST' ) {
443 curl_setopt( $ch, CURLOPT_POST, 1 );
444 curl_setopt( $ch, CURLOPT_POSTFIELDS, $req[
'body'] );
447 if ( is_resource( $req[
'body'] ) || $req[
'body'] !==
'' ) {
448 throw new InvalidArgumentException(
"HTTP body specified for a non PUT/POST request." );
450 $req[
'headers'][
'content-length'] = 0;
453 if ( !isset( $req[
'headers'][
'user-agent'] ) ) {
458 foreach ( $req[
'headers'] as $name => $value ) {
459 if ( strpos( $name,
':' ) !==
false ) {
460 throw new InvalidArgumentException(
"Header name must not contain colon-space." );
462 $headers[] = $name .
': ' . trim( $value );
464 curl_setopt( $ch, CURLOPT_HTTPHEADER,
$headers );
466 curl_setopt( $ch, CURLOPT_HEADERFUNCTION,
467 static function ( $ch, $header ) use ( &$req ) {
468 if ( !empty( $req[
'flags'][
'relayResponseHeaders'] ) && trim( $header ) !==
'' ) {
471 $length = strlen( $header );
473 if ( preg_match(
"/^(HTTP\/(?:1\.[01]|2)) (\d{3}) (.*)/", $header,
$matches ) ) {
474 $req[
'response'][
'code'] = (int)
$matches[2];
475 $req[
'response'][
'reason'] = trim(
$matches[3] );
478 $req[
'response'][
'headers'] = [];
481 if ( strpos( $header,
":" ) ===
false ) {
484 [ $name, $value ] = explode(
":", $header, 2 );
485 $name = strtolower( $name );
486 $value = trim( $value );
487 if ( isset( $req[
'response'][
'headers'][$name] ) ) {
488 $req[
'response'][
'headers'][$name] .=
', ' . $value;
490 $req[
'response'][
'headers'][$name] = $value;
497 $hasOutputStream = isset( $req[
'stream'] );
498 curl_setopt( $ch, CURLOPT_WRITEFUNCTION,
499 static function ( $ch, $data ) use ( &$req, $hasOutputStream ) {
500 if ( $hasOutputStream ) {
502 return fwrite( $req[
'stream'], $data );
505 $req[
'response'][
'body'] .= $data;
507 return strlen( $data );
523 $cmh = curl_multi_init();
526 curl_multi_setopt(
$cmh, CURLMOPT_MAXCONNECTS, (
int)$this->maxConnsPerHost );
530 $curlVersion = curl_version()[
'version'];
533 if ( version_compare( $curlVersion,
'7.30.0',
'>=' ) ) {
536 curl_multi_setopt( $this->cmh, CURLMOPT_MAX_HOST_CONNECTIONS, (
int)$maxHostConns );
539 if ( $opts[
'usePipelining'] ?? $this->usePipelining ) {
540 if ( version_compare( $curlVersion,
'7.43',
'<' ) ) {
543 } elseif ( version_compare( $curlVersion,
'7.62',
'<' ) ) {
545 $pipelining = CURLPIPE_HTTP1 | CURLPIPE_MULTIPLEX;
548 $pipelining = CURLPIPE_MULTIPLEX;
552 @curl_multi_setopt( $this->cmh, CURLMOPT_PIPELINING, $pipelining );
568 private function getCurlTime( $ch, $oldOption, $newConstName ): string {
569 if ( defined( $newConstName ) ) {
570 return sprintf(
"%.6F", curl_getinfo( $ch, constant( $newConstName ) ) / 1e6 );
572 return (
string)curl_getinfo( $ch, $oldOption );
591 private function runMultiHttp( array $reqs, array $opts = [] ) {
593 'timeout' => $opts[
'reqTimeout'] ?? $this->reqTimeout,
594 'connectTimeout' => $opts[
'connTimeout'] ?? $this->connTimeout,
595 'logger' => $this->logger,
596 'caInfo' => $this->caBundlePath,
598 foreach ( $reqs as &$req ) {
599 $reqOptions = $httpOptions + [
600 'method' => $req[
'method'],
601 'proxy' => $req[
'proxy'] ?? $this->proxy,
602 'userAgent' => $req[
'headers'][
'user-agent'] ?? $this->userAgent,
603 'postData' => $req[
'body'],
607 $query = http_build_query( $req[
'query'],
'',
'&', PHP_QUERY_RFC3986 );
608 if ( $query !=
'' ) {
609 $url .= strpos( $req[
'url'],
'?' ) ===
false ?
"?$query" :
"&$query";
612 $httpRequest = MediaWikiServices::getInstance()->getHttpRequestFactory()->create(
613 $url, $reqOptions, __METHOD__ );
614 $httpRequest->setLogger( $this->logger );
615 foreach ( $req[
'headers'] as $header => $value ) {
616 $httpRequest->setHeader( $header, $value );
618 $sv = $httpRequest->execute()->getStatusValue();
620 $respHeaders = array_map(
621 static function ( $v ) {
622 return implode(
', ', $v );
624 $httpRequest->getResponseHeaders() );
627 'code' => $httpRequest->getStatus(),
629 'headers' => $respHeaders,
630 'body' => $httpRequest->getContent(),
634 if ( !$sv->isOK() ) {
635 $svErrors = $sv->getErrors();
636 if ( isset( $svErrors[0] ) ) {
637 $req[
'response'][
'error'] = $svErrors[0][
'message'];
640 if ( isset( $svErrors[0][
'params'][0] ) ) {
641 if ( is_numeric( $svErrors[0][
'params'][0] ) ) {
642 if ( isset( $svErrors[0][
'params'][1] ) ) {
644 $req[
'response'][
'reason'] = $svErrors[0][
'params'][1];
647 $req[
'response'][
'reason'] = $svErrors[0][
'params'][0];
653 $req[
'response'][0] = $req[
'response'][
'code'];
654 $req[
'response'][1] = $req[
'response'][
'reason'];
655 $req[
'response'][2] = $req[
'response'][
'headers'];
656 $req[
'response'][3] = $req[
'response'][
'body'];
657 $req[
'response'][4] = $req[
'response'][
'error'];
668 private function normalizeHeaders( array $headers ): array {
670 foreach ( $headers as $name => $value ) {
671 $normalized[strtolower( $name )] = $value;
681 private function normalizeRequests( array &$reqs ) {
682 foreach ( $reqs as &$req ) {
690 if ( isset( $req[0] ) ) {
691 $req[
'method'] = $req[0];
694 if ( isset( $req[1] ) ) {
695 $req[
'url'] = $req[1];
698 if ( !isset( $req[
'method'] ) ) {
699 throw new InvalidArgumentException(
"Request has no 'method' field set." );
700 } elseif ( !isset( $req[
'url'] ) ) {
701 throw new InvalidArgumentException(
"Request has no 'url' field set." );
703 if ( $this->localProxy !==
false && $this->isLocalURL( $req[
'url'] ) ) {
704 $this->useReverseProxy( $req, $this->localProxy );
706 $req[
'query'] ??= [];
707 $req[
'headers'] = $this->normalizeHeaders(
710 $this->telemetry ? $this->telemetry->getRequestHeaders() : [],
711 $req[
'headers'] ?? []
715 if ( !isset( $req[
'body'] ) ) {
717 $req[
'headers'][
'content-length'] = 0;
720 $logHeaders = $req[
'headers'];
721 foreach ( $logHeaders as $header => $value ) {
722 if ( preg_match( self::SENSITIVE_HEADERS, $header ) === 1 ) {
723 $logHeaders[$header] =
'[redacted]';
726 $this->logger->debug(
"HTTP start: {method} {url}",
728 'method' => $req[
'method'],
729 'url' => $req[
'url'],
730 'headers' => $logHeaders,
733 $req[
'flags'] ??= [];
737 private function useReverseProxy( array &$req,
string $proxy ) {
738 $parsedProxy = parse_url( $proxy );
739 if ( $parsedProxy ===
false ) {
740 throw new InvalidArgumentException(
"Invalid reverseProxy configured: $proxy" );
742 $parsedUrl = parse_url( $req[
'url'] );
743 if ( $parsedUrl ===
false ) {
744 throw new InvalidArgumentException(
"Invalid url specified: {$req['url']}" );
748 $req[
'headers'][
'Host'] = $parsedUrl[
'host'];
751 $parsedUrl[
'scheme'] = $parsedProxy[
'scheme'];
753 $parsedUrl[
'host'] = $parsedProxy[
'host'];
754 if ( isset( $parsedProxy[
'port'] ) ) {
755 $parsedUrl[
'port'] = $parsedProxy[
'port'];
757 unset( $parsedUrl[
'port'] );
759 $req[
'url'] = self::assembleUrl( $parsedUrl );
762 $req[
'proxy'] =
false;
775 private static function assembleUrl( array $urlParts ): string {
776 $result = isset( $urlParts[
'scheme'] ) ? $urlParts[
'scheme'] .
'://' :
'';
778 if ( isset( $urlParts[
'host'] ) ) {
779 if ( isset( $urlParts[
'user'] ) ) {
780 $result .= $urlParts[
'user'];
781 if ( isset( $urlParts[
'pass'] ) ) {
782 $result .=
':' . $urlParts[
'pass'];
787 $result .= $urlParts[
'host'];
789 if ( isset( $urlParts[
'port'] ) ) {
790 $result .=
':' . $urlParts[
'port'];
794 if ( isset( $urlParts[
'path'] ) ) {
795 $result .= $urlParts[
'path'];
798 if ( isset( $urlParts[
'query'] ) && $urlParts[
'query'] !==
'' ) {
799 $result .=
'?' . $urlParts[
'query'];
802 if ( isset( $urlParts[
'fragment'] ) ) {
803 $result .=
'#' . $urlParts[
'fragment'];
816 private function isLocalURL(
$url ) {
817 if ( !$this->localVirtualHosts ) {
824 if ( preg_match(
'!^https?://([\w.-]+)[/:].*$!',
$url,
$matches ) ) {
827 $domainParts = explode(
'.', $host );
829 $domainParts = array_reverse( $domainParts );
832 $countParts = count( $domainParts );
833 for ( $i = 0; $i < $countParts; $i++ ) {
834 $domainPart = $domainParts[$i];
836 $domain = $domainPart;
838 $domain = $domainPart .
'.' . $domain;
841 if ( in_array( $domain, $this->localVirtualHosts ) ) {
856 private function getSelectTimeout( $opts ) {
857 $connTimeout = $opts[
'connTimeout'] ?? $this->connTimeout;
858 $reqTimeout = $opts[
'reqTimeout'] ?? $this->reqTimeout;
859 $timeouts = array_filter( [ $connTimeout, $reqTimeout ] );
860 if ( count( $timeouts ) === 0 ) {
864 $selectTimeout = min( $timeouts ) * self::TIMEOUT_ACCURACY_FACTOR;
866 if ( $selectTimeout < 10e-6 ) {
867 $selectTimeout = 10e-6;
869 return $selectTimeout;
875 public function setLogger( LoggerInterface $logger ): void {
876 $this->logger = $logger;
881 curl_multi_close( $this->cmh );