proxy.py

#!/usr/bin/env python3
"""
Transparent JS-rendering proxy for text browsers like w3m.
Routes requests through Lightpanda to render JavaScript before returning HTML.
"""

import argparse
import hashlib
import os
import re
import shutil
import subprocess
import time
from http.server import HTTPServer, BaseHTTPRequestHandler
from pathlib import Path
from urllib.parse import urljoin, urlparse

# Cache settings
CACHE_DIR = Path.home() / ".cache" / "js-proxy"
CACHE_TTL = 300  # 5 minutes default

class JSRenderProxy(BaseHTTPRequestHandler):
    lightpanda_path = "lightpanda"
    use_cache = True
    cache_ttl = CACHE_TTL
    timeout = 30
    verbose = False
    proxy_base = "http://127.0.0.1:8080"  # Set by run_proxy

    def log_message(self, format, *args):
        if self.verbose:
            super().log_message(format, *args)


    def do_GET(self):
        try:
            self.handle_request()
        except BrokenPipeError:
            pass  # Client disconnected

    def do_POST(self):
        try:
            self.handle_post()
        except BrokenPipeError:
            pass  # Client disconnected

    def handle_post(self):
        """Forward POST requests via curl since Lightpanda is GET-only."""
        path = self.path

        if path.startswith('/http://') or path.startswith('/https://'):
            url = path[1:]
        elif path.startswith('http://') or path.startswith('https://'):
            url = path
        else:
            host = self.headers.get('Host', '')
            if host:
                url = f"http://{host}{path}"
            else:
                self.send_error(400, "Cannot determine target URL")
                return

        content_length = int(self.headers.get('Content-Length', 0))
        post_data = self.rfile.read(content_length) if content_length > 0 else b''
        content_type = self.headers.get('Content-Type', 'application/x-www-form-urlencoded')

        try:
            result = subprocess.run(
                ['curl', '-sL', '-X', 'POST',
                 '-H', f'Content-Type: {content_type}',
                 '-A', 'Mozilla/5.0',
                 '--data-raw', post_data.decode('utf-8', errors='replace'),
                 url],
                capture_output=True,
                text=True,
                timeout=15
            )
            if result.returncode == 0 and result.stdout:
                html = self.rewrite_urls(result.stdout, url)
                self.send_response(200)
                self.send_header('Content-Type', 'text/html; charset=utf-8')
                self.send_header('Content-Length', str(len(html.encode('utf-8'))))
                self.end_headers()
                self.wfile.write(html.encode('utf-8'))
                return
        except Exception:
            pass

        self.send_error(502, "Failed to forward POST request")

    def do_CONNECT(self):
        """Handle HTTPS CONNECT tunneling - render via Lightpanda"""
        # Extract host:port from path
        host_port = self.path
        if ':' not in host_port:
            host_port += ':443'

        url = f"https://{host_port.split(':')[0]}/"

        # Send 200 Connection established
        self.send_response(200, "Connection established")
        self.end_headers()

        # For CONNECT, we can't easily intercept SSL
        # Instead, suggest using the direct URL mode
        self.wfile.write(b"HTTPS tunneling not supported in render mode.\n")
        self.wfile.write(f"Use: http://localhost:PORT/{url}\n".encode())

    def handle_request(self):
        # Parse the URL - support both proxy mode and direct URL mode
        # Direct mode: http://localhost:8080/https://github.com
        # Proxy mode: configured HTTP_PROXY, requesting http://example.com

        path = self.path

        # Direct URL mode: /https://... or /http://...
        if path.startswith('/http://') or path.startswith('/https://'):
            url = path[1:]  # strip leading /
        elif path.startswith('http://') or path.startswith('https://'):
            # Proxy mode
            url = path
        else:
            # Relative path with Host header (proxy mode)
            host = self.headers.get('Host', '')
            if host:
                url = f"http://{host}{path}"
            else:
                self.send_error(400, "Cannot determine target URL")
                return

        # Check cache
        cache_file = self.get_cache_file(url)
        if self.use_cache and cache_file.exists():
            age = time.time() - cache_file.stat().st_mtime
            if age < self.cache_ttl:
                if self.verbose:
                    print(f"[CACHE HIT] {url}")
                self.serve_cached(cache_file)
                return

        # Render with Lightpanda
        if self.verbose:
            print(f"[RENDER] {url}")

        html = self.render_with_lightpanda(url)

        if html is None:
            self.send_error(502, "Failed to render page")
            return

        # Rewrite URLs to go through proxy
        html = self.rewrite_urls(html, url)

        # Cache the result (with rewritten URLs)
        if self.use_cache:
            CACHE_DIR.mkdir(parents=True, exist_ok=True)
            cache_file.write_text(html)

        # Send response
        self.send_response(200)
        self.send_header('Content-Type', 'text/html; charset=utf-8')
        self.send_header('Content-Length', str(len(html.encode('utf-8'))))
        self.send_header('X-Rendered-By', 'Lightpanda')
        self.end_headers()
        self.wfile.write(html.encode('utf-8'))

    def get_cache_file(self, url):
        """Generate cache filename from URL"""
        url_hash = hashlib.md5(url.encode()).hexdigest()
        return CACHE_DIR / f"{url_hash}.html"

    def serve_cached(self, cache_file):
        """Serve content from cache"""
        html = cache_file.read_text()
        self.send_response(200)
        self.send_header('Content-Type', 'text/html; charset=utf-8')
        self.send_header('Content-Length', str(len(html.encode('utf-8'))))
        self.send_header('X-Rendered-By', 'Lightpanda (cached)')
        self.end_headers()
        self.wfile.write(html.encode('utf-8'))

    def render_with_lightpanda(self, url):
        """Render URL using Lightpanda and return HTML"""
        try:
            result = subprocess.run(
                [self.lightpanda_path, 'fetch', '--dump', url],
                capture_output=True,
                text=True,
                timeout=self.timeout,
                env={**os.environ, 'LIGHTPANDA_DISABLE_TELEMETRY': 'true'}
            )

            if result.returncode == 0:
                return result.stdout
            else:
                if self.verbose:
                    print(f"[ERROR] Lightpanda failed: {result.stderr}")
                # Try fallback to curl for simple pages
                return self.fallback_curl(url)

        except subprocess.TimeoutExpired:
            if self.verbose:
                print(f"[TIMEOUT] {url}")
            return self.fallback_curl(url)
        except FileNotFoundError:
            print(f"[ERROR] Lightpanda not found at: {self.lightpanda_path}")
            print("Run setup.sh to install or specify path with --lightpanda")
            return self.fallback_curl(url)

    def fallback_curl(self, url):
        """Fallback to curl for pages that don't need JS"""
        try:
            result = subprocess.run(
                ['curl', '-sL', '-A', 'Mozilla/5.0', url],
                capture_output=True,
                text=True,
                timeout=15
            )
            if result.returncode == 0:
                return result.stdout
        except:
            pass
        return None

    def rewrite_urls(self, html, base_url):
        """Rewrite all URLs in HTML to go through the proxy."""
        parsed_base = urlparse(base_url)
        base_origin = f"{parsed_base.scheme}://{parsed_base.netloc}"

        def make_proxied_url(url):
            """Convert a URL to go through the proxy."""
            url = url.strip()

            # Skip empty, javascript:, data:, mailto:, #anchors, template vars
            if not url or url.startswith(('javascript:', 'data:', 'mailto:', '#', '{')):
                return None

            # Skip already-proxied URLs
            if url.startswith(self.proxy_base):
                return None

            # Convert to absolute URL
            if url.startswith('//'):
                # Protocol-relative URL
                absolute_url = f"{parsed_base.scheme}:{url}"
            elif url.startswith('/'):
                # Root-relative URL
                absolute_url = f"{base_origin}{url}"
            elif url.startswith(('http://', 'https://')):
                # Already absolute
                absolute_url = url
            else:
                # Relative URL
                absolute_url = urljoin(base_url, url)

            return f"{self.proxy_base}/{absolute_url}"

        def make_absolute_url(url):
            """Convert a relative URL to absolute (no proxy prefix)."""
            url = url.strip()
            if not url or url.startswith(('javascript:', 'data:', 'mailto:', '#', '{')):
                return None
            if url.startswith(('http://', 'https://')):
                return None  # already absolute
            if url.startswith('//'):
                return f"{parsed_base.scheme}:{url}"
            elif url.startswith('/'):
                return f"{base_origin}{url}"
            else:
                return urljoin(base_url, url)

        def rewrite_quoted(match):
            """Rewrite href/src/action with quoted values."""
            attr = match.group(1)
            quote = match.group(2)
            url = match.group(3)
            # Form actions: make absolute but don't proxy (let w3m submit directly)
            if attr.lower() == 'action':
                absolute = make_absolute_url(url)
                if absolute:
                    return f'{attr}={quote}{absolute}{quote}'
                return match.group(0)
            proxied = make_proxied_url(url)
            if proxied:
                return f'{attr}={quote}{proxied}{quote}'
            return match.group(0)

        def rewrite_unquoted(match):
            """Rewrite href/src/action with unquoted values."""
            attr = match.group(1)
            url = match.group(2)
            if attr.lower() == 'action':
                absolute = make_absolute_url(url)
                if absolute:
                    return f'{attr}="{absolute}"'
                return match.group(0)
            proxied = make_proxied_url(url)
            if proxied:
                return f'{attr}="{proxied}"'
            return match.group(0)

        # Rewrite quoted URLs: href="..." or href='...'
        pattern_quoted = r'(href|src|action)\s*=\s*(["\'])([^"\']*)\2'
        html = re.sub(pattern_quoted, rewrite_quoted, html, flags=re.IGNORECASE)

        # Rewrite unquoted URLs: href=/path (stop at space or >)
        pattern_unquoted = r'(href|src|action)\s*=\s*([^\s"\'<>]+)'
        html = re.sub(pattern_unquoted, rewrite_unquoted, html, flags=re.IGNORECASE)

        return html


