endpoint-checker.py

#!/usr/bin/env python3
"""
Sbow-Endpointer
A script to check multiple endpoints for multiple websites with advanced filtering
and output options including host information, IP addresses, and CSV export.
"""

import requests
import sys
import argparse
from urllib.parse import urljoin, urlparse
from datetime import datetime
import time
import os
import socket
import csv
from bs4 import BeautifulSoup
import re

VERSION = "1.0.0"
TOOL_NAME = "Sbow-Endpointer"

def print_banner():
    """Print the tool banner with ASCII art."""
    # ANSI color codes
    RED = '\033[91m'
    GREEN = '\033[92m'
    RESET = '\033[0m'
    banner = f"""{RED}
   ██████╗ ██████╗  ██████╗ ██╗    ██╗
  ██╔════╝ ██╔══██╗██╔═══██╗██║    ██║
  ███████╗ ██████╔╝██║   ██║██║ █╗ ██║
  ╚════██║ ██╔══██╗██║   ██║██║███╗██║
  ███████║ ██████╔╝╚██████╔╝╚███╔███╔╝
  ╚══════╝ ╚═════╝  ╚═════╝  ╚══╝╚══╝{RESET}
"""
    print(banner)
    print(f"\t\t\t{GREEN}Sbow-Endpointer v{VERSION}{RESET}")
    print(f"\t\t\t{GREEN}Developed by Sidharth Bahuguna{RESET}")
    print()

class EndpointChecker:
    def __init__(self, timeout=10, user_agent=None):
        self.timeout = timeout
        self.session = requests.Session()
        if user_agent:
            self.session.headers.update({'User-Agent': user_agent})
        else:
            self.session.headers.update({'User-Agent': 'Website-Endpoint-Checker/3.0'})
    
    def get_ip_address(self, hostname):
        """Get IP address for a hostname"""
        try:
            # Remove protocol and path, get just hostname
            if '://' in hostname:
                hostname = hostname.split('://')[1].split('/')[0]
            hostname = hostname.split(':')[0]  # Remove port if present
            return socket.gethostbyname(hostname)
        except socket.gaierror:
            return None
    
    def extract_title(self, html_content):
        """Extract title from HTML content"""
        try:
            soup = BeautifulSoup(html_content, 'html.parser')
            title_tag = soup.find('title')
            if title_tag:
                return title_tag.get_text().strip()
        except:
            pass
        return None
    
    def get_location_from_headers(self, headers):
        """Extract location from response headers"""
        return headers.get('Location', headers.get('location', None))
    
    def normalize_url(self, website, endpoint):
        """Normalize website URL and combine with endpoint"""
        if not website.startswith(('http://', 'https://')):
            website = 'https://' + website
        
        # Remove trailing slash from website
        website = website.rstrip('/')
        
        # Ensure endpoint starts with /
        if not endpoint.startswith('/'):
            endpoint = '/' + endpoint
        
        return urljoin(website, endpoint)
    
    def check_endpoint(self, website, endpoint, keyword=None, get_host_info=False):
        """Check a specific endpoint for a website"""
        url = self.normalize_url(website, endpoint)
        
        result = {
            'website': website,
            'endpoint': endpoint,
            'url': url,
            'status_code': None,
            'status_text': None,
            'response_time': None,
            'content_length': None,
            'keyword_found': False,
            'ip_address': None,
            'title': None,
            'location': None,
            'server': None,
            'error': None
        }
        
        # Get IP address if requested
        if get_host_info:
            result['ip_address'] = self.get_ip_address(website)
        
        try:
            response = self.session.get(url, timeout=self.timeout, allow_redirects=True)
            
            result.update({
                'status_code': response.status_code,
                'status_text': response.reason,
                'response_time': response.elapsed.total_seconds(),
                'content_length': len(response.content),
                'server': response.headers.get('Server', response.headers.get('server', None)),
                'location': self.get_location_from_headers(response.headers)
            })
            
            # Extract title if requested and content is HTML
            if get_host_info and response.status_code == 200:
                content_type = response.headers.get('content-type', '').lower()
                if 'text/html' in content_type:
                    result['title'] = self.extract_title(response.text)
            
            # Check for keyword in response content if provided
            if keyword and response.status_code == 200:
                try:
                    content = response.text.lower()
                    result['keyword_found'] = keyword.lower() in content
                except:
                    result['keyword_found'] = False
            
            return result
            
        except requests.exceptions.RequestException as e:
            result['error'] = str(e)
            return result
    
    def check_multiple_combinations(self, websites, endpoints, keyword=None, delay=0, get_host_info=False):
        """Check multiple endpoints for multiple websites"""
        results = []
        total_checks = len(websites) * len(endpoints)
        current_check = 0
        
        for website in websites:
            for endpoint in endpoints:
                current_check += 1
                print(f"Checking {current_check}/{total_checks}: {website}{endpoint}")
                result = self.check_endpoint(website.strip(), endpoint.strip(), keyword, get_host_info)
                results.append(result)
                
                # Add delay between requests to be respectful
                if delay > 0 and current_check < total_checks:
                    time.sleep(delay)
        
        return results
    
    def filter_results(self, results, status_codes=None, keyword=None):
        """Filter results based on criteria"""
        filtered = []
        
        for result in results:
            if result['error']:
                continue
            
            # Check if should include based on status code
            include_status = True
            if status_codes:
                include_status = result['status_code'] in status_codes
            
            # Check if should include based on keyword
            include_keyword = not keyword or result['keyword_found']
            
            if include_status and include_keyword:
                filtered.append(result)
        
        return filtered

