AppArmor, SELinux, Pi detection, Bionic support

.circleci/config.yml

@@ -10,7 +10,7 @@ jobs:
           command: |
             apt update
             apt install -y python3-pip python3-netifaces python3-requests python3-tz python3-iptables python3-systemd \
-            python3-psutil python3-apt devscripts python3-all ruby-dev
+            python3-psutil python3-apt python3-selinux python3-apparmor devscripts python3-all ruby-dev
             pip3 install -r requirements-dev.txt
       - run:
           name: Run linting and metrics
@@ -23,19 +23,27 @@ jobs:
       - run:
           name: Build dpkg packages
           command: |
-            mkdir -p /tmp/deb/jessie
+            mkdir -p /tmp/deb/jessie /tmp/deb/stretch /tmp/deb/buster
             python3 -c 'import version; version.write_changelog()'
+
+            debuild -e CIRCLE_BUILD_NUM --set-envvar=EXTRA_DEPS=python3-distutils -i -us -uc -b
+            mv ../*.deb /tmp/deb/buster
+
             debuild -e CIRCLE_BUILD_NUM -i -us -uc -b
-            mv ../*.deb /tmp/deb
+            mv ../*.deb /tmp/deb/stretch
+
             cp -v debian/jessie/* debian/
             debuild -e CIRCLE_BUILD_NUM -i -us -uc -b
             mv ../*.deb /tmp/deb/jessie
       - deploy:
           command: |
             if [ "${CIRCLE_BRANCH}" == "master" ]; then
               gem install rake package_cloud
-              for distro in debian/stretch raspbian/stretch debian/buster raspbian/buster ubuntu/xenial; do
-                package_cloud push wott/agent/$distro /tmp/deb/*.deb
+              for distro in debian/buster raspbian/buster ubuntu/bionic; do
+                package_cloud push wott/agent/$distro /tmp/deb/buster/*.deb
+              done
+              for distro in debian/stretch raspbian/stretch ubuntu/xenial; do
+                package_cloud push wott/agent/$distro /tmp/deb/stretch/*.deb
               done
               for distro in debian/jessie raspbian/jessie ubuntu/trusty; do
                 package_cloud push wott/agent/$distro /tmp/deb/jessie/*.deb

Dockerfile

@@ -6,7 +6,7 @@ RUN apt-get update && \
         curl pkg-config libsystemd-dev nmap python3-setuptools python3-all python3-pkg-resources python3-iptables \
         python3-psutil python3-certifi python3-cffi python3-chardet python3-cryptography python3-idna \
         python3-netifaces python3-openssl python3-tz python3-requests python3-sh python3-systemd python3-venv \
-        python3-pip python3-apt
+        python3-pip python3-apt python3-selinux python3-apparmor
 
 COPY requirements-dev.txt ./
 RUN pip3 install -r requirements-dev.txt

agent/rpi_helper.py

@@ -1,5 +1,6 @@
 import hashlib
 import os
+from pathlib import Path
 from enum import Enum
 import pkg_resources
 
@@ -13,18 +14,16 @@ def detect_raspberry_pi():
         'serial_number': None
     }
 
-    with open('/proc/cpuinfo') as f:
-        cpuinfo = f.readlines()
+    proc_model = Path('/proc/device-tree/model')
+    proc_serial = Path('/proc/device-tree/serial-number')
 
-    # Assume it is a Raspberry Pi if these three elements are present.
-    metadata['is_raspberry_pi'] = 'Hardware' in str(cpuinfo) and 'Revision' in str(cpuinfo) and 'Serial' in str(cpuinfo)
+    if proc_model.is_file():
+        model = proc_model.open().read().strip('\0')
+        metadata['hardware_model'] = model
+        metadata['is_raspberry_pi'] = model.startswith('Raspberry Pi')
 
-    if metadata['is_raspberry_pi']:
-        for line in cpuinfo:
-            if line.startswith('Revision'):
-                metadata['hardware_model'] = line.split()[-1]
-            if line.startswith('Serial'):
-                metadata['serial_number'] = line.split()[-1]
+    if proc_serial.is_file():
+        metadata['serial_number'] = proc_serial.open().read().strip('\0')
 
     return metadata
 

agent/security_helper.py

@@ -2,9 +2,9 @@
 import socket
 from pathlib import Path
 from socket import SocketKind
-import spwd
 
 import psutil
+import spwd
 
 
 def check_for_default_passwords(config_path):
@@ -83,38 +83,58 @@ def is_app_armor_enabled():
     Returns a True/False if AppArmor is enabled.
     """
     try:
-        from sh import aa_status
+        import LibAppArmor
     except ImportError:
-        return False
-
-    # Returns 0 if enabled and 1 if disable
-    get_aa_status = aa_status(['--enabled'], _ok_code=[0, 1]).exit_code
-    if get_aa_status == 1:
-        return False
-    return True
+        # If Python bindings for AppArmor are not installed (if we're
+        # running on Jessie where we can't build python3-apparmor package)
+        # we resort to calling aa-status executable.
+        try:
+            from sh import aa_status
+        except ImportError:
+            return False
+
+        # Return codes (as per aa-status(8)):
+        # 0   if apparmor is enabled and policy is loaded.
+        # 1   if apparmor is not enabled/loaded.
+        # 2   if apparmor is enabled but no policy is loaded.
+        # 3   if the apparmor control files aren't available under /sys/kernel/security/.
+        # 4   if the user running the script doesn't have enough privileges to read the apparmor
+        #    control files.
+        return aa_status(['--enabled'], _ok_code=[0, 1, 2, 3, 4]).exit_code in [0, 2]
+    else:
+        return LibAppArmor.aa_is_enabled() == 1
 
 
 def selinux_status():
     """
     Returns a dict as similar to:
         {'enabled': False, 'mode': 'enforcing'}
     """
-    selinux_enabled = None
+    selinux_enabled = False
     selinux_mode = None
 
     try:
-        from sh import sestatus
+        import selinux
     except ImportError:
-        return {'enabled': False}
-
-    # Manually parse out the output for SELinux status
-    for line in sestatus().stdout.split(b'\n'):
-        row = line.split(b':')
-
-        if row[0].startswith(b'SELinux status'):
-            selinux_enabled = row[1].strip() == b'enabled'
-
-        if row[0].startswith(b'Current mode'):
-            selinux_mode = row[1].strip()
-
+        # If Python bindings for SELinux are not installed (if we're
+        # running on Jessie where we can't build python3-selinux package)
+        # we resort to calling sestatus executable.
+        try:
+            from sh import sestatus
+        except ImportError:
+            return {'enabled': False}
+
+        # Manually parse out the output for SELinux status
+        for line in sestatus().stdout.split(b'\n'):
+            row = line.split(b':')
+
+            if row[0].startswith(b'SELinux status'):
+                selinux_enabled = row[1].strip() == b'enabled'
+
+            if row[0].startswith(b'Current mode'):
+                selinux_mode = row[1].strip()
+    else:
+        if selinux.is_selinux_enabled() == 1:
+            selinux_enabled = True
+            selinux_mode = {-1: None, 0: 'permissive', 1: 'enforcing'}[selinux.security_getenforce()]
     return {'enabled': selinux_enabled, 'mode': selinux_mode}

debian/control

@@ -16,8 +16,8 @@ X-Python3-Version: >= 3.5
 Package: wott-agent
 Architecture: all
 Section: python
-Depends: ${misc:Depends}, ${python3:Depends}, 
- python3-apt,
+Depends: ${misc:Depends}, ${python3:Depends}, ${dist:Depends},
+ python3-apt, python3-selinux, python3-apparmor,
  ghostunnel,
  rng-tools
 Description: Let's Encrypt for IoT (with more bells and whistles).

debian/jessie/control

@@ -39,5 +39,7 @@ Depends: ghostunnel,
  python3.5-pycparser,
  python3.5-ply,
  python3.5-apt,
+ python3.5-selinux,
+ python3.5-apparmor,
  systemd, systemd-sysv
 Description: Let's Encrypt for IoT (with more bells and whistles).

debian/rules

@@ -12,3 +12,6 @@ export PYBUILD_NAME=wott-agent
 
 override_dh_auto_test:
 	# Don't run the tests!
+
+override_dh_gencontrol:
+	dh_gencontrol -- -Vdist:Depends="$(EXTRA_DEPS)"

tests/test_agent.py

@@ -12,22 +12,44 @@
 from agent.journal_helper import logins_last_hour
 from agent.rpi_helper import detect_raspberry_pi
 from agent.iptables_helper import block_networks, block_ports, OUTPUT_CHAIN, INPUT_CHAIN
-from agent.security_helper import check_for_default_passwords
+from agent.security_helper import check_for_default_passwords, selinux_status
 from agent import executor
 import pwd
 from os import getenv
 
 
-def test_detect_raspberry_pi(raspberry_cpuinfo):
-    with mock.patch(
-            'builtins.open',
-            mock.mock_open(read_data=raspberry_cpuinfo),
-            create=True
-    ):
+def test_detect_raspberry_pi():
+    class mockPath():
+        def __init__(self, filename):
+            self._filename = filename
+
+        def is_file(self):
+            return True
+
+        def open(self):
+            return mock_open(self._filename)
+
+    def mock_open(filename, mode='r'):
+        """
+        This will return either a Unicode string needed for "r" mode or bytes for "rb" mode.
+        The contents are still the same which is the mock sshd_config. But they are only interpreted
+        by audit_sshd.
+        """
+        if filename == '/proc/device-tree/model':
+            content = 'Raspberry Pi 3 Model B Plus Rev 1.3\x00'
+        elif filename == '/proc/device-tree/serial-number':
+            content = '0000000060e3b222\x00'
+        else:
+            raise FileNotFoundError
+        file_object = mock.mock_open(read_data=content).return_value
+        file_object.__iter__.return_value = content.splitlines(True)
+        return file_object
+
+    with mock.patch('agent.rpi_helper.Path', mockPath):
         metadata = detect_raspberry_pi()
         assert metadata['is_raspberry_pi']
-        assert metadata['hardware_model'] == '900092'
-        assert metadata['serial_number'] == '00000000ebd5f1e8'
+        assert metadata['hardware_model'] == 'Raspberry Pi 3 Model B Plus Rev 1.3'
+        assert metadata['serial_number'] == '0000000060e3b222'
 
 
 def test_failed_logins():
@@ -532,8 +554,7 @@ def test_fetch_device_metadata(tmpdir):
             'string': 'test string value',
             'array': [1, 2, 3, 4, 5, 'penelopa'],
             'test': 'value',
-            'model': 'a020d3',
-            'model-decoded': 'Pi 3 Model B+'
+            'model': 'Pi 3 Model B+'
         }
     )
     mock_resp.return_value.ok = True
@@ -556,8 +577,7 @@ def test_fetch_device_metadata(tmpdir):
                 'string': 'test string value',
                 'array': [1, 2, 3, 4, 5, 'penelopa'],
                 'test': 'value',
-                'model': 'a020d3',
-                'model-decoded': 'Pi 3 Model B+'
+                'model': 'Pi 3 Model B+'
             }
 
         chm.assert_has_calls([
@@ -950,3 +970,19 @@ def test_no_locker(tmpdir):
 def test_independent_lockers(tmpdir):
     one, two, both = _is_parallel(tmpdir, True, True)
     assert (one, two, both) == (False, False, True)
+
+
+def test_selinux_status():
+    with mock.patch('selinux.is_selinux_enabled') as selinux_enabled,\
+            mock.patch('selinux.security_getenforce') as getenforce:
+
+        selinux_enabled.return_value = 1
+        getenforce.return_value = 1
+        assert selinux_status() == {'enabled': True, 'mode': 'enforcing'}
+
+        selinux_enabled.return_value = 1
+        getenforce.return_value = 0
+        assert selinux_status() == {'enabled': True, 'mode': 'permissive'}
+
+        selinux_enabled.return_value = 0
+        assert selinux_status() == {'enabled': False, 'mode': None}