verbosity + refactoring

Author: DominiqueDevinci

utils/oscap-docker.in

@@ -21,8 +21,9 @@
 ''' oscap docker command '''
 
 import argparse
-from oscap_docker_python.oscap_docker_util import OscapAtomicScan,\
-OscapDockerScan, isAtomicLoaded
+from oscap_docker_python.oscap_docker_util import OscapAtomicScan, \
+    OscapDockerScan, isAtomicLoaded
+
 import docker
 import traceback
 import sys
@@ -44,8 +45,11 @@ if __name__ == '__main__':
     parser = argparse.ArgumentParser(description='oscap docker',
                                      epilog='See `man oscap` to learn \
                                      more about OSCAP-ARGUMENTS')
-    parser.add_argument('--oscap', dest='oscap_binary', default='', help='Set the oscap binary to use')
-    parser.add_argument('--disable-atomic', dest='noatomic', action='store_true', help="Force to use native docker API instead of atomic")
+    parser.add_argument('--oscap', dest='oscap_binary', default='',
+                        help='Set the oscap binary to use')
+
+    parser.add_argument('--disable-atomic', dest='noatomic', action='store_true',
+                        help="Force to use native docker API instead of atomic")
     subparser = parser.add_subparsers(help="commands")
 
     # Scan CVEs in image
@@ -91,6 +95,7 @@ if __name__ == '__main__':
 
     try:
         if isAtomicLoaded and not args.noatomic:
+            print("Using Atomic API")
             OS = OscapAtomicScan(oscap_binary=args.oscap_binary)
             if args.action == "scan":
                 rc = OscapAtomicScan.scan(OS, args.scan_target, leftover_args)
@@ -102,7 +107,7 @@ if __name__ == '__main__':
 
         else:  # without atomic
             if args.noatomic:
-                print("Running oscap-docker with native docker api instead of atomic ...")
+                print("Using native Docker API")
 
             ODS = OscapDockerScan(args.scan_target, args.is_image, args.oscap_binary)
             if args.action == "scan":
@@ -113,16 +118,13 @@ if __name__ == '__main__':
                 parser.print_help()
                 sys.exit(2)
 
-    except ValueError as e:
-        raise e
-        sys.exit(255)
-    except RuntimeError as e:
+    except (ValueError, RuntimeError) as e:
         raise e
         sys.exit(255)
     except Exception as exc:
         traceback.print_exc(file=sys.stdout)
-        sys.stderr.write("!!! WARNING !!! This software have crashed, so you should "
-        "check that no temporary container is still running\n")
+        sys.stderr.write("!!! WARNING !!! This software has crashed, so you should "
+                         "check that no temporary container is still running\n")
         sys.exit(255)
 
     sys.exit(rc)

utils/oscap_docker_python/oscap_docker_common.py

