From dda1087a2e6e0f5fcb8623983f5296b8a426cfc5 Mon Sep 17 00:00:00 2001
From: Christian Heimes <cheimes@redhat.com>
Date: Fri, 8 Dec 2017 12:38:41 +0100
Subject: [PATCH] Add safe DirectiveSetter context manager
installutils.set_directive() is both inefficient and potentially
dangerous. It does not ensure that the whole file is written and
properly synced to disk. In worst case it could lead to partially
written or destroyed config files.
The new DirectiveSetter context manager wraps everything under an easy
to use interface.
https://pagure.io/freeipa/issue/7312
Signed-off-by: Christian Heimes <cheimes@redhat.com>
Reviewed-By: Florence Blanc-Renaud <frenaud@redhat.com>
---
ipaserver/install/cainstance.py | 109 ++++++++++-----------
ipaserver/install/installutils.py | 87 +++++++++++++++-
.../test_install/test_installutils.py | 67 +++++++++++++
3 files changed, 205 insertions(+), 58 deletions(-)
diff --git a/ipaserver/install/cainstance.py b/ipaserver/install/cainstance.py
index dd410d1c97cb4b27d35086bb2f511c42c02d022f..20635eae22268ff72de73b8b9c430050114bb45b 100644
--- a/ipaserver/install/cainstance.py
+++ b/ipaserver/install/cainstance.py
@@ -908,62 +908,61 @@ class CAInstance(DogtagInstance):
https://access.redhat.com/knowledge/docs/en-US/Red_Hat_Certificate_System/8.0/html/Admin_Guide/Setting_up_Publishing.html
"""
- def put(k, v):
- installutils.set_directive(
- paths.CA_CS_CFG_PATH, k, v, quotes=False, separator='=')
+ with installutils.DirectiveSetter(paths.CA_CS_CFG_PATH,
+ quotes=False, separator='=') as ds:
- # Enable file publishing, disable LDAP
- put('ca.publish.enable', 'true')
- put('ca.publish.ldappublish.enable', 'false')
-
- # Create the file publisher, der only, not b64
- put('ca.publish.publisher.impl.FileBasedPublisher.class',
- 'com.netscape.cms.publish.publishers.FileBasedPublisher')
- put('ca.publish.publisher.instance.FileBaseCRLPublisher.crlLinkExt',
- 'bin')
- put('ca.publish.publisher.instance.FileBaseCRLPublisher.directory',
- self.prepare_crl_publish_dir())
- put('ca.publish.publisher.instance.FileBaseCRLPublisher.latestCrlLink',
- 'true')
- put('ca.publish.publisher.instance.FileBaseCRLPublisher.pluginName',
- 'FileBasedPublisher')
- put('ca.publish.publisher.instance.FileBaseCRLPublisher.timeStamp',
- 'LocalTime')
- put('ca.publish.publisher.instance.FileBaseCRLPublisher.zipCRLs',
- 'false')
- put('ca.publish.publisher.instance.FileBaseCRLPublisher.zipLevel', '9')
- put('ca.publish.publisher.instance.FileBaseCRLPublisher.Filename.b64',
- 'false')
- put('ca.publish.publisher.instance.FileBaseCRLPublisher.Filename.der',
- 'true')
-
- # The publishing rule
- put('ca.publish.rule.instance.FileCrlRule.enable', 'true')
- put('ca.publish.rule.instance.FileCrlRule.mapper', 'NoMap')
- put('ca.publish.rule.instance.FileCrlRule.pluginName', 'Rule')
- put('ca.publish.rule.instance.FileCrlRule.predicate', '')
- put('ca.publish.rule.instance.FileCrlRule.publisher',
- 'FileBaseCRLPublisher')
- put('ca.publish.rule.instance.FileCrlRule.type', 'crl')
-
- # Now disable LDAP publishing
- put('ca.publish.rule.instance.LdapCaCertRule.enable', 'false')
- put('ca.publish.rule.instance.LdapCrlRule.enable', 'false')
- put('ca.publish.rule.instance.LdapUserCertRule.enable', 'false')
- put('ca.publish.rule.instance.LdapXCertRule.enable', 'false')
-
- # If we are the initial master then we are the CRL generator, otherwise
- # we point to that master for CRLs.
- if not self.clone:
- # These next two are defaults, but I want to be explicit that the
- # initial master is the CRL generator.
- put('ca.crl.MasterCRL.enableCRLCache', 'true')
- put('ca.crl.MasterCRL.enableCRLUpdates', 'true')
- put('ca.listenToCloneModifications', 'true')
- else:
- put('ca.crl.MasterCRL.enableCRLCache', 'false')
- put('ca.crl.MasterCRL.enableCRLUpdates', 'false')
- put('ca.listenToCloneModifications', 'false')
+ # Enable file publishing, disable LDAP
+ ds.set('ca.publish.enable', 'true')
+ ds.set('ca.publish.ldappublish.enable', 'false')
+
+ # Create the file publisher, der only, not b64
+ ds.set(
+ 'ca.publish.publisher.impl.FileBasedPublisher.class',
+ 'com.netscape.cms.publish.publishers.FileBasedPublisher'
+ )
+ prefix = 'ca.publish.publisher.instance.FileBaseCRLPublisher.'
+ ds.set(prefix + 'crlLinkExt', 'bin')
+ ds.set(prefix + 'directory', self.prepare_crl_publish_dir())
+ ds.set(prefix + 'latestCrlLink', 'true')
+ ds.set(prefix + 'pluginName', 'FileBasedPublisher')
+ ds.set(prefix + 'timeStamp', 'LocalTime')
+ ds.set(prefix + 'zipCRLs', 'false')
+ ds.set(prefix + 'zipLevel', '9')
+ ds.set(prefix + 'Filename.b64', 'false')
+ ds.set(prefix + 'Filename.der', 'true')
+
+ # The publishing rule
+ ds.set('ca.publish.rule.instance.FileCrlRule.enable', 'true')
+ ds.set('ca.publish.rule.instance.FileCrlRule.mapper', 'NoMap')
+ ds.set('ca.publish.rule.instance.FileCrlRule.pluginName', 'Rule')
+ ds.set('ca.publish.rule.instance.FileCrlRule.predicate', '')
+ ds.set(
+ 'ca.publish.rule.instance.FileCrlRule.publisher',
+ 'FileBaseCRLPublisher'
+ )
+ ds.set('ca.publish.rule.instance.FileCrlRule.type', 'crl')
+
+ # Now disable LDAP publishing
+ ds.set('ca.publish.rule.instance.LdapCaCertRule.enable', 'false')
+ ds.set('ca.publish.rule.instance.LdapCrlRule.enable', 'false')
+ ds.set(
+ 'ca.publish.rule.instance.LdapUserCertRule.enable',
+ 'false'
+ )
+ ds.set('ca.publish.rule.instance.LdapXCertRule.enable', 'false')
+
+ # If we are the initial master then we are the CRL generator,
+ # otherwise we point to that master for CRLs.
+ if not self.clone:
+ # These next two are defaults, but I want to be explicit
+ # that the initial master is the CRL generator.
+ ds.set('ca.crl.MasterCRL.enableCRLCache', 'true')
+ ds.set('ca.crl.MasterCRL.enableCRLUpdates', 'true')
+ ds.set('ca.listenToCloneModifications', 'true')
+ else:
+ ds.set('ca.crl.MasterCRL.enableCRLCache', 'false')
+ ds.set('ca.crl.MasterCRL.enableCRLUpdates', 'false')
+ ds.set('ca.listenToCloneModifications', 'false')
def uninstall(self):
# just eat state
diff --git a/ipaserver/install/installutils.py b/ipaserver/install/installutils.py
index b56cf591496c679e5fcf3e94f458c286216eb1e4..08e098f8656c8fe61a87fb046c01e1487c25da7f 100644
--- a/ipaserver/install/installutils.py
+++ b/ipaserver/install/installutils.py
@@ -24,6 +24,7 @@ import errno
import socket
import getpass
import gssapi
+import io
import ldif
import os
import re
@@ -31,6 +32,7 @@ import fileinput
import sys
import tempfile
import shutil
+import stat # pylint: disable=bad-python3-import
import traceback
import textwrap
from contextlib import contextmanager
@@ -428,6 +430,82 @@ def unquote_directive_value(value, quote_char):
return unescaped_value
+_SENTINEL = object()
+
+
+class DirectiveSetter(object):
+ """Safe directive setter
+
+ with DirectiveSetter('/path/to/conf') as ds:
+ ds.set(key, value)
+ """
+ def __init__(self, filename, quotes=True, separator=' '):
+ self.filename = os.path.abspath(filename)
+ self.quotes = quotes
+ self.separator = separator
+ self.lines = None
+ self.stat = None
+
+ def __enter__(self):
+ with io.open(self.filename) as f:
+ self.stat = os.fstat(f.fileno())
+ self.lines = list(f)
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ if exc_type is not None:
+ # something went wrong, reset
+ self.lines = None
+ self.stat = None
+ return
+
+ directory, prefix = os.path.split(self.filename)
+ # use tempfile in same directory to have atomic rename
+ fd, name = tempfile.mkstemp(prefix=prefix, dir=directory, text=True)
+ with io.open(fd, mode='w', closefd=True) as f:
+ for line in self.lines:
+ if not isinstance(line, six.text_type):
+ line = line.decode('utf-8')
+ f.write(line)
+ self.lines = None
+ os.fchmod(f.fileno(), stat.S_IMODE(self.stat.st_mode))
+ os.fchown(f.fileno(), self.stat.st_uid, self.stat.st_gid)
+ self.stat = None
+ # flush and sync tempfile inode
+ f.flush()
+ os.fsync(f.fileno())
+
+ # rename file and sync directory inode
+ os.rename(name, self.filename)
+ dirfd = os.open(directory, os.O_RDONLY | os.O_DIRECTORY)
+ try:
+ os.fsync(dirfd)
+ finally:
+ os.close(dirfd)
+
+ def set(self, directive, value, quotes=_SENTINEL, separator=_SENTINEL):
+ """Set a single directive
+ """
+ if quotes is _SENTINEL:
+ quotes = self.quotes
+ if separator is _SENTINEL:
+ separator = self.separator
+ # materialize lines
+ # set_directive_lines() modify item, shrink or enlage line count
+ self.lines = list(set_directive_lines(
+ quotes, separator, directive, value, self.lines
+ ))
+
+ def setitems(self, items):
+ """Set multiple directives from a dict or list with key/value pairs
+ """
+ if isinstance(items, dict):
+ # dict-like, use sorted for stable order
+ items = sorted(items.items())
+ for k, v in items:
+ self.set(k, v)
+
+
def set_directive(filename, directive, value, quotes=True, separator=' '):
"""Set a name/value pair directive in a configuration file.
@@ -447,8 +525,10 @@ def set_directive(filename, directive, value, quotes=True, separator=' '):
st = os.stat(filename)
with open(filename, 'r') as f:
lines = list(f) # read the whole file
- new_lines = set_directive_lines(
- quotes, separator, directive, value, lines)
+ # materialize new list
+ new_lines = list(set_directive_lines(
+ quotes, separator, directive, value, lines
+ ))
with open(filename, 'w') as f:
# don't construct the whole string; write line-wise
for line in new_lines:
@@ -472,8 +552,9 @@ def set_directive_lines(quotes, separator, k, v, lines):
new_line = ''.join([k, separator, v_quoted, '\n'])
found = False
+ matcher = re.compile(r'\s*{}'.format(re.escape(k + separator)))
for line in lines:
- if re.match(r'\s*{}'.format(re.escape(k + separator)), line):
+ if matcher.match(line):
found = True
if v is not None:
yield new_line
diff --git a/ipatests/test_ipaserver/test_install/test_installutils.py b/ipatests/test_ipaserver/test_install/test_installutils.py
index cc8fd3cf3f82c2d9af48287f506a566ffbfc39f6..3c992c9ab06ddc3af557329de81debe704b0fb16 100644
--- a/ipatests/test_ipaserver/test_install/test_installutils.py
+++ b/ipatests/test_ipaserver/test_install/test_installutils.py
@@ -3,8 +3,11 @@
#
import os
+import shutil
import tempfile
+import pytest
+
from ipaserver.install import installutils
EXAMPLE_CONFIG = [
@@ -13,6 +16,17 @@ EXAMPLE_CONFIG = [
]
+@pytest.fixture
+def tempdir(request):
+ tempdir = tempfile.mkdtemp()
+
+ def fin():
+ shutil.rmtree(tempdir)
+
+ request.addfinalizer(fin)
+ return tempdir
+
+
class test_set_directive_lines(object):
def test_remove_directive(self):
lines = installutils.set_directive_lines(
@@ -55,3 +69,56 @@ class test_set_directive(object):
finally:
os.remove(filename)
+
+
+def test_directivesetter(tempdir):
+ filename = os.path.join(tempdir, 'example.conf')
+ with open(filename, 'w') as f:
+ for line in EXAMPLE_CONFIG:
+ f.write(line)
+
+ ds = installutils.DirectiveSetter(filename)
+ assert ds.lines is None
+ with ds:
+ assert ds.lines == EXAMPLE_CONFIG
+ ds.set('foo', '3') # quoted, space separated, doesn't change 'foo='
+ ds.set('foobar', None, separator='=') # remove
+ ds.set('baz', '4', False, '=') # add
+ ds.setitems([
+ ('list1', 'value1'),
+ ('list2', 'value2'),
+ ])
+ ds.setitems({
+ 'dict1': 'value1',
+ 'dict2': 'value2',
+ })
+
+ with open(filename, 'r') as f:
+ lines = list(f)
+
+ assert lines == [
+ 'foo=1\n',
+ 'foo "3"\n',
+ 'baz=4\n',
+ 'list1 "value1"\n',
+ 'list2 "value2"\n',
+ 'dict1 "value1"\n',
+ 'dict2 "value2"\n',
+ ]
+
+ with installutils.DirectiveSetter(filename, True, '=') as ds:
+ ds.set('foo', '4') # doesn't change 'foo '
+
+ with open(filename, 'r') as f:
+ lines = list(f)
+
+ assert lines == [
+ 'foo="4"\n',
+ 'foo "3"\n',
+ 'baz=4\n',
+ 'list1 "value1"\n',
+ 'list2 "value2"\n',
+ 'dict1 "value1"\n',
+ 'dict2 "value2"\n',
+
+ ]
--
2.13.6