Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(terraform): include pods of kubernetes_deployment in kubernetes_pod checks (1/4) #3691

Merged
merged 1 commit into from
Nov 7, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,8 @@ def __init__(self):

name = "Containers should not run with allowPrivilegeEscalation"
id = "CKV_K8S_20"
supported_resources = ['kubernetes_pod', 'kubernetes_pod_v1']
supported_resources = ['kubernetes_pod', 'kubernetes_pod_v1',
'kubernetes_deployment', 'kubernetes_deployment_v1']
categories = [CheckCategories.GENERAL_SECURITY]
super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)

Expand All @@ -24,6 +25,16 @@ def scan_resource_conf(self, conf) -> CheckResult:
return CheckResult.UNKNOWN
spec = spec_list[0]
if spec:
evaluated_keys_path = "spec"

template = spec.get("template")
if template and isinstance(template, list):
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
if template and isinstance(template, list):
if template and isinstance(template, list) and len(template) == 1:

template = template[0]
template_spec = template.get("spec")
if template_spec and isinstance(template_spec, list):
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
if template_spec and isinstance(template_spec, list):
if template_spec and isinstance(template_spec, list) and len(template_spec) == 1:

spec = template_spec[0]
evaluated_keys_path = f'{evaluated_keys_path}/[0]/template/[0]/spec'

containers = spec.get("container")
if not containers:
return CheckResult.UNKNOWN
Expand All @@ -34,7 +45,7 @@ def scan_resource_conf(self, conf) -> CheckResult:
context = container.get("security_context")[0]
if context.get("allow_privilege_escalation"):
if context.get("allow_privilege_escalation") == [True]:
self.evaluated_keys = [f'spec/[0]/container/[{idx}]/security_context/[0]/'
self.evaluated_keys = [f'{evaluated_keys_path}/[0]/container/[{idx}]/security_context/[0]/'
f'allow_privilege_escalation']
return CheckResult.FAILED
return CheckResult.PASSED
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,12 +12,23 @@ def __init__(self):

id = "CKV_K8S_25"

supported_resources = ['kubernetes_pod', 'kubernetes_pod_v1']
supported_resources = ['kubernetes_pod', 'kubernetes_pod_v1',
'kubernetes_deployment', 'kubernetes_deployment_v1']
categories = [CheckCategories.GENERAL_SECURITY]
super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)

def scan_resource_conf(self, conf) -> CheckResult:
spec = conf.get('spec', [None])[0]
evaluated_keys_path = "spec"

template = spec.get("template")
if template and isinstance(template, list):
template = template[0]
template_spec = template.get("spec")
if template_spec and isinstance(template_spec, list):
spec = template_spec[0]
evaluated_keys_path = f'{evaluated_keys_path}/[0]/template/[0]/spec'

if isinstance(spec, dict) and spec.get("container"):
containers = spec.get("container")

Expand All @@ -31,8 +42,9 @@ def scan_resource_conf(self, conf) -> CheckResult:
if capabilities.get("add"):
add = capabilities.get("add")[0]
if add:
self.evaluated_keys = [f'spec/[0]/container/[{idx}]/'
self.evaluated_keys = [f'{evaluated_keys_path}/[0]/container/[{idx}]/'
f'security_context/[0]/capabilities/add']

return CheckResult.FAILED
return CheckResult.PASSED

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,12 +10,23 @@ def __init__(self):
# https://kubernetes.io/docs/tasks/configure-pod-container/security-context/
id = "CKV_K8S_39"

supported_resources = ['kubernetes_pod', 'kubernetes_pod_v1']
supported_resources = ['kubernetes_pod', 'kubernetes_pod_v1',
'kubernetes_deployment', 'kubernetes_deployment_v1']
categories = [CheckCategories.GENERAL_SECURITY]
super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)

def scan_resource_conf(self, conf) -> CheckResult:
spec = conf.get('spec', [None])[0]
evaluated_keys_path = "spec"

template = spec.get("template")
if template and isinstance(template, list):
template = template[0]
template_spec = template.get("spec")
if template_spec and isinstance(template_spec, list):
spec = template_spec[0]
evaluated_keys_path = f'{evaluated_keys_path}/[0]/template/[0]/spec'

if isinstance(spec, dict) and spec.get("container"):
containers = spec.get("container")

Expand All @@ -29,7 +40,7 @@ def scan_resource_conf(self, conf) -> CheckResult:
if capabilities.get("add") and isinstance(capabilities.get("add"), list):
add = capabilities.get("add")[0]
if "SYS_ADMIN" in add:
self.evaluated_keys = [f'spec/[0]/container/[{idx}]/'
self.evaluated_keys = [f'{evaluated_keys_path}/[0]/container/[{idx}]/'
f'security_context/[0]/capabilities/add']
return CheckResult.FAILED
return CheckResult.PASSED
Expand Down
18 changes: 14 additions & 4 deletions checkov/terraform/checks/resource/kubernetes/CPULimits.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,8 @@ class CPULimits(BaseResourceCheck):
def __init__(self) -> None:
name = "CPU Limits should be set"
id = "CKV_K8S_11"
supported_resources = ["kubernetes_pod", "kubernetes_pod_v1"]
supported_resources = ['kubernetes_pod', 'kubernetes_pod_v1',
'kubernetes_deployment', 'kubernetes_deployment_v1']
categories = [CheckCategories.GENERAL_SECURITY]
super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)

