socket.rs

use crossbeam_utils::atomic::AtomicCell;
use gethostname::gethostname;
#[cfg(all(unix, not(target_os = "redox")))]
use nix::unistd::sethostname;
use num_traits::ToPrimitive;
use socket2::{Domain, Protocol, Socket, Type as SocketType};
use std::convert::TryFrom;
use std::mem::MaybeUninit;
use std::net::{self, Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr, ToSocketAddrs};
use std::time::{Duration, Instant};
use std::{ffi, io};

use crate::builtins::int;
use crate::builtins::pystr::PyStrRef;
use crate::builtins::pytype::PyTypeRef;
use crate::builtins::tuple::PyTupleRef;
use crate::byteslike::{ArgBytesLike, ArgMemoryBuffer};
use crate::common::lock::{PyMappedRwLockReadGuard, PyRwLock, PyRwLockReadGuard};
use crate::exceptions::{IntoPyException, PyBaseExceptionRef};
use crate::function::{FuncArgs, OptionalArg, OptionalOption};
use crate::utils::{Either, ToCString};
use crate::VirtualMachine;
use crate::{
    IntoPyObject, PyClassImpl, PyObjectRef, PyRef, PyResult, PyValue, StaticType,
    TryFromBorrowedObject, TryFromObject, TypeProtocol,
};

#[cfg(unix)]
type RawSocket = std::os::unix::io::RawFd;
#[cfg(windows)]
type RawSocket = std::os::windows::raw::SOCKET;

#[cfg(unix)]
macro_rules! errcode {
    ($e:ident) => {
        c::$e
    };
}
#[cfg(windows)]
macro_rules! errcode {
    ($e:ident) => {
        paste::paste!(c::[<WSA $e>])
    };
}

#[cfg(unix)]
use libc as c;
#[cfg(windows)]
mod c {
    pub use winapi::shared::ifdef::IF_MAX_STRING_SIZE as IF_NAMESIZE;
    pub use winapi::shared::netioapi::{if_indextoname, if_nametoindex};
    pub use winapi::shared::ws2def::*;
    pub use winapi::um::winsock2::{
        SD_BOTH as SHUT_RDWR, SD_RECEIVE as SHUT_RD, SD_SEND as SHUT_WR, SOCK_DGRAM, SOCK_RAW,
        SOCK_RDM, SOCK_SEQPACKET, SOCK_STREAM, SOL_SOCKET, SO_BROADCAST, SO_ERROR, SO_LINGER,
        SO_OOBINLINE, SO_REUSEADDR, SO_TYPE, *,
    };
}
#[cfg(windows)]
use winapi::shared::netioapi;

fn get_raw_sock(obj: PyObjectRef, vm: &VirtualMachine) -> PyResult<RawSocket> {
    #[cfg(unix)]
    type CastFrom = libc::c_long;
    #[cfg(windows)]
    type CastFrom = libc::c_longlong;

    // should really just be to_index() but test_socket tests the error messages explicitly
    if obj.isinstance(&vm.ctx.types.float_type) {
        return Err(vm.new_type_error("integer argument expected, got float".to_owned()));
    }
    let int = vm
        .to_index_opt(obj)
        .unwrap_or_else(|| Err(vm.new_type_error("an integer is required".to_owned())))?;
    int::try_to_primitive::<CastFrom>(int.as_bigint(), vm).map(|sock| sock as RawSocket)
}

#[cfg(unix)]
mod nullable_socket {
    use super::*;
    use std::os::unix::io::AsRawFd;

    #[derive(Debug)]
    #[repr(transparent)]
    pub struct NullableSocket(Option<socket2::Socket>);
    impl From<socket2::Socket> for NullableSocket {
        fn from(sock: socket2::Socket) -> Self {
            NullableSocket(Some(sock))
        }
    }
    impl NullableSocket {
        pub fn invalid() -> Self {
            Self(None)
        }
        pub fn get(&self) -> Option<&socket2::Socket> {
            self.0.as_ref()
        }
        pub fn fd(&self) -> RawSocket {
            self.get().map_or(INVALID_SOCKET, |sock| sock.as_raw_fd())
        }
        pub fn insert(&mut self, sock: socket2::Socket) -> &mut socket2::Socket {
            self.0.insert(sock)
        }
    }
}
#[cfg(windows)]
mod nullable_socket {
    use super::*;
    use std::os::windows::io::{AsRawSocket, FromRawSocket};

