githush.py

import requests
import time
import re
import argparse
import os
import sqlite3
import json
import tempfile
from base64 import b64decode

# ANSI color codes
RED = '\033[91m'
GREEN = '\033[92m'
YELLOW = '\033[93m'
PURPLE = '\033[95m'
CYAN = '\033[96m'
RESET = '\033[0m'

# References:
    # Mackenzie Jackson: Miscreants Field Manual for Exploiting Secrets (BSides Leeds 2024) (https://x.com/advocatemack/)
    # https://hackage.haskell.org/package/github-types-0.2.1/docs/GitHub-Types-Events.html
    # https://www.gitguardian.com/


def banner():
    print(f"""
          

  ▄████  ██▓▄▄▄█████▓    ██░ ██  █    ██   ██████  ██░ ██ 
 ██▒ ▀█▒▓██▒▓  ██▒ ▓▒   ▓██░ ██▒ ██  ▓██▒▒██    ▒ ▓██░ ██▒
▒██░▄▄▄░▒██▒▒ ▓██░ ▒░   ▒██▀▀██░▓██  ▒██░░ ▓██▄   ▒██▀▀██░
░▓█  ██▓░██░░ ▓██▓ ░    ░▓█ ░██ ▓▓█  ░██░  ▒   ██▒░▓█ ░██ 
░▒▓███▀▒░██░  ▒██▒ ░    ░▓█▒░██▓▒▒█████▓ ▒██████▒▒░▓█▒░██▓
{RED} ░▒   ▒ ░▓    ▒ ░░       ▒ ░░▒░▒░▒▓▒ ▒ ▒ ▒ ▒▓▒ ▒ ░ ▒ ░░▒░▒
  ░   ░  ▒ ░    ░        ▒ ░▒░ ░░░▒░ ░ ░ ░ ░▒  ░ ░ ▒ ░▒░ ░
░ ░   ░  ▒ ░  ░          ░  ░░ ░ ░░░ ░ ░ ░  ░  ░   ░  ░░ ░
      ░  ░               ░  ░  ░   ░           ░   ░  ░  ░{RESET}
                                                                                       
        Developer: {CYAN}@stuub{RESET}
        Github: {YELLOW}https://github.com/stuub{RESET}
        Coffee? {GREEN}https://buymeacoffee.com/stuub{RESET}
        Who needs Git Blame?

        """)

def get_github_token():
    """
    Retrieve the GitHub token from either the command line arguments or environment variable.
    """
    parser = argparse.ArgumentParser(description='Script to use GitHub API')
    parser.add_argument('--github-token', type=str, help='GitHub access token')
    
    args = parser.parse_args()
    
    # First, lemme check if the argument is provided
    github_token = args.github_token
    
    # If not provided, fall back to env vars
    if not github_token:
        github_token = os.getenv('GITHUB_ACCESS_TOKEN')
    
    if not github_token:
        raise ValueError("GitHub token not provided. Set the --github-token argument or the GITHUB_ACCESS_TOKEN environment variable.")
    
    return github_token

def get_headers():
    """
    Get the headers required for the GitHub API requests.
    """
    token = get_github_token()
    return {'Authorization': f'token {token}'}


def get_events(headers, since=None):
    '''
    Fetch the events from the Github API.
    Returns the list of events and the response headers.
    '''

    url = 'https://api.github.com/events'
    if since:
        url += f'?since={since}'
    
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json(), response.headers

def fetch_commit_files(headers, commit_url):
    '''
    Fetch the files modified in the commit from the given URL.
    Returns the list of files modified in the commit.
    '''

    response = requests.get(commit_url, headers=headers)
    response.raise_for_status()
    return response.json().get('files', [])

def fetch_file_content(headers, file_url):
    '''
    Fetch the content of the file from the given URL.
    Decode the content from base64 encoding because Github API returns the content in base64.

    '''

    response = requests.get(file_url, headers=headers)
    response.raise_for_status()
    content = response.json().get('content', '')

    try:
        decoded_content = b64decode(content).decode('utf-8')
    except UnicodeDecodeError:
        print(f"{RED}    [-] Failed to decode file content as UTF-8 for {file_url}{RESET}")
        decoded_content = ""

    return decoded_content

