mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-09 21:14:22 +00:00
feat(codecommit): add codecommit service and codecommit_repository_no_secrets check (#11846)
Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>
This commit is contained in:
Siddhant Jadhav
co-authored by
Daniel Barranquero
Daniel Barranquero
parent
7a62926a09
commit
fc0204a40d
@@ -0,0 +1 @@
|
||||
`codecommit` service and `codecommit_repository_no_secrets` check for AWS provider, scanning files tracked at the tip of each repository's default branch for hardcoded secrets
|
||||
@@ -467,9 +467,11 @@ aws:
|
||||
secrets_ignore_patterns: []
|
||||
|
||||
# aws.awslambda_function_no_secrets_in_code
|
||||
# Glob patterns of file names inside the Lambda deployment package to skip
|
||||
# when scanning for secrets. Useful to suppress known false positives such
|
||||
# as .NET dependency manifests.
|
||||
# aws.codecommit_repository_no_secrets
|
||||
# Glob patterns of file names inside the Lambda deployment package or the
|
||||
# CodeCommit repository to skip when scanning for secrets. Useful to
|
||||
# suppress known false positives such as .NET dependency manifests or
|
||||
# package lock files.
|
||||
# Example:
|
||||
# secrets_ignore_files:
|
||||
# - "*.deps.json"
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
from prowler.providers.aws.services.codecommit.codecommit_service import CodeCommit
|
||||
from prowler.providers.common.provider import Provider
|
||||
|
||||
codecommit_client = CodeCommit(Provider.get_global_provider())
|
||||
+48
@@ -0,0 +1,48 @@
|
||||
{
|
||||
"Provider": "aws",
|
||||
"CheckID": "codecommit_repository_no_secrets",
|
||||
"CheckTitle": "CodeCommit repository has no secrets in its default branch",
|
||||
"CheckType": [
|
||||
"Software and Configuration Checks/AWS Security Best Practices",
|
||||
"TTPs/Credential Access",
|
||||
"Effects/Data Exposure",
|
||||
"Sensitive Data Identifications/Security"
|
||||
],
|
||||
"ServiceName": "codecommit",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "arn:partition:codecommit:region:account-id:repository-name",
|
||||
"Severity": "high",
|
||||
"ResourceType": "AwsCodeCommitRepository",
|
||||
"ResourceGroup": "devops",
|
||||
"Description": "**AWS CodeCommit repositories** are scanned for **hardcoded secrets** such as API keys, tokens, and passwords in every file tracked at the tip of the default branch. CodeCommit repositories, including their branches and commit history, are a top source of leaked credentials, and secrets can persist in history even after being removed from HEAD.",
|
||||
"Risk": "Hardcoded secrets committed to a CodeCommit repository can be read by any principal with `codecommit:GetFile` or `codecommit:GetBlob` permission, and remain in the commit history even after being deleted from the tip of a branch. Exposed credentials enable unauthorized access to downstream systems and services, leading to data exfiltration and further compromise.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://docs.aws.amazon.com/codecommit/latest/userguide/how-to-view-repository-details.html",
|
||||
"https://docs.aws.amazon.com/secretsmanager/latest/userguide/integrating_how-services-use-secrets_codecommit.html",
|
||||
"https://docs.prowler.com/developer-guide/secret-scanning-checks"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "aws codecommit get-file --repository-name <repository-name> --file-path <file-path>\ngit filter-repo --path <file-path> --invert-paths",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Identify every file and commit that contains the hardcoded secret using `git log -p` or the file and line reported by this check.\n2. Revoke and rotate the exposed credential immediately, since it may already be present in the commit history.\n3. Remove the secret from the current file content and replace it with a reference to AWS Secrets Manager or AWS Systems Manager Parameter Store.\n4. If the secret must be purged from history, rewrite the repository history (for example with `git filter-repo` or the BFG Repo-Cleaner) and force-push the cleaned branches.\n5. Enable a pre-commit or pre-receive hook (such as `detect-secrets` or AWS CodeGuru Secrets Detector) to prevent future commits containing secrets.",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Remove hardcoded credentials from all files committed to CodeCommit repositories. Store secrets in **AWS Secrets Manager** or **AWS Systems Manager Parameter Store** and reference them at runtime instead of committing them. Rotate any credential that has been committed, even if it is later removed, since it remains recoverable from commit history. Apply least-privilege IAM policies to repository access and consider using pre-commit hooks or AWS CodeGuru Secrets Detector to catch secrets before they are pushed.",
|
||||
"Url": "https://hub.prowler.com/check/codecommit_repository_no_secrets"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"secrets",
|
||||
"ci-cd"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [
|
||||
"awslambda_function_no_secrets_in_code",
|
||||
"codepipeline_project_repo_private",
|
||||
"codebuild_project_no_secrets_in_variables"
|
||||
],
|
||||
"Notes": "Severity is High by default. It may be treated as Critical if the detected secret is validated as active. Only the tip of the repository's default branch is scanned by default; full commit history is not walked, so secrets removed from HEAD but still present in earlier commits are not detected by this check. Files whose content cannot be retrieved through the CodeCommit API (for example, files larger than 6 MB) are not scanned and the repository is reported as MANUAL for review. False positives can be suppressed via the secrets_ignore_patterns audit config, and file paths can be excluded from scanning via the secrets_ignore_files audit config."
|
||||
}
|
||||
+163
@@ -0,0 +1,163 @@
|
||||
import fnmatch
|
||||
from collections import defaultdict
|
||||
from typing import List
|
||||
|
||||
from prowler.lib.check.models import Check, Check_Report_AWS
|
||||
from prowler.lib.utils.utils import (
|
||||
SecretsScanError,
|
||||
annotate_verified_secrets,
|
||||
detect_secrets_scan_batch,
|
||||
)
|
||||
from prowler.providers.aws.services.codecommit.codecommit_client import (
|
||||
codecommit_client,
|
||||
)
|
||||
|
||||
|
||||
class codecommit_repository_no_secrets(Check):
|
||||
"""Ensure CodeCommit repositories do not contain hardcoded secrets.
|
||||
|
||||
Scans every file tracked at the tip of each repository's default branch
|
||||
for embedded credentials such as API keys, tokens, or passwords. Only the
|
||||
default branch at its current commit is scanned by default; full commit
|
||||
history is not walked, since secrets that persist in history but have
|
||||
been removed from the tip of the branch are out of scope for this check.
|
||||
|
||||
- PASS: No secrets are detected in the files of the default branch.
|
||||
- FAIL: A secret is detected in one or more files of the default branch.
|
||||
- MANUAL: The secret scan could not be completed — either the scanner
|
||||
failed or some file content could not be retrieved (for example, files
|
||||
larger than 6 MB that the CodeCommit API refuses to return) — and
|
||||
manual review is required.
|
||||
"""
|
||||
|
||||
def execute(self) -> List[Check_Report_AWS]:
|
||||
"""Execute the CodeCommit repository secrets check.
|
||||
|
||||
Iterates over all discovered repositories, scans every file at the
|
||||
tip of the default branch for secrets in batched invocations, and
|
||||
reports any findings, including the repository, branch, commit, and
|
||||
file where the secret was found (never the secret value itself).
|
||||
|
||||
Returns:
|
||||
List[Check_Report_AWS]: A list of report objects with check results.
|
||||
"""
|
||||
findings = []
|
||||
secrets_ignore_patterns = codecommit_client.audit_config.get(
|
||||
"secrets_ignore_patterns", []
|
||||
)
|
||||
# Glob patterns of file paths inside the repository to skip when
|
||||
# scanning for secrets (e.g. "*.deps.json" or "package-lock.json").
|
||||
secrets_ignore_files = (
|
||||
codecommit_client.audit_config.get("secrets_ignore_files", []) or []
|
||||
)
|
||||
validate = codecommit_client.audit_config.get("secrets_validate", False)
|
||||
repositories = (
|
||||
list(codecommit_client.repositories.values())
|
||||
if codecommit_client.repositories
|
||||
else []
|
||||
)
|
||||
|
||||
# Collect every file tracked at the tip of each repository's default
|
||||
# branch and scan them in batched invocations instead of one
|
||||
# subprocess per file. Findings are keyed by (repository index, file
|
||||
# path) so they can be grouped back per repository. Files whose
|
||||
# content could not be retrieved are recorded so the repository is
|
||||
# reported as MANUAL instead of a false PASS.
|
||||
unscanned_files_by_repository = defaultdict(list)
|
||||
|
||||
def payloads():
|
||||
for repo_index, repository in enumerate(repositories):
|
||||
if (
|
||||
not repository.default_branch
|
||||
or not repository.default_branch_commit_id
|
||||
):
|
||||
continue
|
||||
for (
|
||||
file_path,
|
||||
file_content,
|
||||
) in codecommit_client.get_repository_files_content(repository):
|
||||
if any(
|
||||
fnmatch.fnmatch(file_path.lstrip("/"), pattern)
|
||||
for pattern in secrets_ignore_files
|
||||
):
|
||||
continue
|
||||
if file_content is None:
|
||||
unscanned_files_by_repository[repo_index].append(file_path)
|
||||
continue
|
||||
if not file_content:
|
||||
continue
|
||||
yield (repo_index, file_path), file_content.decode("latin-1")
|
||||
|
||||
scan_error = None
|
||||
try:
|
||||
batch_results = detect_secrets_scan_batch(
|
||||
payloads(), excluded_secrets=secrets_ignore_patterns, validate=validate
|
||||
)
|
||||
except SecretsScanError as error:
|
||||
batch_results = {}
|
||||
scan_error = error
|
||||
|
||||
findings_by_repository = defaultdict(dict)
|
||||
for (repo_index, file_path), file_findings in batch_results.items():
|
||||
findings_by_repository[repo_index][file_path] = file_findings
|
||||
|
||||
for repo_index, repository in enumerate(repositories):
|
||||
report = Check_Report_AWS(metadata=self.metadata(), resource=repository)
|
||||
report.status = "PASS"
|
||||
report.status_extended = f"CodeCommit repository {repository.name} does not have secrets in its default branch."
|
||||
|
||||
has_default_branch = bool(
|
||||
repository.default_branch and repository.default_branch_commit_id
|
||||
)
|
||||
|
||||
if not has_default_branch:
|
||||
report.status_extended = (
|
||||
f"CodeCommit repository {repository.name} has no default "
|
||||
f"branch, so there is no content to scan for secrets."
|
||||
)
|
||||
findings.append(report)
|
||||
continue
|
||||
|
||||
if scan_error:
|
||||
report.status = "MANUAL"
|
||||
report.status_extended = (
|
||||
f"Could not scan CodeCommit repository {repository.name} "
|
||||
f"default branch for secrets: {scan_error}; manual review is required."
|
||||
)
|
||||
findings.append(report)
|
||||
continue
|
||||
|
||||
files_with_secrets = findings_by_repository.get(repo_index)
|
||||
unscanned_files = unscanned_files_by_repository.get(repo_index)
|
||||
if files_with_secrets:
|
||||
all_secrets = []
|
||||
secrets_found = []
|
||||
for file_path, file_findings in files_with_secrets.items():
|
||||
all_secrets.extend(file_findings)
|
||||
secrets_found.extend(
|
||||
f"{file_path} on line {secret['line_number']} ({secret['type']})"
|
||||
for secret in file_findings
|
||||
)
|
||||
|
||||
secrets_string = ", ".join(secrets_found)
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
f"CodeCommit repository {repository.name} has "
|
||||
f"{'secrets' if len(secrets_found) > 1 else 'a secret'} in branch "
|
||||
f"{repository.default_branch} (commit {repository.default_branch_commit_id}) -> {secrets_string}."
|
||||
)
|
||||
annotate_verified_secrets(report, all_secrets)
|
||||
elif unscanned_files:
|
||||
report.status = "MANUAL"
|
||||
report.status_extended = (
|
||||
f"Could not retrieve the content of "
|
||||
f"{', '.join(unscanned_files)} in CodeCommit repository "
|
||||
f"{repository.name} default branch (for example, files "
|
||||
f"larger than 6 MB cannot be downloaded through the "
|
||||
f"CodeCommit API), so they were not scanned for secrets; "
|
||||
f"manual review is required."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
@@ -0,0 +1,246 @@
|
||||
from typing import Generator, Optional, Tuple
|
||||
|
||||
from botocore.exceptions import ClientError
|
||||
from pydantic import BaseModel
|
||||
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.providers.aws.lib.service.service import AWSService
|
||||
|
||||
|
||||
class CodeCommit(AWSService):
|
||||
"""AWS CodeCommit service class for managing repository resources.
|
||||
|
||||
This class handles interactions with AWS CodeCommit service, including
|
||||
listing repositories and retrieving their metadata (default branch and
|
||||
the commit it currently points to). The actual file content of a
|
||||
repository is fetched lazily, on demand, via `get_repository_files_content`,
|
||||
since walking a repository tree and downloading every blob can be an
|
||||
expensive operation.
|
||||
|
||||
Attributes:
|
||||
repositories: Dictionary mapping repository ARNs to Repository objects.
|
||||
"""
|
||||
|
||||
def __init__(self, provider):
|
||||
"""Initializes the CodeCommit service class.
|
||||
|
||||
Args:
|
||||
provider: AWS provider instance for making API calls.
|
||||
"""
|
||||
super().__init__(__class__.__name__, provider)
|
||||
self.repositories = {}
|
||||
self.__threading_call__(self._list_repositories)
|
||||
if self.repositories:
|
||||
self.__threading_call__(self._get_repository, self.repositories.values())
|
||||
self.__threading_call__(
|
||||
self._list_tags_for_resource, self.repositories.values()
|
||||
)
|
||||
|
||||
def _list_repositories(self, regional_client):
|
||||
"""Lists all CodeCommit repositories in the specified region.
|
||||
|
||||
Retrieves all repositories using pagination and creates Repository
|
||||
objects for each repository found.
|
||||
|
||||
Args:
|
||||
regional_client: AWS regional client for CodeCommit service.
|
||||
|
||||
Note:
|
||||
AWS API errors are caught and logged internally; this method
|
||||
does not raise them to the caller.
|
||||
"""
|
||||
logger.info("CodeCommit - Listing repositories...")
|
||||
try:
|
||||
if self.repositories is None:
|
||||
self.repositories = {}
|
||||
list_repositories_paginator = regional_client.get_paginator(
|
||||
"list_repositories"
|
||||
)
|
||||
for page in list_repositories_paginator.paginate():
|
||||
for repository in page["repositories"]:
|
||||
repository_arn = f"arn:{self.audited_partition}:codecommit:{regional_client.region}:{self.audited_account}:{repository['repositoryName']}"
|
||||
self.repositories[repository_arn] = Repository(
|
||||
repository_id=repository["repositoryId"],
|
||||
name=repository["repositoryName"],
|
||||
arn=repository_arn,
|
||||
region=regional_client.region,
|
||||
)
|
||||
except ClientError as error:
|
||||
if error.response["Error"]["Code"] in (
|
||||
"AccessDenied",
|
||||
"AccessDeniedException",
|
||||
):
|
||||
logger.error(
|
||||
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
if not self.repositories:
|
||||
self.repositories = None
|
||||
else:
|
||||
logger.error(
|
||||
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def _get_repository(self, repository):
|
||||
"""Retrieves repository metadata and the tip commit of the default branch.
|
||||
|
||||
Args:
|
||||
repository: Repository object to retrieve metadata for.
|
||||
|
||||
Note:
|
||||
AWS API errors are caught and logged internally; this method
|
||||
does not raise them to the caller.
|
||||
"""
|
||||
logger.info("CodeCommit - Getting repository metadata...")
|
||||
try:
|
||||
regional_client = self.regional_clients[repository.region]
|
||||
repository_metadata = regional_client.get_repository(
|
||||
repositoryName=repository.name
|
||||
)["repositoryMetadata"]
|
||||
repository.default_branch = repository_metadata.get("defaultBranch")
|
||||
|
||||
if repository.default_branch:
|
||||
try:
|
||||
branch_info = regional_client.get_branch(
|
||||
repositoryName=repository.name,
|
||||
branchName=repository.default_branch,
|
||||
)["branch"]
|
||||
repository.default_branch_commit_id = branch_info.get("commitId")
|
||||
except ClientError as error:
|
||||
logger.warning(
|
||||
f"{repository.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
except ClientError as error:
|
||||
if error.response["Error"]["Code"] in (
|
||||
"RepositoryDoesNotExistException",
|
||||
"EncryptionKeyAccessDeniedException",
|
||||
):
|
||||
logger.warning(
|
||||
f"{repository.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
else:
|
||||
logger.error(
|
||||
f"{repository.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{repository.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def _list_tags_for_resource(self, resource):
|
||||
"""Lists tags for a given resource.
|
||||
|
||||
Args:
|
||||
resource: Resource object to retrieve tags for.
|
||||
"""
|
||||
logger.info("CodeCommit - Listing Tags...")
|
||||
try:
|
||||
tags_response = self.regional_clients[
|
||||
resource.region
|
||||
].list_tags_for_resource(resourceArn=resource.arn)
|
||||
resource.tags = tags_response.get("tags", {})
|
||||
except ClientError as error:
|
||||
if error.response["Error"]["Code"] == "ResourceNotFoundException":
|
||||
logger.warning(
|
||||
f"{resource.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
else:
|
||||
logger.error(
|
||||
f"{resource.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{resource.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def get_repository_files_content(
|
||||
self, repository: "Repository"
|
||||
) -> Generator[Tuple[str, Optional[bytes]], None, None]:
|
||||
"""Walks the repository tree for the default branch and yields file content.
|
||||
|
||||
This performs the (potentially expensive) tree walk and blob download
|
||||
lazily so it is only paid for by checks that actually need file
|
||||
content, and only for repositories that have a default branch.
|
||||
|
||||
Args:
|
||||
repository: Repository object to fetch files for.
|
||||
|
||||
Yields:
|
||||
Tuple[str, Optional[bytes]]: The absolute file (or folder) path and
|
||||
its raw content. The content is None when it could not be
|
||||
retrieved (for example, files larger than 6 MB raise
|
||||
FileTooLargeException, or a folder listing fails), so callers can
|
||||
tell unscannable content apart from empty files.
|
||||
"""
|
||||
if not repository.default_branch or not repository.default_branch_commit_id:
|
||||
return
|
||||
|
||||
regional_client = self.regional_clients[repository.region]
|
||||
folders_to_process = ["/"]
|
||||
|
||||
while folders_to_process:
|
||||
folder_path = folders_to_process.pop()
|
||||
try:
|
||||
folder = regional_client.get_folder(
|
||||
repositoryName=repository.name,
|
||||
commitSpecifier=repository.default_branch_commit_id,
|
||||
folderPath=folder_path,
|
||||
)
|
||||
except ClientError as error:
|
||||
logger.error(
|
||||
f"{repository.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
yield folder_path, None
|
||||
continue
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{repository.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
yield folder_path, None
|
||||
continue
|
||||
|
||||
for sub_folder in folder.get("subFolders", []):
|
||||
folders_to_process.append(sub_folder["absolutePath"])
|
||||
|
||||
for file_info in folder.get("files", []):
|
||||
try:
|
||||
blob = regional_client.get_blob(
|
||||
repositoryName=repository.name,
|
||||
blobId=file_info["blobId"],
|
||||
)
|
||||
yield file_info["absolutePath"], blob.get("content")
|
||||
except ClientError as error:
|
||||
logger.error(
|
||||
f"{repository.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
yield file_info["absolutePath"], None
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{repository.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
yield file_info["absolutePath"], None
|
||||
|
||||
|
||||
class Repository(BaseModel):
|
||||
"""Model representing an AWS CodeCommit repository.
|
||||
|
||||
Attributes:
|
||||
repository_id: The repository ID.
|
||||
name: The name of the repository.
|
||||
arn: The ARN (Amazon Resource Name) of the repository.
|
||||
region: The AWS region where the repository exists.
|
||||
default_branch: The name of the repository's default branch, if any.
|
||||
default_branch_commit_id: The commit ID the default branch currently points to.
|
||||
tags: Optional dictionary of repository tags.
|
||||
"""
|
||||
|
||||
repository_id: str
|
||||
name: str
|
||||
arn: str
|
||||
region: str
|
||||
default_branch: Optional[str] = None
|
||||
default_branch_commit_id: Optional[str] = None
|
||||
tags: Optional[dict] = {}
|
||||
+519
@@ -0,0 +1,519 @@
|
||||
from unittest import mock
|
||||
|
||||
from tests.providers.aws.utils import (
|
||||
AWS_ACCOUNT_NUMBER,
|
||||
AWS_COMMERCIAL_PARTITION,
|
||||
AWS_REGION_EU_WEST_1,
|
||||
)
|
||||
|
||||
repository_name = "test-repo"
|
||||
repository_arn = f"arn:{AWS_COMMERCIAL_PARTITION}:codecommit:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:{repository_name}"
|
||||
|
||||
|
||||
class Test_codecommit_repository_no_secrets:
|
||||
def test_no_resources(self):
|
||||
"""No findings are returned when there are no repositories."""
|
||||
codecommit_client = mock.MagicMock()
|
||||
codecommit_client.repositories = {}
|
||||
codecommit_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_service.CodeCommit",
|
||||
codecommit_client,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets.codecommit_client",
|
||||
codecommit_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets import (
|
||||
codecommit_repository_no_secrets,
|
||||
)
|
||||
|
||||
check = codecommit_repository_no_secrets()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 0
|
||||
|
||||
def test_repository_no_default_branch(self):
|
||||
"""A repository without a default branch (e.g. empty repo) passes the check."""
|
||||
from prowler.providers.aws.services.codecommit.codecommit_service import (
|
||||
Repository,
|
||||
)
|
||||
|
||||
codecommit_client = mock.MagicMock()
|
||||
codecommit_client.repositories = {
|
||||
repository_arn: Repository(
|
||||
repository_id="repo-id-1",
|
||||
name=repository_name,
|
||||
arn=repository_arn,
|
||||
region=AWS_REGION_EU_WEST_1,
|
||||
)
|
||||
}
|
||||
codecommit_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_service.CodeCommit",
|
||||
codecommit_client,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets.codecommit_client",
|
||||
codecommit_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets import (
|
||||
codecommit_repository_no_secrets,
|
||||
)
|
||||
|
||||
check = codecommit_repository_no_secrets()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"CodeCommit repository {repository_name} has no default branch, so there is no content to scan for secrets."
|
||||
)
|
||||
assert result[0].resource_id == repository_name
|
||||
assert result[0].resource_arn == repository_arn
|
||||
assert result[0].region == AWS_REGION_EU_WEST_1
|
||||
codecommit_client.get_repository_files_content.assert_not_called()
|
||||
|
||||
def test_repository_no_secrets(self):
|
||||
"""A repository whose default branch files contain no secrets passes the check."""
|
||||
from prowler.providers.aws.services.codecommit.codecommit_service import (
|
||||
Repository,
|
||||
)
|
||||
|
||||
codecommit_client = mock.MagicMock()
|
||||
codecommit_client.repositories = {
|
||||
repository_arn: Repository(
|
||||
repository_id="repo-id-1",
|
||||
name=repository_name,
|
||||
arn=repository_arn,
|
||||
region=AWS_REGION_EU_WEST_1,
|
||||
default_branch="main",
|
||||
default_branch_commit_id="commit-1234",
|
||||
)
|
||||
}
|
||||
codecommit_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
codecommit_client.get_repository_files_content.return_value = iter(
|
||||
[
|
||||
("README.md", b"# Test repository\n"),
|
||||
("app.py", b"print('hello world')\n"),
|
||||
]
|
||||
)
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_service.CodeCommit",
|
||||
codecommit_client,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets.codecommit_client",
|
||||
codecommit_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets import (
|
||||
codecommit_repository_no_secrets,
|
||||
)
|
||||
|
||||
check = codecommit_repository_no_secrets()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"CodeCommit repository {repository_name} does not have secrets in its default branch."
|
||||
)
|
||||
assert result[0].resource_id == repository_name
|
||||
assert result[0].resource_arn == repository_arn
|
||||
assert result[0].region == AWS_REGION_EU_WEST_1
|
||||
|
||||
def test_repository_with_secrets(self):
|
||||
"""A repository with a hardcoded credential in a tracked file fails the check."""
|
||||
from prowler.providers.aws.services.codecommit.codecommit_service import (
|
||||
Repository,
|
||||
)
|
||||
|
||||
codecommit_client = mock.MagicMock()
|
||||
codecommit_client.repositories = {
|
||||
repository_arn: Repository(
|
||||
repository_id="repo-id-1",
|
||||
name=repository_name,
|
||||
arn=repository_arn,
|
||||
region=AWS_REGION_EU_WEST_1,
|
||||
default_branch="main",
|
||||
default_branch_commit_id="commit-1234",
|
||||
)
|
||||
}
|
||||
codecommit_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
codecommit_client.get_repository_files_content.return_value = iter(
|
||||
[
|
||||
("README.md", b"# Test repository\n"),
|
||||
(
|
||||
"src/secrets.py",
|
||||
# Realistic fake JWT that Kingfisher detects. A generic
|
||||
# placeholder value (e.g. a plain "test-password" string)
|
||||
# is suppressed by Kingfisher's low-confidence rules, so a
|
||||
# detectable provider-shaped secret is used instead (same
|
||||
# value used by codebuild's equivalent test).
|
||||
b'AUTH_TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.dozjgNryP4J3jVmNHl0w5N_XgL0n3I9PlFUP0THsR8U"\n',
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_service.CodeCommit",
|
||||
codecommit_client,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets.codecommit_client",
|
||||
codecommit_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets import (
|
||||
codecommit_repository_no_secrets,
|
||||
)
|
||||
|
||||
check = codecommit_repository_no_secrets()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "src/secrets.py" in result[0].status_extended
|
||||
assert "JSON Web Token" in result[0].status_extended
|
||||
assert "main" in result[0].status_extended
|
||||
assert "commit-1234" in result[0].status_extended
|
||||
assert result[0].resource_id == repository_name
|
||||
assert result[0].resource_arn == repository_arn
|
||||
assert result[0].region == AWS_REGION_EU_WEST_1
|
||||
|
||||
def test_repository_with_verified_secret_escalates_severity(self):
|
||||
"""A verified secret escalates the check severity to critical."""
|
||||
from prowler.lib.check.models import Severity
|
||||
from prowler.providers.aws.services.codecommit.codecommit_service import (
|
||||
Repository,
|
||||
)
|
||||
|
||||
codecommit_client = mock.MagicMock()
|
||||
codecommit_client.repositories = {
|
||||
repository_arn: Repository(
|
||||
repository_id="repo-id-1",
|
||||
name=repository_name,
|
||||
arn=repository_arn,
|
||||
region=AWS_REGION_EU_WEST_1,
|
||||
default_branch="main",
|
||||
default_branch_commit_id="commit-1234",
|
||||
)
|
||||
}
|
||||
codecommit_client.audit_config = {
|
||||
"secrets_ignore_patterns": [],
|
||||
"secrets_validate": True,
|
||||
}
|
||||
codecommit_client.get_repository_files_content.return_value = iter(
|
||||
[
|
||||
(
|
||||
"src/secrets.py",
|
||||
b'AWS_ACCESS_KEY_ID = "AKIAIOSFODNN7EXAMPLE"\n',
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_service.CodeCommit",
|
||||
codecommit_client,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets.codecommit_client",
|
||||
codecommit_client,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets.detect_secrets_scan_batch",
|
||||
return_value={
|
||||
(0, "src/secrets.py"): [
|
||||
{
|
||||
"line_number": 1,
|
||||
"type": "AWS Access Key",
|
||||
"filename": "src/secrets.py",
|
||||
"hashed_secret": "x",
|
||||
"is_verified": True,
|
||||
}
|
||||
]
|
||||
},
|
||||
) as mock_scan,
|
||||
):
|
||||
from prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets import (
|
||||
codecommit_repository_no_secrets,
|
||||
)
|
||||
|
||||
check = codecommit_repository_no_secrets()
|
||||
result = check.execute()
|
||||
|
||||
# The check must forward secrets_validate from the config to the scan.
|
||||
assert mock_scan.call_args.kwargs.get("validate") is True
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].check_metadata.Severity == Severity.critical
|
||||
assert "confirmed to be live" in result[0].status_extended
|
||||
assert result[0].resource_id == repository_name
|
||||
assert result[0].resource_arn == repository_arn
|
||||
assert result[0].region == AWS_REGION_EU_WEST_1
|
||||
|
||||
def test_repository_empty_file_content(self):
|
||||
"""A file with empty content is safely skipped and the check still passes."""
|
||||
from prowler.providers.aws.services.codecommit.codecommit_service import (
|
||||
Repository,
|
||||
)
|
||||
|
||||
codecommit_client = mock.MagicMock()
|
||||
codecommit_client.repositories = {
|
||||
repository_arn: Repository(
|
||||
repository_id="repo-id-1",
|
||||
name=repository_name,
|
||||
arn=repository_arn,
|
||||
region=AWS_REGION_EU_WEST_1,
|
||||
default_branch="main",
|
||||
default_branch_commit_id="commit-1234",
|
||||
)
|
||||
}
|
||||
codecommit_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
codecommit_client.get_repository_files_content.return_value = iter(
|
||||
[
|
||||
("empty.txt", b""),
|
||||
]
|
||||
)
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_service.CodeCommit",
|
||||
codecommit_client,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets.codecommit_client",
|
||||
codecommit_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets import (
|
||||
codecommit_repository_no_secrets,
|
||||
)
|
||||
|
||||
check = codecommit_repository_no_secrets()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
def test_repository_unretrievable_file_reports_manual(self):
|
||||
"""A file whose content could not be retrieved (e.g. larger than 6 MB)
|
||||
is reported as MANUAL instead of a false PASS."""
|
||||
from prowler.providers.aws.services.codecommit.codecommit_service import (
|
||||
Repository,
|
||||
)
|
||||
|
||||
codecommit_client = mock.MagicMock()
|
||||
codecommit_client.repositories = {
|
||||
repository_arn: Repository(
|
||||
repository_id="repo-id-1",
|
||||
name=repository_name,
|
||||
arn=repository_arn,
|
||||
region=AWS_REGION_EU_WEST_1,
|
||||
default_branch="main",
|
||||
default_branch_commit_id="commit-1234",
|
||||
)
|
||||
}
|
||||
codecommit_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
codecommit_client.get_repository_files_content.return_value = iter(
|
||||
[
|
||||
("app.py", b"print('hello world')\n"),
|
||||
("large-file.bin", None),
|
||||
]
|
||||
)
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_service.CodeCommit",
|
||||
codecommit_client,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets.codecommit_client",
|
||||
codecommit_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets import (
|
||||
codecommit_repository_no_secrets,
|
||||
)
|
||||
|
||||
check = codecommit_repository_no_secrets()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "MANUAL"
|
||||
assert "Could not retrieve the content of large-file.bin" in (
|
||||
result[0].status_extended
|
||||
)
|
||||
assert result[0].resource_id == repository_name
|
||||
assert result[0].resource_arn == repository_arn
|
||||
assert result[0].region == AWS_REGION_EU_WEST_1
|
||||
|
||||
def test_repository_with_secrets_and_unretrievable_file_still_fails(self):
|
||||
"""A detected secret takes precedence over unretrievable files: the
|
||||
repository is reported as FAIL, not MANUAL."""
|
||||
from prowler.providers.aws.services.codecommit.codecommit_service import (
|
||||
Repository,
|
||||
)
|
||||
|
||||
codecommit_client = mock.MagicMock()
|
||||
codecommit_client.repositories = {
|
||||
repository_arn: Repository(
|
||||
repository_id="repo-id-1",
|
||||
name=repository_name,
|
||||
arn=repository_arn,
|
||||
region=AWS_REGION_EU_WEST_1,
|
||||
default_branch="main",
|
||||
default_branch_commit_id="commit-1234",
|
||||
)
|
||||
}
|
||||
codecommit_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
codecommit_client.get_repository_files_content.return_value = iter(
|
||||
[
|
||||
("large-file.bin", None),
|
||||
(
|
||||
"src/secrets.py",
|
||||
b'AUTH_TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.dozjgNryP4J3jVmNHl0w5N_XgL0n3I9PlFUP0THsR8U"\n',
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_service.CodeCommit",
|
||||
codecommit_client,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets.codecommit_client",
|
||||
codecommit_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets import (
|
||||
codecommit_repository_no_secrets,
|
||||
)
|
||||
|
||||
check = codecommit_repository_no_secrets()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "src/secrets.py" in result[0].status_extended
|
||||
|
||||
def test_repository_secrets_ignore_files(self):
|
||||
"""A secret inside a file matched by secrets_ignore_files is skipped."""
|
||||
from prowler.providers.aws.services.codecommit.codecommit_service import (
|
||||
Repository,
|
||||
)
|
||||
|
||||
codecommit_client = mock.MagicMock()
|
||||
codecommit_client.repositories = {
|
||||
repository_arn: Repository(
|
||||
repository_id="repo-id-1",
|
||||
name=repository_name,
|
||||
arn=repository_arn,
|
||||
region=AWS_REGION_EU_WEST_1,
|
||||
default_branch="main",
|
||||
default_branch_commit_id="commit-1234",
|
||||
)
|
||||
}
|
||||
codecommit_client.audit_config = {
|
||||
"secrets_ignore_patterns": [],
|
||||
"secrets_ignore_files": ["package-lock.json"],
|
||||
}
|
||||
codecommit_client.get_repository_files_content.return_value = iter(
|
||||
[
|
||||
("README.md", b"# Test repository\n"),
|
||||
(
|
||||
"/package-lock.json",
|
||||
b'AUTH_TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.dozjgNryP4J3jVmNHl0w5N_XgL0n3I9PlFUP0THsR8U"\n',
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_service.CodeCommit",
|
||||
codecommit_client,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets.codecommit_client",
|
||||
codecommit_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets import (
|
||||
codecommit_repository_no_secrets,
|
||||
)
|
||||
|
||||
check = codecommit_repository_no_secrets()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"CodeCommit repository {repository_name} does not have secrets in its default branch."
|
||||
)
|
||||
|
||||
def test_scan_failure_reports_manual(self):
|
||||
"""A secret scanner failure is reported as MANUAL, never as a silent PASS."""
|
||||
from prowler.lib.utils.utils import SecretsScanError
|
||||
from prowler.providers.aws.services.codecommit.codecommit_service import (
|
||||
Repository,
|
||||
)
|
||||
|
||||
codecommit_client = mock.MagicMock()
|
||||
codecommit_client.repositories = {
|
||||
repository_arn: Repository(
|
||||
repository_id="repo-id-1",
|
||||
name=repository_name,
|
||||
arn=repository_arn,
|
||||
region=AWS_REGION_EU_WEST_1,
|
||||
default_branch="main",
|
||||
default_branch_commit_id="commit-1234",
|
||||
)
|
||||
}
|
||||
codecommit_client.audit_config = {"secrets_ignore_patterns": []}
|
||||
codecommit_client.get_repository_files_content.return_value = iter(
|
||||
[
|
||||
("app.py", b"print('hello world')\n"),
|
||||
]
|
||||
)
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_service.CodeCommit",
|
||||
codecommit_client,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets.codecommit_client",
|
||||
codecommit_client,
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets.detect_secrets_scan_batch",
|
||||
side_effect=SecretsScanError("Kingfisher exited with code 1"),
|
||||
),
|
||||
):
|
||||
from prowler.providers.aws.services.codecommit.codecommit_repository_no_secrets.codecommit_repository_no_secrets import (
|
||||
codecommit_repository_no_secrets,
|
||||
)
|
||||
|
||||
check = codecommit_repository_no_secrets()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "MANUAL"
|
||||
assert "Could not scan" in result[0].status_extended
|
||||
assert result[0].resource_id == repository_name
|
||||
assert result[0].resource_arn == repository_arn
|
||||
@@ -0,0 +1,572 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import botocore
|
||||
from botocore.exceptions import ClientError
|
||||
from moto import mock_aws
|
||||
|
||||
from prowler.providers.aws.services.codecommit.codecommit_service import (
|
||||
CodeCommit,
|
||||
Repository,
|
||||
)
|
||||
from tests.providers.aws.utils import (
|
||||
AWS_ACCOUNT_NUMBER,
|
||||
AWS_COMMERCIAL_PARTITION,
|
||||
AWS_REGION_EU_WEST_1,
|
||||
set_mocked_aws_provider,
|
||||
)
|
||||
|
||||
repository_name = "test-repo"
|
||||
repository_id = "repo-id-1234"
|
||||
repository_arn = f"arn:{AWS_COMMERCIAL_PARTITION}:codecommit:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:{repository_name}"
|
||||
default_branch = "main"
|
||||
commit_id = "commit-1234"
|
||||
|
||||
# Mocking API calls
|
||||
make_api_call = botocore.client.BaseClient._make_api_call
|
||||
|
||||
|
||||
def mock_make_api_call(self, operation_name, kwarg):
|
||||
if operation_name == "ListRepositories":
|
||||
return {
|
||||
"repositories": [
|
||||
{"repositoryName": repository_name, "repositoryId": repository_id}
|
||||
]
|
||||
}
|
||||
elif operation_name == "GetRepository":
|
||||
return {
|
||||
"repositoryMetadata": {
|
||||
"repositoryId": repository_id,
|
||||
"repositoryName": repository_name,
|
||||
"defaultBranch": default_branch,
|
||||
}
|
||||
}
|
||||
elif operation_name == "GetBranch":
|
||||
return {"branch": {"branchName": default_branch, "commitId": commit_id}}
|
||||
elif operation_name == "ListTagsForResource":
|
||||
return {"tags": {"Environment": "Test"}}
|
||||
elif operation_name == "GetFolder":
|
||||
if kwarg["folderPath"] == "/":
|
||||
return {
|
||||
"commitId": commit_id,
|
||||
"folderPath": "/",
|
||||
"subFolders": [{"absolutePath": "/src"}],
|
||||
"files": [{"absolutePath": "README.md", "blobId": "blob-readme"}],
|
||||
}
|
||||
elif kwarg["folderPath"] == "/src":
|
||||
return {
|
||||
"commitId": commit_id,
|
||||
"folderPath": "/src",
|
||||
"subFolders": [],
|
||||
"files": [
|
||||
{"absolutePath": "/src/secrets.py", "blobId": "blob-secrets"}
|
||||
],
|
||||
}
|
||||
elif operation_name == "GetBlob":
|
||||
if kwarg["blobId"] == "blob-readme":
|
||||
return {"content": b"# Test repository\n"}
|
||||
elif kwarg["blobId"] == "blob-secrets":
|
||||
return {"content": b'AWS_ACCESS_KEY_ID = "AKIAIOSFODNN7EXAMPLE"\n'}
|
||||
return make_api_call(self, operation_name, kwarg)
|
||||
|
||||
|
||||
# Mock generate_regional_clients()
|
||||
def mock_generate_regional_clients(provider, service):
|
||||
regional_client = provider._session.current_session.client(
|
||||
service, region_name=AWS_REGION_EU_WEST_1
|
||||
)
|
||||
regional_client.region = AWS_REGION_EU_WEST_1
|
||||
return {AWS_REGION_EU_WEST_1: regional_client}
|
||||
|
||||
|
||||
def mock_make_api_call_list_repositories_access_denied(self, operation_name, kwarg):
|
||||
if operation_name == "ListRepositories":
|
||||
# CodeCommit is a JSON-protocol API, so real IAM denials surface as
|
||||
# AccessDeniedException (the service also accepts plain AccessDenied).
|
||||
raise ClientError(
|
||||
{"Error": {"Code": "AccessDeniedException", "Message": "Access Denied"}},
|
||||
operation_name,
|
||||
)
|
||||
return make_api_call(self, operation_name, kwarg)
|
||||
|
||||
|
||||
def mock_make_api_call_list_repositories_client_error(self, operation_name, kwarg):
|
||||
if operation_name == "ListRepositories":
|
||||
raise ClientError(
|
||||
{"Error": {"Code": "ThrottlingException", "Message": "Rate exceeded"}},
|
||||
operation_name,
|
||||
)
|
||||
return make_api_call(self, operation_name, kwarg)
|
||||
|
||||
|
||||
def mock_make_api_call_repository_errors(self, operation_name, kwarg):
|
||||
if operation_name == "ListRepositories":
|
||||
return {
|
||||
"repositories": [
|
||||
{"repositoryName": repository_name, "repositoryId": repository_id},
|
||||
{"repositoryName": "repo-not-exist", "repositoryId": "repo-id-2"},
|
||||
{"repositoryName": "repo-other-error", "repositoryId": "repo-id-3"},
|
||||
{"repositoryName": "repo-branch-error", "repositoryId": "repo-id-4"},
|
||||
]
|
||||
}
|
||||
elif operation_name == "GetRepository":
|
||||
name = kwarg["repositoryName"]
|
||||
if name == "repo-not-exist":
|
||||
raise ClientError(
|
||||
{
|
||||
"Error": {
|
||||
"Code": "RepositoryDoesNotExistException",
|
||||
"Message": "Repository does not exist",
|
||||
}
|
||||
},
|
||||
operation_name,
|
||||
)
|
||||
if name == "repo-other-error":
|
||||
raise ClientError(
|
||||
{"Error": {"Code": "InternalServerException", "Message": "Boom"}},
|
||||
operation_name,
|
||||
)
|
||||
return {
|
||||
"repositoryMetadata": {
|
||||
"repositoryId": name,
|
||||
"repositoryName": name,
|
||||
"defaultBranch": default_branch,
|
||||
}
|
||||
}
|
||||
elif operation_name == "GetBranch":
|
||||
if kwarg["repositoryName"] == "repo-branch-error":
|
||||
raise ClientError(
|
||||
{"Error": {"Code": "InternalServerException", "Message": "Boom"}},
|
||||
operation_name,
|
||||
)
|
||||
return {"branch": {"branchName": default_branch, "commitId": commit_id}}
|
||||
elif operation_name == "ListTagsForResource":
|
||||
return {"tags": {}}
|
||||
return make_api_call(self, operation_name, kwarg)
|
||||
|
||||
|
||||
def mock_make_api_call_tags_errors(self, operation_name, kwarg):
|
||||
if operation_name == "ListRepositories":
|
||||
return {
|
||||
"repositories": [
|
||||
{"repositoryName": "tags-not-found", "repositoryId": "repo-id-1"},
|
||||
{"repositoryName": "tags-other-error", "repositoryId": "repo-id-2"},
|
||||
]
|
||||
}
|
||||
elif operation_name == "GetRepository":
|
||||
name = kwarg["repositoryName"]
|
||||
return {
|
||||
"repositoryMetadata": {
|
||||
"repositoryId": name,
|
||||
"repositoryName": name,
|
||||
"defaultBranch": default_branch,
|
||||
}
|
||||
}
|
||||
elif operation_name == "GetBranch":
|
||||
return {"branch": {"branchName": default_branch, "commitId": commit_id}}
|
||||
elif operation_name == "ListTagsForResource":
|
||||
if "tags-not-found" in kwarg["resourceArn"]:
|
||||
raise ClientError(
|
||||
{
|
||||
"Error": {
|
||||
"Code": "ResourceNotFoundException",
|
||||
"Message": "Not found",
|
||||
}
|
||||
},
|
||||
operation_name,
|
||||
)
|
||||
if "tags-other-error" in kwarg["resourceArn"]:
|
||||
raise ClientError(
|
||||
{"Error": {"Code": "InternalServerException", "Message": "Boom"}},
|
||||
operation_name,
|
||||
)
|
||||
return make_api_call(self, operation_name, kwarg)
|
||||
|
||||
|
||||
def mock_make_api_call_files_errors(self, operation_name, kwarg):
|
||||
if operation_name == "ListRepositories":
|
||||
return {
|
||||
"repositories": [
|
||||
{"repositoryName": repository_name, "repositoryId": repository_id}
|
||||
]
|
||||
}
|
||||
elif operation_name == "GetRepository":
|
||||
return {
|
||||
"repositoryMetadata": {
|
||||
"repositoryId": repository_id,
|
||||
"repositoryName": repository_name,
|
||||
"defaultBranch": default_branch,
|
||||
}
|
||||
}
|
||||
elif operation_name == "GetBranch":
|
||||
return {"branch": {"branchName": default_branch, "commitId": commit_id}}
|
||||
elif operation_name == "ListTagsForResource":
|
||||
return {"tags": {}}
|
||||
elif operation_name == "GetFolder":
|
||||
if kwarg["folderPath"] == "/":
|
||||
return {
|
||||
"commitId": commit_id,
|
||||
"folderPath": "/",
|
||||
"subFolders": [
|
||||
{"absolutePath": "/broken-folder"},
|
||||
{"absolutePath": "/broken-folder-2"},
|
||||
{"absolutePath": "/src"},
|
||||
],
|
||||
"files": [
|
||||
{"absolutePath": "README.md", "blobId": "blob-readme"},
|
||||
{"absolutePath": "bad-blob.txt", "blobId": "blob-bad"},
|
||||
{"absolutePath": "error-blob.txt", "blobId": "blob-error"},
|
||||
],
|
||||
}
|
||||
elif kwarg["folderPath"] == "/broken-folder":
|
||||
raise ClientError(
|
||||
{
|
||||
"Error": {
|
||||
"Code": "EncryptionKeyAccessDeniedException",
|
||||
"Message": "Access denied",
|
||||
}
|
||||
},
|
||||
operation_name,
|
||||
)
|
||||
elif kwarg["folderPath"] == "/broken-folder-2":
|
||||
raise Exception("Generic folder error")
|
||||
elif kwarg["folderPath"] == "/src":
|
||||
return {
|
||||
"commitId": commit_id,
|
||||
"folderPath": "/src",
|
||||
"subFolders": [],
|
||||
"files": [
|
||||
{"absolutePath": "/src/secrets.py", "blobId": "blob-secrets"}
|
||||
],
|
||||
}
|
||||
elif operation_name == "GetBlob":
|
||||
if kwarg["blobId"] == "blob-readme":
|
||||
return {"content": b"# Test repository\n"}
|
||||
elif kwarg["blobId"] == "blob-secrets":
|
||||
return {"content": b'AWS_ACCESS_KEY_ID = "AKIAIOSFODNN7EXAMPLE"\n'}
|
||||
elif kwarg["blobId"] == "blob-bad":
|
||||
raise ClientError(
|
||||
{
|
||||
"Error": {
|
||||
"Code": "BlobIdDoesNotExistException",
|
||||
"Message": "Blob not found",
|
||||
}
|
||||
},
|
||||
operation_name,
|
||||
)
|
||||
elif kwarg["blobId"] == "blob-error":
|
||||
raise Exception("Generic blob error")
|
||||
return make_api_call(self, operation_name, kwarg)
|
||||
|
||||
|
||||
class Test_CodeCommit_Service:
|
||||
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
|
||||
@patch(
|
||||
"prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
|
||||
new=mock_generate_regional_clients,
|
||||
)
|
||||
@mock_aws
|
||||
def test_codecommit_service(self):
|
||||
codecommit = CodeCommit(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
|
||||
|
||||
assert codecommit.session.__class__.__name__ == "Session"
|
||||
assert codecommit.service == "codecommit"
|
||||
|
||||
# Test repository properties
|
||||
assert len(codecommit.repositories) == 1
|
||||
assert isinstance(codecommit.repositories, dict)
|
||||
assert isinstance(codecommit.repositories[repository_arn], Repository)
|
||||
|
||||
repository = codecommit.repositories[repository_arn]
|
||||
assert repository.repository_id == repository_id
|
||||
assert repository.name == repository_name
|
||||
assert repository.arn == repository_arn
|
||||
assert repository.region == AWS_REGION_EU_WEST_1
|
||||
assert repository.default_branch == default_branch
|
||||
assert repository.default_branch_commit_id == commit_id
|
||||
|
||||
# Test tags
|
||||
assert repository.tags == {"Environment": "Test"}
|
||||
|
||||
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
|
||||
@patch(
|
||||
"prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
|
||||
new=mock_generate_regional_clients,
|
||||
)
|
||||
@mock_aws
|
||||
def test_get_repository_files_content(self):
|
||||
codecommit = CodeCommit(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
|
||||
repository = codecommit.repositories[repository_arn]
|
||||
|
||||
files = dict(codecommit.get_repository_files_content(repository))
|
||||
|
||||
assert files == {
|
||||
"README.md": b"# Test repository\n",
|
||||
"/src/secrets.py": b'AWS_ACCESS_KEY_ID = "AKIAIOSFODNN7EXAMPLE"\n',
|
||||
}
|
||||
|
||||
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
|
||||
@patch(
|
||||
"prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
|
||||
new=mock_generate_regional_clients,
|
||||
)
|
||||
@mock_aws
|
||||
def test_get_repository_files_content_no_default_branch(self):
|
||||
codecommit = CodeCommit(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
|
||||
repository = Repository(
|
||||
repository_id="empty-repo-id",
|
||||
name="empty-repo",
|
||||
arn=f"arn:{AWS_COMMERCIAL_PARTITION}:codecommit:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:empty-repo",
|
||||
region=AWS_REGION_EU_WEST_1,
|
||||
)
|
||||
|
||||
files = list(codecommit.get_repository_files_content(repository))
|
||||
|
||||
assert files == []
|
||||
|
||||
@patch(
|
||||
"botocore.client.BaseClient._make_api_call",
|
||||
new=mock_make_api_call_list_repositories_access_denied,
|
||||
)
|
||||
@patch(
|
||||
"prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
|
||||
new=mock_generate_regional_clients,
|
||||
)
|
||||
@mock_aws
|
||||
def test_list_repositories_access_denied_sets_none(self):
|
||||
"""An AccessDenied error with no repositories collected yet sets repositories to None."""
|
||||
codecommit = CodeCommit(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
|
||||
|
||||
assert codecommit.repositories is None
|
||||
|
||||
@patch(
|
||||
"botocore.client.BaseClient._make_api_call",
|
||||
new=mock_make_api_call_list_repositories_client_error,
|
||||
)
|
||||
@patch(
|
||||
"prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
|
||||
new=mock_generate_regional_clients,
|
||||
)
|
||||
@mock_aws
|
||||
def test_list_repositories_other_client_error(self):
|
||||
"""A non-AccessDenied ClientError is logged but does not set repositories to None."""
|
||||
codecommit = CodeCommit(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
|
||||
|
||||
assert codecommit.repositories == {}
|
||||
|
||||
def test_list_repositories_generic_exception(self):
|
||||
"""A non-ClientError exception while listing repositories is caught and logged."""
|
||||
codecommit = CodeCommit.__new__(CodeCommit)
|
||||
codecommit.repositories = {}
|
||||
codecommit.audited_partition = AWS_COMMERCIAL_PARTITION
|
||||
codecommit.audited_account = AWS_ACCOUNT_NUMBER
|
||||
|
||||
regional_client = MagicMock()
|
||||
regional_client.region = AWS_REGION_EU_WEST_1
|
||||
regional_client.get_paginator.side_effect = Exception("Generic error")
|
||||
|
||||
codecommit._list_repositories(regional_client)
|
||||
|
||||
assert codecommit.repositories == {}
|
||||
|
||||
def test_list_repositories_reinitializes_after_none(self):
|
||||
"""A region that succeeds after another region hit AccessDenied (leaving
|
||||
repositories as None) reinitializes it to a dict instead of crashing."""
|
||||
codecommit = CodeCommit.__new__(CodeCommit)
|
||||
codecommit.repositories = None
|
||||
codecommit.audited_partition = AWS_COMMERCIAL_PARTITION
|
||||
codecommit.audited_account = AWS_ACCOUNT_NUMBER
|
||||
|
||||
regional_client = MagicMock()
|
||||
regional_client.region = AWS_REGION_EU_WEST_1
|
||||
paginator = MagicMock()
|
||||
paginator.paginate.return_value = [
|
||||
{
|
||||
"repositories": [
|
||||
{"repositoryName": repository_name, "repositoryId": repository_id}
|
||||
]
|
||||
}
|
||||
]
|
||||
regional_client.get_paginator.return_value = paginator
|
||||
|
||||
codecommit._list_repositories(regional_client)
|
||||
|
||||
assert isinstance(codecommit.repositories, dict)
|
||||
assert codecommit.repositories[repository_arn].name == repository_name
|
||||
|
||||
def test_list_repositories_access_denied_keeps_existing_repositories(self):
|
||||
"""An AccessDenied error hit after repositories were already collected
|
||||
(e.g. in another region) does not wipe out the ones already found."""
|
||||
codecommit = CodeCommit.__new__(CodeCommit)
|
||||
codecommit.repositories = {}
|
||||
codecommit.audited_partition = AWS_COMMERCIAL_PARTITION
|
||||
codecommit.audited_account = AWS_ACCOUNT_NUMBER
|
||||
|
||||
healthy_client = MagicMock()
|
||||
healthy_client.region = AWS_REGION_EU_WEST_1
|
||||
healthy_paginator = MagicMock()
|
||||
healthy_paginator.paginate.return_value = [
|
||||
{
|
||||
"repositories": [
|
||||
{"repositoryName": repository_name, "repositoryId": repository_id}
|
||||
]
|
||||
}
|
||||
]
|
||||
healthy_client.get_paginator.return_value = healthy_paginator
|
||||
|
||||
codecommit._list_repositories(healthy_client)
|
||||
assert repository_arn in codecommit.repositories
|
||||
|
||||
failing_client = MagicMock()
|
||||
failing_client.region = "us-east-1"
|
||||
failing_paginator = MagicMock()
|
||||
failing_paginator.paginate.side_effect = ClientError(
|
||||
{"Error": {"Code": "AccessDenied", "Message": "Access Denied"}},
|
||||
"ListRepositories",
|
||||
)
|
||||
failing_client.get_paginator.return_value = failing_paginator
|
||||
|
||||
codecommit._list_repositories(failing_client)
|
||||
|
||||
assert codecommit.repositories is not None
|
||||
assert repository_arn in codecommit.repositories
|
||||
|
||||
def test_get_repository_no_default_branch(self):
|
||||
"""A repository with no default branch (e.g. a brand-new empty repo)
|
||||
is left without one, and GetBranch is never called."""
|
||||
codecommit = CodeCommit.__new__(CodeCommit)
|
||||
regional_client = MagicMock()
|
||||
regional_client.get_repository.return_value = {
|
||||
"repositoryMetadata": {
|
||||
"repositoryId": repository_id,
|
||||
"repositoryName": repository_name,
|
||||
}
|
||||
}
|
||||
codecommit.regional_clients = {AWS_REGION_EU_WEST_1: regional_client}
|
||||
|
||||
repository = Repository(
|
||||
repository_id=repository_id,
|
||||
name=repository_name,
|
||||
arn=repository_arn,
|
||||
region=AWS_REGION_EU_WEST_1,
|
||||
)
|
||||
|
||||
codecommit._get_repository(repository)
|
||||
|
||||
assert repository.default_branch is None
|
||||
assert repository.default_branch_commit_id is None
|
||||
regional_client.get_branch.assert_not_called()
|
||||
|
||||
@patch(
|
||||
"botocore.client.BaseClient._make_api_call",
|
||||
new=mock_make_api_call_repository_errors,
|
||||
)
|
||||
@patch(
|
||||
"prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
|
||||
new=mock_generate_regional_clients,
|
||||
)
|
||||
@mock_aws
|
||||
def test_get_repository_error_branches(self):
|
||||
"""GetRepository/GetBranch errors are caught per-repository without affecting others."""
|
||||
codecommit = CodeCommit(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
|
||||
|
||||
assert len(codecommit.repositories) == 4
|
||||
|
||||
def arn_for(name):
|
||||
return f"arn:{AWS_COMMERCIAL_PARTITION}:codecommit:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:{name}"
|
||||
|
||||
not_exist = codecommit.repositories[arn_for("repo-not-exist")]
|
||||
assert not_exist.default_branch is None
|
||||
assert not_exist.default_branch_commit_id is None
|
||||
|
||||
other_error = codecommit.repositories[arn_for("repo-other-error")]
|
||||
assert other_error.default_branch is None
|
||||
assert other_error.default_branch_commit_id is None
|
||||
|
||||
branch_error = codecommit.repositories[arn_for("repo-branch-error")]
|
||||
assert branch_error.default_branch == default_branch
|
||||
assert branch_error.default_branch_commit_id is None
|
||||
|
||||
healthy = codecommit.repositories[arn_for(repository_name)]
|
||||
assert healthy.default_branch == default_branch
|
||||
assert healthy.default_branch_commit_id == commit_id
|
||||
|
||||
def test_get_repository_generic_exception(self):
|
||||
"""A non-ClientError exception while getting repository metadata is caught and logged."""
|
||||
codecommit = CodeCommit.__new__(CodeCommit)
|
||||
codecommit.regional_clients = {AWS_REGION_EU_WEST_1: MagicMock()}
|
||||
codecommit.regional_clients[AWS_REGION_EU_WEST_1].get_repository.side_effect = (
|
||||
Exception("Generic error")
|
||||
)
|
||||
|
||||
repository = Repository(
|
||||
repository_id=repository_id,
|
||||
name=repository_name,
|
||||
arn=repository_arn,
|
||||
region=AWS_REGION_EU_WEST_1,
|
||||
)
|
||||
|
||||
codecommit._get_repository(repository)
|
||||
|
||||
assert repository.default_branch is None
|
||||
|
||||
@patch(
|
||||
"botocore.client.BaseClient._make_api_call",
|
||||
new=mock_make_api_call_tags_errors,
|
||||
)
|
||||
@patch(
|
||||
"prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
|
||||
new=mock_generate_regional_clients,
|
||||
)
|
||||
@mock_aws
|
||||
def test_list_tags_for_resource_error_branches(self):
|
||||
"""ListTagsForResource errors are caught per-repository and tags stay empty."""
|
||||
codecommit = CodeCommit(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
|
||||
|
||||
assert len(codecommit.repositories) == 2
|
||||
for repository in codecommit.repositories.values():
|
||||
assert repository.tags == {}
|
||||
|
||||
def test_list_tags_for_resource_generic_exception(self):
|
||||
"""A non-ClientError exception while listing tags is caught and logged."""
|
||||
codecommit = CodeCommit.__new__(CodeCommit)
|
||||
codecommit.regional_clients = {AWS_REGION_EU_WEST_1: MagicMock()}
|
||||
codecommit.regional_clients[
|
||||
AWS_REGION_EU_WEST_1
|
||||
].list_tags_for_resource.side_effect = Exception("Generic error")
|
||||
|
||||
repository = Repository(
|
||||
repository_id=repository_id,
|
||||
name=repository_name,
|
||||
arn=repository_arn,
|
||||
region=AWS_REGION_EU_WEST_1,
|
||||
)
|
||||
|
||||
codecommit._list_tags_for_resource(repository)
|
||||
|
||||
assert repository.tags == {}
|
||||
|
||||
@patch(
|
||||
"botocore.client.BaseClient._make_api_call",
|
||||
new=mock_make_api_call_files_errors,
|
||||
)
|
||||
@patch(
|
||||
"prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
|
||||
new=mock_generate_regional_clients,
|
||||
)
|
||||
@mock_aws
|
||||
def test_get_repository_files_content_handles_errors(self):
|
||||
"""Broken folders and blobs are yielded with None content so callers
|
||||
can report them as unscanned instead of silently passing."""
|
||||
codecommit = CodeCommit(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
|
||||
repository = codecommit.repositories[repository_arn]
|
||||
|
||||
files = dict(codecommit.get_repository_files_content(repository))
|
||||
|
||||
assert files == {
|
||||
"README.md": b"# Test repository\n",
|
||||
"/src/secrets.py": b'AWS_ACCESS_KEY_ID = "AKIAIOSFODNN7EXAMPLE"\n',
|
||||
"/broken-folder": None,
|
||||
"/broken-folder-2": None,
|
||||
"bad-blob.txt": None,
|
||||
"error-blob.txt": None,
|
||||
}
|
||||
Reference in New Issue
Block a user