| Server IP : 91.108.119.194 / Your IP : 216.73.216.122 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/dhcp/ |
Upload File : |
'''
IPv4 DHCP socket
================
'''
import asyncio
import logging
import socket
from typing import Optional
from pyroute2.compat import ETHERTYPE_IP
from pyroute2.dhcp.dhcp4msg import dhcp4msg
from pyroute2.dhcp.messages import ReceivedDHCPMessage, SentDHCPMessage
from pyroute2.ext.bpf import BPF
from pyroute2.ext.rawsocket import AsyncRawSocket
from pyroute2.protocols import ethmsg, ip4msg, udp4_pseudo_header, udpmsg
LOG = logging.getLogger(__name__)
UDP_HEADER_SIZE = 8
IPV4_HEADER_SIZE = 20
SKF_AD_OFF = -0x1000
SKF_AD_VLAN_TAG_PRESENT = 48
def listen_udp_port(port: int = 68) -> list[list[int]]:
'''BPF filter that matches Ethernet + IPv4 + UDP on the given port.
Packets tagged on a vlan are also dropped, see
https://lore.kernel.org/netdev/51FB6A9D.2050002@redhat.com/T/
'''
bpf_code = [
# Load vlan presence indicator
[BPF.LD + BPF.B + BPF.ABS, 0, 0, SKF_AD_OFF + SKF_AD_VLAN_TAG_PRESENT],
# bail out immediately if there is one and we don't want it
[BPF.JMP + BPF.JEQ + BPF.K, 0, 10, 0],
# Load eth type at offset 12 and check it's IPv4
[BPF.LD + BPF.H + BPF.ABS, 0, 0, 12],
[BPF.JMP + BPF.JEQ + BPF.K, 0, 8, 0x0800],
# Load IP proto at offset 23 and check it's UDP
[BPF.LD + BPF.B + BPF.ABS, 0, 0, 23],
[BPF.JMP + BPF.JEQ + BPF.K, 0, 6, socket.IPPROTO_UDP],
# load frag offset at offset 20
[BPF.LD + BPF.H + BPF.ABS, 0, 0, 20],
# Check mask & drop fragmented packets
[BPF.JMP + BPF.JSET + BPF.K, 4, 0, 8191],
# load ip header length at offset 14
[BPF.LDX + BPF.B + BPF.MSH, 0, 0, 14],
# load udp dport from that offset + 16 and check it
[BPF.LD + BPF.H + BPF.IND, 0, 0, 16],
[BPF.JMP + BPF.JEQ + BPF.K, 0, 1, port],
# allow packet
[BPF.RET + BPF.K, 0, 0, 65535],
# drop packet
[BPF.RET + BPF.K, 0, 0, 0],
]
return bpf_code
class AsyncDHCP4Socket(AsyncRawSocket):
'''
Parameters:
* ifname -- interface name to work on
* port -- UDP port to listen on
This raw socket binds to an interface and installs a BPF filter
to receive (non-VLAN) messages only on the specified UDP port.
It can be used in poll/select and implements the async context manager
protocol, so can be used in `aysnc with` statements.
'''
def __init__(self, ifname: str, port: int = 68):
AsyncRawSocket.__init__(self, ifname, listen_udp_port(port))
self.port = port
self._loop = asyncio.get_running_loop()
@property
def loop(self) -> asyncio.AbstractEventLoop:
# We define this as a property because it's easier to patch in tests
return self._loop
async def put(self, msg: SentDHCPMessage) -> SentDHCPMessage:
'''Send a DHCP message.
This encapsulates the `SentDHCPMessage` into Ethernet, IPv4, and UDP.
If not provided, both the Ethernet source and the DHCP chaddr are set
to the MAC address of the underlying socket's interface.
Example::
msg = SentDHCPMessage(
dhcp=dhcp4msg({
'op': bootp.MessageType.BOOTREQUEST,
'flags': bootp.Flag.BROADCAST,
'options': {
'message_type': dhcp.MessageType.DISCOVER,
'parameter_list': [
dhcp.Option.SUBNET_MASK,
dhcp.Option.ROUTER,
dhcp.Option.NAME_SERVER,
dhcp.Option.LEASE_TIME,
],
},
}),
)
await sock.put(msg)
'''
if msg.sport != self.port:
raise ValueError(
f"Client source port is set to {self.port}, "
f"cannot send message from port {msg.sport}."
)
if not msg.eth_src:
msg.eth_src = self.l2addr
# DHCP layer
dhcp = msg.dhcp
# auto add src addr
if dhcp['chaddr'] is None:
dhcp['chaddr'] = msg.eth_src
data = dhcp.encode().buf
dhcp_payload_size = len(data)
# UDP layer
udp = udpmsg(
{
'sport': self.port,
'dport': msg.dport,
'len': UDP_HEADER_SIZE + dhcp_payload_size,
}
)
# Pseudo UDP header, only for checksum purposes
udph = udp4_pseudo_header(
{
'src': msg.ip_src,
'dst': msg.ip_dst,
'len': UDP_HEADER_SIZE + dhcp_payload_size,
}
)
udp['csum'] = self.csum(udph.encode().buf + udp.encode().buf + data)
udp.reset()
# IPv4 layer
ip4 = ip4msg(
{
'len': IPV4_HEADER_SIZE + UDP_HEADER_SIZE + dhcp_payload_size,
'proto': socket.IPPROTO_UDP,
'dst': msg.ip_dst,
'src': msg.ip_src,
}
)
ip4['csum'] = self.csum(ip4.encode().buf)
ip4.reset()
# MAC layer
eth = ethmsg(
{'dst': msg.eth_dst, 'src': msg.eth_src, 'type': ETHERTYPE_IP}
)
data = eth.encode().buf + ip4.encode().buf + udp.encode().buf + data
await self.loop.sock_sendall(self, data)
dhcp.reset()
return msg
async def get(self) -> ReceivedDHCPMessage:
'''
Get the next incoming packet from the socket and try
to decode it as IPv4 DHCP.
Packets that cannot be decoded are logged & discarded.
Invalid/truncated packets raise `ValueError`.
Example::
# Send a DISCOVER and read an OFFER
disco = messages.discover(parameter_list=[
dhcp.Option.SUBNET_MASK,
dhcp.Option.ROUTER,
dhcp.Option.NAME_SERVER,
dhcp.Option.LEASE_TIME,
])
await sock.put(disco)
offer = await sock.get()
print('received', offer.message_type,
'from', offer.eth_src,
'lease time', offer.dhcp['options']['lease_time'],
)
'''
msg: Optional[ReceivedDHCPMessage] = None
while not msg:
raw = await self.loop.sock_recv(self, 4096)
try:
msg = self._decode_msg(raw)
except ValueError as err:
LOG.error('%s', err)
return msg
@classmethod
def _decode_msg(cls, data: bytes) -> ReceivedDHCPMessage:
'''Decode an IPv4 DHCP packet from bytes.
No analysis is done here. The MAC/IPv4/UDP headers are stripped out,
the relevant values are stored in the `ReceivedDHCPMessage` metadata,
and the rest is interpreted as DHCP.
'''
eth = ethmsg(buf=data).decode()
ip4 = ip4msg(buf=data, offset=eth.offset).decode()
udp = udpmsg(buf=data, offset=ip4.offset).decode()
dhcp = dhcp4msg(buf=data, offset=udp.offset).decode()
return ReceivedDHCPMessage(
dhcp=dhcp,
eth_src=eth['src'],
eth_dst=eth['dst'],
ip_src=ip4['src'],
ip_dst=ip4['dst'],
sport=udp['sport'],
dport=udp['dport'],
)