SIPRTPRecorder

#!/usr/bin/perl

$main::VERSION = sprintf "%d.%03d",q$Revision: 1.3 $ =~ /(\d+)\.(\d+)/;

use strict;
use warnings;

use sigtrap qw(handler signal_handler normal-signals);

use threads;
use threads::shared;
use Thread::Queue;

use Pod::Usage;
use Getopt::Long qw(:config posix_default no_ignore_case);
use File::Temp qw(tempfile tempdir);
use File::Path qw(rmtree);
use Net::Pcap;
use NetPacket::Ethernet;
use NetPacket::IP qw(IP_PROTO_TCP IP_PROTO_UDP);
use NetPacket::TCP;
use NetPacket::UDP;
use Net::RTP::Packet;
use Audio::SndFile;
use HTTP::Date;
use LWP::UserAgent;

my $interface = 'any';
my $dump = '';
my $snaplen = 2048;
my $url = "http://btnwfsqa02.btn.tech.xo.com:8080/FileUploadServlet";
my $base = "/record";
my $verbose = 0;
my $debug = 0;
my $help = 0;
my $manual = 0;
my $version = 0;

GetOptions('interface|i=s' => \$interface,
           'dump=s' => \$dump,
           'snaplen=i' => \$snaplen,
           'url|u=s' => \$url,
           'base|b=s' => \$base,
           'verbose|v!' => \$verbose,
           'debug|d+' => \$debug,
           'help|?!' => \$help,
           'manual|m!' => \$manual,
           'version|V!' => \$version)
  or pod2usage(-exitval => 1, -verbose => 1);

pod2usage(-exitval => 0,
          -verbose => 2,
          -noperldoc => 1)
  if $help;

pod2usage(-exitval => 0,
          -verbose => 2)
  if $manual;

pod2usage(-exitval => 0,
          -verbose => 0,
          -message => "Version $main::VERSION")
  if $version;

pod2usage(-exitval => 1,
          -verbose => 1,
          -message => "Missing --interface argument")
  unless $interface;

my $layer2_unpack;
if ($interface eq 'any') {
  # Linux cooked capture
  $layer2_unpack = "x[nnnH16]na*";
}
else {
  # Ethernet
  $layer2_unpack = "x[NnNn]na*";
}

my $temp_dir = tempdir(CLEANUP => 1);
chmod 0755, $temp_dir;

my $ua = LWP::UserAgent->new;
$ua->agent("SIPRTPRecorder/" . $main::VERSION . " (" . $ua->agent . ")");

my %active_calls : shared = ();
my %timeout_calls : shared = ();
my %active_rtp : shared = ();

my $pcap_error;

my $capture =
  Net::Pcap::open_live($interface, $snaplen, 0, 0, \$pcap_error);
unless (defined $capture) {
  die 'Unable to open device ', $interface, ': ', $pcap_error;
}

my $dumper;
if ($dump) {
  $dumper = Net::Pcap::dump_open($capture, $dump);
  unless (defined $dumper) {
    die 'Unable to create dump file ', $dump, ': ',
      Net::Pcap::geterr($capture);
  }
}

my $filter_str = 'port 5060';
my $filter;
if (Net::Pcap::compile($capture,
                       \$filter,
                       $filter_str,
                       1,
                       0)) {
  die 'Unable to compile filter "', $filter_str, '": ',
    Net::Pcap::geterr($capture);
}

if (Net::Pcap::setfilter($capture, $filter)) {
  die 'Unable to set packet capture filter: ',
    Net::Pcap::geterr($capture);
}


# Recording Queue handles initiation of Indbound and Outbound, and Cleanup of
# RTP streams

my $recording_queue = Thread::Queue->new;

