utils.add_platform_rule.createFunc() - Code Metrics - Inspection of "Merge pull request #6299 from ggbecker/readthedocs..." - ComplianceAsCode/content - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 42467a...9d011f )

by Matěj

created 2020-11-11 16:04 UTC

utils.add_platform_rule.createFunc() C

↳ Parent: utils.add_platform_rule

Complexity

Conditions

Size

Total Lines	45
Code Lines	40

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	132

Importance

Changes

Metric	Value
cc	11
eloc	40
nop	1
dl	0
loc	45
ccs	0
cts	30
cp	0
crap	132
rs	5.4
c	0
b	0
f	0

How to fix Complexity

#!/usr/bin/python3

import argparse
import subprocess
import pathlib
import sys
import textwrap
import os
import time

PROG_DESC = (''' Create and test content files for Kubernetes API checks.

This script is intended to help content writers create a new application check
for OCP4/Kubernetes.

- The 'create' subcommand creates the initial files for a new rule and fetches
  the raw URL of the object in question (unless you specify the URL).

- The 'test' subcommand builds your content locally and tests directly using an
  openscap podman container. The scan container will test against yaml files
  staged under --objectdir.

- The 'cluster-test' subcommand pushes the content to your cluster, and then
  runs a Platform scan for your rule with compliance-operator.

Example workflow:

$ utils/add_platform_rule.py create --rule=ocp_proxy_has_ca \
  --type="proxies.config" --name="cluster" \
  --yamlpath=".spec.trustedCA.name" --match="[a-zA-Z0-9]*"
creating check for "/apis/config.openshift.io/v1/proxies/cluster" with yamlpath ".spec.trustedCA.name" satisfying match of "[a-zA-Z0-9]*"
wrote applications/openshift/ocp_proxy_has_ca/rule.yml

$ mkdir -p /tmp/apis/config.openshift.io/v1/proxies/
$ oc get proxies.config/cluster -o yaml > /tmp/apis/config.openshift.io/v1/proxies/cluster
$ utils/add_platform_rule.py test --rule=ocp_proxy_has_ca
testing rule ocp_proxy_has_ca locally
Title
        None
Rule
        xccdf_org.ssgproject.content_rule_ocp_proxy_has_ca
Ident
        CCE-84209-6
Result
        pass

$ utils/add_platform_rule.py cluster-test --rule=ocp_proxy_has_ca
testing rule ocp_proxy_has_ca in-cluster
deploying compliance-operator
pushing image build to cluster
waiting for cleanup from previous test run
output from last phase check: LAUNCHING NOT-AVAILABLE
output from last phase check: RUNNING NOT-AVAILABLE
output from last phase check: AGGREGATING NOT-AVAILABLE
output from last phase check: DONE COMPLIANT
COMPLIANT

''')

PLATFORM_RULE_DIR = 'applications/openshift'
OSCAP_TEST_IMAGE = 'quay.io/compliance-operator/openscap-ocp:1.3.4'
OSCAP_CMD_OPTS = 'oscap xccdf eval --verbose INFO --fetch-remote-resources --profile xccdf_org.ssgproject.content_profile_test --results-arf /tmp/report-arf.xml /content/ssg-ocp4-ds.xml'
PROFILE_PATH = 'ocp4/profiles/test.profile'

MOCK_VERSION = ('''status:
  versions:
  - name: operator
    version: 4.6.0-0.ci-2020-06-15-112708
  - name: openshift-apiserver
    version: 4.6.0-0.ci-2020-06-15-112708
''')

RULE_TEMPLATE = ('''prodtype: ocp4

title: {TITLE}

description: {DESC}

rationale: TBD

identifiers: {{}}

severity: {SEV}

warnings:
- general: |-
    {{{{{{ openshift_cluster_setting("{URL}") | indent(4) }}}}}}

template:
  name: yamlfile_value
  vars:
    ocp_data: "true"{ENTITY_CHECK}{CHECK_EXISTENCE}
    filepath: {URL}
    yamlpath: "{YAMLPATH}"
    values:
    - value: "{MATCH}"{CHECK_TYPE}
''')


