| 142 | ) |
| 143 | return "" |
| 144 | |
| 145 | |
| 146 | def _cert_not_after(cert_file: str) -> Optional[datetime]: |
| 147 | """Return a PEM cert's notAfter (UTC), or None if it can't be determined.""" |
| 148 | result = exec_command( |
| 149 | ["openssl", "x509", "-enddate", "-noout", "-in", cert_file], |
| 150 | capture_output=True, |
| 151 | ) |
| 152 | if result.returncode != 0 or not result.stdout: |
| 153 | return None |
| 154 | # stdout looks like: "notAfter=Jun 18 12:00:00 2026 GMT" |
| 155 | raw = result.stdout.strip().split("=", 1)[-1].strip() |
| 156 | try: |
| 157 | return datetime.strptime(raw, "%b %d %H:%M:%S %Y %Z").replace( |
| 158 | tzinfo=timezone.utc |
| 159 | ) |
| 160 | except ValueError: |
| 161 | return None |
| 162 | |
| 163 | |
| 164 | class XrayConfig: |
| 165 | def __init__(self) -> None: |
| 166 | """Initialize instance attributes to hold shared variables.""" |
| 167 | self.env_config: Dict[str, str] = {} |
| 168 | self.is_debug_enabled: bool = False |
| 169 | self.anti_abuse: bool = False |
| 170 | self.config_id: str = "" |
| 171 | self.config_uuid: str = "" |
| 172 | self.cf_api_token: Optional[str] = None |
| 173 | self.cf_zone_id: Optional[str] = None |
| 174 | self.nginx_path: Optional[str] = None |
| 175 | self.xray_inbounds: Dict[str, int] = {} |
| 176 | self.server_ip: str = "Unknown" |
| 177 | self.domain: Optional[str] = None |
| 178 | self.subdomain: Optional[str] = None |
| 179 | self.direct_subdomain: Optional[str] = None |
| 180 | self._cert_serial: int = 0 |
| 181 | self.cf_clean_ip_domain: str = "npmjs.com" |
| 182 | self.reality_private_key: str = "" |
| 183 | self.reality_public_key: str = "" |
| 184 | self.reality_sni: str = "" |
| 185 | # VLESS Encryption (post-quantum AEAD) for the non-TLS httpupgrade |
| 186 | # inbounds: derived deterministically like REALITY so links survive |
| 187 | # redeploys. Empty until _setup_vless_encryption populates them. |
| 188 | self.vless_enc_private_key: str = "" |
| 189 | self.vless_enc_password: str = "" |
| 190 | self.vless_enc_decryption: str = "" |
| 191 | self.vless_enc_encryption: str = "" |
| 192 | self.configured_inbounds: List[Dict[str, Any]] = [] |
| 193 | self.xray_config: Dict[str, Any] = {} |
| 194 | self.warps_ready: bool = False |
| 195 | self.wg_configs: Dict[str, str] = {} |
| 196 | self.warps: List[Dict[str, Any]] = [] |
| 197 | self.nginx_locations: Dict[int, str] = {} |
| 198 | self.initialized: bool = False |
| 199 | |
| 200 | def _get_domain(self) -> None: |
| 201 | """Internal method to retrieve domain from Cloudflare API.""" |