mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-23 20:04:27 +00:00
merge branch 'master' into prwlr-7751-github-app-authentication-incorrectly-handles-key-parameter-and-environment-variables
This commit is contained in:
commit
efbf641bb3
1054 files changed
+237818
-21061
No files matched your search
@@ -12,6 +12,7 @@ ACCOUNT_URL = "/user"
|
||||
PAT_TOKEN = "github-token"
|
||||
OAUTH_TOKEN = "oauth-token"
|
||||
APP_ID = "app-id"
|
||||
APP_NAME = "app-name"
|
||||
APP_KEY = "app-key"
|
||||
|
||||
|
||||
|
||||
@@ -27,6 +27,7 @@ from tests.providers.github.github_fixtures import (
|
||||
ACCOUNT_URL,
|
||||
APP_ID,
|
||||
APP_KEY,
|
||||
APP_NAME,
|
||||
OAUTH_TOKEN,
|
||||
PAT_TOKEN,
|
||||
)
|
||||
@@ -135,6 +136,8 @@ class TestGitHubProvider:
|
||||
"prowler.providers.github.github_provider.GithubProvider.setup_identity",
|
||||
return_value=GithubAppIdentityInfo(
|
||||
app_id=APP_ID,
|
||||
app_name=APP_NAME,
|
||||
installations=["test-org"],
|
||||
),
|
||||
),
|
||||
):
|
||||
@@ -147,7 +150,9 @@ class TestGitHubProvider:
|
||||
|
||||
assert provider._type == "github"
|
||||
assert provider.session == GithubSession(token="", id=APP_ID, key=APP_KEY)
|
||||
assert provider.identity == GithubAppIdentityInfo(app_id=APP_ID)
|
||||
assert provider.identity == GithubAppIdentityInfo(
|
||||
app_id=APP_ID, app_name=APP_NAME, installations=["test-org"]
|
||||
)
|
||||
assert provider._audit_config == {
|
||||
"inactive_not_archived_days_threshold": 180,
|
||||
}
|
||||
@@ -206,7 +211,9 @@ class TestGitHubProvider:
|
||||
),
|
||||
patch(
|
||||
"prowler.providers.github.github_provider.GithubProvider.setup_identity",
|
||||
return_value=GithubAppIdentityInfo(app_id=APP_ID),
|
||||
return_value=GithubAppIdentityInfo(
|
||||
app_id=APP_ID, app_name=APP_NAME, installations=["test-org"]
|
||||
),
|
||||
),
|
||||
):
|
||||
connection = GithubProvider.test_connection(
|
||||
|
||||
@@ -63,7 +63,12 @@ class Test_Organization_Scoping:
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_user = MagicMock()
|
||||
mock_user.get_orgs.return_value = [self.mock_org1, self.mock_org2]
|
||||
mock_orgs = MagicMock()
|
||||
mock_orgs.totalCount = 2
|
||||
mock_orgs.__iter__ = MagicMock(
|
||||
return_value=iter([self.mock_org1, self.mock_org2])
|
||||
)
|
||||
mock_user.get_orgs.return_value = mock_orgs
|
||||
mock_client.get_user.return_value = mock_user
|
||||
|
||||
with patch(
|
||||
@@ -81,6 +86,95 @@ class Test_Organization_Scoping:
|
||||
assert orgs[1].name == "test-org1"
|
||||
assert orgs[2].name == "test-org2"
|
||||
|
||||
def test_no_organizations_found_for_user(self):
|
||||
"""Test that when user has no organizations, empty dict is returned"""
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = []
|
||||
provider.organizations = []
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_user = MagicMock()
|
||||
mock_orgs = MagicMock()
|
||||
mock_orgs.totalCount = 0
|
||||
mock_user.get_orgs.return_value = mock_orgs
|
||||
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) == 0
|
||||
|
||||
def test_github_app_organization_scoping(self):
|
||||
"""Test GitHub App organization listing"""
|
||||
from prowler.providers.github.models import GithubAppIdentityInfo
|
||||
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = []
|
||||
provider.organizations = []
|
||||
# Set GitHub App identity
|
||||
provider.identity = GithubAppIdentityInfo(
|
||||
app_id="test-app-id",
|
||||
app_name="test-app",
|
||||
installations=["test-org1", "test-org2"],
|
||||
)
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_orgs = MagicMock()
|
||||
mock_orgs.totalCount = 2
|
||||
mock_orgs.__iter__ = MagicMock(
|
||||
return_value=iter([self.mock_org1, self.mock_org2])
|
||||
)
|
||||
mock_client.get_organizations.return_value = mock_orgs
|
||||
|
||||
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_github_app_no_organizations_found(self):
|
||||
"""Test GitHub App when no organizations are accessible"""
|
||||
from prowler.providers.github.models import GithubAppIdentityInfo
|
||||
|
||||
provider = set_mocked_github_provider()
|
||||
provider.repositories = []
|
||||
provider.organizations = []
|
||||
# Set GitHub App identity
|
||||
provider.identity = GithubAppIdentityInfo(
|
||||
app_id="test-app-id", app_name="test-app", installations=[]
|
||||
)
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_orgs = MagicMock()
|
||||
mock_orgs.totalCount = 0
|
||||
mock_client.get_organizations.return_value = mock_orgs
|
||||
|
||||
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) == 0
|
||||
|
||||
def test_specific_organization_scoping(self):
|
||||
"""Test that only specified organizations are returned"""
|
||||
provider = set_mocked_github_provider()
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in new issue
Block a user