Client Async

DatagramProtocolClient

Bases: Protocol

Source code in pyrad2/client_async.py

class DatagramProtocolClient(asyncio.Protocol):
    def __init__(
        self,
        server: str,
        port: int,
        client: "ClientAsync",
        retries: int = 3,
        timeout: int = 30,
    ):
        self.port = port
        self.server = server
        self.retries = retries
        self.timeout = timeout
        self.client = client

        # Map of pending requests
        self.pending_requests: dict[int, dict] = {}

        # Use cryptographic-safe random generator as provided by the OS.
        random_generator = random.SystemRandom()
        self.packet_id = random_generator.randrange(0, 256)

        self.timeout_future = None

    async def __timeout_handler__(self):
        """Background task that retries or fails timed-out pending requests.

        Runs once per `next_wake_up` seconds. For each pending request we
        compare elapsed time against `self.timeout`; if elapsed has expired,
        we either resend the packet (consuming one retry) or surface a
        TimeoutError on the request's future. `next_wake_up` is the minimum
        remaining-time-to-timeout across all pending requests, so the loop
        wakes exactly when the next request needs servicing.
        """
        try:
            while True:
                req2delete = []
                now = datetime.now()
                next_wake_up = float(self.timeout)

                for id, req in self.pending_requests.items():
                    # send_date is always <= now, so compute elapsed as a
                    # positive float. Using timedelta.seconds here would
                    # wrap negative deltas to ~86399 and prematurely
                    # trigger the timeout branch.
                    elapsed = (now - req["send_date"]).total_seconds()
                    if elapsed >= self.timeout:
                        if req["retries"] >= self.retries:
                            logger.debug(
                                "[{}:{}] For request {} execute all retries",
                                self.server,
                                self.port,
                                id,
                            )
                            req["future"].set_exception(
                                TimeoutError("Timeout on Reply")
                            )
                            req2delete.append(id)
                        else:
                            # Send again packet
                            req["send_date"] = now
                            req["retries"] += 1
                            logger.debug(
                                "[{}:{}] For request {} execute retry {}",
                                self.server,
                                self.port,
                                id,
                                req["retries"],
                            )
                            self.transport.sendto(req["packet"].request_packet())
                    else:
                        remaining = self.timeout - elapsed
                        if remaining < next_wake_up:
                            next_wake_up = remaining

                for id in req2delete:
                    # Remove request for map
                    del self.pending_requests[id]

                # Floor sleeps at 0 so a just-expired request gets serviced
                # on the next loop iteration instead of busy-spinning.
                await asyncio.sleep(max(0.0, next_wake_up))

        except asyncio.CancelledError:
            pass

    def send_packet(self, packet: PacketImplementation, future: asyncio.Future):
        if packet.id in self.pending_requests:
            raise Exception("Packet with id %d already present" % packet.id)

        # Store packet on pending requests map
        self.pending_requests[packet.id] = {
            "packet": packet,
            "creation_date": datetime.now(),
            "retries": 0,
            "future": future,
            "send_date": datetime.now(),
        }

        # In queue packet raw on socket buffer
        self.transport.sendto(packet.request_packet())

    def connection_made(self, transport: asyncio.BaseTransport):
        assert isinstance(transport, asyncio.DatagramTransport), (
            "Expected DatagramTransport"
        )
        self.transport: asyncio.DatagramTransport = transport

        socket = transport.get_extra_info("socket")
        logger.info(
            "[{}:{}] Transport created with binding in {}:{}",
            self.server,
            self.port,
            socket.getsockname()[0],
            socket.getsockname()[1],
        )

        # loop = asyncio.get_event_loop()
        # asyncio.set_event_loop(loop=asyncio.get_event_loop())
        # Start asynchronous timer handler
        self.timeout_future = asyncio.ensure_future(self.__timeout_handler__())
        # asyncio.set_event_loop(loop=pre_loop)

    def error_received(self, exc: Exception) -> None:
        logger.error("[{}:{}] Error received: {}", self.server, self.port, exc)

    def connection_lost(self, exc) -> None:
        if exc:
            logger.warning(
                "[{}:{}] Connection lost: {}", self.server, self.port, str(exc)
            )
        else:
            logger.info("[{}:{}] Transport closed", self.server, self.port)

    def datagram_received(self, data: bytes, addr: str):
        try:
            reply = Packet(packet=data, dict=self.client.dict)

            if reply.code and reply.id in self.pending_requests:
                req = self.pending_requests[reply.id]
                packet = req["packet"]

                reply.dict = packet.dict
                reply.secret = packet.secret

                if packet.verify_reply(reply, data, enforce_ma=self.client.enforce_ma):
                    req["future"].set_result(reply)
                    # Remove request for map
                    del self.pending_requests[reply.id]
                else:
                    logger.warning(
                        "[{}:{}] Ignore invalid reply for id {}: {}",
                        self.server,
                        self.port,
                        reply.id,
                        data,
                    )
            else:
                logger.warning(
                    "[{}:{}] Ignore invalid reply: {}", self.server, self.port, data
                )

        except Exception as exc:
            logger.error(
                "[{}:{}] Error on decode packet: {}", self.server, self.port, exc
            )

    async def close_transport(self) -> None:
        if self.transport:
            logger.debug("[{}:{}] Closing transport...", self.server, self.port)
            self.transport.close()
            self.transport = None  # type: ignore
        if self.timeout_future:
            self.timeout_future.cancel()
            await self.timeout_future
            self.timeout_future = None

    def create_id(self) -> int:
        self.packet_id = (self.packet_id + 1) % 256
        return self.packet_id

    def __str__(self) -> str:
        return "DatagramProtocolClient(server?=%s, port=%d)" % (self.server, self.port)

    # Used as protocol_factory
    def __call__(self):
        return self

