StreamBuffer.php 9.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319
  1. <?php
  2. /*
  3. * This file is part of SwiftMailer.
  4. * (c) 2004-2009 Chris Corbyn
  5. *
  6. * For the full copyright and license information, please view the LICENSE
  7. * file that was distributed with this source code.
  8. */
  9. /**
  10. * A generic IoBuffer implementation supporting remote sockets and local processes.
  11. *
  12. * @author Chris Corbyn
  13. */
  14. class Swift_Transport_StreamBuffer extends Swift_ByteStream_AbstractFilterableInputStream implements Swift_Transport_IoBuffer
  15. {
  16. /** A primary socket */
  17. private $stream;
  18. /** The input stream */
  19. private $in;
  20. /** The output stream */
  21. private $out;
  22. /** Buffer initialization parameters */
  23. private $params = [];
  24. /** The ReplacementFilterFactory */
  25. private $replacementFactory;
  26. /** Translations performed on data being streamed into the buffer */
  27. private $translations = [];
  28. /**
  29. * Create a new StreamBuffer using $replacementFactory for transformations.
  30. */
  31. public function __construct(Swift_ReplacementFilterFactory $replacementFactory)
  32. {
  33. $this->replacementFactory = $replacementFactory;
  34. }
  35. /**
  36. * Perform any initialization needed, using the given $params.
  37. *
  38. * Parameters will vary depending upon the type of IoBuffer used.
  39. */
  40. public function initialize(array $params)
  41. {
  42. $this->params = $params;
  43. switch ($params['type']) {
  44. case self::TYPE_PROCESS:
  45. $this->establishProcessConnection();
  46. break;
  47. case self::TYPE_SOCKET:
  48. default:
  49. $this->establishSocketConnection();
  50. break;
  51. }
  52. }
  53. /**
  54. * Set an individual param on the buffer (e.g. switching to SSL).
  55. *
  56. * @param string $param
  57. * @param mixed $value
  58. */
  59. public function setParam($param, $value)
  60. {
  61. if (isset($this->stream)) {
  62. switch ($param) {
  63. case 'timeout':
  64. if ($this->stream) {
  65. stream_set_timeout($this->stream, $value);
  66. }
  67. break;
  68. case 'blocking':
  69. if ($this->stream) {
  70. stream_set_blocking($this->stream, 1);
  71. }
  72. }
  73. }
  74. $this->params[$param] = $value;
  75. }
  76. public function startTLS()
  77. {
  78. // STREAM_CRYPTO_METHOD_TLS_CLIENT only allow tls1.0 connections (some php versions)
  79. // To support modern tls we allow explicit tls1.0, tls1.1, tls1.2
  80. // Ssl3 and older are not allowed because they are vulnerable
  81. // @TODO make tls arguments configurable
  82. return stream_socket_enable_crypto($this->stream, true, STREAM_CRYPTO_METHOD_TLSv1_0_CLIENT | STREAM_CRYPTO_METHOD_TLSv1_1_CLIENT | STREAM_CRYPTO_METHOD_TLSv1_2_CLIENT);
  83. }
  84. /**
  85. * Perform any shutdown logic needed.
  86. */
  87. public function terminate()
  88. {
  89. if (isset($this->stream)) {
  90. switch ($this->params['type']) {
  91. case self::TYPE_PROCESS:
  92. fclose($this->in);
  93. fclose($this->out);
  94. proc_close($this->stream);
  95. break;
  96. case self::TYPE_SOCKET:
  97. default:
  98. fclose($this->stream);
  99. break;
  100. }
  101. }
  102. $this->stream = null;
  103. $this->out = null;
  104. $this->in = null;
  105. }
  106. /**
  107. * Set an array of string replacements which should be made on data written
  108. * to the buffer.
  109. *
  110. * This could replace LF with CRLF for example.
  111. *
  112. * @param string[] $replacements
  113. */
  114. public function setWriteTranslations(array $replacements)
  115. {
  116. foreach ($this->translations as $search => $replace) {
  117. if (!isset($replacements[$search])) {
  118. $this->removeFilter($search);
  119. unset($this->translations[$search]);
  120. }
  121. }
  122. foreach ($replacements as $search => $replace) {
  123. if (!isset($this->translations[$search])) {
  124. $this->addFilter(
  125. $this->replacementFactory->createFilter($search, $replace), $search
  126. );
  127. $this->translations[$search] = true;
  128. }
  129. }
  130. }
  131. /**
  132. * Get a line of output (including any CRLF).
  133. *
  134. * The $sequence number comes from any writes and may or may not be used
  135. * depending upon the implementation.
  136. *
  137. * @param int $sequence of last write to scan from
  138. *
  139. * @return string
  140. *
  141. * @throws Swift_IoException
  142. */
  143. public function readLine($sequence)
  144. {
  145. if (isset($this->out) && !feof($this->out)) {
  146. $line = fgets($this->out);
  147. if (0 == \strlen($line)) {
  148. $metas = stream_get_meta_data($this->out);
  149. if ($metas['timed_out']) {
  150. throw new Swift_IoException('Connection to '.$this->getReadConnectionDescription().' Timed Out');
  151. }
  152. }
  153. return $line;
  154. }
  155. }
  156. /**
  157. * Reads $length bytes from the stream into a string and moves the pointer
  158. * through the stream by $length.
  159. *
  160. * If less bytes exist than are requested the remaining bytes are given instead.
  161. * If no bytes are remaining at all, boolean false is returned.
  162. *
  163. * @param int $length
  164. *
  165. * @return string|bool
  166. *
  167. * @throws Swift_IoException
  168. */
  169. public function read($length)
  170. {
  171. if (isset($this->out) && !feof($this->out)) {
  172. $ret = fread($this->out, $length);
  173. if (0 == \strlen($ret)) {
  174. $metas = stream_get_meta_data($this->out);
  175. if ($metas['timed_out']) {
  176. throw new Swift_IoException('Connection to '.$this->getReadConnectionDescription().' Timed Out');
  177. }
  178. }
  179. return $ret;
  180. }
  181. }
  182. /** Not implemented */
  183. public function setReadPointer($byteOffset)
  184. {
  185. }
  186. /** Flush the stream contents */
  187. protected function flush()
  188. {
  189. if (isset($this->in)) {
  190. fflush($this->in);
  191. }
  192. }
  193. /** Write this bytes to the stream */
  194. protected function doCommit($bytes)
  195. {
  196. if (isset($this->in)) {
  197. $bytesToWrite = \strlen($bytes);
  198. $totalBytesWritten = 0;
  199. while ($totalBytesWritten < $bytesToWrite) {
  200. $bytesWritten = fwrite($this->in, substr($bytes, $totalBytesWritten));
  201. if (false === $bytesWritten || 0 === $bytesWritten) {
  202. break;
  203. }
  204. $totalBytesWritten += $bytesWritten;
  205. }
  206. if ($totalBytesWritten > 0) {
  207. return ++$this->sequence;
  208. }
  209. }
  210. }
  211. /**
  212. * Establishes a connection to a remote server.
  213. */
  214. private function establishSocketConnection()
  215. {
  216. $host = $this->params['host'];
  217. if (!empty($this->params['protocol'])) {
  218. $host = $this->params['protocol'].'://'.$host;
  219. }
  220. $timeout = 15;
  221. if (!empty($this->params['timeout'])) {
  222. $timeout = $this->params['timeout'];
  223. }
  224. $options = [];
  225. if (!empty($this->params['sourceIp'])) {
  226. $options['socket']['bindto'] = $this->params['sourceIp'].':0';
  227. }
  228. if (isset($this->params['stream_context_options'])) {
  229. $options = array_merge($options, $this->params['stream_context_options']);
  230. }
  231. $streamContext = stream_context_create($options);
  232. set_error_handler(function ($type, $msg) {
  233. throw new Swift_TransportException('Connection could not be established with host '.$this->params['host'].' :'.$msg);
  234. });
  235. try {
  236. $this->stream = stream_socket_client($host.':'.$this->params['port'], $errno, $errstr, $timeout, STREAM_CLIENT_CONNECT, $streamContext);
  237. } finally {
  238. restore_error_handler();
  239. }
  240. if (!empty($this->params['blocking'])) {
  241. stream_set_blocking($this->stream, 1);
  242. } else {
  243. stream_set_blocking($this->stream, 0);
  244. }
  245. stream_set_timeout($this->stream, $timeout);
  246. $this->in = &$this->stream;
  247. $this->out = &$this->stream;
  248. }
  249. /**
  250. * Opens a process for input/output.
  251. */
  252. private function establishProcessConnection()
  253. {
  254. $command = $this->params['command'];
  255. $descriptorSpec = [
  256. 0 => ['pipe', 'r'],
  257. 1 => ['pipe', 'w'],
  258. 2 => ['pipe', 'w'],
  259. ];
  260. $pipes = [];
  261. $this->stream = proc_open($command, $descriptorSpec, $pipes);
  262. stream_set_blocking($pipes[2], 0);
  263. if ($err = stream_get_contents($pipes[2])) {
  264. throw new Swift_TransportException('Process could not be started ['.$err.']');
  265. }
  266. $this->in = &$pipes[0];
  267. $this->out = &$pipes[1];
  268. }
  269. private function getReadConnectionDescription()
  270. {
  271. switch ($this->params['type']) {
  272. case self::TYPE_PROCESS:
  273. return 'Process '.$this->params['command'];
  274. break;
  275. case self::TYPE_SOCKET:
  276. default:
  277. $host = $this->params['host'];
  278. if (!empty($this->params['protocol'])) {
  279. $host = $this->params['protocol'].'://'.$host;
  280. }
  281. $host .= ':'.$this->params['port'];
  282. return $host;
  283. break;
  284. }
  285. }
  286. }