diff --git a/docs/tutorials/github/getting-started-github.md b/docs/tutorials/github/getting-started-github.md index 3515b2ca23..2d3580410d 100644 --- a/docs/tutorials/github/getting-started-github.md +++ b/docs/tutorials/github/getting-started-github.md @@ -24,7 +24,56 @@ Personal Access Tokens provide the simplest GitHub authentication method and sup - Scroll down the left sidebar - Click "Developer settings" -3. **Generate New Token** +3. **Generate Fine-Grained Token** + - Click "Personal access tokens" + - Select "Fine-grained tokens" + - Click "Generate new token" + +4. **Configure Token Settings** + - **Token name**: Give your token a descriptive name (e.g., "Prowler Security Scanner") + - **Expiration**: Set an appropriate expiration date (recommended: 90 days or less) + - **Repository access**: Choose "All repositories" or "Only select repositories" based on your needs + + ???+ note "Public repositories" + Even if you select 'Only select repositories', the token will have access to the public repositories that you own or are a member of. + +5. **Configure Token Permissions** + To enable Prowler functionality, configure the following permissions: + + - **Repository permissions:** + - **Contents**: Read-only access + - **Metadata**: Read-only access + - **Pull requests**: Read-only access + - **Security advisories**: Read-only access + - **Statuses**: Read-only access + + - **Organization permissions:** + - **Members**: Read-only access + + - **Account permissions:** + - **Email addresses**: Read-only access + +6. **Copy and Store the Token** + - Copy the generated token immediately (GitHub displays tokens only once) + - Store tokens securely using environment variables + +![GitHub Personal Access Token Permissions](./img/github-pat-permissions.png) + +#### **Option 2: Create a Classic Personal Access Token (Not Recommended)** + +???+ warning "Security Risk" + Classic tokens provide broad permissions that may exceed what Prowler actually needs. Use fine-grained tokens instead for better security. + +1. **Navigate to GitHub Settings** + - Open [GitHub](https://github.com) and sign in + - Click the profile picture in the top right corner + - Select "Settings" from the dropdown menu + +2. **Access Developer Settings** + - Scroll down the left sidebar + - Click "Developer settings" + +3. **Generate Classic Token** - Click "Personal access tokens" - Select "Tokens (classic)" - Click "Generate new token" diff --git a/docs/tutorials/github/img/github-pat-permissions.png b/docs/tutorials/github/img/github-pat-permissions.png new file mode 100644 index 0000000000..d666c70baa Binary files /dev/null and b/docs/tutorials/github/img/github-pat-permissions.png differ diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index fd6a207cf5..aaf2cc47f5 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -21,6 +21,7 @@ All notable changes to the **Prowler SDK** are documented in this file. - Azure `app_http_logs_enabled` check false positives [(#8507)](https://github.com/prowler-cloud/prowler/pull/8507) - Azure `storage_geo_redundant_enabled` check false positives [(#8504)](https://github.com/prowler-cloud/prowler/pull/8504) - AWS `kafka_cluster_is_public` check false positives [(#8514)](https://github.com/prowler-cloud/prowler/pull/8514) +- List all accessible repositories in GitHub [(#8522)](https://github.com/prowler-cloud/prowler/pull/8522) - GitHub CIS 1.0 Compliance Reports [(#8519)](https://github.com/prowler-cloud/prowler/pull/8519) --- diff --git a/prowler/providers/github/github_provider.py b/prowler/providers/github/github_provider.py index d338933f20..4153b679da 100644 --- a/prowler/providers/github/github_provider.py +++ b/prowler/providers/github/github_provider.py @@ -133,6 +133,12 @@ class GithubProvider(Provider): """ logger.info("Instantiating GitHub Provider...") + # Mute GitHub library logs to reduce noise since it is already handled by the Prowler logger + import logging + + logging.getLogger("github").setLevel(logging.CRITICAL) + logging.getLogger("github.GithubRetry").setLevel(logging.CRITICAL) + # Set repositories and organizations for scoping self._repositories = repositories or [] self._organizations = organizations or [] diff --git a/prowler/providers/github/services/repository/repository_service.py b/prowler/providers/github/services/repository/repository_service.py index dbec869cb2..674383a641 100644 --- a/prowler/providers/github/services/repository/repository_service.py +++ b/prowler/providers/github/services/repository/repository_service.py @@ -2,6 +2,7 @@ from datetime import datetime from typing import Optional import github +import requests from pydantic.v1 import BaseModel from prowler.lib.logger import logger @@ -50,25 +51,57 @@ class Repository(GithubService): return True + def _get_accessible_repos_graphql(self) -> list[str]: + """ + Use the GitHub GraphQL API to list all repositories that the authentication token has access to. + This works with high-granularity (fine-grained) PATs. + """ + graphql_url = "https://api.github.com/graphql" + token = self.provider.session.token + headers = { + "Authorization": f"bearer {token}", + "Content-Type": "application/json", + } + query = """ + { + viewer { + repositories(first: 100, affiliations: [OWNER, ORGANIZATION_MEMBER]) { + nodes { + nameWithOwner + } + } + } + } + """ + + try: + response = requests.post( + graphql_url, json={"query": query}, headers=headers + ) + response.raise_for_status() + data = response.json() + + if "errors" in data: + logger.error(f"Error in GraphQL query: {data['errors']}") + return [] + + repo_nodes = ( + data.get("data", {}) + .get("viewer", {}) + .get("repositories", {}) + .get("nodes", []) + ) + return [repo["nameWithOwner"] for repo in repo_nodes] + + except requests.exceptions.RequestException as error: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + return [] + 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 = {} @@ -109,11 +142,38 @@ class Repository(GithubService): error, "processing organization", org_name ) else: - for repo in client.get_user().get_repos(): - self._process_repository(repo, repos) + logger.info( + "No repository or organization specified, discovering accessible repositories via GraphQL API..." + ) + accessible_repo_names = self._get_accessible_repos_graphql() + + if not accessible_repo_names: + logger.warning( + "Could not find any accessible repositories with the provided token." + ) + + for repo_name in accessible_repo_names: + try: + repo = client.get_repo(repo_name) + logger.info( + f"Processing repository found via GraphQL: {repo.full_name}" + ) + self._process_repository(repo, repos) + except Exception as error: + if hasattr(self, "_handle_github_api_error"): + self._handle_github_api_error( + error, + "accessing repository discovered via GraphQL", + repo_name, + ) + else: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + 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 + raise except github.GithubException as error: logger.error(f"GitHub API error while listing repositories: {error}") except Exception as error: @@ -124,158 +184,167 @@ class Repository(GithubService): 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 + 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: - 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}" - ) + codeowners_exists = False + delete_branch_on_merge = ( + repo.delete_branch_on_merge + if repo.delete_branch_on_merge is not None + else False + ) - 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" - ) + 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: - # 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 + 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: - 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}" + # If the branch is not found, it is not protected + if "404" in str(error): + logger.warning( + f"{repo.full_name}: {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) - dependabot_alerts_enabled = None + # 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"{repo.full_name}: {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"{repo.full_name}: {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + dependabot_alerts_enabled = None + except Exception as error: + logger.error( + f"{repo.full_name}: {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, + ) except Exception as error: logger.error( - f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + f"{repo.full_name}: {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""" name: str - protected: bool + protected: Optional[bool] default_branch: bool require_pull_request: Optional[bool] approval_count: Optional[int] diff --git a/tests/providers/github/services/repository/repository_service_test.py b/tests/providers/github/services/repository/repository_service_test.py index ada89ee072..20e5887d54 100644 --- a/tests/providers/github/services/repository/repository_service_test.py +++ b/tests/providers/github/services/repository/repository_service_test.py @@ -1,6 +1,7 @@ from datetime import datetime, timezone from unittest.mock import MagicMock, patch +import requests from github import GithubException, RateLimitExceededException from prowler.providers.github.services.repository.repository_service import ( @@ -40,6 +41,7 @@ def mock_list_repositories(_): archived=False, pushed_at=datetime.now(timezone.utc), delete_branch_on_merge=True, + dependabot_alerts_enabled=True, ), } @@ -110,6 +112,105 @@ class Test_Repository_FileExists: assert mock_logger.error.called +class Test_Repository_GraphQL: + 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 = False + self.mock_repo1.security_and_analysis = None + self.mock_repo1.get_contents.side_effect = [None, None, None] + self.mock_repo1.get_branch.side_effect = Exception("404 Not Found") + self.mock_repo1.get_dependabot_alerts.side_effect = Exception("403 Forbidden") + + def test_no_scoping_uses_graphql(self): + """Test that no scoping triggers the GraphQL discovery method successfully""" + provider = set_mocked_github_provider() + provider.repositories = [] + provider.organizations = [] + + with patch.object(Repository, "__init__", lambda x, y: None): + repository_service = Repository(provider) + mock_client = MagicMock() + repository_service.clients = [mock_client] + repository_service.provider = provider + + with patch.object( + repository_service, + "_get_accessible_repos_graphql", + return_value=["owner1/repo1"], + ) as mock_graphql_call: + mock_client.get_repo.return_value = self.mock_repo1 + + repos = repository_service._list_repositories() + + assert len(repos) == 1 + assert 1 in repos + assert repos[1].name == "repo1" + + mock_graphql_call.assert_called_once() + mock_client.get_repo.assert_called_once_with("owner1/repo1") + + def test_graphql_call_api_error(self): + """Test that an error during the GraphQL call is handled gracefully""" + provider = set_mocked_github_provider() + provider.repositories = [] + provider.organizations = [] + + with patch.object(Repository, "__init__", lambda x, y: None): + repository_service = Repository(provider) + repository_service.clients = [MagicMock()] + repository_service.provider = provider + + with patch( + "requests.post", + side_effect=requests.exceptions.RequestException("API Error"), + ): + with patch( + "prowler.providers.github.services.repository.repository_service.logger" + ) as mock_logger: + + repos = repository_service._list_repositories() + + assert len(repos) == 0 + mock_logger.error.assert_called_once() + + log_output = str(mock_logger.error.call_args) + assert "RequestException" in log_output + assert "API Error" in log_output + + def test_graphql_returns_empty_list(self): + """Test the case where GraphQL returns no repositories""" + provider = set_mocked_github_provider() + provider.repositories = [] + provider.organizations = [] + + with patch.object(Repository, "__init__", lambda x, y: None): + repository_service = Repository(provider) + repository_service.clients = [MagicMock()] + repository_service.provider = provider + + with patch.object( + repository_service, "_get_accessible_repos_graphql", return_value=[] + ): + with patch( + "prowler.providers.github.services.repository.repository_service.logger" + ) as mock_logger: + + repos = repository_service._list_repositories() + + assert len(repos) == 0 + mock_logger.warning.assert_called_with( + "Could not find any accessible repositories with the provided token." + ) + + class Test_Repository_Scoping: def setup_method(self): self.mock_repo1 = MagicMock() @@ -123,7 +224,7 @@ class Test_Repository_Scoping: 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_contents.side_effect = [None, None, None] self.mock_repo1.get_branch.side_effect = Exception("404 Not Found") self.mock_repo1.get_dependabot_alerts.side_effect = Exception("404 Not Found") @@ -138,200 +239,10 @@ class Test_Repository_Scoping: 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_contents.side_effect = [None, None, 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()