__timeout_handler__() async

Background task that retries or fails timed-out pending requests.

Runs once per next_wake_up seconds. For each pending request we compare elapsed time against self.timeout; if elapsed has expired, we either resend the packet (consuming one retry) or surface a TimeoutError on the request's future. next_wake_up is the minimum remaining-time-to-timeout across all pending requests, so the loop wakes exactly when the next request needs servicing.

Source code in pyrad2/client_async.py

async def __timeout_handler__(self):
    """Background task that retries or fails timed-out pending requests.

    Runs once per `next_wake_up` seconds. For each pending request we
    compare elapsed time against `self.timeout`; if elapsed has expired,
    we either resend the packet (consuming one retry) or surface a
    TimeoutError on the request's future. `next_wake_up` is the minimum
    remaining-time-to-timeout across all pending requests, so the loop
    wakes exactly when the next request needs servicing.
    """
    try:
        while True:
            req2delete = []
            now = datetime.now()
            next_wake_up = float(self.timeout)

            for id, req in self.pending_requests.items():
                # send_date is always <= now, so compute elapsed as a
                # positive float. Using timedelta.seconds here would
                # wrap negative deltas to ~86399 and prematurely
                # trigger the timeout branch.
                elapsed = (now - req["send_date"]).total_seconds()
                if elapsed >= self.timeout:
                    if req["retries"] >= self.retries:
                        logger.debug(
                            "[{}:{}] For request {} execute all retries",
                            self.server,
                            self.port,
                            id,
                        )
                        req["future"].set_exception(
                            TimeoutError("Timeout on Reply")
                        )
                        req2delete.append(id)
                    else:
                        # Send again packet
                        req["send_date"] = now
                        req["retries"] += 1
                        logger.debug(
                            "[{}:{}] For request {} execute retry {}",
                            self.server,
                            self.port,
                            id,
                            req["retries"],
                        )
                        self.transport.sendto(req["packet"].request_packet())
                else:
                    remaining = self.timeout - elapsed
                    if remaining < next_wake_up:
                        next_wake_up = remaining

            for id in req2delete:
                # Remove request for map
                del self.pending_requests[id]

            # Floor sleeps at 0 so a just-expired request gets serviced
            # on the next loop iteration instead of busy-spinning.
            await asyncio.sleep(max(0.0, next_wake_up))

    except asyncio.CancelledError:
        pass

ClientAsync