def operation_value(value):
    if value:
        return '\n      operation: "pattern match"\n      type: "string"'
    else:
        return ''


def entity_value(value):
    if value is not None:
        return '\n    entity_check: "%s"' % value
    else:
        return ''

def check_existence_value(value):
    if value is not None:
        return '\n    check_existence: "%s"' % value
    else:
        return ''


PROFILE_TEMPLATE = ('''documentation_complete: true

title: 'Test Profile for {RULE_NAME}'

platform: ocp4

description: Test Profile
selections:
- {RULE_NAME}
''')


TEST_SCAN_TEMPLATE = ('''apiVersion: compliance.openshift.io/v1alpha1
kind: ComplianceScan
metadata:
  name: test
spec:
  scanType: Platform
  profile: {PROFILE}
  content: ssg-ocp4-ds.xml
  contentImage: image-registry.openshift-image-registry.svc:5000/openshift-compliance/openscap-ocp4-ds:latest
  debug: true
''')


def needs_oc(func):
    def wrapper(args):
        if which('oc') is None:
            print('oc is required for this command.')
            return 1

        return func(args)
    return wrapper


def needs_working_cluster(func):
    def wrapper(args):
        ret_code, output = subprocess.getstatusoutput(
            'oc whoami')
        if ret_code != 0:
            print("* Error connecting to cluster")
            print(output)
            return ret_code

        return func(args)
    return wrapper

def which(program):
    fpath, fname = os.path.split(program)
    if fpath:
        if os.path.isfile(fpath) and os.access(fpath, os.X_OK):
            return program
    else:
        for path in os.environ["PATH"].split(os.pathsep):
            exe_file = os.path.join(path, program)
            if os.path.isfile(exe_file) and os.access(exe_file, os.X_OK):
                return exe_file

    return None


@needs_oc
def createFunc(args):
    url = args.url
    retries = 0
    namespace_flag = ''
    if args.namespace is not None:
        namespace_flag = '-n ' + args.namespace

    group_path = os.path.join(PLATFORM_RULE_DIR, args.group)
    if args.group:
        if not os.path.isdir(group_path):
            print("ERROR: The specified group '%s' doesn't exist in the '%s' directory" % (
                args.group, PLATFORM_RULE_DIR))
            return 0

    rule_path = os.path.join(group_path, args.rule)
    while url is None and retries < 5:
        retries += 1
        ret_code, output = subprocess.getstatusoutput(
            'oc get %s/%s -o template="{{.metadata.selfLink}}" %s' % (args.type, args.name, namespace_flag))
        if ret_code != 0:
            print('error running oc, check connection to the cluster: %d\n %s' % (
                ret_code, output))
            continue
        if len(output) > 0 and '/api' in output:
            url = output

    if url == None:
        print('there was a problem finding the URL from the oc debug output. Hint: override this automatic check with --url')
        return 1

    print('* Creating check for "%s" with yamlpath "%s" satisfying match of "%s"' % (
        url, args.yamlpath, args.match))
    rule_yaml_path = os.path.join(rule_path, 'rule.yml')

    pathlib.Path(rule_path).mkdir(parents=True, exist_ok=True)
    with open(rule_yaml_path, 'w') as f:
        f.write(RULE_TEMPLATE.format(URL=url, TITLE=args.title, SEV=args.severity, IDENT=args.identifiers,
                                     DESC=args.description, YAMLPATH=args.yamlpath, MATCH=args.match,
                                     NEGATE=str(args.negate).lower(),
                                     CHECK_TYPE=operation_value(args.regex),
                                     CHECK_EXISTENCE=check_existence_value(args.check_existence),
                                     ENTITY_CHECK=entity_value(args.match_entity)))
    print('* Wrote ' + rule_yaml_path)
    return 0


