update_ghas_secret_alerts_from_gitguardian_data.py

# Copyright 2025 Cisco Systems, Inc. and its affiliates
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# SPDX-License-Identifier: Apache-2.0

"""Closes GitHub Advanced Security secret scanning alerts if a matching secret is found
in a GitGuardian export."""
import csv
import os
import json
import io
import re
import logging
import argparse
from typing import Dict, cast
import datetime

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

import gitguardian_mappers
import crypto_utils

LOGGER = logging.getLogger(__name__)

# Configure requests session with retry strategy
def create_requests_session():
    """Create a requests session with retry strategy for robustness"""
    session = requests.Session()
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE", "PATCH"]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session

# Global session for reuse
requests_session = create_requests_session()

# Parsed from the full list here, these are queried separately from the rest
# https://docs.github.com/en/enterprise-cloud@latest/code-security/secret-scanning/introduction/supported-secret-scanning-patterns
GENERIC_SECRET_ALERT_TYPES = [
    "ec_private_key",
    "http_basic_authentication_header",
    "http_bearer_authentication_header",
    "mongodb_connection_string",
    "mysql_connection_string",
    "openssh_private_key",
    "pgp_private_key",
    "postgres_connection_string",
    "rsa_private_key",
    "password"
]

# Global set for tracking the number of JSON decode errors
json_decode_errors = set()


def get_ghas_headers() -> dict:
    """
    Returns GitHub API headers with authorization token.
    GITHUB_TOKEN needs to be able to write to all secret scanning alerts within the org.
    """
    github_token = os.getenv('GITHUB_TOKEN')
    if not github_token:
        raise EnvironmentError('GITHUB_TOKEN environment variable not set')
    
    return {
        'Authorization': f'Bearer {github_token}',
        'Accept': 'application/vnd.github+json',
        'X-GitHub-Api-Version': '2022-11-28'
    }

def get_gitguardian_headers() -> dict:
    """
    Returns GitGuardian API headers with authorization token.
    GITGUARDIAN_TOKEN needs Members:read, Incidents:read permissions.
    """
    gitguardian_token = os.getenv('GITGUARDIAN_TOKEN')
    if not gitguardian_token:
        raise EnvironmentError('GITGUARDIAN_TOKEN environment variable not set')
    
    return {
        "Authorization": f"Bearer {gitguardian_token}",
        "Content-Type": "application/json"
    }

