NSSDB: Let certutil decide its default db type

CertDB no longer makes any assumptions about the default db type of a NSS
DB. Instead it let's certutil decide when dbtype is set to 'auto'. This
makes it much easier to support F27 and F28 from a single code base.

Signed-off-by: Christian Heimes <cheimes@redhat.com>
Reviewed-By: Stanislav Laznicka <slaznick@redhat.com>

ipaplatform/_importhook.py

@@ -70,7 +70,7 @@ class IpaMetaImporter(object):
         if sys.platform.startswith('linux'):
             # Linux, get distribution from /etc/os-release
             try:
-                platforms.extend(self._parse_osrelease())
+                platforms.extend(self._parse_platform())
             except Exception as e:
                 warnings.warn("Failed to read /etc/os-release: {}".format(e))
         elif sys.platform == 'win32':
@@ -91,14 +91,17 @@ class IpaMetaImporter(object):
 
         return platforms
 
-    def _parse_osrelease(self, filename='/etc/os-release'):
+    def parse_osrelease(self, filename='/etc/os-release'):
         release = {}
         with io.open(filename, encoding='utf-8') as f:
             for line in f:
                 mo = _osrelease_line.match(line)
                 if mo is not None:
                     release[mo.group('name')] = mo.group('value')
+        return release
 
+    def _parse_platform(self, filename='/etc/os-release'):
+        release = self.parse_osrelease(filename)
         platforms = [
             release['ID'],
         ]

ipaplatform/base/constants.py

@@ -37,8 +37,6 @@ class BaseConstantsNamespace(object):
         'httpd_dbus_sssd': 'on',
     }
     SSSD_USER = "sssd"
-    # sql (new format), dbm (old format)
-    NSS_DEFAULT_DBTYPE = 'dbm'
     # WSGI module override, only used on Fedora
     MOD_WSGI_PYTHON2 = None
     MOD_WSGI_PYTHON3 = None

ipapython/certdb.py

@@ -224,18 +224,11 @@ class NSSDatabase(object):
         if nssdir is None:
             self.secdir = tempfile.mkdtemp()
             self._is_temporary = True
-            if dbtype == 'auto':
-                dbtype = constants.NSS_DEFAULT_DBTYPE
         else:
             self.secdir = nssdir
             self._is_temporary = False
             if dbtype == 'auto':
-                if os.path.isfile(os.path.join(self.secdir, "cert9.db")):
-                    dbtype = 'sql'
-                elif os.path.isfile(os.path.join(self.secdir, "cert8.db")):
-                    dbtype = 'dbm'
-                else:
-                    dbtype = constants.NSS_DEFAULT_DBTYPE
+                dbtype = self._detect_dbtype()
 
         self.pwd_file = os.path.join(self.secdir, 'pwdfile.txt')
         self.dbtype = None
@@ -246,6 +239,14 @@ class NSSDatabase(object):
         self.backup_filenames = ()
         self._set_filenames(dbtype)
 
+    def _detect_dbtype(self):
+        if os.path.isfile(os.path.join(self.secdir, "cert9.db")):
+            return 'sql'
+        elif os.path.isfile(os.path.join(self.secdir, "cert8.db")):
+            return 'dbm'
+        else:
+            return 'auto'
+
     def _set_filenames(self, dbtype):
         self.dbtype = dbtype
         dbmfiles = (
@@ -260,16 +261,15 @@ class NSSDatabase(object):
         )
         if dbtype == 'dbm':
             self.certdb, self.keydb, self.secmod = dbmfiles
+            self.filenames = dbmfiles + (self.pwd_file,)
         elif dbtype == 'sql':
             self.certdb, self.keydb, self.secmod = sqlfiles
+            self.filenames = sqlfiles + (self.pwd_file,)
+        elif dbtype == 'auto':
+            self.certdb = self.keydb = self.secmod = None
+            self.filenames = None
         else:
             raise ValueError(dbtype)
-        self.filenames = (
-            self.certdb,
-            self.keydb,
-            self.secmod,
-            self.pwd_file,
-        )
         self.backup_filenames = (
             self.pwd_file,
         ) + sqlfiles + dbmfiles
@@ -284,7 +284,14 @@ class NSSDatabase(object):
     def __exit__(self, type, value, tb):
         self.close()
 
+    def _check_db(self):
+        if self.filenames is None:
+            raise RuntimeError(
+                "NSSDB '{}' not initialized.".format(self.secdir)
+            )
+
     def run_certutil(self, args, stdin=None, **kwargs):