@@ -0,0 +1,86 @@
+# Copyright (C) 2015 Brent Baude <[email protected]>
+# Copyright (C) 2019 Dominique Blaze <[email protected]>
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the
+# Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+# Boston, MA 02111-1307, USA.
+
+import subprocess
+import platform
+import os
+import collections
+
+
+class OscapError(Exception):
+    ''' oscap Error'''
+    pass
+
+OscapResult = collections.namedtuple("OscapResult", ("returncode", "stdout", "stderr"))
+
+
+def oscap_chroot(chroot_path, oscap_binary, oscap_args, target_name, local_env=[]):
+        '''
+        Wrapper running oscap_chroot on an OscapDockerScan OscapAtomicScan object
+        '''
+        os.environ["OSCAP_PROBE_ARCHITECTURE"] = platform.processor()
+        os.environ["OSCAP_PROBE_ROOT"] = os.path.join(chroot_path)
+        os.environ["OSCAP_PROBE_OS_NAME"] = platform.system()
+        os.environ["OSCAP_PROBE_OS_VERSION"] = platform.release()
+
+        os.environ["OSCAP_EVALUATION_TARGET"] = target_name
+
+        for var in local_env:
+            vname, val = var.split("=", 1)
+            os.environ["OSCAP_OFFLINE_" + vname] = val
+
+        cmd = [oscap_binary] + [x for x in oscap_args]
+
+        oscap_process = subprocess.Popen(cmd, stdout=subprocess.PIPE,
+                                         stderr=subprocess.PIPE)
+        oscap_stdout, oscap_stderr = oscap_process.communicate()
+        return OscapResult(oscap_process.returncode,
+                           oscap_stdout.decode("utf-8"),
+                           oscap_stderr.decode("utf-8"))
+
+# TODO replace by _get_cpe (in order to indentify any containerized system)
+
+
+def get_dist(mountpoint, oscap_binary, local_env):
+    CPE_RHEL = 'oval:org.open-scap.cpe.rhel:def:'
+    DISTS = ["8", "7", "6", "5"]
+
+    '''
+    Test the chroot and determine what RHEL dist it is; returns
+    an integer representing the dist
+    '''
+
+    cpe_dict = '/usr/share/openscap/cpe/openscap-cpe-oval.xml'
+    if not os.path.exists(cpe_dict):
+        # sometime it's installed into /usr/local/share instead of /usr/local
+        cpe_dict = '/usr/local/share/openscap/cpe/openscap-cpe-oval.xml'
+        if not os.path.exists(cpe_dict):
+            raise OscapError()
+
+    for dist in DISTS:
+        result = oscap_chroot(
+            mountpoint, oscap_binary,
+            ("oval", "eval", "--id", CPE_RHEL + dist, cpe_dict,
+             mountpoint, "2>&1", ">", "/dev/null"),
+            '*',
+            local_env
+        )
+
+        if "{0}{1}: true".format(CPE_RHEL, dist) in result.stdout:
+            print("This system seems based on RHEL{0}.".format(dist))
+            return dist

utils/oscap_docker_python/oscap_docker_util.py

@@ -30,6 +30,8 @@
 import docker
 import collections
 from oscap_docker_python.oscap_docker_util_noatomic import OscapDockerScan
+from oscap_docker_python.oscap_docker_common import oscap_chroot, get_dist, \
+    OscapResult, OscapError
 
 atomic_loaded = False
 
@@ -86,14 +88,6 @@ def isAtomicLoaded():
     return atomic_loaded
 
 
-class OscapError(Exception):
-    ''' oscap Error'''
-    pass
-
-
-OscapResult = collections.namedtuple("OscapResult", ("returncode", "stdout", "stderr"))
-
-
 class OscapHelpers(object):
     ''' oscap class full of helpers for scanning '''
     CPE = 'oval:org.open-scap.cpe.rhel:def:'
@@ -120,24 +114,6 @@ def _rm_tmp_dir(tmp_dir):
         '''
         shutil.rmtree(tmp_dir)
 
-    def _get_dist(self, chroot, target):
-        '''
-        Test the chroot and determine what RHEL dist it is; returns
-        an integer representing the dist
-        '''
-        cpe_dict = '/usr/share/openscap/cpe/openscap-cpe-oval.xml'
-        if not os.path.exists(cpe_dict):
-            cpe_dict = '/usr/local/share/openscap/cpe/openscap-cpe-oval.xml'
-            if not os.path.exists(cpe_dict):
-                raise OscapError()
-
-        for dist in self.DISTS:
-            result = self.oscap_chroot(chroot, target, 'oval', 'eval',
-                                       '--id', self.CPE + dist, cpe_dict,
-                                       '2>&1', '>', '/dev/null')
-            if "{0}{1}: true".format(self.CPE, dist) in result.stdout:
-                return dist
-
     def _get_target_name_and_config(self, target):
         '''
         Determines if target is image or container. For images returns full
@@ -166,40 +142,30 @@ def _get_target_name_and_config(self, target):
             except docker.errors.NotFound:
                 return "unknown", {}
 