class GitGuardianIncidentRow:
    """Represents a row in a GitGuardian CSV export file containing the following columns:
           incident_id	created_at	secret_hash	detector_name	matches	occurrences_count	assignees	resolved_at	ignored_at	gitguardian_url	severity	validity	status	ignore_reason	secret_revoked	tags
    """
    def __init__(self, row: dict, line_num: int, debug_log_secrets: bool) -> None:
        self.debug_log_secrets = debug_log_secrets
        self.detector_name = self._retrieve_string_row(row, 'detector_name', line_num)
        self.gitguardian_url = self._retrieve_string_row(row, 'gitguardian_url', line_num)
        self.status = self._retrieve_string_row(row, 'status', line_num)
        self.ignore_reason = self._retrieve_string_row(row, 'ignore_reason', line_num, optional=True)
        self.secret_revoked = self._retrieve_string_row(row, 'secret_revoked', line_num)
        self.ignored_at = self._retrieve_string_row(row, 'ignored_at', line_num, optional=True)
        self.resolved_at = self._retrieve_string_row(row, 'resolved_at', line_num, optional=True)
        self.incident_id = int( cast(str, row.get('incident_id')) )

        matches = self._retrieve_string_row(row, 'matches', line_num)
        self.matches = self._parse_matches(matches)

        self.cached_secret: str | None = None

    def _retrieve_string_row(self, row: dict, key: str, line_num: int, optional: bool = False) -> str:
        if optional:
            x = row.get(key, '')
        else:
            x = row.get(key)
            if not x:
                raise ValueError(f'{key} is missing from the row at line {line_num}')
        return cast(str, x)

    def store_cached_secret(self, secret: str) -> None:
        """Store the GitGuardian representation of the secret for faster calculations"""
        self.cached_secret = secret

    def get_cached_secret(self) -> str | None:
        """Returns any previously cached GitGuardian representation of this secret or None
           if it hasn't been calculated yet."""
        return self.cached_secret

    @staticmethod
    def _fix_the_json(matches_str: str) -> str:
        """Fixes bad strings in the 'matches' column of GG export to be valid JSON"""
        # An example would be:
        # {'host': 'myhost', 'port': '5672', 'scheme': 'a', 'database': "%%2f'", 'password': 'asdf', 'username': 'asdf', 'connection_uri': "amqp://asdf@asdf:5672/%%2f'"}

        pattern = r"'([^']+)':\s*(?:'([^']*)'|\"([^\"]*)\")"

        result = {}
        for match in re.finditer(pattern, matches_str):
            key = match.group(1)
            value = match.group(2) if match.group(2) is not None else match.group(3)
            result[key] = value
        cleaned = json.dumps(result, indent=2)
        #if "\"-" in matches_str:
        #    print(f"matches_str: {matches_str}")
        #    print(f"cleaned: {cleaned}")
        return cleaned

    def _parse_matches(self, matches_str: str) -> dict:
        """
        Parse the 'matches' column from the CSV row into a dictionary.
        """
        try:
            # In order to get the "JSON" to parse, handle private keys separately
            # since there are a lot of issues with how they're stored in the CSV file
            if crypto_utils.is_any_private_key(matches_str):
                fixed = matches_str.replace("'", '"')
                # Match stuff like '-----BEGIN PRIVATE KEY-----\\n"+            + "MIIE...
                # or '-----BEGIN PRIVATE KEY-----\\n"+            b"MEc
                fixed = re.sub(r'"\s*[+-]\s*[b+-]?\s*"', '', fixed)
                fixed = fixed.replace("\\\\n", "")
                fixed = fixed.replace("\\\\r", "")
                fixed = fixed.replace("\" \"", "")
                return json.loads(fixed)
            fixed = GitGuardianIncidentRow._fix_the_json(matches_str)
            return json.loads(fixed)

        # If there are errors track the number of rows we couldn't parse
        except json.JSONDecodeError:
            if self.debug_log_secrets:
                LOGGER.debug(f"Failed to decode JSON from matches: {matches_str}")
                LOGGER.debug(f"Cleaned string: {fixed}")
            json_decode_errors.add(matches_str)
            return {}

class GitGuardianIncidentDictionary:
    """Contains GitGuardianIncidentRows organized by GitGuardian secret type"""

    def __init__(self) -> None:
        self.alerts: Dict[str, list[GitGuardianIncidentRow]] = {}

    def add_row(self, row: GitGuardianIncidentRow) -> None:
        """
        Add ignored or resolved+revoked rows to a dictionary organized by GitGuardian detector name
        """
        if (row.status == "IGNORED" and row.ignore_reason in ("false_positive", "test_credential", "low_risk")) or \
               (row.status == "RESOLVED" and row.secret_revoked == "TRUE"):

            detector_name = row.detector_name
            if detector_name not in self.alerts:
                self.alerts[detector_name] = []
            self.alerts[detector_name].append(row)
        else:
            LOGGER.debug(f"Not considering incident with status {row.status}, ignore_reason {row.ignore_reason}, secret_revoked {row.secret_revoked}")

    def get_rows(self, detector_names: list[str]) -> list[GitGuardianIncidentRow]:
        relevant_rows: list[GitGuardianIncidentRow] = []
        for d in detector_names:
            relevant_rows.extend(self.alerts.get(d, []))
        return relevant_rows


def load_gitguardian_incidents(incident_csv_file: io.TextIOWrapper, debug_log_secrets: bool) -> GitGuardianIncidentDictionary:
    """
    Load the dismissed alerts from the CSV file
    """
    LOGGER.info(f'Loading incidents from {incident_csv_file.name}')
    incident_rows = []
    reader = csv.DictReader(incident_csv_file)
    for row in reader:
        ggir = GitGuardianIncidentRow(row, line_num=reader.line_num, debug_log_secrets=debug_log_secrets)
        incident_rows.append(ggir)
    LOGGER.info(f'Loaded {len(incident_rows)} incidents')
    incident_csv_file.close() # close file handle

    # Create a dictionary to store the dismissed alerts
    gid = GitGuardianIncidentDictionary()
    for irow in incident_rows:
        # Add the row to the dictionary
        gid.add_row(irow)
    return gid

