create_opensips_cfg

#! /usr/bin/python3
import pymysql
import pymysql.cursors
import secrets
import os
from pybars import Compiler



def check_support(param):
    if param == 1:
        return True
    return False

def create_proto_support_code( public_ipv4, proto, proto_port, proto_autoscaling=False ):
    if not proto_autoscaling:
        return f"""listen={proto}:{public_ipv4}:{proto_port}"""

    profile = f"PROFILE_{proto.upper()}"
    workers = f"{proto.upper()}_workers="
    return f"""listen={proto}:{public_ipv4}:{proto_port} use_auto_scaling_profile {profile}
 
"""

def create_proto_support_code_v2( public_ipv4, proto, proto_port, proto_autoscaling=False ):
    profile = f"PROFILE_{proto.upper()}"
    workers = f"{proto.upper()}_workers="
    autoscaling_code = ""
    if proto_autoscaling:
        autoscaling_code = f"""auto_scaling_profile = {profile}
    scale up to 6 on 70% for 4 cycles within 5   
    scale down to 2 on 18% for 10 cycles"""
    socket_listening_code = f"socket={proto}:0.0.0.0:{proto_port}"

    full_proto_code = f"""{autoscaling_code}
{socket_listening_code}"""

    return full_proto_code

def connect_to_db():
    host = os.getenv("DB_HOST")
    user = os.getenv("DB_USER")
    password = os.getenv("DB_PASS")
    db_name = os.getenv("DB_NAME")
    db = pymysql.connect(host,user,password,db_name)
    return db

def get_db_params():
    mock_data = True
    if not mock_data:
        db = connect_to_db()
        # populate database parameters
        cursor = db.cursor()
        cursor = db.cursor(pymysql.cursors.DictCursor)

        cursor.execute("SELECT opensips_config, udp_support, udp_port, udp_autoscaling, tcp_support, tcp_port, tcp_autoscaling, tls_support, tls_port, ws_support, ws_port, wss_support, wss_port FROM customizations")
        data = cursor.fetchone()

        # disconnect from server
        db.close()
        return data

    data = {}
    # open template opensips config
    data['opensips_config'] = open("/tmp/opensips.cfg", "r+").read()
    data['udp_support'] = 1
    data['udp_port'] = '5060'
    data['udp_autoscaling'] = 1
    data['tcp_support'] = 0
    data['tcp_port'] = '5080'
    data['tcp_autoscaling'] = 0
    data['tls_support'] = 0
    data['tls_port'] = '5061'
    data['ws_support'] = 1
    data['ws_port'] = '10061'
    data['wss_support'] = 0
    data['wss_port'] = '10062'
    return data

def generate_carrier_key(length=16):
    carrier_key = secrets.token_urlsafe(length)
    return carrier_key

print("updating OpenSIPs config with database parameters")
cfg_path = "/etc/opensips/opensips.cfg"
cfg_path_new = "/etc/opensips/opensips.cfg.new"
public_ipv4 = os.getenv('PUBLIC_IPV4')
private_ipv4 = os.getenv('PRIVATE_IPV4')
data = get_db_params()
source = data['opensips_config']