def fetch_and_process_db_file(headers, file_url):
    # Fetch the file
    response = requests.get(file_url, headers=headers)
    response.raise_for_status()

    # Create a temporary file
    with tempfile.NamedTemporaryFile(suffix=".db") as temp_db:
        # Write the content of the response to the temporary file
        temp_db.write(response.content)
        temp_db.flush()

        # Check if the file is a valid SQLite database
        with open(temp_db.name, 'rb') as f:
            header = f.read(16)
            if not header.startswith(b'SQLite format 3'):
                print(f"{RED}    [-] File {file_url} is not a valid SQLite database.{RESET}")
                return

        # Connect to the SQLite database
        conn = sqlite3.connect(temp_db.name)
        try:
            # Create a cursor
            cursor = conn.cursor()

            # Get the list of tables in the database
            cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
            tables = cursor.fetchall()

            # For each table, select all data and check for secrets
            for table in tables:
                cursor.execute(f"SELECT * FROM {table[0]};")
                rows = cursor.fetchall()
                for row in rows:
                    for item in row:
                        if isinstance(item, str):
                            secrets = detect_secrets_in_text(item)
                            for secret in secrets:
                                print(f"Secret detected: {secret}")
        except sqlite3.DatabaseError as e:
            print(f"{RED}    [-] Error processing database file {file_url}: {e}{RESET}")
        finally:
            # Close the connection
            conn.close()

def detect_secrets_in_text(text):
    '''
    Detect secrets in the given response using regex patterns.
    Returns the secret and the type of secret if found.
    '''

    patterns = [
        # Generic tokens
        (re.compile(r'\btoken\s*[:=]\s*[\'"]?([a-zA-Z0-9_\-]{20,})[\'"]?\b'), "Generic Token"),
        (re.compile(r'\bapi[_-]?key\s*[:=]\s*[\'"]?([a-zA-Z0-9_\-]{20,})[\'"]?\b'), "API Key"),
        (re.compile(r'\bauth[_-]?key\s*[:=]\s*[\'"]?([a-zA-Z0-9_\-]{20,})[\'"]?\b'), "Auth Key"),
        (re.compile(r'\bsecret[_-]?key\s*[:=]\s*[\'"]?([a-zA-Z0-9_\-]{20,})[\'"]?\b'), "Secret Key"),
        (re.compile(r'\boidc[_-]?token\s*[:=]\s*[\'"]?([a-zA-Z0-9_\-]{20,})[\'"]?\b', re.IGNORECASE), "OIDC Token"),
        (re.compile(r'\bjwt[_-]?token\s*[:=]\s*[\'"]?([a-zA-Z0-9_\-]{20,})[\'"]?\b', re.IGNORECASE), "JWT Token"),
        
        # AWS keys and secrets
        (re.compile(r'\bAWS[^\w]*[A-Za-z0-9/+=]{40}\b'), "AWS Access Key"),
        (re.compile(r'\bAKIA[0-9A-Z]{16}\b'), "AWS Secret Key"),
        
        # Password patterns with contextual clues
        (re.compile(r'\b(?:password|passwd|pass|pwd)\s*[:=]\s*[\'"]?([a-zA-Z0-9@#$%^&*()\-_=+!]{8,})[\'"]?\b'), "Password"),
        
        # Email addresses
        (re.compile(r'\b[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b'), "Email Address"),
        
        # SSH keys
        (re.compile(r'\b(?:ssh-(?:rsa|dss|ed25519)|ecdsa-[a-zA-Z0-9]+) [A-Za-z0-9+/=]+(?: [^\n]+)?\b'), "SSH Key"),
        (re.compile(r'\b-----BEGIN [A-Z ]+-----[\s\S]+?-----END [A-Z ]+-----\b', re.DOTALL), "Private Key"),
        
        # Connection strings
        (re.compile(r'\bmongodb\+srv://[^:]+:[^@]+@[^/]+/[^?]+\b'), "MongoDB Connection String"),
        (re.compile(r'\bmysql:\/\/[^:]+:[^@]+@[^/]+/[^?]+\b'), "MySQL Connection String"),
        (re.compile(r'\bpostgresql:\/\/[^:]+:[^@]+@[^/]+/[^?]+\b'), "PostgreSQL Connection String"),

        # SMTP credentials
        (re.compile(r'\b(?:smtp|mail)\s*[:=]\s*[\'"]?([a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,})[\'"]?\b'), "SMTP Email"),
        (re.compile(r"(?i)(smtp_(username|user|password|pass|server|host|port))\s*[:=]\s*[\"']?([^\"'\s]+)[\"']?"), "SMTP Credentials"),
        
        # Service-specific tokens - Currently focussing common WP & CodeIgniter patterns (promise i'll add more :3)

        (re.compile(r"(?i)define\s*\(\s*['\"](DB_NAME|DB_USER|DB_PASSWORD|DB_HOST|AUTH_KEY|SECURE_AUTH_KEY|LOGGED_IN_KEY|NONCE_KEY)['\"]\s*,\s*['\"]([^'\"]+)['\"]\s*\)"), "wp-config.php Credentials"),
        (re.compile(r"(?i)\$mail->(Host|Username|Password|Port)\s*=\s*['\"]([^'\"]+)['\"]"), "PHPMailer SMTP Credential"),
        (re.compile(r"(?i)\$config\['(username|password)'\]\s*=\s*['\"]([^'\"]+)['\"]"), "CodeIgniter Config Credential"),
        (re.compile(r"(?i)\$db_(host|user|pass|name)\s*=\s*['\"]([^'\"]+)['\"]"), "PHP DB Config Variable"),
        (re.compile(r"(?i)^(DB_USER|DB_PASSWORD|DB_HOST|DB_NAME|SMTP_HOST|SMTP_USERNAME|SMTP_PASSWORD|API_KEY|SECRET_KEY)\s*=\s*([^\s]+)"), "ENV Variable Credential"),

    ]

    secrets = []
    for pattern, description in patterns:
        matches = pattern.findall(text)
        for match in matches:
            secret = match
            # if is_valid_secret(secret, description, text):
            secrets.append((secret, description))
    return secrets

