zapret-kvn/xray_fluent/engines/runtime_security.py
loop-uh d5a4715771
Some checks failed
Windows project source guards / test (push) Has been cancelled
feat: use official Amnezia transport and organize runtime modules
2026-09-06 00:14:08 +03:00

82 lines
2.5 KiB
Python

from __future__ import annotations
import secrets
import string
from typing import Any
_PROXY_INBOUND_TYPES = {"socks", "http", "mixed"}
def generate_local_proxy_credentials(*, prefix: str = "local", password_length: int = 24) -> tuple[str, str]:
suffix = secrets.token_hex(3)
username = f"{prefix}-{suffix}"
alphabet = string.ascii_letters + string.digits
password = "".join(secrets.choice(alphabet) for _ in range(password_length))
return username, password
def strip_xray_proxy_inbounds(payload: dict[str, Any], *, keep_tags: set[str] | None = None) -> int:
return _strip_proxy_inbounds(payload, protocol_key="protocol", keep_tags=keep_tags)
def strip_singbox_proxy_inbounds(payload: dict[str, Any], *, keep_tags: set[str] | None = None) -> int:
return _strip_proxy_inbounds(payload, protocol_key="type", keep_tags=keep_tags)
def _strip_proxy_inbounds(
payload: dict[str, Any],
*,
protocol_key: str,
keep_tags: set[str] | None,
) -> int:
inbounds = payload.get("inbounds")
if not isinstance(inbounds, list):
return 0
keep = {str(tag).strip() for tag in (keep_tags or set()) if str(tag).strip()}
filtered: list[Any] = []
removed = 0
for inbound in inbounds:
if not isinstance(inbound, dict):
filtered.append(inbound)
continue
tag = str(inbound.get("tag") or "").strip()
inbound_type = str(inbound.get(protocol_key) or inbound.get("protocol") or "").strip().lower()
if tag in keep or inbound_type not in _PROXY_INBOUND_TYPES:
filtered.append(inbound)
continue
removed += 1
if removed:
payload["inbounds"] = filtered
return removed
def set_xray_socks_inbound_auth(
payload: dict[str, Any],
*,
username: str,
password: str,
tag: str | None = None,
) -> bool:
inbounds = payload.get("inbounds")
if not isinstance(inbounds, list):
return False
for inbound in inbounds:
if not isinstance(inbound, dict):
continue
if str(inbound.get("protocol") or "").strip().lower() != "socks":
continue
inbound_tag = str(inbound.get("tag") or "").strip()
if tag is not None and inbound_tag != tag:
continue
settings = inbound.get("settings")
if not isinstance(settings, dict):
settings = {}
inbound["settings"] = settings
settings["auth"] = "password"
settings["accounts"] = [{"user": username, "pass": password}]
return True
return False