flayer.profiles.vless

Module source

flayer.profiles.vless

Prepare private VLESS Reality profiles for the pinned Xray implementation.

VLESS_TRANSPORT module-attribute

VLESS_TRANSPORT = 'vless-reality'

VLESS_FLOW module-attribute

VLESS_FLOW = 'xtls-rprx-vision'

VLESS_CAPABILITIES module-attribute

VLESS_CAPABILITIES = VpnCapabilities(VLESS_TRANSPORT, 'application-proxy', ('full', 'split'))

XRAY_VERSION module-attribute

XRAY_VERSION = '26.9.30'

VlessError

Bases: ContractError

VLESS intent or credentials cannot safely produce a supported configuration.

VlessRealitySettings dataclass

VlessRealitySettings(target_host: str, server_name: str, target_port: int = 443, fingerprint: str = 'chrome', socks_port: int = 10808)

Public Reality handshake target and local application-proxy listener settings.

target_host instance-attribute

target_host: str

server_name instance-attribute

server_name: str

target_port class-attribute instance-attribute

target_port: int = 443

fingerprint class-attribute instance-attribute

fingerprint: str = 'chrome'

socks_port class-attribute instance-attribute

socks_port: int = 10808

VlessDeviceCredential dataclass

VlessDeviceCredential(device_id: str, user_id: str)

One revocable per-device VLESS UUID, excluded from object representations.

device_id instance-attribute

device_id: str

user_id class-attribute instance-attribute

user_id: str = field(repr=False)

VlessMaterial dataclass

VlessMaterial(server_private_key: bytes, server_public_key: bytes, short_id: str, devices: tuple[VlessDeviceCredential, ...])

Explicit secret material retained in private bundles, never generic lifecycle state.

server_private_key class-attribute instance-attribute

server_private_key: bytes = field(repr=False)

server_public_key class-attribute instance-attribute

server_public_key: bytes = field(repr=False)

short_id class-attribute instance-attribute

short_id: str = field(repr=False)

devices class-attribute instance-attribute

devices: tuple[VlessDeviceCredential, ...] = field(repr=False)

GenerateVlessMaterial

GenerateVlessMaterial(profile: VpnProfile) -> VlessMaterial

Generate fresh server and device credentials only for an explicitly validated profile.

Source code in installed/flayer/profiles/vless.py

def GenerateVlessMaterial(profile: VpnProfile) -> VlessMaterial:
    """Generate fresh server and device credentials only for an explicitly validated profile."""

    ValidateVpnTransport(profile, VLESS_CAPABILITIES)
    private_key = secrets.token_bytes(32)

    return VlessMaterial(
        private_key, _PublicKey(private_key), secrets.token_hex(8),
        tuple(VlessDeviceCredential(device.device_id, str(uuid.uuid4())) for device in profile.devices),
    )

EncodeVlessMaterial

EncodeVlessMaterial(material: VlessMaterial) -> bytes

Serialize credentials only for a caller-owned sensitive artifact, never general state.

Source code in installed/flayer/profiles/vless.py

def EncodeVlessMaterial(material: VlessMaterial) -> bytes:
    """Serialize credentials only for a caller-owned sensitive artifact, never general state."""

    if not isinstance(material, VlessMaterial):
        raise VlessError("Expected validated VLESS material")

    return _Json({
        "schema_version": 1,
        "server_private_key": _Encode(material.server_private_key),
        "server_public_key": _Encode(material.server_public_key),
        "short_id": material.short_id,
        "devices": [asdict(item) for item in material.devices],
    })

DecodeVlessMaterial

DecodeVlessMaterial(content: bytes) -> VlessMaterial

Read an explicitly supplied bounded private snapshot without consulting local defaults.

Source code in installed/flayer/profiles/vless.py