+        self._check_db()
         new_args = [
             paths.CERTUTIL,
             "-d", '{}:{}'.format(self.dbtype, self.secdir)
@@ -294,6 +301,7 @@ class NSSDatabase(object):
         return ipautil.run(new_args, stdin, **kwargs)
 
     def run_pk12util(self, args, stdin=None, **kwargs):
+        self._check_db()
         new_args = [
             paths.PK12UTIL,
             "-d", '{}:{}'.format(self.dbtype, self.secdir)
@@ -302,6 +310,7 @@ class NSSDatabase(object):
         return ipautil.run(new_args, stdin, **kwargs)
 
     def run_modutil(self, args, stdin=None, **kwargs):
+        self._check_db()
         new_args = [
             paths.MODUTIL,
             '-dbdir', '{}:{}'.format(self.dbtype, self.secdir)
@@ -312,6 +321,8 @@ class NSSDatabase(object):
     def exists(self):
         """Check DB exists (all files are present)
         """
+        if self.filenames is None:
+            return False
         return all(os.path.isfile(filename) for filename in self.filenames)
 
     def create_db(self, user=None, group=None, mode=None, backup=False):
@@ -353,8 +364,26 @@ class NSSDatabase(object):
                 f.write(ipautil.ipa_generate_password())
                 f.flush()
 
-        # -@ in case it's an old db and it must be migrated
-        self.run_certutil(['-N', '-@', self.pwd_file])
+        # In case dbtype is auto, let certutil decide which type of DB
+        # to create.
+        if self.dbtype == 'auto':
+            dbdir = self.secdir
+        else:
+            dbdir = '{}:{}'.format(self.dbtype, self.secdir)
+        ipautil.run([
+            paths.CERTUTIL,
+            '-d', dbdir,
+            '-N',
+            '-f', self.pwd_file,
+            # -@ in case it's an old db and it must be migrated
+            '-@', self.pwd_file,
+        ])
+        self._set_filenames(self._detect_dbtype())
+        if self.filenames is None:
+            # something went wrong...
+            raise ValueError(
+                "Failed to create NSSDB at '{}'".format(self.secdir)
+            )
 
         # Finally fix up perms
         os.chown(self.secdir, uid, gid)
@@ -420,15 +449,14 @@ class NSSDatabase(object):
                 os.rename(oldname, oldname + '.migrated')
 
     def restore(self):
-        for filename in self.filenames:
-            path = os.path.join(self.secdir, filename)
-            backup_path = path + '.orig'
-            save_path = path + '.ipasave'
+        for filename in self.backup_filenames:
+            backup_path = filename + '.orig'
+            save_path = filename + '.ipasave'
             try:
-                if os.path.exists(path):
-                    os.rename(path, save_path)
+                if os.path.exists(filename):
+                    os.rename(filename, save_path)
                 if os.path.exists(backup_path):
-                    os.rename(backup_path, path)
+                    os.rename(backup_path, filename)
             except OSError as e:
                 logger.debug('%s', e)
 

ipaserver/install/certs.py

@@ -45,7 +45,6 @@ from ipalib.errors import CertificateOperationError
 from ipalib.install import certstore
 from ipalib.util import strip_csr_header
 from ipalib.text import _
-from ipaplatform.constants import constants
 from ipaplatform.paths import paths
 
 
@@ -725,7 +724,6 @@ class CertDB(object):
         """
         return (
             self.nssdb.dbtype == 'dbm' and
-            self.nssdb.dbtype != constants.NSS_DEFAULT_DBTYPE and
             self.exists()
         )
 

ipatests/pytest_plugins/integration/tasks.py

@@ -35,7 +35,6 @@ from six import StringIO
 
 from ipapython import ipautil
 from ipaplatform.paths import paths
-from ipaplatform.constants import constants
 from ipapython.dn import DN
 from ipalib import errors
 from ipalib.util import get_reverse_zone_default, verify_host_resolvable
@@ -1267,9 +1266,8 @@ def run_server_del(host, server_to_delete, force=False,
 
 def run_certutil(host, args, reqdir, dbtype=None,
                  stdin=None, raiseonerr=True):
-    if dbtype is None:
-        dbtype = constants.NSS_DEFAULT_DBTYPE
-    new_args = [paths.CERTUTIL, '-d', '{}:{}'.format(dbtype, reqdir)]
+    dbdir = reqdir if dbtype is None else '{}:{}'.format(dbtype, reqdir)
+    new_args = [paths.CERTUTIL, '-d', dbdir]
     new_args.extend(args)
     return host.run_command(new_args, raiseonerr=raiseonerr,
                             stdin_text=stdin)

ipatests/test_ipaplatform/test_importhook.py

@@ -50,5 +50,5 @@ def test_importhook(mod, name):
     (os.path.join(DATA, 'os-release-ubuntu'), ['ubuntu', 'debian']),
 ])
 def test_parse_os_release(filename, expected_platforms):
-    parsed = metaimporter._parse_osrelease(filename)
+    parsed = metaimporter._parse_platform(filename)
     assert parsed == expected_platforms

ipatests/test_ipapython/test_certdb.py

@@ -1,9 +1,21 @@
 import os
 
-from ipapython.certdb import NSSDatabase, TRUSTED_PEER_TRUST_FLAGS
+import pytest
 
+from ipapython.certdb import NSSDatabase, TRUSTED_PEER_TRUST_FLAGS
+from ipaplatform._importhook import metaimporter
+
+OSRELEASE = metaimporter.parse_osrelease()
 CERTNICK = 'testcert'
 
+if OSRELEASE['ID'] == 'fedora':
+    if int(OSRELEASE['VERSION_ID']) >= 28:
+        NSS_DEFAULT = 'sql'
+    else:
+        NSS_DEFAULT = 'dbm'
+else:
+    NSS_DEFAULT = None
+
 
 def create_selfsigned(nssdb):
     # create self-signed cert + key
@@ -137,3 +149,20 @@ def test_convert_db_nokey():
         assert nssdb.certdb in nssdb.filenames
         assert os.path.basename(nssdb.keydb) == 'key4.db'
         assert os.path.basename(nssdb.secmod) == 'pkcs11.txt'
+
+
+def test_auto_db():
+    with NSSDatabase() as nssdb:
+        assert nssdb.dbtype == 'auto'
+        assert nssdb.filenames is None
+        assert not nssdb.exists()
+        with pytest.raises(RuntimeError):
+            nssdb.list_certs()
+
+        nssdb.create_db()
+        assert nssdb.dbtype in ('dbm', 'sql')
+        if NSS_DEFAULT is not None:
+            assert nssdb.dbtype == NSS_DEFAULT
+        assert nssdb.filenames is not None
+        assert nssdb.exists()
+        nssdb.list_certs()