Client Async

ClientAsync

Basic RADIUS client. This class implements a basic RADIUS client. It can send requests to a RADIUS server, taking care of timeouts and retries, and validate its replies.

Source code in pyrad2/client_async.py

class ClientAsync:
    """Basic RADIUS client.
    This class implements a basic RADIUS client. It can send requests
    to a RADIUS server, taking care of timeouts and retries, and
    validate its replies.
    """

    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,
    ):
        """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.
        """
        self.server = server
        self.secret = secret
        self.retries = retries
        self.timeout = timeout
        self.dict = dict

        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

    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 CreateAuthPacket(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,
            **args,
        )

    def CreateAcctPacket(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 CreateCoAPacket(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 CreatePacket(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 SendPacket(self, pkt: Packet) -> asyncio.Future:
        """Send a packet to a RADIUS server.

        Args:
            pkt (Packet): The packet to send

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

        ans: asyncio.Future = asyncio.Future(loop=asyncio.get_event_loop())

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

            self.protocol_auth.send_packet(pkt, ans)

        elif 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

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

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

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,
):
    """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.
    """
    self.server = server
    self.secret = secret
    self.retries = retries
    self.timeout = timeout
    self.dict = dict

    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

CreateAuthPacket(**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 CreateAuthPacket(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,
        **args,
    )

CreateAcctPacket(**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 CreateAcctPacket(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,
    )

CreateCoAPacket(**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 CreateCoAPacket(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
    )

SendPacket(pkt)

Send a packet to a RADIUS server.

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 SendPacket(self, pkt: Packet) -> asyncio.Future:
    """Send a packet to a RADIUS server.

    Args:
        pkt (Packet): The packet to send

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

    ans: asyncio.Future = asyncio.Future(loop=asyncio.get_event_loop())

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

        self.protocol_auth.send_packet(pkt, ans)

    elif 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