-    def oscap_chroot(self, chroot_path, target, *oscap_args):
-        '''
-        Wrapper function for executing oscap in a subprocess
-        '''
-        os.environ["OSCAP_PROBE_ARCHITECTURE"] = platform.processor()
-        os.environ["OSCAP_PROBE_ROOT"] = os.path.join(chroot_path)
-        os.environ["OSCAP_PROBE_OS_NAME"] = platform.system()
-        os.environ["OSCAP_PROBE_OS_VERSION"] = platform.release()
-        name, conf = self._get_target_name_and_config(target)
-        os.environ["OSCAP_EVALUATION_TARGET"] = name
-        for var in conf.get("Env", []):
-            vname, val = var.split("=", 1)
-            os.environ["OSCAP_OFFLINE_" + vname] = val
-        cmd = [self.oscap_binary] + [x for x in oscap_args]
-        oscap_process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        oscap_stdout, oscap_stderr = oscap_process.communicate()
-        return OscapResult(oscap_process.returncode,
-                           oscap_stdout.decode("utf-8"), oscap_stderr.decode("utf-8"))
-
     def _scan_cve(self, chroot, target, dist, scan_args):
         '''
         Scan a chroot for cves
         '''
         cve_input = getInputCVE.dist_cve_name.format(dist)
-        tmp_tuple = ('oval', 'eval') + tuple(scan_args) + \
-            (os.path.join(self.cve_input_dir, cve_input),)
-        return self.oscap_chroot(chroot, target, *tmp_tuple)
+
+        args = ("oval", "eval")
+        for a in scan_args:
+            args += (a,)
+        args += (os.path.join(self.cve_input_dir, cve_input),)
+
+        name, conf = self._get_target_name_and_config(target)
+
+        return oscap_chroot(chroot, self.oscap_binary, args, name,
+                            conf.get("Env", []) or [])
 
     def _scan(self, chroot, target, scan_args):
         '''
         Scan a container or image
         '''
-        tmp_tuple = tuple(scan_args)
-        return self.oscap_chroot(chroot, target, *tmp_tuple)
+
+        name, conf = self._get_target_name_and_config(target)
+        return oscap_chroot(chroot, target, scan_args, name,
+                            conf.get("Env", []) or [])
 
     def resolve_image(self, image):
         '''
@@ -287,7 +253,8 @@ def scan_cve(self, image, scan_args):
             chroot = self._find_chroot_path(_tmp_mnt_dir)
 
             # Figure out which RHEL dist is in the chroot
-            dist = self.helper._get_dist(chroot, image)
+            name, conf = self.helper._get_target_name_and_config(image)
+            dist = get_dist(chroot, self.helper.oscap_binary, conf.get("Env", []) or [])
 
             if dist is None:
                 sys.stderr.write("{0} is not based on RHEL\n".format(image))

utils/oscap_docker_python/oscap_docker_util_noatomic.py

@@ -20,14 +20,14 @@
 
 import os
 import tempfile
-import subprocess
-import platform
 import shutil
 from oscap_docker_python.get_cve_input import getInputCVE
 import sys
 import docker
 import uuid
 import collections
+from oscap_docker_python.oscap_docker_common import oscap_chroot, get_dist, \
+    OscapResult, OscapError
 
 
 class OscapError(Exception):
@@ -39,8 +39,6 @@ class OscapError(Exception):
 
 
 class OscapDockerScan(object):
-    CPE_RHEL = 'oval:org.open-scap.cpe.rhel:def:'
-    DISTS = ["8", "7", "6", "5"]
 
     def __init__(self, target, is_image=False, oscap_binary='oscap'):
 
@@ -107,7 +105,7 @@ def __init__(self, target, is_image=False, oscap_binary='oscap'):
 
         if self._check_container_mountpoint():
             self.mountpoint = "/proc/{0}/root".format(self.pid)
-            print("Docker container {0} ready to be scanned !"
+            print("Docker container {0} ready to be scanned."
                   .format(self.container_name))
         else:
             self._end()
@@ -166,29 +164,6 @@ def _check_container_mountpoint(self):
         '''
         return os.access("/proc/{0}/root".format(self.pid), os.R_OK)
 
