localportmanager.py

#!/usr/bin/env python3
"""
LocalPortManager: Zero-Dependency Local Reverse Proxy
======================================================

A lightweight, Python 3.8+ compatible reverse proxy for managing local services
dynamically without external dependencies.

Features:
- Dynamic port allocation (4000-4999)
- Service registry with persistent state
- HTTP reverse proxy with Host header routing
- Zero external dependencies (stdlib only)
- Thread-safe operations

Author: SHAdd0WTAka
License: MIT
"""

import socket
import threading
import json
import os
import sys
import argparse
import signal
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.request import urlopen, Request
from urllib.error import URLError
from typing import Optional, Dict


__version__ = "1.0.3"
__author__ = "SHAdd0WTAka"


class PortRegistry:
    """
    Manages service name to local port mappings with persistent state.

    Thread-safe registry that stores mappings in a JSON file for
    persistence across restarts.
    """

    def __init__(self, state_file: str = "/tmp/localportmanager_registry.json"):  # nosec B108
        self.state_file = state_file
        self.lock = threading.Lock()
        self.mappings: Dict[str, int] = self._load()

    def _load(self) -> Dict[str, int]:
        """Load mappings from state file."""
        if os.path.exists(self.state_file):
            try:
                with open(self.state_file, 'r') as f:
                    data = json.load(f)
                    return {k: int(v) if isinstance(v, (int, str)) else v
                            for k, v in data.items()}
            except (json.JSONDecodeError, ValueError, TypeError):
                return {}
        return {}

    def _save(self) -> None:
        """Save mappings to state file."""
        with open(self.state_file, 'w') as f:
            json.dump(self.mappings, f, indent=2)

    def register(self, name: str, port: int) -> None:
        """Register a service with a port."""
        with self.lock:
            self.mappings[name] = port
            self._save()

    def unregister(self, name: str) -> bool:
        """Unregister a service. Returns True if removed."""
        with self.lock:
            if name in self.mappings:
                del self.mappings[name]
                self._save()
                return True
            return False

    def get_port(self, name: str) -> Optional[int]:
        """Get port for a service name."""
        with self.lock:
            port = self.mappings.get(name)
            return int(port) if port is not None else None

    def list_services(self) -> Dict[str, int]:
        """Return all registered services."""
        with self.lock:
            return dict(self.mappings)

    def find_free_port(self, start: int = 4000, end: int = 4999) -> int:
        """Find first available port in range."""
        used_ports = set(self.mappings.values())

        for port in range(start, end + 1):
            if port in used_ports:
                continue
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                if s.connect_ex(('127.0.0.1', port)) != 0:
                    return port
        raise RuntimeError("No free ports available in range")


class ReverseProxyHandler(BaseHTTPRequestHandler):
    """HTTP request handler that proxies requests to backend services."""

    protocol_version = 'HTTP/1.1'

    def log_message(self, format: str, *args) -> None:
        """Custom logging with timestamp."""
        import datetime
        timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        sys.stderr.write(f"[{timestamp}] {format % args}\n")

    def do_GET(self) -> None:
        self._proxy_request('GET')

    def do_POST(self) -> None:
        self._proxy_request('POST')

    def do_PUT(self) -> None:
        self._proxy_request('PUT')

    def do_DELETE(self) -> None:
        self._proxy_request('DELETE')

    def do_PATCH(self) -> None:
        self._proxy_request('PATCH')

    def do_HEAD(self) -> None:
        self._proxy_request('HEAD')

    def do_OPTIONS(self) -> None:
        self._proxy_request('OPTIONS')

    def _proxy_request(self, method: str) -> None:
        """Proxy request to backend service."""
        host = self.headers.get('Host', '').split(':')[0]

        # Remove .localhost suffix
        if host.endswith('.localhost'):
            service_name = host.replace('.localhost', '')
        else:
            service_name = host

        # Path-based fallback routing
        if service_name in ['localhost', '127.0.0.1'] and self.path.startswith('/'):
            path_parts = self.path.strip('/').split('/')
            if path_parts and path_parts[0]:
                potential_service = path_parts[0]
                registry = PortRegistry()
                if registry.get_port(potential_service):
                    service_name = potential_service
                    self.path = '/' + '/'.join(path_parts[1:]) if len(path_parts) > 1 else '/'

        registry = PortRegistry()
        target_port = registry.get_port(service_name)

        if not target_port:
            self.send_error(404, f"Service '{service_name}' not registered")
            return

        # Read request body
        body = None
        if method in ['POST', 'PUT', 'PATCH']:
            content_length = self.headers.get('Content-Length')
            if content_length:
                body = self.rfile.read(int(content_length))

        # Proxy to backend
        try:
            backend_url = f"http://127.0.0.1:{target_port}{self.path}"
            req = Request(backend_url, method=method, data=body)

            # Copy headers (except hop-by-hop headers)
            hop_by_hop = ['connection', 'keep-alive', 'proxy-authenticate',
                         'proxy-authorization', 'te', 'trailers',
                         'transfer-encoding', 'upgrade', 'host']

            for header, value in self.headers.items():
                if header.lower() not in hop_by_hop:
                    req.add_header(header, value)

            with urlopen(req, timeout=30) as response:
                self.send_response(response.status)
                for header, value in response.headers.items():
                    if header.lower() not in ['transfer-encoding', 'connection',
                                               'keep-alive', 'upgrade']:
                        self.send_header(header, value)
                self.end_headers()
                data = response.read()
                if data:
                    self.wfile.write(data)

        except URLError as e:
            self.send_error(502, f"Backend unreachable: {e}")
        except Exception as e:
            self.send_error(500, f"Proxy error: {e}")