Expand All @@ -19,6 +20,15 @@ def scan_resource_conf(self, conf: dict[str, list[Any]]) -> CheckResult:
self.evaluated_keys = [""]
return CheckResult.FAILED
spec = conf['spec'][0]
evaluated_keys_path = "spec"

template = spec.get("template")
if template and isinstance(template, list):
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
if template and isinstance(template, list):
if template and isinstance(template, list) and len(template) == 1:

template = template[0]
template_spec = template.get("spec")
if template_spec and isinstance(template_spec, list):
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
if template_spec and isinstance(template_spec, list):
if template_spec and isinstance(template_spec, list) and len(template_spec) == 1:

spec = template_spec[0]
evaluated_keys_path = f'{evaluated_keys_path}/[0]/template/[0]/spec'

containers = spec.get("container")
if not containers:
Expand All @@ -32,11 +42,11 @@ def scan_resource_conf(self, conf: dict[str, list[Any]]) -> CheckResult:
limits = resources.get('limits')[0]
if isinstance(limits, dict) and limits.get('cpu'):
return CheckResult.PASSED
self.evaluated_keys = [f'spec/[0]/container/[{idx}]/resources/[0]/limits']
self.evaluated_keys = [f'{evaluated_keys_path}/[0]/container/[{idx}]/resources/[0]/limits']
return CheckResult.FAILED
self.evaluated_keys = [f'spec/[0]/container/[{idx}]/resources']
self.evaluated_keys = [f'{evaluated_keys_path}/[0]/container/[{idx}]/resources']
return CheckResult.FAILED
self.evaluated_keys = [f'spec/[0]/container/[{idx}]']
self.evaluated_keys = [f'{evaluated_keys_path}/[0]/container/[{idx}]']
return CheckResult.FAILED
return CheckResult.PASSED

Expand Down
18 changes: 14 additions & 4 deletions checkov/terraform/checks/resource/kubernetes/CPURequests.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,8 @@ class CPURequests(BaseResourceCheck):
def __init__(self):
name = "CPU requests should be set"
id = "CKV_K8S_10"
supported_resources = ["kubernetes_pod", "kubernetes_pod_v1"]
supported_resources = ['kubernetes_pod', 'kubernetes_pod_v1',
'kubernetes_deployment', 'kubernetes_deployment_v1']
categories = [CheckCategories.GENERAL_SECURITY]
super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)

Expand All @@ -15,6 +16,15 @@ def scan_resource_conf(self, conf) -> CheckResult:
self.evaluated_keys = [""]
return CheckResult.FAILED
spec = conf['spec'][0]
evaluated_keys_path = "spec"

template = spec.get("template")
if template and isinstance(template, list):
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
if template and isinstance(template, list):
if template and isinstance(template, list) and len(template) == 1:

template = template[0]
template_spec = template.get("spec")
if template_spec and isinstance(template_spec, list):
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
if template_spec and isinstance(template_spec, list):
if template_spec and isinstance(template_spec, list) and len(template_spec) == 1:

spec = template_spec[0]
evaluated_keys_path = f'{evaluated_keys_path}/[0]/template/[0]/spec'

containers = spec.get("container")
if containers is None:
Expand All @@ -28,11 +38,11 @@ def scan_resource_conf(self, conf) -> CheckResult:
limits = resources.get('requests')[0]
if isinstance(limits, dict) and limits.get('cpu'):
return CheckResult.PASSED
self.evaluated_keys = [f'spec/[0]/container/[{idx}]/resources/[0]/requests']
self.evaluated_keys = [f'{evaluated_keys_path}/[0]/container/[{idx}]/resources/[0]/requests']
return CheckResult.FAILED
self.evaluated_keys = [f'spec/[0]/container/[{idx}]/resources']
self.evaluated_keys = [f'{evaluated_keys_path}/[0]/container/[{idx}]/resources']
return CheckResult.FAILED
self.evaluated_keys = [f'spec/[0]/container/[{idx}]']
self.evaluated_keys = [f'{evaluated_keys_path}/[0]/container/[{idx}]']
return CheckResult.FAILED
return CheckResult.PASSED

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,20 +11,31 @@ def __init__(self):
# Location: container .securityContext
id = "CKV_K8S_30"

supported_resources = ['kubernetes_pod', 'kubernetes_pod_v1']
supported_resources = ['kubernetes_pod', 'kubernetes_pod_v1',
'kubernetes_deployment', 'kubernetes_deployment_v1']
categories = [CheckCategories.GENERAL_SECURITY]
super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)

def scan_resource_conf(self, conf) -> CheckResult:
spec = conf.get('spec', [None])[0]
evaluated_keys_path = "spec"

template = spec.get("template")
if template and isinstance(template, list):
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
if template and isinstance(template, list):
if template and isinstance(template, list) and len(template) == 1:

template = template[0]
template_spec = template.get("spec")
if template_spec and isinstance(template_spec, list):
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
if template_spec and isinstance(template_spec, list):
if template_spec and isinstance(template_spec, list) and and len(template_spec) == 1:

spec = template_spec[0]
evaluated_keys_path = f'{evaluated_keys_path}/[0]/template/[0]/spec'

if isinstance(spec, dict) and spec.get("container"):
containers = spec.get("container")

for idx, container in enumerate(containers):
if type(container) != dict:
return CheckResult.UNKNOWN
if not container.get("security_context"):
self.evaluated_keys = [f'spec/[0]/container/[{idx}]/security_context']
self.evaluated_keys = [f'{evaluated_keys_path}/[0]/container/[{idx}]/security_context']
return CheckResult.FAILED
return CheckResult.PASSED

Expand Down
Loading