73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from http import HTTPStatus
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
from certificates.renderer import RenderError, render_certificate_to_bytes, validate_certificate_context
|
|
|
|
|
|
class RenderHandler(BaseHTTPRequestHandler):
|
|
server_version = "CertRenderer/1.0"
|
|
|
|
def do_POST(self):
|
|
if self.path != "/render":
|
|
self._write_json(HTTPStatus.NOT_FOUND, {"error": "not found"})
|
|
return
|
|
|
|
expected_token = os.getenv("CERTIFICATE_RENDERER_TOKEN", "")
|
|
if expected_token:
|
|
provided_token = self.headers.get("X-Renderer-Token", "")
|
|
if provided_token != expected_token:
|
|
self._write_json(HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"})
|
|
return
|
|
|
|
content_length = int(self.headers.get("Content-Length", "0") or "0")
|
|
if content_length <= 0 or content_length > 8192:
|
|
self._write_json(HTTPStatus.BAD_REQUEST, {"error": "invalid payload size"})
|
|
return
|
|
|
|
raw_body = self.rfile.read(content_length)
|
|
try:
|
|
payload = json.loads(raw_body)
|
|
context = validate_certificate_context(payload.get("context", {}))
|
|
pdf_bytes = render_certificate_to_bytes(context)
|
|
except (json.JSONDecodeError, TypeError):
|
|
self._write_json(HTTPStatus.BAD_REQUEST, {"error": "invalid json"})
|
|
return
|
|
except ValueError as exc:
|
|
self._write_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
|
|
return
|
|
except RenderError as exc:
|
|
self._write_json(HTTPStatus.UNPROCESSABLE_ENTITY, {"error": str(exc)})
|
|
return
|
|
|
|
self.send_response(HTTPStatus.OK)
|
|
self.send_header("Content-Type", "application/pdf")
|
|
self.send_header("Content-Length", str(len(pdf_bytes)))
|
|
self.end_headers()
|
|
self.wfile.write(pdf_bytes)
|
|
|
|
def log_message(self, format, *args):
|
|
return
|
|
|
|
def _write_json(self, status: HTTPStatus, body: dict):
|
|
encoded = json.dumps(body).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(encoded)))
|
|
self.end_headers()
|
|
self.wfile.write(encoded)
|
|
|
|
|
|
def main() -> None:
|
|
host = os.getenv("CERTIFICATE_RENDERER_HOST", "0.0.0.0")
|
|
port = int(os.getenv("CERTIFICATE_RENDERER_PORT", "8090"))
|
|
server = ThreadingHTTPServer((host, port), RenderHandler)
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|