#! /usr/bin/python3# This is just enough of the Insights REST API to make the following# work:## insights-client --register# insights-client --status# insights-client --check-results# insights-client --unregister## You need these in your insights-client.conf:## auto_config=False# auto_update=False# base_url=127.0.0.1:8443/r/insights# cert_verify=/var/lib/insights/mock-certs/ca.crt# username=admin# password=foobarimport emailimport jsonimport osimport reimport sslimport subprocessfrom http.server import BaseHTTPRequestHandler, HTTPServersystems: dict[str, object] = {}class handler(BaseHTTPRequestHandler):def match(self, p):return re.fullmatch(p, self.path)def do_GET(self):m = self.match("/r/insights")if m:self.send_response(200)self.end_headers()self.wfile.write(b"lub-dup")returnm = self.match("/r/insights/v1/static/uploader.v2.json")if m:# This is not a valid response and will cause the client# to fall back to the builtin rules.self.send_response(200)self.end_headers()self.wfile.write(b"{ }\n")returnm = self.match("/r/insights/v1/systems/([^/]+)")if m:machine_id = m[1]for system in systems.values():if system["machine_id"] == machine_id:self.send_response(200)self.end_headers()self.wfile.write(json.dumps(system).encode() + b"\n")returnself.send_response(404)self.end_headers()returnm = self.match("/r/insights/v1/branch_info")if m:self.send_response(200)self.end_headers()self.wfile.write(b'{ "remote_branch": -1, "remote_leaf": -1 }\n')returnm = self.match("/r/insights/platform/inventory/v1/hosts\\?insights_id=(.*)")if m:insights_id = m[1]self.send_response(200)self.end_headers()res = {"total": 0,"results": [],}for system in systems.values():if system["insights_id"] == insights_id:res["total"] += 1res["results"].append(system)self.wfile.write(json.dumps(res).encode("utf-8") + b"\n")returnm = self.match("/r/insights/platform/inventory/v1/host_exists\\?insights_id=(.*)")if m:insights_id = m[1]self.send_response(200)self.end_headers()res = {}for system in systems.values():if system["insights_id"] == insights_id:res["id"] = system["id"]breakself.wfile.write(json.dumps(res).encode("utf-8") + b"\n")returnm = self.match("/r/insights/platform/insights/v1/system/([^/]+)/reports/")if m:inventory_id = m[1]self.send_response(200)self.end_headers()if inventory_id in systems:self.wfile.write(b'[ { "rule": { "total_risk": 3 } }, { "rule": { "total_risk": 2 } }, { "rule": { "total_risk": 1 } }]\n')else:self.wfile.write(b'[ ]\n')returnself.send_response(404)self.end_headers()def do_POST(self):content_length = int(self.headers.get('content-length', 0))data = self.rfile.read(content_length)m = self.match("/r/insights/v1/systems")if m:s = json.loads(data)s["unregistered_at"] = Nones["account_number"] = "123456"s["insights_id"] = s["machine_id"]s["id"] = "123-nice-id"print(s)systems[s["id"]] = sself.send_response(200)self.end_headers()self.wfile.write(data)returnm = self.match("/r/insights/uploads/([^/])+")if m:self.send_response(200)self.end_headers()self.wfile.write(b'{ "reports": [ "foo", "bar" ] }\n')returnm = self.match("/r/insights/platform/ingress/v1/upload")if m:# The metadata of the system is in the multipart MIME data# sent by the system for the upload; to pick it and use it# we need to unpack the multipart MIME data.# First, create prologue to the data, so it can be recognized# as multipart MIME.multipart_data = (b"""MIME-Version: 1.0Content-Type: """+ self.headers.get("content-type").encode("utf-8")+ b"""""")multipart_data += data# Parse the multipart MIME data, and then look for a "metadata"# file part which contains the system metadata as JSON.message = email.message_from_bytes(multipart_data)for part in message.walk():if part.get_filename() == "metadata":s = json.loads(part.get_payload())s["id"] = "123-nice-id"print(s)systems[s["id"]] = sself.send_response(202)self.end_headers()self.wfile.write(b"{"b' "request_id": "some-upload", 'b' "upload": { 'b' "account": "123456", 'b' "org_id": "123456" 'b" }"b"}\n")returnself.send_response(400)self.end_headers()returnself.send_response(404)self.end_headers()def do_DELETE(self):m = self.match("/r/insights/v1/systems/([^/]+)")if m:machine_id = m[1]self.send_response(200)self.end_headers()systems.pop(machine_id, None)returnm = self.match("/r/insights/platform/inventory/v1/hosts/([^/]+)")if m:inventory_id = m[1]if inventory_id in systems:del systems[inventory_id]self.send_response(200)self.end_headers()returnself.send_response(404)self.end_headers()returnself.send_response(404)self.end_headers()def insights_server(port):# Let's put the certs into /var/lib/insights so that SELinux# allows insights-client to actually read ca.crt when running as a# systemd service.certdir = "/var/lib/insights/mock-certs"if not os.path.exists(certdir):os.makedirs(certdir)subprocess.check_call(["sscg"], cwd=certdir)httpd = HTTPServer(('', port), handler)ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)ctx.check_hostname = Falsectx.verify_mode = ssl.CERT_OPTIONAL# with newer Pythons this is ctx.minimum_version = ssl.TLSVersion.TLSv1_2ctx.options |= ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1ctx.load_cert_chain(f'{certdir}/service.pem', f'{certdir}/service-key.pem')httpd.socket = ctx.wrap_socket(httpd.socket, server_side=True)httpd.serve_forever()insights_server(8443)