oportunidades 7

tools/portal-country-https-proxy.py

@@ -0,0 +1,41 @@
+"""Loopback QA TLS termination, sanitized logs, no unsafe certificate bypass."""
+from pathlib import Path
+import json,ssl,threading,http.client
+from http.server import ThreadingHTTPServer,BaseHTTPRequestHandler
+ROOT=Path(__file__).resolve().parent.parent
+meta=json.loads((ROOT/'storage/app/private/investmoz-v2/portal-audit/country-https-latest.json').read_text())
+WORK=Path(meta['work'])
+class Proxy(BaseHTTPRequestHandler):
+    protocol_version='HTTP/1.1'
+    def log_message(self,*args):pass
+    def handle_request(self):
+        if self.headers.get('Host')!='country02-qa.test:18443':self.send_error(421);return
+        length=int(self.headers.get('Content-Length','0'))
+        if length>2_000_000:self.send_error(413);return
+        body=self.rfile.read(length) if length else None
+        headers={k:v for k,v in self.headers.items() if k.lower() not in ['connection','transfer-encoding','x-qa-tls-token']}
+        headers['X-QA-TLS-Token']=meta['qa_proxy_token']
+        c=http.client.HTTPConnection('127.0.0.1',18104,timeout=60)
+        try:
+            c.request(self.command,self.path,body,headers);r=c.getresponse();data=r.read()
+            self.send_response(r.status)
+            for k,v in r.getheaders():
+                if k.lower() not in ['connection','transfer-encoding','content-length','server','date']:self.send_header(k,v)
+            self.send_header('Content-Length',str(len(data)));self.end_headers();self.wfile.write(data)
+        except (BrokenPipeError,ConnectionResetError):pass
+        finally:c.close()
+    do_GET=do_POST=do_PUT=do_DELETE=do_PATCH=do_HEAD=handle_request
+class Redirect(BaseHTTPRequestHandler):
+    def log_message(self,*args):pass
+    def do_GET(self):
+        if self.headers.get('Host')!='country02-qa.test:18105':self.send_error(421);return
+        self.send_response(308);self.send_header('Location',meta['base']+self.path);self.send_header('Content-Length','0');self.end_headers()
+    do_HEAD=do_GET
+    def do_POST(self):self.send_error(405,'Credentials must be submitted on HTTPS')
+tls=ThreadingHTTPServer(('127.0.0.1',18443),Proxy)
+ctx=ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER);ctx.minimum_version=ssl.TLSVersion.TLSv1_2
+ctx.load_cert_chain(str(WORK/'qa-server.pem'),str(WORK/'qa-server-key.pem'));tls.socket=ctx.wrap_socket(tls.socket,server_side=True)
+redirect=ThreadingHTTPServer(('127.0.0.1',18105),Redirect)
+threading.Thread(target=redirect.serve_forever,daemon=True).start()
+print(json.dumps({'result':'QA HTTPS serving','origin':meta['base'],'active_config_changed':False}),flush=True)
+tls.serve_forever()