mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 13:01:56 +00:00
refactor(iac): import checkov python library (#8093)
This commit is contained in:
Sergio Garcia
Generated
+5
-117
@@ -1,4 +1,4 @@
|
||||
# This file is automatically @generated by Poetry 2.1.1 and should not be changed by hand.
|
||||
# This file is automatically @generated by Poetry 2.1.3 and should not be changed by hand.
|
||||
|
||||
[[package]]
|
||||
name = "about-time"
|
||||
@@ -844,28 +844,6 @@ test = ["beautifulsoup4 (>=4.8.0)", "coverage (>=4.5.4)", "fixtures (>=3.0.0)",
|
||||
toml = ["tomli (>=1.1.0) ; python_version < \"3.11\""]
|
||||
yaml = ["PyYAML"]
|
||||
|
||||
[[package]]
|
||||
name = "bc-detect-secrets"
|
||||
version = "1.5.41"
|
||||
description = "Tool for detecting secrets in the codebase"
|
||||
optional = false
|
||||
python-versions = ">=3.8"
|
||||
groups = ["main"]
|
||||
markers = "python_version >= \"3.10\""
|
||||
files = [
|
||||
{file = "bc_detect_secrets-1.5.41-py3-none-any.whl", hash = "sha256:629df912f2a4f4d5039cc1fece906c34700586f7db1ae6a8d1c830c25df6db9b"},
|
||||
{file = "bc_detect_secrets-1.5.41.tar.gz", hash = "sha256:4bd08292a975bfc9b95771e118dd1131e1afbd479610eb29e4e0c15bd33677fc"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
pyyaml = "*"
|
||||
requests = "*"
|
||||
unidiff = "*"
|
||||
|
||||
[package.extras]
|
||||
gibberish = ["gibberish-detector"]
|
||||
word-list = ["pyahocorasick"]
|
||||
|
||||
[[package]]
|
||||
name = "bc-detect-secrets"
|
||||
version = "1.5.44"
|
||||
@@ -873,7 +851,6 @@ description = "Tool for detecting secrets in the codebase"
|
||||
optional = false
|
||||
python-versions = ">=3.8"
|
||||
groups = ["main"]
|
||||
markers = "python_version < \"3.10\""
|
||||
files = [
|
||||
{file = "bc_detect_secrets-1.5.44-py3-none-any.whl", hash = "sha256:0ab63d6c4f6680ec2dbe42cc3c63480568c55dbb6254afcc5bb6d4375a4e1d27"},
|
||||
{file = "bc_detect_secrets-1.5.44.tar.gz", hash = "sha256:bebd82c56055c600335f85db95f7ca3b434087f16292a0396a60705de1b94183"},
|
||||
@@ -1341,76 +1318,14 @@ files = [
|
||||
|
||||
[[package]]
|
||||
name = "checkov"
|
||||
version = "3.2.439"
|
||||
version = "3.2.445"
|
||||
description = "Infrastructure as code static analysis"
|
||||
optional = false
|
||||
python-versions = ">=3.8"
|
||||
groups = ["main"]
|
||||
markers = "python_version >= \"3.10\""
|
||||
files = [
|
||||
{file = "checkov-3.2.439-py3-none-any.whl", hash = "sha256:ab3a6b58989c979a5a6c337282a33f372fc1158c73215e3f488adffb56133f6f"},
|
||||
{file = "checkov-3.2.439.tar.gz", hash = "sha256:31b0c1ea0cc99141cf333a20f6fff42d82130d5eb2ba12ef4e4e3b92c7bce68e"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
aiodns = ">=3.0.0,<4.0.0"
|
||||
aiohttp = ">=3.8.0,<4.0.0"
|
||||
aiomultiprocess = ">=0.9.0,<0.10.0"
|
||||
argcomplete = ">=3.0.0,<4.0.0"
|
||||
asteval = "1.0.5"
|
||||
bc-detect-secrets = "1.5.41"
|
||||
bc-jsonpath-ng = "1.6.1"
|
||||
bc-python-hcl2 = "0.4.2"
|
||||
boto3 = "1.35.49"
|
||||
cachetools = ">=5.2.0,<6.0.0"
|
||||
charset-normalizer = ">=3.1.0,<4.0.0"
|
||||
click = ">=8.1.0,<9.0.0"
|
||||
cloudsplaining = ">=0.7.0,<0.8.0"
|
||||
colorama = ">=0.4.3,<0.5.0"
|
||||
configargparse = ">=1.5.3,<2.0.0"
|
||||
cyclonedx-python-lib = ">=6.0.0,<8.0.0"
|
||||
docker = ">=6.0.1,<8.0.0"
|
||||
dockerfile-parse = ">=2.0.0,<3.0.0"
|
||||
dpath = "2.1.3"
|
||||
gitpython = ">=3.1.30,<4.0.0"
|
||||
importlib-metadata = ">=6.0.0,<8.0.0"
|
||||
jmespath = ">=1.0.0,<2.0.0"
|
||||
jsonschema = ">=4.17.0,<5.0.0"
|
||||
junit-xml = ">=1.9,<2.0"
|
||||
license-expression = ">=30.1.0,<31.0.0"
|
||||
networkx = "<2.7"
|
||||
packageurl-python = ">=0.11.1,<0.14.0"
|
||||
packaging = ">=23.0,<24.0"
|
||||
prettytable = ">=3.6.0,<4.0.0"
|
||||
pycep-parser = "0.5.1"
|
||||
pydantic = ">=2.0.0,<3.0.0"
|
||||
pyston = {version = "2.3.5", markers = "python_version < \"3.11\" and (sys_platform == \"linux\" or sys_platform == \"darwin\") and platform_machine == \"x86_64\" and implementation_name == \"cpython\""}
|
||||
pyston-autoload = {version = "2.3.5", markers = "python_version < \"3.11\" and (sys_platform == \"linux\" or sys_platform == \"darwin\") and platform_machine == \"x86_64\" and implementation_name == \"cpython\""}
|
||||
pyyaml = ">=6.0.0,<7.0.0"
|
||||
requests = ">=2.28.0,<3.0.0"
|
||||
rustworkx = ">=0.13.0,<1.0.0"
|
||||
schema = "<=0.7.5"
|
||||
spdx-tools = ">=0.8.0,<0.9.0"
|
||||
tabulate = ">=0.9.0,<0.10.0"
|
||||
termcolor = ">=1.1.0,<2.4.0"
|
||||
tqdm = ">=4.65.0,<5.0.0"
|
||||
typing-extensions = ">=4.5.0,<5.0.0"
|
||||
yarl = ">=1.9.1,<2.0.0"
|
||||
|
||||
[package.extras]
|
||||
dev = ["GitPython (==3.1.41)", "bandit", "coverage (==7.6.1)", "coverage-badge", "jsonschema", "pytest (<8.0.0)"]
|
||||
|
||||
[[package]]
|
||||
name = "checkov"
|
||||
version = "3.2.442"
|
||||
description = "Infrastructure as code static analysis"
|
||||
optional = false
|
||||
python-versions = ">=3.8"
|
||||
groups = ["main"]
|
||||
markers = "python_version < \"3.10\""
|
||||
files = [
|
||||
{file = "checkov-3.2.442-py3-none-any.whl", hash = "sha256:e94a3283bff9b4a81e54e57b4a00b02259dec0d85b17bf17e00652d137bd1a6d"},
|
||||
{file = "checkov-3.2.442.tar.gz", hash = "sha256:e5206872de63d389cfb1b7c1212ce4b5a147986152e890461d87251726b4b0e7"},
|
||||
{file = "checkov-3.2.445-py3-none-any.whl", hash = "sha256:c0cd1109cbcf1c764198a7c64b911845b83776587a65e0677880ed67a39d1f73"},
|
||||
{file = "checkov-3.2.445.tar.gz", hash = "sha256:78705d34a9c7234bd7076c1970daedd18e823f79e7d83d2998bb84695cc33ebc"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
@@ -2738,8 +2653,6 @@ python-versions = "*"
|
||||
groups = ["dev"]
|
||||
files = [
|
||||
{file = "jsonpath-ng-1.7.0.tar.gz", hash = "sha256:f6f5f7fd4e5ff79c785f1573b394043b39849fb2bb47bcead935d12b00beab3c"},
|
||||
{file = "jsonpath_ng-1.7.0-py2-none-any.whl", hash = "sha256:898c93fc173f0c336784a3fa63d7434297544b7198124a68f9a3ef9597b0ae6e"},
|
||||
{file = "jsonpath_ng-1.7.0-py3-none-any.whl", hash = "sha256:f3d7f9e848cba1b6da28c55b1c26ff915dc9e0b1ba7e752a53d6da8d5cbd00b6"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
@@ -5673,7 +5586,6 @@ files = [
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp310-cp310-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:f66efbc1caa63c088dead1c4170d148eabc9b80d95fb75b6c92ac0aad2437d76"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp310-cp310-musllinux_1_1_i686.whl", hash = "sha256:22353049ba4181685023b25b5b51a574bce33e7f51c759371a7422dcae5402a6"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:932205970b9f9991b34f55136be327501903f7c66830e9760a8ffb15b07f05cd"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:a52d48f4e7bf9005e8f0a89209bf9a73f7190ddf0489eee5eb51377385f59f2a"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp310-cp310-win32.whl", hash = "sha256:3eac5a91891ceb88138c113f9db04f3cebdae277f5d44eaa3651a4f573e6a5da"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp310-cp310-win_amd64.whl", hash = "sha256:ab007f2f5a87bd08ab1499bdf96f3d5c6ad4dcfa364884cb4549aa0154b13a28"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp311-cp311-macosx_13_0_arm64.whl", hash = "sha256:4a6679521a58256a90b0d89e03992c15144c5f3858f40d7c18886023d7943db6"},
|
||||
@@ -5682,7 +5594,6 @@ files = [
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:811ea1594b8a0fb466172c384267a4e5e367298af6b228931f273b111f17ef52"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp311-cp311-musllinux_1_1_i686.whl", hash = "sha256:cf12567a7b565cbf65d438dec6cfbe2917d3c1bdddfce84a9930b7d35ea59642"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:7dd5adc8b930b12c8fc5b99e2d535a09889941aa0d0bd06f4749e9a9397c71d2"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:1492a6051dab8d912fc2adeef0e8c72216b24d57bd896ea607cb90bb0c4981d3"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp311-cp311-win32.whl", hash = "sha256:bd0a08f0bab19093c54e18a14a10b4322e1eacc5217056f3c063bd2f59853ce4"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp311-cp311-win_amd64.whl", hash = "sha256:a274fb2cb086c7a3dea4322ec27f4cb5cc4b6298adb583ab0e211a4682f241eb"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp312-cp312-macosx_14_0_arm64.whl", hash = "sha256:20b0f8dc160ba83b6dcc0e256846e1a02d044e13f7ea74a3d1d56ede4e48c632"},
|
||||
@@ -5691,7 +5602,6 @@ files = [
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:749c16fcc4a2b09f28843cda5a193e0283e47454b63ec4b81eaa2242f50e4ccd"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp312-cp312-musllinux_1_1_i686.whl", hash = "sha256:bf165fef1f223beae7333275156ab2022cffe255dcc51c27f066b4370da81e31"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:32621c177bbf782ca5a18ba4d7af0f1082a3f6e517ac2a18b3974d4edf349680"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:b82a7c94a498853aa0b272fd5bc67f29008da798d4f93a2f9f289feb8426a58d"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp312-cp312-win32.whl", hash = "sha256:e8c4ebfcfd57177b572e2040777b8abc537cdef58a2120e830124946aa9b42c5"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp312-cp312-win_amd64.whl", hash = "sha256:0467c5965282c62203273b838ae77c0d29d7638c8a4e3a1c8bdd3602c10904e4"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp313-cp313-macosx_14_0_arm64.whl", hash = "sha256:4c8c5d82f50bb53986a5e02d1b3092b03622c02c2eb78e29bec33fd9593bae1a"},
|
||||
@@ -5700,7 +5610,6 @@ files = [
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp313-cp313-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:96777d473c05ee3e5e3c3e999f5d23c6f4ec5b0c38c098b3a5229085f74236c6"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp313-cp313-musllinux_1_1_i686.whl", hash = "sha256:3bc2a80e6420ca8b7d3590791e2dfc709c88ab9152c00eeb511c9875ce5778bf"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp313-cp313-musllinux_1_1_x86_64.whl", hash = "sha256:e188d2699864c11c36cdfdada94d781fd5d6b0071cd9c427bceb08ad3d7c70e1"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:4f6f3eac23941b32afccc23081e1f50612bdbe4e982012ef4f5797986828cd01"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp313-cp313-win32.whl", hash = "sha256:6442cb36270b3afb1b4951f060eccca1ce49f3d087ca1ca4563a6eb479cb3de6"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp313-cp313-win_amd64.whl", hash = "sha256:e5b8daf27af0b90da7bb903a876477a9e6d7270be6146906b276605997c7e9a3"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp39-cp39-macosx_12_0_arm64.whl", hash = "sha256:fc4b630cd3fa2cf7fce38afa91d7cfe844a9f75d7f0f36393fa98815e911d987"},
|
||||
@@ -5709,7 +5618,6 @@ files = [
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp39-cp39-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:e2f1c3765db32be59d18ab3953f43ab62a761327aafc1594a2a1fbe038b8b8a7"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp39-cp39-musllinux_1_1_i686.whl", hash = "sha256:d85252669dc32f98ebcd5d36768f5d4faeaeaa2d655ac0473be490ecdae3c285"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:e143ada795c341b56de9418c58d028989093ee611aa27ffb9b7f609c00d813ed"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp39-cp39-musllinux_1_2_aarch64.whl", hash = "sha256:2c59aa6170b990d8d2719323e628aaf36f3bfbc1c26279c0eeeb24d05d2d11c7"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp39-cp39-win32.whl", hash = "sha256:beffaed67936fbbeffd10966a4eb53c402fafd3d6833770516bf7314bc6ffa12"},
|
||||
{file = "ruamel.yaml.clib-0.2.12-cp39-cp39-win_amd64.whl", hash = "sha256:040ae85536960525ea62868b642bdb0c2cc6021c9f9d507810c0c604e66f5a7b"},
|
||||
{file = "ruamel.yaml.clib-0.2.12.tar.gz", hash = "sha256:6c8fbb13ec503f99a91901ab46e0b07ae7941cd527393187039aec586fdfd36f"},
|
||||
@@ -6296,7 +6204,6 @@ description = "HTTP library with thread-safe connection pooling, file post, and
|
||||
optional = false
|
||||
python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,>=2.7"
|
||||
groups = ["main", "dev", "docs"]
|
||||
markers = "python_version < \"3.10\""
|
||||
files = [
|
||||
{file = "urllib3-1.26.20-py2.py3-none-any.whl", hash = "sha256:0ed14ccfbf1c30a9072c7ca157e4319b70d65f623e91e7b32fadb2853431016e"},
|
||||
{file = "urllib3-1.26.20.tar.gz", hash = "sha256:40c2dc0c681e47eb8f90e7e27bf6ff7df2e677421fd46756da1161c39ca70d32"},
|
||||
@@ -6307,25 +6214,6 @@ brotli = ["brotli (==1.0.9) ; os_name != \"nt\" and python_version < \"3\" and p
|
||||
secure = ["certifi", "cryptography (>=1.3.4)", "idna (>=2.0.0)", "ipaddress ; python_version == \"2.7\"", "pyOpenSSL (>=0.14)", "urllib3-secure-extra"]
|
||||
socks = ["PySocks (>=1.5.6,!=1.5.7,<2.0)"]
|
||||
|
||||
[[package]]
|
||||
name = "urllib3"
|
||||
version = "2.5.0"
|
||||
description = "HTTP library with thread-safe connection pooling, file post, and more."
|
||||
optional = false
|
||||
python-versions = ">=3.9"
|
||||
groups = ["main", "dev", "docs"]
|
||||
markers = "python_version >= \"3.10\""
|
||||
files = [
|
||||
{file = "urllib3-2.5.0-py3-none-any.whl", hash = "sha256:e6b01673c0fa6a13e374b50871808eb3bf7046c4b125b216f6bf1cc604cff0dc"},
|
||||
{file = "urllib3-2.5.0.tar.gz", hash = "sha256:3fc47733c7e419d4bc3f6b3dc2b4f890bb743906a30d56ba4a5bfa4bbff92760"},
|
||||
]
|
||||
|
||||
[package.extras]
|
||||
brotli = ["brotli (>=1.0.9) ; platform_python_implementation == \"CPython\"", "brotlicffi (>=0.8.0) ; platform_python_implementation != \"CPython\""]
|
||||
h2 = ["h2 (>=4,<5)"]
|
||||
socks = ["pysocks (>=1.5.6,!=1.5.7,<2.0)"]
|
||||
zstd = ["zstandard (>=0.18.0)"]
|
||||
|
||||
[[package]]
|
||||
name = "virtualenv"
|
||||
version = "20.31.2"
|
||||
@@ -6708,4 +6596,4 @@ type = ["pytest-mypy"]
|
||||
[metadata]
|
||||
lock-version = "2.1"
|
||||
python-versions = ">3.9.1,<3.13"
|
||||
content-hash = "d72c55b52949ba94f0c68004d5b778edb69514a05bbb7aba8d641b5058a99fd5"
|
||||
content-hash = "c442552635c8e904d1c7a50f4787c8e90ec90787960ee1867f2235a7aa2205f0"
|
||||
|
||||
@@ -40,6 +40,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
|
||||
- `keyvault_ensure_public_network_access_disabled` check for Azure provider. [(#8072)](https://github.com/prowler-cloud/prowler/pull/8072)
|
||||
- New check `monitor_alert_service_health_exists` for Azure provider [(#8067)](https://github.com/prowler-cloud/prowler/pull/8067)
|
||||
- Replace `Domain.Read.All` with `Directory.Read.All` in Azure and M365 docs [(#8075)](https://github.com/prowler-cloud/prowler/pull/8075)
|
||||
- Refactor IaC provider to use Checkov as Python library [(#8093)](https://github.com/prowler-cloud/prowler/pull/8093)
|
||||
|
||||
### Fixed
|
||||
- Consolidate Azure Storage file service properties to the account level, improving the accuracy of the `storage_ensure_file_shares_soft_delete_is_enabled` check [(#8087)](https://github.com/prowler-cloud/prowler/pull/8087)
|
||||
|
||||
@@ -28,6 +28,7 @@ Mutelist:
|
||||
Tags:
|
||||
- "test=test" # Will ignore every resource containing the string "test" and the tags 'test=test' and
|
||||
- "project=test|project=stage" # either of ('project=test' OR project=stage) in account 123456789012 and every region
|
||||
- "environment=prod" # Will ignore every resource except in account 123456789012 except the ones containing the string "test" and tag environment=prod
|
||||
|
||||
"*":
|
||||
Checks:
|
||||
@@ -46,9 +47,6 @@ Mutelist:
|
||||
- "*"
|
||||
Tags:
|
||||
- "environment=dev" # Will ignore every resource containing the tag 'environment=dev' in every account and region
|
||||
|
||||
"*":
|
||||
Checks:
|
||||
"ecs_task_definitions_no_environment_secrets":
|
||||
Regions:
|
||||
- "*"
|
||||
@@ -60,16 +58,3 @@ Mutelist:
|
||||
Regions:
|
||||
- "eu-west-1"
|
||||
- "eu-south-2" # Will ignore every resource in check ecs_task_definitions_no_environment_secrets except the ones in account 0123456789012 located in eu-south-2 or eu-west-1
|
||||
|
||||
"123456789012":
|
||||
Checks:
|
||||
"*":
|
||||
Regions:
|
||||
- "*"
|
||||
Resources:
|
||||
- "*"
|
||||
Exceptions:
|
||||
Resources:
|
||||
- "test"
|
||||
Tags:
|
||||
- "environment=prod" # Will ignore every resource except in account 123456789012 except the ones containing the string "test" and tag environment=prod
|
||||
|
||||
@@ -28,9 +28,6 @@ Mutelist:
|
||||
Tags:
|
||||
- "test=test" # Will ignore every resource containing the string "test" and the tags 'test=test' and
|
||||
- "project=test|project=stage" # either of ('project=test' OR project=stage) in Azure subscription 1 and every location
|
||||
|
||||
"*":
|
||||
Checks:
|
||||
"admincenter_*":
|
||||
Regions:
|
||||
- "*"
|
||||
|
||||
@@ -7,6 +7,7 @@ from dataclasses import asdict, dataclass, is_dataclass
|
||||
from enum import Enum
|
||||
from typing import Any, Dict, Optional, Set
|
||||
|
||||
from checkov.common.output.record import Record
|
||||
from pydantic.v1 import BaseModel, ValidationError, validator
|
||||
|
||||
from prowler.config.config import Provider
|
||||
@@ -441,6 +442,8 @@ class Check_Report:
|
||||
self.resource = resource.to_dict()
|
||||
elif is_dataclass(resource):
|
||||
self.resource = asdict(resource)
|
||||
elif hasattr(resource, "__dict__"):
|
||||
self.resource = resource.__dict__
|
||||
else:
|
||||
logger.error(
|
||||
f"Resource metadata {type(resource)} in {self.check_metadata.CheckID} could not be converted to dict"
|
||||
@@ -622,7 +625,7 @@ class CheckReportIAC(Check_Report):
|
||||
resource_path: str
|
||||
resource_line_range: str
|
||||
|
||||
def __init__(self, metadata: dict = {}, finding: dict = {}) -> None:
|
||||
def __init__(self, metadata: dict = {}, resource: Record = None) -> None:
|
||||
"""
|
||||
Initialize the IAC Check's finding information from a Checkov failed_check dict.
|
||||
|
||||
@@ -630,11 +633,10 @@ class CheckReportIAC(Check_Report):
|
||||
metadata (Dict): Optional check metadata (can be None).
|
||||
failed_check (dict): A single failed_check result from Checkov's JSON output.
|
||||
"""
|
||||
super().__init__(metadata, finding)
|
||||
|
||||
self.resource_name = getattr(finding, "resource", "")
|
||||
self.resource_path = getattr(finding, "file_path", "")
|
||||
self.resource_line_range = getattr(finding, "file_line_range", "")
|
||||
super().__init__(metadata, resource)
|
||||
self.resource_name = resource.resource
|
||||
self.resource_path = resource.file_path
|
||||
self.resource_line_range = resource.file_line_range
|
||||
|
||||
|
||||
@dataclass
|
||||
|
||||
@@ -286,8 +286,8 @@ class Finding(BaseModel):
|
||||
output_data["auth_method"] = "local" # Until we support remote repos
|
||||
output_data["account_uid"] = "iac"
|
||||
output_data["account_name"] = "iac"
|
||||
output_data["resource_name"] = check_output.resource["resource"]
|
||||
output_data["resource_uid"] = check_output.resource["resource"]
|
||||
output_data["resource_name"] = check_output.resource_name
|
||||
output_data["resource_uid"] = check_output.resource_name
|
||||
output_data["region"] = check_output.resource_path
|
||||
output_data["resource_line_range"] = check_output.resource_line_range
|
||||
output_data["framework"] = (
|
||||
|
||||
@@ -246,6 +246,8 @@ class Provider(ABC):
|
||||
elif "iac" in provider_class_name.lower():
|
||||
provider_class(
|
||||
scan_path=arguments.scan_path,
|
||||
frameworks=arguments.frameworks,
|
||||
exclude_path=arguments.exclude_path,
|
||||
config_path=arguments.config_file,
|
||||
fixer_config=fixer_config,
|
||||
)
|
||||
|
||||
@@ -1,8 +1,39 @@
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from typing import List
|
||||
|
||||
from checkov.ansible.runner import Runner as AnsibleRunner
|
||||
from checkov.argo_workflows.runner import Runner as ArgoWorkflowsRunner
|
||||
from checkov.arm.runner import Runner as ArmRunner
|
||||
from checkov.azure_pipelines.runner import Runner as AzurePipelinesRunner
|
||||
from checkov.bicep.runner import Runner as BicepRunner
|
||||
from checkov.bitbucket.runner import Runner as BitbucketRunner
|
||||
from checkov.bitbucket_pipelines.runner import Runner as BitbucketPipelinesRunner
|
||||
from checkov.cdk.runner import CdkRunner
|
||||
from checkov.circleci_pipelines.runner import Runner as CircleciPipelinesRunner
|
||||
from checkov.cloudformation.runner import Runner as CfnRunner
|
||||
from checkov.common.output.record import Record
|
||||
from checkov.common.output.report import Report
|
||||
from checkov.common.runners.runner_registry import RunnerRegistry
|
||||
from checkov.dockerfile.runner import Runner as DockerfileRunner
|
||||
from checkov.github.runner import Runner as GithubRunner
|
||||
from checkov.github_actions.runner import Runner as GithubActionsRunner
|
||||
from checkov.gitlab.runner import Runner as GitlabRunner
|
||||
from checkov.gitlab_ci.runner import Runner as GitlabCiRunner
|
||||
from checkov.helm.runner import Runner as HelmRunner
|
||||
from checkov.json_doc.runner import Runner as JsonDocRunner
|
||||
from checkov.kubernetes.runner import Runner as K8sRunner
|
||||
from checkov.kustomize.runner import Runner as KustomizeRunner
|
||||
from checkov.openapi.runner import Runner as OpenapiRunner
|
||||
from checkov.runner_filter import RunnerFilter
|
||||
from checkov.sast.runner import Runner as SastRunner
|
||||
from checkov.sca_image.runner import Runner as ScaImageRunner
|
||||
from checkov.sca_package_2.runner import Runner as ScaPackage2Runner
|
||||
from checkov.secrets.runner import Runner as SecretsRunner
|
||||
from checkov.serverless.runner import Runner as ServerlessRunner
|
||||
from checkov.terraform.runner import Runner as TerraformRunner
|
||||
from checkov.terraform_json.runner import TerraformJsonRunner
|
||||
from checkov.yaml_doc.runner import Runner as YamlDocRunner
|
||||
from colorama import Fore, Style
|
||||
|
||||
from prowler.config.config import (
|
||||
@@ -23,6 +54,8 @@ class IacProvider(Provider):
|
||||
def __init__(
|
||||
self,
|
||||
scan_path: str = ".",
|
||||
frameworks: list[str] = ["all"],
|
||||
exclude_path: list[str] = [],
|
||||
config_path: str = None,
|
||||
config_content: dict = None,
|
||||
fixer_config: dict = {},
|
||||
@@ -30,6 +63,8 @@ class IacProvider(Provider):
|
||||
logger.info("Instantiating IAC Provider...")
|
||||
|
||||
self.scan_path = scan_path
|
||||
self.frameworks = frameworks
|
||||
self.exclude_path = exclude_path
|
||||
self.region = "global"
|
||||
self.audited_account = "local-iac"
|
||||
self._session = None
|
||||
@@ -86,7 +121,9 @@ class IacProvider(Provider):
|
||||
"""IAC provider doesn't need a session since it uses Checkov directly"""
|
||||
return None
|
||||
|
||||
def _process_check(self, finding: dict, check: dict, status: str) -> CheckReportIAC:
|
||||
def _process_check(
|
||||
self, finding: Report, check: Record, status: str
|
||||
) -> CheckReportIAC:
|
||||
"""
|
||||
Process a single check (failed or passed) and create a CheckReportIAC object.
|
||||
|
||||
@@ -98,109 +135,110 @@ class IacProvider(Provider):
|
||||
Returns:
|
||||
CheckReportIAC: The processed check report
|
||||
"""
|
||||
metadata_dict = {
|
||||
"Provider": "iac",
|
||||
"CheckID": check.get("check_id", ""),
|
||||
"CheckTitle": check.get("check_name", ""),
|
||||
"CheckType": ["Infrastructure as Code"],
|
||||
"ServiceName": finding["check_type"],
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": (
|
||||
check.get("severity", "low").lower() if check.get("severity") else "low"
|
||||
),
|
||||
"ResourceType": "iac",
|
||||
"Description": check.get("check_name", ""),
|
||||
"Risk": "",
|
||||
"RelatedUrl": (
|
||||
check.get("guideline", "") if check.get("guideline") else ""
|
||||
),
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"NativeIaC": "",
|
||||
"Terraform": "",
|
||||
"CLI": "",
|
||||
"Other": "",
|
||||
try:
|
||||
metadata_dict = {
|
||||
"Provider": "iac",
|
||||
"CheckID": check.check_id,
|
||||
"CheckTitle": check.check_name,
|
||||
"CheckType": ["Infrastructure as Code"],
|
||||
"ServiceName": finding.check_type,
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": (check.severity.lower() if check.severity else "low"),
|
||||
"ResourceType": finding.check_type,
|
||||
"Description": check.check_name,
|
||||
"Risk": "",
|
||||
"RelatedUrl": (check.guideline if check.guideline else ""),
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"NativeIaC": "",
|
||||
"Terraform": "",
|
||||
"CLI": "",
|
||||
"Other": "",
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "",
|
||||
"Url": (check.guideline if check.guideline else ""),
|
||||
},
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "",
|
||||
"Url": (
|
||||
check.get("guideline", "") if check.get("guideline") else ""
|
||||
),
|
||||
},
|
||||
},
|
||||
"Categories": [],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "",
|
||||
}
|
||||
"Categories": [],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "",
|
||||
}
|
||||
|
||||
# Convert metadata dict to JSON string
|
||||
metadata = json.dumps(metadata_dict)
|
||||
# Convert metadata dict to JSON string
|
||||
metadata = json.dumps(metadata_dict)
|
||||
|
||||
report = CheckReportIAC(metadata=metadata, finding=check)
|
||||
report.status = status
|
||||
report.resource_tags = check.get("entity_tags", {})
|
||||
report.status_extended = check.get("check_name", "")
|
||||
if status == "MUTED":
|
||||
report.muted = True
|
||||
return report
|
||||
report = CheckReportIAC(metadata=metadata, resource=check)
|
||||
report.status = status
|
||||
report.resource_tags = check.entity_tags
|
||||
report.status_extended = check.check_name
|
||||
if status == "MUTED":
|
||||
report.muted = True
|
||||
return report
|
||||
except Exception as error:
|
||||
logger.critical(
|
||||
f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
def run(self) -> List[CheckReportIAC]:
|
||||
return self.run_scan(self.scan_path)
|
||||
return self.run_scan(self.scan_path, self.frameworks, self.exclude_path)
|
||||
|
||||
def run_scan(self, directory: str) -> List[CheckReportIAC]:
|
||||
def run_scan(
|
||||
self, directory: str, frameworks: list[str], exclude_path: list[str]
|
||||
) -> List[CheckReportIAC]:
|
||||
try:
|
||||
logger.info(f"Running IaC scan on {directory}...")
|
||||
|
||||
# Run Checkov with JSON output
|
||||
process = subprocess.run(
|
||||
["checkov", "-d", directory, "-o", "json"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
runners = [
|
||||
TerraformRunner(),
|
||||
CfnRunner(),
|
||||
K8sRunner(),
|
||||
ArmRunner(),
|
||||
ServerlessRunner(),
|
||||
DockerfileRunner(),
|
||||
YamlDocRunner(),
|
||||
OpenapiRunner(),
|
||||
SastRunner(),
|
||||
ScaImageRunner(),
|
||||
ScaPackage2Runner(),
|
||||
SecretsRunner(),
|
||||
AnsibleRunner(),
|
||||
ArgoWorkflowsRunner(),
|
||||
BitbucketRunner(),
|
||||
BitbucketPipelinesRunner(),
|
||||
CdkRunner(),
|
||||
CircleciPipelinesRunner(),
|
||||
GithubRunner(),
|
||||
GithubActionsRunner(),
|
||||
GitlabRunner(),
|
||||
GitlabCiRunner(),
|
||||
HelmRunner(),
|
||||
JsonDocRunner(),
|
||||
TerraformJsonRunner(),
|
||||
KustomizeRunner(),
|
||||
AzurePipelinesRunner(),
|
||||
BicepRunner(),
|
||||
]
|
||||
runner_filter = RunnerFilter(
|
||||
framework=frameworks, excluded_paths=exclude_path
|
||||
)
|
||||
# Log Checkov's error output if any
|
||||
if process.stderr:
|
||||
logger.error(process.stderr)
|
||||
|
||||
try:
|
||||
output = json.loads(process.stdout)
|
||||
if not output:
|
||||
logger.warning("No findings returned from Checkov scan")
|
||||
return []
|
||||
except Exception as error:
|
||||
logger.critical(
|
||||
f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
|
||||
)
|
||||
sys.exit(1)
|
||||
registry = RunnerRegistry("", runner_filter, *runners)
|
||||
checkov_reports = registry.run(root_folder=directory)
|
||||
|
||||
reports = []
|
||||
reports: List[CheckReportIAC] = []
|
||||
for report in checkov_reports:
|
||||
|
||||
# If only one framework has findings, the output is a dict, otherwise it's a list of dicts
|
||||
if isinstance(output, dict):
|
||||
output = [output]
|
||||
for failed in report.failed_checks:
|
||||
reports.append(self._process_check(report, failed, "FAIL"))
|
||||
|
||||
# Process all frameworks findings
|
||||
for finding in output:
|
||||
results = finding.get("results", {})
|
||||
for passed in report.passed_checks:
|
||||
reports.append(self._process_check(report, passed, "PASS"))
|
||||
|
||||
# Process failed checks
|
||||
failed_checks = results.get("failed_checks", [])
|
||||
for failed_check in failed_checks:
|
||||
report = self._process_check(finding, failed_check, "FAIL")
|
||||
reports.append(report)
|
||||
|
||||
# Process passed checks
|
||||
passed_checks = results.get("passed_checks", [])
|
||||
for passed_check in passed_checks:
|
||||
report = self._process_check(finding, passed_check, "PASS")
|
||||
reports.append(report)
|
||||
|
||||
# Process skipped checks (muted)
|
||||
skipped_checks = results.get("skipped_checks", [])
|
||||
for skipped_check in skipped_checks:
|
||||
report = self._process_check(finding, skipped_check, "MUTED")
|
||||
reports.append(report)
|
||||
for skipped in report.skipped_checks:
|
||||
reports.append(self._process_check(report, skipped, "MUTED"))
|
||||
|
||||
return reports
|
||||
|
||||
@@ -214,5 +252,12 @@ class IacProvider(Provider):
|
||||
report_lines = [
|
||||
f"Directory: {Fore.YELLOW}{self.scan_path}{Style.RESET_ALL}",
|
||||
]
|
||||
if self.exclude_path:
|
||||
report_lines.append(
|
||||
f"Excluded paths: {Fore.YELLOW}{', '.join(self.exclude_path)}{Style.RESET_ALL}"
|
||||
)
|
||||
report_lines.append(
|
||||
f"Frameworks: {Fore.YELLOW}{', '.join(self.frameworks)}{Style.RESET_ALL}"
|
||||
)
|
||||
report_title = f"{Style.BRIGHT}Scanning local IaC directory:{Style.RESET_ALL}"
|
||||
print_boxes(report_lines, report_title)
|
||||
|
||||
@@ -1,3 +1,36 @@
|
||||
FRAMEWORK_CHOICES = [
|
||||
"ansible",
|
||||
"argo_workflows",
|
||||
"arm",
|
||||
"azure_pipelines",
|
||||
"bicep",
|
||||
"bitbucket",
|
||||
"bitbucket_pipelines",
|
||||
"cdk",
|
||||
"circleci_pipelines",
|
||||
"cloudformation",
|
||||
"dockerfile",
|
||||
"github",
|
||||
"github_actions",
|
||||
"gitlab",
|
||||
"gitlab_ci",
|
||||
"helm",
|
||||
"json_doc",
|
||||
"kubernetes",
|
||||
"kustomize",
|
||||
"openapi",
|
||||
"policies_3d",
|
||||
"sast",
|
||||
"sca_image",
|
||||
"sca_package_2",
|
||||
"secrets",
|
||||
"serverless",
|
||||
"terraform",
|
||||
"terraform_json",
|
||||
"yaml_doc",
|
||||
]
|
||||
|
||||
|
||||
def init_parser(self):
|
||||
"""Init the IAC Provider CLI parser"""
|
||||
iac_parser = self.subparsers.add_parser(
|
||||
@@ -13,3 +46,20 @@ def init_parser(self):
|
||||
default=".",
|
||||
help="Path to the folder containing your infrastructure-as-code files. Default: current directory",
|
||||
)
|
||||
iac_scan_subparser.add_argument(
|
||||
"--frameworks",
|
||||
"-f",
|
||||
"--framework",
|
||||
dest="frameworks",
|
||||
nargs="+",
|
||||
default=["all"],
|
||||
choices=FRAMEWORK_CHOICES,
|
||||
help="Comma-separated list of frameworks to scan. Default: all",
|
||||
)
|
||||
iac_scan_subparser.add_argument(
|
||||
"--exclude-path",
|
||||
dest="exclude_path",
|
||||
nargs="+",
|
||||
default=[],
|
||||
help="Comma-separated list of paths to exclude from the scan. Default: none",
|
||||
)
|
||||
|
||||
+1
-1
@@ -59,7 +59,7 @@ dependencies = [
|
||||
"slack-sdk==3.34.0",
|
||||
"tabulate==0.9.0",
|
||||
"tzlocal==5.3.1",
|
||||
"checkov (>=3.2.434,<4.0.0)",
|
||||
"checkov==3.2.445",
|
||||
"py-iam-expand==0.1.0"
|
||||
]
|
||||
description = "Prowler is an Open Source security tool to perform AWS, GCP and Azure security best practices assessments, audits, incident response, continuous monitoring, hardening and forensics readiness. It contains hundreds of controls covering CIS, NIST 800, NIST CSF, CISA, RBI, FedRAMP, PCI-DSS, GDPR, HIPAA, FFIEC, SOC2, GXP, AWS Well-Architected Framework Security Pillar, AWS Foundational Technical Review (FTR), ENS (Spanish National Security Scheme) and your custom security frameworks."
|
||||
|
||||
@@ -1,67 +1,257 @@
|
||||
from checkov.common.models.enums import CheckResult
|
||||
from checkov.common.output.record import Record
|
||||
from checkov.common.output.report import Report
|
||||
|
||||
# IAC Provider Constants
|
||||
DEFAULT_SCAN_PATH = "."
|
||||
|
||||
# Sample Checkov Output
|
||||
SAMPLE_CHECKOV_OUTPUT = [
|
||||
{
|
||||
"check_type": "terraform",
|
||||
"results": {
|
||||
"failed_checks": [
|
||||
{
|
||||
"check_id": "CKV_AWS_1",
|
||||
"check_name": "Ensure S3 bucket has encryption enabled",
|
||||
"guideline": "https://docs.bridgecrew.io/docs/s3_1-s3-bucket-has-encryption-enabled",
|
||||
"severity": "low",
|
||||
},
|
||||
{
|
||||
"check_id": "CKV_AWS_2",
|
||||
"check_name": "Ensure S3 bucket has public access blocked",
|
||||
"guideline": "https://docs.bridgecrew.io/docs/s3_2-s3-bucket-has-public-access-blocked",
|
||||
"severity": "low",
|
||||
},
|
||||
],
|
||||
"passed_checks": [
|
||||
{
|
||||
"check_id": "CKV_AWS_3",
|
||||
"check_name": "Ensure S3 bucket has versioning enabled",
|
||||
"guideline": "https://docs.bridgecrew.io/docs/s3_3-s3-bucket-has-versioning-enabled",
|
||||
"severity": "low",
|
||||
}
|
||||
],
|
||||
},
|
||||
}
|
||||
]
|
||||
|
||||
# Sample Finding Data
|
||||
SAMPLE_FINDING = SAMPLE_CHECKOV_OUTPUT[0]
|
||||
SAMPLE_FINDING = Report(check_type="terraform")
|
||||
SAMPLE_FAILED_CHECK = Record(
|
||||
check_id="CKV_AWS_1",
|
||||
check_name="Ensure S3 bucket has encryption enabled",
|
||||
severity="low",
|
||||
file_path="test.tf",
|
||||
file_line_range=[1, 2],
|
||||
resource="aws_s3_bucket.test_bucket",
|
||||
evaluations=[],
|
||||
check_class="terraform",
|
||||
check_result=CheckResult.FAILED,
|
||||
code_block=[],
|
||||
file_abs_path="test.tf",
|
||||
)
|
||||
SAMPLE_FAILED_CHECK.guideline = (
|
||||
"https://docs.bridgecrew.io/docs/s3_1-s3-bucket-has-encryption-enabled"
|
||||
)
|
||||
|
||||
SAMPLE_FAILED_CHECK = {
|
||||
"check_id": "CKV_AWS_1",
|
||||
"check_name": "Ensure S3 bucket has encryption enabled",
|
||||
"guideline": "https://docs.bridgecrew.io/docs/s3_1-s3-bucket-has-encryption-enabled",
|
||||
"severity": "low",
|
||||
}
|
||||
SAMPLE_PASSED_CHECK = Record(
|
||||
check_id="CKV_AWS_3",
|
||||
check_name="Ensure S3 bucket has versioning enabled",
|
||||
severity="low",
|
||||
file_path="test.tf",
|
||||
file_line_range=[1, 2],
|
||||
resource="aws_s3_bucket.test_bucket",
|
||||
evaluations=[],
|
||||
check_class="terraform",
|
||||
check_result=CheckResult.PASSED,
|
||||
code_block=[],
|
||||
file_abs_path="test.tf",
|
||||
)
|
||||
SAMPLE_PASSED_CHECK.guideline = (
|
||||
"https://docs.bridgecrew.io/docs/s3_3-s3-bucket-has-versioning-enabled"
|
||||
)
|
||||
|
||||
SAMPLE_PASSED_CHECK = {
|
||||
"check_id": "CKV_AWS_3",
|
||||
"check_name": "Ensure S3 bucket has versioning enabled",
|
||||
"guideline": "https://docs.bridgecrew.io/docs/s3_3-s3-bucket-has-versioning-enabled",
|
||||
"severity": "low",
|
||||
}
|
||||
# Additional test fixtures for comprehensive testing
|
||||
SAMPLE_SKIPPED_CHECK = Record(
|
||||
check_id="CKV_AWS_2",
|
||||
check_name="Ensure S3 bucket has public access blocked",
|
||||
severity="high",
|
||||
file_path="test.tf",
|
||||
file_line_range=[3, 4],
|
||||
resource="aws_s3_bucket.test_bucket",
|
||||
evaluations=[],
|
||||
check_class="terraform",
|
||||
check_result=CheckResult.SKIPPED,
|
||||
code_block=[],
|
||||
file_abs_path="test.tf",
|
||||
)
|
||||
SAMPLE_SKIPPED_CHECK.guideline = (
|
||||
"https://docs.bridgecrew.io/docs/s3_2-s3-bucket-has-public-access-blocked"
|
||||
)
|
||||
|
||||
SAMPLE_HIGH_SEVERITY_CHECK = Record(
|
||||
check_id="CKV_AWS_4",
|
||||
check_name="Ensure S3 bucket has logging enabled",
|
||||
severity="HIGH",
|
||||
file_path="test.tf",
|
||||
file_line_range=[5, 6],
|
||||
resource="aws_s3_bucket.test_bucket",
|
||||
evaluations=[],
|
||||
check_class="terraform",
|
||||
check_result=CheckResult.FAILED,
|
||||
code_block=[],
|
||||
file_abs_path="test.tf",
|
||||
)
|
||||
SAMPLE_HIGH_SEVERITY_CHECK.guideline = (
|
||||
"https://docs.bridgecrew.io/docs/s3_4-s3-bucket-has-logging-enabled"
|
||||
)
|
||||
|
||||
def get_sample_checkov_json_output():
|
||||
"""Return sample Checkov JSON output as string"""
|
||||
import json
|
||||
SAMPLE_KUBERNETES_CHECK = Record(
|
||||
check_id="CKV_K8S_1",
|
||||
check_name="Ensure API server has audit logging enabled",
|
||||
severity="medium",
|
||||
file_path="deployment.yaml",
|
||||
file_line_range=[1, 10],
|
||||
resource="kubernetes_deployment.test_deployment",
|
||||
evaluations=[],
|
||||
check_class="kubernetes",
|
||||
check_result=CheckResult.FAILED,
|
||||
code_block=[],
|
||||
file_abs_path="deployment.yaml",
|
||||
)
|
||||
SAMPLE_KUBERNETES_CHECK.guideline = (
|
||||
"https://docs.bridgecrew.io/docs/k8s_1-api-server-has-audit-logging-enabled"
|
||||
)
|
||||
|
||||
return json.dumps(SAMPLE_CHECKOV_OUTPUT)
|
||||
SAMPLE_CLOUDFORMATION_CHECK = Record(
|
||||
check_id="CKV_AWS_5",
|
||||
check_name="Ensure CloudFormation stacks are not publicly accessible",
|
||||
severity="critical",
|
||||
file_path="template.yaml",
|
||||
file_line_range=[1, 20],
|
||||
resource="AWS::CloudFormation::Stack",
|
||||
evaluations=[],
|
||||
check_class="cloudformation",
|
||||
check_result=CheckResult.PASSED,
|
||||
code_block=[],
|
||||
file_abs_path="template.yaml",
|
||||
)
|
||||
SAMPLE_CLOUDFORMATION_CHECK.guideline = "https://docs.bridgecrew.io/docs/cfn_1-cloudformation-stacks-are-not-publicly-accessible"
|
||||
|
||||
# Sample findings for different frameworks
|
||||
SAMPLE_KUBERNETES_FINDING = Report(check_type="kubernetes")
|
||||
SAMPLE_CLOUDFORMATION_FINDING = Report(check_type="cloudformation")
|
||||
|
||||
def get_empty_checkov_output():
|
||||
"""Return empty Checkov output as string"""
|
||||
return "[]"
|
||||
# Additional fixtures for different test scenarios
|
||||
SAMPLE_CHECK_WITHOUT_GUIDELINE = Record(
|
||||
check_id="CKV_AWS_6",
|
||||
check_name="Test check without guideline",
|
||||
severity="low",
|
||||
file_path="test.tf",
|
||||
file_line_range=[1, 2],
|
||||
resource="aws_s3_bucket.test_bucket",
|
||||
evaluations=[],
|
||||
check_class="terraform",
|
||||
check_result=CheckResult.FAILED,
|
||||
code_block=[],
|
||||
file_abs_path="test.tf",
|
||||
)
|
||||
# Note: No guideline attribute set
|
||||
|
||||
SAMPLE_MEDIUM_SEVERITY_CHECK = Record(
|
||||
check_id="CKV_AWS_7",
|
||||
check_name="Ensure S3 bucket has proper access controls",
|
||||
severity="MEDIUM",
|
||||
file_path="test.tf",
|
||||
file_line_range=[7, 8],
|
||||
resource="aws_s3_bucket.test_bucket",
|
||||
evaluations=[],
|
||||
check_class="terraform",
|
||||
check_result=CheckResult.FAILED,
|
||||
code_block=[],
|
||||
file_abs_path="test.tf",
|
||||
)
|
||||
SAMPLE_MEDIUM_SEVERITY_CHECK.guideline = (
|
||||
"https://docs.bridgecrew.io/docs/s3_7-s3-bucket-has-proper-access-controls"
|
||||
)
|
||||
|
||||
def get_invalid_checkov_output():
|
||||
"""Return invalid JSON output as string"""
|
||||
return "invalid json output"
|
||||
SAMPLE_CRITICAL_SEVERITY_CHECK = Record(
|
||||
check_id="CKV_AWS_8",
|
||||
check_name="Ensure S3 bucket has encryption at rest",
|
||||
severity="CRITICAL",
|
||||
file_path="test.tf",
|
||||
file_line_range=[9, 10],
|
||||
resource="aws_s3_bucket.test_bucket",
|
||||
evaluations=[],
|
||||
check_class="terraform",
|
||||
check_result=CheckResult.FAILED,
|
||||
code_block=[],
|
||||
file_abs_path="test.tf",
|
||||
)
|
||||
SAMPLE_CRITICAL_SEVERITY_CHECK.guideline = (
|
||||
"https://docs.bridgecrew.io/docs/s3_8-s3-bucket-has-encryption-at-rest"
|
||||
)
|
||||
|
||||
# Sample reports for different frameworks
|
||||
SAMPLE_TERRAFORM_REPORT = Report(check_type="terraform")
|
||||
SAMPLE_KUBERNETES_REPORT = Report(check_type="kubernetes")
|
||||
SAMPLE_CLOUDFORMATION_REPORT = Report(check_type="cloudformation")
|
||||
SAMPLE_DOCKERFILE_REPORT = Report(check_type="dockerfile")
|
||||
SAMPLE_YAML_REPORT = Report(check_type="yaml")
|
||||
|
||||
# Sample checks for different frameworks
|
||||
SAMPLE_DOCKERFILE_CHECK = Record(
|
||||
check_id="CKV_DOCKER_1",
|
||||
check_name="Ensure base image is not using latest tag",
|
||||
severity="high",
|
||||
file_path="Dockerfile",
|
||||
file_line_range=[1, 1],
|
||||
resource="Dockerfile",
|
||||
evaluations=[],
|
||||
check_class="dockerfile",
|
||||
check_result=CheckResult.FAILED,
|
||||
code_block=[],
|
||||
file_abs_path="Dockerfile",
|
||||
)
|
||||
SAMPLE_DOCKERFILE_CHECK.guideline = (
|
||||
"https://docs.bridgecrew.io/docs/docker_1-base-image-not-using-latest-tag"
|
||||
)
|
||||
|
||||
SAMPLE_YAML_CHECK = Record(
|
||||
check_id="CKV_YAML_1",
|
||||
check_name="Ensure YAML file has proper indentation",
|
||||
severity="low",
|
||||
file_path="config.yaml",
|
||||
file_line_range=[1, 5],
|
||||
resource="config.yaml",
|
||||
evaluations=[],
|
||||
check_class="yaml",
|
||||
check_result=CheckResult.PASSED,
|
||||
code_block=[],
|
||||
file_abs_path="config.yaml",
|
||||
)
|
||||
SAMPLE_YAML_CHECK.guideline = (
|
||||
"https://docs.bridgecrew.io/docs/yaml_1-proper-indentation"
|
||||
)
|
||||
|
||||
# Sample checks with different statuses for comprehensive testing
|
||||
SAMPLE_ANOTHER_FAILED_CHECK = Record(
|
||||
check_id="CKV_AWS_9",
|
||||
check_name="Ensure S3 bucket has lifecycle policy",
|
||||
severity="medium",
|
||||
file_path="test.tf",
|
||||
file_line_range=[11, 12],
|
||||
resource="aws_s3_bucket.test_bucket",
|
||||
evaluations=[],
|
||||
check_class="terraform",
|
||||
check_result=CheckResult.FAILED,
|
||||
code_block=[],
|
||||
file_abs_path="test.tf",
|
||||
)
|
||||
SAMPLE_ANOTHER_FAILED_CHECK.guideline = (
|
||||
"https://docs.bridgecrew.io/docs/s3_9-s3-bucket-has-lifecycle-policy"
|
||||
)
|
||||
|
||||
SAMPLE_ANOTHER_PASSED_CHECK = Record(
|
||||
check_id="CKV_AWS_10",
|
||||
check_name="Ensure S3 bucket has proper tags",
|
||||
severity="low",
|
||||
file_path="test.tf",
|
||||
file_line_range=[13, 14],
|
||||
resource="aws_s3_bucket.test_bucket",
|
||||
evaluations=[],
|
||||
check_class="terraform",
|
||||
check_result=CheckResult.PASSED,
|
||||
code_block=[],
|
||||
file_abs_path="test.tf",
|
||||
)
|
||||
SAMPLE_ANOTHER_PASSED_CHECK.guideline = (
|
||||
"https://docs.bridgecrew.io/docs/s3_10-s3-bucket-has-proper-tags"
|
||||
)
|
||||
|
||||
SAMPLE_ANOTHER_SKIPPED_CHECK = Record(
|
||||
check_id="CKV_AWS_11",
|
||||
check_name="Ensure S3 bucket has cross-region replication",
|
||||
severity="high",
|
||||
file_path="test.tf",
|
||||
file_line_range=[15, 16],
|
||||
resource="aws_s3_bucket.test_bucket",
|
||||
evaluations=[],
|
||||
check_class="terraform",
|
||||
check_result=CheckResult.SKIPPED,
|
||||
code_block=[],
|
||||
file_abs_path="test.tf",
|
||||
)
|
||||
SAMPLE_ANOTHER_SKIPPED_CHECK.guideline = (
|
||||
"https://docs.bridgecrew.io/docs/s3_11-s3-bucket-has-cross-region-replication"
|
||||
)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -6,12 +6,24 @@ from prowler.lib.check.models import CheckReportIAC
|
||||
from prowler.providers.iac.iac_provider import IacProvider
|
||||
from tests.providers.iac.iac_fixtures import (
|
||||
DEFAULT_SCAN_PATH,
|
||||
SAMPLE_ANOTHER_FAILED_CHECK,
|
||||
SAMPLE_ANOTHER_PASSED_CHECK,
|
||||
SAMPLE_ANOTHER_SKIPPED_CHECK,
|
||||
SAMPLE_CHECK_WITHOUT_GUIDELINE,
|
||||
SAMPLE_CLOUDFORMATION_CHECK,
|
||||
SAMPLE_CRITICAL_SEVERITY_CHECK,
|
||||
SAMPLE_DOCKERFILE_CHECK,
|
||||
SAMPLE_DOCKERFILE_REPORT,
|
||||
SAMPLE_FAILED_CHECK,
|
||||
SAMPLE_FINDING,
|
||||
SAMPLE_HIGH_SEVERITY_CHECK,
|
||||
SAMPLE_KUBERNETES_CHECK,
|
||||
SAMPLE_KUBERNETES_FINDING,
|
||||
SAMPLE_MEDIUM_SEVERITY_CHECK,
|
||||
SAMPLE_PASSED_CHECK,
|
||||
get_empty_checkov_output,
|
||||
get_invalid_checkov_output,
|
||||
get_sample_checkov_json_output,
|
||||
SAMPLE_SKIPPED_CHECK,
|
||||
SAMPLE_YAML_CHECK,
|
||||
SAMPLE_YAML_REPORT,
|
||||
)
|
||||
|
||||
|
||||
@@ -46,10 +58,10 @@ class TestIacProvider:
|
||||
assert report.status == "FAIL"
|
||||
|
||||
assert report.check_metadata.Provider == "iac"
|
||||
assert report.check_metadata.CheckID == SAMPLE_FAILED_CHECK["check_id"]
|
||||
assert report.check_metadata.CheckTitle == SAMPLE_FAILED_CHECK["check_name"]
|
||||
assert report.check_metadata.CheckID == SAMPLE_FAILED_CHECK.check_id
|
||||
assert report.check_metadata.CheckTitle == SAMPLE_FAILED_CHECK.check_name
|
||||
assert report.check_metadata.Severity == "low"
|
||||
assert report.check_metadata.RelatedUrl == SAMPLE_FAILED_CHECK["guideline"]
|
||||
assert report.check_metadata.RelatedUrl == SAMPLE_FAILED_CHECK.guideline
|
||||
|
||||
def test_iac_provider_process_check_passed(self):
|
||||
"""Test processing a passed check"""
|
||||
@@ -61,72 +73,473 @@ class TestIacProvider:
|
||||
assert report.status == "PASS"
|
||||
|
||||
assert report.check_metadata.Provider == "iac"
|
||||
assert report.check_metadata.CheckID == SAMPLE_PASSED_CHECK["check_id"]
|
||||
assert report.check_metadata.CheckTitle == SAMPLE_PASSED_CHECK["check_name"]
|
||||
assert report.check_metadata.CheckID == SAMPLE_PASSED_CHECK.check_id
|
||||
assert report.check_metadata.CheckTitle == SAMPLE_PASSED_CHECK.check_name
|
||||
assert report.check_metadata.Severity == "low"
|
||||
assert report.check_metadata.RelatedUrl == SAMPLE_PASSED_CHECK.guideline
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_iac_provider_run_scan_success(self, mock_subprocess):
|
||||
"""Test successful IAC scan with Checkov"""
|
||||
def test_iac_provider_process_check_skipped(self):
|
||||
"""Test processing a skipped check"""
|
||||
provider = IacProvider()
|
||||
|
||||
mock_subprocess.return_value = MagicMock(
|
||||
stdout=get_sample_checkov_json_output(), stderr=""
|
||||
)
|
||||
report = provider._process_check(SAMPLE_FINDING, SAMPLE_SKIPPED_CHECK, "MUTED")
|
||||
|
||||
reports = provider.run_scan("/test/directory")
|
||||
assert isinstance(report, CheckReportIAC)
|
||||
assert report.status == "MUTED"
|
||||
assert report.muted is True
|
||||
|
||||
# Should have 2 failed checks + 1 passed check = 3 total reports
|
||||
assert len(reports) == 3
|
||||
assert report.check_metadata.Provider == "iac"
|
||||
assert report.check_metadata.CheckID == SAMPLE_SKIPPED_CHECK.check_id
|
||||
assert report.check_metadata.CheckTitle == SAMPLE_SKIPPED_CHECK.check_name
|
||||
assert report.check_metadata.Severity == "high"
|
||||
assert report.check_metadata.RelatedUrl == SAMPLE_SKIPPED_CHECK.guideline
|
||||
|
||||
# Check that we have both failed and passed reports
|
||||
failed_reports = [r for r in reports if r.status == "FAIL"]
|
||||
passed_reports = [r for r in reports if r.status == "PASS"]
|
||||
|
||||
assert len(failed_reports) == 2
|
||||
assert len(passed_reports) == 1
|
||||
|
||||
# Verify subprocess was called correctly
|
||||
mock_subprocess.assert_called_once_with(
|
||||
["checkov", "-d", "/test/directory", "-o", "json"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_iac_provider_run_scan_empty_output(self, mock_subprocess):
|
||||
"""Test IAC scan with empty Checkov output"""
|
||||
def test_iac_provider_process_check_high_severity(self):
|
||||
"""Test processing a high severity check"""
|
||||
provider = IacProvider()
|
||||
|
||||
mock_subprocess.return_value = MagicMock(
|
||||
stdout=get_empty_checkov_output(), stderr=""
|
||||
report = provider._process_check(
|
||||
SAMPLE_FINDING, SAMPLE_HIGH_SEVERITY_CHECK, "FAIL"
|
||||
)
|
||||
|
||||
reports = provider.run_scan("/test/directory")
|
||||
assert isinstance(report, CheckReportIAC)
|
||||
assert report.status == "FAIL"
|
||||
assert report.check_metadata.Severity == "high"
|
||||
|
||||
assert len(reports) == 0
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_iac_provider_run_scan_invalid_json(self, mock_subprocess):
|
||||
"""Test IAC scan with invalid JSON output"""
|
||||
def test_iac_provider_process_check_different_framework(self):
|
||||
"""Test processing a check from a different framework (Kubernetes)"""
|
||||
provider = IacProvider()
|
||||
|
||||
mock_subprocess.return_value = MagicMock(
|
||||
stdout=get_invalid_checkov_output(), stderr=""
|
||||
report = provider._process_check(
|
||||
SAMPLE_KUBERNETES_FINDING, SAMPLE_KUBERNETES_CHECK, "FAIL"
|
||||
)
|
||||
|
||||
with pytest.raises(SystemExit) as excinfo:
|
||||
provider.run_scan("/test/directory")
|
||||
assert isinstance(report, CheckReportIAC)
|
||||
assert report.status == "FAIL"
|
||||
assert report.check_metadata.ServiceName == "kubernetes"
|
||||
assert report.check_metadata.CheckID == SAMPLE_KUBERNETES_CHECK.check_id
|
||||
|
||||
assert excinfo.value.code == 1
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_iac_provider_run_scan_null_output(self, mock_subprocess):
|
||||
"""Test IAC scan with null Checkov output"""
|
||||
def test_iac_provider_process_check_no_guideline(self):
|
||||
"""Test processing a check without guideline URL"""
|
||||
provider = IacProvider()
|
||||
|
||||
mock_subprocess.return_value = MagicMock(stdout="null", stderr="")
|
||||
report = provider._process_check(
|
||||
SAMPLE_FINDING, SAMPLE_CHECK_WITHOUT_GUIDELINE, "FAIL"
|
||||
)
|
||||
|
||||
reports = provider.run_scan("/test/directory")
|
||||
assert isinstance(report, CheckReportIAC)
|
||||
assert report.status == "FAIL"
|
||||
assert report.check_metadata.RelatedUrl == ""
|
||||
|
||||
assert len(reports) == 0
|
||||
def test_iac_provider_process_check_medium_severity(self):
|
||||
"""Test processing a medium severity check"""
|
||||
provider = IacProvider()
|
||||
|
||||
report = provider._process_check(
|
||||
SAMPLE_FINDING, SAMPLE_MEDIUM_SEVERITY_CHECK, "FAIL"
|
||||
)
|
||||
|
||||
assert isinstance(report, CheckReportIAC)
|
||||
assert report.status == "FAIL"
|
||||
assert report.check_metadata.Severity == "medium"
|
||||
|
||||
def test_iac_provider_process_check_critical_severity(self):
|
||||
"""Test processing a critical severity check"""
|
||||
provider = IacProvider()
|
||||
|
||||
report = provider._process_check(
|
||||
SAMPLE_FINDING, SAMPLE_CRITICAL_SEVERITY_CHECK, "FAIL"
|
||||
)
|
||||
|
||||
assert isinstance(report, CheckReportIAC)
|
||||
assert report.status == "FAIL"
|
||||
assert report.check_metadata.Severity == "critical"
|
||||
|
||||
def test_iac_provider_process_check_dockerfile(self):
|
||||
"""Test processing a Dockerfile check"""
|
||||
provider = IacProvider()
|
||||
|
||||
report = provider._process_check(
|
||||
SAMPLE_DOCKERFILE_REPORT, SAMPLE_DOCKERFILE_CHECK, "FAIL"
|
||||
)
|
||||
|
||||
assert isinstance(report, CheckReportIAC)
|
||||
assert report.status == "FAIL"
|
||||
assert report.check_metadata.ServiceName == "dockerfile"
|
||||
assert report.check_metadata.CheckID == SAMPLE_DOCKERFILE_CHECK.check_id
|
||||
|
||||
def test_iac_provider_process_check_yaml(self):
|
||||
"""Test processing a YAML check"""
|
||||
provider = IacProvider()
|
||||
|
||||
report = provider._process_check(SAMPLE_YAML_REPORT, SAMPLE_YAML_CHECK, "PASS")
|
||||
|
||||
assert isinstance(report, CheckReportIAC)
|
||||
assert report.status == "PASS"
|
||||
assert report.check_metadata.ServiceName == "yaml"
|
||||
assert report.check_metadata.CheckID == SAMPLE_YAML_CHECK.check_id
|
||||
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerRegistry")
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerFilter")
|
||||
@patch("prowler.providers.iac.iac_provider.logger")
|
||||
def test_run_scan_success_with_failed_and_passed_checks(
|
||||
self, mock_logger, mock_runner_filter, mock_runner_registry
|
||||
):
|
||||
"""Test successful run_scan with both failed and passed checks"""
|
||||
# Setup mocks
|
||||
mock_registry_instance = Mock()
|
||||
mock_runner_registry.return_value = mock_registry_instance
|
||||
|
||||
# Create mock reports with failed and passed checks
|
||||
mock_report = Mock()
|
||||
mock_report.check_type = "terraform" # Set the check_type attribute
|
||||
mock_report.failed_checks = [SAMPLE_FAILED_CHECK]
|
||||
mock_report.passed_checks = [SAMPLE_PASSED_CHECK]
|
||||
mock_report.skipped_checks = []
|
||||
|
||||
mock_registry_instance.run.return_value = [mock_report]
|
||||
|
||||
provider = IacProvider()
|
||||
result = provider.run_scan("/test/directory", ["terraform"], [])
|
||||
|
||||
# Verify logger was called
|
||||
mock_logger.info.assert_called_with("Running IaC scan on /test/directory...")
|
||||
|
||||
# Verify RunnerFilter was created with correct parameters
|
||||
mock_runner_filter.assert_called_with(
|
||||
framework=["terraform"], excluded_paths=[]
|
||||
)
|
||||
|
||||
# Verify RunnerRegistry was created and run was called
|
||||
mock_runner_registry.assert_called_once()
|
||||
mock_registry_instance.run.assert_called_with(root_folder="/test/directory")
|
||||
|
||||
# Verify results
|
||||
assert len(result) == 2
|
||||
assert all(isinstance(report, CheckReportIAC) for report in result)
|
||||
|
||||
# Check that we have one FAIL and one PASS report
|
||||
statuses = [report.status for report in result]
|
||||
assert "FAIL" in statuses
|
||||
assert "PASS" in statuses
|
||||
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerRegistry")
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerFilter")
|
||||
@patch("prowler.providers.iac.iac_provider.logger")
|
||||
def test_run_scan_with_skipped_checks(
|
||||
self, mock_logger, mock_runner_filter, mock_runner_registry
|
||||
):
|
||||
"""Test run_scan with skipped checks (muted)"""
|
||||
# Setup mocks
|
||||
mock_registry_instance = Mock()
|
||||
mock_runner_registry.return_value = mock_registry_instance
|
||||
|
||||
# Create mock report with skipped checks
|
||||
mock_report = Mock()
|
||||
mock_report.check_type = "terraform" # Set the check_type attribute
|
||||
mock_report.failed_checks = []
|
||||
mock_report.passed_checks = []
|
||||
mock_report.skipped_checks = [SAMPLE_SKIPPED_CHECK]
|
||||
|
||||
mock_registry_instance.run.return_value = [mock_report]
|
||||
|
||||
provider = IacProvider()
|
||||
result = provider.run_scan("/test/directory", ["all"], ["exclude/path"])
|
||||
|
||||
# Verify RunnerFilter was created with correct parameters
|
||||
mock_runner_filter.assert_called_with(
|
||||
framework=["all"], excluded_paths=["exclude/path"]
|
||||
)
|
||||
|
||||
# Verify results
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], CheckReportIAC)
|
||||
assert result[0].status == "MUTED"
|
||||
assert result[0].muted is True
|
||||
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerRegistry")
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerFilter")
|
||||
@patch("prowler.providers.iac.iac_provider.logger")
|
||||
def test_run_scan_empty_results(
|
||||
self, mock_logger, mock_runner_filter, mock_runner_registry
|
||||
):
|
||||
"""Test run_scan with no findings"""
|
||||
# Setup mocks
|
||||
mock_registry_instance = Mock()
|
||||
mock_runner_registry.return_value = mock_registry_instance
|
||||
|
||||
# Create mock report with no checks
|
||||
mock_report = Mock()
|
||||
mock_report.check_type = "terraform" # Set the check_type attribute
|
||||
mock_report.failed_checks = []
|
||||
mock_report.passed_checks = []
|
||||
mock_report.skipped_checks = []
|
||||
|
||||
mock_registry_instance.run.return_value = [mock_report]
|
||||
|
||||
provider = IacProvider()
|
||||
result = provider.run_scan("/test/directory", ["kubernetes"], [])
|
||||
|
||||
# Verify results
|
||||
assert len(result) == 0
|
||||
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerRegistry")
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerFilter")
|
||||
@patch("prowler.providers.iac.iac_provider.logger")
|
||||
def test_run_scan_multiple_reports(
|
||||
self, mock_logger, mock_runner_filter, mock_runner_registry
|
||||
):
|
||||
"""Test run_scan with multiple reports from different frameworks"""
|
||||
# Setup mocks
|
||||
mock_registry_instance = Mock()
|
||||
mock_runner_registry.return_value = mock_registry_instance
|
||||
|
||||
# Create multiple mock reports
|
||||
mock_report1 = Mock()
|
||||
mock_report1.check_type = "terraform" # Set the check_type attribute
|
||||
mock_report1.failed_checks = [SAMPLE_FAILED_CHECK]
|
||||
mock_report1.passed_checks = []
|
||||
mock_report1.skipped_checks = []
|
||||
|
||||
mock_report2 = Mock()
|
||||
mock_report2.check_type = "kubernetes" # Set the check_type attribute
|
||||
mock_report2.failed_checks = []
|
||||
mock_report2.passed_checks = [SAMPLE_PASSED_CHECK]
|
||||
mock_report2.skipped_checks = []
|
||||
|
||||
mock_registry_instance.run.return_value = [mock_report1, mock_report2]
|
||||
|
||||
provider = IacProvider()
|
||||
result = provider.run_scan("/test/directory", ["terraform", "kubernetes"], [])
|
||||
|
||||
# Verify results
|
||||
assert len(result) == 2
|
||||
assert all(isinstance(report, CheckReportIAC) for report in result)
|
||||
|
||||
# Check that we have one FAIL and one PASS report
|
||||
statuses = [report.status for report in result]
|
||||
assert "FAIL" in statuses
|
||||
assert "PASS" in statuses
|
||||
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerRegistry")
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerFilter")
|
||||
@patch("prowler.providers.iac.iac_provider.logger")
|
||||
@patch("prowler.providers.iac.iac_provider.sys")
|
||||
def test_run_scan_exception_handling(
|
||||
self, mock_sys, mock_logger, mock_runner_filter, mock_runner_registry
|
||||
):
|
||||
"""Test run_scan exception handling"""
|
||||
# Setup mocks to raise an exception
|
||||
mock_registry_instance = Mock()
|
||||
mock_runner_registry.return_value = mock_registry_instance
|
||||
mock_registry_instance.run.side_effect = Exception("Test exception")
|
||||
|
||||
# Configure sys.exit to raise SystemExit
|
||||
mock_sys.exit.side_effect = SystemExit(1)
|
||||
|
||||
provider = IacProvider()
|
||||
|
||||
# The function should call sys.exit(1) when an exception occurs
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
provider.run_scan("/test/directory", ["terraform"], [])
|
||||
|
||||
assert exc_info.value.code == 1
|
||||
|
||||
# Verify logger was called with error information
|
||||
mock_logger.critical.assert_called_once()
|
||||
critical_call_args = mock_logger.critical.call_args[0][0]
|
||||
assert "Exception" in critical_call_args
|
||||
assert "Test exception" in critical_call_args
|
||||
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerRegistry")
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerFilter")
|
||||
@patch("prowler.providers.iac.iac_provider.logger")
|
||||
def test_run_scan_with_different_frameworks(
|
||||
self, mock_logger, mock_runner_filter, mock_runner_registry
|
||||
):
|
||||
"""Test run_scan with different framework configurations"""
|
||||
# Setup mocks
|
||||
mock_registry_instance = Mock()
|
||||
mock_runner_registry.return_value = mock_registry_instance
|
||||
|
||||
mock_report = Mock()
|
||||
mock_report.check_type = "terraform" # Set the check_type attribute
|
||||
mock_report.failed_checks = []
|
||||
mock_report.passed_checks = [SAMPLE_PASSED_CHECK]
|
||||
mock_report.skipped_checks = []
|
||||
|
||||
mock_registry_instance.run.return_value = [mock_report]
|
||||
|
||||
provider = IacProvider()
|
||||
|
||||
# Test with specific frameworks
|
||||
frameworks = ["terraform", "kubernetes", "cloudformation"]
|
||||
result = provider.run_scan("/test/directory", frameworks, [])
|
||||
|
||||
# Verify RunnerFilter was created with correct frameworks
|
||||
mock_runner_filter.assert_called_with(framework=frameworks, excluded_paths=[])
|
||||
|
||||
# Verify results
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerRegistry")
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerFilter")
|
||||
@patch("prowler.providers.iac.iac_provider.logger")
|
||||
def test_run_scan_with_exclude_paths(
|
||||
self, mock_logger, mock_runner_filter, mock_runner_registry
|
||||
):
|
||||
"""Test run_scan with exclude paths"""
|
||||
# Setup mocks
|
||||
mock_registry_instance = Mock()
|
||||
mock_runner_registry.return_value = mock_registry_instance
|
||||
|
||||
mock_report = Mock()
|
||||
mock_report.check_type = "terraform" # Set the check_type attribute
|
||||
mock_report.failed_checks = []
|
||||
mock_report.passed_checks = [SAMPLE_PASSED_CHECK]
|
||||
mock_report.skipped_checks = []
|
||||
|
||||
mock_registry_instance.run.return_value = [mock_report]
|
||||
|
||||
provider = IacProvider()
|
||||
|
||||
# Test with exclude paths
|
||||
exclude_paths = ["node_modules", ".git", "vendor"]
|
||||
result = provider.run_scan("/test/directory", ["all"], exclude_paths)
|
||||
|
||||
# Verify RunnerFilter was created with correct exclude paths
|
||||
mock_runner_filter.assert_called_with(
|
||||
framework=["all"], excluded_paths=exclude_paths
|
||||
)
|
||||
|
||||
# Verify results
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerRegistry")
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerFilter")
|
||||
@patch("prowler.providers.iac.iac_provider.logger")
|
||||
def test_run_scan_all_check_types(
|
||||
self, mock_logger, mock_runner_filter, mock_runner_registry
|
||||
):
|
||||
"""Test run_scan with all types of checks (failed, passed, skipped)"""
|
||||
# Setup mocks
|
||||
mock_registry_instance = Mock()
|
||||
mock_runner_registry.return_value = mock_registry_instance
|
||||
|
||||
mock_report = Mock()
|
||||
mock_report.check_type = "terraform" # Set the check_type attribute
|
||||
mock_report.failed_checks = [SAMPLE_FAILED_CHECK, SAMPLE_HIGH_SEVERITY_CHECK]
|
||||
mock_report.passed_checks = [SAMPLE_PASSED_CHECK, SAMPLE_CLOUDFORMATION_CHECK]
|
||||
mock_report.skipped_checks = [SAMPLE_SKIPPED_CHECK]
|
||||
|
||||
mock_registry_instance.run.return_value = [mock_report]
|
||||
|
||||
provider = IacProvider()
|
||||
result = provider.run_scan("/test/directory", ["all"], [])
|
||||
|
||||
# Verify results
|
||||
assert len(result) == 5 # 2 failed + 2 passed + 1 skipped
|
||||
|
||||
# Check status distribution
|
||||
statuses = [report.status for report in result]
|
||||
assert statuses.count("FAIL") == 2
|
||||
assert statuses.count("PASS") == 2
|
||||
assert statuses.count("MUTED") == 1
|
||||
|
||||
# Check that muted reports have muted=True
|
||||
muted_reports = [report for report in result if report.status == "MUTED"]
|
||||
assert all(report.muted for report in muted_reports)
|
||||
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerRegistry")
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerFilter")
|
||||
@patch("prowler.providers.iac.iac_provider.logger")
|
||||
def test_run_scan_no_reports_returned(
|
||||
self, mock_logger, mock_runner_filter, mock_runner_registry
|
||||
):
|
||||
"""Test run_scan when no reports are returned from registry"""
|
||||
# Setup mocks
|
||||
mock_registry_instance = Mock()
|
||||
mock_runner_registry.return_value = mock_registry_instance
|
||||
|
||||
# Return empty list of reports
|
||||
mock_registry_instance.run.return_value = []
|
||||
|
||||
provider = IacProvider()
|
||||
result = provider.run_scan("/test/directory", ["terraform"], [])
|
||||
|
||||
# Verify results
|
||||
assert len(result) == 0
|
||||
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerRegistry")
|
||||
@patch("prowler.providers.iac.iac_provider.RunnerFilter")
|
||||
@patch("prowler.providers.iac.iac_provider.logger")
|
||||
def test_run_scan_multiple_frameworks_with_different_checks(
|
||||
self, mock_logger, mock_runner_filter, mock_runner_registry
|
||||
):
|
||||
"""Test run_scan with multiple frameworks and different types of checks"""
|
||||
# Setup mocks
|
||||
mock_registry_instance = Mock()
|
||||
mock_runner_registry.return_value = mock_registry_instance
|
||||
|
||||
# Create reports for different frameworks
|
||||
terraform_report = Mock()
|
||||
terraform_report.check_type = "terraform"
|
||||
terraform_report.failed_checks = [
|
||||
SAMPLE_FAILED_CHECK,
|
||||
SAMPLE_ANOTHER_FAILED_CHECK,
|
||||
]
|
||||
terraform_report.passed_checks = [SAMPLE_PASSED_CHECK]
|
||||
terraform_report.skipped_checks = []
|
||||
|
||||
kubernetes_report = Mock()
|
||||
kubernetes_report.check_type = "kubernetes"
|
||||
kubernetes_report.failed_checks = [SAMPLE_KUBERNETES_CHECK]
|
||||
kubernetes_report.passed_checks = []
|
||||
kubernetes_report.skipped_checks = [SAMPLE_ANOTHER_SKIPPED_CHECK]
|
||||
|
||||
cloudformation_report = Mock()
|
||||
cloudformation_report.check_type = "cloudformation"
|
||||
cloudformation_report.failed_checks = []
|
||||
cloudformation_report.passed_checks = [
|
||||
SAMPLE_CLOUDFORMATION_CHECK,
|
||||
SAMPLE_ANOTHER_PASSED_CHECK,
|
||||
]
|
||||
cloudformation_report.skipped_checks = []
|
||||
|
||||
mock_registry_instance.run.return_value = [
|
||||
terraform_report,
|
||||
kubernetes_report,
|
||||
cloudformation_report,
|
||||
]
|
||||
|
||||
provider = IacProvider()
|
||||
result = provider.run_scan(
|
||||
"/test/directory", ["terraform", "kubernetes", "cloudformation"], []
|
||||
)
|
||||
|
||||
# Verify results
|
||||
assert (
|
||||
len(result) == 7
|
||||
) # 2 failed + 1 passed (terraform) + 1 failed + 1 skipped (kubernetes) + 2 passed (cloudformation)
|
||||
|
||||
# Check status distribution
|
||||
statuses = [report.status for report in result]
|
||||
assert statuses.count("FAIL") == 3
|
||||
assert statuses.count("PASS") == 3
|
||||
assert statuses.count("MUTED") == 1
|
||||
|
||||
def test_run_method_calls_run_scan(self):
|
||||
"""Test that the run method calls run_scan with correct parameters"""
|
||||
provider = IacProvider(
|
||||
scan_path="/custom/path", frameworks=["terraform"], exclude_path=["exclude"]
|
||||
)
|
||||
|
||||
with patch.object(provider, "run_scan") as mock_run_scan:
|
||||
mock_run_scan.return_value = []
|
||||
provider.run()
|
||||
|
||||
mock_run_scan.assert_called_once_with(
|
||||
"/custom/path", ["terraform"], ["exclude"]
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user