SdkStreamHandler.php 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519
  1. <?php
  2. /**
  3. * Copyright (c) 2011-2018 Michael Dowling, https://github.com/mtdowling <mtdowling@gmail.com>
  4. * Permission is hereby granted, free of charge, to any person obtaining a copy
  5. * of this software and associated documentation files (the "Software"), to deal
  6. * in the Software without restriction, including without limitation the rights
  7. * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  8. * copies of the Software, and to permit persons to whom the Software is
  9. * furnished to do so, subject to the following conditions:
  10. * The above copyright notice and this permission notice shall be included in
  11. * all copies or substantial portions of the Software.
  12. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  13. * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  14. * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  15. * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  16. * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  17. * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  18. * THE SOFTWARE.
  19. */
  20. namespace Obs\Internal\Common;
  21. use GuzzleHttp\Exception\RequestException;
  22. use GuzzleHttp\Exception\ConnectException;
  23. use GuzzleHttp\Promise\FulfilledPromise;
  24. use GuzzleHttp\Promise\PromiseInterface;
  25. use GuzzleHttp\Psr7;
  26. use GuzzleHttp\TransferStats;
  27. use Psr\Http\Message\RequestInterface;
  28. use Psr\Http\Message\ResponseInterface;
  29. use Psr\Http\Message\StreamInterface;
  30. class SdkStreamHandler
  31. {
  32. private $lastHeaders = [];
  33. public function __invoke(RequestInterface $request, array $options)
  34. {
  35. if (isset($options['delay'])) {
  36. usleep($options['delay'] * 1000);
  37. }
  38. $startTime = isset($options['on_stats']) ? microtime(true) : null;
  39. try {
  40. $request = $request->withoutHeader('Expect');
  41. if (0 === $request->getBody()->getSize()) {
  42. $request = $request->withHeader('Content-Length', 0);
  43. }
  44. return $this->createResponse(
  45. $request,
  46. $options,
  47. $this->createStream($request, $options),
  48. $startTime
  49. );
  50. } catch (\InvalidArgumentException $e) {
  51. throw $e;
  52. } catch (\Exception $e) {
  53. $message = $e->getMessage();
  54. if (strpos($message, 'getaddrinfo')
  55. || strpos($message, 'Connection refused')
  56. || strpos($message, "couldn't connect to host")
  57. ) {
  58. $e = new ConnectException($e->getMessage(), $request, $e);
  59. }
  60. $e = RequestException::wrapException($request, $e);
  61. $this->invokeStats($options, $request, $startTime, null, $e);
  62. return \GuzzleHttp\Promise\rejection_for($e);
  63. }
  64. }
  65. private function invokeStats(
  66. array $options,
  67. RequestInterface $request,
  68. $startTime,
  69. ResponseInterface $response = null,
  70. $error = null
  71. ) {
  72. if (isset($options['on_stats'])) {
  73. $stats = new TransferStats(
  74. $request,
  75. $response,
  76. microtime(true) - $startTime,
  77. $error,
  78. []
  79. );
  80. call_user_func($options['on_stats'], $stats);
  81. }
  82. }
  83. private function createResponse(
  84. RequestInterface $request,
  85. array $options,
  86. $stream,
  87. $startTime
  88. ) {
  89. $hdrs = $this->lastHeaders;
  90. $this->lastHeaders = [];
  91. $parts = explode(' ', array_shift($hdrs), 3);
  92. $ver = explode('/', $parts[0])[1];
  93. $status = $parts[1];
  94. $reason = isset($parts[2]) ? $parts[2] : null;
  95. $headers = \GuzzleHttp\headers_from_lines($hdrs);
  96. list ($stream, $headers) = $this->checkDecode($options, $headers, $stream);
  97. try {
  98. $stream = Psr7\stream_for($stream);
  99. } catch (\Throwable $e) {
  100. $stream = Psr7\Utils::streamFor($stream);
  101. }
  102. $sink = $stream;
  103. if (strcasecmp('HEAD', $request->getMethod())) {
  104. $sink = $this->createSink($stream, $options);
  105. }
  106. $response = new Psr7\Response($status, $headers, $sink, $ver, $reason);
  107. if (isset($options['on_headers'])) {
  108. try {
  109. $options['on_headers']($response);
  110. } catch (\Exception $e) {
  111. $msg = 'An error was encountered during the on_headers event';
  112. $ex = new RequestException($msg, $request, $response, $e);
  113. return \GuzzleHttp\Promise\rejection_for($ex);
  114. }
  115. }
  116. if ($sink !== $stream) {
  117. $this->drain(
  118. $stream,
  119. $sink,
  120. $response->getHeaderLine('Content-Length')
  121. );
  122. }
  123. $this->invokeStats($options, $request, $startTime, $response, null);
  124. return new FulfilledPromise($response);
  125. }
  126. private function createSink(StreamInterface $stream, array $options)
  127. {
  128. if (!empty($options['stream'])) {
  129. return $stream;
  130. }
  131. $sink = isset($options['sink'])
  132. ? $options['sink']
  133. : fopen('php://temp', 'r+');
  134. if (is_string($sink)) {
  135. return new Psr7\LazyOpenStream($sink, 'w+');
  136. }
  137. try {
  138. return Psr7\stream_for($sink);
  139. } catch (\Throwable $e) {
  140. return Psr7\Utils::streamFor($sink);
  141. }
  142. }
  143. private function checkDecode(array $options, array $headers, $stream)
  144. {
  145. if (!empty($options['decode_content'])) {
  146. $normalizedKeys = \GuzzleHttp\normalize_header_keys($headers);
  147. if (isset($normalizedKeys['content-encoding'])) {
  148. $encoding = $headers[$normalizedKeys['content-encoding']];
  149. if ($encoding[0] === 'gzip' || $encoding[0] === 'deflate') {
  150. try {
  151. $stream = new Psr7\InflateStream(
  152. Psr7\stream_for($stream)
  153. );
  154. } catch (\Throwable $th) {
  155. $stream = new Psr7\InflateStream(
  156. Psr7\Utils::streamFor($stream)
  157. );
  158. }
  159. $headers['x-encoded-content-encoding']
  160. = $headers[$normalizedKeys['content-encoding']];
  161. unset($headers[$normalizedKeys['content-encoding']]);
  162. if (isset($normalizedKeys['content-length'])) {
  163. $headers['x-encoded-content-length']
  164. = $headers[$normalizedKeys['content-length']];
  165. $length = (int) $stream->getSize();
  166. if ($length === 0) {
  167. unset($headers[$normalizedKeys['content-length']]);
  168. } else {
  169. $headers[$normalizedKeys['content-length']] = [$length];
  170. }
  171. }
  172. }
  173. }
  174. }
  175. return [$stream, $headers];
  176. }
  177. private function drain(
  178. StreamInterface $source,
  179. StreamInterface $sink,
  180. $contentLength
  181. ) {
  182. Psr7\copy_to_stream(
  183. $source,
  184. $sink,
  185. (strlen($contentLength) > 0 && (int) $contentLength > 0) ? (int) $contentLength : -1
  186. );
  187. $sink->seek(0);
  188. $source->close();
  189. return $sink;
  190. }
  191. private function createResource(callable $callback)
  192. {
  193. $errors = null;
  194. set_error_handler(function ($_, $msg, $file, $line) use (&$errors) {
  195. $errors[] = [
  196. 'message' => $msg,
  197. 'file' => $file,
  198. 'line' => $line
  199. ];
  200. return true;
  201. });
  202. $resource = $callback();
  203. restore_error_handler();
  204. if (!$resource) {
  205. $message = 'Error creating resource: ';
  206. foreach ($errors as $err) {
  207. foreach ($err as $key => $value) {
  208. $message .= "[$key] $value" . PHP_EOL;
  209. }
  210. }
  211. throw new \RuntimeException(trim($message));
  212. }
  213. return $resource;
  214. }
  215. private function createStream(RequestInterface $request, array $options)
  216. {
  217. static $methods;
  218. if (!$methods) {
  219. $methods = array_flip(get_class_methods(__CLASS__));
  220. }
  221. if ($request->getProtocolVersion() == '1.1'
  222. && !$request->hasHeader('Connection')
  223. ) {
  224. $request = $request->withHeader('Connection', 'close');
  225. }
  226. if (!isset($options['verify'])) {
  227. $options['verify'] = true;
  228. }
  229. $params = [];
  230. $context = $this->getDefaultContext($request, $options);
  231. if (isset($options['on_headers']) && !is_callable($options['on_headers'])) {
  232. throw new \InvalidArgumentException('on_headers must be callable');
  233. }
  234. if (!empty($options)) {
  235. foreach ($options as $key => $value) {
  236. $method = "add_{$key}";
  237. if (isset($methods[$method])) {
  238. $this->{$method}($request, $context, $value, $params);
  239. }
  240. }
  241. }
  242. if (isset($options['stream_context'])) {
  243. if (!is_array($options['stream_context'])) {
  244. throw new \InvalidArgumentException('stream_context must be an array');
  245. }
  246. $context = array_replace_recursive(
  247. $context,
  248. $options['stream_context']
  249. );
  250. }
  251. if (isset($options['auth'])
  252. && is_array($options['auth'])
  253. && isset($options['auth'][2])
  254. && 'ntlm' == $options['auth'][2]
  255. ) {
  256. throw new \InvalidArgumentException('Microsoft NTLM authentication only supported with curl handler');
  257. }
  258. $uri = $this->resolveHost($request, $options);
  259. $context = $this->createResource(
  260. function () use ($context, $params) {
  261. return stream_context_create($context, $params);
  262. }
  263. );
  264. return $this->createResource(
  265. function () use ($uri, &$http_response_header, $context, $options) {
  266. $resource = fopen((string) $uri, 'r', null, $context);
  267. $this->lastHeaders = $http_response_header;
  268. if (isset($options['read_timeout'])) {
  269. $readTimeout = $options['read_timeout'];
  270. $sec = (int) $readTimeout;
  271. $usec = ($readTimeout - $sec) * 100000;
  272. stream_set_timeout($resource, $sec, $usec);
  273. }
  274. return $resource;
  275. }
  276. );
  277. }
  278. private function resolveHost(RequestInterface $request, array $options)
  279. {
  280. $uri = $request->getUri();
  281. if (isset($options['force_ip_resolve']) && !filter_var($uri->getHost(), FILTER_VALIDATE_IP)) {
  282. if ('v4' === $options['force_ip_resolve']) {
  283. $records = dns_get_record($uri->getHost(), DNS_A);
  284. if (!isset($records[0]['ip'])) {
  285. throw new ConnectException(sprintf("Could not resolve IPv4 address for host '%s'", $uri->getHost()), $request);
  286. }
  287. $uri = $uri->withHost($records[0]['ip']);
  288. } elseif ('v6' === $options['force_ip_resolve']) {
  289. $records = dns_get_record($uri->getHost(), DNS_AAAA);
  290. if (!isset($records[0]['ipv6'])) {
  291. throw new ConnectException(sprintf("Could not resolve IPv6 address for host '%s'", $uri->getHost()), $request);
  292. }
  293. $uri = $uri->withHost('[' . $records[0]['ipv6'] . ']');
  294. }
  295. }
  296. return $uri;
  297. }
  298. private function getDefaultContext(RequestInterface $request)
  299. {
  300. $headers = '';
  301. foreach ($request->getHeaders() as $name => $value) {
  302. foreach ($value as $val) {
  303. $headers .= "$name: $val\r\n";
  304. }
  305. }
  306. $context = [
  307. 'http' => [
  308. 'method' => $request->getMethod(),
  309. 'header' => $headers,
  310. 'protocol_version' => $request->getProtocolVersion(),
  311. 'ignore_errors' => true,
  312. 'follow_location' => 0,
  313. ],
  314. ];
  315. $body = (string) $request->getBody();
  316. if (!empty($body)) {
  317. $context['http']['content'] = $body;
  318. if (!$request->hasHeader('Content-Type')) {
  319. $context['http']['header'] .= "Content-Type:\r\n";
  320. }
  321. }
  322. $context['http']['header'] = rtrim($context['http']['header']);
  323. return $context;
  324. }
  325. private function add_proxy(RequestInterface $request, &$options, $value, &$params)
  326. {
  327. if (!is_array($value)) {
  328. $options['http']['proxy'] = $value;
  329. } else {
  330. $scheme = $request->getUri()->getScheme();
  331. if (isset($value[$scheme])) {
  332. if (!isset($value['no'])
  333. || !\GuzzleHttp\is_host_in_noproxy(
  334. $request->getUri()->getHost(),
  335. $value['no']
  336. )
  337. ) {
  338. $options['http']['proxy'] = $value[$scheme];
  339. }
  340. }
  341. }
  342. }
  343. private function add_timeout(RequestInterface $request, &$options, $value, &$params)
  344. {
  345. if ($value > 0) {
  346. $options['http']['timeout'] = $value;
  347. }
  348. }
  349. private function add_verify(RequestInterface $request, &$options, $value, &$params)
  350. {
  351. if ($value === true) {
  352. if (PHP_VERSION_ID < 50600) {
  353. $options['ssl']['cafile'] = \GuzzleHttp\default_ca_bundle();
  354. }
  355. } elseif (is_string($value)) {
  356. $options['ssl']['cafile'] = $value;
  357. if (!file_exists($value)) {
  358. throw new \RuntimeException("SSL CA bundle not found: $value");
  359. }
  360. } elseif ($value === false) {
  361. $options['ssl']['verify_peer'] = false;
  362. $options['ssl']['verify_peer_name'] = false;
  363. return;
  364. } else {
  365. throw new \InvalidArgumentException('Invalid verify request option');
  366. }
  367. $options['ssl']['verify_peer'] = true;
  368. $options['ssl']['verify_peer_name'] = true;
  369. $options['ssl']['allow_self_signed'] = false;
  370. }
  371. private function add_cert(RequestInterface $request, &$options, $value, &$params)
  372. {
  373. if (is_array($value)) {
  374. $options['ssl']['passphrase'] = $value[1];
  375. $value = $value[0];
  376. }
  377. if (!file_exists($value)) {
  378. throw new \RuntimeException("SSL certificate not found: {$value}");
  379. }
  380. $options['ssl']['local_cert'] = $value;
  381. }
  382. private function add_progress(RequestInterface $request, &$options, $value, &$params)
  383. {
  384. $this->addNotification(
  385. $params,
  386. function ($code, $a, $b, $c, $transferred, $total) use ($value) {
  387. if ($code == STREAM_NOTIFY_PROGRESS) {
  388. $value($total, $transferred, null, null);
  389. }
  390. }
  391. );
  392. }
  393. private function add_debug(RequestInterface $request, &$options, $value, &$params)
  394. {
  395. if ($value === false) {
  396. return;
  397. }
  398. static $map = [
  399. STREAM_NOTIFY_CONNECT => 'CONNECT',
  400. STREAM_NOTIFY_AUTH_REQUIRED => 'AUTH_REQUIRED',
  401. STREAM_NOTIFY_AUTH_RESULT => 'AUTH_RESULT',
  402. STREAM_NOTIFY_MIME_TYPE_IS => 'MIME_TYPE_IS',
  403. STREAM_NOTIFY_FILE_SIZE_IS => 'FILE_SIZE_IS',
  404. STREAM_NOTIFY_REDIRECTED => 'REDIRECTED',
  405. STREAM_NOTIFY_PROGRESS => 'PROGRESS',
  406. STREAM_NOTIFY_FAILURE => 'FAILURE',
  407. STREAM_NOTIFY_COMPLETED => 'COMPLETED',
  408. STREAM_NOTIFY_RESOLVE => 'RESOLVE',
  409. ];
  410. static $args = ['severity', 'message', 'message_code',
  411. 'bytes_transferred', 'bytes_max'];
  412. $value = \GuzzleHttp\debug_resource($value);
  413. $ident = $request->getMethod() . ' ' . $request->getUri()->withFragment('');
  414. $this->addNotification(
  415. $params,
  416. function () use ($ident, $value, $map, $args) {
  417. $passed = func_get_args();
  418. $code = array_shift($passed);
  419. fprintf($value, '<%s> [%s] ', $ident, $map[$code]);
  420. foreach (array_filter($passed) as $i => $v) {
  421. fwrite($value, $args[$i] . ': "' . $v . '" ');
  422. }
  423. fwrite($value, "\n");
  424. }
  425. );
  426. }
  427. private function addNotification(array &$params, callable $notify)
  428. {
  429. if (!isset($params['notification'])) {
  430. $params['notification'] = $notify;
  431. } else {
  432. $params['notification'] = $this->callArray([
  433. $params['notification'],
  434. $notify
  435. ]);
  436. }
  437. }
  438. private function callArray(array $functions)
  439. {
  440. return function () use ($functions) {
  441. $args = func_get_args();
  442. foreach ($functions as $fn) {
  443. call_user_func_array($fn, $args);
  444. }
  445. };
  446. }
  447. }