CorsOne.py

#!/usr/bin/env python3
"""
CorsOne - CORS Misconfiguration Discovery Tool
A fast, reliable, and feature-rich CORS vulnerability scanner.

Usage:
    python3 CorsOne.py -u https://example.com
    python3 CorsOne.py -l targets.txt -w 10
    cat domains.txt | python3 CorsOne.py -w 20

Version: 1.1.0
Author: Mohammad Reza Omrani
License: MIT
"""

from __future__ import annotations

import argparse
import asyncio  # PERF 1
import json
import logging
import sys
from contextlib import asynccontextmanager  # PERF 3
from dataclasses import dataclass, asdict, field
from pathlib import Path
from time import sleep
from threading import Lock
from typing import Dict, List, Optional, Tuple, Set
from urllib.parse import unquote, urlparse

import aiohttp  # PERF 1
import aiodns  # PERF 2
from aiohttp import ClientTimeout, TCPConnector  # PERF 1
import validators
from colorama import Fore, Style, init

# Suppress SSL warnings (logger only; not importing urllib3 directly)
urllib3_logger = logging.getLogger('urllib3')
urllib3_logger.setLevel(logging.WARNING)

# Initialize colorama
init(autoreset=True)

# Global locks for thread-safe operations
output_lock = Lock()
log_lock = Lock()

logger = logging.getLogger(__name__)  # FIX C1

def _setup_logging(verbose: bool = False, log_file: Optional[str] = None) -> None:  # FIX C1
    level = logging.DEBUG if verbose else logging.INFO
    logger.setLevel(level)
    if logger.handlers:
        logger.handlers.clear()
    fmt = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    ch = logging.StreamHandler(sys.stderr)
    ch.setLevel(level)
    ch.setFormatter(fmt)
    logger.addHandler(ch)
    if log_file:
        try:
            fh = logging.FileHandler(log_file)
            fh.setLevel(logging.DEBUG)
            fh.setFormatter(fmt)
            logger.addHandler(fh)
        except IOError as e:
            logger.warning(f"Could not create log file: {e}")


# ============================================================================
# Configuration & Data Classes
# ============================================================================

@dataclass
class ScanResult:
    """Represents a single CORS bypass test result."""
    url: str
    bypass_name: str
    bypass_value: str
    is_vulnerable: bool
    response_code: int = 0
    acac: Optional[str] = None
    acao: Optional[str] = None
    error: Optional[str] = None
    timestamp: float = field(default_factory=__import__('time').time)

    def to_dict(self) -> Dict:
        """Convert result to dictionary."""
        result = asdict(self)
        # Remove None values from output
        return {k: v for k, v in result.items() if v is not None}

    def __str__(self) -> str:
        """String representation."""
        status = '[VULNERABLE]' if self.is_vulnerable else '[SAFE]'
        return f"{self.url} {status} {self.bypass_name}: {self.bypass_value}"


@dataclass
class ScanConfig:
    """Configuration for CORS vulnerability scanning."""
    url: str
    method: str = "GET"
    custom_domain: str = "attacker.com"
    rate_limit: float = 0.0
    timeout: int = 10
    retries: int = 3
    backoff_factor: float = 0.5
    max_workers: int = 5
    stop_on_first: bool = False
    no_color: bool = False
    output_file: Optional[str] = None
    output_format: str = "txt"
    output_log: Optional[str] = None
    custom_headers: Optional[Dict[str, str]] = None
    proxy: Optional[Dict[str, str]] = None
    verbose: bool = False
    vulnerable_only: bool = False


