| 22 | |
| 23 | |
| 24 | class XrayService: |
| 25 | def __init__(self) -> None: |
| 26 | self.app: Flask = Flask(__name__) |
| 27 | self.valid_configs: Dict[str, Dict[str, str]] = {} |
| 28 | self.latest_metrics: str = "" |
| 29 | self.instance_location_info: Optional[Union[str, Dict[str, str]]] = None |
| 30 | self._shutdown_event: threading.Event = threading.Event() |
| 31 | self._setup_routes() |
| 32 | |
| 33 | def _setup_routes(self) -> None: |
| 34 | @self.app.route("/config") |
| 35 | def get_xray_config(): |
| 36 | if not config.initialized: |
| 37 | log.info( |
| 38 | "Xray Config requested but not yet initialized", hypothesisId="XRAY" |
| 39 | ) |
| 40 | return "Not Ready", 503 |
| 41 | # Strip top-level null keys; some Xray versions reject them |
| 42 | clean = {k: v for k, v in config.xray_config.items() if v is not None} |
| 43 | return json.dumps(clean, indent=4) |
| 44 | |
| 45 | @self.app.route("/subdomain") |
| 46 | def export_certs(): |
| 47 | if not config.initialized or not config.direct_subdomain: |
| 48 | log.info("Subdomain requested but not yet ready", hypothesisId="XRAY") |
| 49 | return "Not Ready", 503 |
| 50 | return config.direct_subdomain |
| 51 | |
| 52 | @self.app.route("/warps") |
| 53 | def get_warps(): |
| 54 | if not config.warps_ready: |
| 55 | abort(404) |
| 56 | return json.dumps(config.warps, indent=4) |
| 57 | |
| 58 | @self.app.route("/wg-configs") |
| 59 | def get_wg_configs(): |
| 60 | if not config.warps_ready: |
| 61 | abort(404) |
| 62 | return json.dumps(config.wg_configs, indent=4) |
| 63 | |
| 64 | @self.app.route("/nginx-locations") |
| 65 | def get_nginx_locations(): |
| 66 | if not config.initialized: |
| 67 | return "Not Ready", 503 |
| 68 | return json.dumps(config.nginx_locations, indent=4) |
| 69 | |
| 70 | @self.app.route("/metrics") |
| 71 | def metrics(): |
| 72 | return self.latest_metrics |
| 73 | |
| 74 | def update_metrics(self, configs: Dict[str, Dict[str, str]]) -> bool: |
| 75 | if not self.instance_location_info: |
| 76 | self.instance_location_info = get_public_ip(extra=True) |
| 77 | |
| 78 | if isinstance(self.instance_location_info, dict): |
| 79 | instance_ip = self.instance_location_info.get("ip", "Unknown") |
| 80 | instance_country = self.instance_location_info.get("country", "Unknown") |
| 81 | else: |