-    # TODO replace by _get_cpe (in order to indentify any containerized system)
-    def _get_dist(self):
-        '''
-        Test the chroot and determine what RHEL dist it is; returns
-        an integer representing the dist
-        '''
-        cpe_dict = '/usr/share/openscap/cpe/openscap-cpe-oval.xml'
-        if not os.path.exists(cpe_dict):
-            cpe_dict = '/usr/local/share/openscap/cpe/openscap-cpe-oval.xml'
-            if not os.path.exists(cpe_dict):
-                raise OscapError()
-
-        for dist in self.DISTS:
-            result = self.oscap_chroot(
-                self.mountpoint, self.oscap_binary,
-                ("oval", "eval", "--id", self.CPE_RHEL + dist, cpe_dict,
-                 self.mountpoint, "2>&1", ">", "/dev/null")
-            )
-
-            if "{0}{1}: true".format(self.CPE_RHEL, dist) in result.stdout:
-                print("This system seems bases on RHEL{0}.".format(dist))
-                return dist
-
     def scan_cve(self, scan_args):
         '''
         Wrapper function for scanning cve of a mounted container
@@ -197,7 +172,8 @@ def scan_cve(self, scan_args):
         tmp_dir = tempfile.mkdtemp()
 
         # Figure out which RHEL dist is in the chroot
-        dist = self._get_dist()
+        dist = get_dist(self.mountpoint, self.oscap_binary,
+                        self.config["Config"].get("Env", []) or [])
 
         if dist is None:
             sys.stderr.write("{0} is not based on RHEL\n"
@@ -215,8 +191,11 @@ def scan_cve(self, scan_args):
             args += (a,)
         args += (cve_file,)
 
-        scan_result = self.oscap_chroot(
-            self.mountpoint, self.oscap_binary, args)
+        scan_result = oscap_chroot(
+            self.mountpoint, self.oscap_binary, args,
+            self.image_name or self.container_name,
+            self.config["Config"].get("Env", []) or []  # because Env can exists but be None
+        )
 
         print(scan_result.stdout)
         print(scan_result.stderr, file=sys.stderr)
@@ -233,38 +212,16 @@ def scan(self, scan_args):
         '''
         Wrapper function forwarding oscap args for an offline scan
         '''
-        scan_result = self.oscap_chroot(
+        scan_result = oscap_chroot(
             "/proc/{0}/root".format(self.pid),
-            self.oscap_binary, scan_args
+            self.oscap_binary, scan_args,
+            self.image_name or self.container_name,
+            self.config["Config"].get("Env", []) or []  # because Env can exists but be None
         )
+
         print(scan_result.stdout)
         print(scan_result.stderr, file=sys.stderr)
 
         self._end()
 
         return scan_result.returncode
-
-    def oscap_chroot(self, chroot_path, oscap_binary, oscap_args):
-        '''
-        Wrapper running oscap_chroot on an OscapDockerScan object
-        '''
-        os.environ["OSCAP_PROBE_ARCHITECTURE"] = platform.processor()
-        os.environ["OSCAP_PROBE_ROOT"] = os.path.join(chroot_path)
-        os.environ["OSCAP_PROBE_OS_NAME"] = platform.system()
-        os.environ["OSCAP_PROBE_OS_VERSION"] = platform.release()
-
-        os.environ["OSCAP_EVALUATION_TARGET"] = self.container_name
-
-        if self.config["Config"].get("Env"):  # Env can be exists but be None
-            for var in self.config["Config"].get("Env", []):
-                vname, val = var.split("=", 1)
-                os.environ["OSCAP_OFFLINE_" + vname] = val
-
-        cmd = [oscap_binary] + [x for x in oscap_args]
-
-        oscap_process = subprocess.Popen(cmd, stdout=subprocess.PIPE,
-                                         stderr=subprocess.PIPE)
-        oscap_stdout, oscap_stderr = oscap_process.communicate()
-        return OscapResult(oscap_process.returncode,
-                           oscap_stdout.decode("utf-8"),
-                           oscap_stderr.decode("utf-8"))