def DecodeVlessMaterial(content: bytes) -> VlessMaterial:
    """Read an explicitly supplied bounded private snapshot without consulting local defaults."""

    try:
        if not isinstance(content, bytes) or not 1 <= len(content) <= 16384:
            raise ValueError("Invalid input size")

        data = json.loads(content)
        expected = {"schema_version", "server_private_key", "server_public_key", "short_id", "devices"}

        if not isinstance(data, dict) or set(data) != expected or type(data["schema_version"]) is not int:
            raise ValueError("Invalid schema")

        if data["schema_version"] != 1 or not isinstance(data["devices"], list):
            raise ValueError("Invalid schema version")

        keys = []

        for name in ("server_private_key", "server_public_key"):
            value = data[name]

            if not isinstance(value, str) or re.fullmatch(r"[a-zA-Z0-9_-]{43}", value) is None:
                raise ValueError("Invalid key encoding")

            raw = base64.urlsafe_b64decode(value + "=")

            if _Encode(raw) != value:
                raise ValueError("Noncanonical key encoding")

            keys.append(raw)

        devices = []

        for device in data["devices"]:
            if not isinstance(device, dict) or set(device) != {"device_id", "user_id"}:
                raise ValueError("Invalid device credential")

            devices.append(VlessDeviceCredential(device["device_id"], device["user_id"]))

        material = VlessMaterial(keys[0], keys[1], data["short_id"], tuple(devices))

        if not secrets.compare_digest(_PublicKey(material.server_private_key), material.server_public_key):
            raise ValueError("Inconsistent keys")

        return material

    except (ValueError, TypeError, KeyError, RecursionError):
        raise VlessError("Private VLESS material does not satisfy its schema or key contract") from None

RenderVlessServer

RenderVlessServer(profile: VpnProfile, settings: VlessRealitySettings, material: VlessMaterial) -> bytes

Build a Reality-protected IPv4 proxy without device traffic or credential logging.

Source code in installed/flayer/profiles/vless.py

def RenderVlessServer(
    profile: VpnProfile, settings: VlessRealitySettings, material: VlessMaterial,
) -> bytes:
    """Build a Reality-protected IPv4 proxy without device traffic or credential logging."""

    port = _Validate(profile, settings, material)
    blocked = [
        "0.0.0.0/8", "10.0.0.0/8", "100.64.0.0/10", "127.0.0.0/8", "169.254.0.0/16",
        "172.16.0.0/12", "192.168.0.0/16", "224.0.0.0/4", "240.0.0.0/4", "::/0",
    ]

    return _Json({
        "log": {"loglevel": "none", "access": "none", "error": "none"},
        "dns": {"servers": list(profile.routes.dns_servers), "queryStrategy": "UseIPv4"},
        "inbounds": [{
            "tag": "vless-reality", "listen": "0.0.0.0", "port": port, "protocol": "vless",
            "settings": {
                "clients": [{"id": item.user_id, "flow": VLESS_FLOW} for item in material.devices],
                "decryption": "none",
            },
            "streamSettings": {
                "network": "tcp", "security": "reality",
                "realitySettings": {
                    "show": False, "target": f"{settings.target_host}:{settings.target_port}",
                    "xver": 0, "serverNames": [settings.server_name],
                    "privateKey": _Encode(material.server_private_key), "shortIds": [material.short_id],
                },
            },
        }],
        "outbounds": [
            {"tag": "internet", "protocol": "freedom", "settings": {"domainStrategy": "UseIPv4"}},
            {"tag": "blocked", "protocol": "blackhole"},
        ],
        "routing": {
            "domainStrategy": "IPOnDemand",
            "rules": [{"type": "field", "ip": blocked, "outboundTag": "blocked"}],
        },
    })

RenderVlessClient

RenderVlessClient(profile: VpnProfile, endpoint: VpnEndpoint, settings: VlessRealitySettings, material: VlessMaterial, device_id: str) -> bytes

Render a loopback SOCKS client; split destinations outside the allowlist are rejected.

Source code in installed/flayer/profiles/vless.py

def RenderVlessClient(
    profile: VpnProfile, endpoint: VpnEndpoint, settings: VlessRealitySettings,
    material: VlessMaterial, device_id: str,
) -> bytes:
    """Render a loopback SOCKS client; split destinations outside the allowlist are rejected."""

    port = _Validate(profile, settings, material)

    if not isinstance(endpoint, VpnEndpoint) or endpoint.port != port:
        raise VlessError("VLESS endpoint must match the declared TCP listener port")

    credential = _Credential(material, device_id)
    routing_rules: list[dict[str, object]] = [
        {"type": "field", "ip": ["::/0"], "outboundTag": "blocked"},
        {"type": "field", "ip": list(profile.routes.dns_servers), "port": "53",
         "outboundTag": "vpn"},
        {"type": "field", "ip": list(profile.routes.ipv4_cidrs), "outboundTag": "vpn"},
        {"type": "field", "network": "tcp,udp", "outboundTag": "blocked"},
    ]

    return _Json({
        "log": {"loglevel": "none", "access": "none", "error": "none"},
        "dns": {"servers": list(profile.routes.dns_servers), "queryStrategy": "UseIPv4"},
        "inbounds": [{
            "tag": "local-socks", "listen": "127.0.0.1", "port": settings.socks_port,
            "protocol": "socks", "settings": {"auth": "noauth", "udp": True, "ip": "127.0.0.1"},
        }],
        "outbounds": [{
            "tag": "vpn", "protocol": "vless",
            "settings": {"vnext": [{
                "address": endpoint.host, "port": endpoint.port,
                "users": [{"id": credential.user_id, "encryption": "none", "flow": VLESS_FLOW}],
            }]},
            "streamSettings": {
                "network": "tcp", "security": "reality",
                "realitySettings": {
                    "fingerprint": settings.fingerprint, "serverName": settings.server_name,
                    "publicKey": _Encode(material.server_public_key), "shortId": material.short_id,
                    "spiderX": "/",
                },
            },
        }, {"tag": "blocked", "protocol": "blackhole"}],
        "routing": {"domainStrategy": "IPOnDemand", "rules": routing_rules},
    })

