Merge pull request repejota#122 from tomponline/develop

Adds TLS support

Author: repejota

src/Nats/Connection.php

@@ -229,10 +229,10 @@ public function isConnected()
      * @param string $address Server url string.
      * @param float  $timeout Number of seconds until the connect() system call should timeout.
      *
-     * @return resource
      * @throws \Exception Exception raised if connection fails.
+     * @return resource
      */
-    private function getStream($address, $timeout)
+    private function getStream($address, $timeout, $context)
     {
         $errno  = null;
         $errstr = null;
@@ -242,7 +242,8 @@ function () {
                 return true;
             }
         );
-        $fp = stream_socket_client($address, $errno, $errstr, $timeout, STREAM_CLIENT_CONNECT);
+
+        $fp = stream_socket_client($address, $errno, $errstr, $timeout, STREAM_CLIENT_CONNECT, $context);
         restore_error_handler();
 
         if ($fp === false) {
@@ -315,8 +316,8 @@ public function __construct(ConnectionOptions $options = null)
      *
      * @param string $payload Message data.
      *
-     * @return void
      * @throws \Exception Raises if fails sending data.
+     * @return void
      */
     private function send($payload)
     {
@@ -394,8 +395,8 @@ private function handlePING()
      *
      * @param string $line Message command from Nats.
      *
-     * @return             void
      * @throws             Exception If subscription not found.
+     * @return             void
      * @codeCoverageIgnore
      */
     private function handleMSG($line)
@@ -408,7 +409,7 @@ private function handleMSG($line)
         if (count($parts) === 5) {
             $length  = trim($parts[4]);
             $subject = $parts[3];
-        } else if (count($parts) === 4) {
+        } elseif (count($parts) === 4) {
             $length  = trim($parts[3]);
             $subject = $parts[1];
         }
@@ -443,19 +444,34 @@ public function connect($timeout = null)
         }
 
         $this->timeout      = $timeout;
-        $this->streamSocket = $this->getStream($this->options->getAddress(), $timeout);
+        $this->streamSocket = $this->getStream(
+            $this->options->getAddress(), $timeout, $this->options->getStreamContext());
         $this->setStreamTimeout($timeout);
 
-        $msg = 'CONNECT '.$this->options;
-        $this->send($msg);
-        $connectResponse = $this->receive();
+        $infoResponse = $this->receive();
 
-        if ($this->isErrorResponse($connectResponse) === true) {
-            throw Exception::forFailedConnection($connectResponse);
+        if ($this->isErrorResponse($infoResponse) === true) {
+            throw Exception::forFailedConnection($infoResponse);
         } else {
-            $this->processServerInfo($connectResponse);
+            $this->processServerInfo($infoResponse);
+            if ($this->serverInfo->isTLSRequired()) {
+                set_error_handler(
+                    function ($errno, $errstr, $errfile, $errline) {
+                        restore_error_handler();
+                        throw Exception::forFailedConnection($errstr);
+                    });
+
+                if (!stream_socket_enable_crypto(
+                        $this->streamSocket, true, STREAM_CRYPTO_METHOD_TLSv1_2_CLIENT)) {
+                    throw Exception::forFailedConnection('Error negotiating crypto');
+                }
+
+                restore_error_handler();
+            }
         }
 
+        $msg = 'CONNECT '.$this->options;
+        $this->send($msg);
         $this->ping();
         $pingResponse = $this->receive();
 
@@ -560,9 +576,9 @@ public function unsubscribe($sid, $quantity = null)
      * @param string $payload Message data.
      * @param string $inbox   Message inbox.
      *
+     * @throws Exception If subscription not found.
      * @return void
      *
-     * @throws Exception If subscription not found.
      */
     public function publish($subject, $payload = null, $inbox = null)
     {

src/Nats/ConnectionOptions.php

@@ -81,6 +81,13 @@ class ConnectionOptions
      */
     private $reconnect = true;
 
+    /**
+     * Stream context to use.
+     *
+     * @var resource
+     */
+    private $streamContext = null;
+
     /**
      * Allows to define parameters which can be set by passing them to the class constructor.
      *
@@ -97,6 +104,7 @@ class ConnectionOptions
                              'verbose',
                              'pedantic',
                              'reconnect',
+                             'streamContext',
                             ];
 
 
@@ -120,6 +128,9 @@ class ConnectionOptions
      */
     public function __construct($options = null)
     {
+        //Default stream context
+        $this->streamContext = stream_context_get_default();
+
         if (empty($options) === false) {
             $this->initialize($options);
         }
@@ -420,6 +431,30 @@ public function setReconnect($reconnect)
         return $this;
     }
 
+    /**
+     * Get stream context.
+     *
+     * @return resource
+     */
+    public function getStreamContext()
+    {
+        return $this->streamContext;
+    }
+
+    /**
+     * Set stream context.
+     *
+     * @param resource $streamContext Stream context.
+     *
+     * @return $this
+     */
+    public function setStreamContext($streamContext)
+    {
+        $this->streamContext = $streamContext;
+
+        return $this;
+    }
+
     /**
      * Set the connection options.
      *

src/Nats/ServerInfo.php

@@ -102,9 +102,9 @@ public function __construct($connectionResponse)
         $this->setPort($data['port']);
         $this->setVersion($data['version']);
         $this->setGoVersion($data['go']);
-        $this->setAuthRequired($data['auth_required']);
-        $this->setTLSRequired($data['tls_required']);
-        $this->setTLSVerify($data['tls_verify']);
+        $this->setAuthRequired(isset($data['auth_required']) ? $data['auth_required'] : false);
+        $this->setTLSRequired(isset($data['tls_required']) ? $data['tls_required'] : false);
+        $this->setTLSVerify(isset($data['tls_verify']) ? $data['tls_verify'] : false);
         $this->setMaxPayload($data['max_payload']);
 
         if (version_compare($data['version'], '1.1.0') === -1) {