Merge remote-tracking branch 'xcat/master'

Author: Jarrod Johnson

.github/workflows/ci.yml

@@ -0,0 +1,64 @@
+name: CI
+
+on:
+  push:
+  pull_request:
+
+permissions:
+  contents: read
+
+jobs:
+  ShellCheck:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+      - name: Run ShellCheck (errors only)
+        # Check every tracked file that has a .sh extension or an sh/bash
+        # shebang. SC2148 (missing shebang) is excluded because many .sh
+        # files are sourced fragments or dracut hooks; ShellCheck then
+        # falls back to checking them as bash.
+        run: |
+          {
+            git ls-files '*.sh'
+            git ls-files | while IFS= read -r f; do
+              [ -f "$f" ] || continue
+              head -c 200 "$f" | head -n 1 | \
+                grep -qE '^#!.*[/ ](sh|bash|dash|ash|ksh)([ \t]|$)' && echo "$f"
+            done
+          } | sort -u | xargs -d '\n' shellcheck --severity=error --exclude=SC2148
+
+  python-compileall:
+    name: Python compileall
+    runs-on: ubuntu-latest
+    env:
+      # One entry per Python version shipped by the distros confluent
+      # targets, limited to versions actions/setup-python still provides
+      # on current runners (sles15/alma8 ship 3.6, which is unavailable).
+      # Newline-separated so it feeds both setup-python (multiline input)
+      # and the shell loop below (word-split on whitespace).
+      PYTHON_VERSIONS: |
+        3.8
+        3.9
+        3.10
+        3.12
+        3.13
+        3.14
+    steps:
+      - uses: actions/checkout@v7
+      - uses: actions/setup-python@v6
+        with:
+          python-version: ${{ env.PYTHON_VERSIONS }}
+      - name: Compile all Python files
+        run: |
+          rc=0
+          for v in $PYTHON_VERSIONS; do
+            echo "::group::Python $v"
+            if "python$v" -W error -m compileall -q -x '/\.git/' .; then
+              echo "::endgroup::"
+            else
+              echo "::endgroup::"
+              echo "::error::Python $v compileall failed"
+              rc=1
+            fi
+          done
+          exit "$rc"

confluent_osdeploy/el7-diskless/profiles/default/scripts/image2disk.py

@@ -173,8 +173,8 @@ def fixup(rootdir, vols):
     for vol in vols:
         if vol['mount'] == '/boot/efi':
             targdev = vol['targetdisk']
-            partnum = re.search('(\d+)$', targdev).group(1)
-            targblock = re.search('(.*)\d+$', targdev).group(1)
+            partnum = re.search(r'(\d+)$', targdev).group(1)
+            targblock = re.search(r'(.*)\d+$', targdev).group(1)
     if targblock:
         if targblock.endswith('p') and 'nvme' in targblock:
             targblock = targblock[:-1]
@@ -231,7 +231,7 @@ def install_to_disk(imgpath):
             deflvmsize += fs['initsize']
             minlvmsize += fs['minsize']
         else:
-            plainvols[int(re.search('(\d+)$', fs['device'])[0])] = fs
+            plainvols[int(re.search(r'(\d+)$', fs['device'])[0])] = fs
     with open('/tmp/installdisk') as diskin:
         instdisk = diskin.read()
     instdisk = '/dev/' + instdisk

confluent_osdeploy/ubuntu20.04-diskless/profiles/default/scripts/image2disk.py

@@ -204,8 +204,8 @@ def fixup(rootdir, vols):
     for vol in vols:
         if vol['mount'] == '/boot/efi':
             targdev = vol['targetdisk']
-            partnum = re.search('(\d+)$', targdev).group(1)
-            targblock = re.search('(.*)\d+$', targdev).group(1)
+            partnum = re.search(r'(\d+)$', targdev).group(1)
+            targblock = re.search(r'(.*)\d+$', targdev).group(1)
     if targblock:
         if targblock.endswith('p') and 'nvme' in targblock:
             targblock = targblock[:-1]
@@ -263,7 +263,7 @@ def install_to_disk(imgpath):
             deflvmsize += fs['initsize']
             minlvmsize += fs['minsize']
         else:
-            plainvols[int(re.search('(\d+)$', fs['device'])[0])] = fs
+            plainvols[int(re.search(r'(\d+)$', fs['device'])[0])] = fs
     with open('/tmp/installdisk') as diskin:
         instdisk = diskin.read()
     instdisk = '/dev/' + instdisk

confluent_server/builddeb

@@ -50,12 +50,30 @@ if ! grep wheezy /etc/os-release; then
 fi
 head -n -1 debian/control > debian/control1
 mv debian/control1 debian/control