class CORSBypassPayloads:
    """CORS bypass payload generation and management."""

    @staticmethod
    def generate(origin: str, malicious_domain: str) -> Dict[str, str]:
        """
        Generate CORS bypass payloads dynamically.
        
        Args:
            origin: Target domain origin
            malicious_domain: Attacker domain for bypass attempts
            
        Returns:
            Dictionary of bypass names and their payloads
        """
        return {
            'Reflected Origin': f'https://{malicious_domain}',
            'Breaking TLS': f'http://{origin}',
            'Trusted Subdomains': f'https://subdomain.{origin}',
            'Unencrypted Subdomains': f'http://subdomain.{origin}',
            'Null Origin': 'null',
            'Unencrypted domain ends allow': f'http://attacker{origin}',
            'Domain ends allow': f'https://attacker{origin}',
            'Unencrypted localhost regex': f'http://localhost.{malicious_domain}',
            'Localhost regex': f'https://localhost.{malicious_domain}',
            'Bypass 1': f'http://{malicious_domain}.{origin}',
            'Bypass 2': f'https://{malicious_domain}.{origin}',
            'Bypass 3': f'https://{origin}._.{malicious_domain}',
            'Bypass 4': f'https://{origin}.-.{malicious_domain}',
            'Bypass 5': f'https://{origin}.,.{malicious_domain}',
            'Bypass 6': f'https://{origin}.;.{malicious_domain}',
            'Bypass 7': f'https://{origin}.!.{malicious_domain}',
            'Bypass 8': f"https://{origin}.' .{malicious_domain}",
            'Bypass 9': f'https://{origin}".{malicious_domain}',
            'Bypass 10': f'https://{origin}.({malicious_domain}',
            'Bypass 11': f'https://{origin}.){malicious_domain}',
            'Bypass 12': f'https://{origin}' + '.{' + f'{malicious_domain}',
            'Bypass 13': f'https://{origin}' + '.}' + f'{malicious_domain}',
            'Bypass 14': f'https://{origin}.*.{malicious_domain}',
            'Bypass 15': f'https://{origin}.&.{malicious_domain}',
            'Bypass 16': f'https://{origin}.`.{malicious_domain}',
            'Bypass 17': f'https://{origin}.+.{malicious_domain}',
            'Bypass 18': f'https://{origin}.{malicious_domain}',
            'Bypass 19': f'https://{origin}.=.{malicious_domain}',
            'Bypass 20': f'https://{origin}.~.{malicious_domain}',
            'Bypass 21': f'https://{origin}.$.{malicious_domain}',
            'Bypass 22': f'http://s{origin}',
            'Bypass 23': f'https://{origin.replace(".", "x")}',
            'Regexp bypass 1': f'{origin},.{malicious_domain}',
            'Regexp bypass 2': f'{origin}&.{malicious_domain}',
            'Regexp bypass 3': f"{origin}'.{malicious_domain}",
            'Regexp bypass 4': f'{origin}".{malicious_domain}',
            'Regexp bypass 5': f'{origin};.{malicious_domain}',
            'Regexp bypass 6': f'{origin}!.{malicious_domain}',
            'Regexp bypass 7': f'{origin}$.{malicious_domain}',
            'Regexp bypass 8': f'{origin}^.{malicious_domain}',
            'Regexp bypass 9': f'{origin}*.{malicious_domain}',
            'Regexp bypass 10': f'{origin}(.{malicious_domain}',
            'Regexp bypass 11': f'{origin}).{malicious_domain}',
            'Regexp bypass 12': f'{origin}+.{malicious_domain}',
            'Regexp bypass 13': f'{origin}=.{malicious_domain}',
            'Regexp bypass 14': f'{origin}`.{malicious_domain}',
            'Regexp bypass 15': f'{origin}~.{malicious_domain}',
            'Regexp bypass 16': f'{origin}-.{malicious_domain}',
            'Regexp bypass 17': f'{origin}_.{malicious_domain}',
            'Regexp bypass 18': f'{origin}|.{malicious_domain}',
            'Regexp bypass 19': f'https://{origin}' + '.{' + f'{malicious_domain}',
            'Regexp bypass 21': f'{origin}%.{malicious_domain}',
        }