def fetch_all_secret_alerts(org: str, repo: str | None, alert_number: int, secret_type: str, include_closed_alerts: bool) -> list:
    """
    Fetch all secret scanning alerts, optionally returning a single alert or filtering
    on all secrets of a particular type.
    """
    if alert_number:
        if not repo:
            raise ValueError('alert_number can only be used with a repo')
        return fetch_single_open_secret_scanning_alert(org, repo, alert_number, include_closed_alerts)

    # Otherwise fetch all alerts
    # Limit to a specific secret type
    if secret_type:
        secret_types = [secret_type]
        alerts = fetch_secret_scanning_alerts(org, repo, secret_types, include_closed_alerts)
    else:
        # Or first fetch the default pattern alerts
        default_pattern_alerts = fetch_secret_scanning_alerts(org, repo, None, include_closed_alerts)
        # and then the generic alerts
        generic_pattern_alerts = fetch_secret_scanning_alerts(org, repo, GENERIC_SECRET_ALERT_TYPES, include_closed_alerts)
        alerts = default_pattern_alerts + generic_pattern_alerts
    LOGGER.info(f'Found {len(alerts)} open secret scanning alerts')
    return alerts

def fetch_single_open_secret_scanning_alert(org: str, repo: str, alert_number: int, include_closed_alerts: bool) -> list:
    org_repo_pair = f'{org}/{repo}'
    api_url = f'https://api.github.com/repos/{org_repo_pair}/secret-scanning/alerts/{alert_number}'
    try:
        response = requests_session.get(api_url, headers=get_ghas_headers(), timeout=30)
        if response.status_code != 200:
            raise Exception(f"Failed to fetch data from {api_url}. HTTP Status Code: {response.status_code}")
        response_alerts = response.json()
        if not include_closed_alerts and response_alerts.get('state') != 'open':
            raise Exception(f'Alert {alert_number} is already dismissed')
        return [response_alerts]
    except (requests.exceptions.ConnectionError, requests.exceptions.Timeout, requests.exceptions.RequestException):
        LOGGER.exception(f"Network error fetching alert {alert_number}")
        raise

def fetch_secret_scanning_alerts(org: str,
                                 repo: str | None,
                                 secret_types: list | None, 
                                 include_closed_alerts: bool) -> list:
    """
    Fetch secret scanning alerts from GitHub Actions Secret Scanning API.
    
    Args:
        org (str): The organization name.
        repo (str | None): The repository name. If None, then fetch all alerts for the organization.
        secret_types (list | None): GHAS secret types to fetch using the API, if None fetch all default pattern alerts
        include_closed_alerts (bool): If True, then fetch also closed alerts.
        
    Returns:
        list: A list of secret scanning alerts.
    """
    alerts = []
    page = 1

    if repo:
        org_repo_pair = f'{org}/{repo}'
        api_url = f'https://api.github.com/repos/{org_repo_pair}/secret-scanning/alerts'
    else:
        api_url = f'https://api.github.com/orgs/{org}/secret-scanning/alerts'

    while True:
        params = {
            'page': str(page),  # Ensure page is a string
            'per_page': '100'   # Ensure per_page is a string
        }
        if secret_types:
            secret_types_str = [str(secret) for secret in secret_types]
            params.update({'secret_type': ','.join(secret_types_str)})
        # Query only open alerts unless we want all
        if not include_closed_alerts:
            params['state'] = 'open'

        try:
            response = requests_session.get(api_url, params=params, headers=get_ghas_headers(), timeout=30)
            if response.status_code != 200:
                raise Exception(f'Error fetching alerts: HTTP Status Code {response.status_code} - {response.reason}')
            response_alerts = response.json()
        except (requests.exceptions.ConnectionError, requests.exceptions.Timeout, requests.exceptions.RequestException) as e:
            LOGGER.exception(f"Network error fetching alerts page # {page}")
            raise
        if not response_alerts:
            break
        alerts.extend(response_alerts)
        page += 1
    return alerts

