net-kingdom/identity-provisioner/server.py

from dataclasses import asdict
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import os
import secrets

from provisioner import LLDAPProvisioner, dispatch


class Handler(BaseHTTPRequestHandler):
    provisioner: LLDAPProvisioner
    service_token: str

    def do_GET(self):
        if self.path == "/healthz":
            return self._send(200, {"status": "ok"})
        self._send(404, {"error": "not_found"})

    def do_POST(self):
        supplied = self.headers.get("Authorization", "").removeprefix("Bearer ")
        if not secrets.compare_digest(supplied, self.service_token):
            return self._send(403, {"error": "access_denied"})
        try:
            length = min(int(self.headers.get("Content-Length", "0")), 65536)
            payload = json.loads(self.rfile.read(length))
            result = dispatch(self.provisioner, self.path, payload)
        except KeyError:
            return self._send(404, {"error": "not_found"})
        except (ValueError, json.JSONDecodeError) as exc:
            return self._send(400, {"error": "invalid_request", "message": str(exc)})
        self._send(200, asdict(result))

    def _send(self, status: int, payload: dict):
        body = json.dumps(payload, separators=(",", ":")).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Cache-Control", "no-store")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format, *args):
        return


def main():
    Handler.provisioner = LLDAPProvisioner(
        base_url=os.environ["LLDAP_URL"],
        admin_password=os.environ["LLDAP_ADMIN_PASSWORD"],
    )
    Handler.service_token = os.environ["PROVISIONER_SERVICE_TOKEN"].strip()
    if not Handler.service_token:
        raise ValueError("PROVISIONER_SERVICE_TOKEN must not be empty")
    ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever()


if __name__ == "__main__":
    main()