Asyncio-based RADIUS client.

Sends Access-Request, Accounting-Request, CoA, and Status-Server packets over UDP, validates replies (including Message-Authenticator when present), and retries timed-out requests up to retries times with a per-request budget of timeout seconds.

EAP-MD5 is handled transparently: setting auth_type="eap-md5" on the request makes send_packet perform the EAP-Identity / Access-Challenge / EAP-MD5-Response round-trip and return only the final reply.

Source code in pyrad2/client_async.py

class ClientAsync:
    """Asyncio-based RADIUS client.

    Sends Access-Request, Accounting-Request, CoA, and Status-Server
    packets over UDP, validates replies (including
    ``Message-Authenticator`` when present), and retries timed-out
    requests up to ``retries`` times with a per-request budget of
    ``timeout`` seconds.

    EAP-MD5 is handled transparently: setting ``auth_type="eap-md5"``
    on the request makes ``send_packet`` perform the EAP-Identity /
    Access-Challenge / EAP-MD5-Response round-trip and return only
    the final reply.
    """

    def __init__(
        self,
        server: str,
        auth_port: int = 1812,
        acct_port: int = 1813,
        coa_port: int = 3799,
        secret: bytes = b"",
        dict: Optional[Dictionary] = None,
        retries: int = 3,
        timeout: int = 30,
        enforce_ma: bool = False,
    ):
        """Initializes an async RADIUS client.

        Args:
            server (str): Hostname or IP address of the RADIUS server.
            auth_port (int): Port to use for authentication packets.
            acct_port (int): Port to use for accounting packets.
            coa_port (int): Port to use for CoA packets.
            secret (bytes): RADIUS secret.
            dict (pyrad.dictionary.Dictionary): RADIUS dictionary.
            retries (int): Number of times to retry sending a RADIUS request.
            timeout (int): Number of seconds to wait for an answer.
            enforce_ma (bool): Enforce usage of Message-Authenticator.
        """
        self.server = server
        self.secret = secret
        self.retries = retries
        self.timeout = timeout
        self.dict = dict
        self.enforce_ma = enforce_ma

        self.auth_port = auth_port
        self.protocol_auth: Optional[DatagramProtocolClient] = None

        self.acct_port = acct_port
        self.protocol_acct: Optional[DatagramProtocolClient] = None

        self.protocol_coa: Optional[DatagramProtocolClient] = None
        self.coa_port = coa_port

    def _prepare_outgoing_packet(self, pkt: Packet) -> None:
        """Apply Message-Authenticator policy before a packet is sent."""
        prepare_request_message_authenticator(
            pkt,
            require_message_authenticator=self.enforce_ma,
        )

    async def initialize_transports(
        self,
        enable_acct: bool = False,
        enable_auth: bool = False,
        enable_coa: bool = False,
        local_addr: Optional[str] = None,
        local_auth_port: Optional[int] = None,
        local_acct_port: Optional[int] = None,
        local_coa_port: Optional[int] = None,
    ):
        task_list = []

        if not enable_acct and not enable_auth and not enable_coa:
            raise Exception("No transports selected")

        loop = asyncio.get_event_loop()
        if enable_acct and not self.protocol_acct:
            self.protocol_acct = DatagramProtocolClient(
                self.server,
                self.acct_port,
                self,
                retries=self.retries,
                timeout=self.timeout,
            )
            bind_addr = None
            if local_addr and local_acct_port:
                bind_addr = (local_addr, local_acct_port)

            acct_connect = loop.create_datagram_endpoint(
                self.protocol_acct,
                reuse_port=True,
                remote_addr=(self.server, self.acct_port),
                local_addr=bind_addr,
            )
            task_list.append(acct_connect)

        if enable_auth and not self.protocol_auth:
            self.protocol_auth = DatagramProtocolClient(
                self.server,
                self.auth_port,
                self,
                retries=self.retries,
                timeout=self.timeout,
            )
            bind_addr = None
            if local_addr and local_auth_port:
                bind_addr = (local_addr, local_auth_port)

            auth_connect = loop.create_datagram_endpoint(
                self.protocol_auth,
                reuse_port=True,
                remote_addr=(self.server, self.auth_port),
                local_addr=bind_addr,
            )
            task_list.append(auth_connect)

        if enable_coa and not self.protocol_coa:
            self.protocol_coa = DatagramProtocolClient(
                self.server,
                self.coa_port,
                self,
                retries=self.retries,
                timeout=self.timeout,
            )
            bind_addr = None
            if local_addr and local_coa_port:
                bind_addr = (local_addr, local_coa_port)

            coa_connect = loop.create_datagram_endpoint(
                self.protocol_coa,
                reuse_port=True,
                remote_addr=(self.server, self.coa_port),
                local_addr=bind_addr,
            )
            task_list.append(coa_connect)

        await asyncio.ensure_future(
            asyncio.gather(
                *task_list,
                return_exceptions=False,
            ),
            loop=loop,
        )

    async def deinitialize_transports(
        self,
        deinit_coa: bool = True,
        deinit_auth: bool = True,
        deinit_acct: bool = True,
    ) -> None:
        if self.protocol_coa and deinit_coa:
            await self.protocol_coa.close_transport()
            del self.protocol_coa
            self.protocol_coa = None
        if self.protocol_auth and deinit_auth:
            await self.protocol_auth.close_transport()
            del self.protocol_auth
            self.protocol_auth = None
        if self.protocol_acct and deinit_acct:
            await self.protocol_acct.close_transport()
            del self.protocol_acct
            self.protocol_acct = None

    def create_auth_packet(self, **args) -> AuthPacket:
        """Create a new RADIUS packet.
        This utility function creates a new RADIUS packet which can
        be used to communicate with the RADIUS server this client
        talks to. This is initializing the new packet with the
        dictionary and secret used for the client.

        Returns:
            packet.Packet: A new empty packet instance
        """
        if not self.protocol_auth:
            raise Exception("Transport not initialized")

        return AuthPacket(
            dict=self.dict,
            id=self.protocol_auth.create_id(),
            secret=self.secret,
            message_authenticator=True if self.enforce_ma else False,
            **args,
        )

    def create_acct_packet(self, **args) -> AcctPacket:
        """Create a new RADIUS packet.
        This utility function creates a new RADIUS packet which can
        be used to communicate with the RADIUS server this client
        talks to. This is initializing the new packet with the
        dictionary and secret used for the client.

        Returns:
            packet.Packet: A new empty packet instance
        """
        if not self.protocol_acct:
            raise Exception("Transport not initialized")

        return AcctPacket(
            id=self.protocol_acct.create_id(),
            dict=self.dict,
            secret=self.secret,
            **args,
        )

    def create_coa_packet(self, **args) -> CoAPacket:
        """Create a new RADIUS packet.
        This utility function creates a new RADIUS packet which can
        be used to communicate with the RADIUS server this client
        talks to. This is initializing the new packet with the
        dictionary and secret used for the client.

        Returns:
            packet.Packet: A new empty packet instance
        """

        if not self.protocol_coa:
            raise Exception("Transport not initialized")

        return CoAPacket(
            id=self.protocol_coa.create_id(), dict=self.dict, secret=self.secret, **args
        )

    def _status_protocol(self, port: str) -> DatagramProtocolClient:
        """Return the protocol used for a Status-Server health check."""
        if port == "auth":
            if not self.protocol_auth:
                raise Exception("Auth transport not initialized")
            return self.protocol_auth
        if port == "acct":
            if not self.protocol_acct:
                raise Exception("Accounting transport not initialized")
            return self.protocol_acct
        raise ValueError("Status-Server port must be 'auth' or 'acct'")

    def create_status_packet(self, *, port: str = "auth", **args) -> StatusPacket:
        """Create an RFC 5997 Status-Server health-check packet."""
        protocol = self._status_protocol(port)
        return StatusPacket(
            id=protocol.create_id(),
            dict=self.dict,
            secret=self.secret,
            **args,
        )

    def create_packet(self, id: int, **args) -> Packet:
        if not id:
            raise Exception("Missing mandatory packet id")

        return Packet(id=id, dict=self.dict, secret=self.secret, **args)

    def send_status_packet(
        self, pkt: Optional[StatusPacket] = None, *, port: str = "auth"
    ) -> asyncio.Future:
        """Send a Status-Server packet to the auth or accounting port."""
        protocol = self._status_protocol(port)
        if pkt is None:
            pkt = self.create_status_packet(port=port)

        ans: asyncio.Future = asyncio.Future(loop=asyncio.get_event_loop())
        self._prepare_outgoing_packet(pkt)
        protocol.send_packet(pkt, ans)
        return ans

    def send_packet(self, pkt: Packet) -> asyncio.Future:
        """Send a packet to a RADIUS server.

        Handles EAP-MD5 challenge/response automatically when
        ``pkt.auth_type == "eap-md5"``: an EAP-Identity is injected
        before the first send, and an Access-Challenge reply triggers
        a transparent second exchange that carries the MD5 response
        back to the server. The returned Future resolves with the
        final reply (Access-Accept or Access-Reject) or rejects with
        ``TimeoutError`` if retries are exhausted.

        Args:
            pkt (Packet): The packet to send

        Returns:
            asyncio.Future: Future related with packet to send
        """

        if isinstance(pkt, StatusPacket):
            return self.send_status_packet(pkt)

        if isinstance(pkt, AuthPacket):
            if not self.protocol_auth:
                raise Exception("Transport not initialized")
            return self._send_auth_packet(pkt)

        ans: asyncio.Future = asyncio.Future(loop=asyncio.get_event_loop())
        self._prepare_outgoing_packet(pkt)

        if isinstance(pkt, AcctPacket):
            if not self.protocol_acct:
                raise Exception("Transport not initialized")

            self.protocol_acct.send_packet(pkt, ans)

        elif isinstance(pkt, CoAPacket):
            if not self.protocol_coa:
                raise Exception("Transport not initialized")

            self.protocol_coa.send_packet(pkt, ans)

        else:
            raise Exception("Unsupported packet")

        return ans

    def _send_auth_packet(self, pkt: AuthPacket) -> asyncio.Future:
        """Send an Access-Request, handling the EAP-MD5 challenge round-trip."""
        assert self.protocol_auth is not None
        # Capture the protocol locally so the nested callback can use it
        # without re-asserting (mypy cannot prove self.protocol_auth is
        # still non-None when the callback fires).
        protocol = self.protocol_auth

        if pkt.auth_type == "eap-md5":
            eap.inject_eap_identity(pkt)

        outer: asyncio.Future = asyncio.Future(loop=asyncio.get_event_loop())
        self._prepare_outgoing_packet(pkt)

        first: asyncio.Future = asyncio.Future(loop=asyncio.get_event_loop())
        protocol.send_packet(pkt, first)

        def _on_first_reply(fut: asyncio.Future) -> None:
            if outer.done():
                return
            if fut.cancelled():
                outer.cancel()
                return
            first_exc = fut.exception()
            if first_exc is not None:
                outer.set_exception(first_exc)
                return

            reply = fut.result()
            if (
                pkt.auth_type == "eap-md5"
                and reply is not None
                and reply.code == PacketType.AccessChallenge
            ):
                try:
                    eap.apply_eap_md5_challenge(pkt, reply)
                except Exception as challenge_exc:  # noqa: BLE001
                    outer.set_exception(challenge_exc)
                    return

                # The retry uses the same Packet object, so it needs a
                # fresh id/authenticator before re-entering the transport.
                pkt.id = protocol.create_id()
                pkt.authenticator = pkt.create_authenticator()
                self._prepare_outgoing_packet(pkt)

                second: asyncio.Future = asyncio.Future(
                    loop=asyncio.get_event_loop()
                )
                protocol.send_packet(pkt, second)

                def _on_second_reply(fut2: asyncio.Future) -> None:
                    if outer.done():
                        return
                    if fut2.cancelled():
                        outer.cancel()
                        return
                    exc2 = fut2.exception()
                    if exc2 is not None:
                        outer.set_exception(exc2)
                    else:
                        outer.set_result(fut2.result())

                second.add_done_callback(_on_second_reply)
                return

            outer.set_result(reply)

        first.add_done_callback(_on_first_reply)
        return outer

