har_endpoint_extractor.py

#!/usr/bin/env python3
"""
HAR File API Endpoint Extractor
Extract real API endpoints from browser HAR files (from Burp, DevTools, etc.)
and generate targets for Nuclei scanning.

Usage:
    python har_endpoint_extractor.py --har-file traffic.har --output nuclei_targets.txt
"""

import json
import sys
import argparse
import logging
from pathlib import Path
from typing import Set, Dict, List
from urllib.parse import urlparse, urljoin
from collections import defaultdict

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


class HAREndpointExtractor:
    def __init__(self, har_file: str):
        self.har_file = har_file
        self.entries = []
        self.api_endpoints: Set[str] = set()
        self.api_calls: Dict[str, List[Dict]] = defaultdict(list)
        
    def load_har(self) -> bool:
        """Load and parse HAR file"""
        try:
            with open(self.har_file, 'r') as f:
                har_data = json.load(f)
            
            self.entries = har_data.get('log', {}).get('entries', [])
            logger.info(f"[+] Loaded {len(self.entries)} entries from HAR file")
            return True
        except Exception as e:
            logger.error(f"[-] Error loading HAR file: {e}")
            return False
    
    def extract_endpoints(self, api_prefix: str = "/api/") -> Set[str]:
        """Extract API endpoints from HAR entries"""
        endpoints = set()
        
        for entry in self.entries:
            try:
                request = entry.get('request', {})
                url = request.get('url', '')
                method = request.get('method', 'GET')
                
                # Filter to API endpoints
                if api_prefix.lower() not in url.lower():
                    continue
                
                # Parse URL and remove query params/fragments
                parsed = urlparse(url)
                clean_url = f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
                
                endpoints.add(clean_url)
                
                # Store request details
                self.api_calls[clean_url].append({
                    'method': method,
                    'status': entry.get('response', {}).get('status', '?'),
                    'headers': {h['name']: h['value'] for h in request.get('headers', [])},
                })
                
            except Exception as e:
                logger.debug(f"[-] Error processing entry: {e}")
                continue
        
        self.api_endpoints = endpoints
        logger.info(f"[+] Extracted {len(endpoints)} unique API endpoints")
        return endpoints
    
    def classify_endpoints(self) -> Dict[str, List[str]]:
        """Classify endpoints by type (patient, admin, appointment, etc.)"""
        classified = defaultdict(list)
        
        keyword_map = {
            'patient': ['patient', 'user', 'profile', 'account', 'me'],
            'appointment': ['appointment', 'booking', 'slot', 'schedule'],
            'admin': ['admin', 'management', 'staff', 'doctor', 'specialist'],
            'auth': ['login', 'logout', 'auth', 'token', 'session'],
            'config': ['config', 'settings', 'info', 'health', 'status'],
        }
        
        for endpoint in self.api_endpoints:
            url_lower = endpoint.lower()
            matched = False
            
            for category, keywords in keyword_map.items():
                if any(kw in url_lower for kw in keywords):
                    classified[category].append(endpoint)
                    matched = True
            
            if not matched:
                classified['other'].append(endpoint)
        
        return dict(classified)
    
    def filter_by_status(self, status_codes: List[int] = None) -> Set[str]:
        """Filter endpoints by HTTP response status"""
        if status_codes is None:
            status_codes = [200, 201]
        
        filtered = set()
        
        for endpoint in self.api_endpoints:
            for call in self.api_calls[endpoint]:
                if call['status'] in status_codes:
                    filtered.add(endpoint)
                    break
        
        logger.info(f"[+] Found {len(filtered)} endpoints with status codes {status_codes}")
        return filtered
    
    def save_targets(self, output_file: str, endpoints: Set[str] = None):
        """Save endpoints as Nuclei target file"""
        if endpoints is None:
            endpoints = self.api_endpoints
        
        with open(output_file, 'w') as f:
            for endpoint in sorted(endpoints):
                f.write(f"{endpoint}\n")
        
        logger.info(f"[+] Saved {len(endpoints)} targets to {output_file}")
    
    def save_json_targets(self, output_file: str, endpoints: Set[str] = None):
        """Save endpoints as JSON with metadata"""
        if endpoints is None:
            endpoints = self.api_endpoints
        
        targets = []
        for endpoint in sorted(endpoints):
            calls = self.api_calls[endpoint]
            target = {
                "url": endpoint,
                "methods": list(set(c['method'] for c in calls)),
                "status_codes": list(set(c['status'] for c in calls)),
                "request_count": len(calls),
                "headers": calls[0]['headers'] if calls else {}
            }
            targets.append(target)
        
        with open(output_file, 'w') as f:
            json.dump(targets, f, indent=2)
        
        logger.info(f"[+] Saved {len(targets)} targets with metadata to {output_file}")
    
    def save_classification_report(self, output_file: str):
        """Save classified endpoints report"""
        classified = self.classify_endpoints()
        
        report = {
            "total_endpoints": len(self.api_endpoints),
            "by_category": {
                cat: {
                    "count": len(endpoints),
                    "endpoints": sorted(endpoints)
                }
                for cat, endpoints in classified.items()
            }
        }
        
        with open(output_file, 'w') as f:
            json.dump(report, f, indent=2)
        
        logger.info(f"[+] Saved classification report to {output_file}")
        
        # Print summary
        logger.info("\n=== Endpoint Classification Summary ===")
        for category, endpoints_list in sorted(classified.items()):
            logger.info(f"{category}: {len(endpoints_list)} endpoints")
            for ep in sorted(endpoints_list)[:5]:  # Show first 5
                logger.info(f"  - {ep}")
            if len(endpoints_list) > 5:
                logger.info(f"  ... and {len(endpoints_list) - 5} more")
    
    def generate_nuclei_commands(self, template_dir: str = ".") -> List[str]:
        """Generate recommended Nuclei commands"""
        classified = self.classify_endpoints()
        commands = []
        
        # Command for authenticated endpoints (patient data)
        patient_endpoints = classified.get('patient', [])
        if patient_endpoints:
            endpoints_file = "nuclei_targets_patient.txt"
            self.save_targets(endpoints_file, set(patient_endpoints))
            cmd = (f"nuclei -l {endpoints_file} "
                   f"-t {template_dir}/netbear-auth-bypass.yaml "
                   f"-t {template_dir}/netbear-idor.yaml "
                   f"-o nuclei_results_patient.txt -silent")
            commands.append(("Patient/User endpoints (IDOR/Auth Bypass)", cmd))
        
        # Command for API exposure
        api_endpoints = classified.get('config', [])
        if api_endpoints:
            endpoints_file = "nuclei_targets_config.txt"
            self.save_targets(endpoints_file, set(api_endpoints))
            cmd = (f"nuclei -l {endpoints_file} "
                   f"-t {template_dir}/netbear-exposure.yaml "
                   f"-o nuclei_results_config.txt -silent")
            commands.append(("Config/Info endpoints (API Exposure)", cmd))
        
        # Command for all endpoints
        all_file = "nuclei_targets_all.txt"
        self.save_targets(all_file, self.api_endpoints)
        cmd = (f"nuclei -l {all_file} "
               f"-t {template_dir}/netbear-*.yaml "
               f"-o nuclei_results_all.txt -silent")
        commands.append(("All endpoints (Comprehensive scan)", cmd))
        
        return commands