#print("source ", source)
with open(cfg_path_new, "w") as file:
    data = get_db_params()
    print ("UDP support=" + str( data['udp_support'] ))
    print ("UDP auto scaling=" + str( data['udp_autoscaling'] ))
    print ("TCP support=" + str( data['tcp_support'] ))
    print ("TCP auto scaling=" + str( data['tcp_autoscaling'] ))
    print ("TLS support=" + str( data['tls_support'] ))

    udp_support = check_support( data['udp_support'] )
    udp_autoscaling = check_support( data['udp_autoscaling'] )
    udp_port = data['udp_port']
    tcp_support = check_support( data['tcp_support'] )
    tcp_autoscaling = check_support( data['tcp_autoscaling'] )
    tcp_port = data['tcp_port']
    tls_support = check_support( data['tls_support'] )
    tls_port = data['tls_port']
    tls_support = True
    tls_ip_addr = ""
    tls_port = ""
    cert_dir = ""
    tls_client_ip_addr = ""
    tls_client_port = ""
    ws_port = data['ws_port']
    ws_support = data['ws_support']
    wss_port = data['wss_port']
    wss_support = data['wss_support']

    compiler = Compiler()
    template = compiler.compile(source)
    replacements = {}

    replacements['api_url'] = os.getenv('API_URL')
    replacements['private_ipv4'] = os.getenv('PRIVATE_IPV4')
    replacements['public_ipv4'] = os.getenv('PUBLIC_IPV4')
    replacements['rtpproxy_ipv4'] = os.getenv('RTPPROXY_IPV4')
    #replacements['advertized_ipv4'] = os.getenv('ADVERTIZED_IPV4')
    replacements['advertized_ipv4'] = os.getenv('PUBLIC_IPV4')
    replacements['db_user'] = os.getenv('DB_USER')
    replacements['db_name'] = os.getenv('DB_NAME')
    replacements['db_pass'] = os.getenv('DB_PASS')
    replacements['db_host'] = os.getenv('DB_HOST')
    replacements['db_opensips'] = os.getenv('DB_OPENSIPS')
    replacements['db_homer'] = os.getenv('DB_HOMER')
    replacements['lineblocs_key'] = os.getenv('LINEBLOCS_KEY')
    replacements['deployment_domain'] = os.getenv('DEPLOYMENT_DOMAIN')
    carrier_key = os.getenv('CARRIER_KEY') 
    if carrier_key is None:
        carrier_key = generate_carrier_key()
        print(f"generated carrier key: {carrier_key}")

    replacements['carrier_key'] = carrier_key


    if data['udp_support']:
        udp_code = create_proto_support_code_v2( public_ipv4, 'udp', udp_port, udp_autoscaling )
        replacements['udp_support'] = udp_code
        replacements['udp_load_module'] = f"""loadmodule "proto_udp.so"
modparam("proto_udp", "udp_port", {udp_port})"""
    if data['tcp_support']:
        tcp_code = create_proto_support_code_v2( public_ipv4, 'tcp', tcp_port, tcp_autoscaling )
        replacements['tcp_support'] = tcp_code
        replacements['tcp_load_module'] = f"""loadmodule "proto_tcp.so"
modparam("proto_tcp", "tcp_port", {tcp_port})"""
    if data['ws_support']:
        ws_code = create_proto_support_code_v2( public_ipv4, 'ws', ws_port)
        replacements['ws_support'] = ws_code
        replacements['ws_load_module'] = f"""loadmodule "proto_ws.so"
"""
    if data['wss_support']:
        wss_code = create_proto_support_code( public_ipv4, 'wss', wss_port)
        replacements['wss_support'] = wss_code
        replacements['wss_load_module'] = f"""loadmodule "proto_wss.so"
modparam("proto_wss", "certificate", "{cert_dir}/cert.pem")
modparam("proto_wss", "private_key", "{cert_dir}/key.pem")
modparam("proto_wss", "verify_cert", 0)"""
    if data['tls_support']:
        tls_code = create_proto_support_code( public_ipv4, 'tls', tls_port, False)
        replacements['tls_support'] = tls_code
        replacements['tls_load_module'] = f"""loadmodule "proto_tls.so"
modparam("proto_tls", "tls_port", {tls_port})
modparam("proto_tls", "verify_cert", "0")
modparam("proto_tls", "require_cert", "0")
modparam("proto_tls", "tls_method", "TLSv1")
modparam("proto_tls", "certificate", "{cert_dir}/rootCA/cacert.pem")           
modparam("proto_tls", "private_key", "{cert_dir}/rootCA/private/cakey.pem")    
modparam("proto_tls", "ca_list", "{cert_dir}/rootCA/cacert.pem")                
modparam("proto_tls", "ca_dir", "{cert_dir}/rootCA/") 

#first the  server domain
modparam("proto_tls", "server_domain", "sv_dom={tls_ip_addr}:{tls_port}")           
modparam("proto_tls", "certificate", "sv_dom:{cert_dir}/rootCA/cacert.pem")           
modparam("proto_tls", "private_key", "sv_dom:{cert_dir}/rootCA/private/cakey.pem")    
modparam("proto_tls", "ca_list", "sv_dom:{cert_dir}/rootCA/cacert.pem")  

#and the client domain                                                               
modparam("proto_tls", "client_domain", "cl_dom={tls_client_ip_addr}:{tls_client_port}")            
modparam("proto_tls", "certificate", "cl_dom:{cert_dir}/user/user-cert.pem")         
modparam("proto_tls", "private_key", "cl_dom:{cert_dir}/user/user-privkey.pem")       
modparam("proto_tls", "ca_list", "cl_dom:{cert_dir}/user/user-calist.pem")"""

    replacements['user_agent']= "Lineblocs SIP router (powered by OpenSIPs)"

    output = template(replacements)
    #print("contents are now")
    #print(output)
    print("writing new opensips.cfg contents to path " + cfg_path_new)

    print("contents ", output)
    file.write( output )

os.replace(cfg_path_new, cfg_path)