def retrieve_incident_closer_id_from_gitguardian(gitguardian_api_url: str, incident_id: int) -> int:
    url = f"{gitguardian_api_url}/incidents/secrets/{incident_id}"

    logging.debug(f"Fetching incident details from GitGuardian API: {url}")
    try:
        response = requests_session.get(url, headers=get_gitguardian_headers(), timeout=30)
        if response.status_code != 200:
            raise Exception(f"Error fetching incident details: HTTP Status Code {response.status_code} - {response.reason}")

        json_response = response.json()
        status = json_response.get('status')
        incident_closer_id = None
        if status == 'IGNORED':
            incident_closer_id = json_response.get('ignorer_id')
        elif status == 'RESOLVED':
            incident_closer_id = json_response.get('resolver_id')
        else:
            raise Exception(f"Unexpected status '{status}' for incident ID {incident_id}")
        return incident_closer_id
    except (requests.exceptions.ConnectionError, requests.exceptions.Timeout, requests.exceptions.RequestException):
        LOGGER.exception(f"Network error fetching GitGuardian incident {incident_id}")
        raise

def retrieve_name_and_email_from_gitguardian(gitguardian_api_url: str, user_id: int) -> tuple[str, str]:
    url = f"{gitguardian_api_url}/members/{user_id}"
    try:
        response = requests_session.get(url, headers=get_gitguardian_headers(), timeout=30)
        if response.status_code != 200:
            raise Exception(f"Error fetching user details: HTTP Status Code {response.status_code} - {response.reason}")

        json_response = response.json()
        name = json_response.get('name')
        email = json_response.get('email')

        return (name, email)
    except (requests.exceptions.ConnectionError, requests.exceptions.Timeout, requests.exceptions.RequestException):
        LOGGER.exception(f"Network error fetching GitGuardian user {user_id}")
        raise

def parse_incident_close_datetime_from_gitguardian(incident_row: GitGuardianIncidentRow) -> datetime.datetime:
    if incident_row.status == 'IGNORED':
        close_datetime = incident_row.ignored_at
    elif incident_row.status == 'RESOLVED':
        close_datetime = incident_row.resolved_at
    else:
        raise Exception(f"Unexpected status '{incident_row.status}' for incident ID {incident_row.detector_name}")

    return datetime.datetime.fromisoformat(close_datetime)

def search_gitguardian_incidents_for_alert(alert_secret: str,
                                           ghas_secret_type: str,
                                           gitguardian_incident_dictionary: GitGuardianIncidentDictionary,
                                           debug_log_secrets: bool) -> GitGuardianIncidentRow | None:
    """Search for the secret within the GitGuardian Incident Dictionray.
       Return a GitGuardianIncidentRow if the secret was matched and the alert was resolved/ignored in GitGuardian.  None otherwise"""
    try:
        converters = gitguardian_mappers.retrieve_mapper(ghas_secret_type, debug_log_secrets)
    except ValueError:
        logging.warning(f'WARNING: No detector implemented for secret type {ghas_secret_type} skipping!')
        return None

    if debug_log_secrets:
        LOGGER.debug(f'Checking if secret {alert_secret} is dismissed in GitGuardian for type {ghas_secret_type}')

    # A list of the GitGuardian detector names
    gg_types = [converter.get_detector_name() for converter in converters]

    # Iterate over the rows dealing with those detector names
    for row in gitguardian_incident_dictionary.get_rows(gg_types):
        try:
            for converter in converters:
                present = converter.secret_present_in_row(alert_secret, row)
                if present:
                    LOGGER.info(f"GitGuardian detector {converter.get_detector_name()} found secret")
                    return row
            continue
        except Exception:
            LOGGER.exception(f"Error processing secret with converter {converter.get_detector_name()}, Skipping Additional Processing")
            if debug_log_secrets:
                LOGGER.debug(f"Raw secret value: {alert_secret}")
            raise
    return None

def write_match_csv_row_if_writer(match_csv_writer: csv.DictWriter | None,
                                  alert: dict,
                                  found: bool) -> None:
    if match_csv_writer:
        match_csv_writer.writerow({'alert_number': alert['number'],
                             'url': alert['html_url'],
                             'secret_type': alert['secret_type'],
                             'found': found})