    // TODO: may change if windows changes its TcpStream repr
    #[derive(Debug)]
    #[repr(transparent)]
    pub struct NullableSocket(socket2::Socket);
    impl From<socket2::Socket> for NullableSocket {
        fn from(sock: socket2::Socket) -> Self {
            NullableSocket(sock)
        }
    }
    impl NullableSocket {
        pub fn invalid() -> Self {
            // TODO: may become UB in the future; maybe see rust-lang/rust#74699
            Self(unsafe { socket2::Socket::from_raw_socket(INVALID_SOCKET) })
        }
        pub fn get(&self) -> Option<&socket2::Socket> {
            (self.0.as_raw_socket() != INVALID_SOCKET).then(|| &self.0)
        }
        pub fn fd(&self) -> RawSocket {
            self.0.as_raw_socket()
        }
        pub fn insert(&mut self, sock: socket2::Socket) -> &mut socket2::Socket {
            self.0 = sock;
            &mut self.0
        }
    }
}
use nullable_socket::NullableSocket;
impl Default for NullableSocket {
    fn default() -> Self {
        Self::invalid()
    }
}

#[pyclass(module = "socket", name = "socket")]
#[derive(Debug, PyValue)]
pub struct PySocket {
    kind: AtomicCell<i32>,
    family: AtomicCell<i32>,
    proto: AtomicCell<i32>,
    pub(crate) timeout: AtomicCell<f64>,
    sock: PyRwLock<NullableSocket>,
}

impl Default for PySocket {
    fn default() -> Self {
        PySocket {
            kind: AtomicCell::default(),
            family: AtomicCell::default(),
            proto: AtomicCell::default(),
            timeout: AtomicCell::new(-1.0),
            sock: PyRwLock::new(NullableSocket::invalid()),
        }
    }
}

pub type PySocketRef = PyRef<PySocket>;