def format_output(results, show_details=False, show_host_info=False):
    """Format results for output"""
    output_lines = []
    
    if not results:
        return "No results match the specified criteria."
    
    output_lines.append(f"Results found: {len(results)}")
    output_lines.append("-" * 100)
    
    for result in results:
        if show_details and show_host_info:
            # Full detailed output with host info
            line = f"URL: {result['url']}"
            line += f"\n  Status: {result['status_code']} {result['status_text']}"
            line += f"\n  Response Time: {result['response_time']:.2f}s"
            line += f"\n  Content Length: {result['content_length']} bytes"
            if result['ip_address']:
                line += f"\n  IP Address: {result['ip_address']}"
            if result['title']:
                line += f"\n  Title: {result['title']}"
            if result['server']:
                line += f"\n  Server: {result['server']}"
            if result['location']:
                line += f"\n  Location: {result['location']}"
            if result['keyword_found']:
                line += f"\n  Keyword: FOUND"
            line += "\n"
        elif show_details:
            # Standard detailed output
            line = f"{result['url']} "
            line += f"[{result['status_code']} {result['status_text']}] "
            line += f"({result['response_time']:.2f}s, {result['content_length']} bytes)"
            if result['keyword_found']:
                line += " [KEYWORD FOUND]"
        else:
            # Simple output
            line = f"{result['url']} -> {result['status_code']} {result['status_text']}"
            if result['keyword_found']:
                line += " [KEYWORD FOUND]"
        
        output_lines.append(line)
    
    return "\n".join(output_lines)