-cat > debian/postinst << EOF
-if ! getent passwd confluent > /dev/null; then 
+cat > debian/postinst << \EOF
+if ! getent passwd confluent > /dev/null; then
     useradd -r confluent -d /var/lib/confluent -s /usr/sbin/nologin
-    mkdir -p /etc/confluent
-    chown confluent /etc/confluent
 fi
+mkdir -p /etc/confluent /var/lib/confluent /var/log/confluent /var/cache/confluent
+chown confluent:confluent /etc/confluent /var/lib/confluent /var/log/confluent /var/cache/confluent
+
+sysctl -p /usr/lib/sysctl.d/confluent.conf > /dev/null 2>&1
+NEEDCHOWN=0
+NEEDSTART=0
+[ -n "$(find /etc/confluent /var/log/confluent /var/cache/confluent -uid 0 -print -quit 2>/dev/null)" ] && NEEDCHOWN=1
+if [ $NEEDCHOWN = 1 ]; then
+    if systemctl is-active confluent > /dev/null; then
+        NEEDSTART=1
+        systemctl stop confluent
+    fi
+    chown -R confluent:confluent /etc/confluent /var/log/confluent /var/cache/confluent
+fi
+systemctl daemon-reload
+if systemctl is-active confluent > /dev/null || [ $NEEDSTART = 1 ]; then systemctl restart confluent > /dev/null 2>&1; fi
+if [ ! -e /etc/pam.d/confluent ]; then
+    ln -s /etc/pam.d/sshd /etc/pam.d/confluent
+fi
+true
 EOF
 echo 'export PYBUILD_INSTALL_ARGS=--install-lib=/opt/confluent/lib/python' >> debian/rules
 #echo 'Provides: python-'$DPKGNAME >> debian/control

confluent_server/confluent/certutil.py

@@ -207,85 +207,98 @@ def substitute_cfg(setting, key, val, newval, cfgfile, line):
     return False
 
 async def create_full_ca(certout):
