Server

RemoteHost dataclass

Remote RADIUS capable host we can talk to.

Parameters:

Name	Type	Description	Default
address	str	IP address.	required
secret	bytes	RADIUS secret. If connecting to a RadSec server, the secret should be radsec.	required
name	str	Short name (used for logging only).	required
authport	int	Port used for authentication packets.	1812
acctport	int	Port used for accounting packets.	1813
coaport	int	Port used for CoA packets.	3799

Source code in pyrad2/server.py

@dataclass
class RemoteHost:
    """Remote RADIUS capable host we can talk to.

    Args:
        address (str): IP address.
        secret (bytes): RADIUS secret. If connecting to a RadSec server, the secret should be `radsec`.
        name (str): Short name (used for logging only).
        authport (int): Port used for authentication packets.
        acctport (int): Port used for accounting packets.
        coaport (int): Port used for CoA packets.
    """

    address: str
    secret: bytes
    name: str
    authport: int = 1812
    acctport: int = 1813
    coaport: int = 3799

Server

Bases: Host

Basic RADIUS server. This class implements the basics of a RADIUS server. It takes care of the details of receiving and decoding requests; processing of the requests should be done by overloading the appropriate methods in derived classes.

Attributes:

Name	Type	Description
hosts	dict	Hosts who are allowed to talk to us. Dictionary of Host class instances.
_poll	poll	Poll object for network sockets.
_fdmap	dict	Map of file descriptors to network sockets.
MaxPacketSize	int	Maximum size of a RADIUS packet. (class variable)

Source code in pyrad2/server.py