def log_secret(logfile, filename, commit_sha, commit_url, secret, secret_type):
    log_entry = {
        "filename": filename,
        "commit_sha": commit_sha,
        "commit_url": commit_url,
        "secret": secret,
        "secret_type": secret_type
    }

    # a bird told me jsonl is the new black
    with open(logfile, "a", encoding='utf-8') as f:
        f.write(json.dumps(log_entry) + "\n")

def process_event(headers, event):
    '''
    Process a single event and look for secrets in the commits.
    If a secret is found, log it to logfile.
    '''

    print(f"{YELLOW}[!]{RESET} Processing event: {YELLOW}{event['id']}{RESET} - {YELLOW}{event['type']}{RESET}")

    absolutes = (
        'wp-config.php',
        'phpmailer.php',
        'config.php',
        'config.inc.php',
        'database.php',
        '.env.local',
    )

    extensions = (
        '.env', '.ini', '.cfg', '.conf',
        '.json', '.yaml', '.yml',
        '.py', '.js', '.ts', '.php', '.rb', '.go', '.java', '.cs',
        '.sh', '.bash', '.zsh', '.ps1', '.bat',
        '.pem', '.cer', '.crt', '.p12', '.pfx', '.key', '.rsa', '.jks',
        '.tf', '.tfvars',
        '.sql', '.sqlite', '.xml', '.properties',
        '.gradle', '.pom', '.Dockerfile',
        '.md', '.markdown'
    )

    if event['type'] not in ('PushEvent', 'PullRequestEvent'):
        return

    repo_name = event['repo']['name']
    commits = []

    if event['type'] == 'PushEvent':
        commits = event['payload'].get('commits', [])

    elif event['type'] == 'PullRequestEvent':
        pr = event['payload'].get('pull_request', {})
        commits_url = pr.get('commits_url')
        if commits_url:
            try:
                response = requests.get(commits_url, headers=headers)
                response.raise_for_status()
                commits = response.json()
            except requests.RequestException as e:
                print(f"{RED}    [-] Failed to fetch PR commits: {e}{RESET}")
                return

    for commit in commits:
        commit_sha = commit['sha']
        commit_url = commit['url']
        print(f"{YELLOW}[!]{RESET} Processing commit: {YELLOW}{commit_sha}{RESET}")
        files = fetch_commit_files(headers, commit_url)

        for file in files:
            filename = file['filename'].lower()

            if 'wp-config-sample.php' in filename:
                continue

            if filename.endswith('.db'):
                file_url = file['contents_url']
                print(f"{YELLOW}    [*] Fetching file: {file['filename']}{RESET}")
                fetch_and_process_db_file(headers, file_url)
                continue

            if any(hvf in filename for hvf in absolutes) or filename.endswith(extensions):
                file_url = file['contents_url']
                print(f"{YELLOW}    [*] Fetching file: {file['filename']}{RESET}")
                file_content = fetch_file_content(headers, file_url)

                if not file_content:
                    print(f"{RED}    [-] Failed to fetch file content for {file['filename']}{RESET}")
                    continue

                detected_secrets = detect_secrets_in_text(file_content)
                if not detected_secrets:
                    print(f"{RED}    [-] No secrets found in {file['filename']}{RESET}")
                    continue

                commit_web_url = f"https://github.com/{repo_name}/commit/{commit_sha}"
                print(f"{GREEN}[+]{RESET} Secrets detected in {GREEN}{file['filename']}{RESET} at commit {GREEN}{commit_sha}{RESET}")
                print(f"{GREEN}[+]{RESET} URL: {GREEN}{commit_web_url}{RESET}")

                secrets_by_type = {}
                for secret, secret_type in detected_secrets:
                    secrets_by_type.setdefault(secret_type, []).append(secret)

                for secret_type, secrets in secrets_by_type.items():
                    print(f"{CYAN}[*]{RESET} Secret Type: {CYAN}{secret_type}{RESET}")
                    for secret in secrets:
                        print(f"{GREEN}[+]{RESET} Secret: {GREEN}{secret}{RESET}")
                        log_secret(logfile, file['filename'], commit_sha, commit_web_url, secret, secret_type)


