config_generator.py

import json 
import uuid 
import secrets 
import string 
from typing import Optional 
from shared .schemas import VlessConfig 






def generate_uuid ()->str :
    return str (uuid .uuid4 ())


def generate_short_id (length :int =8 )->str :

    return secrets .token_hex (length //2 )


def random_service_name (seed :int )->str :

    rng_chars =string .ascii_lowercase +string .digits 
    r =__import__ ("random").Random (seed )
    length =r .randint (6 ,20 )
    return "".join (r .choice (rng_chars )for _ in range (length ))






def build_server_config (
cfg :VlessConfig ,
vless_uuid :str ,
private_key :str ,
public_key :str ,
listen_ip :str ="0.0.0.0",
)->dict :


    stream =_build_stream_settings_server (cfg )


    inbound ={
    "tag":"vless-in",
    "listen":listen_ip ,
    "port":cfg .proxy_port ,
    "protocol":"vless",
    "settings":{
    "clients":[
    {
    "id":vless_uuid ,
    "flow":_pick_flow (cfg ),
    }
    ],
    "decryption":"none",
    },
    "streamSettings":stream ,
    "sniffing":{
    "enabled":True ,
    "destOverride":["http","tls","quic"],
    },
    }


    inbound ["streamSettings"]["realitySettings"]={
    "show":False ,
    "dest":f"{cfg .dest_domain }:443",
    "xver":0 ,
    "serverNames":[cfg .dest_domain ],
    "privateKey":private_key ,
    "shortIds":[cfg .short_id ],
    "spiderX":cfg .spider_x ,
    }

    config ={
    "log":{"loglevel":"warning"},
    "inbounds":[inbound ],
    "outbounds":[
    {"tag":"direct","protocol":"freedom"},
    {"tag":"block","protocol":"blackhole"},
    ],
    "routing":{
    "rules":[
    {"type":"field","ip":["geoip:private"],"outboundTag":"block"},
    ]
    },
    }

    return config 


def _pick_flow (cfg :VlessConfig )->str :
    if cfg .transport_type =="tcp":
        return "xtls-rprx-vision"
    return ""


def _build_stream_settings_server (cfg :VlessConfig )->dict :
    base ={
    "network":cfg .transport_type ,
    "security":"reality",
    }

    if cfg .transport_type =="grpc":
        base ["grpcSettings"]={
        "serviceName":cfg .grpc_service_name ,
        "multiMode":False ,
        }
    elif cfg .transport_type =="xhttp":
        base ["xhttpSettings"]={
        "mode":cfg .xhttp_mode ,
        "path":cfg .spider_x or "/",
        "host":cfg .dest_domain ,
        }

    return base 






def build_client_config (
cfg :VlessConfig ,
vless_uuid :str ,
server_ip :str ,
public_key :str ,
socks_port :int =10808 ,
http_port :int =10809 ,
)->dict :
    
    stream =_build_stream_settings_client (cfg ,public_key )

    outbound ={
    "tag":"proxy",
    "protocol":"vless",
    "settings":{
    "vnext":[
    {
    "address":server_ip ,
    "port":cfg .proxy_port ,
    "users":[
    {
    "id":vless_uuid ,
    "encryption":"none",
    "flow":_pick_flow (cfg ),
    }
    ],
    }
    ]
    },
    "streamSettings":stream ,
    }


    if cfg .mux_concurrency >0 :
        outbound ["mux"]={
        "enabled":True ,
        "concurrency":cfg .mux_concurrency ,
        "xudpConcurrency":cfg .mux_concurrency ,
        }


    if cfg .transport_type =="tcp"and cfg .fragment_strategy !="none":
        outbound ["streamSettings"]["sockopt"]={
        "dialerProxy":"fragment",
        }

    config ={
    "log":{"loglevel":"warning"},
    "inbounds":[
    {
    "tag":"socks",
    "listen":"127.0.0.1",
    "port":socks_port ,
    "protocol":"socks",
    "settings":{"auth":"noauth","udp":True },
    },
    {
    "tag":"http",
    "listen":"127.0.0.1",
    "port":http_port ,
    "protocol":"http",
    },
    ],
    "outbounds":[outbound ,{"tag":"direct","protocol":"freedom"}],
    }


    if cfg .transport_type =="tcp"and cfg .fragment_strategy !="none":
        config ["outbounds"].append (_build_fragment_outbound (cfg ))

    return config 


def _build_stream_settings_client (cfg :VlessConfig ,public_key :str )->dict :
    base :dict ={
    "network":cfg .transport_type ,
    "security":"reality",
    "realitySettings":{
    "fingerprint":cfg .fingerprint ,
    "serverName":cfg .dest_domain ,
    "publicKey":public_key ,
    "shortId":cfg .short_id ,
    "spiderX":cfg .spider_x ,
    },
    }

    if cfg .alpn :
        base ["realitySettings"]["alpn"]=cfg .alpn 

    if cfg .transport_type =="grpc":
        base ["grpcSettings"]={
        "serviceName":cfg .grpc_service_name ,
        }
    elif cfg .transport_type =="xhttp":
        headers ={"Host":cfg .dest_domain }
        headers .update (cfg .extra_headers )
        base ["xhttpSettings"]={
        "mode":cfg .xhttp_mode ,
        "path":cfg .spider_x or "/",
        "headers":headers ,
        }

    return base 


def _build_fragment_outbound (cfg :VlessConfig )->dict :
    return {
    "tag":"fragment",
    "protocol":"freedom",
    "settings":{
    "domainStrategy":"AsIs",
    },
    "streamSettings":{
    "sockopt":{
    "dialerProxy":"",
    "tcpKeepAliveInterval":0 ,
    }
    },
    "fragment":{
    "packets":cfg .fragment_strategy ,
    "length":f"{cfg .fragment_length_min }-{cfg .fragment_length_max }",
    "interval":f"{cfg .fragment_interval_min }-{cfg .fragment_interval_max }",
    },
    }






def generate_reality_keys ()->tuple [str ,str ]:

    import subprocess 
    result =subprocess .run (
    ["xray","x25519"],
    capture_output =True ,text =True ,timeout =10 
    )
    lines =result .stdout .strip ().splitlines ()
    priv =lines [0 ].split (": ")[1 ].strip ()
    pub =lines [1 ].split (": ")[1 ].strip ()
    return priv ,pub 


def save_config (config :dict ,path :str ):
    with open (path ,"w")as f :
        json .dump (config ,f ,indent =2 )