def run_proxy(host='127.0.0.1', port=8080, **kwargs):
    """Start the proxy server"""
    # Configure handler class attributes
    for key, value in kwargs.items():
        if hasattr(JSRenderProxy, key):
            setattr(JSRenderProxy, key, value)

    # Set proxy base URL for URL rewriting
    JSRenderProxy.proxy_base = f"http://{host}:{port}"

    server = HTTPServer((host, port), JSRenderProxy)
    print(f"JS-Rendering Proxy started on http://{host}:{port}")
    print(f"")
    print(f"Usage with w3m:")
    print(f"  w3m http://{host}:{port}/https://github.com")
    print(f"")
    print(f"Or set as HTTP proxy:")
    print(f"  export http_proxy=http://{host}:{port}")
    print(f"  w3m http://example.com")
    print(f"")
    print(f"Press Ctrl+C to stop")

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nShutting down...")
        server.shutdown()


def main():
    parser = argparse.ArgumentParser(
        description='Transparent JS-rendering proxy for text browsers'
    )
    parser.add_argument(
        '--host', default='127.0.0.1',
        help='Host to bind to (default: 127.0.0.1)'
    )
    parser.add_argument(
        '--port', '-p', type=int, default=8080,
        help='Port to listen on (default: 8080)'
    )
    parser.add_argument(
        '--lightpanda', default='lightpanda',
        help='Path to lightpanda binary'
    )
    parser.add_argument(
        '--no-cache', action='store_true',
        help='Disable caching'
    )
    parser.add_argument(
        '--cache-ttl', type=int, default=CACHE_TTL,
        help=f'Cache TTL in seconds (default: {CACHE_TTL})'
    )
    parser.add_argument(
        '--timeout', type=int, default=30,
        help='Render timeout in seconds (default: 30)'
    )
    parser.add_argument(
        '--verbose', '-v', action='store_true',
        help='Verbose logging'
    )
    parser.add_argument(
        '--clear-cache', action='store_true',
        help='Clear cache and exit'
    )

    args = parser.parse_args()

    if args.clear_cache:
        if CACHE_DIR.exists():
            shutil.rmtree(CACHE_DIR)
            print(f"Cleared cache at {CACHE_DIR}")
        else:
            print("Cache directory doesn't exist")
        return

    run_proxy(
        host=args.host,
        port=args.port,
        lightpanda_path=args.lightpanda,
        use_cache=not args.no_cache,
        cache_ttl=args.cache_ttl,
        timeout=args.timeout,
        verbose=args.verbose
    )


if __name__ == '__main__':
    main()