Merge pull request #9 from secynic/dev

CHANGES.rst

@@ -6,6 +6,7 @@ Changelog
 
 - Added syslog-ng support (#2)
 - Added sudo arg to utils.popen_wrapper() - code consolidation
+- Fixed bytes to str decoding issue on Python 3
 - Fixed splitlines list[bytes] decode on Python 3
 - Logging output tweaks
 - Fixed redundant TCPDump.check_packet_print() in nfsinkhole-setup.py

README.rst

@@ -17,16 +17,8 @@ nfsinkhole
 .. image:: https://img.shields.io/badge/docs-dev-yellow.svg?style=flat
     :target: https://nfsinkhole.readthedocs.io/en/dev
 
-.. warning::
-
-    This version is considered experimental. Do not attempt to use this
-    library in production until tests via travis and docker are setup, stable,
-    and sufficiently covered.
-
-.. attention::
-
-    You are responsible for rotating log files (/var/log/nfsinkhole*), and
-    syslog forwarding must be configured manually (automation pending).
+Summary
+=======
 
 nfsinkhole is a Python library and scripts for setting up a Unix server
 as a sinkhole (monitor, log/capture, and drop all traffic to a secondary
@@ -41,6 +33,17 @@ you can enable tcpdump to output packet capture text to
 /var/log/nfsinkhole-pcap.log if your version of tcpdump supports packet
 printing; otherwise reverts to /var/log/nfsinkhole.pcap.
 
+.. warning::
+
+    This version is considered experimental. Do not attempt to use this
+    library in production until tests via travis and docker are setup, stable,
+    and sufficiently covered.
+
+.. attention::
+
+    You are responsible for rotating log files (/var/log/nfsinkhole*), and
+    syslog forwarding must be configured manually (automation pending).
+
 Features
 ========
 

nfsinkhole/iptables.py

@@ -108,14 +108,14 @@ def list_existing_rules(self, filter_io_drop=False):
 
             out, err = popen_wrapper(cmd, sudo=True)
 
-        except OSError as e:
+        except OSError as e:  # pragma: no cover
 
             raise IPTablesError('Error encountered when running process "{0}":'
                                 '\n{1}'.format(' '.join(cmd), e))
 
         # If any errors, iterate them and write to log, then raise
         # IPTablesError.
-        if err:
+        if err:  # pragma: no cover
 
             arr = err.splitlines()
             raise IPTablesError('Error encountered when running process "{0}":'
@@ -265,7 +265,7 @@ def create_drop_rule(self):
 
             popen_wrapper(cmd_arr=tmp_arr, raise_err=True, sudo=True)
 
-        except SubprocessError as e:
+        except SubprocessError as e:  # pragma: no cover
 
             raise IPTablesError(e)
 

nfsinkhole/tests/__init__.py

@@ -35,3 +35,10 @@ def assertIsInstance(self, obj, cls, msg=None):
                     msg,
                     '{0} is not an instance of {1}'.format(obj, cls)
                 ))
+
+    if not hasattr(unittest.TestCase, 'assertSequenceEqual'):
+        def assertSequenceEqual(self, seq1, seq2, msg=None, seq_type=None):
+
+            # Basic Python 2.6 check instead of copying the whole
+            # assertSequenceEqual function from later Python versions
+            self.assertEqual(list(seq1), list(seq2))

nfsinkhole/tests/test_iptables.py

@@ -12,14 +12,13 @@
 
 class TestIPTablesSinkhole(TestCommon):
 
-    def test_drop(self):
+    def _test_create_drop_rule(self):
 
         # Bad interface
         myobj = IPTablesSinkhole(
             interface='~!@#$%^&*()',
         )
         myobj.create_drop_rule()
-        self.assertRaises(IPTablesExists, myobj.create_drop_rule)
 
         # Success
         myobj = IPTablesSinkhole(
@@ -29,3 +28,79 @@ def test_drop(self):
 
         # Exists
         self.assertRaises(IPTablesExists, myobj.create_drop_rule)
+
+        # Content check
+        expected = [
+            '-A INPUT -i eth1 -j DROP',
+            '-A OUTPUT -o eth1 -j DROP'
+        ]
+        existing = myobj.list_existing_rules(filter_io_drop=True)
+        self.assertSequenceEqual(existing, expected, seq_type=list)
+
+    def _test_delete_drop_rule(self):
+
+        # Bad interface
+        myobj = IPTablesSinkhole(
+            interface='~!@#$%^&*()',
+        )
+        myobj.delete_drop_rule()
+        self.assertRaises(IPTablesNotExists, myobj.delete_drop_rule)
+
+        # Success
+        myobj = IPTablesSinkhole(
+            interface='eth1',
+        )
+        myobj.delete_drop_rule()
+
+        # Exists
+        self.assertRaises(IPTablesNotExists, myobj.delete_drop_rule)
+
+    def test_drop_rule(self):
+
+        self._test_create_drop_rule()
+        self._test_delete_drop_rule()
+
+    def _test_create_rules(self):
+
+        # Success
+        myobj = IPTablesSinkhole(
+            interface='eth1',
+            interface_addr='127.0.0.1',
+            protocol='tcp',
+            dport='0:53'
+        )
+        myobj.create_rules()
+
+        # Exists
+        self.assertRaises(IPTablesExists, myobj.create_rules)
+
+        # Content check
+        expected = [
+            '-N SINKHOLE',
+            '-A INPUT -d 127.0.0.1/32 -i eth1 -p tcp -m hashlimit '
+            '--hashlimit-upto 1/hour --hashlimit-burst 1 --hashlimit-mode '
+            'srcip,dstip,dstport --hashlimit-name sinkhole -m multiport '
+            '--dports 0:53 -j SINKHOLE',
+            '-A SINKHOLE -s 127.0.0.1/32 -j RETURN',
+            '-A SINKHOLE -j LOG --log-prefix "\\"[nfsinkhole] \\""',
+            '-A SINKHOLE -j NFLOG'
+        ]
+        existing = myobj.list_existing_rules()
+        self.assertSequenceEqual(existing, expected, seq_type=list)
+
+    def _test_delete_rules(self):
+
+        # Success
+        myobj = IPTablesSinkhole(
+            interface='eth1',
+            interface_addr='127.0.0.1'
+        )
+        myobj.delete_rules()
+
+        # Exists
+        self.assertRaises(IPTablesNotExists, myobj.delete_rules)
+
+    def test_rules(self):
+
+        self._test_create_rules()
+        self._test_delete_rules()

nfsinkhole/tests/test_utils.py

@@ -12,6 +12,18 @@
 
 class TestIPTablesSinkhole(TestCommon):
 
+    def test_popen_wrapper(self):
+
+        # Argument checks
+        self.assertRaises(ValueError, popen_wrapper)
+        self.assertRaises(TypeError, popen_wrapper, 'notalist')
+
+        # raise_err test
+        self.assertRaises(SubprocessError, popen_wrapper, **dict(
+            cmd_arr=['asdasd'],
+            raise_err=True
+        ))
+
     def test_timezone(self):
 
         set_system_timezone('UTC')

nfsinkhole/utils.py

@@ -68,14 +68,14 @@ def popen_wrapper(cmd_arr=None, raise_err=False, log_stdout_line=True,
             raise_err must be True for this.
     """
 
-    log.debug('Running: {0}'.format(' '.join(cmd_arr)))
-
     if not cmd_arr:
         raise ValueError('cmd_arr is required (list of commands)')
 
     if not isinstance(cmd_arr, list):
         raise TypeError('cmd_arr must be a list of commands')
 
+    log.debug('Running: {0}'.format(' '.join(cmd_arr)))
+
     # If sudo, run /usr/bin/sudo if not root
     if sudo and uid != 0:
         cmd_arr = ['/usr/bin/sudo'] + cmd_arr
@@ -102,9 +102,14 @@ def popen_wrapper(cmd_arr=None, raise_err=False, log_stdout_line=True,
         out_arr = out.splitlines(True) if out else []
         for line in out_arr:
 
+            try:
+                tmp = line.decode('ascii', 'ignore').replace('\n', '')
+            except AttributeError:  # pragma: no cover
+                tmp = line.replace('\n', '')
+
             log.debug('[{0}] {1}'.format(
                 ' '.join(cmd_arr),
-                line.replace(b'\n', b'').decode('ascii', 'ignore')
+                tmp
             ))
 
     # Log stdout as a single entry.
@@ -116,18 +121,29 @@ def popen_wrapper(cmd_arr=None, raise_err=False, log_stdout_line=True,
     err_arr = err.splitlines(True) if err else []
     for line in err_arr:
 
+        try:
+            tmp = line.decode('ascii', 'ignore').replace('\n', '')
+        except AttributeError:  # pragma: no cover
+            tmp = line.replace('\n', '')
+
         log.error('[{0}] {1}'.format(
             ' '.join(cmd_arr),
-            line.replace(b'\n', b'').decode('ascii', 'ignore')
+            tmp
         ))
 
     # If any errors, iterate them and write to log, then raise
     # SubprocessError.
     if raise_err and err:
         arr = err.splitlines()
+
+        try:
+            tmp = b'\n'.join(arr).decode('ascii', 'ignore')
+        except TypeError:  # pragma: no cover
+            tmp = '\n'.join(arr)
+
         raise SubprocessError(
             'Error encountered when running process "{0}":\n{1}'.format(
-                ' '.join(cmd_arr), b'\n'.join(arr).decode('ascii', 'ignore')
+                ' '.join(cmd_arr), tmp
             )
         )