ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 13:01:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(api): add GitHub integration for sending findings as issues

Browse Source 
Add complete GitHub integration to Prowler that allows sending security
findings as GitHub Issues, working similarly to the existing Jira integration.

Features:
- GitHub API client with Personal Access Token (PAT) authentication
- Support for creating issues from Prowler findings
- Automatic repository discovery and validation
- Label support for issue categorization
- Rich markdown-formatted issues with detailed finding information
- Full API integration with CRUD operations
- Async task processing for bulk operations
- Connection testing and validation

Implementation:
- Add GitHub API client in prowler/lib/outputs/github/
- Add GitHub to Integration model choices
- Create serializers and validators for GitHub credentials and configuration
- Implement IntegrationGitHubViewSet for API endpoints
- Add async tasks and job processing for sending findings
- Add URL routing for /integrations/{id}/github/dispatches endpoint

API Changes:
- New integration type: "github"
- Credentials: token (required), owner (optional)
- Configuration: repositories dict, owner string
- Dispatch endpoint accepts repository and optional labels

Issue Format:
- Title: [Prowler] SEVERITY - CHECK_ID - RESOURCE_UID
- Body includes: finding details, risk description, recommendations,
  remediation code (CLI/Terraform/IaC), resource tags, compliance info

Security:
- GitHub PAT encrypted with Fernet before storage
- Repository access validated before dispatch
- All API calls use HTTPS
- Comprehensive error handling and logging

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Toni de la Fuente
2026-01-07 21:38:16 +01:00
parent df8d82345d
commit a8c244849f
13 changed files with 1070 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/backend/api/filters.py
+20
View File
@@ -957,6 +957,26 @@ class ProcessorFilter(FilterSet):
)
class IntegrationGitHubFindingsFilter(FilterSet):
# To be expanded as needed
finding_id = UUIDFilter(field_name="id", lookup_expr="exact")
finding_id__in = UUIDInFilter(field_name="id", lookup_expr="in")
class Meta:
model = Finding
fields = {}
def filter_queryset(self, queryset):
# Validate that there is at least one filter provided
if not self.data:
raise ValidationError(
{
"findings": "No finding filters provided. At least one filter is required."
}
)
return super().filter_queryset(queryset)
class IntegrationJiraFindingsFilter(FilterSet):
# To be expanded as needed
finding_id = UUIDFilter(field_name="id", lookup_expr="exact")
api/src/backend/api/models.py
+1
View File
@@ -1586,6 +1586,7 @@ class Integration(RowLevelSecurityProtectedModel):
class IntegrationChoices(models.TextChoices):
AMAZON_S3 = "amazon_s3", _("Amazon S3")
AWS_SECURITY_HUB = "aws_security_hub", _("AWS Security Hub")
GITHUB = "github", _("GitHub")
JIRA = "jira", _("JIRA")
SLACK = "slack", _("Slack")
api/src/backend/api/utils.py
+29 -2
View File
@@ -287,6 +287,20 @@ def prowler_integration_connection_test(integration: Integration) -> Connection:
integration.save()
return connection
elif integration.integration_type == Integration.IntegrationChoices.GITHUB:
from prowler.lib.outputs.github.github import GitHub
github_connection = GitHub.test_connection(
**integration.credentials,
raise_on_exception=False,
)
repositories = (
github_connection.repositories if github_connection.is_connected else {}
)
with rls_transaction(str(integration.tenant_id)):
integration.configuration["repositories"] = repositories
integration.save()
return github_connection
elif integration.integration_type == Integration.IntegrationChoices.JIRA:
jira_connection = Jira.test_connection(
**integration.credentials,
@@ -406,9 +420,22 @@ def get_findings_metadata_no_aggregations(tenant_id: str, filtered_queryset):
return serializer.data
def initialize_prowler_integration(integration: Integration) -> Jira:
def initialize_prowler_integration(integration: Integration):
# TODO Refactor other integrations to use this function
if integration.integration_type == Integration.IntegrationChoices.JIRA:
if integration.integration_type == Integration.IntegrationChoices.GITHUB:
from prowler.lib.outputs.github.exceptions import GitHubAuthenticationError
from prowler.lib.outputs.github.github import GitHub
try:
return GitHub(**integration.credentials)
except GitHubAuthenticationError as github_auth_error:
with rls_transaction(str(integration.tenant_id)):
integration.configuration["repositories"] = {}
integration.connected = False
integration.connection_last_checked_at = datetime.now(tz=timezone.utc)
integration.save()
raise github_auth_error
elif integration.integration_type == Integration.IntegrationChoices.JIRA:
try:
return Jira(**integration.credentials)
except JiraBasicAuthError as jira_auth_error:
api/src/backend/api/v1/serializer_utils/integrations.py
+41
View File
@@ -67,6 +67,14 @@ class SecurityHubConfigSerializer(BaseValidateSerializer):
resource_name = "integrations"
class GitHubConfigSerializer(BaseValidateSerializer):
owner = serializers.CharField(read_only=True)
repositories = serializers.DictField(read_only=True)
class Meta:
resource_name = "integrations"
class JiraConfigSerializer(BaseValidateSerializer):
domain = serializers.CharField(read_only=True)
issue_types = serializers.ListField(
@@ -93,6 +101,14 @@ class AWSCredentialSerializer(BaseValidateSerializer):
resource_name = "integrations"
class GitHubCredentialSerializer(BaseValidateSerializer):
token = serializers.CharField(required=True)
owner = serializers.CharField(required=False)
class Meta:
resource_name = "integrations"
class JiraCredentialSerializer(BaseValidateSerializer):
user_mail = serializers.EmailField(required=True)
api_token = serializers.CharField(required=True)
@@ -153,6 +169,23 @@ class JiraCredentialSerializer(BaseValidateSerializer):
},
},
},
{
"type": "object",
"title": "GitHub Credentials",
"properties": {
"token": {
"type": "string",
"description": "GitHub Personal Access Token (PAT) with repo scope. Can be generated from "
"GitHub Settings > Developer settings > Personal access tokens.",
},
"owner": {
"type": "string",
"description": "Repository owner (username or organization name). Optional - if not provided, "
"all accessible repositories will be available.",
},
},
"required": ["token"],
},
{
"type": "object",
"title": "JIRA Credentials",
@@ -221,6 +254,14 @@ class IntegrationCredentialField(serializers.JSONField):
},
},
},
{
"type": "object",
"title": "GitHub",
"description": "GitHub integration does not accept any configuration in the payload. Leave it as an "
"empty JSON object (`{}`).",
"properties": {},
"additionalProperties": False,
},
{
"type": "object",
"title": "JIRA",
api/src/backend/api/v1/serializers.py
+74 -1
View File
@@ -54,6 +54,8 @@ from api.models import (
from api.rls import Tenant
from api.v1.serializer_utils.integrations import (
AWSCredentialSerializer,
GitHubConfigSerializer,
GitHubCredentialSerializer,
IntegrationConfigField,
IntegrationCredentialField,
JiraConfigSerializer,
@@ -2432,6 +2434,28 @@ class BaseWriteIntegrationSerializer(BaseWriteSerializer):
)
config_serializer = SecurityHubConfigSerializer
credentials_serializers = [AWSCredentialSerializer]
elif integration_type == Integration.IntegrationChoices.GITHUB:
if providers:
raise serializers.ValidationError(
{
"providers": "Relationship field is not accepted. This integration applies to all providers."
}
)
if configuration:
raise serializers.ValidationError(
{
"configuration": "This integration does not support custom configuration."
}
)
config_serializer = GitHubConfigSerializer
# Create non-editable configuration for GitHub integration
configuration.update(
{
"repositories": {},
"owner": credentials.get("owner", ""),
}
)
credentials_serializers = [GitHubCredentialSerializer]
elif integration_type == Integration.IntegrationChoices.JIRA:
if providers:
raise serializers.ValidationError(
@@ -2519,7 +2543,11 @@ class IntegrationSerializer(RLSSerializer):
for provider in representation["providers"]
if provider["id"] in allowed_provider_ids
]
if instance.integration_type == Integration.IntegrationChoices.JIRA:
if instance.integration_type == Integration.IntegrationChoices.GITHUB:
representation["configuration"].update(
{"owner": instance.credentials.get("owner", "")}
)
elif instance.integration_type == Integration.IntegrationChoices.JIRA:
representation["configuration"].update(
{"domain": instance.credentials.get("domain")}
)
@@ -2666,6 +2694,51 @@ class IntegrationUpdateSerializer(BaseWriteIntegrationSerializer):
return representation
class IntegrationGitHubDispatchSerializer(BaseSerializerV1):
"""
Serializer for dispatching findings to GitHub integration.
"""
repository = serializers.CharField(required=True)
labels = serializers.ListField(
child=serializers.CharField(), required=False, default=list
)
class JSONAPIMeta:
resource_name = "integrations-github-dispatches"
def validate(self, attrs):
validated_attrs = super().validate(attrs)
integration_instance = Integration.objects.get(
id=self.context.get("integration_id")
)
if (
integration_instance.integration_type
!= Integration.IntegrationChoices.GITHUB
):
raise ValidationError(
{
"integration_type": "The given integration is not a GitHub integration"
}
)
if not integration_instance.enabled:
raise ValidationError(
{"integration": "The given integration is not enabled"}
)
repository = attrs.get("repository")
if repository not in integration_instance.configuration.get("repositories", {}):
raise ValidationError(
{
"repository": "The given repository is not available for this GitHub integration. Refresh the "
"connection if this is an error."
}
)
return validated_attrs
class IntegrationJiraDispatchSerializer(BaseSerializerV1):
"""
Serializer for dispatching findings to JIRA integration.
api/src/backend/api/v1/urls.py
+4
View File
@@ -12,6 +12,7 @@ from api.v1.views import (
FindingViewSet,
GithubSocialLoginView,
GoogleSocialLoginView,
IntegrationGitHubViewSet,
IntegrationJiraViewSet,
IntegrationViewSet,
InvitationAcceptViewSet,
@@ -94,6 +95,9 @@ users_router.register(r"memberships", MembershipViewSet, basename="user-membersh
integrations_router = routers.NestedSimpleRouter(
router, r"integrations", lookup="integration"
)
integrations_router.register(
r"github", IntegrationGitHubViewSet, basename="integration-github"
)
integrations_router.register(
r"jira", IntegrationJiraViewSet, basename="integration-jira"
)
api/src/backend/api/v1/views.py
+83
View File
@@ -83,6 +83,7 @@ from tasks.tasks import (
check_provider_connection_task,
delete_provider_task,
delete_tenant_task,
github_integration_task,
jira_integration_task,
mute_historical_findings_task,
perform_scan_task,
@@ -105,6 +106,7 @@ from api.filters import (
DailySeveritySummaryFilter,
FindingFilter,
IntegrationFilter,
IntegrationGitHubFindingsFilter,
IntegrationJiraFindingsFilter,
InvitationFilter,
LatestFindingFilter,
@@ -190,6 +192,7 @@ from api.v1.serializers import (
FindingSerializer,
FindingsSeverityOverTimeSerializer,
IntegrationCreateSerializer,
IntegrationGitHubDispatchSerializer,
IntegrationJiraDispatchSerializer,
IntegrationSerializer,
IntegrationUpdateSerializer,
@@ -5131,6 +5134,86 @@ class IntegrationViewSet(BaseRLSViewSet):
)
@extend_schema_view(
dispatches=extend_schema(
tags=["Integration"],
summary="Send findings to a GitHub integration",
description="Send a set of filtered findings to the given GitHub integration as issues. At least one finding "
"filter must be provided.",
responses={202: OpenApiResponse(response=TaskSerializer)},
filters=True,
)
)
class IntegrationGitHubViewSet(BaseRLSViewSet):
queryset = Finding.all_objects.all()
serializer_class = IntegrationGitHubDispatchSerializer
http_method_names = ["post"]
filter_backends = [CustomDjangoFilterBackend]
filterset_class = IntegrationGitHubFindingsFilter
# RBAC required permissions
required_permissions = [Permissions.MANAGE_INTEGRATIONS]
@extend_schema(exclude=True)
def create(self, request, *args, **kwargs):
raise MethodNotAllowed(method="POST")
def get_queryset(self):
tenant_id = self.request.tenant_id
user_roles = get_role(self.request.user)
if user_roles.unlimited_visibility:
# User has unlimited visibility, return all findings
queryset = Finding.all_objects.filter(tenant_id=tenant_id)
else:
# User lacks permission, filter findings based on provider groups associated with the role
queryset = Finding.all_objects.filter(
scan__provider__in=get_providers(user_roles)
)
return queryset
@action(detail=False, methods=["post"], url_name="dispatches")
def dispatches(self, request, integration_pk=None):
get_object_or_404(Integration, pk=integration_pk)
serializer = self.get_serializer(
data=request.data, context={"integration_id": integration_pk}
)
serializer.is_valid(raise_exception=True)
if self.filter_queryset(self.get_queryset()).count() == 0:
raise ValidationError(
{"findings": "No findings match the provided filters"}
)
finding_ids = [
str(finding_id)
for finding_id in self.filter_queryset(self.get_queryset()).values_list(
"id", flat=True
)
]
repository = serializer.validated_data["repository"]
labels = serializer.validated_data.get("labels", [])
with transaction.atomic():
task = github_integration_task.delay(
tenant_id=self.request.tenant_id,
integration_id=integration_pk,
repository=repository,
labels=labels,
finding_ids=finding_ids,
)
prowler_task = Task.objects.get(id=task.id)
serializer = TaskSerializer(prowler_task)
return Response(
data=serializer.data,
status=status.HTTP_202_ACCEPTED,
headers={
"Content-Location": reverse(
"task-detail", kwargs={"pk": prowler_task.id}
)
},
)
@extend_schema_view(
dispatches=extend_schema(
tags=["Integration"],
api/src/backend/tasks/jobs/integrations.py
+77 -2
View File
@@ -17,11 +17,11 @@ from prowler.lib.outputs.html.html import HTML
from prowler.lib.outputs.ocsf.ocsf import OCSF
from prowler.providers.aws.aws_provider import AwsProvider
from prowler.providers.aws.lib.s3.s3 import S3
from prowler.providers.aws.lib.security_hub.security_hub import SecurityHub
from prowler.providers.common.models import Connection
from prowler.providers.aws.lib.security_hub.exceptions.exceptions import (
SecurityHubNoEnabledRegionsError,
)
from prowler.providers.aws.lib.security_hub.security_hub import SecurityHub
from prowler.providers.common.models import Connection
logger = get_task_logger(__name__)
@@ -436,6 +436,81 @@ def upload_security_hub_integration(
return False
def send_findings_to_github(
tenant_id: str,
integration_id: str,
repository: str,
labels: list[str],
finding_ids: list[str],
):
with rls_transaction(tenant_id):
integration = Integration.objects.get(id=integration_id)
github_integration = initialize_prowler_integration(integration)
num_issues_created = 0
for finding_id in finding_ids:
with rls_transaction(tenant_id):
finding_instance = (
Finding.all_objects.select_related("scan__provider")
.prefetch_related("resources")
.get(id=finding_id)
)
# Extract resource information
resource = (
finding_instance.resources.first()
if finding_instance.resources.exists()
else None
)
resource_uid = resource.uid if resource else ""
resource_name = resource.name if resource else ""
resource_tags = {}
if resource and hasattr(resource, "tags"):
resource_tags = resource.get_tags(tenant_id)
# Get region
region = resource.region if resource and resource.region else ""
# Extract remediation information from check_metadata
check_metadata = finding_instance.check_metadata
remediation = check_metadata.get("remediation", {})
recommendation = remediation.get("recommendation", {})
remediation_code = remediation.get("code", {})
# Send the individual finding to GitHub
result = github_integration.send_finding(
check_id=finding_instance.check_id,
check_title=check_metadata.get("checktitle", ""),
severity=finding_instance.severity,
status=finding_instance.status,
status_extended=finding_instance.status_extended or "",
provider=finding_instance.scan.provider.provider,
region=region,
resource_uid=resource_uid,
resource_name=resource_name,
risk=check_metadata.get("risk", ""),
recommendation_text=recommendation.get("text", ""),
recommendation_url=recommendation.get("url", ""),
remediation_code_native_iac=remediation_code.get("nativeiac", ""),
remediation_code_terraform=remediation_code.get("terraform", ""),
remediation_code_cli=remediation_code.get("cli", ""),
remediation_code_other=remediation_code.get("other", ""),
resource_tags=resource_tags,
compliance=finding_instance.compliance or {},
repository=repository,
issue_labels=labels,
)
if result:
num_issues_created += 1
else:
logger.error(f"Failed to send finding {finding_id} to GitHub")
return {
"created_count": num_issues_created,
"failed_count": len(finding_ids) - num_issues_created,
}
def send_findings_to_jira(
tenant_id: str,
integration_id: str,
api/src/backend/tasks/tasks.py
+18
View File
@@ -28,6 +28,7 @@ from tasks.jobs.export import (
_upload_to_s3,
)
from tasks.jobs.integrations import (
send_findings_to_github,
send_findings_to_jira,
upload_s3_integration,
upload_security_hub_integration,
@@ -791,6 +792,23 @@ def security_hub_integration_task(
return upload_security_hub_integration(tenant_id, provider_id, scan_id)
@shared_task(
base=RLSTask,
name="integration-github",
queue="integrations",
)
def github_integration_task(
tenant_id: str,
integration_id: str,
repository: str,
labels: list[str],
finding_ids: list[str],
):
return send_findings_to_github(
tenant_id, integration_id, repository, labels, finding_ids
)
@shared_task(
base=RLSTask,
name="integration-jira",
prowler/lib/outputs/github/__init__.py
+5
View File
@@ -0,0 +1,5 @@
"""GitHub Integration Package."""
from prowler.lib.outputs.github.github import GitHub, GitHubConnection
__all__ = ["GitHub", "GitHubConnection"]
prowler/lib/outputs/github/exceptions/__init__.py
+35
View File
@@ -0,0 +1,35 @@
"""GitHub Integration Exceptions Package."""
from prowler.lib.outputs.github.exceptions.exceptions import (
GitHubAuthenticationError,
GitHubBaseException,
GitHubCreateIssueError,
GitHubCreateIssueResponseError,
GitHubGetLabelsError,
GitHubGetLabelsResponseError,
GitHubGetRepositoriesError,
GitHubGetRepositoriesResponseError,
GitHubInvalidParameterError,
GitHubInvalidRepositoryError,
GitHubNoRepositoriesError,
GitHubSendFindingsResponseError,
GitHubTestConnectionError,
GitHubTokenError,
)
__all__ = [
"GitHubAuthenticationError",
"GitHubBaseException",
"GitHubCreateIssueError",
"GitHubCreateIssueResponseError",
"GitHubGetLabelsError",
"GitHubGetLabelsResponseError",
"GitHubGetRepositoriesError",
"GitHubGetRepositoriesResponseError",
"GitHubInvalidParameterError",
"GitHubInvalidRepositoryError",
"GitHubNoRepositoriesError",
"GitHubSendFindingsResponseError",
"GitHubTestConnectionError",
"GitHubTokenError",
]
prowler/lib/outputs/github/exceptions/exceptions.py
+57
View File
@@ -0,0 +1,57 @@
"""GitHub Integration Exceptions."""
class GitHubBaseException(Exception):
"""Base exception for all GitHub integration errors."""
class GitHubAuthenticationError(GitHubBaseException):
"""Exception raised when GitHub authentication fails."""
class GitHubTokenError(GitHubBaseException):
"""Exception raised when GitHub token is invalid or missing."""
class GitHubGetRepositoriesError(GitHubBaseException):
"""Exception raised when fetching repositories fails."""
class GitHubGetRepositoriesResponseError(GitHubBaseException):
"""Exception raised when the response from GitHub repositories API is invalid."""
class GitHubNoRepositoriesError(GitHubBaseException):
"""Exception raised when no repositories are found."""
class GitHubInvalidRepositoryError(GitHubBaseException):
"""Exception raised when an invalid repository is specified."""
class GitHubCreateIssueError(GitHubBaseException):
"""Exception raised when creating a GitHub issue fails."""
class GitHubCreateIssueResponseError(GitHubBaseException):
"""Exception raised when the response from GitHub create issue API is invalid."""
class GitHubTestConnectionError(GitHubBaseException):
"""Exception raised when testing the connection to GitHub fails."""
class GitHubInvalidParameterError(GitHubBaseException):
"""Exception raised when an invalid parameter is provided."""
class GitHubSendFindingsResponseError(GitHubBaseException):
"""Exception raised when sending findings to GitHub fails."""
class GitHubGetLabelsError(GitHubBaseException):
"""Exception raised when fetching repository labels fails."""
class GitHubGetLabelsResponseError(GitHubBaseException):
"""Exception raised when the response from GitHub labels API is invalid."""
prowler/lib/outputs/github/github.py
+626
View File
@@ -0,0 +1,626 @@
"""GitHub Integration Module."""
import os
from dataclasses import dataclass
from typing import Dict, List
import requests
from prowler.lib.logger import logger
from prowler.lib.outputs.github.exceptions.exceptions import (
GitHubAuthenticationError,
GitHubGetLabelsError,
GitHubGetLabelsResponseError,
GitHubGetRepositoriesError,
GitHubGetRepositoriesResponseError,
GitHubInvalidParameterError,
GitHubInvalidRepositoryError,
GitHubNoRepositoriesError,
GitHubTestConnectionError,
)
from prowler.providers.common.models import Connection
@dataclass
class GitHubConnection(Connection):
"""
Represents a GitHub connection object.
Attributes:
repositories (dict): Dictionary of repositories in GitHub.
"""
repositories: dict = None
class GitHub:
"""
GitHub class to interact with the GitHub API
This integration supports creating GitHub Issues from Prowler findings.
It uses Personal Access Token (PAT) authentication.
Attributes:
- _token: The Personal Access Token
- _owner: The repository owner (user or organization)
- _api_url: The GitHub API base URL (defaults to https://api.github.com)
Methods:
- __init__: Initialize the GitHub object
- test_connection: Test the connection to GitHub and return a Connection object
- get_repositories: Get the accessible repositories from GitHub
- get_repository_labels: Get the available labels for a repository
- send_finding: Send a finding to GitHub and create an issue
Raises:
- GitHubAuthenticationError: Failed to authenticate
- GitHubTokenError: Token is invalid or missing
- GitHubNoRepositoriesError: No repositories found
- GitHubGetRepositoriesError: Failed to get repositories
- GitHubGetRepositoriesResponseError: Failed to get repositories, response code did not match 200
- GitHubInvalidRepositoryError: The repository is invalid
- GitHubCreateIssueError: Failed to create an issue
- GitHubCreateIssueResponseError: Failed to create an issue, response code did not match 201
- GitHubTestConnectionError: Failed to test the connection
- GitHubInvalidParameterError: Invalid parameters provided
- GitHubGetLabelsError: Failed to get labels
- GitHubGetLabelsResponseError: Failed to get labels, response code did not match 200
Usage:
github = GitHub(
token="ghp_xxxxxxxxxxxx",
owner="myorg"
)
github.send_finding(
check_id="aws_ec2_instance_public_ip",
severity="high",
repository="myorg/myrepo",
...
)
"""
_token: str = None
_owner: str = None
_api_url: str = "https://api.github.com"
HEADER_TEMPLATE = {
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
}
def __init__(
self,
token: str = None,
owner: str = None,
api_url: str = None,
):
"""
Initialize the GitHub client.
Args:
token: GitHub Personal Access Token
owner: Repository owner (user or organization)
api_url: GitHub API base URL (defaults to https://api.github.com for GitHub.com,
use https://github.example.com/api/v3 for GitHub Enterprise)
"""
if not token:
raise GitHubInvalidParameterError(
message="GitHub token is required",
file=os.path.basename(__file__),
)
self._token = token
self._owner = owner
if api_url:
self._api_url = api_url.rstrip("/")
# Test authentication
try:
self._authenticate()
except Exception as e:
raise GitHubAuthenticationError(
message=f"Failed to authenticate with GitHub: {str(e)}",
file=os.path.basename(__file__),
)
def _get_headers(self) -> Dict:
"""Get the headers for GitHub API requests."""
headers = self.HEADER_TEMPLATE.copy()
headers["Authorization"] = f"Bearer {self._token}"
return headers
def _authenticate(self) -> bool:
"""
Authenticate with GitHub by testing the token.
Returns:
True if authentication successful
Raises:
GitHubAuthenticationError: If authentication fails
"""
try:
response = requests.get(
f"{self._api_url}/user",
headers=self._get_headers(),
timeout=10,
)
if response.status_code == 200:
return True
elif response.status_code == 401:
raise GitHubAuthenticationError(
message="Invalid or expired GitHub token",
file=os.path.basename(__file__),
)
else:
raise GitHubAuthenticationError(
message=f"GitHub authentication failed with status {response.status_code}",
file=os.path.basename(__file__),
)
except requests.exceptions.RequestException as e:
raise GitHubAuthenticationError(
message=f"Failed to connect to GitHub: {str(e)}",
file=os.path.basename(__file__),
)
@staticmethod
def test_connection(
token: str = None,
owner: str = None,
api_url: str = None,
raise_on_exception: bool = True,
) -> GitHubConnection:
"""
Test the connection to GitHub.
Args:
token: GitHub Personal Access Token
owner: Repository owner (optional)
api_url: GitHub API base URL (optional)
raise_on_exception: Whether to raise exceptions or return error in Connection object
Returns:
GitHubConnection object with connection status and repositories
"""
try:
github = GitHub(token=token, owner=owner, api_url=api_url)
repositories = github.get_repositories()
return GitHubConnection(
is_connected=True,
error=None,
repositories=repositories,
)
except Exception as e:
logger.error(f"GitHub connection test failed: {str(e)}")
if raise_on_exception:
raise GitHubTestConnectionError(
message=f"Failed to test GitHub connection: {str(e)}",
file=os.path.basename(__file__),
)
return GitHubConnection(
is_connected=False,
error=str(e),
repositories={},
)
def get_repositories(self) -> Dict[str, str]:
"""
Get accessible repositories from GitHub.
Returns:
Dictionary with repository full names as keys and names as values
Example: {"owner/repo1": "repo1", "owner/repo2": "repo2"}
Raises:
GitHubGetRepositoriesError: If getting repositories fails
GitHubGetRepositoriesResponseError: If response is invalid
GitHubNoRepositoriesError: If no repositories found
"""
try:
repositories = {}
page = 1
per_page = 100
while True:
# Get repositories for the authenticated user
response = requests.get(
f"{self._api_url}/user/repos",
headers=self._get_headers(),
params={
"per_page": per_page,
"page": page,
"sort": "updated",
"affiliation": "owner,collaborator,organization_member",
},
timeout=10,
)
if response.status_code != 200:
raise GitHubGetRepositoriesResponseError(
message=f"Failed to get repositories: {response.status_code} - {response.text}",
file=os.path.basename(__file__),
)
repos = response.json()
if not repos:
break
for repo in repos:
full_name = repo.get("full_name")
name = repo.get("name")
if full_name and name:
repositories[full_name] = name
# Check if there are more pages
if len(repos) < per_page:
break
page += 1
if not repositories:
raise GitHubNoRepositoriesError(
message="No repositories found for the authenticated user",
file=os.path.basename(__file__),
)
return repositories
except GitHubNoRepositoriesError:
raise
except GitHubGetRepositoriesResponseError:
raise
except Exception as e:
raise GitHubGetRepositoriesError(
message=f"Failed to get repositories: {str(e)}",
file=os.path.basename(__file__),
)
def get_repository_labels(self, repository: str) -> List[str]:
"""
Get available labels for a repository.
Args:
repository: Repository full name (e.g., "owner/repo")
Returns:
List of label names
Raises:
GitHubGetLabelsError: If getting labels fails
GitHubGetLabelsResponseError: If response is invalid
"""
try:
response = requests.get(
f"{self._api_url}/repos/{repository}/labels",
headers=self._get_headers(),
params={"per_page": 100},
timeout=10,
)
if response.status_code != 200:
raise GitHubGetLabelsResponseError(
message=f"Failed to get labels: {response.status_code} - {response.text}",
file=os.path.basename(__file__),
)
labels = response.json()
return [label.get("name") for label in labels if label.get("name")]
except GitHubGetLabelsResponseError:
raise
except Exception as e:
raise GitHubGetLabelsError(
message=f"Failed to get repository labels: {str(e)}",
file=os.path.basename(__file__),
)
@staticmethod
def _get_severity_label(severity: str) -> str:
"""Get a severity label with color indicator."""
severity_lower = severity.lower()
emoji_map = {
"critical": "🔴",
"high": "🟠",
"medium": "🟡",
"low": "🟢",
"informational": "🔵",
}
emoji = emoji_map.get(severity_lower, "⚪")
return f"{emoji} {severity.upper()}"
@staticmethod
def _get_status_label(status: str) -> str:
"""Get a status label with indicator."""
status_lower = status.lower()
if "fail" in status_lower:
return "❌ FAIL"
elif "pass" in status_lower:
return "✅ PASS"
else:
return f"ℹ️ {status.upper()}"
def _build_issue_body(
self,
check_id: str = "",
check_title: str = "",
severity: str = "",
status: str = "",
status_extended: str = "",
provider: str = "",
region: str = "",
resource_uid: str = "",
resource_name: str = "",
risk: str = "",
recommendation_text: str = "",
recommendation_url: str = "",
remediation_code_native_iac: str = "",
remediation_code_terraform: str = "",
remediation_code_cli: str = "",
remediation_code_other: str = "",
resource_tags: dict = None,
compliance: dict = None,
finding_url: str = "",
tenant_info: str = "",
) -> str:
"""
Build the markdown body for the GitHub issue.
GitHub natively supports markdown, so we can use standard markdown formatting.
"""
body_parts = []
# Header with severity and status
body_parts.append("## Prowler Security Finding\n")
# Metadata table
body_parts.append("### Finding Details\n")
body_parts.append("| Field | Value |")
body_parts.append("|-------|-------|")
if check_id:
body_parts.append(f"| **Check ID** | `{check_id}` |")
if check_title:
body_parts.append(f"| **Check Title** | {check_title} |")
if severity:
body_parts.append(
f"| **Severity** | {self._get_severity_label(severity)} |"
)
if status:
body_parts.append(f"| **Status** | {self._get_status_label(status)} |")
if status_extended:
body_parts.append(f"| **Status Details** | {status_extended} |")
if provider:
body_parts.append(f"| **Provider** | {provider.upper()} |")
if region:
body_parts.append(f"| **Region** | {region} |")
if resource_uid:
body_parts.append(f"| **Resource UID** | `{resource_uid}` |")
if resource_name:
body_parts.append(f"| **Resource Name** | {resource_name} |")
if tenant_info:
body_parts.append(f"| **Tenant** | {tenant_info} |")
body_parts.append("")
# Risk description
if risk:
body_parts.append("### Risk\n")
body_parts.append(risk)
body_parts.append("")
# Recommendation
if recommendation_text or recommendation_url:
body_parts.append("### Recommendation\n")
if recommendation_text:
body_parts.append(recommendation_text)
if recommendation_url:
body_parts.append(f"\n[View Recommendation]({recommendation_url})")
body_parts.append("")
# Remediation code
if any(
[
remediation_code_native_iac,
remediation_code_terraform,
remediation_code_cli,
remediation_code_other,
]
):
body_parts.append("### Remediation\n")
if remediation_code_cli:
body_parts.append("#### CLI")
body_parts.append("```bash")
body_parts.append(remediation_code_cli.strip())
body_parts.append("```\n")
if remediation_code_terraform:
body_parts.append("#### Terraform")
body_parts.append("```hcl")
body_parts.append(remediation_code_terraform.strip())
body_parts.append("```\n")
if remediation_code_native_iac:
body_parts.append("#### Native IaC")
body_parts.append("```yaml")
body_parts.append(remediation_code_native_iac.strip())
body_parts.append("```\n")
if remediation_code_other:
body_parts.append("#### Other")
body_parts.append("```")
body_parts.append(remediation_code_other.strip())
body_parts.append("```\n")
# Resource tags
if resource_tags:
body_parts.append("### Resource Tags\n")
for key, value in resource_tags.items():
body_parts.append(f"- **{key}**: {value}")
body_parts.append("")
# Compliance
if compliance:
body_parts.append("### Compliance Frameworks\n")
for framework, requirements in compliance.items():
if requirements:
body_parts.append(f"- **{framework}**: {', '.join(requirements)}")
body_parts.append("")
# Finding URL
if finding_url:
body_parts.append(f"[View Finding in Prowler]({finding_url})\n")
# Footer
body_parts.append("---")
body_parts.append("*This issue was automatically created by Prowler*")
return "\n".join(body_parts)
def send_finding(
self,
check_id: str = "",
check_title: str = "",
severity: str = "",
status: str = "",
status_extended: str = "",
provider: str = "",
region: str = "",
resource_uid: str = "",
resource_name: str = "",
risk: str = "",
recommendation_text: str = "",
recommendation_url: str = "",
remediation_code_native_iac: str = "",
remediation_code_terraform: str = "",
remediation_code_cli: str = "",
remediation_code_other: str = "",
resource_tags: dict = None,
compliance: dict = None,
repository: str = "",
issue_labels: list = None,
finding_url: str = "",
tenant_info: str = "",
) -> bool:
"""
Send a finding to GitHub as an issue.
Args:
check_id: The check ID
check_title: The check title
severity: The severity level
status: The status
status_extended: Extended status information
provider: The cloud provider
region: The region
resource_uid: The resource UID
resource_name: The resource name
risk: Risk description
recommendation_text: Recommendation text
recommendation_url: Recommendation URL
remediation_code_native_iac: Native IaC remediation code
remediation_code_terraform: Terraform remediation code
remediation_code_cli: CLI remediation code
remediation_code_other: Other remediation code
resource_tags: Resource tags dictionary
compliance: Compliance frameworks dictionary
repository: Repository full name (e.g., "owner/repo")
issue_labels: List of label names to apply
finding_url: URL to the finding in Prowler
tenant_info: Tenant information
Returns:
True if the issue was created successfully, False otherwise
Raises:
GitHubInvalidRepositoryError: If repository is invalid
GitHubCreateIssueError: If issue creation fails
"""
try:
if not repository:
raise GitHubInvalidParameterError(
message="Repository is required",
file=os.path.basename(__file__),
)
# Validate repository exists
repositories = self.get_repositories()
if repository not in repositories:
raise GitHubInvalidRepositoryError(
message=f"Repository '{repository}' not found or not accessible",
file=os.path.basename(__file__),
)
# Build issue title
title_parts = ["[Prowler]"]
if severity:
title_parts.append(severity.upper())
if check_id:
title_parts.append(check_id)
if resource_uid:
title_parts.append(resource_uid)
title = " - ".join(title_parts[1:])
title = f"{title_parts[0]} {title}"
# Build issue body
body = self._build_issue_body(
check_id=check_id,
check_title=check_title,
severity=severity,
status=status,
status_extended=status_extended,
provider=provider,
region=region,
resource_uid=resource_uid,
resource_name=resource_name,
risk=risk,
recommendation_text=recommendation_text,
recommendation_url=recommendation_url,
remediation_code_native_iac=remediation_code_native_iac,
remediation_code_terraform=remediation_code_terraform,
remediation_code_cli=remediation_code_cli,
remediation_code_other=remediation_code_other,
resource_tags=resource_tags or {},
compliance=compliance or {},
finding_url=finding_url,
tenant_info=tenant_info,
)
# Build payload
payload = {
"title": title,
"body": body,
}
if issue_labels:
payload["labels"] = issue_labels
# Create issue
response = requests.post(
f"{self._api_url}/repos/{repository}/issues",
headers=self._get_headers(),
json=payload,
timeout=10,
)
if response.status_code != 201:
try:
response_json = response.json()
error_message = response_json.get("message", response.text)
except (ValueError, requests.exceptions.JSONDecodeError):
error_message = response.text
logger.error(
f"Failed to create GitHub issue: {response.status_code} - {error_message}"
)
return False
response_json = response.json()
issue_url = response_json.get("html_url", "")
logger.info(f"GitHub issue created successfully: {issue_url}")
return True
except GitHubInvalidRepositoryError as e:
logger.error(f"Invalid repository: {str(e)}")
return False
except Exception as e:
logger.error(f"Failed to send finding to GitHub: {str(e)}")
return False