class Server(host.Host):
    """Basic RADIUS server.
    This class implements the basics of a RADIUS server. It takes care
    of the details of receiving and decoding requests; processing of
    the requests should be done by overloading the appropriate methods
    in derived classes.

    Attributes:
        hosts (dict): Hosts who are allowed to talk to us. Dictionary of Host class instances.
        _poll (select.poll): Poll object for network sockets.
        _fdmap (dict): Map of file descriptors to network sockets.
        MaxPacketSize (int): Maximum size of a RADIUS packet. (class variable)
    """

    MAX_PACKET_SIZE = 8192

    def __init__(
        self,
        addresses: Optional[list[str]] = None,
        authport: int = 1812,
        acctport: int = 1813,
        coaport: int = 3799,
        hosts: Optional[dict] = None,
        dict: Optional[Dictionary] = None,
        auth_enabled: bool = True,
        acct_enabled: bool = True,
        coa_enabled: bool = False,
        require_message_authenticator: bool = False,
        require_eap_message_authenticator: bool = True,
        dedup_enabled: bool = True,
        dedup_ttl: float = 30.0,
        dedup_max_entries: int = 4096,
        dedup_cache: Optional[dedup.ResponseCache] = None,
    ):
        """Initializes a sync server.

        Args:
            addresses (Sequence[str]): IP addresses to listen on.
            authport (int): Port to listen on for authentication packets.
            acctport (int): Port to listen on for accounting packets.
            coaport (int): Port to listen on for CoA packets.
            hosts (dict[str, RemoteHost]): Hosts who we can talk to. A dictionary mapping IP to RemoteHost class instances.
            dict (Dictionary): RADIUS dictionary to use.
            auth_enabled (bool): Enable auth server (default: True).
            acct_enabled (bool): Enable accounting server (default: True).
            coa_enabled (bool): Enable CoA server (default: False).
            require_message_authenticator (bool): Require
                Message-Authenticator on incoming packets.
            require_eap_message_authenticator (bool): Require
                Message-Authenticator on packets containing EAP-Message.
            dedup_enabled (bool): Enable RFC 5080 duplicate detection and
                response caching (default: True). Retransmissions of an
                Access-Request, Accounting-Request, CoA-Request, or
                Disconnect-Request will be answered by replaying the cached
                reply bytes instead of re-running the handler.
            dedup_ttl (float): Lifetime in seconds of a cached reply.
            dedup_max_entries (int): Maximum number of cached replies before
                LRU eviction kicks in.
            dedup_cache (ResponseCache): Provide a pre-built cache to share
                between servers or to inject a custom clock for tests.
        """
        super().__init__(authport, acctport, coaport, dict)

        self.hosts = hosts or {}
        self.auth_enabled = auth_enabled
        self.authfds: list[socket.socket] = []
        self.acct_enabled = acct_enabled
        self.acctfds: list = []
        self.coa_enabled = coa_enabled
        self.coafds: list = []
        self.require_message_authenticator = require_message_authenticator
        self.require_eap_message_authenticator = require_eap_message_authenticator
        if dedup_cache is not None:
            self._dedup_cache: Optional[dedup.ResponseCache] = dedup_cache
        elif dedup_enabled:
            self._dedup_cache = dedup.ResponseCache(
                ttl=dedup_ttl, max_entries=dedup_max_entries
            )
        else:
            self._dedup_cache = None

        if addresses:
            for addr in addresses:
                self.bind_to_address(addr)

    def _validate_message_authenticator_policy(self, pkt: packet.Packet) -> None:
        """Validate incoming Message-Authenticator policy for a packet."""
        if not isinstance(pkt, packet.Packet):
            return
        pkt.validate_message_authenticator_policy(
            require_message_authenticator=self.require_message_authenticator,
            require_eap_message_authenticator=self.require_eap_message_authenticator,
        )

    def _send_status_response(self, pkt: packet.Packet, code: PacketType) -> None:
        """Reply to Status-Server without invoking normal request handlers."""
        reply = self.create_reply_packet(pkt, code=code)
        if hasattr(pkt, "fd"):
            self.send_reply_packet(pkt.fd, reply)

    def _handle_status_packet(self, pkt: packet.Packet, code: PacketType) -> bool:
        """Handle Status-Server packets before normal request dispatch."""
        if pkt.code != PacketType.StatusServer:
            return False
        self._validate_message_authenticator_policy(pkt)
        logger.debug(
            "Received Status-Server from {}; replying with {}",
            getattr(pkt, "source", None),
            code.name,
        )
        self._send_status_response(pkt, code)
        return True

    def _get_addr_info(
        self, addr: str
    ) -> set[tuple[socket.AddressFamily, str | int]] | list:
        """Use getaddrinfo to lookup all addresses for each address.

        Returns a list of tuples or an empty list:
          [(family, address)]

        Args:
            adddr (str): IP address to lookup
        """
        results = set()
        try:
            tmp = socket.getaddrinfo(addr, 80)
        except socket.gaierror:
            return []

        for el in tmp:
            results.add((el[0], el[4][0]))

        return results

    def bind_to_address(self, addr: str) -> None:
        """Add an address to listen on a specific interface.
        String "0.0.0.0" indicates you want to listen on all interfaces.

        Args:
            addr (str): IP address to listen on
        """
        addr_family = self._get_addr_info(addr)
        for family, address in addr_family:
            if self.auth_enabled:
                authfd = socket.socket(family, socket.SOCK_DGRAM)
                authfd.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                authfd.bind((address, self.authport))
                self.authfds.append(authfd)

            if self.acct_enabled:
                acctfd = socket.socket(family, socket.SOCK_DGRAM)
                acctfd.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                acctfd.bind((address, self.acctport))
                self.acctfds.append(acctfd)

            if self.coa_enabled:
                coafd = socket.socket(family, socket.SOCK_DGRAM)
                coafd.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                coafd.bind((address, self.coaport))
                self.coafds.append(coafd)

    def handle_auth_packet(self, pkt: packet.Packet):
        """Authentication packet handler.
        This is an empty function that is called when a valid
        authentication packet has been received. It can be overriden in
        derived classes to add custom behaviour.

        Args:
            pkt (packet.Packet): Packet to process
        """

    def handle_acct_packet(self, pkt: packet.Packet):
        """Accounting packet handler.
        This is an empty function that is called when a valid
        accounting packet has been received. It can be overriden in
        derived classes to add custom behaviour.

        Args:
            pkt (packet.Packet): Packet to process
        """

    def handle_coa_packet(self, pkt: packet.Packet):
        """CoA packet handler.
        This is an empty function that is called when a valid
        accounting packet has been received. It can be overriden in
        derived classes to add custom behaviour.

        Args:
            pkt (packet.Packet): Packet to process
        """

    def handle_disconnect_packet(self, pkt: packet.Packet):
        """CoA packet handler.
        This is an empty function that is called when a valid
        accounting packet has been received. It can be overriden in
        derived classes to add custom behaviour.

        Args:
            pkt (packet.Packet): Packet to process
        """

    def _dedup_dispatch(
        self, pkt: packet.Packet, handler: Callable[[packet.Packet], None]
    ) -> None:
        """Wrap ``handler(pkt)`` with RFC 5080 dedup.

        On a cached duplicate, replays the stored reply bytes via the
        request's socket and skips the handler entirely. On an in-flight
        duplicate, drops silently. Otherwise runs the handler and lets
        ``send_reply_packet`` populate the cache.
        """
        key = dedup.key_for(pkt) if self._dedup_cache is not None else None
        fd = getattr(pkt, "fd", None)

        def _resend(raw: bytes) -> None:
            if fd is not None:
                fd.sendto(raw, pkt.source)

        action = dedup.consult_cache(self._dedup_cache, key, _resend)
        if action is dedup.DispatchAction.DROP:
            logger.debug(
                "Dropping duplicate in-flight request from {}", pkt.source
            )
            return
        if action is dedup.DispatchAction.RESENT:
            logger.debug(
                "Resent cached reply for duplicate request from {}", pkt.source
            )
            return

        if key is not None:
            pkt._dedup_key = key  # type: ignore[attr-defined]
        try:
            handler(pkt)
        finally:
            if key is not None and self._dedup_cache is not None:
                # No-op if the handler already recorded a reply.
                self._dedup_cache.drop_in_flight(key)

    def _add_secret(self, pkt: packet.Packet) -> None:
        """Add secret to packets received and raise ServerPacketError
        for unknown hosts.

        Args:
            pkt (packet.Packet): Packet to process
        """
        if pkt.source[0] in self.hosts:
            pkt.secret = self.hosts[pkt.source[0]].secret
        elif "0.0.0.0" in self.hosts:
            pkt.secret = self.hosts["0.0.0.0"].secret
        else:
            raise ServerPacketError("Received packet from unknown host")

    def _handle_auth_packet(self, pkt: packet.Packet) -> None:
        """Process a packet received on the authentication port.
        If this packet should be dropped instead of processed a
        ServerPacketError exception should be raised. The main loop will
        drop the packet and log the reason.

        Args:
            pkt (packet.Packet): Packet to process
        """
        self._add_secret(pkt)
        if self._handle_status_packet(pkt, PacketType.AccessAccept):
            return
        if pkt.code != PacketType.AccessRequest:
            raise ServerPacketError(
                "Received non-authentication packet on authentication port"
            )
        self._validate_message_authenticator_policy(pkt)
        self._dedup_dispatch(pkt, self.handle_auth_packet)

    def _handle_acct_packet(self, pkt: packet.Packet) -> None:
        """Process a packet received on the accounting port.
        If this packet should be dropped instead of processed a
        ServerPacketError exception should be raised. The main loop will
        drop the packet and log the reason.

        Args:
            pkt (packet.Packet): Packet to process
        """
        self._add_secret(pkt)
        if self._handle_status_packet(pkt, PacketType.AccountingResponse):
            return
        if pkt.code not in [
            PacketType.AccountingRequest,
            PacketType.AccountingResponse,
        ]:
            raise ServerPacketError("Received non-accounting packet on accounting port")
        self._validate_message_authenticator_policy(pkt)
        self._dedup_dispatch(pkt, self.handle_acct_packet)

    def _handle_coa_packet(self, pkt: packet.Packet) -> None:
        """Process a packet received on the coa port.
        If this packet should be dropped instead of processed a
        ServerPacketError exception should be raised. The main loop will
        drop the packet and log the reason.

        Args:
            pkt (packet.Packet): Packet to process
        """
        self._add_secret(pkt)
        if pkt.code == PacketType.CoARequest:
            self._validate_message_authenticator_policy(pkt)
            self._dedup_dispatch(pkt, self.handle_coa_packet)
        elif pkt.code == PacketType.DisconnectRequest:
            self._validate_message_authenticator_policy(pkt)
            self._dedup_dispatch(pkt, self.handle_disconnect_packet)
        else:
            raise ServerPacketError("Received non-coa packet on coa port")

    def _grab_packet(self, pktgen: Callable, fd: socket.socket) -> packet.Packet:
        """Read a packet from a network connection.
        This method assumes there is data waiting for to be read.

        Args:
            fd (socket.socket): Socket to read packet from

        Returns:
            packet.Packet: RADIUS packet
        """
        (data, source) = fd.recvfrom(self.MAX_PACKET_SIZE)
        pkt = pktgen(data)
        pkt.source = source
        pkt.fd = fd
        return pkt

    def _prepare_sockets(self) -> None:
        """Prepare all sockets to receive packets."""
        for fd in self.authfds + self.acctfds + self.coafds:
            self._fdmap[fd.fileno()] = fd
            if os.name == "nt":
                self._sel.register(fd.fileno(), selectors.EVENT_READ)
            else:
                self._poll.register(
                    fd.fileno(), select.POLLIN | select.POLLPRI | select.POLLERR
                )
        if self.auth_enabled:
            self._realauthfds = list(map(lambda x: x.fileno(), self.authfds))
        if self.acct_enabled:
            self._realacctfds = list(map(lambda x: x.fileno(), self.acctfds))
        if self.coa_enabled:
            self._realcoafds = list(map(lambda x: x.fileno(), self.coafds))

    def create_reply_packet(self, pkt: packet.Packet, **attributes) -> packet.Packet:
        """Create a reply packet.
        Create a new packet which can be returned as a reply to a received
        packet.

        Args:
            pkt (packet.Packet): Packet to process
        """
        reply = pkt.create_reply(**attributes)
        reply.source = pkt.source
        packet.prepare_reply_message_authenticator(
            pkt,
            reply,
            require_message_authenticator=self.require_message_authenticator,
            require_eap_message_authenticator=self.require_eap_message_authenticator,
        )
        # Carry the request's dedup key forward so send_reply_packet can
        # cache the resulting bytes without re-deriving the key.
        request_key = getattr(pkt, "_dedup_key", None)
        if request_key is not None:
            reply._dedup_key = request_key  # type: ignore[attr-defined]
        return reply

    def send_reply_packet(self, fd: socket.socket, pkt: packet.Packet) -> None:
        """Send a reply packet after applying Message-Authenticator policy."""
        if self.require_message_authenticator or (
            self.require_eap_message_authenticator
            and isinstance(pkt, packet.Packet)
            and pkt.has_eap_message()
        ):
            pkt.ensure_message_authenticator()
        # Encode once: we need the exact bytes for RFC 5080 replay so a
        # retransmission gets a byte-identical answer (which matters for
        # the EAP State attribute and the Message-Authenticator).
        raw = pkt.reply_packet()
        fd.sendto(raw, pkt.source)  # type: ignore[call-overload]
        dedup.record_if_keyed(self._dedup_cache, pkt, raw)

    def _process_input(self, fd: socket.socket) -> None:
        """Process available data.
        If this packet should be dropped instead of processed a
        PacketError exception should be raised. The main loop will
        drop the packet and log the reason.

        This function calls either handle_auth_packet() or
        handle_acct_packet() depending on which socket is being
        processed.

        Args:
            fd (socket.socket): Socket to read the packet from
        """
        if self.auth_enabled and fd.fileno() in self._realauthfds:
            pkt = self._grab_packet(
                lambda data, s=self: packet.parse_packet(data, b"", s.dict), fd
            )
            self._handle_auth_packet(pkt)
        elif self.acct_enabled and fd.fileno() in self._realacctfds:
            pkt = self._grab_packet(
                lambda data, s=self: packet.parse_packet(data, b"", s.dict), fd
            )
            self._handle_acct_packet(pkt)
        elif self.coa_enabled:
            pkt = self._grab_packet(
                lambda data, s=self: packet.parse_packet(data, b"", s.dict), fd
            )
            self._handle_coa_packet(pkt)
        else:
            raise ServerPacketError("Received packet for unknown handler")

    def run(self) -> None:
        """Main loop.
        This method is the main loop for a RADIUS server. It waits
        for packets to arrive via the network and calls other methods
        to process them.
        """
        if os.name == "nt":
            self._sel = selectors.DefaultSelector()
        else:
            self._poll = select.poll()
        self._fdmap: dict[int, socket.socket] = {}
        self._prepare_sockets()

        while True:
            if os.name == "nt":
                for key, mask in self._sel.select(timeout=1):
                    if mask & selectors.EVENT_READ:
                        try:
                            fdo = self._fdmap[key.fd]
                            self._process_input(fdo)
                        except ServerPacketError as err:
                            logger.info("Dropping packet: " + str(err))
                        except packet.PacketError as err:
                            logger.info("Received a broken packet: " + str(err))
                    else:
                        logger.error("Unexpected event in server main loop")
            else:
                for fd, event in self._poll.poll():
                    if event == select.POLLIN:
                        try:
                            fdo = self._fdmap[fd]
                            self._process_input(fdo)
                        except ServerPacketError as err:
                            logger.info("Dropping packet: " + str(err))
                        except packet.PacketError as err:
                            logger.info("Received a broken packet: " + str(err))
                    else:
                        logger.error("Unexpected event in server main loop")