def main():
    '''
    Fetch events and process em.
    In case of rate limit exceeded, it will wait for the reset time.
    '''
    last_event_id = None
    retry_count = 0
    max_retries = 5
    try:
        token = get_github_token()
        print(f'Using GitHub token: {token}')
        headers = get_headers()

    except ValueError as e:
        print(f"{RED}[-]{RESET} Error: GitHub token not provided. Please set the --github-token argument or the GITHUB_ACCESS_TOKEN environment variable.")
        exit(1)

    while True:
        try:
            print(f"{YELLOW}[*] Fetching events...{RESET}")
            events, response_headers = get_events(headers, since=last_event_id)
            if events:
                last_event_id = events[0]['id']
                for event in events:
                    detected_secrets = process_event(headers, event) or []
                    for secret in detected_secrets:
                        print(f"\n{YELLOW}[!] Secret detected in repository: {secret['repo']}{RESET}")
                        print(f"    File: {secret['file']}")
                        print(f"    Secret Type: {secret['type']}")
                        print(f"    Secret: {secret['secret']}")
            else:
                print(f"{RED}[!] No new events.{RESET}")
            
            remaining_requests = int(response_headers.get('X-RateLimit-Remaining', 0))
            reset_time = int(response_headers.get('X-RateLimit-Reset', time.time()))
            
            if 'Retry-After' in response_headers:
                retry_after = int(response_headers['Retry-After'])
                print(f"{YELLOW}[*] Rate limit hit. Retrying after {retry_after} seconds.{RESET}")
                time.sleep(retry_after)
            elif remaining_requests == 0:
                sleep_time = max(0, reset_time - int(time.time()))
                print(f"{YELLOW}[*] Rate limit exceeded. Waiting for {sleep_time} seconds.{RESET}")
                time.sleep(sleep_time)
            else:
                print(f"{YELLOW}[*] Remaining requests: {remaining_requests}{RESET}")
                print(f"{YELLOW}[*] Waiting for next poll...{RESET}")
                time.sleep(10)
            
            retry_count = 0  # Reset retry count after good req

        except (requests.exceptions.HTTPError, requests.exceptions.ConnectionError) as e:
            print(f"{RED}[-] Error: {e}{RESET}")
            if retry_count < max_retries:
                retry_count += 1
                sleep_time = min(2 ** retry_count, 60)
                print(f"{RED}[-] Retrying in {sleep_time} seconds... (Retry {retry_count}/{max_retries}){RESET}")
                time.sleep(sleep_time)
            else:
                print(f"{RED}[-] Max retries reached. Rate Limit Exceeded. Exiting.{RESET}")
                break
        except KeyboardInterrupt:
            print(f"{RED}[-] Exiting...{RESET}")
            break

if __name__ == '__main__':
    logfile = 'goodie_bag.jsonl'
    banner()
    main()