Function ql_syscall_connect

qiling/os/posix/syscall/socket.py:257–309

(ql: Qiling, sockfd: int, addr: int, addrlen: int)

def ql_syscall_connect(ql: Qiling, sockfd: int, addr: int, addrlen: int):
    sock = get_opened_socket(ql.os, sockfd)

    if isinstance(sock, int):
        return sock

    data = ql.mem.read(addr, addrlen)

    abits = ql.arch.bits
    endian = ql.arch.endian

    sockaddr = make_sockaddr(abits, endian)
    sockaddr_obj = sockaddr.from_buffer(data)

    if sock.family != sockaddr_obj.sa_family:
        return -1

    if sock.family == AF_UNIX:
        hpath, vpath = ql_unix_socket_path(ql, data[2:])

        # TODO: support connecting to fs_mapped unix sockets
        ql.log.debug(f'Connecting to "{vpath}"')
        dest = hpath

    elif sock.family == AF_INET:
        sockaddr_in = make_sockaddr_in(abits, endian)
        sockaddr_obj = sockaddr_in.from_buffer(data)

        port = ntohs(ql, sockaddr_obj.sin_port)
        host = inet_htoa(ql, sockaddr_obj.sin_addr.s_addr)

        ql.log.debug(f'Connecting to {host}:{port}')
        dest = (host, port)

    elif sock.family == AF_INET6 and ql.os.ipv6:
        sockaddr_in6 = make_sockaddr_in6(abits, endian)
        sockaddr_obj = sockaddr_in6.from_buffer(data)

        port = ntohs(ql, sockaddr_obj.sin6_port)
        host = inet6_htoa(ql, sockaddr_obj.sin6_addr.s6_addr)

        ql.log.debug(f'Connecting to {host}:{port}')
        dest = (host, port)

    else:
        return -EAFNOSUPPORT

    try:
        sock.connect(dest)
    except (ConnectionError, FileNotFoundError) as ex:
        return -ex.errno

    return 0


def ql_syscall_getsockopt(ql: Qiling, sockfd: int, level: int, optname: int, optval_addr: int, optlen_addr: int):