sub recording_queue_reader {
  while (my $recording_element = $recording_queue->dequeue) {
    my ($type, $call_id) = @$recording_element;
    if ($type eq 'Exit') {
      return;
    }
    elsif ($type eq 'Cleanup') {
      my $inbound_address = $active_calls{$call_id}{inbound_address};
      my $inbound_port = $active_calls{$call_id}{inbound_port};
      my $outbound_address = $active_calls{$call_id}{outbound_address};
      my $outbound_port = $active_calls{$call_id}{outbound_port};

      print "* Cleaning Call-Id $call_id, ",
        "Record-Id: ", $active_calls{$call_id}{record_id}, ", ",
          "Inbound-RTP: $inbound_address:$inbound_port, ",
            "Outbound-RTP $outbound_address:$outbound_port\n"
              if $debug;

      delete $active_rtp{"$inbound_address:$inbound_port"};
      delete $active_rtp{"$outbound_address:$outbound_port"};

      my $expression = " or (host $inbound_address and dst port $inbound_port)";
      if ($filter_str =~ /(.*)\Q$expression\E(.*)?/) {
        $filter_str = "$1$2";
      }
      $expression = " or (host $outbound_address and dst port $outbound_port)";
      if ($filter_str =~ /(.*)\Q$expression\E(.*)?/) {
        $filter_str = "$1$2";
      }

      if (Net::Pcap::compile($capture,
                             \$filter,
                             $filter_str,
                             1,
                             0)) {
        warn 'Unable to compile filter "', $filter_str, '": ',
          Net::Pcap::geterr($capture);
        kill 6, $$;
        return;
      }

      if (Net::Pcap::setfilter($capture, $filter)) {
        warn 'Unable to set packet capture filter: ',
          Net::Pcap::geterr($capture);
        kill 6, $$;
        return;
      }

      print "* Filter '$filter_str'\n" if $debug > 1;

      $active_calls{$call_id}{cleaned} = 1;
    }
    else {
      my ($address, $port);
      if ($type eq 'Inbound') {
        $address = $active_calls{$call_id}{inbound_address};
        $port = $active_calls{$call_id}{inbound_port};
      }
      elsif ($type eq 'Outbound') {
        $address = $active_calls{$call_id}{outbound_address};
        $port = $active_calls{$call_id}{outbound_port};
      }
      print "* Recording $type Call-Id: $call_id, ",
        "Record-Id: ", $active_calls{$call_id}{record_id}, ", ",
          "RTP: $address:$port\n",
            if $verbose;

      $filter_str .= " or (host $address and dst port $port)";

      my $previous_filter = $filter;
      if (Net::Pcap::compile($capture,
                             \$filter,
                             $filter_str,
                             1,
                             0)) {
        die 'Unable to compile filter "', $filter_str, '": ',
          Net::Pcap::geterr($capture);
      }
      if (Net::Pcap::setfilter($capture, $filter)) {
        die 'Unable to set packet capture filter: ',
          Net::Pcap::geterr($capture);
      }
      if (Net::Pcap::freecode($previous_filter)) {
        die 'Unable to free previous filter: ',
          Net::Pcap::geterr($capture);
      }

      print "* Filter '$filter_str'\n" if $debug > 1;
    }
  }
}

my $recording_thread = threads->create(\&recording_queue_reader);


# Writer Queue handles file operations for RTP streams

my $writer_queue = Thread::Queue->new;
my %writer_files = ();

sub writer_queue_reader {
  while (my $writer_element = $writer_queue->dequeue) {
    my ($type, $call_id, $direction, $data) = @$writer_element;
    if ($type eq 'Exit') {
      return;
    }
    elsif ($type eq 'Close') {
      my %upload_parts = ();

      foreach my $file_type (qw(inbound outbound dump)) {
        if (exists $active_calls{$call_id}{$file_type . "_file"}) {
          my $file = $active_calls{$call_id}{$file_type . "_file"};
          if ($file_type eq "dump") {
            my $dump_handle = $active_calls{$call_id}{dump_handle};
            if (Net::Pcap::dump_close($dump_handle)) {
              warn 'Unable to close dump file ', $file, ': ',
                Net::Pcap::geterr($capture);
            }

            delete $active_calls{$call_id}{dump_handle};

            print "* Closed dump ", $file, "\n"
              if ($debug);

            $upload_parts{$file_type} =
              [ $file =>
                $base
                . "/"
                . $active_calls{$call_id}{record_id}
                . ".pcap"
              ];
          }
          else {
            if (exists $writer_files{$file}) {
              my $sndfile = $writer_files{$file}{sndfile};
              $sndfile->close;

              my $fh = $writer_files{$file}{fh};
              close $fh;

              print "* Closed file ", $file, "\n"
                if ($debug);

              delete $writer_files{$file};

              $upload_parts{$file_type} =
                [ $file =>
                  $base
                  . "/"
                  . $active_calls{$call_id}{record_id}
                  . "_"
                  . $file_type
                  . ".wav"
                ];
            }
          }
        }
      }

      my $upload_response =
        $ua->post($url, \%upload_parts, Content_Type => "form-data");

      if ($upload_response->is_success) {
        print "* Uploaded files to ", $base, "/",
          $active_calls{$call_id}{record_id}, " at ", $url, "\n"
            if $debug;
      }
      else {
        warn "Failed to upload files to $url: ",
          $upload_response->status_line, "\n";
      }

      foreach my $file_type (qw(inbound_file outbound_file dump_file)) {
        if (exists $active_calls{$call_id}{$file_type}) {
          my $file = $active_calls{$call_id}{$file_type};
          unlink $file;
        }
      }

      $active_calls{$call_id}{closed} = 1;
    }
    elsif ($type eq 'Write') {
      my $file;
      if (exists $active_calls{$call_id}{$direction . "_file"}) {
        $file = $active_calls{$call_id}{$direction . "_file"};
      }
      my $sndfile;
      if (defined $file && exists $writer_files{$file}) {
        $sndfile = $writer_files{$file}{sndfile};
      }
      else {
        my $fh;
        ($fh, $file) = File::Temp::tempfile(DIR => $temp_dir, SUFFIX => ".wav");
        chmod 0644, $file;
        $sndfile =
          Audio::SndFile->open(">",
                               $fh,
                               type => "wav",
                               subtype => "ulaw",
                               endianness => "file",
                               channels => 1,
                               samplerate => 8000);
        $sndfile->set_title($active_calls{$call_id}{record_id});

        $active_calls{$call_id}{$direction . "_file"} = $file;
        $writer_files{$file}{sndfile} = $sndfile;
        $writer_files{$file}{fh} = $fh;

        print "* Created file ", $file, " for ", $direction, " RTP stream ",
          $active_calls{$call_id}{$direction . "_address"}, ":",
            $active_calls{$call_id}{$direction . "_port"}, "\n"
            if ($debug);
      }

      $sndfile->write_raw($data);
    }
  }
}

