Spaces:
Paused
Paused
File size: 3,913 Bytes
a5784e9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 | import argparse
import asyncio
import logging
import sys
from pathlib import Path
from typing import Any, Optional
from logging_utils import GridFormatter, set_source
from stream.proxy_server import ProxyServer
def parse_args() -> argparse.Namespace:
"""Parse command line arguments"""
parser = argparse.ArgumentParser(
description="HTTPS Proxy Server with SSL Inspection"
)
parser.add_argument(
"--host", default="127.0.0.1", help="Host to bind the proxy server"
)
parser.add_argument(
"--port", type=int, default=3120, help="Port to bind the proxy server"
)
parser.add_argument(
"--domains",
nargs="+",
default=["*.google.com"],
help="List of domain patterns to intercept (regex)",
)
parser.add_argument(
"--proxy", help="Upstream proxy URL (e.g., http://user:pass@host:port)"
)
return parser.parse_args()
async def main() -> None:
"""Main entry point"""
args = parse_args()
# Set up logging with GridFormatter for consistent output
set_source("PROXY")
console_handler = logging.StreamHandler(sys.stderr)
console_handler.setFormatter(GridFormatter(show_tree=True, colorize=True))
console_handler.setLevel(logging.INFO)
# Configure proxy_server logger specifically (not root)
logger = logging.getLogger("proxy_server")
logger.handlers.clear() # Remove any existing handlers
logger.addHandler(console_handler)
logger.setLevel(logging.INFO)
logger.propagate = False # Prevent double logging
logging.getLogger("asyncio").setLevel(logging.ERROR)
logging.getLogger("websockets").setLevel(logging.ERROR)
# Create certs directory
cert_dir = Path("certs")
cert_dir.mkdir(exist_ok=True)
# Print startup information
logger.info(f"Starting proxy server on {args.host}:{args.port}")
logger.info(f"Intercepting domains: {args.domains}")
if args.proxy:
logger.info(f"Using upstream proxy: {args.proxy}")
# Create and start the proxy server
proxy_server = ProxyServer(
host=args.host,
port=args.port,
intercept_domains=args.domains,
upstream_proxy=args.proxy,
queue=None,
)
try:
await proxy_server.start()
except KeyboardInterrupt:
logger.info("Shutting down proxy server")
except asyncio.CancelledError:
raise
except Exception as e:
logger.error(f"Error starting proxy server: {e}", exc_info=True)
sys.exit(1)
async def builtin(
queue: Optional[Any] = None, port: Optional[int] = None, proxy: Optional[str] = None
) -> None:
# Set up logging with GridFormatter for consistent output
set_source("PROXY")
console_handler = logging.StreamHandler(sys.stderr)
console_handler.setFormatter(GridFormatter(show_tree=True, colorize=True))
console_handler.setLevel(logging.INFO)
# Configure proxy_server logger specifically (not root)
logger = logging.getLogger("proxy_server")
logger.handlers.clear() # Remove any existing handlers
logger.addHandler(console_handler)
logger.setLevel(logging.INFO)
logger.propagate = False # Prevent double logging
# Create certs directory
cert_dir = Path("certs")
cert_dir.mkdir(exist_ok=True)
if port is None:
port = 3120
# Create and start the proxy server
proxy_server = ProxyServer(
host="127.0.0.1",
port=port,
intercept_domains=["*.google.com"],
upstream_proxy=proxy,
queue=queue,
)
try:
await proxy_server.start()
except KeyboardInterrupt:
logger.info("Shutting down proxy server")
except asyncio.CancelledError:
raise
except Exception as e:
logger.error(f"Error starting proxy server: {e}", exc_info=True)
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
|