class CORSVulnerabilityScanner:

    def __init__(self, config: ScanConfig):
        """
        Initialize scanner.
        
        Args:
            config: Scan configuration
        """
        self.config = config
        self.logger = logger  # FIX C1
        self.results: List[ScanResult] = []  # NOTE: do not reuse scanner across URLs  # FIX C2
        self.vulnerable_results: List[ScanResult] = []
        self.error_count: int = 0
        self.http_status_codes: Dict[int, int] = {}

    async def _test_bypass_async(
        self,
        session: aiohttp.ClientSession,
        url: str,
        bypass_name: str,
        bypass_value: str,
    ) -> ScanResult:  # PERF 1
        headers = {"Origin": bypass_value}
        if self.config.custom_headers:
            headers.update({k: v for k, v in self.config.custom_headers.items()})

        try:
            async with session.request(
                self.config.method,
                url,
                headers=headers,
                proxy=self._proxy_url(),
                allow_redirects=False,
            ) as resp:
                acac = resp.headers.get("Access-Control-Allow-Credentials")
                acao = resp.headers.get("Access-Control-Allow-Origin")
                is_vulnerable = acac == "true" and acao == bypass_value

                with log_lock:  # PERF 1
                    self.http_status_codes[resp.status] = (
                        self.http_status_codes.get(resp.status, 0) + 1
                    )

                result = ScanResult(
                    url=url,
                    bypass_name=bypass_name,
                    bypass_value=bypass_value,
                    is_vulnerable=is_vulnerable,
                    response_code=resp.status,
                    acac=acac,
                    acao=acao,
                )

        except asyncio.TimeoutError:  # PERF 1
            result = ScanResult(
                url=url,
                bypass_name=bypass_name,
                bypass_value=bypass_value,
                is_vulnerable=False,
                error="Timeout",
            )
            self.error_count += 1
        except aiohttp.ClientError as exc:  # PERF 1
            result = ScanResult(
                url=url,
                bypass_name=bypass_name,
                bypass_value=bypass_value,
                is_vulnerable=False,
                error=str(exc),
            )
            self.error_count += 1

        self._print_result(result)
        if self.config.rate_limit:
            await asyncio.sleep(self.config.rate_limit)  # PERF 1
        return result

    @asynccontextmanager
    async def _make_session(self):  # PERF 3
        connector = TCPConnector(
            ssl=False,
            use_dns_cache=True,
            ttl_dns_cache=300,
            limit=self.config.max_workers * 2,
            resolver=aiohttp.AsyncResolver(),
        )  # PERF 2
        timeout = ClientTimeout(total=self.config.timeout, connect=5)
        default_headers = {
            "User-Agent": "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:122.0) Gecko/20100101 Firefox/122.0",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
            "Accept-Language": "en-US,en;q=0.9",
            "Accept-Encoding": "gzip, deflate",
            "Connection": "keep-alive",
        }
        async with aiohttp.ClientSession(
            connector=connector,
            headers=default_headers,
            timeout=timeout,
        ) as session:
            yield session

    def _proxy_url(self) -> Optional[str]:  # PERF 3
        if not self.config.proxy:
            return None
        return self.config.proxy.get("https") or self.config.proxy.get("http")

    def scan(self, urls: Optional[List[str]] = None) -> Tuple[List[ScanResult], int]:  # PERF 1
        return asyncio.run(self._async_scan(urls or [self.config.url]))

    async def _async_scan(self, urls: List[str]) -> Tuple[List[ScanResult], int]:  # PERF 1
        if self.config.verbose:
            self.logger.info(f"Using {self.config.max_workers} workers")

        async with self._make_session() as session:
            for url in urls:
                url = unquote(url, encoding="utf-8")
                origin = urlparse(url).netloc

                if self.config.verbose:
                    self.logger.info(f"Starting scan on {url}")

                payloads = CORSBypassPayloads.generate(
                    origin, self.config.custom_domain
                )

                semaphore = asyncio.Semaphore(self.config.max_workers)

                async def bounded(name: str, value: str) -> ScanResult:
                    async with semaphore:
                        return await self._test_bypass_async(
                            session, url, name, value
                        )

                if self.config.stop_on_first:
                    for name, value in payloads.items():
                        result = await self._test_bypass_async(
                            session, url, name, value
                        )
                        self.results.append(result)
                        if result.is_vulnerable:
                            self.vulnerable_results.append(result)
                            break
                else:
                    tasks = [bounded(name, value) for name, value in payloads.items()]
                    url_results = await asyncio.gather(*tasks)
                    self.results.extend(url_results)
                    self.vulnerable_results.extend(
                        [r for r in url_results if r.is_vulnerable]
                    )

        return self.results, len(self.vulnerable_results)

    def _print_result(self, result: ScanResult) -> None:  # FIX C3
        """
        Print result with appropriate formatting.
        
        Args:
            result: ScanResult to print
        """
        if self.config.vulnerable_only and not result.is_vulnerable:
            return

        with output_lock:
            status = '[VULNERABLE]' if result.is_vulnerable else '[SAFE]'
            output = f"{status} {result.bypass_name}: {result.bypass_value}"
            
            if self.config.no_color:
                print(output)
            else:
                color = Fore.GREEN if result.is_vulnerable else Fore.RED
                print(f"{color}{output}{Style.RESET_ALL}")

    def _generate_sarif_report(self) -> Dict:  # FIX C3
        """Generate SARIF (Static Analysis Results Interchange Format) report."""
        sarif_results = []
        
        for result in self.results:
            if self.config.vulnerable_only and not result.is_vulnerable:
                continue
            
            # Determine severity and level
            if result.is_vulnerable:
                level = "warning"
                message = f"CORS misconfiguration detected via {result.bypass_name}"
            else:
                level = "note"
                message = f"No CORS misconfiguration found for {result.bypass_name}"
            
            sarif_result = {
                "ruleId": "cors-bypass",
                "level": level,
                "message": {
                    "text": message
                },
                "locations": [
                    {
                        "physicalLocation": {
                            "address": {
                                "uri": result.url
                            }
                        }
                    }
                ],
                "properties": {
                    "bypass_name": result.bypass_name,
                    "bypass_value": result.bypass_value,
                    "response_code": result.response_code,
                    "access_control_allow_credentials": result.acac,
                    "access_control_allow_origin": result.acao,
                    "vulnerability_type": "CORS",
                    "timestamp": str(result.timestamp)
                }
            }
            
            if result.error:
                sarif_result["properties"]["error"] = result.error
            
            sarif_results.append(sarif_result)
        
        # Build SARIF report structure
        sarif_report = {
            "$schema": "https://raw.githubusercontent.com/oasis-tcs/sarif-spec/master/Schemata/sarif-schema-2.1.0.json",
            "version": "2.1.0",
            "runs": [
                {
                    "tool": {
                        "driver": {
                            "name": "Security Scanner",
                            "version": "1.1.0",
                            "informationUri": "https://github.com/omranisecurity/CorsOne",
                            "rules": [
                                {
                                    "id": "cors-bypass",
                                    "name": "CORS Misconfiguration",
                                    "shortDescription": {
                                        "text": "Detection of Cross-Origin Resource Sharing (CORS) configuration vulnerabilities"
                                    },
                                    "fullDescription": {
                                        "text": "Tests for CORS misconfigurations that could allow unauthorized cross-origin requests with credentials"
                                    },
                                    "defaultConfiguration": {
                                        "level": "warning"
                                    },
                                    "properties": {
                                        "category": "Security",
                                        "tags": ["cors", "security", "misconfiguration"]
                                    }
                                }
                            ]
                        }
                    },
                    "results": sarif_results
                }
            ]
        }
        
        return sarif_report

    def save_results(self) -> None:  # FIX C3
        """Save results to output file in specified format."""
        if not self.config.output_file:
            return
        
        try:
            output_path = Path(self.config.output_file)
            output_format = self.config.output_format.lower()

            if output_format not in ['json', 'txt', 'sarif']:
                self.logger.error(f"Unsupported format: {output_format}. Use 'txt', 'json', or 'sarif'.")
                return

            final_path = output_path.with_suffix('.json' if output_format in ['json', 'sarif'] else '.txt')

            if output_format == 'sarif':
                sarif_report = self._generate_sarif_report()
                with open(final_path, 'w') as f:
                    json.dump(sarif_report, f, indent=2, default=str)
                self.logger.info(f"Results saved to {final_path}")
            elif output_format == 'json':
                results_to_save = [r.to_dict() for r in self.results if not self.config.vulnerable_only or r.is_vulnerable]
                with open(final_path, 'w') as f:
                    json.dump(
                        results_to_save,
                        f,
                        indent=2,
                        default=str
                    )
                self.logger.info(f"Results saved to {final_path}")
            else:
                with open(final_path, 'w') as f:
                    if self.config.vulnerable_only:
                        results_to_save = [r for r in self.results if r.is_vulnerable]
                    elif self.config.stop_on_first and self.vulnerable_results:
                        results_to_save = [self.vulnerable_results[0]]
                    else:
                        results_to_save = self.results

                    for result in results_to_save:
                        f.write(str(result) + '\n')
                self.logger.info(f"Results saved to {final_path}")

        except IOError as e:
            self.logger.error(f"Failed to save results: {e}")

    def print_summary(self) -> None:  # FIX C3
        """Print scan summary."""
        total = len(self.results)
        vulnerable = len(self.vulnerable_results)
        safe = total - vulnerable - self.error_count
        
        # Helper function to apply color if enabled
        def colorize(text: str, color) -> str:
            if self.config.no_color:
                return text
            return f"{color}{text}{Style.RESET_ALL}"
        
        print(f"\n{'='*70}")
        print(f"{'SCAN SUMMARY':^70}")
        print(f"{'='*70}")
        print(f"Total tests:          {total}")
        print(f"Vulnerable:           {colorize(str(vulnerable), Fore.GREEN)}")
        print(f"Safe:                 {colorize(str(safe), Fore.RED)}")
        print(f"Errors:               {colorize(str(self.error_count), Fore.YELLOW)}")
        
        if self.http_status_codes:
            print(f"\n{colorize('HTTP Status Codes:', Fore.CYAN)}")
            for status_code in sorted(self.http_status_codes.keys()):
                count = self.http_status_codes[status_code]
                print(f"  • {status_code}: {count}")
        
        if vulnerable > 0:
            print(f"\n{colorize('Vulnerable bypasses:', Fore.GREEN)}")
            for result in self.vulnerable_results:
                print(f"  • {result.bypass_name}: {result.bypass_value}")
        
        print(f"{'='*70}\n")