-    mkdirp('/etc/confluent/tls/ca/private')
-    keyout = '/etc/confluent/tls/ca/private/cakey.pem'
-    csrout = '/etc/confluent/tls/ca/ca.csr'
-    mkdirp('/etc/confluent/tls/ca/newcerts')
-    with open('/etc/confluent/tls/ca/index.txt', 'w') as idx:
-        pass
-    with open('/etc/confluent/tls/ca/index.txt.attr', 'w') as idx:
-        idx.write('unique_subject = no')
-    with open('/etc/confluent/tls/ca/serial', 'w') as srl:
-        srl.write('01')
-    sslcfg = get_openssl_conf_location()
-    newcfg = '/etc/confluent/tls/ca/openssl.cfg'
-    settings = {
-        'dir': '/etc/confluent/tls/ca',
-        'certificate': '$dir/cacert.pem',
-        'private_key': '$dir/private/cakey.pem',
-        'countryName': 'optional',
-        'stateOrProvinceName': 'optional',
-        'organizationName': 'optional',
-    }
-    subj = '/CN=Confluent TLS Certificate authority ({0})'.format(socket.gethostname())
-    if len(subj) > 68:
-        subj = subj[:68]
-    with open(sslcfg, 'r') as cfgin:
-        with open(newcfg, 'w') as cfgfile:
-            for line in cfgin.readlines():
-                cfg = line.split('#')[0]
-                if '=' in cfg:
-                    key, val = cfg.split('=', 1)
-                    for stg in settings:
-                        if substitute_cfg(stg, key, val, settings[stg], cfgfile, line):
-                            break
-                    else:
-                        cfgfile.write(line.strip() + '\n')
-                    continue
-                cfgfile.write(line.strip() + '\n')
-            cfgfile.write('\n[CACert]\nbasicConstraints = critical,CA:true\nkeyUsage = critical,keyCertSign,cRLSign\n[ca_confluent]\n')
-    await util.check_call(
-        'openssl', 'ecparam', '-name', 'secp384r1', '-genkey', '-out',
-        keyout)
-    await util.check_call(
-        'openssl', 'req', '-new', '-key', keyout, '-out', csrout, '-subj', subj)
-    await util.check_call(
-        'openssl', 'ca', '-config', newcfg, '-batch', '-selfsign',
-        '-extensions', 'CACert', '-extfile', newcfg, 
-        '-notext', '-md', 'sha384', '-startdate',
-         '19700101010101Z', '-enddate', '21000101010101Z', '-keyfile',
-         keyout, '-out', '/etc/confluent/tls/ca/cacert.pem', '-in', csrout
-    )
-    shutil.copy2('/etc/confluent/tls/ca/cacert.pem', certout)
+    # The CA is used by the confluent service, which runs as the owner of
+    # /etc/confluent rather than root; create the CA material as that user
+    # so the service can use the database for issuing certificates
+    ouid = normalize_uid()
+    try:
+        mkdirp('/etc/confluent/tls/ca/private')
+        keyout = '/etc/confluent/tls/ca/private/cakey.pem'
+        csrout = '/etc/confluent/tls/ca/ca.csr'
+        mkdirp('/etc/confluent/tls/ca/newcerts')
+        with open('/etc/confluent/tls/ca/index.txt', 'w') as idx:
+            pass
+        with open('/etc/confluent/tls/ca/index.txt.attr', 'w') as idx:
+            idx.write('unique_subject = no')
+        with open('/etc/confluent/tls/ca/serial', 'w') as srl:
+            srl.write('01')
+        sslcfg = get_openssl_conf_location()
+        newcfg = '/etc/confluent/tls/ca/openssl.cfg'
+        settings = {
+            'dir': '/etc/confluent/tls/ca',
+            'certificate': '$dir/cacert.pem',
+            'private_key': '$dir/private/cakey.pem',
+            'countryName': 'optional',
+            'stateOrProvinceName': 'optional',
+            'organizationName': 'optional',
+        }
+        subj = '/CN=Confluent TLS Certificate authority ({0})'.format(socket.gethostname())
+        if len(subj) > 68:
+            subj = subj[:68]
+        with open(sslcfg, 'r') as cfgin:
+            with open(newcfg, 'w') as cfgfile:
+                for line in cfgin.readlines():
+                    cfg = line.split('#')[0]
+                    if '=' in cfg:
+                        key, val = cfg.split('=', 1)
+                        for stg in settings:
+                            if substitute_cfg(stg, key, val, settings[stg], cfgfile, line):
+                                break
+                        else:
+                            cfgfile.write(line.strip() + '\n')
+                        continue
+                    cfgfile.write(line.strip() + '\n')
+                cfgfile.write('\n[CACert]\nbasicConstraints = critical,CA:true\nkeyUsage = critical,keyCertSign,cRLSign\n[ca_confluent]\n')
+        await util.check_call(
+            'openssl', 'ecparam', '-name', 'secp384r1', '-genkey', '-out',
+            keyout)
+        await util.check_call(
+            'openssl', 'req', '-new', '-key', keyout, '-out', csrout, '-subj', subj)
+        await util.check_call(
+            'openssl', 'ca', '-config', newcfg, '-batch', '-selfsign',
+            '-extensions', 'CACert', '-extfile', newcfg,
+            '-notext', '-md', 'sha384', '-startdate',
+             '19700101010101Z', '-enddate', '21000101010101Z', '-keyfile',
+             keyout, '-out', '/etc/confluent/tls/ca/cacert.pem', '-in', csrout
+        )
+        shutil.copy2('/etc/confluent/tls/ca/cacert.pem', certout)
+    finally:
+        os.seteuid(ouid)
 #openssl ca -config openssl.cnf -selfsign -keyfile cakey.pem -startdate 20150214120000Z -enddate 20160214120000Z
 #20160107071311Z -enddate 20170106071311Z
 
 async def create_simple_ca(keyout, certout):
+    # As with create_full_ca, the CA material must be owned by the owner
+    # of /etc/confluent for use by the confluent service
+    ouid = normalize_uid()
     try:
-        os.makedirs('/etc/confluent/tls')
-    except OSError as e:
-        if e.errno != 17:
-            raise
-    sslcfg = get_openssl_conf_location()
-    tmphdl, tmpconfig = tempfile.mkstemp()
-    os.close(tmphdl)
-    shutil.copy2(sslcfg, tmpconfig)
-    await util.check_call(
-            'openssl', 'ecparam', '-name', 'secp384r1', '-genkey', '-out',
-            keyout)
-    try:
-        subj = '/CN=Confluent TLS Certificate authority ({0})'.format(socket.gethostname())
-        if len(subj) > 68:
-            subj = subj[:68]
-        with open(tmpconfig, 'a') as cfgfile:
-            cfgfile.write('\n[CACert]\nbasicConstraints = critical,CA:true\n')
+        try:
+            os.makedirs('/etc/confluent/tls')
+        except OSError as e:
+            if e.errno != 17:
+                raise
+        sslcfg = get_openssl_conf_location()
+        tmphdl, tmpconfig = tempfile.mkstemp()
+        os.close(tmphdl)
+        shutil.copy2(sslcfg, tmpconfig)
         await util.check_call(
-                'openssl', 'req', '-new', '-x509', '-key', keyout, '-days',
-                '27300', '-out', certout, '-subj', subj,
-                '-extensions', 'CACert', '-config', tmpconfig
-            )
+                'openssl', 'ecparam', '-name', 'secp384r1', '-genkey', '-out',
+                keyout)
+        try:
+            subj = '/CN=Confluent TLS Certificate authority ({0})'.format(socket.gethostname())
+            if len(subj) > 68:
+                subj = subj[:68]
+            with open(tmpconfig, 'a') as cfgfile:
+                cfgfile.write('\n[CACert]\nbasicConstraints = critical,CA:true\n')
+            await util.check_call(
+                    'openssl', 'req', '-new', '-x509', '-key', keyout, '-days',
+                    '27300', '-out', certout, '-subj', subj,
+                    '-extensions', 'CACert', '-config', tmpconfig
+                )
+        finally:
+            os.remove(tmpconfig)
     finally:
