mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 13:01:56 +00:00
feat(github): add repository and organization scoping support (#8329)
This commit is contained in:
Andoni Alonso
@@ -10,6 +10,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
|
||||
- CIS 4.0 for the Azure provider [(#7782)](https://github.com/prowler-cloud/prowler/pull/7782)
|
||||
- `vm_desired_sku_size` check for Azure provider [(#8191)](https://github.com/prowler-cloud/prowler/pull/8191)
|
||||
- `vm_scaleset_not_empty` check for Azure provider [(#8192)](https://github.com/prowler-cloud/prowler/pull/8192)
|
||||
- GitHub repository and organization scoping support with `--repository/respositories` and `--organization/organizations` flags [(#8329)](https://github.com/prowler-cloud/prowler/pull/8329)
|
||||
|
||||
### Changed
|
||||
- Handle some AWS errors as warnings instead of errors [(#8347)](https://github.com/prowler-cloud/prowler/pull/8347)
|
||||
|
||||
@@ -242,6 +242,8 @@ class Provider(ABC):
|
||||
github_app_id=arguments.github_app_id,
|
||||
mutelist_path=arguments.mutelist_file,
|
||||
config_path=arguments.config_file,
|
||||
repositories=arguments.repository,
|
||||
organizations=arguments.organization,
|
||||
)
|
||||
elif "iac" in provider_class_name.lower():
|
||||
provider_class(
|
||||
|
||||
@@ -82,6 +82,8 @@ class GithubProvider(Provider):
|
||||
_audit_config (dict): The audit configuration for the provider.
|
||||
_fixer_config (dict): The fixer configuration for the provider.
|
||||
_mutelist (Mutelist): The mutelist for the provider.
|
||||
_repositories (list): List of repository names to scan in 'owner/repo-name' format.
|
||||
_organizations (list): List of organization or user names to scan repositories for.
|
||||
audit_metadata (Audit_Metadata): The audit metadata for the provider.
|
||||
"""
|
||||
|
||||
@@ -91,6 +93,8 @@ class GithubProvider(Provider):
|
||||
_identity: GithubIdentityInfo
|
||||
_audit_config: dict
|
||||
_mutelist: Mutelist
|
||||
_repositories: list
|
||||
_organizations: list
|
||||
audit_metadata: Audit_Metadata
|
||||
|
||||
def __init__(
|
||||
@@ -107,6 +111,8 @@ class GithubProvider(Provider):
|
||||
fixer_config: dict = {},
|
||||
mutelist_path: str = None,
|
||||
mutelist_content: dict = None,
|
||||
repositories: list = None,
|
||||
organizations: list = None,
|
||||
):
|
||||
"""
|
||||
GitHub Provider constructor
|
||||
@@ -122,9 +128,15 @@ class GithubProvider(Provider):
|
||||
fixer_config (dict): Fixer configuration content.
|
||||
mutelist_path (str): Path to the mutelist file.
|
||||
mutelist_content (dict): Mutelist content.
|
||||
repositories (list): List of repository names to scan in 'owner/repo-name' format.
|
||||
organizations (list): List of organization or user names to scan repositories for.
|
||||
"""
|
||||
logger.info("Instantiating GitHub Provider...")
|
||||
|
||||
# Set repositories and organizations for scoping
|
||||
self._repositories = repositories or []
|
||||
self._organizations = organizations or []
|
||||
|
||||
self._session = GithubProvider.setup_session(
|
||||
personal_access_token,
|
||||
oauth_app_token,
|
||||
@@ -213,6 +225,20 @@ class GithubProvider(Provider):
|
||||
"""
|
||||
return self._mutelist
|
||||
|
||||
@property
|
||||
def repositories(self) -> list:
|
||||
"""
|
||||
repositories method returns the provider's repository list for scoping.
|
||||
"""
|
||||
return self._repositories
|
||||
|
||||
@property
|
||||
def organizations(self) -> list:
|
||||
"""
|
||||
organizations method returns the provider's organization list for scoping.
|
||||
"""
|
||||
return self._organizations
|
||||
|
||||
@staticmethod
|
||||
def setup_session(
|
||||
personal_access_token: str = None,
|
||||
@@ -295,7 +321,7 @@ class GithubProvider(Provider):
|
||||
|
||||
except Exception as error:
|
||||
logger.critical(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
raise GithubSetUpSessionError(
|
||||
original_exception=error,
|
||||
@@ -344,7 +370,7 @@ class GithubProvider(Provider):
|
||||
|
||||
except Exception as error:
|
||||
logger.critical(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
raise GithubSetUpIdentityError(
|
||||
original_exception=error,
|
||||
|
||||
@@ -37,3 +37,21 @@ def init_parser(self):
|
||||
default=None,
|
||||
metavar="GITHUB_APP_KEY",
|
||||
)
|
||||
|
||||
github_scoping_subparser = github_parser.add_argument_group("Scan Scoping")
|
||||
github_scoping_subparser.add_argument(
|
||||
"--repository",
|
||||
"--repositories",
|
||||
nargs="*",
|
||||
help="Repository name(s) to scan in 'owner/repo-name' format",
|
||||
default=None,
|
||||
metavar="REPOSITORY",
|
||||
)
|
||||
github_scoping_subparser.add_argument(
|
||||
"--organization",
|
||||
"--organizations",
|
||||
nargs="*",
|
||||
help="Organization or user name(s) to scan repositories for",
|
||||
default=None,
|
||||
metavar="ORGANIZATION",
|
||||
)
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import github
|
||||
from github import Auth, Github, GithubIntegration
|
||||
from github.GithubRetry import GithubRetry
|
||||
|
||||
@@ -11,6 +12,7 @@ class GithubService:
|
||||
service: str,
|
||||
provider: GithubProvider,
|
||||
):
|
||||
self.provider = provider
|
||||
self.clients = self.__set_clients__(
|
||||
provider.session,
|
||||
)
|
||||
@@ -41,3 +43,53 @@ class GithubService:
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
return clients
|
||||
|
||||
def _handle_github_api_error(
|
||||
self, error, context: str, item_name: str, reraise_rate_limit: bool = False
|
||||
):
|
||||
"""Centralized GitHub API error handling"""
|
||||
if isinstance(error, github.RateLimitExceededException):
|
||||
logger.error(f"Rate limit exceeded while {context} '{item_name}': {error}")
|
||||
if reraise_rate_limit:
|
||||
raise
|
||||
elif isinstance(error, github.GithubException):
|
||||
if "404" in str(error):
|
||||
logger.error(f"'{item_name}' not found or not accessible")
|
||||
elif "403" in str(error):
|
||||
logger.error(
|
||||
f"Access denied to '{item_name}' - insufficient permissions"
|
||||
)
|
||||
else:
|
||||
logger.error(f"GitHub API error for '{item_name}': {error}")
|
||||
else:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def _get_repositories_from_owner(self, client, name: str):
|
||||
"""Get repositories from organization or user entity"""
|
||||
try:
|
||||
org = client.get_organization(name)
|
||||
return org.get_repos(), "organization"
|
||||
except github.GithubException as error:
|
||||
if "404" in str(error):
|
||||
logger.info(f"'{name}' not found as organization, trying as user...")
|
||||
try:
|
||||
user = client.get_user(name)
|
||||
return user.get_repos(), "user"
|
||||
except github.GithubException as user_error:
|
||||
self._handle_github_api_error(
|
||||
user_error, "accessing", f"{name} as user"
|
||||
)
|
||||
return [], "none"
|
||||
except Exception as user_error:
|
||||
self._handle_github_api_error(
|
||||
user_error, "accessing", f"{name} as user"
|
||||
)
|
||||
return [], "none"
|
||||
else:
|
||||
self._handle_github_api_error(error, "accessing organization", name)
|
||||
return [], "none"
|
||||
except Exception as error:
|
||||
self._handle_github_api_error(error, "accessing organization", name)
|
||||
return [], "none"
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
from typing import Optional
|
||||
|
||||
import github
|
||||
from pydantic.v1 import BaseModel
|
||||
|
||||
from prowler.lib.logger import logger
|
||||
@@ -12,29 +13,135 @@ class Organization(GithubService):
|
||||
self.organizations = self._list_organizations()
|
||||
|
||||
def _list_organizations(self):
|
||||
"""
|
||||
List organizations based on provider scoping configuration.
|
||||
|
||||
Scoping behavior:
|
||||
- No scoping: Returns all organizations for authenticated user
|
||||
- Organization scoping: Returns only specified organizations
|
||||
Example: --organization org1 org2
|
||||
- Repository + Organization scoping: Returns specified organizations + repository owners
|
||||
Example: --repository owner1/repo1 --organization org2
|
||||
- Repository only: Returns empty (no organization checks)
|
||||
Example: --repository owner1/repo1
|
||||
|
||||
Returns:
|
||||
dict: Dictionary of organization ID to Org objects
|
||||
|
||||
Raises:
|
||||
github.GithubException: When GitHub API access fails
|
||||
github.RateLimitExceededException: When API rate limits are exceeded
|
||||
"""
|
||||
logger.info("Organization - Listing Organizations...")
|
||||
organizations = {}
|
||||
org_names_to_check = set()
|
||||
|
||||
try:
|
||||
for client in self.clients:
|
||||
for org in client.get_user().get_orgs():
|
||||
try:
|
||||
require_mfa = org.two_factor_requirement_enabled
|
||||
except Exception as error:
|
||||
require_mfa = None
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
organizations[org.id] = Org(
|
||||
id=org.id,
|
||||
name=org.login,
|
||||
mfa_required=require_mfa,
|
||||
)
|
||||
if self.provider.organizations:
|
||||
org_names_to_check.update(self.provider.organizations)
|
||||
|
||||
# If repositories are specified without organizations, don't perform organization checks
|
||||
# Only add repository owners to organization checks if organizations are also specified
|
||||
if self.provider.repositories and self.provider.organizations:
|
||||
for repo_name in self.provider.repositories:
|
||||
if "/" in repo_name:
|
||||
owner_name = repo_name.split("/")[0]
|
||||
org_names_to_check.add(owner_name)
|
||||
logger.info(
|
||||
f"Adding owner '{owner_name}' from repository '{repo_name}' to organization check list"
|
||||
)
|
||||
|
||||
# If specific organizations/owners are specified, check them directly
|
||||
if org_names_to_check:
|
||||
for org_name in org_names_to_check:
|
||||
try:
|
||||
try:
|
||||
org = client.get_organization(org_name)
|
||||
self._process_organization(org, organizations)
|
||||
except github.GithubException as org_error:
|
||||
# If organization fails, try as a user (personal account)
|
||||
if "404" in str(org_error):
|
||||
logger.info(
|
||||
f"'{org_name}' not found as organization, trying as user..."
|
||||
)
|
||||
try:
|
||||
user = client.get_user(org_name)
|
||||
# Create a pseudo-organization for the user
|
||||
organizations[user.id] = Org(
|
||||
id=user.id,
|
||||
name=user.login,
|
||||
mfa_required=None, # Users don't have MFA requirements like orgs
|
||||
)
|
||||
logger.info(
|
||||
f"Added user '{user.login}' as organization for checks"
|
||||
)
|
||||
except github.GithubException as user_error:
|
||||
if "404" in str(user_error):
|
||||
logger.warning(
|
||||
f"'{org_name}' not found as organization or user"
|
||||
)
|
||||
elif "403" in str(user_error):
|
||||
logger.warning(
|
||||
f"Access denied to '{org_name}' - insufficient permissions"
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
f"GitHub API error accessing '{org_name}' as user: {user_error}"
|
||||
)
|
||||
except Exception as user_error:
|
||||
logger.error(
|
||||
f"{user_error.__class__.__name__}[{user_error.__traceback__.tb_lineno}]: {user_error}"
|
||||
)
|
||||
elif "403" in str(org_error):
|
||||
logger.warning(
|
||||
f"Access denied to organization '{org_name}' - insufficient permissions"
|
||||
)
|
||||
else:
|
||||
logger.error(
|
||||
f"GitHub API error accessing organization '{org_name}': {org_error}"
|
||||
)
|
||||
except github.RateLimitExceededException as error:
|
||||
logger.error(
|
||||
f"Rate limit exceeded while processing organization '{org_name}': {error}"
|
||||
)
|
||||
raise # Re-raise rate limit errors as they need special handling
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
elif not self.provider.repositories:
|
||||
# Default behavior: get all organizations the user is a member of
|
||||
# Only when no repositories are specified
|
||||
for org in client.get_user().get_orgs():
|
||||
self._process_organization(org, organizations)
|
||||
|
||||
except github.RateLimitExceededException as error:
|
||||
logger.error(f"GitHub API rate limit exceeded: {error}")
|
||||
raise # Re-raise rate limit errors as they need special handling
|
||||
except github.GithubException as error:
|
||||
logger.error(f"GitHub API error while listing organizations: {error}")
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
return organizations
|
||||
|
||||
def _process_organization(self, org, organizations):
|
||||
"""Process a single organization and extract its information."""
|
||||
try:
|
||||
require_mfa = org.two_factor_requirement_enabled
|
||||
except Exception as error:
|
||||
require_mfa = None
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
organizations[org.id] = Org(
|
||||
id=org.id,
|
||||
name=org.login,
|
||||
mfa_required=require_mfa,
|
||||
)
|
||||
|
||||
|
||||
class Org(BaseModel):
|
||||
"""Model for Github Organization"""
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
import github
|
||||
from pydantic.v1 import BaseModel
|
||||
|
||||
from prowler.lib.logger import logger
|
||||
@@ -25,175 +26,250 @@ class Repository(GithubService):
|
||||
)
|
||||
return None
|
||||
|
||||
def _validate_repository_format(self, repo_name: str) -> bool:
|
||||
"""
|
||||
Validate repository name format.
|
||||
|
||||
Args:
|
||||
repo_name: Repository name to validate
|
||||
|
||||
Returns:
|
||||
bool: True if format is valid, False otherwise
|
||||
"""
|
||||
if not repo_name or "/" not in repo_name:
|
||||
return False
|
||||
|
||||
parts = repo_name.split("/")
|
||||
if len(parts) != 2:
|
||||
return False
|
||||
|
||||
owner, repo = parts
|
||||
# Ensure both owner and repo names are non-empty
|
||||
if not owner.strip() or not repo.strip():
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def _list_repositories(self):
|
||||
"""
|
||||
List repositories based on provider scoping configuration.
|
||||
|
||||
Scoping behavior:
|
||||
- No scoping: Returns all accessible repositories for authenticated user
|
||||
- Repository scoping: Returns only specified repositories
|
||||
Example: --repository owner1/repo1 owner2/repo2
|
||||
- Organization scoping: Returns all repositories from specified organizations
|
||||
Example: --organization org1 org2
|
||||
- Combined scoping: Returns specified repositories + all repos from organizations
|
||||
Example: --repository owner1/repo1 --organization org2
|
||||
|
||||
Returns:
|
||||
dict: Dictionary of repository ID to Repo objects
|
||||
|
||||
Raises:
|
||||
github.GithubException: When GitHub API access fails
|
||||
github.RateLimitExceededException: When API rate limits are exceeded
|
||||
"""
|
||||
logger.info("Repository - Listing Repositories...")
|
||||
repos = {}
|
||||
try:
|
||||
for client in self.clients:
|
||||
for repo in client.get_user().get_repos():
|
||||
default_branch = repo.default_branch
|
||||
securitymd_exists = self._file_exists(repo, "SECURITY.md")
|
||||
# CODEOWNERS file can be in .github/, root, or docs/
|
||||
# https://docs.github.com/en/repositories/managing-your-repositorys-settings-and-features/customizing-your-repository/about-code-owners#codeowners-file-location
|
||||
codeowners_paths = [
|
||||
".github/CODEOWNERS",
|
||||
"CODEOWNERS",
|
||||
"docs/CODEOWNERS",
|
||||
]
|
||||
codeowners_files = [
|
||||
self._file_exists(repo, path) for path in codeowners_paths
|
||||
]
|
||||
if True in codeowners_files:
|
||||
codeowners_exists = True
|
||||
elif all(file is None for file in codeowners_files):
|
||||
codeowners_exists = None
|
||||
else:
|
||||
codeowners_exists = False
|
||||
delete_branch_on_merge = (
|
||||
repo.delete_branch_on_merge
|
||||
if repo.delete_branch_on_merge is not None
|
||||
else False
|
||||
)
|
||||
|
||||
require_pr = False
|
||||
approval_cnt = 0
|
||||
branch_protection = False
|
||||
required_linear_history = False
|
||||
allow_force_pushes = True
|
||||
branch_deletion = True
|
||||
require_code_owner_reviews = False
|
||||
require_signed_commits = False
|
||||
status_checks = False
|
||||
enforce_admins = False
|
||||
conversation_resolution = False
|
||||
try:
|
||||
branch = repo.get_branch(default_branch)
|
||||
if branch.protected:
|
||||
protection = branch.get_protection()
|
||||
if protection:
|
||||
require_pr = (
|
||||
protection.required_pull_request_reviews is not None
|
||||
)
|
||||
approval_cnt = (
|
||||
protection.required_pull_request_reviews.required_approving_review_count
|
||||
if require_pr
|
||||
else 0
|
||||
)
|
||||
required_linear_history = (
|
||||
protection.required_linear_history
|
||||
)
|
||||
allow_force_pushes = protection.allow_force_pushes
|
||||
branch_deletion = protection.allow_deletions
|
||||
status_checks = (
|
||||
protection.required_status_checks is not None
|
||||
)
|
||||
enforce_admins = protection.enforce_admins
|
||||
conversation_resolution = (
|
||||
protection.required_conversation_resolution
|
||||
)
|
||||
branch_protection = True
|
||||
require_code_owner_reviews = (
|
||||
protection.required_pull_request_reviews.require_code_owner_reviews
|
||||
if require_pr
|
||||
else False
|
||||
)
|
||||
require_signed_commits = (
|
||||
branch.get_required_signatures()
|
||||
)
|
||||
except Exception as error:
|
||||
# If the branch is not found, it is not protected
|
||||
if "404" in str(error):
|
||||
logger.warning(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
# Any other error, we cannot know if the branch is protected or not
|
||||
else:
|
||||
require_pr = None
|
||||
approval_cnt = None
|
||||
branch_protection = None
|
||||
required_linear_history = None
|
||||
allow_force_pushes = None
|
||||
branch_deletion = None
|
||||
require_code_owner_reviews = None
|
||||
require_signed_commits = None
|
||||
status_checks = None
|
||||
enforce_admins = None
|
||||
conversation_resolution = None
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
secret_scanning_enabled = False
|
||||
dependabot_alerts_enabled = False
|
||||
try:
|
||||
if (
|
||||
repo.security_and_analysis
|
||||
and repo.security_and_analysis.secret_scanning
|
||||
):
|
||||
secret_scanning_enabled = (
|
||||
repo.security_and_analysis.secret_scanning.status
|
||||
== "enabled"
|
||||
)
|
||||
try:
|
||||
# Use get_dependabot_alerts to check if Dependabot alerts are enabled
|
||||
repo.get_dependabot_alerts().totalCount
|
||||
# If the call succeeds, Dependabot is enabled (even if no alerts)
|
||||
dependabot_alerts_enabled = True
|
||||
except Exception as error:
|
||||
error_str = str(error)
|
||||
if (
|
||||
"403" in error_str
|
||||
and "Dependabot alerts are disabled for this repository."
|
||||
in error_str
|
||||
):
|
||||
dependabot_alerts_enabled = False
|
||||
else:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
dependabot_alerts_enabled = None
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
if self.provider.repositories or self.provider.organizations:
|
||||
if self.provider.repositories:
|
||||
logger.info(
|
||||
f"Filtering for specific repositories: {self.provider.repositories}"
|
||||
)
|
||||
secret_scanning_enabled = None
|
||||
dependabot_alerts_enabled = None
|
||||
repos[repo.id] = Repo(
|
||||
id=repo.id,
|
||||
name=repo.name,
|
||||
owner=repo.owner.login,
|
||||
full_name=repo.full_name,
|
||||
default_branch=Branch(
|
||||
name=default_branch,
|
||||
protected=branch_protection,
|
||||
default_branch=True,
|
||||
require_pull_request=require_pr,
|
||||
approval_count=approval_cnt,
|
||||
required_linear_history=required_linear_history,
|
||||
allow_force_pushes=allow_force_pushes,
|
||||
branch_deletion=branch_deletion,
|
||||
status_checks=status_checks,
|
||||
enforce_admins=enforce_admins,
|
||||
conversation_resolution=conversation_resolution,
|
||||
require_code_owner_reviews=require_code_owner_reviews,
|
||||
require_signed_commits=require_signed_commits,
|
||||
),
|
||||
private=repo.private,
|
||||
archived=repo.archived,
|
||||
pushed_at=repo.pushed_at,
|
||||
securitymd=securitymd_exists,
|
||||
codeowners_exists=codeowners_exists,
|
||||
secret_scanning_enabled=secret_scanning_enabled,
|
||||
dependabot_alerts_enabled=dependabot_alerts_enabled,
|
||||
delete_branch_on_merge=delete_branch_on_merge,
|
||||
)
|
||||
for repo_name in self.provider.repositories:
|
||||
if not self._validate_repository_format(repo_name):
|
||||
logger.warning(
|
||||
f"Repository name '{repo_name}' should be in 'owner/repo-name' format. Skipping."
|
||||
)
|
||||
continue
|
||||
try:
|
||||
repo = client.get_repo(repo_name)
|
||||
self._process_repository(repo, repos)
|
||||
except Exception as error:
|
||||
self._handle_github_api_error(
|
||||
error, "accessing repository", repo_name
|
||||
)
|
||||
|
||||
if self.provider.organizations:
|
||||
logger.info(
|
||||
f"Filtering for repositories in organizations: {self.provider.organizations}"
|
||||
)
|
||||
for org_name in self.provider.organizations:
|
||||
try:
|
||||
repos_list, _ = self._get_repositories_from_owner(
|
||||
client, org_name
|
||||
)
|
||||
for repo in repos_list:
|
||||
self._process_repository(repo, repos)
|
||||
except Exception as error:
|
||||
self._handle_github_api_error(
|
||||
error, "processing organization", org_name
|
||||
)
|
||||
else:
|
||||
for repo in client.get_user().get_repos():
|
||||
self._process_repository(repo, repos)
|
||||
except github.RateLimitExceededException as error:
|
||||
logger.error(f"GitHub API rate limit exceeded: {error}")
|
||||
raise # Re-raise rate limit errors as they need special handling
|
||||
except github.GithubException as error:
|
||||
logger.error(f"GitHub API error while listing repositories: {error}")
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
return repos
|
||||
|
||||
def _process_repository(self, repo, repos):
|
||||
"""Process a single repository and extract all its information."""
|
||||
default_branch = repo.default_branch
|
||||
securitymd_exists = self._file_exists(repo, "SECURITY.md")
|
||||
# CODEOWNERS file can be in .github/, root, or docs/
|
||||
# https://docs.github.com/en/repositories/managing-your-repositorys-settings-and-features/customizing-your-repository/about-code-owners#codeowners-file-location
|
||||
codeowners_paths = [
|
||||
".github/CODEOWNERS",
|
||||
"CODEOWNERS",
|
||||
"docs/CODEOWNERS",
|
||||
]
|
||||
codeowners_files = [self._file_exists(repo, path) for path in codeowners_paths]
|
||||
if True in codeowners_files:
|
||||
codeowners_exists = True
|
||||
elif all(file is None for file in codeowners_files):
|
||||
codeowners_exists = None
|
||||
else:
|
||||
codeowners_exists = False
|
||||
delete_branch_on_merge = (
|
||||
repo.delete_branch_on_merge
|
||||
if repo.delete_branch_on_merge is not None
|
||||
else False
|
||||
)
|
||||
|
||||
require_pr = False
|
||||
approval_cnt = 0
|
||||
branch_protection = False
|
||||
required_linear_history = False
|
||||
allow_force_pushes = True
|
||||
branch_deletion = True
|
||||
require_code_owner_reviews = False
|
||||
require_signed_commits = False
|
||||
status_checks = False
|
||||
enforce_admins = False
|
||||
conversation_resolution = False
|
||||
try:
|
||||
branch = repo.get_branch(default_branch)
|
||||
if branch.protected:
|
||||
protection = branch.get_protection()
|
||||
if protection:
|
||||
require_pr = protection.required_pull_request_reviews is not None
|
||||
approval_cnt = (
|
||||
protection.required_pull_request_reviews.required_approving_review_count
|
||||
if require_pr
|
||||
else 0
|
||||
)
|
||||
required_linear_history = protection.required_linear_history
|
||||
allow_force_pushes = protection.allow_force_pushes
|
||||
branch_deletion = protection.allow_deletions
|
||||
status_checks = protection.required_status_checks is not None
|
||||
enforce_admins = protection.enforce_admins
|
||||
conversation_resolution = (
|
||||
protection.required_conversation_resolution
|
||||
)
|
||||
branch_protection = True
|
||||
require_code_owner_reviews = (
|
||||
protection.required_pull_request_reviews.require_code_owner_reviews
|
||||
if require_pr
|
||||
else False
|
||||
)
|
||||
require_signed_commits = branch.get_required_signatures()
|
||||
except Exception as error:
|
||||
# If the branch is not found, it is not protected
|
||||
if "404" in str(error):
|
||||
logger.warning(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
# Any other error, we cannot know if the branch is protected or not
|
||||
else:
|
||||
require_pr = None
|
||||
approval_cnt = None
|
||||
branch_protection = None
|
||||
required_linear_history = None
|
||||
allow_force_pushes = None
|
||||
branch_deletion = None
|
||||
require_code_owner_reviews = None
|
||||
require_signed_commits = None
|
||||
status_checks = None
|
||||
enforce_admins = None
|
||||
conversation_resolution = None
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
secret_scanning_enabled = False
|
||||
dependabot_alerts_enabled = False
|
||||
try:
|
||||
if (
|
||||
repo.security_and_analysis
|
||||
and repo.security_and_analysis.secret_scanning
|
||||
):
|
||||
secret_scanning_enabled = (
|
||||
repo.security_and_analysis.secret_scanning.status == "enabled"
|
||||
)
|
||||
try:
|
||||
# Use get_dependabot_alerts to check if Dependabot alerts are enabled
|
||||
repo.get_dependabot_alerts().totalCount
|
||||
# If the call succeeds, Dependabot is enabled (even if no alerts)
|
||||
dependabot_alerts_enabled = True
|
||||
except Exception as error:
|
||||
error_str = str(error)
|
||||
if (
|
||||
"403" in error_str
|
||||
and "Dependabot alerts are disabled for this repository."
|
||||
in error_str
|
||||
):
|
||||
dependabot_alerts_enabled = False
|
||||
else:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
dependabot_alerts_enabled = None
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
secret_scanning_enabled = None
|
||||
dependabot_alerts_enabled = None
|
||||
repos[repo.id] = Repo(
|
||||
id=repo.id,
|
||||
name=repo.name,
|
||||
owner=repo.owner.login,
|
||||
full_name=repo.full_name,
|
||||
default_branch=Branch(
|
||||
name=default_branch,
|
||||
protected=branch_protection,
|
||||
default_branch=True,
|
||||
require_pull_request=require_pr,
|
||||
approval_count=approval_cnt,
|
||||
required_linear_history=required_linear_history,
|
||||
allow_force_pushes=allow_force_pushes,
|
||||
branch_deletion=branch_deletion,
|
||||
status_checks=status_checks,
|
||||
enforce_admins=enforce_admins,
|
||||
conversation_resolution=conversation_resolution,
|
||||
require_code_owner_reviews=require_code_owner_reviews,
|
||||
require_signed_commits=require_signed_commits,
|
||||
),
|
||||
private=repo.private,
|
||||
archived=repo.archived,
|
||||
pushed_at=repo.pushed_at,
|
||||
securitymd=securitymd_exists,
|
||||
codeowners_exists=codeowners_exists,
|
||||
secret_scanning_enabled=secret_scanning_enabled,
|
||||
dependabot_alerts_enabled=dependabot_alerts_enabled,
|
||||
delete_branch_on_merge=delete_branch_on_merge,
|
||||
)
|
||||
|
||||
|
||||
class Branch(BaseModel):
|
||||
"""Model for Github Branch"""
|
||||
|
||||
@@ -641,3 +641,63 @@ class TestGitHubProvider:
|
||||
|
||||
with pytest.raises(GithubInvalidProviderIdError):
|
||||
GithubProvider.validate_provider_id(mock_session, "invalid-org")
|
||||
|
||||
|
||||
class Test_GithubProvider_Scoping:
|
||||
def setup_method(self):
|
||||
"""Setup mock session and identity for testing"""
|
||||
self.mock_session = GithubSession(
|
||||
token=PAT_TOKEN,
|
||||
key="",
|
||||
id=0,
|
||||
)
|
||||
self.mock_identity = GithubIdentityInfo(
|
||||
account_id=ACCOUNT_ID,
|
||||
account_name=ACCOUNT_NAME,
|
||||
account_url=ACCOUNT_URL,
|
||||
)
|
||||
|
||||
def test_provider_scoping_integration(self):
|
||||
"""Test that provider correctly handles scoping parameters and integrates with services"""
|
||||
repositories = ["owner1/repo1", "owner2/repo2"]
|
||||
organizations = ["org1", "org2"]
|
||||
|
||||
with (
|
||||
patch(
|
||||
"prowler.providers.github.github_provider.GithubProvider.setup_session",
|
||||
return_value=self.mock_session,
|
||||
),
|
||||
patch(
|
||||
"prowler.providers.github.github_provider.GithubProvider.setup_identity",
|
||||
return_value=self.mock_identity,
|
||||
),
|
||||
):
|
||||
# Test with both scoping parameters
|
||||
provider = GithubProvider(
|
||||
personal_access_token=PAT_TOKEN,
|
||||
repositories=repositories,
|
||||
organizations=organizations,
|
||||
)
|
||||
|
||||
assert provider.repositories == repositories
|
||||
assert provider.organizations == organizations
|
||||
assert provider.type == "github"
|
||||
assert provider.auth_method == "Personal Access Token"
|
||||
|
||||
# Test with no scoping (backward compatibility)
|
||||
provider_no_scoping = GithubProvider(
|
||||
personal_access_token=PAT_TOKEN,
|
||||
)
|
||||
|
||||
assert provider_no_scoping.repositories == []
|
||||
assert provider_no_scoping.organizations == []
|
||||
|
||||
# Test with None values (should convert to empty lists)
|
||||
provider_none = GithubProvider(
|
||||
personal_access_token=PAT_TOKEN,
|
||||
repositories=None,
|
||||
organizations=None,
|
||||
)
|
||||
|
||||
assert provider_none.repositories == []
|
||||
assert provider_none.organizations == []
|
||||
|
||||
@@ -0,0 +1,305 @@
|
||||
import argparse
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from prowler.providers.github.lib.arguments import arguments
|
||||
|
||||
|
||||
class Test_GitHubArguments:
|
||||
def setup_method(self):
|
||||
"""Setup mock ArgumentParser for testing"""
|
||||
self.mock_parser = MagicMock()
|
||||
self.mock_subparsers = MagicMock()
|
||||
self.mock_github_parser = MagicMock()
|
||||
self.mock_auth_group = MagicMock()
|
||||
self.mock_scoping_group = MagicMock()
|
||||
|
||||
# Setup the mock chain
|
||||
self.mock_parser.add_subparsers.return_value = self.mock_subparsers
|
||||
self.mock_subparsers.add_parser.return_value = self.mock_github_parser
|
||||
self.mock_github_parser.add_argument_group.side_effect = [
|
||||
self.mock_auth_group,
|
||||
self.mock_scoping_group,
|
||||
]
|
||||
|
||||
def test_init_parser_creates_subparser(self):
|
||||
"""Test that init_parser creates the GitHub subparser correctly"""
|
||||
# Create a mock object that has the necessary attributes
|
||||
mock_github_args = MagicMock()
|
||||
mock_github_args.subparsers = self.mock_subparsers
|
||||
mock_github_args.common_providers_parser = MagicMock()
|
||||
|
||||
# Call init_parser
|
||||
arguments.init_parser(mock_github_args)
|
||||
|
||||
# Verify subparser was created
|
||||
self.mock_subparsers.add_parser.assert_called_once_with(
|
||||
"github",
|
||||
parents=[mock_github_args.common_providers_parser],
|
||||
help="GitHub Provider",
|
||||
)
|
||||
|
||||
def test_init_parser_creates_argument_groups(self):
|
||||
"""Test that init_parser creates the correct argument groups"""
|
||||
mock_github_args = MagicMock()
|
||||
mock_github_args.subparsers = self.mock_subparsers
|
||||
mock_github_args.common_providers_parser = MagicMock()
|
||||
|
||||
arguments.init_parser(mock_github_args)
|
||||
|
||||
# Verify argument groups were created
|
||||
assert self.mock_github_parser.add_argument_group.call_count == 2
|
||||
calls = self.mock_github_parser.add_argument_group.call_args_list
|
||||
assert calls[0][0][0] == "Authentication Modes"
|
||||
assert calls[1][0][0] == "Scan Scoping"
|
||||
|
||||
def test_init_parser_adds_authentication_arguments(self):
|
||||
"""Test that init_parser adds all authentication arguments"""
|
||||
mock_github_args = MagicMock()
|
||||
mock_github_args.subparsers = self.mock_subparsers
|
||||
mock_github_args.common_providers_parser = MagicMock()
|
||||
|
||||
arguments.init_parser(mock_github_args)
|
||||
|
||||
# Verify authentication arguments were added
|
||||
assert self.mock_auth_group.add_argument.call_count == 4
|
||||
|
||||
# Check that all authentication arguments are present
|
||||
calls = self.mock_auth_group.add_argument.call_args_list
|
||||
auth_args = [call[0][0] for call in calls]
|
||||
|
||||
assert "--personal-access-token" in auth_args
|
||||
assert "--oauth-app-token" in auth_args
|
||||
assert "--github-app-id" in auth_args
|
||||
# Check for either form of the github app key argument
|
||||
assert any("--github-app-key" in arg for arg in auth_args)
|
||||
|
||||
def test_init_parser_adds_scoping_arguments(self):
|
||||
"""Test that init_parser adds all scoping arguments"""
|
||||
mock_github_args = MagicMock()
|
||||
mock_github_args.subparsers = self.mock_subparsers
|
||||
mock_github_args.common_providers_parser = MagicMock()
|
||||
|
||||
arguments.init_parser(mock_github_args)
|
||||
|
||||
# Verify scoping arguments were added
|
||||
assert self.mock_scoping_group.add_argument.call_count == 2
|
||||
|
||||
# Check that all scoping arguments are present
|
||||
calls = self.mock_scoping_group.add_argument.call_args_list
|
||||
scoping_args = [call[0][0] for call in calls]
|
||||
|
||||
assert "--repository" in scoping_args
|
||||
assert "--organization" in scoping_args
|
||||
|
||||
def test_repository_argument_configuration(self):
|
||||
"""Test that repository argument is configured correctly"""
|
||||
mock_github_args = MagicMock()
|
||||
mock_github_args.subparsers = self.mock_subparsers
|
||||
mock_github_args.common_providers_parser = MagicMock()
|
||||
|
||||
arguments.init_parser(mock_github_args)
|
||||
|
||||
# Find the repository argument call
|
||||
calls = self.mock_scoping_group.add_argument.call_args_list
|
||||
repo_call = None
|
||||
for call in calls:
|
||||
if call[0][0] == "--repository":
|
||||
repo_call = call
|
||||
break
|
||||
|
||||
assert repo_call is not None
|
||||
|
||||
# Check argument configuration
|
||||
kwargs = repo_call[1]
|
||||
assert kwargs["nargs"] == "*"
|
||||
assert kwargs["default"] is None
|
||||
assert kwargs["metavar"] == "REPOSITORY"
|
||||
assert "owner/repo-name" in kwargs["help"]
|
||||
|
||||
def test_organization_argument_configuration(self):
|
||||
"""Test that organization argument is configured correctly"""
|
||||
mock_github_args = MagicMock()
|
||||
mock_github_args.subparsers = self.mock_subparsers
|
||||
mock_github_args.common_providers_parser = MagicMock()
|
||||
|
||||
arguments.init_parser(mock_github_args)
|
||||
|
||||
# Find the organization argument call
|
||||
calls = self.mock_scoping_group.add_argument.call_args_list
|
||||
org_call = None
|
||||
for call in calls:
|
||||
if call[0][0] == "--organization":
|
||||
org_call = call
|
||||
break
|
||||
|
||||
assert org_call is not None
|
||||
|
||||
# Check argument configuration
|
||||
kwargs = org_call[1]
|
||||
assert kwargs["nargs"] == "*"
|
||||
assert kwargs["default"] is None
|
||||
assert kwargs["metavar"] == "ORGANIZATION"
|
||||
assert "Organization or user name" in kwargs["help"]
|
||||
|
||||
|
||||
class Test_GitHubArguments_Integration:
|
||||
def test_real_argument_parsing_no_scoping(self):
|
||||
"""Test parsing arguments without scoping parameters"""
|
||||
parser = argparse.ArgumentParser()
|
||||
subparsers = parser.add_subparsers()
|
||||
common_parser = argparse.ArgumentParser(add_help=False)
|
||||
|
||||
# Create a mock object that mimics the structure used by the init_parser function
|
||||
mock_github_args = MagicMock()
|
||||
mock_github_args.subparsers = subparsers
|
||||
mock_github_args.common_providers_parser = common_parser
|
||||
|
||||
arguments.init_parser(mock_github_args)
|
||||
|
||||
# Parse arguments without scoping
|
||||
args = parser.parse_args(["github", "--personal-access-token", "test-token"])
|
||||
|
||||
assert args.personal_access_token == "test-token"
|
||||
assert args.repository is None
|
||||
assert args.organization is None
|
||||
|
||||
def test_real_argument_parsing_with_repository(self):
|
||||
"""Test parsing arguments with repository scoping"""
|
||||
parser = argparse.ArgumentParser()
|
||||
subparsers = parser.add_subparsers()
|
||||
common_parser = argparse.ArgumentParser(add_help=False)
|
||||
|
||||
mock_github_args = MagicMock()
|
||||
mock_github_args.subparsers = subparsers
|
||||
mock_github_args.common_providers_parser = common_parser
|
||||
|
||||
arguments.init_parser(mock_github_args)
|
||||
|
||||
# Parse arguments with repository scoping
|
||||
args = parser.parse_args(
|
||||
[
|
||||
"github",
|
||||
"--personal-access-token",
|
||||
"test-token",
|
||||
"--repository",
|
||||
"owner1/repo1",
|
||||
"owner2/repo2",
|
||||
]
|
||||
)
|
||||
|
||||
assert args.personal_access_token == "test-token"
|
||||
assert args.repository == ["owner1/repo1", "owner2/repo2"]
|
||||
assert args.organization is None
|
||||
|
||||
def test_real_argument_parsing_with_organization(self):
|
||||
"""Test parsing arguments with organization scoping"""
|
||||
parser = argparse.ArgumentParser()
|
||||
subparsers = parser.add_subparsers()
|
||||
common_parser = argparse.ArgumentParser(add_help=False)
|
||||
|
||||
mock_github_args = MagicMock()
|
||||
mock_github_args.subparsers = subparsers
|
||||
mock_github_args.common_providers_parser = common_parser
|
||||
|
||||
arguments.init_parser(mock_github_args)
|
||||
|
||||
# Parse arguments with organization scoping
|
||||
args = parser.parse_args(
|
||||
[
|
||||
"github",
|
||||
"--personal-access-token",
|
||||
"test-token",
|
||||
"--organization",
|
||||
"org1",
|
||||
"org2",
|
||||
]
|
||||
)
|
||||
|
||||
assert args.personal_access_token == "test-token"
|
||||
assert args.repository is None
|
||||
assert args.organization == ["org1", "org2"]
|
||||
|
||||
def test_real_argument_parsing_with_both_scoping(self):
|
||||
"""Test parsing arguments with both repository and organization scoping"""
|
||||
parser = argparse.ArgumentParser()
|
||||
subparsers = parser.add_subparsers()
|
||||
common_parser = argparse.ArgumentParser(add_help=False)
|
||||
|
||||
mock_github_args = MagicMock()
|
||||
mock_github_args.subparsers = subparsers
|
||||
mock_github_args.common_providers_parser = common_parser
|
||||
|
||||
arguments.init_parser(mock_github_args)
|
||||
|
||||
# Parse arguments with both scoping types
|
||||
args = parser.parse_args(
|
||||
[
|
||||
"github",
|
||||
"--personal-access-token",
|
||||
"test-token",
|
||||
"--repository",
|
||||
"owner1/repo1",
|
||||
"--organization",
|
||||
"org1",
|
||||
]
|
||||
)
|
||||
|
||||
assert args.personal_access_token == "test-token"
|
||||
assert args.repository == ["owner1/repo1"]
|
||||
assert args.organization == ["org1"]
|
||||
|
||||
def test_real_argument_parsing_single_values(self):
|
||||
"""Test parsing arguments with single repository and organization values"""
|
||||
parser = argparse.ArgumentParser()
|
||||
subparsers = parser.add_subparsers()
|
||||
common_parser = argparse.ArgumentParser(add_help=False)
|
||||
|
||||
mock_github_args = MagicMock()
|
||||
mock_github_args.subparsers = subparsers
|
||||
mock_github_args.common_providers_parser = common_parser
|
||||
|
||||
arguments.init_parser(mock_github_args)
|
||||
|
||||
# Parse arguments with single values
|
||||
args = parser.parse_args(
|
||||
[
|
||||
"github",
|
||||
"--personal-access-token",
|
||||
"test-token",
|
||||
"--repository",
|
||||
"owner1/repo1",
|
||||
"--organization",
|
||||
"org1",
|
||||
]
|
||||
)
|
||||
|
||||
assert args.personal_access_token == "test-token"
|
||||
assert args.repository == ["owner1/repo1"]
|
||||
assert args.organization == ["org1"]
|
||||
|
||||
def test_real_argument_parsing_empty_scoping(self):
|
||||
"""Test parsing arguments with empty scoping values"""
|
||||
parser = argparse.ArgumentParser()
|
||||
subparsers = parser.add_subparsers()
|
||||
common_parser = argparse.ArgumentParser(add_help=False)
|
||||
|
||||
mock_github_args = MagicMock()
|
||||
mock_github_args.subparsers = subparsers
|
||||
mock_github_args.common_providers_parser = common_parser
|
||||
|
||||
arguments.init_parser(mock_github_args)
|
||||
|
||||
# Parse arguments with empty scoping flags
|
||||
args = parser.parse_args(
|
||||
[
|
||||
"github",
|
||||
"--personal-access-token",
|
||||
"test-token",
|
||||
"--repository",
|
||||
"--organization",
|
||||
]
|
||||
)
|
||||
|
||||
assert args.personal_access_token == "test-token"
|
||||
assert args.repository == []
|
||||
assert args.organization == []
|
||||
@@ -1,4 +1,6 @@
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from github import GithubException, RateLimitExceededException
|
||||
|
||||
from prowler.providers.github.services.organization.organization_service import (
|
||||
Org,
|
||||
@@ -35,3 +37,330 @@ class Test_Repository_Service:
|
||||
assert len(repository_service.organizations) == 1
|
||||
assert repository_service.organizations[1].name == "test-organization"
|
||||
assert repository_service.organizations[1].mfa_required
|
||||
|
||||
|
||||
class Test_Organization_Scoping:
|
||||
def setup_method(self):
|
||||
self.mock_org1 = MagicMock()
|
||||
self.mock_org1.id = 1
|
||||
self.mock_org1.login = "test-org1"
|
||||
self.mock_org1.two_factor_requirement_enabled = True
|
||||
|
||||
self.mock_org2 = MagicMock()
|
||||
self.mock_org2.id = 2
|
||||
self.mock_org2.login = "test-org2"
|
||||
self.mock_org2.two_factor_requirement_enabled = False
|
||||
|
||||
self.mock_user = MagicMock()
|
||||
self.mock_user.id = 100
|
||||
self.mock_user.login = "test-user"
|
||||
|
||||
def test_no_organization_scoping(self):
|
||||
"""Test that all user organizations are returned when no scoping is specified"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = []
|
||||
provider.organizations = []
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_user = MagicMock()
|
||||
mock_user.get_orgs.return_value = [self.mock_org1, self.mock_org2]
|
||||
mock_client.get_user.return_value = mock_user
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.organization.organization_service.GithubService.__init__"
|
||||
):
|
||||
organization_service = Organization(provider)
|
||||
organization_service.clients = [mock_client]
|
||||
organization_service.provider = provider
|
||||
|
||||
orgs = organization_service._list_organizations()
|
||||
|
||||
assert len(orgs) == 2
|
||||
assert 1 in orgs
|
||||
assert 2 in orgs
|
||||
assert orgs[1].name == "test-org1"
|
||||
assert orgs[2].name == "test-org2"
|
||||
|
||||
def test_specific_organization_scoping(self):
|
||||
"""Test that only specified organizations are returned"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = []
|
||||
provider.organizations = ["test-org1"]
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.get_organization.return_value = self.mock_org1
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.organization.organization_service.GithubService.__init__"
|
||||
):
|
||||
organization_service = Organization(provider)
|
||||
organization_service.clients = [mock_client]
|
||||
organization_service.provider = provider
|
||||
|
||||
orgs = organization_service._list_organizations()
|
||||
|
||||
assert len(orgs) == 1
|
||||
assert 1 in orgs
|
||||
assert orgs[1].name == "test-org1"
|
||||
assert orgs[1].mfa_required is True
|
||||
mock_client.get_organization.assert_called_once_with("test-org1")
|
||||
|
||||
def test_multiple_organization_scoping(self):
|
||||
"""Test that multiple specified organizations are returned"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = []
|
||||
provider.organizations = ["test-org1", "test-org2"]
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.get_organization.side_effect = [self.mock_org1, self.mock_org2]
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.organization.organization_service.GithubService.__init__"
|
||||
):
|
||||
organization_service = Organization(provider)
|
||||
organization_service.clients = [mock_client]
|
||||
organization_service.provider = provider
|
||||
|
||||
orgs = organization_service._list_organizations()
|
||||
|
||||
assert len(orgs) == 2
|
||||
assert 1 in orgs
|
||||
assert 2 in orgs
|
||||
assert orgs[1].name == "test-org1"
|
||||
assert orgs[2].name == "test-org2"
|
||||
assert mock_client.get_organization.call_count == 2
|
||||
|
||||
def test_organization_as_user_fallback(self):
|
||||
"""Test that organization scoping falls back to user when organization not found"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = []
|
||||
provider.organizations = ["test-user"]
|
||||
|
||||
mock_client = MagicMock()
|
||||
# Organization lookup fails
|
||||
mock_client.get_organization.side_effect = GithubException(
|
||||
404, "Not Found", None
|
||||
)
|
||||
# User lookup succeeds
|
||||
mock_client.get_user.return_value = self.mock_user
|
||||
|
||||
# Create service without calling the parent constructor
|
||||
organization_service = Organization.__new__(Organization)
|
||||
organization_service.clients = [mock_client]
|
||||
organization_service.provider = provider
|
||||
|
||||
orgs = organization_service._list_organizations()
|
||||
|
||||
assert len(orgs) == 1
|
||||
assert 100 in orgs
|
||||
assert orgs[100].name == "test-user"
|
||||
assert orgs[100].mfa_required is None # Users don't have MFA requirements
|
||||
mock_client.get_organization.assert_called_once_with("test-user")
|
||||
mock_client.get_user.assert_called_once_with("test-user")
|
||||
|
||||
def test_repository_only_scoping_no_organization_checks(self):
|
||||
"""Test that repository-only scoping does NOT perform organization checks"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = ["owner1/repo1", "owner2/repo2"]
|
||||
provider.organizations = []
|
||||
|
||||
mock_client = MagicMock()
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.organization.organization_service.GithubService.__init__"
|
||||
):
|
||||
organization_service = Organization(provider)
|
||||
organization_service.clients = [mock_client]
|
||||
organization_service.provider = provider
|
||||
|
||||
orgs = organization_service._list_organizations()
|
||||
|
||||
# Should be empty - no organization checks when only repositories are specified
|
||||
assert len(orgs) == 0
|
||||
# Should not call get_organization at all
|
||||
mock_client.get_organization.assert_not_called()
|
||||
mock_client.get_user.assert_not_called()
|
||||
|
||||
def test_combined_repository_and_organization_scoping(self):
|
||||
"""Test that both repository owners and specified organizations are included"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = ["owner1/repo1"]
|
||||
provider.organizations = ["specific-org"]
|
||||
|
||||
mock_client = MagicMock()
|
||||
# Mock different organizations for owner1 and specific-org
|
||||
mock_owner_org = MagicMock()
|
||||
mock_owner_org.id = 1
|
||||
mock_owner_org.login = "owner1"
|
||||
mock_owner_org.two_factor_requirement_enabled = True
|
||||
|
||||
mock_specific_org = MagicMock()
|
||||
mock_specific_org.id = 2
|
||||
mock_specific_org.login = "specific-org"
|
||||
mock_specific_org.two_factor_requirement_enabled = False
|
||||
|
||||
mock_client.get_organization.side_effect = [
|
||||
mock_owner_org,
|
||||
mock_specific_org,
|
||||
]
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.organization.organization_service.GithubService.__init__"
|
||||
):
|
||||
organization_service = Organization(provider)
|
||||
organization_service.clients = [mock_client]
|
||||
organization_service.provider = provider
|
||||
|
||||
orgs = organization_service._list_organizations()
|
||||
|
||||
assert len(orgs) == 2
|
||||
assert 1 in orgs
|
||||
assert 2 in orgs
|
||||
assert orgs[1].name == "owner1"
|
||||
assert orgs[2].name == "specific-org"
|
||||
assert mock_client.get_organization.call_count == 2
|
||||
|
||||
def test_organization_not_found(self):
|
||||
"""Test that inaccessible organizations are skipped with warning"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = []
|
||||
provider.organizations = ["nonexistent-org"]
|
||||
|
||||
mock_client = MagicMock()
|
||||
# Both organization and user lookups fail
|
||||
mock_client.get_organization.side_effect = Exception("404 Not Found")
|
||||
mock_client.get_user.side_effect = Exception("404 Not Found")
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.organization.organization_service.GithubService.__init__"
|
||||
):
|
||||
organization_service = Organization(provider)
|
||||
organization_service.clients = [mock_client]
|
||||
organization_service.provider = provider
|
||||
|
||||
orgs = organization_service._list_organizations()
|
||||
|
||||
# Should be empty since organization/user wasn't found
|
||||
assert len(orgs) == 0
|
||||
|
||||
def test_organization_error_handling(self):
|
||||
"""Test that other errors (non-404) are handled gracefully"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = []
|
||||
provider.organizations = ["error-org"]
|
||||
|
||||
mock_client = MagicMock()
|
||||
# Organization lookup fails with non-404 error
|
||||
mock_client.get_organization.side_effect = Exception(
|
||||
"500 Internal Server Error"
|
||||
)
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.organization.organization_service.GithubService.__init__"
|
||||
):
|
||||
organization_service = Organization(provider)
|
||||
organization_service.clients = [mock_client]
|
||||
organization_service.provider = provider
|
||||
|
||||
orgs = organization_service._list_organizations()
|
||||
|
||||
# Should be empty since organization wasn't accessible
|
||||
assert len(orgs) == 0
|
||||
|
||||
def test_duplicate_organizations_handling(self):
|
||||
"""Test that duplicate organizations (e.g., from repositories and organizations) are handled correctly"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = ["test-org1/repo1"]
|
||||
provider.organizations = ["test-org1"] # Same organization specified twice
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.get_organization.return_value = self.mock_org1
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.organization.organization_service.GithubService.__init__"
|
||||
):
|
||||
organization_service = Organization(provider)
|
||||
organization_service.clients = [mock_client]
|
||||
organization_service.provider = provider
|
||||
|
||||
orgs = organization_service._list_organizations()
|
||||
|
||||
# Should only have one organization despite being specified twice
|
||||
assert len(orgs) == 1
|
||||
assert 1 in orgs
|
||||
assert orgs[1].name == "test-org1"
|
||||
# Should only call get_organization once due to set deduplication
|
||||
mock_client.get_organization.assert_called_once_with("test-org1")
|
||||
|
||||
|
||||
class Test_Organization_ErrorHandling:
|
||||
def setup_method(self):
|
||||
self.mock_org1 = MagicMock()
|
||||
self.mock_org1.id = 1
|
||||
self.mock_org1.login = "test-org1"
|
||||
self.mock_org1.two_factor_requirement_enabled = True
|
||||
|
||||
def test_github_api_error_handling(self):
|
||||
"""Test that GitHub API errors are handled properly"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = []
|
||||
provider.organizations = ["test-org1"]
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.get_organization.side_effect = GithubException(
|
||||
403, "Forbidden", None
|
||||
)
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.organization.organization_service.GithubService.__init__"
|
||||
):
|
||||
organization_service = Organization(provider)
|
||||
organization_service.clients = [mock_client]
|
||||
organization_service.provider = provider
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.organization.organization_service.logger"
|
||||
) as mock_logger:
|
||||
orgs = organization_service._list_organizations()
|
||||
|
||||
# Should be empty due to API error
|
||||
assert len(orgs) == 0
|
||||
# Should log specific error message
|
||||
mock_logger.warning.assert_called()
|
||||
# Check if Access denied message was logged (could be in warning or error calls)
|
||||
log_messages = [
|
||||
str(call)
|
||||
for call in mock_logger.warning.call_args_list
|
||||
+ mock_logger.error.call_args_list
|
||||
]
|
||||
assert any("Access denied" in msg for msg in log_messages)
|
||||
|
||||
def test_rate_limit_error_handling(self):
|
||||
"""Test that rate limit errors are logged appropriately"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = []
|
||||
provider.organizations = ["test-org1"]
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.get_organization.side_effect = RateLimitExceededException(
|
||||
429, "Rate limit exceeded", None
|
||||
)
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.organization.organization_service.GithubService.__init__"
|
||||
):
|
||||
organization_service = Organization(provider)
|
||||
organization_service.clients = [mock_client]
|
||||
organization_service.provider = provider
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.organization.organization_service.logger"
|
||||
) as mock_logger:
|
||||
# Rate limit errors should be caught and logged at the outer level
|
||||
orgs = organization_service._list_organizations()
|
||||
|
||||
# Should be empty due to rate limit error
|
||||
assert len(orgs) == 0
|
||||
# Should log rate limit error
|
||||
mock_logger.error.assert_called()
|
||||
assert "Rate limit exceeded" in str(mock_logger.error.call_args)
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from github import GithubException, RateLimitExceededException
|
||||
|
||||
from prowler.providers.github.services.repository.repository_service import (
|
||||
Branch,
|
||||
Repo,
|
||||
@@ -106,3 +108,358 @@ class Test_Repository_FileExists:
|
||||
) as mock_logger:
|
||||
assert self.repository._file_exists(self.mock_repo, "errorfile.txt") is None
|
||||
assert mock_logger.error.called
|
||||
|
||||
|
||||
class Test_Repository_Scoping:
|
||||
def setup_method(self):
|
||||
self.mock_repo1 = MagicMock()
|
||||
self.mock_repo1.id = 1
|
||||
self.mock_repo1.name = "repo1"
|
||||
self.mock_repo1.owner.login = "owner1"
|
||||
self.mock_repo1.full_name = "owner1/repo1"
|
||||
self.mock_repo1.default_branch = "main"
|
||||
self.mock_repo1.private = False
|
||||
self.mock_repo1.archived = False
|
||||
self.mock_repo1.pushed_at = datetime.now(timezone.utc)
|
||||
self.mock_repo1.delete_branch_on_merge = True
|
||||
self.mock_repo1.security_and_analysis = None
|
||||
self.mock_repo1.get_contents.return_value = None
|
||||
self.mock_repo1.get_branch.side_effect = Exception("404 Not Found")
|
||||
self.mock_repo1.get_dependabot_alerts.side_effect = Exception("404 Not Found")
|
||||
|
||||
self.mock_repo2 = MagicMock()
|
||||
self.mock_repo2.id = 2
|
||||
self.mock_repo2.name = "repo2"
|
||||
self.mock_repo2.owner.login = "owner2"
|
||||
self.mock_repo2.full_name = "owner2/repo2"
|
||||
self.mock_repo2.default_branch = "main"
|
||||
self.mock_repo2.private = False
|
||||
self.mock_repo2.archived = False
|
||||
self.mock_repo2.pushed_at = datetime.now(timezone.utc)
|
||||
self.mock_repo2.delete_branch_on_merge = True
|
||||
self.mock_repo2.security_and_analysis = None
|
||||
self.mock_repo2.get_contents.return_value = None
|
||||
self.mock_repo2.get_branch.side_effect = Exception("404 Not Found")
|
||||
self.mock_repo2.get_dependabot_alerts.side_effect = Exception("404 Not Found")
|
||||
|
||||
def test_no_repository_scoping(self):
|
||||
"""Test that all repositories are returned when no scoping is specified"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = []
|
||||
provider.organizations = []
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_user = MagicMock()
|
||||
mock_user.get_repos.return_value = [self.mock_repo1, self.mock_repo2]
|
||||
mock_client.get_user.return_value = mock_user
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.repository.repository_service.GithubService.__init__"
|
||||
):
|
||||
repository_service = Repository(provider)
|
||||
repository_service.clients = [mock_client]
|
||||
repository_service.provider = provider
|
||||
|
||||
repos = repository_service._list_repositories()
|
||||
|
||||
assert len(repos) == 2
|
||||
assert 1 in repos
|
||||
assert 2 in repos
|
||||
assert repos[1].name == "repo1"
|
||||
assert repos[2].name == "repo2"
|
||||
|
||||
def test_specific_repository_scoping(self):
|
||||
"""Test that only specified repositories are returned"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = ["owner1/repo1"]
|
||||
provider.organizations = []
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.get_repo.return_value = self.mock_repo1
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.repository.repository_service.GithubService.__init__"
|
||||
):
|
||||
repository_service = Repository(provider)
|
||||
repository_service.clients = [mock_client]
|
||||
repository_service.provider = provider
|
||||
|
||||
repos = repository_service._list_repositories()
|
||||
|
||||
assert len(repos) == 1
|
||||
assert 1 in repos
|
||||
assert repos[1].name == "repo1"
|
||||
assert repos[1].full_name == "owner1/repo1"
|
||||
mock_client.get_repo.assert_called_once_with("owner1/repo1")
|
||||
|
||||
def test_multiple_repository_scoping(self):
|
||||
"""Test that multiple specified repositories are returned"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = ["owner1/repo1", "owner2/repo2"]
|
||||
provider.organizations = []
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.get_repo.side_effect = [self.mock_repo1, self.mock_repo2]
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.repository.repository_service.GithubService.__init__"
|
||||
):
|
||||
repository_service = Repository(provider)
|
||||
repository_service.clients = [mock_client]
|
||||
repository_service.provider = provider
|
||||
|
||||
repos = repository_service._list_repositories()
|
||||
|
||||
assert len(repos) == 2
|
||||
assert 1 in repos
|
||||
assert 2 in repos
|
||||
assert repos[1].name == "repo1"
|
||||
assert repos[2].name == "repo2"
|
||||
assert mock_client.get_repo.call_count == 2
|
||||
|
||||
def test_invalid_repository_format(self):
|
||||
"""Test that invalid repository formats are skipped with warning"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = ["invalid-repo-name", "owner/valid-repo"]
|
||||
provider.organizations = []
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.get_repo.return_value = self.mock_repo1
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.repository.repository_service.GithubService.__init__"
|
||||
):
|
||||
repository_service = Repository(provider)
|
||||
repository_service.clients = [mock_client]
|
||||
repository_service.provider = provider
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.repository.repository_service.logger"
|
||||
) as mock_logger:
|
||||
repos = repository_service._list_repositories()
|
||||
|
||||
# Should only have the valid repository
|
||||
assert len(repos) == 1
|
||||
assert 1 in repos
|
||||
# Should log warning for invalid format
|
||||
assert mock_logger.warning.call_count >= 1
|
||||
# Check that at least one warning is about invalid format
|
||||
warning_calls = [
|
||||
call[0][0] for call in mock_logger.warning.call_args_list
|
||||
]
|
||||
assert any(
|
||||
"should be in 'owner/repo-name' format" in call
|
||||
for call in warning_calls
|
||||
)
|
||||
|
||||
def test_repository_not_found(self):
|
||||
"""Test that inaccessible repositories are skipped with warning"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = ["owner/nonexistent-repo"]
|
||||
provider.organizations = []
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.get_repo.side_effect = Exception("404 Not Found")
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.repository.repository_service.GithubService.__init__"
|
||||
):
|
||||
repository_service = Repository(provider)
|
||||
repository_service.clients = [mock_client]
|
||||
repository_service.provider = provider
|
||||
|
||||
repos = repository_service._list_repositories()
|
||||
|
||||
# Should be empty since repository wasn't found
|
||||
assert len(repos) == 0
|
||||
|
||||
def test_organization_scoping(self):
|
||||
"""Test that repositories from specified organizations are returned"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = []
|
||||
provider.organizations = ["org1"]
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_org = MagicMock()
|
||||
mock_org.get_repos.return_value = [self.mock_repo1]
|
||||
mock_client.get_organization.return_value = mock_org
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.repository.repository_service.GithubService.__init__"
|
||||
):
|
||||
repository_service = Repository(provider)
|
||||
repository_service.clients = [mock_client]
|
||||
repository_service.provider = provider
|
||||
|
||||
repos = repository_service._list_repositories()
|
||||
|
||||
assert len(repos) == 1
|
||||
assert 1 in repos
|
||||
assert repos[1].name == "repo1"
|
||||
mock_client.get_organization.assert_called_once_with("org1")
|
||||
|
||||
def test_organization_as_user_fallback(self):
|
||||
"""Test that organization scoping falls back to user when organization not found"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = []
|
||||
provider.organizations = ["user1"]
|
||||
|
||||
mock_client = MagicMock()
|
||||
# Organization lookup fails
|
||||
mock_client.get_organization.side_effect = GithubException(
|
||||
404, "Not Found", None
|
||||
)
|
||||
# User lookup succeeds
|
||||
mock_user = MagicMock()
|
||||
mock_user.get_repos.return_value = [self.mock_repo1]
|
||||
mock_client.get_user.return_value = mock_user
|
||||
|
||||
# Create service without calling the parent constructor
|
||||
repository_service = Repository.__new__(Repository)
|
||||
repository_service.clients = [mock_client]
|
||||
repository_service.provider = provider
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.repository.repository_service.logger"
|
||||
) as mock_logger:
|
||||
repos = repository_service._list_repositories()
|
||||
|
||||
assert len(repos) == 1
|
||||
assert 1 in repos
|
||||
assert repos[1].name == "repo1"
|
||||
mock_client.get_organization.assert_called_once_with("user1")
|
||||
mock_client.get_user.assert_called_once_with("user1")
|
||||
# Should log info about trying as user
|
||||
mock_logger.info.assert_called()
|
||||
|
||||
def test_combined_repository_and_organization_scoping(self):
|
||||
"""Test that both repository and organization scoping can be used together"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = ["owner1/repo1"]
|
||||
provider.organizations = ["org2"]
|
||||
|
||||
mock_client = MagicMock()
|
||||
# Repository lookup
|
||||
mock_client.get_repo.return_value = self.mock_repo1
|
||||
# Organization lookup
|
||||
mock_org = MagicMock()
|
||||
mock_org.get_repos.return_value = [self.mock_repo2]
|
||||
mock_client.get_organization.return_value = mock_org
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.repository.repository_service.GithubService.__init__"
|
||||
):
|
||||
repository_service = Repository(provider)
|
||||
repository_service.clients = [mock_client]
|
||||
repository_service.provider = provider
|
||||
|
||||
repos = repository_service._list_repositories()
|
||||
|
||||
assert len(repos) == 2
|
||||
assert 1 in repos
|
||||
assert 2 in repos
|
||||
assert repos[1].name == "repo1"
|
||||
assert repos[2].name == "repo2"
|
||||
mock_client.get_repo.assert_called_once_with("owner1/repo1")
|
||||
mock_client.get_organization.assert_called_once_with("org2")
|
||||
|
||||
|
||||
class Test_Repository_Validation:
|
||||
def setup_method(self):
|
||||
self.repository = Repository(set_mocked_github_provider())
|
||||
|
||||
def test_validate_repository_format_valid(self):
|
||||
"""Test repository format validation with valid formats"""
|
||||
# Valid formats
|
||||
assert self.repository._validate_repository_format("owner/repo") is True
|
||||
assert self.repository._validate_repository_format("my-org/my-repo") is True
|
||||
assert (
|
||||
self.repository._validate_repository_format("user123/project_name") is True
|
||||
)
|
||||
assert self.repository._validate_repository_format("org/repo.name") is True
|
||||
|
||||
def test_validate_repository_format_invalid(self):
|
||||
"""Test repository format validation with invalid formats"""
|
||||
# Invalid formats
|
||||
assert self.repository._validate_repository_format("") is False
|
||||
assert self.repository._validate_repository_format("no-slash") is False
|
||||
assert self.repository._validate_repository_format("too/many/slashes") is False
|
||||
assert self.repository._validate_repository_format("/missing-owner") is False
|
||||
assert self.repository._validate_repository_format("missing-repo/") is False
|
||||
|
||||
|
||||
class Test_Repository_ErrorHandling:
|
||||
def setup_method(self):
|
||||
self.mock_repo1 = MagicMock()
|
||||
self.mock_repo1.id = 1
|
||||
self.mock_repo1.name = "repo1"
|
||||
self.mock_repo1.owner.login = "owner1"
|
||||
self.mock_repo1.full_name = "owner1/repo1"
|
||||
self.mock_repo1.default_branch = "main"
|
||||
self.mock_repo1.private = False
|
||||
self.mock_repo1.archived = False
|
||||
self.mock_repo1.pushed_at = datetime.now(timezone.utc)
|
||||
self.mock_repo1.delete_branch_on_merge = True
|
||||
self.mock_repo1.security_and_analysis = None
|
||||
self.mock_repo1.get_contents.return_value = None
|
||||
self.mock_repo1.get_branch.side_effect = Exception("404 Not Found")
|
||||
self.mock_repo1.get_dependabot_alerts.side_effect = Exception("404 Not Found")
|
||||
|
||||
def test_github_api_error_handling(self):
|
||||
"""Test that GitHub API errors are handled properly"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = ["owner/repo1"]
|
||||
provider.organizations = []
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.get_repo.side_effect = GithubException(403, "Forbidden", None)
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.repository.repository_service.GithubService.__init__"
|
||||
):
|
||||
repository_service = Repository(provider)
|
||||
repository_service.clients = [mock_client]
|
||||
repository_service.provider = provider
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.lib.service.service.logger"
|
||||
) as mock_logger:
|
||||
repos = repository_service._list_repositories()
|
||||
|
||||
# Should be empty due to API error
|
||||
assert len(repos) == 0
|
||||
# Should log specific error message
|
||||
mock_logger.error.assert_called()
|
||||
# Check if Access denied message was logged
|
||||
log_messages = [str(call) for call in mock_logger.error.call_args_list]
|
||||
assert any("Access denied" in msg for msg in log_messages)
|
||||
|
||||
def test_rate_limit_error_handling(self):
|
||||
"""Test that rate limit errors are logged appropriately"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = ["owner/repo1"]
|
||||
provider.organizations = []
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.get_repo.side_effect = RateLimitExceededException(
|
||||
429, "Rate limit exceeded", None
|
||||
)
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.services.repository.repository_service.GithubService.__init__"
|
||||
):
|
||||
repository_service = Repository(provider)
|
||||
repository_service.clients = [mock_client]
|
||||
repository_service.provider = provider
|
||||
|
||||
with patch(
|
||||
"prowler.providers.github.lib.service.service.logger"
|
||||
) as mock_logger:
|
||||
# Rate limit errors should be caught and logged at the outer level
|
||||
repos = repository_service._list_repositories()
|
||||
|
||||
# Should be empty due to rate limit error
|
||||
assert len(repos) == 0
|
||||
# Should log rate limit error
|
||||
mock_logger.error.assert_called()
|
||||
assert "Rate limit exceeded" in str(mock_logger.error.call_args)
|
||||
|
||||
Reference in New Issue
Block a user