def createTestProfile(rule):
    # create a solo profile for rule
    with open(PROFILE_PATH, 'w') as f:
        f.write(PROFILE_TEMPLATE.format(RULE_NAME=rule))


@needs_oc
@needs_working_cluster
def clusterTestFunc(args):

    print('* Testing rule %s in-cluster' % args.rule)

    findout = subprocess.getoutput(
        "find %s -name '%s' -type d" % (PLATFORM_RULE_DIR, args.rule))
    if findout == "":
        print('ERROR: no rule for %s, run "create" first' % args.rule)
        return 1

    if not args.skip_deploy:
        subprocess.run("utils/deploy_compliance_operator.sh")

    if not args.skip_build:
        createTestProfile(args.rule)
        print('* Pushing image build to cluster')
        # execute the build_ds_container script
        buildp = subprocess.run(
            ['utils/build_ds_container.sh', '-P', 'ocp4', '-P', 'rhcos4'])
        if buildp.returncode != 0:
            try:
                os.remove(PROFILE_PATH)
            except OSError:
                pass
            return 1

    ret_code, _ = subprocess.getstatusoutput(
        'oc delete compliancescans/test')
    if ret_code == 0:
        # if previous compliancescans were actually deleted, wait a bit to allow resources to clean up.
        print('* Waiting for cleanup from a previous test run')
        time.sleep(20)

    # create a single-rule scan
    print("* Running scan with rule '%s'" % args.rule)
    profile = 'xccdf_org.ssgproject.content_profile_test'
    apply_cmd = ['oc', 'apply', '-f', '-']
    with subprocess.Popen(apply_cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE) as proc:
        _, err = proc.communicate(
            input=TEST_SCAN_TEMPLATE.format(PROFILE=profile).encode())
        if proc.returncode != 0:
            print('Error applying scan object: %s' % err)
            try:
                os.remove(PROFILE_PATH)
            except OSError:
                pass
            return 1

    # poll for the DONE result
    timeout = time.time() + 120   # A couple of minutes is generous for the platform scan.
    scan_result = None
    while True:
        ret_code, output = subprocess.getstatusoutput(
            'oc get compliancescans/test -o template="{{.status.phase}} {{.status.result}}"')
        if output is not None:
            print('> Output from last phase check: %s' % output)
        if output.startswith('DONE'):
            scan_result = output[5:]
            break
        if time.time() >= timeout:
            break
        time.sleep(2)

    if scan_result == None:
        print('ERROR: Timeout waiting for scan to finish')
        return 1

    print("* The result is '%s'" % scan_result)
    return 0


def testFunc(args):
    if which('podman') is None:
        print('podman is required')
        return 1

    print('testing rule %s locally' % args.rule)

    if not args.skip_build:
        createTestProfile(args.rule)
        ret_code, out = subprocess.getstatusoutput('./build_product --datastream-only ocp4')
        if ret_code != 0:
            print('build failed: %s' % out)
            return 1

    # mock a passing result for the implicit ocp4 version check
    version_dir = args.objectdir + '/apis/config.openshift.io/v1/clusteroperators'
    if not os.path.exists(version_dir):
        pathlib.Path(version_dir).mkdir(parents=True, exist_ok=True)
        with open(version_dir + '/openshift-apiserver', 'w') as f:
            f.write(MOCK_VERSION)

    pod_cmd = 'podman run -it --security-opt label=disable -v "%s:/content" -v "%s:/kubernetes-api-resources" %s %s' % (args.contentdir,
                                                                                                                        args.objectdir, OSCAP_TEST_IMAGE, OSCAP_CMD_OPTS)
    print(subprocess.getoutput(pod_cmd))