__init__(server, auth_port=1812, acct_port=1813, coa_port=3799, secret=b'', dict=None, retries=3, timeout=30, enforce_ma=False)

Initializes an async RADIUS client.

Parameters:

Name	Type	Description	Default
server	str	Hostname or IP address of the RADIUS server.	required
auth_port	int	Port to use for authentication packets.	1812
acct_port	int	Port to use for accounting packets.	1813
coa_port	int	Port to use for CoA packets.	3799
secret	bytes	RADIUS secret.	b''
dict	Dictionary	RADIUS dictionary.	None
retries	int	Number of times to retry sending a RADIUS request.	3
timeout	int	Number of seconds to wait for an answer.	30
enforce_ma	bool	Enforce usage of Message-Authenticator.	False

Source code in pyrad2/client_async.py

def __init__(
    self,
    server: str,
    auth_port: int = 1812,
    acct_port: int = 1813,
    coa_port: int = 3799,
    secret: bytes = b"",
    dict: Optional[Dictionary] = None,
    retries: int = 3,
    timeout: int = 30,
    enforce_ma: bool = False,
):
    """Initializes an async RADIUS client.

    Args:
        server (str): Hostname or IP address of the RADIUS server.
        auth_port (int): Port to use for authentication packets.
        acct_port (int): Port to use for accounting packets.
        coa_port (int): Port to use for CoA packets.
        secret (bytes): RADIUS secret.
        dict (pyrad.dictionary.Dictionary): RADIUS dictionary.
        retries (int): Number of times to retry sending a RADIUS request.
        timeout (int): Number of seconds to wait for an answer.
        enforce_ma (bool): Enforce usage of Message-Authenticator.
    """
    self.server = server
    self.secret = secret
    self.retries = retries
    self.timeout = timeout
    self.dict = dict
    self.enforce_ma = enforce_ma

    self.auth_port = auth_port
    self.protocol_auth: Optional[DatagramProtocolClient] = None

    self.acct_port = acct_port
    self.protocol_acct: Optional[DatagramProtocolClient] = None

    self.protocol_coa: Optional[DatagramProtocolClient] = None
    self.coa_port = coa_port