__init__(addresses=None, authport=1812, acctport=1813, coaport=3799, hosts=None, dict=None, auth_enabled=True, acct_enabled=True, coa_enabled=False, require_message_authenticator=False, require_eap_message_authenticator=True, dedup_enabled=True, dedup_ttl=30.0, dedup_max_entries=4096, dedup_cache=None)

Initializes a sync server.

Parameters:

Name	Type	Description	Default
addresses	Sequence[str]	IP addresses to listen on.	None
authport	int	Port to listen on for authentication packets.	1812
acctport	int	Port to listen on for accounting packets.	1813
coaport	int	Port to listen on for CoA packets.	3799
hosts	dict[str, RemoteHost]	Hosts who we can talk to. A dictionary mapping IP to RemoteHost class instances.	None
dict	Dictionary	RADIUS dictionary to use.	None
auth_enabled	bool	Enable auth server (default: True).	True
acct_enabled	bool	Enable accounting server (default: True).	True
coa_enabled	bool	Enable CoA server (default: False).	False
require_message_authenticator	bool	Require Message-Authenticator on incoming packets.	False
require_eap_message_authenticator	bool	Require Message-Authenticator on packets containing EAP-Message.	True
dedup_enabled	bool	Enable RFC 5080 duplicate detection and response caching (default: True). Retransmissions of an Access-Request, Accounting-Request, CoA-Request, or Disconnect-Request will be answered by replaying the cached reply bytes instead of re-running the handler.	True
dedup_ttl	float	Lifetime in seconds of a cached reply.	30.0
dedup_max_entries	int	Maximum number of cached replies before LRU eviction kicks in.	4096
dedup_cache	ResponseCache	Provide a pre-built cache to share between servers or to inject a custom clock for tests.	None