-        os.remove(tmpconfig)
+        os.seteuid(ouid)
 
 async def create_certificate(keyout=None, certout=None, csrfile=None, subj=None, san=None, backdate=True, days=None):
     now_utc = datetime.datetime.now(datetime.timezone.utc)
@@ -413,12 +426,30 @@ async def create_certificate(keyout=None, certout=None, csrfile=None, subj=None,
                 shutil.copy2(cacfgfile, tmpcafile)
                 os.close(tmphdl)
                 cacfgfile = tmpcafile
-            await util.check_call(
-                'openssl', 'ca', '-config', cacfgfile, '-rand_serial',
-                '-in', csrfile, '-out', certout, '-batch', '-notext',
-                '-startdate', startdate, '-enddate', enddate, '-md', 'sha384',
-                '-extfile', extconfig, '-subj', subj
-            )
+                os.chmod(cacfgfile, 0o644)
+                os.chmod(csrfile, 0o644)
+            # openssl ca rewrites the CA database (index, serial) as the
+            # invoking user; run it as the owner of /etc/confluent so the
+            # database remains usable by the confluent service. The chmodded
+            # temporary inputs hold no secrets, and the certificate is
+            # written to a temporary path first, as certout may only be
+            # writable by the original user (e.g. a web server certificate
+            # path during osdeploy initialize -t)
+            os.chmod(extconfig, 0o644)
+            ouid = normalize_uid()
+            try:
+                tmphdl, tmpcertout = tempfile.mkstemp()
+                os.close(tmphdl)
+                await util.check_call(
+                    'openssl', 'ca', '-config', cacfgfile, '-rand_serial',
+                    '-in', csrfile, '-out', tmpcertout, '-batch', '-notext',
+                    '-startdate', startdate, '-enddate', enddate, '-md', 'sha384',
+                    '-extfile', extconfig, '-subj', subj
+                )
+            finally:
+                os.seteuid(ouid)
+            shutil.copy(tmpcertout, certout)
+            os.remove(tmpcertout)
         for keycopy in tlsmateriallocation.get('keys', []):
             if keycopy != keyout:
                 shutil.copy2(keyout, keycopy)

confluent_server/confluent/collective/manager.py

@@ -206,28 +206,30 @@ async def follow_leader(remote, leader):
         exitcause = await cfm.follow_channel(remote)
         newleader = exitcause.get('newleader', None)
     finally:
+        handled = False
         if cleanexit:
             log.log({'info': 'Previous following cleanly closed',
                      'subsystem': 'collective'})
-            return
-        if newleader:
+            handled = True
+        if not handled and newleader:
             log.log(
                 {'info': 'Previous leader directed us to join new leader {}'.format(newleader)})
             try:
                 if await connect_to_leader(None, get_myname(), newleader):
-                    return
+                    handled = True
             except Exception:
                 log.log({'error': 'Unknown error attempting to connect to {}, check trace log'.format(newleader), 'subsystem': 'collective'})
                 cfm.logException()
-        log.log({'info': 'Current leader ({0}) has disappeared, restarting '
-                         'collective membership'.format(leader), 'subsystem': 'collective'})
-        # The leader has folded, time to startup again...
-        follower = None
-        await cfm.stop_following()
-        currentleader = None
-        if retrythread is None:  # start a recovery
-            retrythread = tasks.spawn_task_after(
-                random.random(), start_collective)
+        if not handled:
+            log.log({'info': 'Current leader ({0}) has disappeared, restarting '
+                             'collective membership'.format(leader), 'subsystem': 'collective'})
+            # The leader has folded, time to startup again...
+            follower = None
+            await cfm.stop_following()
+            currentleader = None
+            if retrythread is None:  # start a recovery
+                retrythread = tasks.spawn_task_after(
+                    random.random(), start_collective)
 
 async def _create_tls_connection(host, port):
     cloop = asyncio.get_running_loop()