Mensaje import osimport ssl

import osimport ssl import socketimport threading import subprocessfrom http.server import HTTPServer, BaseHTTPRequestHandler from cryptography import x509from cryptography.x509.oid import NameOID from cryptography.hazmat.primitives import hashes, serializationfrom cryptography.hazmat.primitives.asymmetric import rsa from datetime import datetime, timedelta, timezone # General ConfigurationCERT_DIR = "C:\\certs" CA_CERT_FILE = os.path.join(CERT_DIR, "ca_cert.pem")CA_KEY_FILE = os.path.join(CERT_DIR, "ca_key.pem") SSL_TIMEOUT = 15 # Timeout for SSL connectionsMAX_RETRIES = 3 # Number of retries for connections def log_message(message): print(f"[LOG] {message}") # 1. Generate CA Certificate and Key def generate_ca(): if not os.path.exists(CERT_DIR): os.makedirs(CERT_DIR) log_message("Generating CA certificate and key...") ca_key = rsa.generate_private_key(public_exponent=65537, key_size=4096) subject = x509.Name([ x509.NameAttribute(NameOID.COUNTRY_NAME, "US"), x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Advanced Proxy Authority"), x509.NameAttribute(NameOID.COMMON_NAME, "Advanced Proxy Root CA"), ]) cert = ( x509.CertificateBuilder() .subject_name(subject) .issuer_name(subject) .public_key(ca_key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(datetime.now(timezone.utc)) .not_valid_after(datetime.now(timezone.utc) + timedelta(days=3650)) .sign(ca_key, hashes.SHA256()) ) with open(CA_CERT_FILE, "wb") as f: f.write(cert.public_bytes(serialization.Encoding.PEM)) with open(CA_KEY_FILE, "wb") as f: f.write(ca_key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL, serialization.NoEncryption() )) subprocess.run(["certutil", "-addstore", "Root", CA_CERT_FILE]) log_message("CA certificate installed successfully.") # 2. Proxy Handler class ProxyHandler(BaseHTTPRequestHandler): def do_CONNECT(self): host, port = self.path.split(":") port = int(port) log_message(f"Intercepting HTTPS for {host}") cert_file, key_file = self.generate_cert(host) retries = 0 while retries < MAX_RETRIES: try: with socket.create_connection((host, port), timeout=SSL_TIMEOUT) as remote_socket: self.send_response(200, "Connection Established") self.end_headers() context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) context.minimum_version = ssl.TLSVersion.TLSv1_2 context.load_cert_chain(certfile=cert_file, keyfile=key_file) with context.wrap_socket(self.connection, server_side=True) as client_socket: self.forward_data(client_socket, remote_socket) return except Exception as e: retries += 1 log_message(f"Error during connection ({retries}/{MAX_RETRIES}): {e}") if retries >= MAX_RETRIES: self.send_error(502, "Bad Gateway") def generate_cert(self, domain): cert_file = os.path.join(CERT_DIR, f"{domain}.pem") key_file = os.path.join(CERT_DIR, f"{domain}_key.pem") if os.path.exists(cert_file) and os.path.exists(key_file): log_message(f"Using existing certificate for {domain}.") return cert_file, key_file with open(CA_KEY_FILE, "rb") as f: ca_key = serialization.load_pem_private_key(f.read(), None) with open(CA_CERT_FILE, "rb") as f: ca_cert = x509.load_pem_x509_certificate(f.read()) key = rsa.generate_private_key(public_exponent=65537, key_size=2048) subject = x509.Name([ x509.NameAttribute(NameOID.COMMON_NAME, domain) ]) san = x509.SubjectAlternativeName([ x509.DNSName(domain) ]) cert = ( x509.CertificateBuilder()