#!/usr/bin/env python3 """ external_monitor/server.py -------------------------- Lightweight HTTP API server that checks whether services are reachable from the outside internet. Endpoints: GET /status -> {"online": true} (liveness probe) GET /check/ -> {"online": bool, "message": str} Endpoint definitions live in endpoints.json (same directory). No external dependencies — uses only the Python standard library. Usage: python3 server.py [--port 8080] [--host 0.0.0.0] """ import json import os import socket import subprocess import sys import urllib.error import urllib.request from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.parse import urlparse # ----------------------------------------------------------------------- # Configuration # ----------------------------------------------------------------------- DEFAULT_PORT = 8080 DEFAULT_HOST = "0.0.0.0" ENDPOINTS_FILE = os.path.join(os.path.dirname(__file__), "endpoints.json") # ----------------------------------------------------------------------- # Load endpoints # ----------------------------------------------------------------------- def load_endpoints(): with open(ENDPOINTS_FILE, "r") as f: return json.load(f) # ----------------------------------------------------------------------- # Check functions # ----------------------------------------------------------------------- def check_ping(host: str) -> tuple[bool, str]: """ICMP ping — relies on the system ping binary.""" try: result = subprocess.run( ["ping", "-c", "1", "-W", "1", host], capture_output=True, timeout=4, ) if result.returncode == 0: return True, "Reachable" return False, "Unreachable" except FileNotFoundError: # ping not on PATH — try Windows style try: result = subprocess.run( ["ping", "-n", "1", "-w", "1000", host], capture_output=True, timeout=4, ) return result.returncode == 0, "Reachable" if result.returncode == 0 else "Unreachable" except Exception as e: return False, str(e) except Exception as e: return False, str(e) def check_http(url: str) -> tuple[bool, str]: """HTTP/HTTPS: verifies the server responds with any HTTP status.""" try: req = urllib.request.Request( url, headers={"User-Agent": "ExternalMonitor/1.0"}, ) with urllib.request.urlopen(req, timeout=6) as resp: return True, f"HTTP {resp.status}" except urllib.error.HTTPError as e: # Server responded — it's reachable even if status is 4xx/5xx return True, f"HTTP {e.code}" except Exception as e: return False, str(e)[:80] def check_tcp(host: str, port: int) -> tuple[bool, str]: """TCP port probe — works for any TCP service (web, game servers, etc.).""" try: with socket.create_connection((host, port), timeout=4): return True, f"Port {port} open" except Exception as e: return False, str(e)[:80] def check_udp(host: str, port: int) -> tuple: """UDP probe — sends a minimal packet and checks for a response or ICMP rejection. Result states: True — service responded to the probe (definitively up) False — ICMP port unreachable received (definitively down) None — timed out with no response (open or filtered; ambiguous) """ try: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.settimeout(2) sock.sendto(b"\x00", (host, port)) try: data, _ = sock.recvfrom(256) return True, f"Responded ({len(data)}b)" except socket.timeout: return None, "No response" except ConnectionRefusedError: return False, "ICMP Unreachable" finally: sock.close() except Exception as e: return False, str(e)[:80] def check_dns(host: str) -> tuple[bool, str]: """Send a real DNS query directly to host:53 and verify it responds. Queries for "google.com" type A. Any valid DNS response (even NXDOMAIN/REFUSED) means the nameserver is reachable and functioning. """ query = ( b"\xAB\xCD" # Transaction ID (checked in response) b"\x01\x00" # Flags: standard query, RD=1 b"\x00\x01" # QDCOUNT: 1 b"\x00\x00\x00\x00\x00\x00" # ANCOUNT, NSCOUNT, ARCOUNT: 0 b"\x06google\x03com\x00" # QNAME: google.com b"\x00\x01" # QTYPE: A b"\x00\x01" # QCLASS: IN ) rcodes = {0: "NOERROR", 2: "SERVFAIL", 3: "NXDOMAIN", 5: "REFUSED"} try: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.settimeout(2) sock.sendto(query, (host, 53)) try: data, _ = sock.recvfrom(512) if len(data) >= 12 and data[0:2] == b"\xAB\xCD" and (data[2] & 0x80): rcode = data[3] & 0x0F label = rcodes.get(rcode, f"RCODE {rcode}") return True, f"DNS {label}" return False, "Bad response" except socket.timeout: return False, "No response" except ConnectionRefusedError: return False, "ICMP Unreachable" finally: sock.close() except Exception as e: return False, str(e)[:80] def check_endpoint(ep: dict) -> dict: """Dispatch to the correct check function based on endpoint type.""" type_ = ep.get("type", "http").lower() if type_ == "ping": online, msg = check_ping(ep["host"]) elif type_ in ("http", "https"): if "url" in ep: url = ep["url"] else: scheme = type_ host = ep["host"] port = ep.get("port", 443 if scheme == "https" else 80) path = ep.get("path", "/") url = f"{scheme}://{host}:{port}{path}" online, msg = check_http(url) elif type_ == "tcp": online, msg = check_tcp(ep["host"], int(ep["port"])) elif type_ == "udp": online, msg = check_udp(ep["host"], int(ep["port"])) elif type_ == "dns": online, msg = check_dns(ep["host"]) else: online, msg = False, f"Unknown type: {type_}" return {"online": online, "message": msg} # ----------------------------------------------------------------------- # HTTP request handler # ----------------------------------------------------------------------- class MonitorHandler(BaseHTTPRequestHandler): endpoints = [] # populated at startup def do_GET(self): parsed = urlparse(self.path) path = parsed.path.rstrip("/") if path == "/status": self._json(200, {"online": True}) elif path.startswith("/check/"): raw_index = path.split("/check/", 1)[-1] try: index = int(raw_index) except ValueError: self._json(400, {"error": "Index must be an integer"}) return if index < 0 or index >= len(self.endpoints): self._json(404, {"error": f"No endpoint at index {index}"}) return ep = self.endpoints[index] result = check_endpoint(ep) self._json(200, result) else: self._json(404, {"error": "Not found"}) def _json(self, code: int, payload: dict): body = json.dumps(payload).encode() self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(body) def log_message(self, fmt, *args): # Print concise one-line log to stdout print(f"[{self.address_string()}] {fmt % args}") # ----------------------------------------------------------------------- # Entry point # ----------------------------------------------------------------------- def main(): import argparse parser = argparse.ArgumentParser(description="External Monitor Server") parser.add_argument("--host", default=DEFAULT_HOST) parser.add_argument("--port", type=int, default=DEFAULT_PORT) args = parser.parse_args() # Load endpoints once at startup try: MonitorHandler.endpoints = load_endpoints() print(f"Loaded {len(MonitorHandler.endpoints)} endpoint(s) from {ENDPOINTS_FILE}") except Exception as e: print(f"ERROR: Could not load {ENDPOINTS_FILE}: {e}", file=sys.stderr) sys.exit(1) server = HTTPServer((args.host, args.port), MonitorHandler) print(f"External Monitor listening on {args.host}:{args.port}") print("Routes:") print(f" GET /status -> liveness probe") print(f" GET /check/ -> check endpoint at index ") try: server.serve_forever() except KeyboardInterrupt: print("\nShutting down.") server.server_close() if __name__ == "__main__": main()