my $writer_thread = threads->create(\&writer_queue_reader);


# Timeout calls, usually this simply deletes old %active_calls keys, and
# occasionally closes resources for calls which have not had a SIP BYE message

my $timeout_handler_active : shared = 1;
sub timeout_handler {
  while ($timeout_handler_active) {
    sleep 1;
    my $time = time;
    foreach my $call_id (keys %timeout_calls) {
      my $timeout = $timeout_calls{$call_id};
      if ($time > $timeout) {
        my $close = 0;
        my $cleanup = 0;

        if (exists $active_calls{$call_id}) {
          if (!exists $active_calls{$call_id}{closed}
              || !$active_calls{$call_id}{closed}) {
            $close = 1;
          }
          if (!exists $active_calls{$call_id}{cleaned}
              || !$active_calls{$call_id}{cleaned}) {
            $cleanup = 1;
          }
        }

        if ($close || $cleanup) {
          print "* Timeout reached for $call_id\n" if ($verbose || $debug);
          if ($close) {
            my @writer_element : shared = ('Close', $call_id, '', '');
            $writer_queue->enqueue(\@writer_element);
          }
          if ($cleanup) {
            my @recording_element : shared = ('Cleanup', $call_id);
            $recording_queue->enqueue(\@recording_element);
          }
        }
        else {
          delete $active_calls{$call_id};
          delete $timeout_calls{$call_id};
        }
      }
    }
  }
}

my $timeout_thread = threads->create(\&timeout_handler);


# Packet processing callback