class URLValidator:
    """URL validation and normalization."""

    @staticmethod
    def validate(url: str) -> str:
        """
        Validate and normalize URL.
        
        Args:
            url: URL to validate
            
        Returns:
            Validated URL
            
        Raises:
            ValueError: If URL is invalid
        """
        url = url.strip()
        
        if validators.url(url):
            return url
        elif validators.domain(url):
            return f'https://{url}'
        else:
            raise ValueError(f"Invalid URL: {url}")


class DomainValidator:
    """Validates that custom domain is a base domain without protocol or subdomain."""

    @staticmethod
    def validate(domain: str) -> str:
        """
        Validate custom domain.
        
        Args:
            domain: Domain to validate
            
        Returns:
            Validated domain
            
        Raises:
            ValueError: If domain is invalid
        """
        domain = domain.strip()
        
        # Check for protocol prefixes
        if domain.startswith('http://') or domain.startswith('https://'):
            raise ValueError("Custom domain should not include 'http://' or 'https://' protocol")
        
        # Check if it contains subdomain (multiple dots)
        parts = domain.split('.')
        if len(parts) < 2:
            raise ValueError("Custom domain must be a valid domain (e.g., example.com)")
        
        # Validate domain format
        if validators.domain(domain):
            return domain
        else:
            raise ValueError(f"Invalid domain format: {domain}")