def save_to_txt_file(results, filename=None, urls_only=False, full_details=True):
    """Save results to a text file"""
    if not filename:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"endpoint_check_results_{timestamp}.txt"
    
    try:
        with open(filename, 'w', encoding='utf-8') as f:
            if urls_only:
                # Save only URLs
                for result in results:
                    f.write(f"{result['url']}\n")
            elif full_details:
                # Save all detailed information by default
                content = format_output(results, show_details=True, show_host_info=True)
                f.write(content)
                f.write(f"\n\nGenerated on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
            else:
                # Save formatted output
                content = format_output(results, show_details=True, show_host_info=False)
                f.write(content)
                f.write(f"\n\nGenerated on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        return filename
    except Exception as e:
        print(f"Error saving file: {e}")
        return None

def save_to_csv_file(results, filename=None):
    """Save results to a CSV file"""
    if not filename:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"endpoint_check_results_{timestamp}.csv"
    
    try:
        with open(filename, 'w', newline='', encoding='utf-8') as csvfile:
            fieldnames = [
                'website', 'endpoint', 'url', 'status_code', 'status_text',
                'response_time', 'content_length', 'ip_address', 'title',
                'server', 'location', 'keyword_found'
            ]
            
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            
            for result in results:
                # Clean up the data for CSV
                csv_row = {
                    'website': result['website'],
                    'endpoint': result['endpoint'],
                    'url': result['url'],
                    'status_code': result['status_code'],
                    'status_text': result['status_text'],
                    'response_time': f"{result['response_time']:.3f}" if result['response_time'] else '',
                    'content_length': result['content_length'] or '',
                    'ip_address': result['ip_address'] or '',
                    'title': result['title'] or '',
                    'server': result['server'] or '',
                    'location': result['location'] or '',
                    'keyword_found': result['keyword_found']
                }
                writer.writerow(csv_row)
        
        return filename
    except Exception as e:
        print(f"Error saving CSV file: {e}")
        return None

def read_list_from_file(filename, item_type="items"):
    """Read a list of items from a file"""
    items = []
    
    try:
        with open(filename, 'r', encoding='utf-8') as f:
            for line_num, line in enumerate(f, 1):
                item = line.strip()
                # Skip empty lines and comments
                if item and not item.startswith('#'):
                    items.append(item)
        
        if not items:
            print(f"Warning: No valid {item_type} found in {filename}")
        else:
            print(f"Loaded {len(items)} {item_type} from {filename}")
        
        return items
        
    except FileNotFoundError:
        print(f"Error: File '{filename}' not found.")
        sys.exit(1)
    except PermissionError:
        print(f"Error: Permission denied to read '{filename}'.")
        sys.exit(1)
    except Exception as e:
        print(f"Error reading file '{filename}': {e}")
        sys.exit(1)

def get_items_from_input(item_type="items"):
    """Get items from standard input"""
    items = []
    
    print(f"Enter {item_type}, one per line.")
    print("Press Ctrl+D (Unix) or Ctrl+Z (Windows) when finished:")
    
    try:
        for line in sys.stdin:
            item = line.strip()
            if item:
                items.append(item)
    except KeyboardInterrupt:
        print(f"\nInput interrupted by user.")
        sys.exit(1)
    
    return items

def parse_comma_separated(value):
    """Parse comma-separated values"""
    return [item.strip() for item in value.split(',') if item.strip()]

def parse_status_codes(value):
    """Parse status codes input"""
    codes = []
    for item in value.split(','):
        item = item.strip()
        if item.isdigit():
            codes.append(int(item))
        elif 'x' in item or 'X' in item:
            # Handle ranges like 2xx, 3xx, 4xx, 5xx
            base = item.replace('x', '').replace('X', '')
            if base.isdigit():
                base_code = int(base) * 100
                codes.extend(range(base_code, base_code + 100))
    return codes

def main():
    parser = argparse.ArgumentParser(
        description="Advanced Website Endpoint Checker v3",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Basic endpoint checking
  python endpoint_checker.py -e "/health,/status" -W "site1.com,site2.com"
  
  # Filter by specific status codes
  python endpoint_checker.py -ef endpoints.txt -w websites.txt --status-codes "200,201,301"
  
  # Filter by status code ranges
  python endpoint_checker.py -ef endpoints.txt -w websites.txt --status-codes "2xx,3xx"
  
  # Get detailed host information
  python endpoint_checker.py -e "/health" -W "site1.com" --host-info --details
  
  # Save only URLs to text file
  python endpoint_checker.py -ef endpoints.txt -w websites.txt --save-txt urls.txt --urls-only
  
  # Save minimal details to CSV (without host info)  
  python endpoint_checker.py -e "/health" -w websites.txt --save-csv results.csv --minimal-save

  # Save full details to CSV (default behavior - includes all collected data)
  python endpoint_checker.py -e "/health" -w websites.txt --host-info --save-csv detailed.csv

  # Save formatted report to text with all details (default)
  python endpoint_checker.py -e "/status" -w websites.txt --save-txt report.txt

  # Advanced filtering
  python endpoint_checker.py -e "/admin" -w websites.txt -k "login" --status-codes "200"

Default File Saving Behavior:
  - Text files: Save full detailed output with all collected data (unless --urls-only specified)
  - CSV files: Save all available columns including host info if collected
  - Use --minimal-save to exclude extended host information from files
  - Use --urls-only to save only URLs to text files
        """
    )
    
    # Endpoint input options
    endpoint_group = parser.add_mutually_exclusive_group()
    endpoint_group.add_argument('-e', '--endpoints',
                               help='Comma-separated endpoints (e.g., "/health,/status,/api")')
    endpoint_group.add_argument('-ef', '--endpoints-file',
                               help='File containing endpoints (one per line)')
    
    # Website input options
    website_group = parser.add_mutually_exclusive_group()
    website_group.add_argument('-w', '--websites-file',
                              help='File containing websites (one per line)')
    website_group.add_argument('-W', '--websites',
                              help='Comma-separated websites (e.g., "site1.com,site2.com")')
    
    # Filtering options
    parser.add_argument('--status-codes',
                       help='Filter by HTTP status codes (e.g., "200,301" or "2xx,4xx")')
    parser.add_argument('-k', '--keyword',
                       help='Keyword to search for in response content')
    
    # Output options
    parser.add_argument('--details', action='store_true',
                       help='Show detailed output with response time and size')
    parser.add_argument('--host-info', action='store_true',
                       help='Include host information (IP, title, server, location)')
    
    # File saving options
    save_group = parser.add_mutually_exclusive_group()
    save_group.add_argument('--save-txt', metavar='FILENAME',
                           help='Save results to text file')
    save_group.add_argument('--save-csv', metavar='FILENAME',
                           help='Save results to CSV file')
    
    parser.add_argument('--urls-only', action='store_true',
                       help='When saving to text file, save only URLs (no other data)')
    parser.add_argument('--minimal-save', action='store_true',
                       help='Save minimal data to files (excludes host info by default)')
    
    # Request options
    parser.add_argument('-t', '--timeout', type=int, default=10,
                       help='Request timeout in seconds (default: 10)')
    parser.add_argument('-d', '--delay', type=float, default=0,
                       help='Delay between requests in seconds (default: 0)')
    parser.add_argument('--user-agent',
                       help='Custom User-Agent string')
    
    args = parser.parse_args()
    
    # Get endpoints
    if args.endpoints:
        endpoints = parse_comma_separated(args.endpoints)
    elif args.endpoints_file:
        endpoints = read_list_from_file(args.endpoints_file, "endpoints")
    else:
        print("No endpoints specified. Enter endpoints manually:")
        endpoints = get_items_from_input("endpoints")
    
    if not endpoints:
        print("No endpoints provided.")
        sys.exit(1)
    
    # Get websites
    if args.websites:
        websites = parse_comma_separated(args.websites)
    elif args.websites_file:
        websites = read_list_from_file(args.websites_file, "websites")
    else:
        print("No websites specified. Enter websites manually:")
        websites = get_items_from_input("websites (domains/subdomains)")
    
    if not websites:
        print("No websites provided.")
        sys.exit(1)
    
    # Parse status codes filter
    status_codes = None
    if args.status_codes:
        status_codes = parse_status_codes(args.status_codes)
        print(f"Filtering for status codes: {sorted(set(status_codes))}")
    
    # Print summary
    print(f"Checking {len(websites)} websites with {len(endpoints)} endpoints")
    print(f"Total checks to perform: {len(websites) * len(endpoints)}")
    if args.keyword:
        print(f"Filtering for keyword: '{args.keyword}'")
    if args.host_info:
        print("Host information will be collected")
    
    # Initialize checker
    checker = EndpointChecker(timeout=args.timeout, user_agent=args.user_agent)
    
    # Check all combinations
    all_results = checker.check_multiple_combinations(
        websites, endpoints, args.keyword, args.delay, args.host_info
    )
    
    # Filter results
    filtered_results = checker.filter_results(
        all_results, 
        status_codes=status_codes,
        keyword=args.keyword
    )
    
    # Display output
    if filtered_results:
        output_content = format_output(filtered_results, show_details=args.details, show_host_info=args.host_info)
        print(output_content)
    else:
        print("No results match the specified criteria.")
    
    # Save to file if requested
    if args.save_txt and filtered_results:
        filename = save_to_txt_file(
            filtered_results, 
            args.save_txt, 
            urls_only=args.urls_only, 
            full_details=not args.minimal_save
        )
        if filename:
            print(f"Results saved to text file: {filename}")
    elif args.save_csv and filtered_results:
        filename = save_to_csv_file(filtered_results, args.save_csv)
        if filename:
            print(f"Results saved to CSV file: {filename}")
    elif filtered_results and not (args.save_txt or args.save_csv):
        # Interactive save option
        try:
            save_choice = input(f"\nSave results to file? (t)xt/(c)sv/(n)o: ").lower()
            if save_choice.startswith('t'):
                urls_only = input("Save URLs only? (y/n): ").lower().startswith('y')
                filename = save_to_txt_file(
                    filtered_results, 
                    urls_only=urls_only,
                    full_details=not urls_only  # If not URLs only, save full details by default
                )
                if filename:
                    print(f"Results saved to text file: {filename}")
            elif save_choice.startswith('c'):
                filename = save_to_csv_file(filtered_results)
                if filename:
                    print(f"Results saved to CSV file: {filename}")
        except (KeyboardInterrupt, EOFError):
            pass
    
    # Print summary
    total_success = len([r for r in all_results if not r['error']])
    print(f"\nSummary:")
    print(f"  Total checks: {len(all_results)}")
    print(f"  Successful requests: {total_success}")
    print(f"  Matching criteria: {len(filtered_results)}")
    
    # Show status code breakdown
    status_counts = {}
    for result in all_results:
        if not result['error']:
            code = result['status_code']
            status_counts[code] = status_counts.get(code, 0) + 1
    
    if status_counts:
        print(f"  Status code breakdown:")
        for code in sorted(status_counts.keys()):
            print(f"    {code}: {status_counts[code]}")

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\nOperation cancelled by user.")
        sys.exit(1)