spandspflow2pcap.py

Go to the documentation of this file.

#!/usr/bin/env python
# vim: set ts=8 sw=4 sts=4 et ai tw=79:
'''
Usage: ./spandspflow2pcap.py SPANDSP_LOG SENDFAX_PCAP
 
Takes a log from Asterisk with SpanDSP, extracts the "received" data
and puts it in a pcap file. Use 'fax set debug on' and configure
logger.conf to get fax logs.
 
Input data should look something like this::
 
    [2013-08-07 15:17:34] FAX[23479] res_fax.c: FLOW T.38 Rx     5: IFP c0 ...
 
Output data will look like a valid pcap file ;-)
 
This allows you to reconstruct received faxes into replayable pcaps.
 
Replaying is expected to be done by SIPp with sipp-sendfax.xml. The
SIPp binary used for replaying must have image (fax) support. This means
you'll need a version higher than 3.5.0 (unreleased when writing this),
or the git master branch: https://github.com/SIPp/sipp
 
 
Author: Walter Doekes, OSSO B.V. (2013,2015,2016,2019)
License: Public Domain
'''
from base64 import b16decode
from collections import namedtuple
from datetime import datetime, timedelta
from re import search
from time import mktime
from struct import pack
import os
import sys
 
 
LOSSY = False
EMPTY_RECOVERY = False
 
 
IFP = namedtuple('IFP', 'date seqno data')  # datetime, int, bytearray
 
 
def n2b(text):
    """
    Convert "aa bb cc" to bytearray('\xaa\xbb\xcc').
    """
    return bytearray(
        b16decode(text.replace(' ', '').replace('\n', '').upper()))
 
 
class SkipPacket(Exception):
    pass
 
 
class FaxPcap(object):
    PCAP_PREAMBLE = n2b(
        'd4 c3 b2 a1 02 00 04 00'
        '00 00 00 00 00 00 00 00'
        'ff ff 00 00 71 00 00 00')
 
    def __init__(self, outfile):
        self.outfile = outfile
        self.date = None
        self.seqno = None
        self.udpseqno = 128
        self.prev_data = None
 
        # Only do this if at pos 0?
 
    def add(self, ifp):
        """
        Add the IFP packet.
 
        T.38 basic format of UDPTL payload section with redundancy:
 
        UDPTL_SEQNO
        - 2 sequence number (big endian)
        UDPTL_PRIMARY_PAYLOAD (T30?)
        - 1 subpacket length (excluding this byte)
        - 1 type of message (e.g. 0xd0 for data(?))
        - 1 items in data field (e.g. 0x01)
        - 2 length of data (big endian)
        - N data
        RECOVERY (optional)
        - 2 count of previous seqno packets (big endian)
        - N UDPTL_PRIMARY_PAYLOAD of (seqno-1)
        - N UDPTL_PRIMARY_PAYLOAD of (seqno-2)
        - ...
        """
        # First packet?
        if self.seqno is None:
            # Add preamble.
            self._add_preamble()
            # Start a second late (optional).
            self._add_garbage(ifp.date)
 
            # Set sequence, and fill with missing leading zeroes.
            self.seqno = 0
            for i in range(ifp.seqno):
                self.add(IFP(date=ifp.date, seqno=i, data=bytearray([0])))
 
        # Auto-increasing dates
        if self.date is None or ifp.date > self.date:
            self.date = ifp.date
        elif ifp.date < self.date.replace(microsecond=0):
            assert False, 'More packets than expected in 1s? {!r}/{!r}'.format(
                ifp.date, self.date)
        else:
            self.date += timedelta(microseconds=9000)
 
        # Add packet.
        self.seqno = ifp.seqno
        try:
            self.outfile.write(self._make_packet(ifp.data))
        except SkipPacket:
            pass
 
    def _add_preamble(self):
        self.outfile.write(self.PCAP_PREAMBLE)
 
    def _add_garbage(self, date):
        if self.date is None or date > self.date:
            self.date = date
 
        self.seqno = 0xffff
        self.outfile.write(self._make_packet(
            bytearray(b'GARBAGE'), is_ifp=False))
 
    def _make_packet(self, ifp_data, is_ifp=True):
        sum16 = bytearray(b'\x43\x21')  # the OS fixes the checksums for us
 
        data = bytearray()
        if is_ifp:
            data.append(len(ifp_data))  # length
            data.extend(ifp_data)       # data
            self.prev_data, prev_data = data[:], self.prev_data
        else:
            data.extend(ifp_data)
            prev_data = None
 
        if prev_data:
            if LOSSY and (self.seqno % 3) == 2:
                self.udpseqno += 1
                raise SkipPacket()
 
            if EMPTY_RECOVERY:
                # struct ast_frame f[16], we have room for a few
                # packets.
                packets = 14
                data.extend([0, packets + 1] + [0] * packets)
                data.extend(prev_data)
            else:
                # Add 1 previous packet, without the seqno.
                data.extend([0, 1])
                data.extend(prev_data)
 
        # Wrap it in UDP
        udp = bytearray(
            b'\x00\x01\x00\x02%(len)s%(sum16)s%(seqno)s%(data)s' % {
                b'len': pack('>H', len(data) + 10),
                b'sum16': sum16,
                b'seqno': pack('>H', self.seqno),
                b'data': data})
 
        # Wrap it in IP
        ip = bytearray(
            b'\x45\xb8%(len)s%(udpseqno)s\x00\x00\xf9\x11%(sum16)s'
            b'\x01\x01\x01\x01\x02\x02\x02\x02%(udp)s' % {
                b'len': pack('>H', len(udp) + 20),
                b'udpseqno': pack('>H', self.udpseqno),
                b'sum16': sum16,
                b'udp': udp})
 
        # Wrap it in Ethernet
        ethernet = bytearray(
            b'\x00\x00\x00\x01\x00\x06\x00\x30\x48\xb1\x1c\x34\x00\x00'
            b'\x08\x00%(ip)s' % {b'ip': ip})
 
        # Wrap it in a pcap packet
        packet = bytearray(b'%(prelude)s%(ethernet)s' % {
            b'prelude': pack(
                '<IIII', int(mktime(self.date.timetuple())),
                self.date.microsecond, len(ethernet), len(ethernet)),
            b'ethernet': ethernet})
 
        # Increase values.
        self.udpseqno += 1
 
        return packet
 
 
