File Coverage

blib/lib/AnyEvent/Radius/Client.pm

Criterion	Covered	Total	%
statement	96	161	59.6
branch	13	56	23.2
condition	4	9	44.4
subroutine	19	31	61.2
pod	10	13	76.9
total	142	270	52.5

package AnyEvent::Radius::Client;
# AnyEvent-based radius client
use strict;
use warnings;
use AnyEvent;
use AnyEvent::Handle::UDP;

use base qw(Class::Accessor::Fast);
__PACKAGE__->mk_accessors(qw(
                handler packer send_cache
                queue_cv write_cv read_cv
                sent_cnt reply_cnt queue_cnt
                last_request_id
            ));

use Data::Radius::Constants qw(%RADIUS_PACKET_TYPES);
use Data::Radius::Dictionary ();
use Data::Radius::Packet ();

use constant {
    READ_TIMEOUT_SEC => 5,
    WRITE_TIMEOUT_SEC => 5,
    RADIUS_PORT => 1812,
    MAX_QUEUE => 255,
};

# new 'NAS'
# args:
#   ip
#   port
#   secret
#   dictionary
#   read_timeout
#   write_timeout
#   bind_ip
#- callbacks:
#    on_read
#    on_read_raw
#    on_read_timeout
#    on_write_timeout
#    on_error
sub new {
    my ($class, %h) = @_;

    my $obj = bless {}, $class;

    # either pre-created packer object, or need radius secret to create new one
    # dictionary is optional
    if ( defined $h{packer} ) {
        $obj->packer( $h{packer} );
    } elsif ( defined $h{secret} ) {
        $obj->packer( Data::Radius::Packet->new(dict => $h{dictionary}, secret => $h{secret}) );
    } else {
        die "No radius secret";
    }

    my %udp_handle_args = (
        rtimeout => $h{read_timeout} // READ_TIMEOUT_SEC,
        wtimeout => $h{write_timeout} // WRITE_TIMEOUT_SEC,
    );

    die "No IP argument" if ! exists $h{ip};
    $udp_handle_args{connect} = [ $h{ip}, $h{port} // RADIUS_PORT ];
    $udp_handle_args{bind} = [$h{bind_ip}, 0] if exists $h{bind_ip};

    $udp_handle_args{on_recv} = sub {
        my ($data, $handle, $from) = @_;
        $obj->read_cv->end;
        $obj->reply_cnt($obj->reply_cnt + 1);

        if ($h{on_read_raw}) {
            # dump raw data
            $h{on_read_raw}->($obj, $data, $from);
        }

        # using authenticator from request to verify reply
        my $request_id = $obj->packer()->request_id($data);
        # FIXME how to react on unknown request_id ?
        my $send_info = delete $obj->send_cache()->{ $request_id };
        if (! $send_info ) {
            # got unknown reply (with wrong request id?)
            if ($h{on_error}) {
                $h{on_error}->($obj, 'Unknown reply');
            }
            else {
                warn "Error: unknown reply";
            }
        }
        else {
            my $on_read = $h{on_read};
            my $req_callback = $send_info->{callback};
            if ( $on_read || $req_callback ) {
                # how to decode $from
                # my($port, $host) = AnyEvent::Socket::unpack_sockaddr($from);
                # my $ip = format_ipv4($host);

                my ($type, $req_id, $auth, $av_list) = $obj->packer()->parse($data, $send_info->{authenticator});

                $on_read->($obj, {
                            type => $type,
                            request_id => $req_id,
                            av_list => $av_list,
                            # from is sockaddr binary data
                            from => $from,
                            authenticator => $auth,
                        }) if $on_read;
                $req_callback->($type, $av_list) if $req_callback;
            }
        }

        $obj->queue_cv->end;
    };

    $udp_handle_args{on_rtimeout} = sub {
        my $handle = shift;
        if(! $obj->read_cv->ready) {
            if($h{on_read_timeout}) {
                $h{on_read_timeout}->($obj, $handle);
            }
            $obj->clear_send_cache();
            # stop queue
            $obj->queue_cv->send;
        }
        $handle->clear_rtimeout();
    };

    $udp_handle_args{on_wtimeout}  = sub {
        my $handle = shift;
        if(! $obj->write_cv->ready) {
            if($h{on_write_timeout}) {
                $h{on_write_timeout}->($obj, $handle);
            }
            $obj->clear_send_cache();
            # stop queue
            $obj->queue_cv->send;
        }
        $handle->clear_wtimeout();
    };

    # low-level socket errors
    $udp_handle_args{on_error} = sub {
        my ($handle, $fatal, $error) = @_;
        # abort all
        $handle->clear_wtimeout();
        $handle->clear_rtimeout();
        $obj->clear_send_cache();
        $obj->queue_cv->send;
        if ($h{on_error}) {
            $h{on_error}->($obj, $error);
        }
        else {
            warn "Error occured: $error";
        }
        # the handle::udp self destroys right after calling the on_error_handler
        # so the client have to do the same
        $obj->destroy() if $fatal;
    };

    $obj->handler( AnyEvent::Handle::UDP->new(%udp_handle_args) );

    $obj->init();

    return $obj;
}

sub clear_send_cache {
    my $self = shift;
    my $send_cache = $self->send_cache();
    $self->send_cache({});
    if ($send_cache) {
        my @ordered_reqids = sort { $send_cache->{$a}{time_cached} <=> $send_cache->{$b}{time_cached} } keys %$send_cache;
        foreach my $request_id (@ordered_reqids) {
            if (my $cb = $send_cache->{$request_id}{callback}) {
                $cb->();
            }
        }
    }
}

sub _send_packet {
    my ($self, $packet) = @_;

    $self->queue_cnt($self->queue_cnt() + 1);

    # +1
    $self->queue_cv()->begin;
    $self->write_cv()->begin;
    $self->read_cv()->begin;

    my $cv = AnyEvent->condvar;

    $cv->cb(sub {
        $self->sent_cnt($self->sent_cnt() + 1);
        # -1
        $self->write_cv()->end;
    });

    # cv->send is called by Handle::UDP when packet is sent
    $self->handler()->push_send($packet, undef, $cv);
}

# wait for Handle to send all queued packets (or timeout)
# object is not usable after it - call init()
sub wait {
    my $self = shift;

    $self->queue_cv()->recv();
}

# reset vars - need to be called after wait() or on_ready()
sub init {
    my $self = shift;

    $self->read_cv(AnyEvent->condvar);
    $self->write_cv(AnyEvent->condvar);
    $self->queue_cv(AnyEvent->condvar);
    $self->sent_cnt(0);
    $self->reply_cnt(0);
    $self->queue_cnt(0);
    $self->send_cache({});
    $self->last_request_id( int rand(0x100) );
}

# close open socket, object is unusable after it was called
sub destroy {
    my $self = shift;
    $self->handler()->destroy();
    $self->handler(undef);
}

my $_IN_GLOBAL_DESTRUCTION = 0;
END {
    $_IN_GLOBAL_DESTRUCTION = 1;
}

sub DESTROY {
    my $self = shift;
    if (defined ${^GLOBAL_PHASE}) {
        # >= 5.14
        return if (${^GLOBAL_PHASE} eq 'DESTRUCT');
    }
    else {
        # before 5.14, see also Devel::GlobalDestruction
        return if $_IN_GLOBAL_DESTRUCTION;
    }

    return if (! $self->handler());
    $self->handler()->destroy();
}

# group wait
# cv is AnyEvent condition var passed outside
#
# Example:
#  my $cv = AnyEvent->condvar;
#  $nas1->on_ready($cv);
#  $nas2->on_ready($cv);
#  $nas3->on_ready($cv);
#  $cv->recv;
#
sub on_ready {
    my ($self, $cv) = @_;

    $cv->begin();
    $self->queue_cv()->cb(sub { $cv->end });
}

sub load_dictionary {
    my ($class, $path) = @_;
    my $dict = Data::Radius::Dictionary->load_file($path);

    if(ref($class)) {
        $class->packer()->dict($dict);
    }

    return $dict;
}

sub new_request_id {
    my $self = shift;
    my $last_request_id = $self->last_request_id();
    my $new_request_id = ($last_request_id + 1) & 0xFF;
    my $send_cache = $self->send_cache();
    while (exists $send_cache->{$new_request_id}) {
        $new_request_id = ($new_request_id + 1) & 0xFF;
        return undef if $new_request_id == $last_request_id; # send cache full ??
    }
    return $new_request_id;
}

# add packet to the queue
# type - radius request packet type code or its text alias
# av_list - list of attributes in {Name => ... Value => ... } form
# cb - optional callback to be called on result:
#      - when received response as $cb->($resp_type, $resp_av_list)
#      - when failed (eg time out, invalid or non matching response)
#        with empty parameter list cb->();
sub send_packet {
    my ($self, $type, $av_list, $cb) = @_;

    if ($self->queue_cnt >= MAX_QUEUE) {
        # queue overflow
        return undef;
    }

    my $request_id = $self->new_request_id();
    return undef if !defined $request_id;

    $type = $RADIUS_PACKET_TYPES{$type} if exists $RADIUS_PACKET_TYPES{$type};

    my ($packet, $req_id, $auth) = $self->packer()->build(
                        type => $type,
                        av_list => $av_list,
                        with_msg_auth => 1,
                        request_id => $request_id,
                    );

    $self->last_request_id( $req_id );

    # required to verify reply
    $self->send_cache()->{ $req_id } = {
        authenticator => $auth,
        type => $type,
        callback => $cb,
        time_cached => AE::now(),
    };

    $self->_send_packet($packet);

    return wantarray() ? ($req_id, $auth) : $req_id;
}

# shortcut methods:

sub send_auth {
    my $self = shift;
    return $self->send_packet(AUTH => @_);
}

sub send_acct {
    my $self = shift;
    return $self->send_packet(ACCT => @_);
}

sub send_pod {
    my $self = shift;
    return $self->send_packet(POD => @_);
}

sub send_coa {
    my $self = shift;
    return $self->send_packet(COA => @_);
}

1;

__END__