Source code in pyrad2/server.py

def __init__(
    self,
    addresses: Optional[list[str]] = None,
    authport: int = 1812,
    acctport: int = 1813,
    coaport: int = 3799,
    hosts: Optional[dict] = None,
    dict: Optional[Dictionary] = None,
    auth_enabled: bool = True,
    acct_enabled: bool = True,
    coa_enabled: bool = False,
    require_message_authenticator: bool = False,
    require_eap_message_authenticator: bool = True,
    dedup_enabled: bool = True,
    dedup_ttl: float = 30.0,
    dedup_max_entries: int = 4096,
    dedup_cache: Optional[dedup.ResponseCache] = None,
):
    """Initializes a sync server.

    Args:
        addresses (Sequence[str]): IP addresses to listen on.
        authport (int): Port to listen on for authentication packets.
        acctport (int): Port to listen on for accounting packets.
        coaport (int): Port to listen on for CoA packets.
        hosts (dict[str, RemoteHost]): Hosts who we can talk to. A dictionary mapping IP to RemoteHost class instances.
        dict (Dictionary): RADIUS dictionary to use.
        auth_enabled (bool): Enable auth server (default: True).
        acct_enabled (bool): Enable accounting server (default: True).
        coa_enabled (bool): Enable CoA server (default: False).
        require_message_authenticator (bool): Require
            Message-Authenticator on incoming packets.
        require_eap_message_authenticator (bool): Require
            Message-Authenticator on packets containing EAP-Message.
        dedup_enabled (bool): Enable RFC 5080 duplicate detection and
            response caching (default: True). Retransmissions of an
            Access-Request, Accounting-Request, CoA-Request, or
            Disconnect-Request will be answered by replaying the cached
            reply bytes instead of re-running the handler.
        dedup_ttl (float): Lifetime in seconds of a cached reply.
        dedup_max_entries (int): Maximum number of cached replies before
            LRU eviction kicks in.
        dedup_cache (ResponseCache): Provide a pre-built cache to share
            between servers or to inject a custom clock for tests.
    """
    super().__init__(authport, acctport, coaport, dict)

    self.hosts = hosts or {}
    self.auth_enabled = auth_enabled
    self.authfds: list[socket.socket] = []
    self.acct_enabled = acct_enabled
    self.acctfds: list = []
    self.coa_enabled = coa_enabled
    self.coafds: list = []
    self.require_message_authenticator = require_message_authenticator
    self.require_eap_message_authenticator = require_eap_message_authenticator
    if dedup_cache is not None:
        self._dedup_cache: Optional[dedup.ResponseCache] = dedup_cache
    elif dedup_enabled:
        self._dedup_cache = dedup.ResponseCache(
            ttl=dedup_ttl, max_entries=dedup_max_entries
        )
    else:
        self._dedup_cache = None

    if addresses:
        for addr in addresses:
            self.bind_to_address(addr)