def main():
    parser = argparse.ArgumentParser(
        description="Extract and analyze API endpoints from HAR files"
    )
    parser.add_argument("--har-file", required=True, help="Path to HAR file")
    parser.add_argument("--api-prefix", default="/api/", help="API path prefix to filter")
    parser.add_argument("--output-txt", help="Output file for plain text targets")
    parser.add_argument("--output-json", help="Output file for JSON targets with metadata")
    parser.add_argument("--output-report", help="Output file for classification report")
    parser.add_argument("--template-dir", default=".", help="Directory with Nuclei templates")
    parser.add_argument("--status-filter", help="Comma-separated HTTP status codes to include (default: 200,201)")
    parser.add_argument("--generate-commands", action="store_true", help="Generate Nuclei scan commands")
    
    args = parser.parse_args()
    
    # Load and extract
    extractor = HAREndpointExtractor(args.har_file)
    
    if not extractor.load_har():
        sys.exit(1)
    
    extractor.extract_endpoints(args.api_prefix)
    
    if not extractor.api_endpoints:
        logger.warning("[-] No API endpoints found")
        sys.exit(1)
    
    # Filter by status if specified
    if args.status_filter:
        status_codes = [int(s.strip()) for s in args.status_filter.split(',')]
        endpoints = extractor.filter_by_status(status_codes)
    else:
        endpoints = extractor.api_endpoints
    
    # Save outputs
    if args.output_txt:
        extractor.save_targets(args.output_txt, endpoints)
    
    if args.output_json:
        extractor.save_json_targets(args.output_json, endpoints)
    
    if args.output_report:
        extractor.save_classification_report(args.output_report)
    else:
        # Always save classification report if not specified
        extractor.save_classification_report("endpoint_classification.json")
    
    # Generate Nuclei commands
    if args.generate_commands:
        logger.info("\n=== Recommended Nuclei Commands ===")
        commands = extractor.generate_nuclei_commands(args.template_dir)
        for title, cmd in commands:
            logger.info(f"\n{title}:")
            logger.info(f"  {cmd}")
    
    logger.info("\n[+] Endpoint extraction complete!")


if __name__ == "__main__":
    main()