def create_argument_parser() -> argparse.ArgumentParser:
    """
    Create and configure argument parser.
    
    Returns:
        Configured ArgumentParser
    """
    parser = argparse.ArgumentParser(
        prog='CorsOne',
        description='CORS Misconfiguration Discovery Tool',
        epilog='Version: 1.1.0 | https://github.com/omranisecurity/CorsOne',
        formatter_class=argparse.RawDescriptionHelpFormatter
    )
    
    # Target input
    input_group = parser.add_mutually_exclusive_group()
    input_group.add_argument('-u', '--url', help='Target URL to scan')
    input_group.add_argument('-l', '--list', help='File with URLs (one per line)')
    
    # Scanning options
    parser.add_argument('-m', '--method', choices=['GET', 'POST'], default='GET',
                       help='HTTP method (default: GET)')
    parser.add_argument('-sof', '--stop-on-first', action='store_true',
                       help='Stop after finding first vulnerability')
    parser.add_argument('-cd', '--custom-domain', default='attacker.com',
                       help='Custom domain for payloads (default: attacker.com)')
    parser.add_argument('-H', '--headers', help='Custom headers as JSON (e.g., \'{"Cookie": "session=abc123"}\')')
    parser.add_argument('-p', '--proxy', help='Proxy URL (socks5://host:port)')
    
    # Performance options
    parser.add_argument('-w', '--workers', type=int, default=5,
                       help='Number of concurrent workers (default: 5)')
    parser.add_argument('-rl', '--rate-limit', type=float, default=0,
                       help='Delay between requests in seconds (default: 0)')
    parser.add_argument('-t', '--timeout', type=int, default=10,
                       help='Request timeout in seconds (default: 10)')
    parser.add_argument('-r', '--retries', type=int, default=3,
                       help='Number of retries for failed requests (default: 3)')
    
    # Output options
    parser.add_argument('-o', '--output', help='Output file for results')
    parser.add_argument('-f', '--format', choices=['txt', 'json', 'sarif'], default='txt',
                       help='Output format (txt, json, or sarif). Default: txt. Explicit --format always takes precedence over output file extension.')
    parser.add_argument('--log', help='Log file path. Only creates log file if specified.')
    parser.add_argument('-vo', '--vuln-only', action='store_true',
                       help='Show and save only vulnerable endpoints')
    parser.add_argument('-nc', '--no-color', action='store_true',
                       help='Disable colored output')
    parser.add_argument('-s', '--silent', action='store_true',
                       help='Silent mode (no banner)')
    parser.add_argument('-v', '--verbose', action='store_true',
                       help='Verbose logging')
    
    # Utility options
    parser.add_argument('--version', action='store_true', help='Show version')
    
    return parser