create_auth_packet(**args)

Create a new RADIUS packet. This utility function creates a new RADIUS packet which can be used to communicate with the RADIUS server this client talks to. This is initializing the new packet with the dictionary and secret used for the client.

Returns:

Type	Description
AuthPacket	packet.Packet: A new empty packet instance

Source code in pyrad2/client_async.py

def create_auth_packet(self, **args) -> AuthPacket:
    """Create a new RADIUS packet.
    This utility function creates a new RADIUS packet which can
    be used to communicate with the RADIUS server this client
    talks to. This is initializing the new packet with the
    dictionary and secret used for the client.

    Returns:
        packet.Packet: A new empty packet instance
    """
    if not self.protocol_auth:
        raise Exception("Transport not initialized")

    return AuthPacket(
        dict=self.dict,
        id=self.protocol_auth.create_id(),
        secret=self.secret,
        message_authenticator=True if self.enforce_ma else False,
        **args,
    )

create_acct_packet(**args)

Create a new RADIUS packet. This utility function creates a new RADIUS packet which can be used to communicate with the RADIUS server this client talks to. This is initializing the new packet with the dictionary and secret used for the client.

Returns:

Type	Description
AcctPacket	packet.Packet: A new empty packet instance

Source code in pyrad2/client_async.py

def create_acct_packet(self, **args) -> AcctPacket:
    """Create a new RADIUS packet.
    This utility function creates a new RADIUS packet which can
    be used to communicate with the RADIUS server this client
    talks to. This is initializing the new packet with the
    dictionary and secret used for the client.

    Returns:
        packet.Packet: A new empty packet instance
    """
    if not self.protocol_acct:
        raise Exception("Transport not initialized")

    return AcctPacket(
        id=self.protocol_acct.create_id(),
        dict=self.dict,
        secret=self.secret,
        **args,
    )