bind_to_address(addr)

Add an address to listen on a specific interface. String "0.0.0.0" indicates you want to listen on all interfaces.

Parameters:

Name	Type	Description	Default
addr	str	IP address to listen on	required

Source code in pyrad2/server.py

def bind_to_address(self, addr: str) -> None:
    """Add an address to listen on a specific interface.
    String "0.0.0.0" indicates you want to listen on all interfaces.

    Args:
        addr (str): IP address to listen on
    """
    addr_family = self._get_addr_info(addr)
    for family, address in addr_family:
        if self.auth_enabled:
            authfd = socket.socket(family, socket.SOCK_DGRAM)
            authfd.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            authfd.bind((address, self.authport))
            self.authfds.append(authfd)

        if self.acct_enabled:
            acctfd = socket.socket(family, socket.SOCK_DGRAM)
            acctfd.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            acctfd.bind((address, self.acctport))
            self.acctfds.append(acctfd)

        if self.coa_enabled:
            coafd = socket.socket(family, socket.SOCK_DGRAM)
            coafd.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            coafd.bind((address, self.coaport))
            self.coafds.append(coafd)

handle_auth_packet(pkt)

Authentication packet handler. This is an empty function that is called when a valid authentication packet has been received. It can be overriden in derived classes to add custom behaviour.