def main():
    parser = argparse.ArgumentParser(
        prog="add_platform_rule.py",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description=textwrap.dedent(PROG_DESC))
    subparser = parser.add_subparsers(
        dest='subcommand', title='subcommands', help='pick one')
    create_parser = subparser.add_parser(
        'create', help='Bootstrap the XML and YML files under %s for a new check.' % PLATFORM_RULE_DIR)
    create_parser.add_argument(
        '--rule', required=True, help='The name of the rule to create. Required.')
    create_parser.add_argument(
        '--group', default="", help='The group directory of the rule to create.')
    create_parser.add_argument(
        '--name', required=True, help='The name of the Kubernetes object to check. Required.')
    create_parser.add_argument(
        '--type', required=True, help='The type of Kubernetes object, e.g., configmap. Required.')
    create_parser.add_argument('--yamlpath', required=True,
                               help='The yaml-path of the element to match against.')
    create_parser.add_argument(
        '--match', required=True, help='A string value or regex providing the matching criteria. Required')
    create_parser.add_argument(
        '--namespace', help='The namespace of the Kubernetes object (optional for cluster-scoped objects)', default=None)
    create_parser.add_argument(
        '--title', help='A short description of the check.')
    create_parser.add_argument(
        '--url', help='The direct api path (metadata.selfLink) of the object, which overrides --type --name and --namespace options.')
    create_parser.add_argument(
        '--description', help='A human-readable description of the provided matching criteria.')
    create_parser.add_argument(
        '--regex', default=False, action="store_true", help='treat the --match value as a regex')
    create_parser.add_argument(
        '--match-entity', help='the entity_check value to apply, i.e., "all", "at least one", "none exist"')
    create_parser.add_argument(
        '--check-existence', help='check_existence` value for the `yamlfilecontent_test`.')
    create_parser.add_argument(
        '--negate', default=False, action="store_true", help='negate the given matching criteria (does NOT match). Default is false.')
    create_parser.add_argument(
        '--identifiers', default="TBD", help='an identifier for the rule (CCE number)')
    create_parser.add_argument(
        '--severity', default="unknown", help='the severity of the rule.')
    create_parser.set_defaults(func=createFunc)

    cluster_test_parser = subparser.add_parser(
        'cluster-test', help='Test a rule on a running OCP cluster using the compliance-operator.')
    cluster_test_parser.add_argument(
        '--rule', required=True, help='The name of the rule to test. Required.')
    cluster_test_parser.add_argument(
        '--skip-deploy', default=False, action="store_true", help='Skip deploying the compliance-operator. Default is to deploy.')
    cluster_test_parser.add_argument(
        '--skip-build', default=False, action="store_true", help='Skip building and pushing the datastream. Default is true.')
    cluster_test_parser.set_defaults(func=clusterTestFunc)

    test_parser = subparser.add_parser(
        'test', help='Test a rule locally against a directory of mocked object files using podman and an oscap container.')
    test_parser.add_argument('--rule', required=True,
                             help='The name of the rule to test.')
    test_parser.add_argument(
        '--contentdir', default="./build", help='The path to the directory containing the datastream')
    test_parser.add_argument(
        '--skip-build', default=False, action="store_true", help='Skip building the datastream. Default is false.')
    test_parser.add_argument('--objectdir', default="/tmp",
                             help='The path to a directory structure of yaml objects to test against.')
    test_parser.set_defaults(func=testFunc)

    args = parser.parse_args()

    return args.func(args)


if __name__ == "__main__":
    sys.exit(main())


