| Server IP : 84.32.84.3 / Your IP : 216.73.217.1 Web Server : LiteSpeed System : Linux id-dci-web1986.main-hosting.eu 5.14.0-611.26.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Thu Jan 29 05:24:47 EST 2026 x86_64 User : u686484674 ( 686484674) PHP Version : 8.0.30 Disable Function : system, exec, shell_exec, passthru, mysql_list_dbs, ini_alter, dl, symlink, link, chgrp, leak, popen, apache_child_terminate, virtual, mb_send_mail MySQL : OFF | cURL : ON | WGET : ON | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /proc/self/root/opt/alt/python311/lib/python3.11/site-packages/pyroute2/netlink/uevent/ |
Upload File : |
import struct
from typing import Callable, Generator, Optional
from pyroute2.netlink import NETLINK_KOBJECT_UEVENT, nlmsg
from pyroute2.netlink.nlsocket import (
AsyncNetlinkSocket,
Marshal,
NetlinkSocket,
)
MONITOR_GROUP_NONE = 0
MONITOR_GROUP_KERNEL = 1
MONITOR_GROUP_UDEV = 2
class ueventmsg(nlmsg):
pass
class MarshalUevent(Marshal):
def is_enough(self, _) -> bool:
return False
def parse(
self,
data: bytes,
seq: Optional[int] = None,
callback: Optional[Callable] = None,
skip_alien_seq: bool = False,
) -> Generator[ueventmsg, None, None]:
ret = ueventmsg()
ret['header']['sequence_number'] = 0
# handle udevd messages which have a binary header
if data.startswith(b'libudev\x00'):
# prefix, magic, header_size, properties_off, properties_len,
# filter_subsystem_hash, filter_tag_bloom_hi,
# filter_tag_bloom_low
header_format = '8sIIIIIIII'
header_length = struct.calcsize(header_format)
if len(data) < header_length:
return
lines = data[header_length:].split(b'\x00')
ret['header']['message'] = "libudev"
else:
lines = data.split(b'\x00')
ret['header']['message'] = lines[0].decode('utf-8')
wtf = []
ret['header']['unparsed'] = b''
for line in lines[1:]:
if line.find(b'=') <= 0:
wtf.append(line)
else:
if wtf:
ret['header']['unparsed'] = b'\x00'.join(wtf)
wtf = []
try:
key, value = line.decode('utf-8').split('=', 1)
except UnicodeDecodeError:
# should not happen but let's account for it for
# robustness' sake
wtf.append(line)
continue
ret[key] = value
del ret['value']
yield ret
class AsyncUeventSocket(AsyncNetlinkSocket):
def __init__(self):
super().__init__(NETLINK_KOBJECT_UEVENT)
self.set_marshal(MarshalUevent())
async def bind(self, groups: int = -1) -> None: # type: ignore[override]
return await super().bind(groups=groups)
class UeventSocket(NetlinkSocket):
def __init__(self):
super().__init__(NETLINK_KOBJECT_UEVENT)
self.set_marshal(MarshalUevent())
def bind(self, groups: int = -1) -> None:
return super().bind(groups=groups)