#[cfg(windows)]
const CLOSED_ERR: i32 = c::WSAENOTSOCK;
#[cfg(unix)]
const CLOSED_ERR: i32 = c::EBADF;
#[pyimpl(flags(BASETYPE))]
impl PySocket {
    pub fn sock_opt(&self) -> Option<PyMappedRwLockReadGuard<'_, Socket>> {
        PyRwLockReadGuard::try_map(self.sock.read(), |sock| sock.get()).ok()
    }

    fn sock_io(&self) -> io::Result<PyMappedRwLockReadGuard<'_, Socket>> {
        self.sock_opt()
            .ok_or_else(|| io::Error::from_raw_os_error(CLOSED_ERR))
    }

    pub fn sock(&self, vm: &VirtualMachine) -> PyResult<PyMappedRwLockReadGuard<'_, Socket>> {
        self.sock_io().map_err(|e| e.into_pyexception(vm))
    }

    #[pyslot]
    fn tp_new(cls: PyTypeRef, _args: FuncArgs, vm: &VirtualMachine) -> PyResult {
        Self::default().into_pyresult_with_type(vm, cls)
    }

    #[pymethod(magic)]
    fn init(
        &self,
        family: OptionalArg<i32>,
        socket_kind: OptionalArg<i32>,
        proto: OptionalArg<i32>,
        fileno: OptionalOption<PyObjectRef>,
        vm: &VirtualMachine,
    ) -> PyResult<()> {
        let mut family = family.unwrap_or(-1);
        let mut socket_kind = socket_kind.unwrap_or(-1);
        let mut proto = proto.unwrap_or(-1);
        let fileno = fileno
            .flatten()
            .map(|obj| get_raw_sock(obj, vm))
            .transpose()?;
        let sock;
        if let Some(fileno) = fileno {
            sock = sock_from_raw(fileno, vm)?;
            match sock.local_addr() {
                Ok(addr) if family == -1 => family = addr.family() as i32,
                Err(e)
                    if family == -1
                        || matches!(
                            e.raw_os_error(),
                            Some(errcode!(ENOTSOCK)) | Some(errcode!(EBADF))
                        ) =>
                {
                    std::mem::forget(sock);
                    return Err(e.into_pyexception(vm));
                }
                _ => {}
            }
            if socket_kind == -1 {
                // TODO: when socket2 cuts a new release, type will be available on all os
                // socket_kind = sock.r#type().map_err(|e| e.into_pyexception(vm))?.into();
                let res = unsafe {
                    c::getsockopt(
                        sock_fileno(&sock) as _,
                        c::SOL_SOCKET,
                        c::SO_TYPE,
                        &mut socket_kind as *mut libc::c_int as *mut _,
                        &mut (std::mem::size_of::<i32>() as _),
                    )
                };
                if res < 0 {
                    return Err(super::os::errno_err(vm));
                }
            }
            cfg_if::cfg_if! {
                if #[cfg(any(
                    target_os = "android",
                    target_os = "freebsd",
                    target_os = "fuchsia",
                    target_os = "linux",
                ))] {
                    if proto == -1 {
                        proto = sock.protocol().map_err(|e| e.into_pyexception(vm))?.map_or(0, Into::into);
                    }
                } else {
                    proto = 0;
                }
            }
        } else {
            if family == -1 {
                family = c::AF_INET as i32
            }
            if socket_kind == -1 {
                socket_kind = c::SOCK_STREAM
            }
            if proto == -1 {
                proto = 0
            }
            sock = Socket::new(
                Domain::from(family),
                SocketType::from(socket_kind),
                Some(Protocol::from(proto)),
            )
            .map_err(|err| err.into_pyexception(vm))?;
        };
        self.init_inner(family, socket_kind, proto, sock, vm)
    }

    fn init_inner(
        &self,
        family: i32,
        socket_kind: i32,
        proto: i32,
        sock: Socket,
        vm: &VirtualMachine,
    ) -> PyResult<()> {
        self.family.store(family);
        self.kind.store(socket_kind);
        self.proto.store(proto);
        let mut s = self.sock.write();
        let sock = s.insert(sock);
        let timeout = DEFAULT_TIMEOUT.load();
        self.timeout.store(timeout);
        if timeout >= 0.0 {
            sock.set_nonblocking(true)
                .map_err(|e| e.into_pyexception(vm))?;
        }
        Ok(())
    }

    #[inline]
    fn sock_op<F, R>(&self, vm: &VirtualMachine, select: SelectKind, f: F) -> PyResult<R>
    where
        F: FnMut() -> io::Result<R>,
    {
        self.sock_op_err(vm, select, f)
            .map_err(|e| e.into_pyexception(vm))
    }

    /// returns Err(blocking)
    pub fn get_timeout(&self) -> Result<Duration, bool> {
        let timeout = self.timeout.load();
        if timeout > 0.0 {
            Ok(Duration::from_secs_f64(timeout))
        } else {
            Err(timeout != 0.0)
        }
    }

    fn sock_op_err<F, R>(
        &self,
        vm: &VirtualMachine,
        select: SelectKind,
        f: F,
    ) -> Result<R, IoOrPyException>
    where
        F: FnMut() -> io::Result<R>,
    {
        self.sock_op_timeout_err(vm, select, self.get_timeout().ok(), f)
    }

    fn sock_op_timeout_err<F, R>(
        &self,
        vm: &VirtualMachine,
        select: SelectKind,
        timeout: Option<Duration>,
        mut f: F,
    ) -> Result<R, IoOrPyException>
    where
        F: FnMut() -> io::Result<R>,
    {
        let deadline = timeout.map(Deadline::new);

        loop {
            if deadline.is_some() || matches!(select, SelectKind::Connect) {
                let interval = deadline.as_ref().map(|d| d.time_until()).transpose()?;
                let res = sock_select(&*self.sock(vm)?, select, interval);
                match res {
                    Ok(true) => return Err(IoOrPyException::Timeout),
                    Err(e) if e.kind() == io::ErrorKind::Interrupted => {
                        vm.check_signals()?;
                        continue;
                    }
                    Err(e) => return Err(e.into()),
                    Ok(false) => {} // no timeout, continue as normal
                }
            }

            let err = loop {
                // loop on interrupt
                match f() {
                    Ok(x) => return Ok(x),
                    Err(e) if e.kind() == io::ErrorKind::Interrupted => vm.check_signals()?,
                    Err(e) => break e,
                }
            };
            if timeout.is_some() && err.kind() == io::ErrorKind::WouldBlock {
                continue;
            }
            return Err(err.into());
        }
    }

    fn extract_address(
        &self,
        addr: PyObjectRef,
        caller: &str,
        vm: &VirtualMachine,
    ) -> PyResult<socket2::SockAddr> {
        let family = self.family.load();
        match family {
            #[cfg(unix)]
            c::AF_UNIX => {
                use std::os::unix::ffi::OsStrExt;
                let buf = crate::byteslike::ArgStrOrBytesLike::try_from_object(vm, addr)?;
                let path = &*buf.borrow_bytes();
                if cfg!(any(target_os = "linux", target_os = "android")) && path.first() == Some(&0)
                {
                    use libc::{sa_family_t, socklen_t};
                    use {socket2::SockAddr, std::ptr};
                    unsafe {
                        // based on SockAddr::unix
                        // TODO: upstream or fix socklen check for SockAddr::unix()?
                        SockAddr::init(|storage, len| {
                            // Safety: `SockAddr::init` zeros the address, which is a valid
                            // representation.
                            let storage: &mut libc::sockaddr_un = &mut *storage.cast();
                            let len: &mut socklen_t = &mut *len;

                            let bytes = path;
                            if bytes.len() > storage.sun_path.len() {
                                return Err(io::Error::new(
                                    io::ErrorKind::InvalidInput,
                                    "path must be shorter than SUN_LEN",
                                ));
                            }

                            storage.sun_family = libc::AF_UNIX as sa_family_t;
                            // Safety: `bytes` and `addr.sun_path` are not overlapping and
                            // both point to valid memory.
                            // `SockAddr::init` zeroes the memory, so the path is already
                            // null terminated.
                            ptr::copy_nonoverlapping(
                                bytes.as_ptr(),
                                storage.sun_path.as_mut_ptr() as *mut u8,
                                bytes.len(),
                            );

                            let base = storage as *const _ as usize;
                            let path = &storage.sun_path as *const _ as usize;
                            let sun_path_offset = path - base;
                            let length = sun_path_offset + bytes.len();
                            *len = length as socklen_t;

                            Ok(())
                        })
                    }
                    .map(|(_, addr)| addr)
                } else {
                    socket2::SockAddr::unix(ffi::OsStr::from_bytes(path))
                }
                .map_err(|_| vm.new_os_error("AF_UNIX path too long".to_owned()))
            }
            c::AF_INET => {
                let tuple: PyTupleRef = addr.downcast().map_err(|obj| {
                    vm.new_type_error(format!(
                        "{}(): AF_INET address must be tuple, not {}",
                        caller,
                        obj.class().name()
                    ))
                })?;
                let tuple = tuple.as_slice();
                if tuple.len() != 2 {
                    return Err(
                        vm.new_type_error("AF_INET address must be a pair (host, post)".to_owned())
                    );
                }
                let addr = Address::from_tuple(tuple, vm)?;
                let mut addr4 = get_addr(vm, addr.host, c::AF_INET)?;
                match &mut addr4 {
                    SocketAddr::V4(addr4) => {
                        addr4.set_port(addr.port);
                    }
                    SocketAddr::V6(_) => unreachable!(),
                }
                Ok(addr4.into())
            }
            c::AF_INET6 => {
                let tuple: PyTupleRef = addr.downcast().map_err(|obj| {
                    vm.new_type_error(format!(
                        "{}(): AF_INET6 address must be tuple, not {}",
                        caller,
                        obj.class().name()
                    ))
                })?;
                let tuple = tuple.as_slice();
                match tuple.len() {
                    2 | 3 | 4 => {}
                    _ => {
                        return Err(vm.new_type_error(
                            "AF_INET6 address must be a tuple (host, port[, flowinfo[, scopeid]])"
                                .to_owned(),
                        ))
                    }
                }
                let (addr, flowinfo, scopeid) = Address::from_tuple_ipv6(tuple, vm)?;
                let mut addr6 = get_addr(vm, addr.host, c::AF_INET6)?;
                match &mut addr6 {
                    SocketAddr::V6(addr6) => {
                        addr6.set_port(addr.port);
                        addr6.set_flowinfo(flowinfo);
                        addr6.set_scope_id(scopeid);
                    }
                    SocketAddr::V4(_) => unreachable!(),
                }
                Ok(addr6.into())
            }
            _ => Err(vm.new_os_error(format!("{}(): bad family", caller))),
        }
    }

    fn connect_inner(
        &self,
        address: PyObjectRef,
        caller: &str,
        vm: &VirtualMachine,
    ) -> Result<(), IoOrPyException> {
        let sock_addr = self.extract_address(address, caller, vm)?;

        let err = match self.sock(vm)?.connect(&sock_addr) {
            Ok(()) => return Ok(()),
            Err(e) => e,
        };

        let wait_connect = if err.kind() == io::ErrorKind::Interrupted {
            vm.check_signals()?;
            self.timeout.load() != 0.0
        } else {
            #[cfg(unix)]
            use c::EINPROGRESS;
            #[cfg(windows)]
            use c::WSAEWOULDBLOCK as EINPROGRESS;

            self.timeout.load() > 0.0 && err.raw_os_error() == Some(EINPROGRESS)
        };

        if wait_connect {
            // basically, connect() is async, and it registers an "error" on the socket when it's
            // done connecting. SelectKind::Connect fills the errorfds fd_set, so if we wake up
            // from poll and the error is EISCONN then we know that the connect is done
            self.sock_op_err(vm, SelectKind::Connect, || {
                let sock = self.sock_io()?;
                let err = sock.take_error()?;
                match err {
                    Some(e) if e.raw_os_error() == Some(libc::EISCONN) => Ok(()),
                    Some(e) => Err(e),
                    // TODO: is this accurate?
                    None => Ok(()),
                }
            })
        } else {
            Err(err.into())
        }
    }

    #[pymethod]
    fn connect(&self, address: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
        self.connect_inner(address, "connect", vm)
            .map_err(|e| e.into_pyexception(vm))
    }

    #[pymethod]
    fn connect_ex(&self, address: PyObjectRef, vm: &VirtualMachine) -> PyResult<i32> {
        match self.connect_inner(address, "connect_ex", vm) {
            Ok(()) => Ok(0),
            Err(err) => err.errno(),
        }
    }

    #[pymethod]
    fn bind(&self, address: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
        let sock_addr = self.extract_address(address, "bind", vm)?;
        self.sock(vm)?
            .bind(&sock_addr)
            .map_err(|err| err.into_pyexception(vm))
    }

    #[pymethod]
    fn listen(&self, backlog: OptionalArg<i32>, vm: &VirtualMachine) -> PyResult<()> {
        let backlog = backlog.unwrap_or(128);
        let backlog = if backlog < 0 { 0 } else { backlog };
        self.sock(vm)?
            .listen(backlog)
            .map_err(|err| err.into_pyexception(vm))
    }

    #[pymethod]
    fn _accept(&self, vm: &VirtualMachine) -> PyResult<(RawSocket, PyObjectRef)> {
        let (sock, addr) = self.sock_op(vm, SelectKind::Read, || self.sock_io()?.accept())?;
        let fd = into_sock_fileno(sock);
        Ok((fd, get_addr_tuple(&addr, vm)))
    }

    #[pymethod]
    fn recv(
        &self,
        bufsize: usize,
        flags: OptionalArg<i32>,
        vm: &VirtualMachine,
    ) -> PyResult<Vec<u8>> {
        let flags = flags.unwrap_or(0);
        let mut buffer = Vec::with_capacity(bufsize);
        let sock = self.sock(vm)?;
        let n = self.sock_op(vm, SelectKind::Read, || {
            sock.recv_with_flags(spare_capacity_mut(&mut buffer), flags)
        })?;
        unsafe { buffer.set_len(n) };
        Ok(buffer)
    }

    #[pymethod]
    fn recv_into(
        &self,
        buf: ArgMemoryBuffer,
        flags: OptionalArg<i32>,
        vm: &VirtualMachine,
    ) -> PyResult<usize> {
        let flags = flags.unwrap_or(0);
        let sock = self.sock(vm)?;
        let mut buf = buf.borrow_buf_mut();
        let buf = &mut *buf;
        self.sock_op(vm, SelectKind::Read, || {
            sock.recv_with_flags(slice_as_uninit(buf), flags)
        })
    }

    #[pymethod]
    fn recvfrom(
        &self,
        bufsize: isize,
        flags: OptionalArg<i32>,
        vm: &VirtualMachine,
    ) -> PyResult<(Vec<u8>, PyObjectRef)> {
        let flags = flags.unwrap_or(0);
        let bufsize = bufsize
            .to_usize()
            .ok_or_else(|| vm.new_value_error("negative buffersize in recvfrom".to_owned()))?;
        let mut buffer = Vec::with_capacity(bufsize);
        let (n, addr) = self.sock_op(vm, SelectKind::Read, || {
            self.sock_io()?
                .recv_from_with_flags(spare_capacity_mut(&mut buffer), flags)
        })?;
        unsafe { buffer.set_len(n) };
        Ok((buffer, get_addr_tuple(&addr, vm)))
    }

    #[pymethod]
    fn recvfrom_into(
        &self,
        buf: ArgMemoryBuffer,
        nbytes: OptionalArg<isize>,
        flags: OptionalArg<i32>,
        vm: &VirtualMachine,
    ) -> PyResult<(usize, PyObjectRef)> {
        let mut buf = buf.borrow_buf_mut();
        let buf = &mut *buf;
        let buf = match nbytes {
            OptionalArg::Present(i) => {
                let i = i.to_usize().ok_or_else(|| {
                    vm.new_value_error("negative buffersize in recvfrom_into".to_owned())
                })?;
                buf.get_mut(..i).ok_or_else(|| {
                    vm.new_value_error("nbytes is greater than the length of the buffer".to_owned())
                })?
            }
            OptionalArg::Missing => buf,
        };
        let flags = flags.unwrap_or(0);
        let sock = self.sock(vm)?;
        let (n, addr) = self.sock_op(vm, SelectKind::Read, || {
            sock.recv_from_with_flags(slice_as_uninit(buf), flags)
        })?;
        Ok((n, get_addr_tuple(&addr, vm)))
    }

    #[pymethod]
    fn send(
        &self,
        bytes: ArgBytesLike,
        flags: OptionalArg<i32>,
        vm: &VirtualMachine,
    ) -> PyResult<usize> {
        let flags = flags.unwrap_or(0);
        let buf = bytes.borrow_buf();
        let buf = &*buf;
        self.sock_op(vm, SelectKind::Write, || {
            self.sock_io()?.send_with_flags(buf, flags)
        })
    }

    #[pymethod]
    fn sendall(
        &self,
        bytes: ArgBytesLike,
        flags: OptionalArg<i32>,
        vm: &VirtualMachine,
    ) -> PyResult<()> {
        let flags = flags.unwrap_or(0);

        let timeout = self.get_timeout().ok();

        let deadline = timeout.map(Deadline::new);

        let buf = bytes.borrow_buf();
        let buf = &*buf;
        let mut buf_offset = 0;
        // now we have like 3 layers of interrupt loop :)
        while buf_offset < buf.len() {
            let interval = deadline
                .as_ref()
                .map(|d| d.time_until().map_err(|e| e.into_pyexception(vm)))
                .transpose()?;
            self.sock_op_timeout_err(vm, SelectKind::Write, interval, || {
                let subbuf = &buf[buf_offset..];
                buf_offset += self.sock_io()?.send_with_flags(subbuf, flags)?;
                Ok(())
            })
            .map_err(|e| e.into_pyexception(vm))?;
            vm.check_signals()?;
        }
        Ok(())
    }

    #[pymethod]
    fn sendto(
        &self,
        bytes: ArgBytesLike,
        arg2: PyObjectRef,
        arg3: OptionalArg<PyObjectRef>,
        vm: &VirtualMachine,
    ) -> PyResult<usize> {
        // signature is bytes[, flags], address
        let (flags, address) = match arg3 {
            OptionalArg::Present(arg3) => {
                // should just be i32::try_from_obj but tests check for error message
                let int = vm.to_index_opt(arg2).unwrap_or_else(|| {
                    Err(vm.new_type_error("an integer is required".to_owned()))
                })?;
                let flags = int::try_to_primitive::<i32>(int.as_bigint(), vm)?;
                (flags, arg3)
            }
            OptionalArg::Missing => (0, arg2),
        };
        let addr = self.extract_address(address, "sendto", vm)?;
        let buf = bytes.borrow_buf();
        let buf = &*buf;
        self.sock_op(vm, SelectKind::Write, || {
            self.sock_io()?.send_to_with_flags(buf, &addr, flags)
        })
    }

    #[pymethod]
    fn close(&self, vm: &VirtualMachine) -> PyResult<()> {
        let sock = self.detach();
        if sock != INVALID_SOCKET {
            close_inner(sock, vm)?;
        }
        Ok(())
    }
    #[pymethod]
    #[inline]
    fn detach(&self) -> RawSocket {
        let sock = std::mem::replace(&mut *self.sock.write(), NullableSocket::invalid());
        std::mem::ManuallyDrop::new(sock).fd()
    }

    #[pymethod]
    fn fileno(&self) -> RawSocket {
        self.sock.read().fd()
    }

    #[pymethod]
    fn getsockname(&self, vm: &VirtualMachine) -> PyResult<PyObjectRef> {
        let addr = self
            .sock(vm)?
            .local_addr()
            .map_err(|err| err.into_pyexception(vm))?;

        Ok(get_addr_tuple(&addr, vm))
    }
    #[pymethod]
    fn getpeername(&self, vm: &VirtualMachine) -> PyResult<PyObjectRef> {
        let addr = self
            .sock(vm)?
            .peer_addr()
            .map_err(|err| err.into_pyexception(vm))?;

        Ok(get_addr_tuple(&addr, vm))
    }

    #[pymethod]
    fn gettimeout(&self) -> Option<f64> {
        let timeout = self.timeout.load();
        if timeout >= 0.0 {
            Some(timeout)
        } else {
            None
        }
    }

    #[pymethod]
    fn setblocking(&self, block: bool, vm: &VirtualMachine) -> PyResult<()> {
        self.timeout.store(if block { -1.0 } else { 0.0 });
        self.sock(vm)?
            .set_nonblocking(!block)
            .map_err(|err| err.into_pyexception(vm))
    }

    #[pymethod]
    fn getblocking(&self) -> bool {
        self.timeout.load() != 0.0
    }

    #[pymethod]
    fn settimeout(&self, timeout: Option<Duration>, vm: &VirtualMachine) -> PyResult<()> {
        self.timeout
            .store(timeout.map_or(-1.0, |d| d.as_secs_f64()));
        // even if timeout is > 0 the socket needs to be nonblocking in order for us to select() on
        // it
        self.sock(vm)?
            .set_nonblocking(timeout.is_some())
            .map_err(|err| err.into_pyexception(vm))
    }

    #[pymethod]
    fn getsockopt(
        &self,
        level: i32,
        name: i32,
        buflen: OptionalArg<i32>,
        vm: &VirtualMachine,
    ) -> PyResult {
        let fd = self.sock.read().fd() as _;
        let buflen = buflen.unwrap_or(0);
        if buflen == 0 {
            let mut flag: libc::c_int = 0;
            let mut flagsize = std::mem::size_of::<libc::c_int>() as _;
            let ret = unsafe {
                c::getsockopt(
                    fd,
                    level,
                    name,
                    &mut flag as *mut libc::c_int as *mut _,
                    &mut flagsize,
                )
            };
            if ret < 0 {
                Err(super::os::errno_err(vm))
            } else {
                Ok(vm.ctx.new_int(flag))
            }
        } else {
            if buflen <= 0 || buflen > 1024 {
                return Err(vm.new_os_error("getsockopt buflen out of range".to_owned()));
            }
            let mut buf = vec![0u8; buflen as usize];
            let mut buflen = buflen as _;
            let ret =
                unsafe { c::getsockopt(fd, level, name, buf.as_mut_ptr() as *mut _, &mut buflen) };
            buf.truncate(buflen as usize);
            if ret < 0 {
                Err(super::os::errno_err(vm))
            } else {
                Ok(vm.ctx.new_bytes(buf))
            }
        }
    }

    #[pymethod]
    fn setsockopt(
        &self,
        level: i32,
        name: i32,
        value: Option<Either<ArgBytesLike, i32>>,
        optlen: OptionalArg<u32>,
        vm: &VirtualMachine,
    ) -> PyResult<()> {
        let fd = self.sock.read().fd() as _;
        let ret = match (value, optlen) {
            (Some(Either::A(b)), OptionalArg::Missing) => b.with_ref(|b| unsafe {
                c::setsockopt(fd, level, name, b.as_ptr() as *const _, b.len() as _)
            }),
            (Some(Either::B(ref val)), OptionalArg::Missing) => unsafe {
                c::setsockopt(
                    fd,
                    level,
                    name,
                    val as *const i32 as *const _,
                    std::mem::size_of::<i32>() as _,
                )
            },
            (None, OptionalArg::Present(optlen)) => unsafe {
                c::setsockopt(fd, level, name, std::ptr::null(), optlen as _)
            },
            _ => {
                return Err(
                    vm.new_type_error("expected the value arg xor the optlen arg".to_owned())
                );
            }
        };
        if ret < 0 {
            Err(super::os::errno_err(vm))
        } else {
            Ok(())
        }
    }

    #[pymethod]
    fn shutdown(&self, how: i32, vm: &VirtualMachine) -> PyResult<()> {
        let how = match how {
            c::SHUT_RD => Shutdown::Read,
            c::SHUT_WR => Shutdown::Write,
            c::SHUT_RDWR => Shutdown::Both,
            _ => {
                return Err(
                    vm.new_value_error("`how` must be SHUT_RD, SHUT_WR, or SHUT_RDWR".to_owned())
                )
            }
        };
        self.sock(vm)?
            .shutdown(how)
            .map_err(|err| err.into_pyexception(vm))
    }

    #[pyproperty(name = "type")]
    fn kind(&self) -> i32 {
        self.kind.load()
    }
    #[pyproperty]
    fn family(&self) -> i32 {
        self.family.load()
    }
    #[pyproperty]
    fn proto(&self) -> i32 {
        self.proto.load()
    }

    #[pymethod(magic)]
    fn repr(&self) -> String {
        format!(
            "<socket object, fd={}, family={}, type={}, proto={}>",
            // cast because INVALID_SOCKET is unsigned, so would show usize::MAX instead of -1
            self.sock.read().fd() as i64,
            self.family.load(),
            self.kind.load(),
            self.proto.load(),
        )
    }
}