class LocalPortManager:
    """Main application class for LocalPortManager."""

    def __init__(self, proxy_port: int = 1355, state_file: Optional[str] = None):
        self.proxy_port = proxy_port
        self.registry = PortRegistry(state_file) if state_file else PortRegistry()
        self.server: Optional[ThreadingHTTPServer] = None
        self._shutdown_event = threading.Event()

    def start_proxy(self) -> None:
        """Start the reverse proxy server."""
        self.server = ThreadingHTTPServer(
            ('127.0.0.1', self.proxy_port),
            ReverseProxyHandler
        )

        print(f"[*] LocalPortManager Proxy running on http://127.0.0.1:{self.proxy_port}")
        print(f"[*] Registry file: {self.registry.state_file}")
        print(f"[*] Press Ctrl+C to stop")

        def signal_handler(signum, frame):
            print("\n[!] Shutting down...")
            self._shutdown_event.set()
            if self.server:
                self.server.shutdown()

        signal.signal(signal.SIGINT, signal_handler)
        signal.signal(signal.SIGTERM, signal_handler)

        try:
            self.server.serve_forever()
        except KeyboardInterrupt:
            pass
        finally:
            self.server.server_close()
            print("[*] Proxy stopped")

    def register_service(self, name: str, command: str, auto_start: bool = False) -> int:
        """Register a new service with dynamic port allocation."""
        if not name or not name.replace('-', '').replace('_', '').isalnum():
            raise ValueError("Service name must be alphanumeric with dashes/underscores only")

        port = self.registry.find_free_port()
        self.registry.register(name, port)

        if '{port}' in command:
            cmd = command.replace('{port}', str(port))
        else:
            cmd = f"{command} {port}"

        print(f"[+] Service '{name}' registered:")
        print(f"    Local:    http://127.0.0.1:{port}")
        print(f"    Proxy:    http://{name}.localhost:{self.proxy_port}")
        print(f"    Command:  {cmd}")

        if auto_start:
            print(f"[*] Starting service...")
            os.system(f"{cmd} &")
        else:
            response = input("Start service now? [y/N]: ").lower()
            if response == 'y':
                os.system(f"{cmd} &")

        return port

    def unregister_service(self, name: str) -> bool:
        """Unregister a service."""
        if self.registry.unregister(name):
            print(f"[+] Service '{name}' unregistered")
            return True
        else:
            print(f"[!] Service '{name}' not found")
            return False

    def list_services(self) -> None:
        """List all registered services."""
        services = self.registry.list_services()

        if not services:
            print("[*] No services registered")
            return

        print(f"\n{'Service':<20} {'Port':<8} {'Proxy URL'}")
        print("-" * 60)
        for name, port in sorted(services.items()):
            proxy_url = f"http://{name}.localhost:{self.proxy_port}"
            print(f"{name:<20} {port:<8} {proxy_url}")
        print()


def main():
    parser = argparse.ArgumentParser(
        description='LocalPortManager: Zero-Dependency Local Reverse Proxy',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s proxy                          # Start proxy server
  %(prog)s register webapp "python -m http.server {port}"  # Register service
  %(prog)s list                           # List all services
  %(prog)s unregister webapp              # Remove service
        """
    )
    parser.add_argument('--port', type=int, default=1355, help='Proxy port (default: 1355)')
    parser.add_argument('--state-file', type=str, default='/tmp/localportmanager_registry.json',
                       help='State file path')
    parser.add_argument('--version', action='version', version=f'%(prog)s {__version__}')

    subparsers = parser.add_subparsers(dest='command', help='Commands')

    subparsers.add_parser('proxy', help='Start the proxy server')

    reg_parser = subparsers.add_parser('register', help='Register a service')
    reg_parser.add_argument('name', help='Service name (e.g., webapp-01)')
    reg_parser.add_argument('command', help='Command with {port} placeholder')
    reg_parser.add_argument('-y', '--yes', action='store_true', help='Auto-start without prompting')

    unreg_parser = subparsers.add_parser('unregister', help='Unregister a service')
    unreg_parser.add_argument('name', help='Service name to remove')

    subparsers.add_parser('list', help='List registered services')
    subparsers.add_parser('status', help='Show proxy status')

    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        sys.exit(1)

    lpm = LocalPortManager(args.port, args.state_file)

    if args.command == 'proxy':
        print(f"[*] LocalPortManager v{__version__}")
        print("[*] Zero external dependencies - Python 3.8+")
        lpm.start_proxy()
    elif args.command == 'register':
        try:
            lpm.register_service(args.name, args.command, args.yes)
        except ValueError as e:
            print(f"[!] Error: {e}")
            sys.exit(1)
    elif args.command == 'unregister':
        lpm.unregister_service(args.name)
    elif args.command == 'list':
        lpm.list_services()
    elif args.command == 'status':
        print(f"[*] Proxy port: {args.port}")
        lpm.list_services()


if __name__ == "__main__":
    main()