1  #!/usr/bin/env python3
 2  
 3  # Copyright 2026 Alibaba Group Holding Ltd.
 4  # 
 5  # Licensed under the Apache License, Version 2.0 (the "License");
 6  # you may not use this file except in compliance with the License.
 7  # You may obtain a copy of the License at
 8  # 
 9  #     http://www.apache.org/licenses/LICENSE-2.0
10  # 
11  # Unless required by applicable law or agreed to in writing, software
12  # distributed under the License is distributed on an "AS IS" BASIS,
13  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14  # See the License for the specific language governing permissions and
15  # limitations under the License.
16  
17  """
18  Lightweight HTTP server to receive OPENSANDBOX_EGRESS_DENY_WEBHOOK callbacks.
19  
20  Config:
21  - WEBHOOK_HOST: listen address (default 0.0.0.0)
22  - WEBHOOK_PORT: listen port (default 8000)
23  - WEBHOOK_PATH: webhook path (default /)
24  
25  Run:
26    python webhook_server.py
27  Then point OPENSANDBOX_EGRESS_DENY_WEBHOOK to http://<host>:<port><path>
28  """
29  
30  import http.server
31  import json
32  import os
33  import socketserver
34  from datetime import datetime
35  
36  HOST = os.getenv("WEBHOOK_HOST", "0.0.0.0")
37  PORT = int(os.getenv("WEBHOOK_PORT", "8000"))
38  PATH = os.getenv("WEBHOOK_PATH", "/")
39  
40  
41  class WebhookHandler(http.server.BaseHTTPRequestHandler):
42      def _send(self, code: int = 200, body: str = "ok") -> None:
43          self.send_response(code)
44          self.send_header("Content-Type", "text/plain; charset=utf-8")
45          self.end_headers()
46          self.wfile.write(body.encode("utf-8"))
47  
48      def do_POST(self) -> None:  # noqa: N802 (BaseHTTPRequestHandler API)
49          # Only allow the configured path
50          if self.path != PATH:
51              self._send(404, "not found")
52              return
53  
54          length = int(self.headers.get("Content-Length", 0))
55          raw = self.rfile.read(length) if length else b""
56  
57          payload = raw.decode("utf-8", errors="replace")
58          try:
59              parsed = json.loads(payload)
60          except json.JSONDecodeError:
61              parsed = None
62  
63          # Log request info for debugging
64          print(f"\n[{datetime.utcnow().isoformat()}Z] Received webhook")
65          print(f"Path: {self.path}")
66          print(f"Headers: {dict(self.headers)}")
67          print(f"Raw body: {payload}")
68          if parsed is not None:
69              print("Parsed JSON:")
70              print(json.dumps(parsed, indent=2))
71  
72          self._send(200, "received")
73  
74      # Silence default logging to reduce noise
75      def log_message(self, *args) -> None:
76          return
77  
78  
79  def main() -> None:
80      with socketserver.TCPServer((HOST, PORT), WebhookHandler) as httpd:
81          print(f"Listening on http://{HOST}:{PORT}{PATH} ...")
82          httpd.serve_forever()
83  
84  
85  if __name__ == "__main__":
86      main()