Add scoped LLDAP identity provisioner

identity-provisioner/server.py

@@ -0,0 +1,56 @@
+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"]
+    ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever()
+
+
+if __name__ == "__main__":
+    main()