diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index 8883f6b491..bb81edca20 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -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) diff --git a/prowler/providers/common/provider.py b/prowler/providers/common/provider.py index 47f8d6c4d8..a78c191711 100644 --- a/prowler/providers/common/provider.py +++ b/prowler/providers/common/provider.py @@ -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( diff --git a/prowler/providers/github/github_provider.py b/prowler/providers/github/github_provider.py index 30d3ed865a..d338933f20 100644 --- a/prowler/providers/github/github_provider.py +++ b/prowler/providers/github/github_provider.py @@ -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, diff --git a/prowler/providers/github/lib/arguments/arguments.py b/prowler/providers/github/lib/arguments/arguments.py index 9ba2415cdd..57aed863ab 100644 --- a/prowler/providers/github/lib/arguments/arguments.py +++ b/prowler/providers/github/lib/arguments/arguments.py @@ -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", + ) diff --git a/prowler/providers/github/lib/service/service.py b/prowler/providers/github/lib/service/service.py index 911a22b975..3ece64ee10 100644 --- a/prowler/providers/github/lib/service/service.py +++ b/prowler/providers/github/lib/service/service.py @@ -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" diff --git a/prowler/providers/github/services/organization/organization_service.py b/prowler/providers/github/services/organization/organization_service.py index 51654fc43f..cf57f2c799 100644 --- a/prowler/providers/github/services/organization/organization_service.py +++ b/prowler/providers/github/services/organization/organization_service.py @@ -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""" diff --git a/prowler/providers/github/services/repository/repository_service.py b/prowler/providers/github/services/repository/repository_service.py index dee6f35900..dbec869cb2 100644 --- a/prowler/providers/github/services/repository/repository_service.py +++ b/prowler/providers/github/services/repository/repository_service.py @@ -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""" diff --git a/tests/providers/github/github_provider_test.py b/tests/providers/github/github_provider_test.py index 88ed4dcfdb..a3c74bca86 100644 --- a/tests/providers/github/github_provider_test.py +++ b/tests/providers/github/github_provider_test.py @@ -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 == [] diff --git a/tests/providers/github/lib/arguments/arguments_test.py b/tests/providers/github/lib/arguments/arguments_test.py new file mode 100644 index 0000000000..fd2f813b6e --- /dev/null +++ b/tests/providers/github/lib/arguments/arguments_test.py @@ -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 == [] diff --git a/tests/providers/github/services/organization/organization_service_test.py b/tests/providers/github/services/organization/organization_service_test.py index 7720841698..a51373b8c7 100644 --- a/tests/providers/github/services/organization/organization_service_test.py +++ b/tests/providers/github/services/organization/organization_service_test.py @@ -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) diff --git a/tests/providers/github/services/repository/repository_service_test.py b/tests/providers/github/services/repository/repository_service_test.py index 487ac41e7c..ada89ee072 100644 --- a/tests/providers/github/services/repository/repository_service_test.py +++ b/tests/providers/github/services/repository/repository_service_test.py @@ -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)