Parameters:

Name	Type	Description	Default
pkt	Packet	Packet to process	required

Source code in pyrad2/server.py

def handle_auth_packet(self, pkt: packet.Packet):
    """Authentication packet handler.
    This is an empty function that is called when a valid
    authentication packet has been received. It can be overriden in
    derived classes to add custom behaviour.

    Args:
        pkt (packet.Packet): Packet to process
    """

handle_acct_packet(pkt)

Accounting packet handler. This is an empty function that is called when a valid accounting packet has been received. It can be overriden in derived classes to add custom behaviour.

Parameters:

Name	Type	Description	Default
pkt	Packet	Packet to process	required

Source code in pyrad2/server.py

def handle_acct_packet(self, pkt: packet.Packet):
    """Accounting packet handler.
    This is an empty function that is called when a valid
    accounting packet has been received. It can be overriden in
    derived classes to add custom behaviour.

    Args:
        pkt (packet.Packet): Packet to process
    """

handle_coa_packet(pkt)

CoA packet handler. This is an empty function that is called when a valid accounting packet has been received. It can be overriden in derived classes to add custom behaviour.

Parameters:

Name	Type	Description	Default
pkt	Packet	Packet to process	required

Source code in pyrad2/server.py

def handle_coa_packet(self, pkt: packet.Packet):
    """CoA packet handler.
    This is an empty function that is called when a valid
    accounting packet has been received. It can be overriden in
    derived classes to add custom behaviour.

    Args:
        pkt (packet.Packet): Packet to process
    """