sub capture_packets($$$) {
  my ($user_data, $header, $packet) = @_;

  if (defined $dumper) {
    if (Net::Pcap::dump($dumper, $header, $packet)) {
      warn 'Unable to write dump file ', $dump, ': ',
        Net::Pcap::geterr($capture);
    }
  }

  my ($layer2_type, $layer2_data) = unpack($layer2_unpack, $packet);
  if ($layer2_type != 0x0800) {
    print "Unrecognized Layer 2 type ", $layer2_type, ", expected IPv4 (0x800)"
      if ($verbose || $debug);
    return;
  }

  my $ip = NetPacket::IP->decode($layer2_data);
  my $proto;
  if ($ip->{proto} == IP_PROTO_UDP) {
    $proto = NetPacket::UDP->decode($ip->{data});
  }
  elsif ($ip->{proto} == IP_PROTO_TCP) {
    $proto = NetPacket::TCP->decode($ip->{data});
  }
  else {
    print "Unexpected IP protocol ", $ip->{proto},
      ", expected TCP (", IP_PROTO_TCP, ") or UDP (", IP_PROTO_UDP, ")\n"
        if $verbose || $debug;
    return;
  }

  print $ip->{src_ip}, ":", $proto->{src_port}, " -> ",
    $ip->{dest_ip}, ":", $proto->{dest_port}, "\n",
      ($proto->{src_port} == 5060
       || $proto->{dest_port} == 5060
       || $debug > 2 ? $proto->{data} . "\n" :"")
        if $debug > 1;

  # Process SIP requests
  if ($proto->{src_port} == 5060 || $proto->{dest_port} == 5060) {
    if ($proto->{data} =~ /\bCall-ID: (.*)\b/i) {
      my $call_id = $1;

      print "Examining $call_id\n" if $debug;

      if (exists $active_calls{$call_id}) {
        if (exists $active_calls{$call_id}{dump_file}
            && exists $active_calls{$call_id}{dump_handle}) {
          my $dump_file = $active_calls{$call_id}{dump_file};
          my $dump_handle = $active_calls{$call_id}{dump_handle};
          if (Net::Pcap::dump($dump_handle, $header, $packet)) {
            warn 'Unable to write dump file ', $dump_file, ': ',
              Net::Pcap::geterr($capture);
          }
        }

        if (exists $active_calls{$call_id}{outbound_address}) {
          if ($proto->{data} =~ /\bBYE /i) {
            print "  Found BYE\n" if $debug;
            my @writer_element : shared = ('Close', $call_id, '', '');
            $writer_queue->enqueue(\@writer_element);
            my @recording_element : shared = ('Cleanup', $call_id);
            $recording_queue->enqueue(\@recording_element);
          }
          else {
            print "  Active\n" if $debug;
          }
        }
        else {
          print "  Checking for outbound RTP\n" if $debug;
          if ($proto->{data} =~ /\bc=IN IP4 (.*)\b/i) {
            my $address = $1;
            print "    Address $address\n" if $debug;
            if ($proto->{data} =~ /\bm=audio (\d+) RTP/i) {
              my $port = $1;
              print "    Port $port\n" if $debug;
              $active_calls{$call_id}{outbound_address} = $address;
              $active_calls{$call_id}{outbound_port} = $port;
              $active_rtp{"$address:$port"} = $call_id;

              my @recording_element : shared = ('Outbound', $call_id);
              $recording_queue->enqueue(\@recording_element);
            }
          }
        }
      }
      else {
        if ($proto->{data} =~ /\bRecord-Call: (.*)\b/i) {
          my $record_id = $1;

          my $timeout = time + 300;
          if ($proto->{data} =~ /\bRecord-Until: (.*)\b/i) {
            $timeout = str2time($1);
          }

          print "  Found Record header in $call_id\n" if $debug;
          print "    Timeout ", time2str($timeout), "\n" if $debug;
          if ($proto->{data} =~ /\bc=IN IP4 (.*)\b/i) {
            my $address = $1;
            print "    Address $address\n" if $debug;
            if ($proto->{data} =~ /\bm=audio (\d+) RTP/i) {
              my $port = $1;
              print "    Port $port\n" if $debug;

              $active_calls{$call_id} = &share({});
              $active_calls{$call_id}{record_id} = $record_id;
              $active_calls{$call_id}{inbound_address} = $address;
              $active_calls{$call_id}{inbound_port} = $port;
              $timeout_calls{$call_id} = $timeout;
              $active_rtp{"$address:$port"} = $call_id;

              my ($dump_fh, $dump_file) =
                File::Temp::tempfile(DIR => $temp_dir, SUFFIX => ".pcap");
              close $dump_fh;
              chmod 0644, $dump_file;
              my $dump_handle = Net::Pcap::dump_open($capture, $dump_file);
              unless (defined $dump_handle) {
                warn 'Unable to create dump file ', $dump_file, ': ',
                  Net::Pcap::geterr($capture);
              }

              if (defined $dump_handle) {
                if (Net::Pcap::dump($dump_handle, $header, $packet)) {
                  warn 'Unable to write dump file ', $dump_file, ': ',
                    Net::Pcap::geterr($capture);
                }
                else {
                  $active_calls{$call_id}{dump_file} = $dump_file;
                  $active_calls{$call_id}{dump_handle} = &share($dump_handle);
                }
              }

              my @recording_element : shared = ('Inbound', $call_id);
              $recording_queue->enqueue(\@recording_element);
            }
          }
        }
      }
    }
  }
  # Process RTP streams
  else {
    my $call_id;
    my ($address, $port);
    if (exists $active_rtp{$ip->{src_ip} . ":" . $proto->{src_port}}) {
      $address = $ip->{src_ip};
      $port = $proto->{src_port};
      $call_id = $active_rtp{"$address:$port"};
    }
    elsif (exists $active_rtp{$ip->{dest_ip} . ":" . $proto->{dest_port}}) {
      $address = $ip->{dest_ip};
      $port = $proto->{dest_port};
      $call_id = $active_rtp{"$address:$port"};
    }
    else {
      print "Unexpected RTP stream from ",
        $ip->{src_ip}, ":", $proto->{src_port},
          " to ", $ip->{dest_ip}, ":", $proto->{dest_port},
            " not found in active RTP streams\n",
            if ($verbose || $debug);
      return;
    }

    if (exists $active_calls{$call_id}{dump_file}
        && exists $active_calls{$call_id}{dump_handle}) {
      my $dump_file = $active_calls{$call_id}{dump_file};
      my $dump_handle = $active_calls{$call_id}{dump_handle};
      if (Net::Pcap::dump($dump_handle, $header, $packet)) {
        warn 'Unable to write dump file ', $dump_file, ': ',
          Net::Pcap::geterr($capture);
      }
    }

    my $direction;
    if ($address eq $active_calls{$call_id}{inbound_address}
        && $port == $active_calls{$call_id}{inbound_port}) {
      $direction = "inbound";
    }
    elsif ($address eq $active_calls{$call_id}{outbound_address}
           && $port == $active_calls{$call_id}{outbound_port}) {
      $direction = "outbound";
    }
    else {
      print "Unexpected RTP stream from ",
        $ip->{src_ip}, ":", $proto->{src_port},
          " to ", $ip->{dest_ip}, ":", $proto->{dest_port},
            " does not match active calls\n",
              if ($verbose || $debug);
      return;
    }

    my $rtp = new Net::RTP::Packet($proto->{data});
    if ($rtp->{payload_type} != 0) {
      print "Unrecognized RTP payload type ", $rtp->{payload_type},
        " expected PCMU (0)\n"
          if ($verbose || $debug);
      return;
    }

    my @writer_element : shared =
      ('Write', $call_id, $direction, $rtp->payload);
    $writer_queue->enqueue(\@writer_element);
  }
}