1			#!/usr/bin/python3
2
3			import argparse
4			import subprocess
5			import pathlib
6			import sys
7			import textwrap
8			import os
9			import time
10
11			PROG_DESC = (''' Create and test content files for Kubernetes API checks.
12
13			This script is intended to help content writers create a new application check
14			for OCP4/Kubernetes.
15
16			- The 'create' subcommand creates the initial files for a new rule and fetches
17			the raw URL of the object in question (unless you specify the URL).
18
19			- The 'test' subcommand builds your content locally and tests directly using an
20			openscap podman container. The scan container will test against yaml files
21			staged under --objectdir.
22
23			- The 'cluster-test' subcommand pushes the content to your cluster, and then
24			runs a Platform scan for your rule with compliance-operator.
25
26			Example workflow:
27
28			$ utils/add_platform_rule.py create --rule=ocp_proxy_has_ca \
29			--type="proxies.config" --name="cluster" \
30			--yamlpath=".spec.trustedCA.name" --match="[a-zA-Z0-9]*"
31			creating check for "/apis/config.openshift.io/v1/proxies/cluster" with yamlpath ".spec.trustedCA.name" satisfying match of "[a-zA-Z0-9]*"
32			wrote applications/openshift/ocp_proxy_has_ca/rule.yml
33
34			$ mkdir -p /tmp/apis/config.openshift.io/v1/proxies/
35			$ oc get proxies.config/cluster -o yaml > /tmp/apis/config.openshift.io/v1/proxies/cluster
36			$ utils/add_platform_rule.py test --rule=ocp_proxy_has_ca
37			testing rule ocp_proxy_has_ca locally
38			Title
39			None
40			Rule
41			xccdf_org.ssgproject.content_rule_ocp_proxy_has_ca
42			Ident
43			CCE-84209-6
44			Result
45			pass
46
47			$ utils/add_platform_rule.py cluster-test --rule=ocp_proxy_has_ca
48			testing rule ocp_proxy_has_ca in-cluster
49			deploying compliance-operator
50			pushing image build to cluster
51			waiting for cleanup from previous test run
52			output from last phase check: LAUNCHING NOT-AVAILABLE
53			output from last phase check: RUNNING NOT-AVAILABLE
54			output from last phase check: AGGREGATING NOT-AVAILABLE
55			output from last phase check: DONE COMPLIANT
56			COMPLIANT
57
58			''')
59
60			PLATFORM_RULE_DIR = 'applications/openshift'
61			OSCAP_TEST_IMAGE = 'quay.io/compliance-operator/openscap-ocp:1.3.4'
62			OSCAP_CMD_OPTS = 'oscap xccdf eval --verbose INFO --fetch-remote-resources --profile xccdf_org.ssgproject.content_profile_test --results-arf /tmp/report-arf.xml /content/ssg-ocp4-ds.xml'
63			PROFILE_PATH = 'ocp4/profiles/test.profile'
64
65			MOCK_VERSION = ('''status:
66			versions:
67			- name: operator
68			version: 4.6.0-0.ci-2020-06-15-112708
69			- name: openshift-apiserver
70			version: 4.6.0-0.ci-2020-06-15-112708
71			''')
72
73			RULE_TEMPLATE = ('''prodtype: ocp4
74
75			title: {TITLE}
76
77			description: {DESC}
78
79			rationale: TBD
80
81			identifiers: {{}}
82
83			severity: {SEV}
84
85			warnings:
86			- general: \|-
87			{{{{{{ openshift_cluster_setting("{URL}") \| indent(4) }}}}}}
88
89			template:
90			name: yamlfile_value
91			vars:
92			ocp_data: "true"{ENTITY_CHECK}{CHECK_EXISTENCE}
93			filepath: {URL}
94			yamlpath: "{YAMLPATH}"
95			values:
96			- value: "{MATCH}"{CHECK_TYPE}
97			''')
98
99
100			def operation_value(value):
101			if value:
102			return '\n operation: "pattern match"\n type: "string"'
103			else:
104			return ''
105
106
107			def entity_value(value):
108			if value is not None:
109			return '\n entity_check: "%s"' % value
110			else:
111			return ''
112
113			def check_existence_value(value):
114			if value is not None:
115			return '\n check_existence: "%s"' % value
116			else:
117			return ''
118
119
120			PROFILE_TEMPLATE = ('''documentation_complete: true
121
122			title: 'Test Profile for {RULE_NAME}'
123
124			platform: ocp4
125
126			description: Test Profile
127			selections:
128			- {RULE_NAME}
129			''')
130
131
132			TEST_SCAN_TEMPLATE = ('''apiVersion: compliance.openshift.io/v1alpha1
133			kind: ComplianceScan
134			metadata:
135			name: test
136			spec:
137			scanType: Platform
138			profile: {PROFILE}
139			content: ssg-ocp4-ds.xml
140			contentImage: image-registry.openshift-image-registry.svc:5000/openshift-compliance/openscap-ocp4-ds:latest
141			debug: true
142			''')
143
144
145			def needs_oc(func):
146			def wrapper(args):
147			if which('oc') is None:
148			print('oc is required for this command.')
149			return 1
150
151			return func(args)
152			return wrapper
153
154
155			def needs_working_cluster(func):
156			def wrapper(args):
157			ret_code, output = subprocess.getstatusoutput(
158			'oc whoami')
159			if ret_code != 0:
160			print("* Error connecting to cluster")
161			print(output)
162			return ret_code
163
164			return func(args)
165			return wrapper
166
167			def which(program):
168			fpath, fname = os.path.split(program)
169			if fpath:
170			if os.path.isfile(fpath) and os.access(fpath, os.X_OK):
171			return program
172			else:
173			for path in os.environ["PATH"].split(os.pathsep):
174			exe_file = os.path.join(path, program)
175			if os.path.isfile(exe_file) and os.access(exe_file, os.X_OK):
176			return exe_file
177
178			return None
179
180
181			@needs_oc
182			def createFunc(args):
183			url = args.url
184			retries = 0
185			namespace_flag = ''
186			if args.namespace is not None:
187			namespace_flag = '-n ' + args.namespace
188
189			group_path = os.path.join(PLATFORM_RULE_DIR, args.group)
190			if args.group:
191			if not os.path.isdir(group_path):
192			print("ERROR: The specified group '%s' doesn't exist in the '%s' directory" % (
193			args.group, PLATFORM_RULE_DIR))
194			return 0
195
196			rule_path = os.path.join(group_path, args.rule)
197			while url is None and retries < 5:
198			retries += 1
199			ret_code, output = subprocess.getstatusoutput(
200			'oc get %s/%s -o template="{{.metadata.selfLink}}" %s' % (args.type, args.name, namespace_flag))
201			if ret_code != 0:
202			print('error running oc, check connection to the cluster: %d\n %s' % (
203			ret_code, output))
204			continue
205			if len(output) > 0 and '/api' in output:
206			url = output
207
208			if url == None:
209			print('there was a problem finding the URL from the oc debug output. Hint: override this automatic check with --url')
210			return 1
211
212			print('* Creating check for "%s" with yamlpath "%s" satisfying match of "%s"' % (
213			url, args.yamlpath, args.match))
214			rule_yaml_path = os.path.join(rule_path, 'rule.yml')
215
216			pathlib.Path(rule_path).mkdir(parents=True, exist_ok=True)
217			with open(rule_yaml_path, 'w') as f:
218			f.write(RULE_TEMPLATE.format(URL=url, TITLE=args.title, SEV=args.severity, IDENT=args.identifiers,
219			DESC=args.description, YAMLPATH=args.yamlpath, MATCH=args.match,
220			NEGATE=str(args.negate).lower(),
221			CHECK_TYPE=operation_value(args.regex),
222			CHECK_EXISTENCE=check_existence_value(args.check_existence),
223			ENTITY_CHECK=entity_value(args.match_entity)))
224			print('* Wrote ' + rule_yaml_path)
225			return 0
226
227
228			def createTestProfile(rule):
229			# create a solo profile for rule
230			with open(PROFILE_PATH, 'w') as f:
231			f.write(PROFILE_TEMPLATE.format(RULE_NAME=rule))
232
233
234			@needs_oc
235			@needs_working_cluster
236			def clusterTestFunc(args):
237
238			print('* Testing rule %s in-cluster' % args.rule)
239
240			findout = subprocess.getoutput(
241			"find %s -name '%s' -type d" % (PLATFORM_RULE_DIR, args.rule))
242			if findout == "":
243			print('ERROR: no rule for %s, run "create" first' % args.rule)
244			return 1
245
246			if not args.skip_deploy:
247			subprocess.run("utils/deploy_compliance_operator.sh")
248
249			if not args.skip_build:
250			createTestProfile(args.rule)
251			print('* Pushing image build to cluster')
252			# execute the build_ds_container script
253			buildp = subprocess.run(
254			['utils/build_ds_container.sh', '-P', 'ocp4', '-P', 'rhcos4'])
255			if buildp.returncode != 0:
256			try:
257			os.remove(PROFILE_PATH)
258			except OSError:
259			pass
260			return 1
261
262			ret_code, _ = subprocess.getstatusoutput(
263			'oc delete compliancescans/test')
264			if ret_code == 0:
265			# if previous compliancescans were actually deleted, wait a bit to allow resources to clean up.
266			print('* Waiting for cleanup from a previous test run')
267			time.sleep(20)
268
269			# create a single-rule scan
270			print("* Running scan with rule '%s'" % args.rule)
271			profile = 'xccdf_org.ssgproject.content_profile_test'
272			apply_cmd = ['oc', 'apply', '-f', '-']
273			with subprocess.Popen(apply_cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE) as proc:
274			_, err = proc.communicate(
275			input=TEST_SCAN_TEMPLATE.format(PROFILE=profile).encode())
276			if proc.returncode != 0:
277			print('Error applying scan object: %s' % err)
278			try:
279			os.remove(PROFILE_PATH)
280			except OSError:
281			pass
282			return 1
283
284			# poll for the DONE result
285			timeout = time.time() + 120 # A couple of minutes is generous for the platform scan.
286			scan_result = None
287			while True:
288			ret_code, output = subprocess.getstatusoutput(
289			'oc get compliancescans/test -o template="{{.status.phase}} {{.status.result}}"')
290			if output is not None:
291			print('> Output from last phase check: %s' % output)
292			if output.startswith('DONE'):
293			scan_result = output[5:]
294			break
295			if time.time() >= timeout:
296			break
297			time.sleep(2)
298
299			if scan_result == None:
300			print('ERROR: Timeout waiting for scan to finish')
301			return 1
302
303			print("* The result is '%s'" % scan_result)
304			return 0
305
306
307			def testFunc(args):
308			if which('podman') is None:
309			print('podman is required')
310			return 1
311
312			print('testing rule %s locally' % args.rule)
313
314			if not args.skip_build:
315			createTestProfile(args.rule)
316			ret_code, out = subprocess.getstatusoutput('./build_product --datastream-only ocp4')
317			if ret_code != 0:
318			print('build failed: %s' % out)
319			return 1
320
321			# mock a passing result for the implicit ocp4 version check
322			version_dir = args.objectdir + '/apis/config.openshift.io/v1/clusteroperators'
323			if not os.path.exists(version_dir):
324			pathlib.Path(version_dir).mkdir(parents=True, exist_ok=True)
325			with open(version_dir + '/openshift-apiserver', 'w') as f:
326			f.write(MOCK_VERSION)
327
328			pod_cmd = 'podman run -it --security-opt label=disable -v "%s:/content" -v "%s:/kubernetes-api-resources" %s %s' % (args.contentdir,
329			args.objectdir, OSCAP_TEST_IMAGE, OSCAP_CMD_OPTS)
330			print(subprocess.getoutput(pod_cmd))
331
332
333			def main():
334			parser = argparse.ArgumentParser(
335			prog="add_platform_rule.py",
336			formatter_class=argparse.RawDescriptionHelpFormatter,
337			description=textwrap.dedent(PROG_DESC))
338			subparser = parser.add_subparsers(
339			dest='subcommand', title='subcommands', help='pick one')
340			create_parser = subparser.add_parser(
341			'create', help='Bootstrap the XML and YML files under %s for a new check.' % PLATFORM_RULE_DIR)
342			create_parser.add_argument(
343			'--rule', required=True, help='The name of the rule to create. Required.')
344			create_parser.add_argument(
345			'--group', default="", help='The group directory of the rule to create.')
346			create_parser.add_argument(
347			'--name', required=True, help='The name of the Kubernetes object to check. Required.')
348			create_parser.add_argument(
349			'--type', required=True, help='The type of Kubernetes object, e.g., configmap. Required.')
350			create_parser.add_argument('--yamlpath', required=True,
351			help='The yaml-path of the element to match against.')
352			create_parser.add_argument(
353			'--match', required=True, help='A string value or regex providing the matching criteria. Required')
354			create_parser.add_argument(
355			'--namespace', help='The namespace of the Kubernetes object (optional for cluster-scoped objects)', default=None)
356			create_parser.add_argument(
357			'--title', help='A short description of the check.')
358			create_parser.add_argument(
359			'--url', help='The direct api path (metadata.selfLink) of the object, which overrides --type --name and --namespace options.')
360			create_parser.add_argument(
361			'--description', help='A human-readable description of the provided matching criteria.')
362			create_parser.add_argument(
363			'--regex', default=False, action="store_true", help='treat the --match value as a regex')
364			create_parser.add_argument(
365			'--match-entity', help='the entity_check value to apply, i.e., "all", "at least one", "none exist"')
366			create_parser.add_argument(
367			'--check-existence', help='check_existence` value for the `yamlfilecontent_test`.')
368			create_parser.add_argument(
369			'--negate', default=False, action="store_true", help='negate the given matching criteria (does NOT match). Default is false.')
370			create_parser.add_argument(
371			'--identifiers', default="TBD", help='an identifier for the rule (CCE number)')
372			create_parser.add_argument(
373			'--severity', default="unknown", help='the severity of the rule.')
374			create_parser.set_defaults(func=createFunc)
375
376			cluster_test_parser = subparser.add_parser(
377			'cluster-test', help='Test a rule on a running OCP cluster using the compliance-operator.')
378			cluster_test_parser.add_argument(
379			'--rule', required=True, help='The name of the rule to test. Required.')
380			cluster_test_parser.add_argument(
381			'--skip-deploy', default=False, action="store_true", help='Skip deploying the compliance-operator. Default is to deploy.')
382			cluster_test_parser.add_argument(
383			'--skip-build', default=False, action="store_true", help='Skip building and pushing the datastream. Default is true.')
384			cluster_test_parser.set_defaults(func=clusterTestFunc)
385
386			test_parser = subparser.add_parser(
387			'test', help='Test a rule locally against a directory of mocked object files using podman and an oscap container.')
388			test_parser.add_argument('--rule', required=True,
389			help='The name of the rule to test.')
390			test_parser.add_argument(
391			'--contentdir', default="./build", help='The path to the directory containing the datastream')
392			test_parser.add_argument(
393			'--skip-build', default=False, action="store_true", help='Skip building the datastream. Default is false.')
394			test_parser.add_argument('--objectdir', default="/tmp",
395			help='The path to a directory structure of yaml objects to test against.')
396			test_parser.set_defaults(func=testFunc)
397
398			args = parser.parse_args()
399
400			return args.func(args)
401
402
403			if __name__ == "__main__":
404			sys.exit(main())
405

ComplianceAsCode / content

Push — master ( 42467a...9d011f )

utils.add_platform_rule.createFunc() C

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like