def print_banner() -> None:  # FIX C3
    """Print tool banner."""
    banner_text = """
    ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
    ┃                                                          ┃
    ┃                        CorsOne                           ┃
    ┃         CORS Misconfiguration Discovery Tool v1.0        ┃
    ┃                                                          ┃
    ┃               Fast | Reliable | Feature-Rich             ┃
    ┃                                                          ┃
    ┃        https://github.com/omranisecurity/CorsOne         ┃
    ┃                                                          ┃
    ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
    """
    print(banner_text)


def main() -> None:  # FIX C3
    """Main entry point."""
    parser = create_argument_parser()
    args = parser.parse_args()
    
    # Handle version
    if args.version:
        print("CorsOne v1.1.0")
        sys.exit(0)
    
    # Setup logging
    _setup_logging(verbose=args.verbose, log_file=args.log)
    
    # Print banner unless silent
    if not args.silent:
        print_banner()
    
    # Collect URLs
    urls: List[str] = []
    
    if args.url:
        urls = [args.url]
    elif args.list:
        try:
            with open(args.list, 'r') as f:
                urls = [line.strip() for line in f if line.strip()]
        except IOError as e:
            logger.error(f"Failed to read URL list: {e}")
            sys.exit(1)
    elif not sys.stdin.isatty():
        urls = [line.strip() for line in sys.stdin if line.strip()]
    else:
        parser.print_help()
        sys.exit(1)
    
    # Parse custom headers if provided
    custom_headers: Optional[Dict[str, str]] = None
    if args.headers:
        try:
            custom_headers = json.loads(args.headers)
        except json.JSONDecodeError:
            logger.error("Invalid JSON for headers. Use format: '{\"Header-Name\": \"value\"}'")
            sys.exit(1)
    
    # Parse proxy if provided
    proxy: Optional[Dict[str, str]] = None
    if args.proxy:
        proxy = {'http': args.proxy, 'https': args.proxy}
    
    # Validate custom domain
    try:
        custom_domain = DomainValidator.validate(args.custom_domain)
    except ValueError as e:
        logger.error(f"{e}")
        sys.exit(1)

    validated_urls: List[str] = []
    for url_input in urls:
        try:
            validated_urls.append(URLValidator.validate(url_input))
        except ValueError as e:
            logger.error(f"{e}")
    
    if not validated_urls:
        logger.error("No valid URLs to scan")
        sys.exit(1)
    
    # Create configuration using the first URL as a placeholder
    config = ScanConfig(
        url=validated_urls[0],
        method=args.method,
        custom_domain=custom_domain,
        rate_limit=args.rate_limit,
        timeout=args.timeout,
        retries=args.retries,
        max_workers=args.workers,
        stop_on_first=args.stop_on_first,
        no_color=args.no_color,
        output_file=args.output,
        output_format=args.format,
        output_log=args.log,
        custom_headers=custom_headers,
        proxy=proxy,
        verbose=args.verbose,
        vulnerable_only=args.vuln_only
    )
    
    # Run scanner once for all URLs with a shared session
    scanner = CORSVulnerabilityScanner(config)
    results, vulnerable_count = scanner.scan(validated_urls)
    
    # Save and display results
    scanner.save_results()
    if not args.silent:
        scanner.print_summary()


if __name__ == '__main__':
    main()