def process_ghas_alert_find_gitguardian_incident(alert: dict,
                                                 match_csv_writer: csv.DictWriter | None,
                                                 gitguardian_incident_dictionary: GitGuardianIncidentDictionary,
                                                 debug_log_secrets: bool) -> GitGuardianIncidentRow | None:
    """
    Process a single alert using the secrets CSV file
    Return a GitGuardianIncidentRow if the alert has been resolved/ignored in GitGuardian, None if the secret could not be found or matched.
    """
    alert_secret = cast(str, alert.get('secret'))
    ghas_secret_type = cast(str, alert.get('secret_type'))

    found_incident = search_gitguardian_incidents_for_alert(alert_secret, ghas_secret_type, gitguardian_incident_dictionary, debug_log_secrets)
    if found_incident:
        LOGGER.info(f'Found matching secret incident in GitGuardian for GHAS Alert URL: {alert["html_url"]} type: {alert["secret_type"]}')
    else:
        LOGGER.info(f'Secret not found in GitGuardian for GHAS Alert URL: {alert["html_url"]} type: {alert["secret_type"]}')

    write_match_csv_row_if_writer(match_csv_writer, alert, found_incident is not None)
    return found_incident

def calculate_ghas_resolution(incident_row: GitGuardianIncidentRow) -> str:
    """
    Calculate the resolution for a GitHub Advanced Security alert based on the GitGuardian incident row.
    """
    status = incident_row.status
    ignore_reason = incident_row.ignore_reason
    secret_revoked = incident_row.secret_revoked

    if status == 'IGNORED':
        if ignore_reason == 'false_positive':
            return 'false_positive'
        elif ignore_reason == 'test_credential':
            return 'used_in_tests'
        elif ignore_reason == 'low_risk':
            return 'wont_fix'
        else:
            raise Exception(f'Unknown ignore reason {ignore_reason} for incident {incident_row.detector_name}')
    elif status == 'RESOLVED':
        if secret_revoked == 'TRUE':
            return 'revoked'
        else:
            raise Exception(f'Incident {incident_row.detector_name} is resolved but not revoked')
    else:
        raise Exception(f'Unknown status {status} for incident {incident_row.detector_name}')

def calculate_ghas_resolution_comment(gitguardian_api_url: str, incident_row: GitGuardianIncidentRow) -> tuple[str, str]:
    """
    Return the resolution comment to add to the GHAS Alert and the email of the person who closed the GitGuardian incident
    """
    closer_id = retrieve_incident_closer_id_from_gitguardian(gitguardian_api_url, incident_row.incident_id)
    name, email = retrieve_name_and_email_from_gitguardian(gitguardian_api_url, closer_id)
    close_datetime = parse_incident_close_datetime_from_gitguardian(incident_row)
    comment = f"Closed by {name} <{email}> on {close_datetime.strftime('%Y-%m-%d')} in GitGuardian ({incident_row.gitguardian_url})"
    return (comment, email)

def write_close_csv_row_if_writer(close_csv_writer: csv.DictWriter | None,
                                  incident_row: GitGuardianIncidentRow,
                                  alert: dict,
                                  ghas_resolution_state: str,
                                  email: str
                                  ) -> None:
    #close_fieldnames = ['ghas_url', 'ghas_secret_type', 'ghas_resolution_state', 'gitguardian_url', 'gitguardian_closer_email']
    if close_csv_writer:
        close_csv_writer.writerow({'ghas_url': alert['html_url'],
                                   'ghas_secret_type': alert['secret_type'],
                                   'ghas_resolution_state': ghas_resolution_state,
                                   'gitguardian_url': incident_row.gitguardian_url,
                                   'gitguardian_closer_email': email })

