Fix issues with leftover ssh sessions

Upon connection loss, even though confluent internally
decides it is done with it, it fails to close the session.

Catch a number of these scenarios and ensure the connection closes.

confluent_client/bin/nodeshell

@@ -46,6 +46,8 @@ def run():
                          help='Number of commands to run at a time')
     argparser.add_option('-n', '--nonodeprefix', action='store_true',
                          help='Do not prefix output with node names')
+    argparser.add_option('-p', '--port', type='int', default=0,
+                         help='Specify a custom port for ssh')
     argparser.add_option('-m', '--maxnodes', type='int',
                      help='Specify a maximum number of '
                           'nodes to run remote ssh command to, '
@@ -59,7 +61,7 @@ def run():
         sys.exit(1)
     client.check_globbing(args[0])
     concurrentprocs = options.count
-    c = client.Command()    
+    c = client.Command()
     cmdstr = " ".join(args[1:])
 
     currprocs = 0
@@ -79,7 +81,10 @@ def run():
             cmd = ex[node]['value']
             if not isinstance(cmd, str) and not isinstance(cmd, bytes):
                 cmd = cmd.encode('utf-8')
-            cmdv = ['ssh', node, cmd]
+            if options.port:
+                cmdv = ['ssh', '-p', '{0}'.format(options.port), node, cmd]
+            else:
+                cmdv = ['ssh', node, cmd]
             if currprocs < concurrentprocs:
                 currprocs += 1
                 run_cmdv(node, cmdv, all, pipedesc)

confluent_server/bin/confluentdbutil

@@ -62,8 +62,11 @@ if args[0] == 'restore':
     if pid is not None:
         print("Confluent is running, must shut down to restore db")
         sys.exit(1)
+    password = options.password
+    if options.interactivepassword:
+        password = getpass.getpass('Enter password to restore backup: ')
     try:
-        cfm.restore_db_from_directory(dumpdir, options.password)
+        cfm.restore_db_from_directory(dumpdir, password)
     except Exception as e:
         print(str(e))
         sys.exit(1)
@@ -86,7 +89,7 @@ elif args[0] == 'dump':
     main._initsecurity(conf.get_config())
     if not os.path.exists(dumpdir):
         os.makedirs(dumpdir)
-    cfm.dump_db_to_directory(dumpdir, options.password, options.redact,
+    cfm.dump_db_to_directory(dumpdir, password, options.redact,
                              options.skipkeys)
 
 

confluent_server/builddeb

@@ -36,7 +36,7 @@ if [ "$OPKGNAME" = "confluent-server" ]; then
     if grep wheezy /etc/os-release; then
         sed -i 's/^\(Depends:.*\)/\1, python-confluent-client, python-lxml, python-eficompressor, python-pycryptodomex, python-dateutil, python-pyopenssl, python-msgpack/' debian/control
     else
-        sed -i 's/^\(Depends:.*\)/\1, confluent-client, python3-lxml, python3-eficompressor, python3-pycryptodome, python3-websocket, python3-msgpack/' debian/control
+        sed -i 's/^\(Depends:.*\)/\1, confluent-client, python3-lxml, python3-eficompressor, python3-pycryptodome, python3-websocket, python3-msgpack, python3-eventlet, python3-pyparsing, python3-pyte, python3-pyghmi, python3-paramiko/' debian/control
     fi
     if grep wheezy /etc/os-release; then
         echo 'confluent_client python-confluent-client' >> debian/pydist-overrides
@@ -44,6 +44,9 @@ if [ "$OPKGNAME" = "confluent-server" ]; then
         echo 'confluent_client confluent-client' >> debian/pydist-overrides
     fi
 fi
+if ! grep wheezy /etc/os-release; then
+    sed -i 's/^Package: python3-/Package: /' debian/control
+fi
 head -n -1 debian/control > debian/control1
 mv debian/control1 debian/control
 echo 'export PYBUILD_INSTALL_ARGS=--install-lib=/opt/confluent/lib/python' >> debian/rules

confluent_server/confluent/config/configmanager.py

@@ -117,6 +117,7 @@ def attrib_supports_expression(attrib):
     attrib = _attraliases.get(attrib, attrib)
     if attrib.startswith('secret.') or attrib.startswith('crypted.'):
         return False
+    return True
 
 
 def _mkpath(pathname):
@@ -495,7 +496,8 @@ def crypt_value(value,
         key = _masterkey
     iv = os.urandom(12)
     crypter = AES.new(key, AES.MODE_GCM, nonce=iv)
-    value = confluent.util.stringify(value).encode('utf-8')
+    if not isinstance(value, bytes):
+        value = value.encode('utf-8')
     cryptval, hmac = crypter.encrypt_and_digest(value)
     return iv, cryptval, hmac, b'\x02'
 

confluent_server/confluent/consoleserver.py

@@ -108,10 +108,10 @@ def pytechars2line(chars, maxlen=None):
         char = chars[charidx]
         csi = bytearray([])
         if char.fg != lfg:
-            csi.append(30 + pytecolors2ansi[char.fg])
+            csi.append(30 + pytecolors2ansi.get(char.fg, 9))
             lfg = char.fg
         if char.bg != lbg:
-            csi.append(40 + pytecolors2ansi[char.bg])
+            csi.append(40 + pytecolors2ansi.get(char.bg, 9))
             lbg = char.bg
         if char.bold != lb:
             lb = char.bold
@@ -243,7 +243,7 @@ class ConsoleHandler(object):
     def check_collective(self, attrvalue):
         myc = attrvalue.get(self.node, {}).get('collective.manager', {}).get(
             'value', None)
-        if configmodule.list_collective() and not myc:
+        if list(configmodule.list_collective()) and not myc:
             self._is_local = False
             self._detach()
             self._disconnect()

confluent_server/confluent/core.py

@@ -69,6 +69,10 @@ import sys
 pluginmap = {}
 dispatch_plugins = (b'ipmi', u'ipmi', b'redfish', u'redfish', b'tsmsol', u'tsmsol')
 
+try:
+    unicode
+except NameError:
+    unicode = str
 
 def seek_element(currplace, currkey):
     try:

confluent_server/confluent/discovery/core.py

@@ -1076,7 +1076,7 @@ def discover_node(cfg, handler, info, nodename, manual):
                 traceback.print_exc()
                 return False
             newnodeattribs = {}
-            if cfm.list_collective():
+            if list(cfm.list_collective()):
                 # We are in a collective, check collective.manager
                 cmc = cfg.get_node_attributes(nodename, 'collective.manager')
                 cm = cmc.get(nodename, {}).get('collective.manager', {}).get('value', None)

confluent_server/confluent/discovery/handlers/smm.py

@@ -90,6 +90,13 @@ class NodeHandler(bmchandler.NodeHandler):
             smmip = smmip[-1][0]
             if smmip and ':' in smmip:
                 raise exc.NotImplementedException('IPv6 not supported')
+            wc.request('POST', '/data', 'get=hostname')
+            rsp = wc.getresponse()
+            rspdata = fromstring(util.stringify(rsp.read()))
+            currip = rspdata.find('netConfig').find('ifConfigEntries').find(
+                'ifConfig').find('v4IPAddr').text
+            if currip == smmip:
+                return
             netconfig = netutil.get_nic_config(cfg, nodename, ip=smmip)
             netmask = netutil.cidr_to_mask(netconfig['prefix'])
             setdata = 'set=ifIndex:0,v4DHCPEnabled:0,v4IPAddr:{0},v4NetMask:{1}'.format(smmip, netmask)

confluent_server/confluent/noderange.py

@@ -250,6 +250,8 @@ class NodeRange(object):
                 return nodes
         if ':' in element:  # : range for less ambiguity
             return self.expandrange(element, ':')
+        elif '..' in element:
+            return self.expandrange(element, '..')
         elif '-' in element:
             return self.expandrange(element, '-')
         elif '+' in element:

confluent_server/confluent/plugins/hardwaremanagement/ipmi.py

@@ -1035,6 +1035,8 @@ class IpmiHandler(object):
             return self._create_storage(storelem)
 
     def _delete_storage(self, storelem):
+        if len(storelem) < 2:
+            storelem.append('')
         if len(storelem) < 2 or storelem[0] != 'volumes':
             raise exc.InvalidArgumentException('Must target a specific volume')
         volname = storelem[-1]

confluent_server/confluent/plugins/hardwaremanagement/redfish.py

@@ -894,6 +894,8 @@ class IpmiHandler(object):
             return self._create_storage(storelem)
 
     def _delete_storage(self, storelem):
+        if len(storelem) < 2:
+            storelem.append('')
         if len(storelem) < 2 or storelem[0] != 'volumes':
             raise exc.InvalidArgumentException('Must target a specific volume')
         volname = storelem[-1]

confluent_server/confluent/plugins/shell/ssh.py

@@ -99,8 +99,10 @@ class SshShell(conapi.Console):
     def recvdata(self):
         while self.connected:
             pendingdata = self.shell.recv(8192)
-            if pendingdata == '':
-                self.datacallback(conapi.ConsoleEvent.Disconnect)
+            if not pendingdata:
+                self.ssh.close()
+                if self.datacallback:
+                    self.datacallback(conapi.ConsoleEvent.Disconnect)
                 return
             self.datacallback(pendingdata)
 
@@ -110,7 +112,7 @@ class SshShell(conapi.Console):
         # that would rather not use the nodename as anything but an opaque
         # identifier
         self.datacallback = callback
-        if self.username is not '':
+        if self.username is not b'':
             self.logon()
         else:
             self.inputmode = 0
@@ -126,12 +128,14 @@ class SshShell(conapi.Console):
                              password=self.password, allow_agent=False,
                              look_for_keys=False)
         except paramiko.AuthenticationException:
+            self.ssh.close()
             self.inputmode = 0
             self.username = b''
             self.password = b''
             self.datacallback('\r\nlogin as: ')
             return
         except paramiko.ssh_exception.NoValidConnectionsError as e:
+            self.ssh.close()
             self.datacallback(str(e))
             self.inputmode = 0
             self.username = b''
@@ -139,6 +143,7 @@ class SshShell(conapi.Console):
             self.datacallback('\r\nlogin as: ')
             return
         except cexc.PubkeyInvalid as pi:
+            self.ssh.close()
             self.keyaction = ''
             self.candidatefprint = pi.fingerprint
             self.datacallback(pi.message)
@@ -148,6 +153,7 @@ class SshShell(conapi.Console):
             self.datacallback('\r\nEnter "disconnect" or "accept": ')
             return
         except paramiko.SSHException as pi:
+            self.ssh.close()
             self.inputmode = -2
             warn = str(pi)
             if warnhostkey:
@@ -169,11 +175,11 @@ class SshShell(conapi.Console):
             self.datacallback(conapi.ConsoleEvent.Disconnect)
             return
         elif self.inputmode == -1:
-            while len(data) and data[0] == b'\x7f' and len(self.keyaction):
+            while len(data) and data[0:1] == b'\x7f' and len(self.keyaction):
                 self.datacallback('\b \b')  # erase previously echoed value
                 self.keyaction = self.keyaction[:-1]
                 data = data[1:]
-            while len(data) and data[0] == b'\x7f':
+            while len(data) and data[0:1] == b'\x7f':
                 data = data[1:]
             while b'\x7f' in data:
                 delidx = data.index(b'\x7f')
@@ -195,11 +201,11 @@ class SshShell(conapi.Console):
             elif len(data) > 0:
                 self.datacallback(data)
         elif self.inputmode == 0:
-            while len(data) and data[0] == b'\x7f' and len(self.username):
+            while len(data) and data[0:1] == b'\x7f' and len(self.username):
                 self.datacallback('\b \b')  # erase previously echoed value
                 self.username = self.username[:-1]
                 data = data[1:]
-            while len(data) and data[0] == b'\x7f':
+            while len(data) and data[0:1] == b'\x7f':
                 data = data[1:]
             while b'\x7f' in data:
                 delidx = data.index(b'\x7f')
@@ -216,7 +222,7 @@ class SshShell(conapi.Console):
                 # echo back typed data
                 self.datacallback(data)
         elif self.inputmode == 1:
-            while len(data) > 0 and data[0] == b'\x7f':
+            while len(data) > 0 and data[0:1] == b'\x7f':
                 self.password = self.password[:-1]
                 data = data[1:]
             while b'\x7f' in data:
@@ -237,4 +243,4 @@ class SshShell(conapi.Console):
 
 def create(nodes, element, configmanager, inputdata):
     if len(nodes) == 1:
-        return SshShell(nodes[0], configmanager)
+        return SshShell(nodes[0], configmanager)

confluent_server/confluent/shellserver.py

@@ -111,6 +111,8 @@ class ShellSession(consoleserver.ConsoleSession):
 
     def destroy(self):
         try:
+            activesessions[(self.configmanager.tenant, self.node,
+                            self.username)][self.sessionid].close()
             del activesessions[(self.configmanager.tenant, self.node,
                                 self.username)][self.sessionid]
         except KeyError:

confluent_server/confluent/sockapi.py

@@ -472,10 +472,14 @@ class SockApi(object):
 
     def watch_for_cert(self):
         libc = ctypes.CDLL(ctypes.util.find_library('c'))
-        watcher = libc.inotify_init()
-        if libc.inotify_add_watch(watcher, '/etc/confluent/', 0x100) > -1:
+        watcher = libc.inotify_init1(os.O_NONBLOCK)
+        if libc.inotify_add_watch(watcher, b'/etc/confluent/', 0x100) > -1:
             while True:
                 select.select((watcher,), (), (), 86400)
+                try:
+                    os.read(watcher, 1024)
+                except Exception:
+                    pass
                 if self.should_run_remoteapi():
                     os.close(watcher)
                     self.start_remoteapi()