Perform all initialization logic. Only executes once.
(self)
| 265 | else: |
| 266 | log.error( |
| 267 | "Create DNS failed", |
| 268 | name=name, |
| 269 | status=response.status_code, |
| 270 | body=response.text[:200], |
| 271 | hypothesisId="DNS", |
| 272 | ) |
| 273 | except Exception as e: |
| 274 | log.error(f"Create DNS error {name}", hypothesisId="DNS", error=str(e)) |
| 275 | return None |
| 276 | |
| 277 | server_ip_str = str(self.server_ip) |
| 278 | name = get_identifier() + "-" + server_ip_str.replace(".", "") |
| 279 | res1 = create_dns_record(name + "-direct", False) |
| 280 | res2 = create_dns_record(name, True) |
| 281 | return name if res1 and res2 else None |
| 282 | |
| 283 | def _setup_reality(self) -> None: |
| 284 | """Derive the REALITY keypair and short id deterministically from the |
| 285 | identifier (same idea as the uuid), so links survive redeploys without |
| 286 | persisting any key material.""" |
| 287 | seed = hashlib.sha256(f"{self.config_id}:reality".encode()).digest() |
| 288 | seed_b64 = base64.urlsafe_b64encode(seed).decode().rstrip("=") |
| 289 | result = exec_command(["xray", "x25519", "-i", seed_b64], capture_output=True) |
| 290 | if result.returncode != 0 or not result.stdout: |
| 291 | log.error( |
| 292 | "xray x25519 failed; REALITY inbounds will be skipped", |
| 293 | hypothesisId="CFG", |
| 294 | exit_code=result.returncode, |
| 295 | ) |
| 296 | return |
| 297 | # v26 prints "PrivateKey:" / "Password (PublicKey):"; older builds |
| 298 | # printed "Private key:" / "Public key:" — accept both. |
| 299 | for line in result.stdout.splitlines(): |
| 300 | label, _, value = line.partition(":") |
| 301 | label = label.strip().lower() |
| 302 | value = value.strip() |
| 303 | if label in ("privatekey", "private key"): |
| 304 | self.reality_private_key = value |
| 305 | elif label.startswith("password") or label == "public key": |
| 306 | self.reality_public_key = value |
| 307 | if not self.reality_private_key or not self.reality_public_key: |
| 308 | self.reality_private_key = self.reality_public_key = "" |
| 309 | log.error( |
| 310 | "Could not parse xray x25519 output; REALITY inbounds will be skipped", |
| 311 | hypothesisId="CFG", |
| 312 | ) |
| 313 | return |
| 314 | |
| 315 | def _setup_vless_encryption(self) -> None: |
| 316 | """Derive the VLESS Encryption keypair deterministically from the |
| 317 | identifier (same idea as REALITY) so the non-TLS httpupgrade links |
| 318 | survive redeploys without persisting any key material. |
| 319 | |
| 320 | Authentication uses X25519 (the ephemeral exchange is ML-KEM-768 + |
| 321 | X25519, so it stays post-quantum safe either way). The server keeps the |
| 322 | private key in `decryption`; clients carry the matching public key |
| 323 | ("Password") in `encryption`. Format per Xray-core: |
| 324 | decryption: mlkem768x25519plus.xorpub.600s.<padding>.<PrivateKey> |
no test coverage detected