40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Capturing mock fleet server for agent tests: logs one JSON line per POST."""
|
|
import json
|
|
import os
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
REQLOG = os.environ.get("REQLOG", "/tmp/reqs.jsonl")
|
|
open(REQLOG, "w").close()
|
|
|
|
|
|
class H(BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
n = int(self.headers.get("Content-Length", 0))
|
|
body = self.rfile.read(n)
|
|
ctype = (self.headers.get("Content-Type", "") or "").split(";")[0]
|
|
rec = {"path": self.path, "ctype": ctype,
|
|
"auth": self.headers.get("Authorization", ""), "len": n}
|
|
if ctype == "application/json":
|
|
try:
|
|
rec["json"] = json.loads(body)
|
|
except Exception as e:
|
|
rec["json_err"] = str(e)
|
|
else:
|
|
txt = body.decode("latin-1")
|
|
i = txt.find('name="meta"')
|
|
if i != -1:
|
|
rec["meta"] = txt[i:].split("\r\n\r\n", 1)[-1].split("\r\n", 1)[0]
|
|
with open(REQLOG, "a") as f:
|
|
f.write(json.dumps(rec) + "\n")
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.wfile.write(b'{"ok":true}')
|
|
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
HTTPServer(("127.0.0.1", 8799), H).serve_forever()
|