|
@@ -1,13 +1,16 @@
|
|
|
"""Zookeeper Protocol Connection Handler"""
|
|
"""Zookeeper Protocol Connection Handler"""
|
|
|
|
|
+from binascii import hexlify
|
|
|
|
|
+from contextlib import contextmanager
|
|
|
|
|
+import copy
|
|
|
import logging
|
|
import logging
|
|
|
-import os
|
|
|
|
|
import random
|
|
import random
|
|
|
import select
|
|
import select
|
|
|
import socket
|
|
import socket
|
|
|
|
|
+import ssl
|
|
|
import sys
|
|
import sys
|
|
|
import time
|
|
import time
|
|
|
-from binascii import hexlify
|
|
|
|
|
-from contextlib import contextmanager
|
|
|
|
|
|
|
+
|
|
|
|
|
+import six
|
|
|
|
|
|
|
|
from kazoo.exceptions import (
|
|
from kazoo.exceptions import (
|
|
|
AuthFailedError,
|
|
AuthFailedError,
|
|
@@ -15,9 +18,8 @@ from kazoo.exceptions import (
|
|
|
EXCEPTIONS,
|
|
EXCEPTIONS,
|
|
|
SessionExpiredError,
|
|
SessionExpiredError,
|
|
|
NoNodeError,
|
|
NoNodeError,
|
|
|
- SaslException
|
|
|
|
|
|
|
+ SASLException,
|
|
|
)
|
|
)
|
|
|
-from kazoo.handlers.utils import create_pipe
|
|
|
|
|
from kazoo.loggingsupport import BLATHER
|
|
from kazoo.loggingsupport import BLATHER
|
|
|
from kazoo.protocol.serialization import (
|
|
from kazoo.protocol.serialization import (
|
|
|
Auth,
|
|
Auth,
|
|
@@ -25,13 +27,14 @@ from kazoo.protocol.serialization import (
|
|
|
Connect,
|
|
Connect,
|
|
|
Exists,
|
|
Exists,
|
|
|
GetChildren,
|
|
GetChildren,
|
|
|
|
|
+ GetChildren2,
|
|
|
Ping,
|
|
Ping,
|
|
|
PingInstance,
|
|
PingInstance,
|
|
|
ReplyHeader,
|
|
ReplyHeader,
|
|
|
SASL,
|
|
SASL,
|
|
|
Transaction,
|
|
Transaction,
|
|
|
Watch,
|
|
Watch,
|
|
|
- int_struct
|
|
|
|
|
|
|
+ int_struct,
|
|
|
)
|
|
)
|
|
|
from kazoo.protocol.states import (
|
|
from kazoo.protocol.states import (
|
|
|
Callback,
|
|
Callback,
|
|
@@ -41,10 +44,16 @@ from kazoo.protocol.states import (
|
|
|
)
|
|
)
|
|
|
from kazoo.retry import (
|
|
from kazoo.retry import (
|
|
|
ForceRetryError,
|
|
ForceRetryError,
|
|
|
- RetryFailedError
|
|
|
|
|
|
|
+ RetryFailedError,
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
-import sasl
|
|
|
|
|
|
|
+try:
|
|
|
|
|
+ import puresasl
|
|
|
|
|
+ import puresasl.client
|
|
|
|
|
+ PURESASL_AVAILABLE = True
|
|
|
|
|
+except ImportError:
|
|
|
|
|
+ PURESASL_AVAILABLE = False
|
|
|
|
|
+
|
|
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
|
@@ -136,12 +145,11 @@ class RWServerAvailable(Exception):
|
|
|
|
|
|
|
|
class ConnectionHandler(object):
|
|
class ConnectionHandler(object):
|
|
|
"""Zookeeper connection handler"""
|
|
"""Zookeeper connection handler"""
|
|
|
- def __init__(self, client, retry_sleeper, logger=None, sasl_server_principal=None):
|
|
|
|
|
|
|
+ def __init__(self, client, retry_sleeper, logger=None, sasl_options=None):
|
|
|
self.client = client
|
|
self.client = client
|
|
|
self.handler = client.handler
|
|
self.handler = client.handler
|
|
|
self.retry_sleeper = retry_sleeper
|
|
self.retry_sleeper = retry_sleeper
|
|
|
self.logger = logger or log
|
|
self.logger = logger or log
|
|
|
- self.sasl_server_principal = sasl_server_principal
|
|
|
|
|
|
|
|
|
|
# Our event objects
|
|
# Our event objects
|
|
|
self.connection_closed = client.handler.event_object()
|
|
self.connection_closed = client.handler.event_object()
|
|
@@ -150,8 +158,8 @@ class ConnectionHandler(object):
|
|
|
self.connection_stopped.set()
|
|
self.connection_stopped.set()
|
|
|
self.ping_outstanding = client.handler.event_object()
|
|
self.ping_outstanding = client.handler.event_object()
|
|
|
|
|
|
|
|
- self._read_pipe = None
|
|
|
|
|
- self._write_pipe = None
|
|
|
|
|
|
|
+ self._read_sock = None
|
|
|
|
|
+ self._write_sock = None
|
|
|
|
|
|
|
|
self._socket = None
|
|
self._socket = None
|
|
|
self._xid = None
|
|
self._xid = None
|
|
@@ -160,6 +168,8 @@ class ConnectionHandler(object):
|
|
|
|
|
|
|
|
self._connection_routine = None
|
|
self._connection_routine = None
|
|
|
|
|
|
|
|
|
|
+ self.sasl_options = sasl_options
|
|
|
|
|
+ self.sasl_cli = None
|
|
|
|
|
|
|
|
# This is instance specific to avoid odd thread bug issues in Python
|
|
# This is instance specific to avoid odd thread bug issues in Python
|
|
|
# during shutdown global cleanup
|
|
# during shutdown global cleanup
|
|
@@ -174,7 +184,8 @@ class ConnectionHandler(object):
|
|
|
def start(self):
|
|
def start(self):
|
|
|
"""Start the connection up"""
|
|
"""Start the connection up"""
|
|
|
if self.connection_closed.is_set():
|
|
if self.connection_closed.is_set():
|
|
|
- self._read_pipe, self._write_pipe = create_pipe()
|
|
|
|
|
|
|
+ rw_sockets = self.handler.create_socket_pair()
|
|
|
|
|
+ self._read_sock, self._write_sock = rw_sockets
|
|
|
self.connection_closed.clear()
|
|
self.connection_closed.clear()
|
|
|
if self._connection_routine:
|
|
if self._connection_routine:
|
|
|
raise Exception("Unable to start, connection routine already "
|
|
raise Exception("Unable to start, connection routine already "
|
|
@@ -197,12 +208,12 @@ class ConnectionHandler(object):
|
|
|
if not self.connection_stopped.is_set():
|
|
if not self.connection_stopped.is_set():
|
|
|
raise Exception("Cannot close connection until it is stopped")
|
|
raise Exception("Cannot close connection until it is stopped")
|
|
|
self.connection_closed.set()
|
|
self.connection_closed.set()
|
|
|
- wp, rp = self._write_pipe, self._read_pipe
|
|
|
|
|
- self._write_pipe = self._read_pipe = None
|
|
|
|
|
- if wp is not None:
|
|
|
|
|
- os.close(wp)
|
|
|
|
|
- if rp is not None:
|
|
|
|
|
- os.close(rp)
|
|
|
|
|
|
|
+ ws, rs = self._write_sock, self._read_sock
|
|
|
|
|
+ self._write_sock = self._read_sock = None
|
|
|
|
|
+ if ws is not None:
|
|
|
|
|
+ ws.close()
|
|
|
|
|
+ if rs is not None:
|
|
|
|
|
+ rs.close()
|
|
|
|
|
|
|
|
def _server_pinger(self):
|
|
def _server_pinger(self):
|
|
|
"""Returns a server pinger iterable, that will ping the next
|
|
"""Returns a server pinger iterable, that will ping the next
|
|
@@ -222,13 +233,29 @@ class ConnectionHandler(object):
|
|
|
remaining = length
|
|
remaining = length
|
|
|
with self._socket_error_handling():
|
|
with self._socket_error_handling():
|
|
|
while remaining > 0:
|
|
while remaining > 0:
|
|
|
- s = self.handler.select([self._socket], [], [], timeout)[0]
|
|
|
|
|
- if not s: # pragma: nocover
|
|
|
|
|
- # If the read list is empty, we got a timeout. We don't
|
|
|
|
|
- # have to check wlist and xlist as we don't set any
|
|
|
|
|
- raise self.handler.timeout_exception("socket time-out")
|
|
|
|
|
-
|
|
|
|
|
- chunk = self._socket.recv(remaining)
|
|
|
|
|
|
|
+ # Because of SSL framing, a select may not return when using
|
|
|
|
|
+ # an SSL socket because the underlying physical socket may not
|
|
|
|
|
+ # have anything to select, but the wrapped object may still
|
|
|
|
|
+ # have something to read as it has previously gotten enough
|
|
|
|
|
+ # data from the underlying socket.
|
|
|
|
|
+ if (hasattr(self._socket, "pending") and
|
|
|
|
|
+ self._socket.pending() > 0):
|
|
|
|
|
+ pass
|
|
|
|
|
+ else:
|
|
|
|
|
+ s = self.handler.select([self._socket], [], [], timeout)[0]
|
|
|
|
|
+ if not s: # pragma: nocover
|
|
|
|
|
+ # If the read list is empty, we got a timeout. We don't
|
|
|
|
|
+ # have to check wlist and xlist as we don't set any
|
|
|
|
|
+ raise self.handler.timeout_exception(
|
|
|
|
|
+ "socket time-out during read")
|
|
|
|
|
+ try:
|
|
|
|
|
+ chunk = self._socket.recv(remaining)
|
|
|
|
|
+ except ssl.SSLError as e:
|
|
|
|
|
+ if e.errno in (ssl.SSL_ERROR_WANT_READ,
|
|
|
|
|
+ ssl.SSL_ERROR_WANT_WRITE):
|
|
|
|
|
+ continue
|
|
|
|
|
+ else:
|
|
|
|
|
+ raise
|
|
|
if chunk == b'':
|
|
if chunk == b'':
|
|
|
raise ConnectionDropped('socket connection broken')
|
|
raise ConnectionDropped('socket connection broken')
|
|
|
msgparts.append(chunk)
|
|
msgparts.append(chunk)
|
|
@@ -243,8 +270,8 @@ class ConnectionHandler(object):
|
|
|
if xid:
|
|
if xid:
|
|
|
header, buffer, offset = self._read_header(timeout)
|
|
header, buffer, offset = self._read_header(timeout)
|
|
|
if header.xid != xid:
|
|
if header.xid != xid:
|
|
|
- raise RuntimeError('xids do not match, expected %r received %r',
|
|
|
|
|
- xid, header.xid)
|
|
|
|
|
|
|
+ raise RuntimeError('xids do not match, expected %r '
|
|
|
|
|
+ 'received %r', xid, header.xid)
|
|
|
if header.zxid > 0:
|
|
if header.zxid > 0:
|
|
|
zxid = header.zxid
|
|
zxid = header.zxid
|
|
|
if header.err:
|
|
if header.err:
|
|
@@ -262,8 +289,9 @@ class ConnectionHandler(object):
|
|
|
try:
|
|
try:
|
|
|
obj, _ = request.deserialize(msg, 0)
|
|
obj, _ = request.deserialize(msg, 0)
|
|
|
except Exception:
|
|
except Exception:
|
|
|
- self.logger.exception("Exception raised during deserialization"
|
|
|
|
|
- " of request: %s", request)
|
|
|
|
|
|
|
+ self.logger.exception(
|
|
|
|
|
+ "Exception raised during deserialization "
|
|
|
|
|
+ "of request: %s", request)
|
|
|
|
|
|
|
|
# raise ConnectionDropped so connect loop will retry
|
|
# raise ConnectionDropped so connect loop will retry
|
|
|
raise ConnectionDropped('invalid server response')
|
|
raise ConnectionDropped('invalid server response')
|
|
@@ -281,8 +309,9 @@ class ConnectionHandler(object):
|
|
|
if request.type:
|
|
if request.type:
|
|
|
b.extend(int_struct.pack(request.type))
|
|
b.extend(int_struct.pack(request.type))
|
|
|
b += request.serialize()
|
|
b += request.serialize()
|
|
|
- self.logger.log((BLATHER if isinstance(request, Ping) else logging.DEBUG),
|
|
|
|
|
- "Sending request(xid=%s): %s", xid, request)
|
|
|
|
|
|
|
+ self.logger.log(
|
|
|
|
|
+ (BLATHER if isinstance(request, Ping) else logging.DEBUG),
|
|
|
|
|
+ "Sending request(xid=%s): %s", xid, request)
|
|
|
self._write(int_struct.pack(len(b)) + b, timeout)
|
|
self._write(int_struct.pack(len(b)) + b, timeout)
|
|
|
|
|
|
|
|
def _write(self, msg, timeout):
|
|
def _write(self, msg, timeout):
|
|
@@ -295,9 +324,17 @@ class ConnectionHandler(object):
|
|
|
if not s: # pragma: nocover
|
|
if not s: # pragma: nocover
|
|
|
# If the write list is empty, we got a timeout. We don't
|
|
# If the write list is empty, we got a timeout. We don't
|
|
|
# have to check rlist and xlist as we don't set any
|
|
# have to check rlist and xlist as we don't set any
|
|
|
- raise self.handler.timeout_exception("socket time-out")
|
|
|
|
|
|
|
+ raise self.handler.timeout_exception("socket time-out"
|
|
|
|
|
+ " during write")
|
|
|
msg_slice = buffer(msg, sent)
|
|
msg_slice = buffer(msg, sent)
|
|
|
- bytes_sent = self._socket.send(msg_slice)
|
|
|
|
|
|
|
+ try:
|
|
|
|
|
+ bytes_sent = self._socket.send(msg_slice)
|
|
|
|
|
+ except ssl.SSLError as e:
|
|
|
|
|
+ if e.errno in (ssl.SSL_ERROR_WANT_READ,
|
|
|
|
|
+ ssl.SSL_ERROR_WANT_WRITE):
|
|
|
|
|
+ continue
|
|
|
|
|
+ else:
|
|
|
|
|
+ raise
|
|
|
if not bytes_sent:
|
|
if not bytes_sent:
|
|
|
raise ConnectionDropped('socket connection broken')
|
|
raise ConnectionDropped('socket connection broken')
|
|
|
sent += bytes_sent
|
|
sent += bytes_sent
|
|
@@ -340,8 +377,10 @@ class ConnectionHandler(object):
|
|
|
if header.zxid and header.zxid > 0:
|
|
if header.zxid and header.zxid > 0:
|
|
|
client.last_zxid = header.zxid
|
|
client.last_zxid = header.zxid
|
|
|
if header.xid != xid:
|
|
if header.xid != xid:
|
|
|
- raise RuntimeError('xids do not match, expected %r '
|
|
|
|
|
|
|
+ exc = RuntimeError('xids do not match, expected %r '
|
|
|
'received %r', xid, header.xid)
|
|
'received %r', xid, header.xid)
|
|
|
|
|
+ async_object.set_exception(exc)
|
|
|
|
|
+ raise exc
|
|
|
|
|
|
|
|
# Determine if its an exists request and a no node error
|
|
# Determine if its an exists request and a no node error
|
|
|
exists_error = (header.err == NoNodeError.code and
|
|
exists_error = (header.err == NoNodeError.code and
|
|
@@ -363,8 +402,9 @@ class ConnectionHandler(object):
|
|
|
try:
|
|
try:
|
|
|
response = request.deserialize(buffer, offset)
|
|
response = request.deserialize(buffer, offset)
|
|
|
except Exception as exc:
|
|
except Exception as exc:
|
|
|
- self.logger.exception("Exception raised during deserialization"
|
|
|
|
|
- " of request: %s", request)
|
|
|
|
|
|
|
+ self.logger.exception(
|
|
|
|
|
+ "Exception raised during deserialization "
|
|
|
|
|
+ "of request: %s", request)
|
|
|
async_object.set_exception(exc)
|
|
async_object.set_exception(exc)
|
|
|
return
|
|
return
|
|
|
self.logger.debug(
|
|
self.logger.debug(
|
|
@@ -379,7 +419,7 @@ class ConnectionHandler(object):
|
|
|
# Determine if watchers should be registered
|
|
# Determine if watchers should be registered
|
|
|
watcher = getattr(request, 'watcher', None)
|
|
watcher = getattr(request, 'watcher', None)
|
|
|
if not client._stopped.is_set() and watcher:
|
|
if not client._stopped.is_set() and watcher:
|
|
|
- if isinstance(request, GetChildren):
|
|
|
|
|
|
|
+ if isinstance(request, (GetChildren, GetChildren2)):
|
|
|
client._child_watchers[request.path].add(watcher)
|
|
client._child_watchers[request.path].add(watcher)
|
|
|
else:
|
|
else:
|
|
|
client._data_watchers[request.path].add(watcher)
|
|
client._data_watchers[request.path].add(watcher)
|
|
@@ -420,11 +460,11 @@ class ConnectionHandler(object):
|
|
|
except IndexError:
|
|
except IndexError:
|
|
|
# Not actually something on the queue, this can occur if
|
|
# Not actually something on the queue, this can occur if
|
|
|
# something happens to cancel the request such that we
|
|
# something happens to cancel the request such that we
|
|
|
- # don't clear the pipe below after sending
|
|
|
|
|
|
|
+ # don't clear the socket below after sending
|
|
|
try:
|
|
try:
|
|
|
# Clear possible inconsistence (no request in the queue
|
|
# Clear possible inconsistence (no request in the queue
|
|
|
- # but have data in the read pipe), which causes cpu to spin.
|
|
|
|
|
- os.read(self._read_pipe, 1)
|
|
|
|
|
|
|
+ # but have data in the read socket), which causes cpu to spin.
|
|
|
|
|
+ self._read_sock.recv(1)
|
|
|
except OSError:
|
|
except OSError:
|
|
|
pass
|
|
pass
|
|
|
return
|
|
return
|
|
@@ -440,12 +480,12 @@ class ConnectionHandler(object):
|
|
|
if request.type == Auth.type:
|
|
if request.type == Auth.type:
|
|
|
xid = AUTH_XID
|
|
xid = AUTH_XID
|
|
|
else:
|
|
else:
|
|
|
- self._xid += 1
|
|
|
|
|
|
|
+ self._xid = (self._xid % 2147483647) + 1
|
|
|
xid = self._xid
|
|
xid = self._xid
|
|
|
|
|
|
|
|
self._submit(request, connect_timeout, xid)
|
|
self._submit(request, connect_timeout, xid)
|
|
|
client._queue.popleft()
|
|
client._queue.popleft()
|
|
|
- os.read(self._read_pipe, 1)
|
|
|
|
|
|
|
+ self._read_sock.recv(1)
|
|
|
client._pending.append((request, async_object, xid))
|
|
client._pending.append((request, async_object, xid))
|
|
|
|
|
|
|
|
def _send_ping(self, connect_timeout):
|
|
def _send_ping(self, connect_timeout):
|
|
@@ -479,14 +519,37 @@ class ConnectionHandler(object):
|
|
|
self.client._session_callback(KeeperState.CLOSED)
|
|
self.client._session_callback(KeeperState.CLOSED)
|
|
|
self.logger.log(BLATHER, 'Connection stopped')
|
|
self.logger.log(BLATHER, 'Connection stopped')
|
|
|
|
|
|
|
|
|
|
+ def _expand_client_hosts(self):
|
|
|
|
|
+ # Expand the entire list in advance so we can randomize it if needed
|
|
|
|
|
+ host_ports = []
|
|
|
|
|
+ for host, port in self.client.hosts:
|
|
|
|
|
+ try:
|
|
|
|
|
+ host = host.strip()
|
|
|
|
|
+ for rhost in socket.getaddrinfo(host, port, 0, 0,
|
|
|
|
|
+ socket.IPPROTO_TCP):
|
|
|
|
|
+ host_ports.append((host, rhost[4][0], rhost[4][1]))
|
|
|
|
|
+ except socket.gaierror as e:
|
|
|
|
|
+ # Skip hosts that don't resolve
|
|
|
|
|
+ self.logger.warning("Cannot resolve %s: %s", host, e)
|
|
|
|
|
+ pass
|
|
|
|
|
+ if self.client.randomize_hosts:
|
|
|
|
|
+ random.shuffle(host_ports)
|
|
|
|
|
+ return host_ports
|
|
|
|
|
+
|
|
|
def _connect_loop(self, retry):
|
|
def _connect_loop(self, retry):
|
|
|
# Iterate through the hosts a full cycle before starting over
|
|
# Iterate through the hosts a full cycle before starting over
|
|
|
status = None
|
|
status = None
|
|
|
- for host, port in self.client.hosts:
|
|
|
|
|
|
|
+ host_ports = self._expand_client_hosts()
|
|
|
|
|
+
|
|
|
|
|
+ # Check for an empty hostlist, indicating none resolved
|
|
|
|
|
+ if len(host_ports) == 0:
|
|
|
|
|
+ return STOP_CONNECTING
|
|
|
|
|
+
|
|
|
|
|
+ for host, hostip, port in host_ports:
|
|
|
if self.client._stopped.is_set():
|
|
if self.client._stopped.is_set():
|
|
|
status = STOP_CONNECTING
|
|
status = STOP_CONNECTING
|
|
|
break
|
|
break
|
|
|
- status = self._connect_attempt(host, port, retry)
|
|
|
|
|
|
|
+ status = self._connect_attempt(host, hostip, port, retry)
|
|
|
if status is STOP_CONNECTING:
|
|
if status is STOP_CONNECTING:
|
|
|
break
|
|
break
|
|
|
|
|
|
|
@@ -495,9 +558,9 @@ class ConnectionHandler(object):
|
|
|
else:
|
|
else:
|
|
|
raise ForceRetryError('Reconnecting')
|
|
raise ForceRetryError('Reconnecting')
|
|
|
|
|
|
|
|
- def _connect_attempt(self, host, port, retry):
|
|
|
|
|
|
|
+ def _connect_attempt(self, host, hostip, port, retry):
|
|
|
client = self.client
|
|
client = self.client
|
|
|
- TimeoutError = self.handler.timeout_exception
|
|
|
|
|
|
|
+ KazooTimeoutError = self.handler.timeout_exception
|
|
|
close_connection = False
|
|
close_connection = False
|
|
|
|
|
|
|
|
self._socket = None
|
|
self._socket = None
|
|
@@ -513,46 +576,47 @@ class ConnectionHandler(object):
|
|
|
client._session_callback(KeeperState.CONNECTING)
|
|
client._session_callback(KeeperState.CONNECTING)
|
|
|
|
|
|
|
|
try:
|
|
try:
|
|
|
- read_timeout, connect_timeout = self._connect(host, port)
|
|
|
|
|
|
|
+ self._xid = 0
|
|
|
|
|
+ read_timeout, connect_timeout = self._connect(host, hostip, port)
|
|
|
read_timeout = read_timeout / 1000.0
|
|
read_timeout = read_timeout / 1000.0
|
|
|
connect_timeout = connect_timeout / 1000.0
|
|
connect_timeout = connect_timeout / 1000.0
|
|
|
retry.reset()
|
|
retry.reset()
|
|
|
- self._xid = 0
|
|
|
|
|
-
|
|
|
|
|
- while not close_connection:
|
|
|
|
|
- # Watch for something to read or send
|
|
|
|
|
- jitter_time = random.randint(0, 40) / 100.0
|
|
|
|
|
- # Ensure our timeout is positive
|
|
|
|
|
- timeout = max([read_timeout / 2.0 - jitter_time, jitter_time])
|
|
|
|
|
- s = self.handler.select([self._socket, self._read_pipe],
|
|
|
|
|
- [], [], timeout)[0]
|
|
|
|
|
-
|
|
|
|
|
- if not s:
|
|
|
|
|
- if self.ping_outstanding.is_set():
|
|
|
|
|
- self.ping_outstanding.clear()
|
|
|
|
|
- raise ConnectionDropped(
|
|
|
|
|
- "outstanding heartbeat ping not received")
|
|
|
|
|
- self._send_ping(connect_timeout)
|
|
|
|
|
- elif s[0] == self._socket:
|
|
|
|
|
- response = self._read_socket(read_timeout)
|
|
|
|
|
- close_connection = response == CLOSE_RESPONSE
|
|
|
|
|
- else:
|
|
|
|
|
- self._send_request(read_timeout, connect_timeout)
|
|
|
|
|
-
|
|
|
|
|
|
|
+ self.ping_outstanding.clear()
|
|
|
|
|
+ with self._socket_error_handling():
|
|
|
|
|
+ while not close_connection:
|
|
|
|
|
+ # Watch for something to read or send
|
|
|
|
|
+ jitter_time = random.randint(0, 40) / 100.0
|
|
|
|
|
+ # Ensure our timeout is positive
|
|
|
|
|
+ timeout = max([read_timeout / 2.0 - jitter_time,
|
|
|
|
|
+ jitter_time])
|
|
|
|
|
+ s = self.handler.select([self._socket, self._read_sock],
|
|
|
|
|
+ [], [], timeout)[0]
|
|
|
|
|
+
|
|
|
|
|
+ if not s:
|
|
|
|
|
+ if self.ping_outstanding.is_set():
|
|
|
|
|
+ self.ping_outstanding.clear()
|
|
|
|
|
+ raise ConnectionDropped(
|
|
|
|
|
+ "outstanding heartbeat ping not received")
|
|
|
|
|
+ self._send_ping(connect_timeout)
|
|
|
|
|
+ elif s[0] == self._socket:
|
|
|
|
|
+ response = self._read_socket(read_timeout)
|
|
|
|
|
+ close_connection = response == CLOSE_RESPONSE
|
|
|
|
|
+ else:
|
|
|
|
|
+ self._send_request(read_timeout, connect_timeout)
|
|
|
self.logger.info('Closing connection to %s:%s', host, port)
|
|
self.logger.info('Closing connection to %s:%s', host, port)
|
|
|
client._session_callback(KeeperState.CLOSED)
|
|
client._session_callback(KeeperState.CLOSED)
|
|
|
return STOP_CONNECTING
|
|
return STOP_CONNECTING
|
|
|
- except (ConnectionDropped, TimeoutError) as e:
|
|
|
|
|
|
|
+ except (ConnectionDropped, KazooTimeoutError) as e:
|
|
|
if isinstance(e, ConnectionDropped):
|
|
if isinstance(e, ConnectionDropped):
|
|
|
self.logger.warning('Connection dropped: %s', e)
|
|
self.logger.warning('Connection dropped: %s', e)
|
|
|
else:
|
|
else:
|
|
|
- self.logger.warning('Connection time-out')
|
|
|
|
|
|
|
+ self.logger.warning('Connection time-out: %s', e)
|
|
|
if client._state != KeeperState.CONNECTING:
|
|
if client._state != KeeperState.CONNECTING:
|
|
|
self.logger.warning("Transition to CONNECTING")
|
|
self.logger.warning("Transition to CONNECTING")
|
|
|
client._session_callback(KeeperState.CONNECTING)
|
|
client._session_callback(KeeperState.CONNECTING)
|
|
|
- except AuthFailedError:
|
|
|
|
|
|
|
+ except AuthFailedError as err:
|
|
|
retry.reset()
|
|
retry.reset()
|
|
|
- self.logger.warning('AUTH_FAILED closing')
|
|
|
|
|
|
|
+ self.logger.warning('AUTH_FAILED closing: %s', err)
|
|
|
client._session_callback(KeeperState.AUTH_FAILED)
|
|
client._session_callback(KeeperState.AUTH_FAILED)
|
|
|
return STOP_CONNECTING
|
|
return STOP_CONNECTING
|
|
|
except SessionExpiredError:
|
|
except SessionExpiredError:
|
|
@@ -570,18 +634,27 @@ class ConnectionHandler(object):
|
|
|
if self._socket is not None:
|
|
if self._socket is not None:
|
|
|
self._socket.close()
|
|
self._socket.close()
|
|
|
|
|
|
|
|
- def _connect(self, host, port):
|
|
|
|
|
|
|
+ def _connect(self, host, hostip, port):
|
|
|
client = self.client
|
|
client = self.client
|
|
|
- self.logger.info('Connecting to %s:%s', host, port)
|
|
|
|
|
|
|
+ self.logger.info('Connecting to %s(%s):%s, use_ssl: %r',
|
|
|
|
|
+ host, hostip, port, self.client.use_ssl)
|
|
|
|
|
|
|
|
self.logger.log(BLATHER,
|
|
self.logger.log(BLATHER,
|
|
|
- ' Using session_id: %r session_passwd: %s',
|
|
|
|
|
- client._session_id,
|
|
|
|
|
- hexlify(client._session_passwd))
|
|
|
|
|
|
|
+ ' Using session_id: %r session_passwd: %s',
|
|
|
|
|
+ client._session_id,
|
|
|
|
|
+ hexlify(client._session_passwd))
|
|
|
|
|
|
|
|
with self._socket_error_handling():
|
|
with self._socket_error_handling():
|
|
|
self._socket = self.handler.create_connection(
|
|
self._socket = self.handler.create_connection(
|
|
|
- (host, port), client._session_timeout / 1000.0)
|
|
|
|
|
|
|
+ address=(hostip, port),
|
|
|
|
|
+ timeout=client._session_timeout / 1000.0,
|
|
|
|
|
+ use_ssl=self.client.use_ssl,
|
|
|
|
|
+ keyfile=self.client.keyfile,
|
|
|
|
|
+ certfile=self.client.certfile,
|
|
|
|
|
+ ca=self.client.ca,
|
|
|
|
|
+ keyfile_password=self.client.keyfile_password,
|
|
|
|
|
+ verify_certs=self.client.verify_certs,
|
|
|
|
|
+ )
|
|
|
|
|
|
|
|
self._socket.setblocking(0)
|
|
self._socket.setblocking(0)
|
|
|
|
|
|
|
@@ -589,7 +662,8 @@ class ConnectionHandler(object):
|
|
|
client._session_id or 0, client._session_passwd,
|
|
client._session_id or 0, client._session_passwd,
|
|
|
client.read_only)
|
|
client.read_only)
|
|
|
|
|
|
|
|
- connect_result, zxid = self._invoke(client._session_timeout, connect)
|
|
|
|
|
|
|
+ connect_result, zxid = self._invoke(
|
|
|
|
|
+ client._session_timeout / 1000.0 / len(client.hosts), connect)
|
|
|
|
|
|
|
|
if connect_result.time_out <= 0:
|
|
if connect_result.time_out <= 0:
|
|
|
raise SessionExpiredError("Session has expired")
|
|
raise SessionExpiredError("Session has expired")
|
|
@@ -606,16 +680,13 @@ class ConnectionHandler(object):
|
|
|
client._session_passwd = connect_result.passwd
|
|
client._session_passwd = connect_result.passwd
|
|
|
|
|
|
|
|
self.logger.log(BLATHER,
|
|
self.logger.log(BLATHER,
|
|
|
- 'Session created, session_id: %r session_passwd: %s\n'
|
|
|
|
|
- ' negotiated session timeout: %s\n'
|
|
|
|
|
- ' connect timeout: %s\n'
|
|
|
|
|
- ' read timeout: %s', client._session_id,
|
|
|
|
|
- hexlify(client._session_passwd),
|
|
|
|
|
- negotiated_session_timeout, connect_timeout,
|
|
|
|
|
- read_timeout)
|
|
|
|
|
-
|
|
|
|
|
- if self.sasl_server_principal:
|
|
|
|
|
- self._authenticate_with_sasl(host, connect_timeout / 1000.0)
|
|
|
|
|
|
|
+ 'Session created, session_id: %r session_passwd: %s\n'
|
|
|
|
|
+ ' negotiated session timeout: %s\n'
|
|
|
|
|
+ ' connect timeout: %s\n'
|
|
|
|
|
+ ' read timeout: %s', client._session_id,
|
|
|
|
|
+ hexlify(client._session_passwd),
|
|
|
|
|
+ negotiated_session_timeout, connect_timeout,
|
|
|
|
|
+ read_timeout)
|
|
|
|
|
|
|
|
if connect_result.read_only:
|
|
if connect_result.read_only:
|
|
|
client._session_callback(KeeperState.CONNECTED_RO)
|
|
client._session_callback(KeeperState.CONNECTED_RO)
|
|
@@ -624,41 +695,97 @@ class ConnectionHandler(object):
|
|
|
client._session_callback(KeeperState.CONNECTED)
|
|
client._session_callback(KeeperState.CONNECTED)
|
|
|
self._ro_mode = None
|
|
self._ro_mode = None
|
|
|
|
|
|
|
|
- for scheme, auth in client.auth_data:
|
|
|
|
|
|
|
+ if self.sasl_options is not None:
|
|
|
|
|
+ self._authenticate_with_sasl(host, connect_timeout / 1000.0)
|
|
|
|
|
+
|
|
|
|
|
+ # Get a copy of the auth data before iterating, in case it is
|
|
|
|
|
+ # changed.
|
|
|
|
|
+ client_auth_data_copy = copy.copy(client.auth_data)
|
|
|
|
|
+
|
|
|
|
|
+ for scheme, auth in client_auth_data_copy:
|
|
|
ap = Auth(0, scheme, auth)
|
|
ap = Auth(0, scheme, auth)
|
|
|
- zxid = self._invoke(connect_timeout, ap, xid=AUTH_XID)
|
|
|
|
|
|
|
+ zxid = self._invoke(connect_timeout / 1000.0, ap, xid=AUTH_XID)
|
|
|
if zxid:
|
|
if zxid:
|
|
|
client.last_zxid = zxid
|
|
client.last_zxid = zxid
|
|
|
|
|
|
|
|
return read_timeout, connect_timeout
|
|
return read_timeout, connect_timeout
|
|
|
|
|
|
|
|
def _authenticate_with_sasl(self, host, timeout):
|
|
def _authenticate_with_sasl(self, host, timeout):
|
|
|
- saslc = sasl.Client()
|
|
|
|
|
- saslc.setAttr('host', str(host))
|
|
|
|
|
- saslc.setAttr('service', str(self.sasl_server_principal))
|
|
|
|
|
- saslc.init()
|
|
|
|
|
-
|
|
|
|
|
- ret, chosen_mech, initial_response = saslc.start('GSSAPI')
|
|
|
|
|
- if not ret:
|
|
|
|
|
- raise SaslException(saslc.getError())
|
|
|
|
|
-
|
|
|
|
|
- response = initial_response
|
|
|
|
|
-
|
|
|
|
|
|
|
+ """Establish a SASL authenticated connection to the server.
|
|
|
|
|
+ """
|
|
|
|
|
+ if not PURESASL_AVAILABLE:
|
|
|
|
|
+ raise SASLException('Missing SASL support')
|
|
|
|
|
+
|
|
|
|
|
+ if 'service' not in self.sasl_options:
|
|
|
|
|
+ self.sasl_options['service'] = 'zookeeper'
|
|
|
|
|
+
|
|
|
|
|
+ # NOTE: Zookeeper hardcoded the domain for Digest authentication
|
|
|
|
|
+ # instead of using the hostname. See
|
|
|
|
|
+ # zookeeper/util/SecurityUtils.java#L74 and Server/Client
|
|
|
|
|
+ # initializations.
|
|
|
|
|
+ if self.sasl_options['mechanism'] == 'DIGEST-MD5':
|
|
|
|
|
+ host = 'zk-sasl-md5'
|
|
|
|
|
+
|
|
|
|
|
+ sasl_cli = self.client.sasl_cli = puresasl.client.SASLClient(
|
|
|
|
|
+ host=host,
|
|
|
|
|
+ **self.sasl_options
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ # Inititalize the process with an empty challenge token
|
|
|
|
|
+ challenge = None
|
|
|
xid = 0
|
|
xid = 0
|
|
|
|
|
|
|
|
while True:
|
|
while True:
|
|
|
- xid += 1
|
|
|
|
|
|
|
+ if sasl_cli.complete:
|
|
|
|
|
+ break
|
|
|
|
|
+
|
|
|
|
|
+ try:
|
|
|
|
|
+ response = sasl_cli.process(challenge=challenge)
|
|
|
|
|
+ except puresasl.SASLError as err:
|
|
|
|
|
+ six.reraise(
|
|
|
|
|
+ SASLException,
|
|
|
|
|
+ SASLException('library error: %s' % err.message),
|
|
|
|
|
+ sys.exc_info()[2]
|
|
|
|
|
+ )
|
|
|
|
|
+ except puresasl.SASLProtocolException as err:
|
|
|
|
|
+ six.reraise(
|
|
|
|
|
+ AuthFailedError,
|
|
|
|
|
+ AuthFailedError('protocol error: %s' % err.message),
|
|
|
|
|
+ sys.exc_info()[2]
|
|
|
|
|
+ )
|
|
|
|
|
+ except Exception as err:
|
|
|
|
|
+ six.reraise(
|
|
|
|
|
+ AuthFailedError,
|
|
|
|
|
+ AuthFailedError('Unknown error: %s' % err),
|
|
|
|
|
+ sys.exc_info()[2]
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ if sasl_cli.complete and not response:
|
|
|
|
|
+ break
|
|
|
|
|
+ elif response is None:
|
|
|
|
|
+ response = b''
|
|
|
|
|
+
|
|
|
|
|
+ xid = (xid % 2147483647) + 1
|
|
|
|
|
|
|
|
request = SASL(response)
|
|
request = SASL(response)
|
|
|
self._submit(request, timeout, xid)
|
|
self._submit(request, timeout, xid)
|
|
|
|
|
|
|
|
- header, buffer, offset = self._read_header(timeout)
|
|
|
|
|
|
|
+ try:
|
|
|
|
|
+ header, buffer, offset = self._read_header(timeout)
|
|
|
|
|
+ except ConnectionDropped:
|
|
|
|
|
+ # Zookeeper simply drops connections with failed authentication
|
|
|
|
|
+ six.reraise(
|
|
|
|
|
+ AuthFailedError,
|
|
|
|
|
+ AuthFailedError('Connection dropped in SASL'),
|
|
|
|
|
+ sys.exc_info()[2]
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
if header.xid != xid:
|
|
if header.xid != xid:
|
|
|
raise RuntimeError('xids do not match, expected %r '
|
|
raise RuntimeError('xids do not match, expected %r '
|
|
|
'received %r', xid, header.xid)
|
|
'received %r', xid, header.xid)
|
|
|
|
|
|
|
|
if header.zxid > 0:
|
|
if header.zxid > 0:
|
|
|
- client.last_zxid = zxid
|
|
|
|
|
|
|
+ self.client.last_zxid = header.zxid
|
|
|
|
|
|
|
|
if header.err:
|
|
if header.err:
|
|
|
callback_exception = EXCEPTIONS[header.err]()
|
|
callback_exception = EXCEPTIONS[header.err]()
|
|
@@ -666,11 +793,8 @@ class ConnectionHandler(object):
|
|
|
'Received error(xid=%s) %r', xid, callback_exception)
|
|
'Received error(xid=%s) %r', xid, callback_exception)
|
|
|
raise callback_exception
|
|
raise callback_exception
|
|
|
|
|
|
|
|
- token, _ = SASL.deserialize(buffer, offset)
|
|
|
|
|
-
|
|
|
|
|
- if not token:
|
|
|
|
|
- break
|
|
|
|
|
|
|
+ challenge, _ = SASL.deserialize(buffer, offset)
|
|
|
|
|
|
|
|
- ret, response = saslc.step(token)
|
|
|
|
|
- if not ret:
|
|
|
|
|
- raise SaslException(saslc.getError())
|
|
|
|
|
|
|
+ # If we made it here, authentication is ok, and we are connected.
|
|
|
|
|
+ # Remove sensible information from the object.
|
|
|
|
|
+ sasl_cli.dispose()
|