def close_ghas_alert(gitguardian_api_url: str,
                     incident_row: GitGuardianIncidentRow,
                     alert: dict,
                     dry_run: bool,
                     close_csv_writer: csv.DictWriter | None) -> bool:
    """
    Close a GHAS Alert that matched in GitGuardian
    Returns True if successful, False if failed
    """
    if alert.get('validity') == 'active':
        LOGGER.info(f"Ignoring active alert for alert URL: {alert['html_url']}")
        return True
    if alert.get('state') == 'resolved':
        LOGGER.info(f"Alert {alert['html_url']} is already closed")
        return True

    # GitGuardianIncidentDictionary.add_row only adds rows that should be closed

    try:
        (resolution_comment, email) = calculate_ghas_resolution_comment(gitguardian_api_url, incident_row)
        resolution = calculate_ghas_resolution(incident_row)

        params = {'state': 'resolved',
                  'resolution': resolution,
                  'resolution_comment': resolution_comment}
        if dry_run:
            LOGGER.info(f"DRY RUN: Would have closed alert {alert['html_url']} with params: {params}")
        else:
            api_url = cast(str, alert.get('url'))
            response = requests_session.patch(api_url, headers=get_ghas_headers(), json=params, timeout=30)
            if response.status_code != 200:
                LOGGER.error(f'Error dismissing alert {alert["html_url"]}: HTTP {response.status_code} - {response.text}')
                return False
            LOGGER.info(f"Resolved GHAS Alert: {alert['html_url']} with params: {params}")
        write_close_csv_row_if_writer(close_csv_writer, incident_row, alert, resolution, email)
        return True
    except (requests.exceptions.ConnectionError, requests.exceptions.Timeout, requests.exceptions.RequestException):
        LOGGER.exception(f"Network error closing alert {alert['html_url']}")
        return False
    except Exception:
        LOGGER.exception(f"Unexpected error processing alert {alert['html_url']}")
        return False


def parse_arguments():
    parser = argparse.ArgumentParser(description='A script that imports dismissed/resolved secret incidents from GitGuardian into GitHub Advanced Security')
    parser.add_argument('--org', required=True, help='GitHub organization name')
    parser.add_argument('--repo', required=False, help='GitHub repository name if a single repo should be checked.  If not specified, the entire org is reviewed')
    parser.add_argument('--gitguardian-csv-file', required=True, nargs='?', type=argparse.FileType('r', encoding='UTF-8'), help='GitGuardian incident CSV export file with secret info.  Only export dismissed or resolved findings.')
    parser.add_argument('--gitguardian-api-url', required=True, help='GitGuardian API URL like https://api.gitguardian.com/v1')
    parser.add_argument('--dismiss-alerts', action='store_true', help='The default behavior is to operate in a dry run.  Passing this flag dismisses alerts in GHAS if the secret is found in the GitGuardian CSV file (excluding validity=active secrets)')
    parser.add_argument('--matching-info-csv-output', required=False, nargs='?', type=argparse.FileType('w', encoding='UTF-8'),
                        help='CSV output about which alerts matched')
    parser.add_argument('--close-info-csv-output', required=False, nargs='?', type=argparse.FileType('w', encoding='UTF-8'),
                        help='CSV output with information about the alerts being closed')
    # Some lesser used arguments
    parser.add_argument('-d,', '--debug', action='store_true', help='Enable debug logging')
    parser.add_argument('--debug-log-secrets', action='store_true', help='Enable debug logging of secrets (WARNING: THIS WILL OUTPUT SECRETS TO THE CONSOLE)')
    parser.add_argument('--alert', required=False, nargs='?', type=int, help='Specific secret alert number to process')
    parser.add_argument('--secret-type', required=False, nargs='?', help='Filter alerts by a specific secret type (e.g. aws_secret_access_key)')
    parser.add_argument('--include-closed-alerts', required=False, action='store_true', help="Query closed GHAS alerts (useful for testing)")
    return parser.parse_args()

