42 lines
2.5 KiB
Python
42 lines
2.5 KiB
Python
"""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()
|