create_coa_packet(**args)

Create a new RADIUS packet. This utility function creates a new RADIUS packet which can be used to communicate with the RADIUS server this client talks to. This is initializing the new packet with the dictionary and secret used for the client.

Returns:

Type	Description
CoAPacket	packet.Packet: A new empty packet instance

Source code in pyrad2/client_async.py

def create_coa_packet(self, **args) -> CoAPacket:
    """Create a new RADIUS packet.
    This utility function creates a new RADIUS packet which can
    be used to communicate with the RADIUS server this client
    talks to. This is initializing the new packet with the
    dictionary and secret used for the client.

    Returns:
        packet.Packet: A new empty packet instance
    """

    if not self.protocol_coa:
        raise Exception("Transport not initialized")

    return CoAPacket(
        id=self.protocol_coa.create_id(), dict=self.dict, secret=self.secret, **args
    )

create_status_packet(*, port='auth', **args)

Create an RFC 5997 Status-Server health-check packet.

Source code in pyrad2/client_async.py

def create_status_packet(self, *, port: str = "auth", **args) -> StatusPacket:
    """Create an RFC 5997 Status-Server health-check packet."""
    protocol = self._status_protocol(port)
    return StatusPacket(
        id=protocol.create_id(),
        dict=self.dict,
        secret=self.secret,
        **args,
    )

