mysocketctl/connect.py
def connect(
    ctx,
    port,
    name,
    protected,
    username,
    password,
    host,
    type,
    engine,
    cloudauth,
    allowed_email_addresses,
    allowed_email_domains,
):
    """Quckly connect, Wrapper around sockets and tunnels"""

    if cloudauth:
        cloudauth = True
        allowed_email_addresses_list = []
        if allowed_email_addresses:
            for email in allowed_email_addresses.split(","):
                if validate_email(email.strip()):
                    allowed_email_addresses_list.append(email.strip())
                else:
                    print("Warning: ignoring invalid email " + email.strip())

        allowed_email_domain_list = []
        if allowed_email_domains:
            for domain in allowed_email_domains.split(","):
                allowed_email_domain_list.append(domain.strip())

        # check if both email and domain list are empty and warn
        if not allowed_email_domain_list and not allowed_email_addresses_list:
            print(
                "Error: no allowed email addresses or domains provided. You will be unabled to get to your socket"
            )
            sys.exit(1)
    else:
        cloudauth = False
        allowed_email_domain_list = []
        allowed_email_addresses_list = []

    if protected:
        if not username:
            print("--username required when using --protected")
            sys.exit(1)
        if not password:
            print("--password required when using --protected")
            sys.exit(1)
    if not name:
        name = f"Local port {port}"

    authorization_header = get_auth_header()
    new_conn = new_connection(
        authorization_header,
        name,
        protected,
        str(username),
        str(password),
        str(type),
        cloudauth,
        allowed_email_addresses_list,
        allowed_email_domain_list,
    )
    remote_bind_port = new_conn["tunnels"][0]["local_port"]
    ssh_server = new_conn["tunnels"][0]["tunnel_server"]
    ssh_server = "ssh.mysocket.io"
    ssh_user = str(new_conn["user_name"])

    print_sockets([new_conn])
    if protected:
        print_protected(username, password)
    if cloudauth:
        print_cloudauth(allowed_email_addresses_list, allowed_email_domain_list)

    time.sleep(2)
    ssh_tunnel(port, remote_bind_port, ssh_server, ssh_user, host=host, engine=engine)
    print("cleaning up...")
    ctx.invoke(mysocketctl.socket.delete, socket_id=new_conn["socket_id"])
Quckly connect, Wrapper around sockets and tunnels
Quckly connect, Wrapper around sockets and tunnels
Quckly connect Wrapper around sockets and tunnels
def connect(
    ctx,
    port,
    name,
    protected,
    username,
    password,
    host,
    type,
    engine,
    cloudauth,
    allowed_email_addresses,
    allowed_email_domains,
):
    if cloudauth:
        cloudauth = True
        allowed_email_addresses_list = []
        if allowed_email_addresses:
            for email in allowed_email_addresses.split(","):
                if validate_email(email.strip()):
                    allowed_email_addresses_list.append(email.strip())
                else:
                    print("Warning: ignoring invalid email " + email.strip())
        allowed_email_domain_list = []
        if allowed_email_domains:
            for domain in allowed_email_domains.split(","):
                allowed_email_domain_list.append(domain.strip())
        if not allowed_email_domain_list and not allowed_email_addresses_list:
            print(
                "Error: no allowed email addresses or domains provided. You will be unabled to get to your socket"
            )
            sys.exit(1)
    else:
        cloudauth = False
        allowed_email_domain_list = []
        allowed_email_addresses_list = []
    if protected:
        if not username:
            print("--username required when using --protected")
            sys.exit(1)
        if not password:
            print("--password required when using --protected")
            sys.exit(1)
    if not name:
        name = f"Local port {port}"
    authorization_header = get_auth_header()
    new_conn = new_connection(
        authorization_header,
        name,
        protected,
        str(username),
        str(password),
        str(type),
        cloudauth,
        allowed_email_addresses_list,
        allowed_email_domain_list,
    )
    remote_bind_port = new_conn["tunnels"][0]["local_port"]
    ssh_server = new_conn["tunnels"][0]["tunnel_server"]
    ssh_server = "ssh.mysocket.io"
    ssh_user = str(new_conn["user_name"])
    print_sockets([new_conn])
    if protected:
        print_protected(username, password)
    if cloudauth:
        print_cloudauth(allowed_email_addresses_list, allowed_email_domain_list)
    time.sleep(2)
    ssh_tunnel(port, remote_bind_port, ssh_server, ssh_user, host=host, engine=engine)
    print("cleaning up...")
    ctx.invoke(mysocketctl.socket.delete, socket_id=new_conn["socket_id"])
def connect ( ctx , port , name , protected , username , password , host , type , engine , cloudauth , allowed_email_addresses , allowed_email_domains , ) : if cloudauth : cloudauth = True allowed_email_addresses_list = [ ] if allowed_email_addresses : for email in allowed_email_addresses . split ( "," ) : if validate_email ( email . strip ( ) ) : allowed_email_addresses_list . append ( email . strip ( ) ) else : print ( "Warning: ignoring invalid email " + email . strip ( ) ) allowed_email_domain_list = [ ] if allowed_email_domains : for domain in allowed_email_domains . split ( "," ) : allowed_email_domain_list . append ( domain . strip ( ) ) if not allowed_email_domain_list and not allowed_email_addresses_list : print ( "Error: no allowed email addresses or domains provided. You will be unabled to get to your socket" ) sys . exit ( 1 ) else : cloudauth = False allowed_email_domain_list = [ ] allowed_email_addresses_list = [ ] if protected : if not username : print ( "--username required when using --protected" ) sys . exit ( 1 ) if not password : print ( "--password required when using --protected" ) sys . exit ( 1 ) if not name : name = f"Local port {port}" authorization_header = get_auth_header ( ) new_conn = new_connection ( authorization_header , name , protected , str ( username ) , str ( password ) , str ( type ) , cloudauth , allowed_email_addresses_list , allowed_email_domain_list , ) remote_bind_port = new_conn [ "tunnels" ] [ 0 ] [ "local_port" ] ssh_server = new_conn [ "tunnels" ] [ 0 ] [ "tunnel_server" ] ssh_server = "ssh.mysocket.io" ssh_user = str ( new_conn [ "user_name" ] ) print_sockets ( [ new_conn ] ) if protected : print_protected ( username , password ) if cloudauth : print_cloudauth ( allowed_email_addresses_list , allowed_email_domain_list ) time . sleep ( 2 ) ssh_tunnel ( port , remote_bind_port , ssh_server , ssh_user , host = host , engine = engine ) print ( "cleaning up..." ) ctx . invoke ( mysocketctl . socket . delete , socket_id = new_conn [ "socket_id" ] )
Quckly connect, Wrapper around sockets and tunnels
Quckly connect Wrapper around sockets and tunnels
"""Quckly connect, Wrapper around sockets and tunnels""" # check if both email and domain list are empty and warn
ctx
port
name
protected
username
password
host
type
engine
cloudauth
allowed_email_addresses
allowed_email_domains