def main():
    args = parse_arguments()
    logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO)

    # Check for required environment variables
    if not os.getenv('GITHUB_TOKEN'):
        LOGGER.error('GITHUB_TOKEN environment variable is not set')
        raise EnvironmentError('GITHUB_TOKEN environment variable is required. Please set it before running this script.')
    
    if not os.getenv('GITGUARDIAN_TOKEN'):
        LOGGER.error('GITGUARDIAN_TOKEN environment variable is not set')
        raise EnvironmentError('GITGUARDIAN_TOKEN environment variable is required. Please set it before running this script.')

    gitguardian_incident_dictionary = load_gitguardian_incidents(args.gitguardian_csv_file, args.debug_log_secrets)

    if args.alert and args.secret_type:
        raise ValueError("Cannot specify both --alert and --secret-type. Please choose one.")

    LOGGER.info(f'Fetching Secret Scanning Alerts...')
    alerts = fetch_all_secret_alerts(args.org, args.repo, args.alert, args.secret_type, args.include_closed_alerts)
    number_resolvable = 0

    if len(alerts) > 0:
        try:
            match_csv_writer = None
            if args.matching_info_csv_output:
                LOGGER.info(f'Writing CSV output on which secrets were matched to {args.matching_info_csv_output.name}')
                match_fieldnames = ['alert_number', 'url', 'secret_type', 'found']
                match_csv_writer = csv.DictWriter(args.matching_info_csv_output, fieldnames=match_fieldnames)
                match_csv_writer.writeheader()
            close_csv_writer = None
            if args.close_info_csv_output:
                LOGGER.info(f'Writing CSV output on which secrets are being closed to {args.close_info_csv_output.name}')
                close_fieldnames = ['ghas_url', 'ghas_secret_type', 'ghas_resolution_state', 'gitguardian_url', 'gitguardian_closer_email']
                close_csv_writer = csv.DictWriter(args.close_info_csv_output, fieldnames=close_fieldnames)
                close_csv_writer.writeheader()

            num_alerts_to_process = len(alerts)
            LOGGER.info(f'Number of alerts to process: {num_alerts_to_process}')
            count = 0
            # Process each alert
            successful_closures = 0
            failed_closures = 0
            for alert in alerts:
                try:
                    incident_row = process_ghas_alert_find_gitguardian_incident(alert, match_csv_writer, gitguardian_incident_dictionary, args.debug_log_secrets)
                    if incident_row:
                        number_resolvable += 1
                        dry_run = False if args.dismiss_alerts else True
                        success = close_ghas_alert(args.gitguardian_api_url, incident_row, alert, dry_run, close_csv_writer)
                        if success:
                            successful_closures += 1
                        else:
                            failed_closures += 1
                            LOGGER.error(f"Failed to close alert {alert['html_url']}, continuing with next alert")
                except Exception:
                    failed_closures += 1
                    LOGGER.exception(f"Error processing alert {alert.get('html_url', 'unknown')}")
                    LOGGER.info("Continuing with next alert...")

                count = count + 1
                if count % 10 == 0:
                    if args.dismiss_alerts:
                        LOGGER.info(f'Progress: {count}/{num_alerts_to_process} processed, {successful_closures} successful closures, {failed_closures} failures')
                    else:
                        LOGGER.info(f'Progress: {count}/{num_alerts_to_process} processed, {successful_closures} would be closed, {failed_closures} failures')

        finally:
            if match_csv_writer:
                args.matching_info_csv_output.close()  # close file handle
            if close_csv_writer:
                args.close_info_csv_output.close()  # close file handle

    # The data/code isn't perfect so output info about what couldn't be processed
    if len(gitguardian_mappers.base64_errors) > 0:
        LOGGER.info(f'There were {len(gitguardian_mappers.base64_errors)} secrets with Base64 decoding errors which were skipped.  Pass --debug --debug-log-secrets for details')
    if len(json_decode_errors) > 0:
        LOGGER.info(f'There were {len(json_decode_errors)} incidents with JSON decoding errors of GitGuardian data which were skipped.  Pass --debug --debug-log-secrets for details')
    # Output final statistics about what was done
    if 'successful_closures' in locals() and 'failed_closures' in locals():
        if args.dismiss_alerts:
            LOGGER.info(f'Processed {len(alerts)} alerts. Found {number_resolvable} resolvable alerts. Successfully closed {successful_closures}.  There were {failed_closures} errors.')
        else:
            LOGGER.info(f'DRY RUN: Processed {len(alerts)} alerts. Found {number_resolvable} resolvable alerts that would have been closed.  There were {failed_closures} errors.  No alerts were actually dismissed (use --dismiss-alerts to dismiss them).')
    else:
        LOGGER.info(f'Processed {len(alerts)} alerts. Found {number_resolvable} resolvable alerts')

if __name__ == '__main__':
    main()