BuildVlessUri

BuildVlessUri(profile: VpnProfile, endpoint: VpnEndpoint, settings: VlessRealitySettings, material: VlessMaterial, device_id: str) -> str

Produce a transport import URI; client route and DNS policy need separate configuration.

Source code in installed/flayer/profiles/vless.py

def BuildVlessUri(
    profile: VpnProfile, endpoint: VpnEndpoint, settings: VlessRealitySettings,
    material: VlessMaterial, device_id: str,
) -> str:
    """Produce a transport import URI; client route and DNS policy need separate configuration."""

    RenderVlessClient(profile, endpoint, settings, material, device_id)
    credential = _Credential(material, device_id)
    parameters = urlencode({
        "encryption": "none", "flow": VLESS_FLOW, "security": "reality",
        "sni": settings.server_name, "fp": settings.fingerprint,
        "pbk": _Encode(material.server_public_key), "sid": material.short_id,
        "type": "tcp", "headerType": "none", "spx": "/",
    })
    label = quote(f"{profile.identity.stack}-{device_id}", safe="")

    return f"vless://{credential.user_id}@{endpoint.host}:{endpoint.port}?{parameters}#{label}"

VlessServiceName

VlessServiceName(profile: VpnProfile) -> str

Return the deterministic systemd unit for this exact stack ownership identity.

Source code in installed/flayer/profiles/vless.py

def VlessServiceName(profile: VpnProfile) -> str:
    """Return the deterministic systemd unit for this exact stack ownership identity."""

    ValidateVpnTransport(profile, VLESS_CAPABILITIES)
    owner = hashlib.sha256(_Json(asdict(profile.identity))).hexdigest()

    return f"flayer-vless-{owner[:24]}.service"

PrepareVless

PrepareVless(profile: VpnProfile, endpoint: VpnEndpoint, settings: VlessRealitySettings, material: VlessMaterial) -> PreparedVpnTransport

Prepare an owned installer and one private import bundle per declared device.

Source code in installed/flayer/profiles/vless.py

def PrepareVless(
    profile: VpnProfile, endpoint: VpnEndpoint, settings: VlessRealitySettings,
    material: VlessMaterial,
) -> PreparedVpnTransport:
    """Prepare an owned installer and one private import bundle per declared device."""

    server = RenderVlessServer(profile, settings, material)
    identity = asdict(profile.identity)
    owner = hashlib.sha256(_Json(identity)).hexdigest()
    deployment = _Json({
        "schema_version": 1, "owner": owner, "config_sha256": hashlib.sha256(server).hexdigest(),
        "port": endpoint.port, "xray_version": XRAY_VERSION,
    })
    installer = files("flayer.profiles").joinpath("_vless_guest.py").read_bytes().replace(b"\r\n", b"\n")
    server_bundle = BuildVpnServerBundle(profile, VLESS_TRANSPORT, (
        Artifact("server.json", server), Artifact("install.py", installer),
        Artifact("deployment.json", deployment),
    ))
    device_bundles = tuple(BuildVpnDeviceBundle(profile, VLESS_TRANSPORT, device.device_id, (
        Artifact("client.json", RenderVlessClient(profile, endpoint, settings, material, device.device_id)),
        Artifact("import.txt", (BuildVlessUri(
            profile, endpoint, settings, material, device.device_id,
        ) + "\n").encode("utf-8")),
        *BuildAmneziaArtifacts(
            RenderVlessClient(profile, endpoint, settings, material, device.device_id),
            endpoint.host, device.device_id, profile.routes.dns_servers,
        ),
    )) for device in profile.devices)

    return PreparedVpnTransport(profile, VLESS_CAPABILITIES, server_bundle, device_bundles)