send_status_packet(pkt=None, *, port='auth')

Send a Status-Server packet to the auth or accounting port.

Source code in pyrad2/client_async.py

def send_status_packet(
    self, pkt: Optional[StatusPacket] = None, *, port: str = "auth"
) -> asyncio.Future:
    """Send a Status-Server packet to the auth or accounting port."""
    protocol = self._status_protocol(port)
    if pkt is None:
        pkt = self.create_status_packet(port=port)

    ans: asyncio.Future = asyncio.Future(loop=asyncio.get_event_loop())
    self._prepare_outgoing_packet(pkt)
    protocol.send_packet(pkt, ans)
    return ans

send_packet(pkt)

Send a packet to a RADIUS server.

Handles EAP-MD5 challenge/response automatically when pkt.auth_type == "eap-md5": an EAP-Identity is injected before the first send, and an Access-Challenge reply triggers a transparent second exchange that carries the MD5 response back to the server. The returned Future resolves with the final reply (Access-Accept or Access-Reject) or rejects with TimeoutError if retries are exhausted.

Parameters:

Name	Type	Description	Default
pkt	Packet	The packet to send	required

Returns:

Type	Description
Future	asyncio.Future: Future related with packet to send

Source code in pyrad2/client_async.py

def send_packet(self, pkt: Packet) -> asyncio.Future:
    """Send a packet to a RADIUS server.

    Handles EAP-MD5 challenge/response automatically when
    ``pkt.auth_type == "eap-md5"``: an EAP-Identity is injected
    before the first send, and an Access-Challenge reply triggers
    a transparent second exchange that carries the MD5 response
    back to the server. The returned Future resolves with the
    final reply (Access-Accept or Access-Reject) or rejects with
    ``TimeoutError`` if retries are exhausted.

    Args:
        pkt (Packet): The packet to send

    Returns:
        asyncio.Future: Future related with packet to send
    """

    if isinstance(pkt, StatusPacket):
        return self.send_status_packet(pkt)

    if isinstance(pkt, AuthPacket):
        if not self.protocol_auth:
            raise Exception("Transport not initialized")
        return self._send_auth_packet(pkt)

    ans: asyncio.Future = asyncio.Future(loop=asyncio.get_event_loop())
    self._prepare_outgoing_packet(pkt)

    if isinstance(pkt, AcctPacket):
        if not self.protocol_acct:
            raise Exception("Transport not initialized")

        self.protocol_acct.send_packet(pkt, ans)

    elif isinstance(pkt, CoAPacket):
        if not self.protocol_coa:
            raise Exception("Transport not initialized")

        self.protocol_coa.send_packet(pkt, ans)

    else:
        raise Exception("Unsupported packet")

    return ans