Module rsyscall.sys.un
#include <sys/un.h>
Expand source code Browse git
"`#include <sys/un.h>`"
from __future__ import annotations
import typing as t
from rsyscall._raw import ffi, lib # type: ignore
from rsyscall.sys.socket import AF, Sockaddr, _register_sockaddr
from rsyscall.path import Path
from dataclasses import dataclass
import os
from rsyscall.fcntl import O
if t.TYPE_CHECKING:
from rsyscall.handle import FileDescriptor
from rsyscall.thread import Thread
__all__ = [
"PathTooLongError",
"SockaddrUn",
]
class PathTooLongError(ValueError):
pass
@dataclass
class SockaddrUn(Sockaddr):
path: bytes
family = AF.UNIX
def __post_init__(self) -> None:
if len(self.path) > 108:
raise PathTooLongError("path", self.path, "is longer than the maximum unix address size")
@staticmethod
async def from_path(thr: Thread, path: t.Union[str, os.PathLike]) -> SockaddrUn:
"""Turn this path into a SockaddrUn, hacking around the 108 byte limit on socket addresses.
If the passed path is too long to fit in an address, this function will open the parent
directory with O_PATH and return SockaddrUn("/proc/self/fd/n/name").
"""
try:
return SockaddrUn(os.fsencode(path))
except PathTooLongError:
ppath = Path(path)
fd = await thr.task.open(await thr.ram.ptr(ppath.parent), O.PATH)
return SockaddrUnProcFd(fd, ppath.name)
T = t.TypeVar('T', bound='SockaddrUn')
@classmethod
def from_bytes(cls: t.Type[T], data: bytes) -> T:
header = ffi.sizeof('sa_family_t')
if len(data) < header:
raise Exception("data too smalllll", data)
struct = ffi.cast('struct sockaddr_un*', ffi.from_buffer(data))
cls.check_family(AF(struct.sun_family))
if len(data) == header:
# unnamed socket, name is empty
length = 0
elif struct.sun_path[0] == b'\0':
# abstract socket, entire buffer is part of path
length = len(data) - header
else:
# TODO handle the case where there's no null terminator
# pathname socket, path is null-terminated
length = lib.strlen(struct.sun_path)
return cls(bytes(ffi.buffer(struct.sun_path, length)))
def to_bytes(self) -> bytes:
addr = ffi.new('struct sockaddr_un*', (AF.UNIX, self.path))
real_length = ffi.sizeof('sa_family_t') + len(self.path) + 1
return bytes(ffi.buffer(addr))[:real_length]
@classmethod
def sizeof(cls) -> int:
return ffi.sizeof('struct sockaddr_un')
def __str__(self) -> str:
return f"SockaddrUn({self.path!r})"
async def close(self) -> None:
pass
_register_sockaddr(SockaddrUn)
class SockaddrUnProcFd(SockaddrUn):
def __init__(self, fd: FileDescriptor, name: str) -> None:
super().__init__(os.fsencode(f"/proc/self/fd/{int(fd)}/{name}"))
self.fd = fd
self.name = name
async def close(self) -> None:
await self.fd.close()
#### Tests ####
from unittest import TestCase
class TestUn(TestCase):
def test_sockaddrun(self) -> None:
initial = SockaddrUn(b"asefliasjeflsaifje0.1")
output = SockaddrUn.from_bytes(initial.to_bytes())
self.assertEqual(initial, output)
from rsyscall.sys.socket import SockaddrStorage
out = SockaddrStorage.from_bytes(initial.to_bytes()).parse()
self.assertEqual(initial, output)
Classes
class PathTooLongError (*args, **kwargs)
-
Inappropriate argument value (of correct type).
Expand source code Browse git
class PathTooLongError(ValueError): pass
Ancestors
- builtins.ValueError
- builtins.Exception
- builtins.BaseException
class SockaddrUn (path: bytes)
-
SockaddrUn(path: 'bytes')
Expand source code Browse git
@dataclass class SockaddrUn(Sockaddr): path: bytes family = AF.UNIX def __post_init__(self) -> None: if len(self.path) > 108: raise PathTooLongError("path", self.path, "is longer than the maximum unix address size") @staticmethod async def from_path(thr: Thread, path: t.Union[str, os.PathLike]) -> SockaddrUn: """Turn this path into a SockaddrUn, hacking around the 108 byte limit on socket addresses. If the passed path is too long to fit in an address, this function will open the parent directory with O_PATH and return SockaddrUn("/proc/self/fd/n/name"). """ try: return SockaddrUn(os.fsencode(path)) except PathTooLongError: ppath = Path(path) fd = await thr.task.open(await thr.ram.ptr(ppath.parent), O.PATH) return SockaddrUnProcFd(fd, ppath.name) T = t.TypeVar('T', bound='SockaddrUn') @classmethod def from_bytes(cls: t.Type[T], data: bytes) -> T: header = ffi.sizeof('sa_family_t') if len(data) < header: raise Exception("data too smalllll", data) struct = ffi.cast('struct sockaddr_un*', ffi.from_buffer(data)) cls.check_family(AF(struct.sun_family)) if len(data) == header: # unnamed socket, name is empty length = 0 elif struct.sun_path[0] == b'\0': # abstract socket, entire buffer is part of path length = len(data) - header else: # TODO handle the case where there's no null terminator # pathname socket, path is null-terminated length = lib.strlen(struct.sun_path) return cls(bytes(ffi.buffer(struct.sun_path, length))) def to_bytes(self) -> bytes: addr = ffi.new('struct sockaddr_un*', (AF.UNIX, self.path)) real_length = ffi.sizeof('sa_family_t') + len(self.path) + 1 return bytes(ffi.buffer(addr))[:real_length] @classmethod def sizeof(cls) -> int: return ffi.sizeof('struct sockaddr_un') def __str__(self) -> str: return f"SockaddrUn({self.path!r})" async def close(self) -> None: pass
Ancestors
Subclasses
- rsyscall.sys.un.SockaddrUnProcFd
Class variables
var path : bytes
var family
var T
Static methods
async def from_path(thr: Thread, path: t.Union[str, os.PathLike]) ‑> SockaddrUn
-
Turn this path into a SockaddrUn, hacking around the 108 byte limit on socket addresses.
If the passed path is too long to fit in an address, this function will open the parent directory with O_PATH and return SockaddrUn("/proc/self/fd/n/name").
Expand source code Browse git
@staticmethod async def from_path(thr: Thread, path: t.Union[str, os.PathLike]) -> SockaddrUn: """Turn this path into a SockaddrUn, hacking around the 108 byte limit on socket addresses. If the passed path is too long to fit in an address, this function will open the parent directory with O_PATH and return SockaddrUn("/proc/self/fd/n/name"). """ try: return SockaddrUn(os.fsencode(path)) except PathTooLongError: ppath = Path(path) fd = await thr.task.open(await thr.ram.ptr(ppath.parent), O.PATH) return SockaddrUnProcFd(fd, ppath.name)
Methods
async def close(self) ‑> NoneType
-
Expand source code Browse git
async def close(self) -> None: pass
Inherited members