ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 12:31:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(github): list all accessible repositories (#8522)

Browse Source
This commit is contained in:
Sergio Garcia
2025-08-14 10:38:38 +02:00
committed by GitHub
parent 81478dfed3
commit dfdd45e4d0
6 changed files with 350 additions and 358 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/tutorials/github/getting-started-github.md
+15 -10
View File
@@ -37,26 +37,31 @@ Personal Access Tokens provide the simplest GitHub authentication method and sup
- **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 events**: Read-only access
- **Statuses**: Read-only access
- **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
- **Organization permissions:**
- **Members**: Read-only access
**Account permissions:**
- **Email addresses**: 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"
docs/tutorials/github/img/github-pat-permissions.png
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 89 KiB

prowler/CHANGELOG.md
+1
View File
@@ -25,6 +25,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)
---
prowler/providers/github/github_provider.py
+6
View File
@@ -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 []
prowler/providers/github/services/repository/repository_service.py
+225 -156
View File
@@ -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]
tests/providers/github/services/repository/repository_service_test.py
+103 -192
View File
@@ -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()