class SpandspLog:
    def __init__(self, fp):
        self._fp = fp
 
    def __iter__(self):
        r"""
        Looks for lines line:
 
            [2013-08-07 15:17:34] FAX[23479] res_fax.c: \
              FLOW T.38 Rx     5: IFP c0 01 80 00 00 ff
 
        And yields:
 
            IFP(date=..., seqno=..., data=...)
        """
        prev_seqno = None
 
        for lineno, line in enumerate(self._fp):
            if 'FLOW T.38 Rx' not in line:
                continue
            if 'IFP' not in line:
                continue
 
            match = search(r'(\d{4})-(\d\d)-(\d\d) (\d\d):(\d\d):(\d\d)', line)
            assert match
            date = datetime(*[int(i) for i in match.groups()])
 
            match = search(r'Rx\s*(\d+):', line)
            assert match
            seqno = int(match.groups()[0])
 
            match = search(r': IFP ([0-9a-f ]+)', line)
            assert match
            data = n2b(match.groups()[0])
 
            if prev_seqno is not None:
                # Expected all sequence numbers. But you can safely disable
                # this check.
                assert seqno == prev_seqno + 1, '%s+1 != %s' % (
                    seqno, prev_seqno)
                pass
            prev_seqno = seqno
 
            yield IFP(date=date, seqno=seqno, data=data)
 
 
def main(logname, pcapname):
    with open(sys.argv[1], 'r') as infile:
        log = SpandspLog(infile)
 
        # with open(sys.argv[2], 'xb') as outfile:  # py3 exclusive write, bin
        create_or_fail = os.O_CREAT | os.O_EXCL | os.O_WRONLY
        try:
            fd = os.open(sys.argv[2], create_or_fail, 0o600)
        except Exception:
            raise
        else:
            with os.fdopen(fd, 'wb') as outfile:
                pcap = FaxPcap(outfile)
                for data in log:
                    pcap.add(data)
 
 
if __name__ == '__main__':
    if len(sys.argv) != 3:
        sys.stderr.write('Usage: {} LOGFILE PCAP\n'.format(sys.argv[0]))
        sys.exit(1)
 
    main(sys.argv[1], sys.argv[2])