Merge pull request #1415 from DominiqueDevinci/docker

oscap-docker available without atomic

Author: jan-cerny

.gitignore

@@ -5,6 +5,7 @@
 .libs/
 .*.swp
 tags
+*.pyc
 
 CMakeLists.txt.user
 build/

utils/oscap-docker.in

@@ -1,6 +1,7 @@
 #!@OSCAP_DOCKER_PYTHON@
 
 # 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
@@ -20,8 +21,11 @@
 ''' oscap docker command '''
 
 import argparse
-from oscap_docker_python.oscap_docker_util import OscapScan
+from oscap_docker_python.oscap_docker_util import OscapAtomicScan, \
+    OscapDockerScan, isAtomicLoaded
+
 import docker
+import traceback
 import sys
 from requests import exceptions
 
@@ -41,40 +45,44 @@ 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('--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
     image_cve = subparser.add_parser('image-cve', help='Scan a docker image \
                                     for known vulnerabilities.')
-    image_cve.set_defaults(func=OscapScan.scan_cve)
+    image_cve.set_defaults(action="scan_cve", is_image=True)
     image_cve.add_argument('scan_target', help='Container or image to scan')
 
     # Scan an Image
     image = subparser.add_parser('image', help='Scan a docker image')
     image.add_argument('scan_target',
                        help='Container or image to scan')
 
-    image.set_defaults(func=OscapScan.scan)
+    image.set_defaults(action="scan", is_image=True)
     # Scan a container
     container = subparser.add_parser('container', help='Scan a running docker\
                                       container of given name.')
     container.add_argument('scan_target',
                            help='Container or image to scan')
-    container.set_defaults(func=OscapScan.scan)
+    container.set_defaults(action="scan", is_image=False)
 
     # Scan CVEs in container
     container_cve = subparser.add_parser('container-cve', help='Scan a \
                                          running container for known \
                                          vulnerabilities.')
 
-    container_cve.set_defaults(func=OscapScan.scan_cve)
+    container_cve.set_defaults(action="scan_cve", is_image=False)
     container_cve.add_argument('scan_target',
                                help='Container or image to scan')
 
     args, leftover_args = parser.parse_known_args()
 
-    if "func" not in args:
+    if "action" not in args:
         parser.print_help()
         sys.exit(2)
 
@@ -86,10 +94,40 @@ if __name__ == '__main__':
         sys.exit(1)
 
     try:
-        OS = OscapScan(oscap_binary=args.oscap_binary)
-        rc = args.func(OS, args.scan_target, leftover_args)
+        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)
+            elif args.action == "scan_cve":
+                rc = OscapAtomicScan.scan_cve(OS, args.scan_target, leftover_args)
+            else:
+                parser.print_help()
+                sys.exit(2)
+
+        else:  # without atomic
+            if args.noatomic:
+                print("Using native Docker API")
+
+            ODS = OscapDockerScan(args.scan_target, args.is_image, args.oscap_binary)
+            if args.action == "scan":
+                rc = OscapDockerScan.scan(ODS, leftover_args)
+            elif args.action == "scan_cve":
+                rc = OscapDockerScan.scan_cve(ODS, leftover_args)
+            else:
+                parser.print_help()
+                sys.exit(2)
+
+    except (ValueError, RuntimeError) as e:
+        raise e
+        sys.exit(255)
+    except(FileNotFoundError) as e:
+        sys.stderr.write("Target {0} not found.\n".format(target))
+        sys.exit(255)
     except Exception as exc:
+        traceback.print_exc(file=sys.stdout)
+        sys.stderr.write("!!! WARNING !!! This software has crashed, so you should "
+                         "check that no temporary container is still running\n")
         sys.exit(255)
-        raise exc
 
     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

@@ -1,4 +1,5 @@
 # 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
@@ -28,56 +29,69 @@
 import sys
 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
+
+
+class AtomicError(Exception):
+    """Exception raised when an error happens in atomic import
+    """
+    def __init__(self, message):
+        self.message = message
+
 
 try:
     from Atomic.mount import DockerMount
     from Atomic.mount import MountError
     import inspect
 
     if "mnt_mkdir" not in inspect.getargspec(DockerMount.__init__).args:
-        sys.stderr.write(
+        raise AtomicError(
             "\"Atomic.mount.DockerMount\" has been successfully imported but "
             "it doesn't support the mnt_mkdir argument. Please upgrade your "
             "Atomic installation to 1.4 or higher.\n"
         )
-        sys.exit(1)
 
     # we only care about method names
     member_methods = [
         x[0] for x in
         inspect.getmembers(
-            DockerMount, predicate=lambda member: inspect.isfunction(member) or inspect.ismethod(member)
+            DockerMount, predicate=lambda member:
+                inspect.isfunction(member) or inspect.ismethod(member)
         )
     ]
 
     if "_clean_temp_container_by_path" not in member_methods:
-        sys.stderr.write(
+        raise AtomicError(
             "\"Atomic.mount.DockerMount\" has been successfully imported but "
             "it doesn't have the _clean_temp_container_by_path method. Please "
             "upgrade your Atomic installation to 1.4 or higher.\n"
         )
-        sys.exit(1)
+
+    # if all imports are ok we can use atomic
+    atomic_loaded = True
 
 except ImportError:
     sys.stderr.write(
         "Failed to import \"Atomic.mount.DockerMount\". It seems Atomic has "
         "not been installed.\n"
     )
-    sys.exit(1)
 
+except AtomicError as err:
+    sys.stderr.write(err.message)
 
-class OscapError(Exception):
-    ''' oscap Error'''
-    pass
 
-
-OscapResult = collections.namedtuple("OscapResult", ("returncode", "stdout", "stderr"))
+def isAtomicLoaded():
+    return atomic_loaded
 
 
 class OscapHelpers(object):
     ''' oscap class full of helpers for scanning '''
     CPE = 'oval:org.open-scap.cpe.rhel:def:'
-    DISTS = ["7", "6", "5"]
+    DISTS = ["8", "7", "6", "5"]
 
     def __init__(self, cve_input_dir, oscap_binary):
         self.cve_input_dir = cve_input_dir
@@ -100,21 +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):
-            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
@@ -143,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 config.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, self.oscap_binary, scan_args, name,
+                            conf.get("Env", []) or [])
 
     def resolve_image(self, image):
         '''
@@ -209,7 +198,7 @@ def mount_image_filesystem():
     _tmp_mnt_dir = DM.mount(image)
 
 
-class OscapScan(object):
+class OscapAtomicScan(object):
     def __init__(self, tmp_dir=tempfile.gettempdir(), mnt_dir=None,
                  hours_old=2, oscap_binary=''):
         self.tmp_dir = tmp_dir
@@ -264,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))