UDS under Windows POC

winuds.pyx

from ctypes import windll, Structure, c_char, c_ushort, c_char_p, POINTER, c_int, WinError
from ctypes.wintypes import WORD

from typing import Literal, TypeAlias

__all__ = "test",

_ExtraInfo: TypeAlias = Literal["peername", "socket", "sockname"]


cdef extern from *:
    # language=C
    """
    #undef UNICODE
    #define VC_EXTRALEAN
    #define WIN32_LEAN_AND_MEAN
    #include <windows.h>
    #include <ws2tcpip.h>
    """


cdef extern from "ws2def.h":
    ctypedef unsigned short ADDRESS_FAMILY;
    cdef unsigned short AF_UNIX;
    ctypedef struct SOCKADDR:
        ADDRESS_FAMILY sa_family;
        char sa_data[14];


cdef extern from "WinSock2.h":
    ctypedef unsigned int SOCKET;
    const int SOCK_STREAM;
    const unsigned int INVALID_SOCKET;
    const unsigned int SOCKET_ERROR;

    cdef int bind(SOCKET s, SOCKADDR *name, int namelen);
    cdef int listen(SOCKET s, int backlog);
    cdef SOCKET socket(int af, int type, int protocol);
    cdef int connect(SOCKET s, SOCKADDR *name, int namelen);


cdef extern from "afunix.h":
    const int UNIX_PATH_MAX;

    ctypedef struct SOCKADDR_UN:
        unsigned short sun_family;
        char sun_path[UNIX_PATH_MAX];


def _WSAStart():
    WSADESCRIPTION_LEN = 256
    WSASYS_STATUS_LEN = 128

    class WSADATA(Structure):
        _fields_ = [
            ("wVersion", WORD),
            ("wHighVersion", WORD),
            ("szDescription", c_char * (WSADESCRIPTION_LEN + 1)),
            ("szSystemStatus", c_char * (WSASYS_STATUS_LEN + 1)),
            ("iMaxSockets", c_ushort),
            ("iMaxUdpDg", c_ushort),
            ("lpVendorInfo", c_char_p),
        ]

    LP_WSADATA = POINTER(WSADATA)

    WSAStartup = windll.Ws2_32.WSAStartup
    WSAStartup.argtypes = (WORD, POINTER(WSADATA))
    WSAStartup.restype = c_int

    def MAKEWORD(bLow, bHigh):
        return (bHigh << 8) + bLow

    wsaData = WSADATA()
    ret = WSAStartup(MAKEWORD(2, 2), LP_WSADATA(wsaData))
    if ret != 0:
        raise WinError(ret)


def test():
    _WSAStart()
    cdef SOCKET s = socket(AF_UNIX, SOCK_STREAM, 0)
    if s == INVALID_SOCKET:
        raise RuntimeError("Invalid socket")
    cdef SOCKADDR_UN addr
    addr.sun_family = AF_UNIX
    addr.sun_path = <char *>b"C:\\Users\\User\\Projects\\cpptest\\cmake-build-debug\\server.sock\0"
    if connect(s, <SOCKADDR *>&addr, sizeof(addr)) == SOCKET_ERROR:
        raise RuntimeError("Socket error")
    breakpoint()