handle_disconnect_packet(pkt)

CoA packet handler. This is an empty function that is called when a valid accounting packet has been received. It can be overriden in derived classes to add custom behaviour.

Parameters:

Name	Type	Description	Default
pkt	Packet	Packet to process	required

Source code in pyrad2/server.py

def handle_disconnect_packet(self, pkt: packet.Packet):
    """CoA packet handler.
    This is an empty function that is called when a valid
    accounting packet has been received. It can be overriden in
    derived classes to add custom behaviour.

    Args:
        pkt (packet.Packet): Packet to process
    """

create_reply_packet(pkt, **attributes)

Create a reply packet. Create a new packet which can be returned as a reply to a received packet.

Parameters:

Name	Type	Description	Default
pkt	Packet	Packet to process	required

Source code in pyrad2/server.py

def create_reply_packet(self, pkt: packet.Packet, **attributes) -> packet.Packet:
    """Create a reply packet.
    Create a new packet which can be returned as a reply to a received
    packet.

    Args:
        pkt (packet.Packet): Packet to process
    """
    reply = pkt.create_reply(**attributes)
    reply.source = pkt.source
    packet.prepare_reply_message_authenticator(
        pkt,
        reply,
        require_message_authenticator=self.require_message_authenticator,
        require_eap_message_authenticator=self.require_eap_message_authenticator,
    )
    # Carry the request's dedup key forward so send_reply_packet can
    # cache the resulting bytes without re-deriving the key.
    request_key = getattr(pkt, "_dedup_key", None)
    if request_key is not None:
        reply._dedup_key = request_key  # type: ignore[attr-defined]
    return reply

send_reply_packet(fd, pkt)

Send a reply packet after applying Message-Authenticator policy.

Source code in pyrad2/server.py

def send_reply_packet(self, fd: socket.socket, pkt: packet.Packet) -> None:
    """Send a reply packet after applying Message-Authenticator policy."""
    if self.require_message_authenticator or (
        self.require_eap_message_authenticator
        and isinstance(pkt, packet.Packet)
        and pkt.has_eap_message()
    ):
        pkt.ensure_message_authenticator()
    # Encode once: we need the exact bytes for RFC 5080 replay so a
    # retransmission gets a byte-identical answer (which matters for
    # the EAP State attribute and the Message-Authenticator).
    raw = pkt.reply_packet()
    fd.sendto(raw, pkt.source)  # type: ignore[call-overload]
    dedup.record_if_keyed(self._dedup_cache, pkt, raw)

run()

Main loop. This method is the main loop for a RADIUS server. It waits for packets to arrive via the network and calls other methods to process them.

Source code in pyrad2/server.py

def run(self) -> None:
    """Main loop.
    This method is the main loop for a RADIUS server. It waits
    for packets to arrive via the network and calls other methods
    to process them.
    """
    if os.name == "nt":
        self._sel = selectors.DefaultSelector()
    else:
        self._poll = select.poll()
    self._fdmap: dict[int, socket.socket] = {}
    self._prepare_sockets()

    while True:
        if os.name == "nt":
            for key, mask in self._sel.select(timeout=1):
                if mask & selectors.EVENT_READ:
                    try:
                        fdo = self._fdmap[key.fd]
                        self._process_input(fdo)
                    except ServerPacketError as err:
                        logger.info("Dropping packet: " + str(err))
                    except packet.PacketError as err:
                        logger.info("Received a broken packet: " + str(err))
                else:
                    logger.error("Unexpected event in server main loop")
        else:
            for fd, event in self._poll.poll():
                if event == select.POLLIN:
                    try:
                        fdo = self._fdmap[fd]
                        self._process_input(fdo)
                    except ServerPacketError as err:
                        logger.info("Dropping packet: " + str(err))
                    except packet.PacketError as err:
                        logger.info("Received a broken packet: " + str(err))
                else:
                    logger.error("Unexpected event in server main loop")