unless (Net::Pcap::loop($capture, -1, \&capture_packets, '')) {
  die 'Unable to perform packet capture',
    Net::Pcap::geterr($capture);
}


# Signal handler for cleaning up temporary files

sub signal_handler {
  print "Caught signal ", @_, "\n" if ($verbose || $debug);

  $timeout_handler_active = 0;

  if (defined $writer_queue) {
    my @writer_element : shared = ('Exit', '', '', '');
    $writer_queue->enqueue(\@writer_element);
  }

  if (defined $recording_queue) {
    my @recording_element : shared = ('Exit', '');
    $recording_queue->enqueue(\@recording_element);
  }

  if (defined $timeout_thread) {
    $timeout_thread->join;
  }

  if (defined $writer_thread) {
    $writer_thread->join;
  }

  if (defined $recording_thread) {
    $recording_thread->join;
  }

  if (defined $temp_dir && -d $temp_dir) {
    print "Removing directory $temp_dir\n" if $debug;
    rmtree($temp_dir, ($debug > 1 ? 1 : 0), 0);
  }

  if (defined $capture) {
    Net::Pcap::close($capture);
  }

  if (defined $dumper) {
    if (Net::Pcap::dump_close($dumper)) {
      warn 'Unable to close dump file ', $dump, ': ',
        Net::Pcap::geterr($capture);
    }
  }

  exit;
}


__END__

=head1 NAME

siprtprecorder - Capture RTP streams triggered by SIP requests.

=head1 SYNOPSIS

siprtprecorder [options]

=head1 OPTIONS

=over 8

=item B<--interface, -i>

Interface to capture SIP requests and RTP streams, may be specified multiple
times.  Default value is "any".

=item B<--dump>

Optional file which all captured packets will be written to for debugging
purposes.

=item B<--url, -u>

URL of FileUploadServlet where audio files are uploaded.  Default value is
"http://btnwfsqa02.btn.tech.xo.com:8080/FileUploadServlet".

=item B<--base, -b>

Base path of audio files uploaded to FileUploadServet.  Default is "/record".

=item B<--verbose, v>

Print verbose output.

=item B<--debug, -d>

Print debugging output, may specify multiple times for increased debugging.

=item B<--help, -?>

Print a help message and exit.

=item B<--manual, -m>

Display the complete documentation.

=item B<--version, -V>

Print the version and exit.

=back

=head1 DESCRIPTION

This program monitors a network interface for SIP requests containing a
I<Record> header which triggers the program to capture an I<RTP> stream.

=cut