impl io::Read for PySocketRef {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        <&Socket as io::Read>::read(&mut &*self.sock_io()?, buf)
    }
}
impl io::Write for PySocketRef {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        <&Socket as io::Write>::write(&mut &*self.sock_io()?, buf)
    }
    fn flush(&mut self) -> io::Result<()> {
        <&Socket as io::Write>::flush(&mut &*self.sock_io()?)
    }
}

struct Address {
    host: PyStrRef,
    port: u16,
}

impl ToSocketAddrs for Address {
    type Iter = std::vec::IntoIter<SocketAddr>;
    fn to_socket_addrs(&self) -> io::Result<Self::Iter> {
        (self.host.as_str(), self.port).to_socket_addrs()
    }
}

impl TryFromObject for Address {
    fn try_from_object(vm: &VirtualMachine, obj: PyObjectRef) -> PyResult<Self> {
        let tuple = PyTupleRef::try_from_object(vm, obj)?;
        if tuple.as_slice().len() != 2 {
            Err(vm.new_type_error("Address tuple should have only 2 values".to_owned()))
        } else {
            Self::from_tuple(tuple.as_slice(), vm)
        }
    }
}

impl Address {
    fn from_tuple(tuple: &[PyObjectRef], vm: &VirtualMachine) -> PyResult<Self> {
        let host = PyStrRef::try_from_object(vm, tuple[0].clone())?;
        let port = i32::try_from_borrowed_object(vm, &tuple[1])?;
        let port = port
            .to_u16()
            .ok_or_else(|| vm.new_overflow_error("port must be 0-65535.".to_owned()))?;
        Ok(Address { host, port })
    }
    fn from_tuple_ipv6(tuple: &[PyObjectRef], vm: &VirtualMachine) -> PyResult<(Self, u32, u32)> {
        let addr = Address::from_tuple(tuple, vm)?;
        let flowinfo = tuple
            .get(2)
            .map(|obj| u32::try_from_borrowed_object(vm, obj))
            .transpose()?
            .unwrap_or(0);
        let scopeid = tuple
            .get(3)
            .map(|obj| u32::try_from_borrowed_object(vm, obj))
            .transpose()?
            .unwrap_or(0);
        if flowinfo > 0xfffff {
            return Err(vm.new_overflow_error("flowinfo must be 0-1048575.".to_owned()));
        }
        Ok((addr, flowinfo, scopeid))
    }
}

fn get_ip_addr_tuple(addr: &SocketAddr, vm: &VirtualMachine) -> PyObjectRef {
    match addr {
        SocketAddr::V4(addr) => (addr.ip().to_string(), addr.port()).into_pyobject(vm),
        SocketAddr::V6(addr) => (
            addr.ip().to_string(),
            addr.port(),
            addr.flowinfo(),
            addr.scope_id(),
        )