Simulating a broken LNS

A common requirement when testing a LAC is to confirm its reaction when various failure codes are returned by the LNS. In theory you would expect the LAC to react to an LNS failure in the same way (i.e. try another) irrespective of the error type or code returned, but as we all know theory and practice don't always align and that is why we test.

I recently had to prove exactly this area of functionality and found that, while it is relatively easy to put an LNS together which will terminate sessions, it's actually quite hard to get a real LNS to return error messages. Would you believe that they appear to be designed not to fail?

So the aim was:

• To have an 'LNS' which could be configured to reject incoming start control connection requests (SCCRQs)
• To be able to configure the result code, error code and, to make the packet captures easier to read and more authentic, the error message contained within the StopCCN message
• Ideally, to be able to service requests arriving on multiple IP addresses

As usual, the answer to this problem turned out to be scapy.

Important:

The script shown below does exactly what I needed but doesn't exactly work how you might expect. In order to reduce reconfiguration between test cases I have made it respond to queries arriving on any IP address - it does this by inspecting the incoming SCCRQ's source and destination MAC and IP addresses, then flipping them around on the response. That means that it does not attempt to bind to port 1701 on the host, therefore if the LAC sends an SCCRQ to the host's real IP it will get an ICMP unreachable and a StopCCN back. This is almost certainly not what you want.

The intended use case for this script is to have the LAC attempt to connect to an LNS which is "behind" the host running scapy, i.e. the last hop router should have a static route directing traffic for the LNS via the scapy host, in effect creating the following topology:

Alternatively, you could use a static ARP entry on the gateway router to direct traffic for an address on the attached LAN to the scapy host.

Usage

Usage is simple - firstly run scapy, then call 'execfile("BrokenLNS.py")' to load the script. You must create an instance of "LNS" and then, if the defaults to not suit, set the following member values:
 interface (default "eth1")

• resultcode (default 0)
• errorcode (default 4)
• errormessage (default "Internal error")

The script will sit there and close as many sessions as you care to offer it. Press control-C to stop.

Example

root@scapyhost:~/Projects/BrokenLNS# scapy
WARNING: No route found for IPv6 destination :: (no default route?)
Welcome to Scapy (2.0.1)
>>> execfile('BrokenLNS.py')
>>> lns = LNS()
>>> lns.resultcode = 1
>>> lns.errorcode = 6
>>> lns.errormessage = "Oh, no!"
>>> lns.run()
Received L2TP packet from 1.2.3.4
Got an SCCRQ
Sending spoofed StopCCN from 172.16.0.20  to 1.2.3.4.
.
Sent 1 packets.
Received L2TP packet from 1.2.3.4
^C>>>

Code

import os
# Flags
MANDATORY = 32768
HIDDEN = 16384
CONTROL = 32768
L = 16384
S = 2048
# Types
CONTROLMESSAGE = 0
ERRORMESSAGE = 1
PROTOCOLVERSION = 2
HOSTNAME = 7
RECVWIN = 10
FRAMING = 3
BEARER = 4
FIRMWARE = 6
TUNNELID = 9
CHALLENGE = 11
# Control Message Types
SCCRQ = '\x00\x01'
SCCRP = '\x00\x02'
StopCCN = '\x00\x04'
def word(value):
# Generates a two byte representation of the provided number
  return(chr((value/256)%256)+chr(value%256))
def AVP(bitmask, vendor, attribute_type, data):
# Generates an L2TP AVP using the given attribute number and payload
  length = len(data) + 6
  return(word(bitmask + (length % 1024)) + word(0) + word(attribute_type) + data)
def genL2TP(flags, tunid, sessid, ns, nr, payload):
# Generates an L2TP payload with the given parameters and AVP payload
  length = len(payload) + 12
  return(word(flags | 2) + word(length) + tunid + sessid + word(ns) + word(nr) + payload)
def getAVP(avp, payload):
  loc = 0
  while(loc < len(payload)):
    avp_type = payload[loc+2:loc+6]
    avp_len = ((ord(payload[loc:loc+1]) & 3) * 256) + ord(payload[loc+1:loc+2])
    # Uncomment the following line if you want to see info on every AVP checked
#    print "Got AVP " + str(ord(avp_type[0:1])).zfill(2) + str(ord(avp_type[1:2])).zfill(2)  + str(ord(avp_type[2:3])).zfill(2) + str(ord(avp_type[3:4])).zfill(2) + " of length " + str(avp_len) + " value " + payload[loc+6:loc+avp_len]
    if avp_type == avp:
      return(payload[loc+6:loc+avp_len])
    loc = loc + avp_len
class LNS(Automaton):
  interface = "eth1"
  resultcode = 0
  errorcode = 4
  errormessage = "Internal error"
# Define possible states
# Since this is so simple we only need one state :)
  @ATMT.state(initial=1)
  def WAIT(self):
    pass
# Define transitions
# Transitions from WAIT
  @ATMT.receive_condition(WAIT)
  def receive_sccrq(self,pkt):
    if (UDP in pkt) and pkt.dport==1701:
      print "Received L2TP packet from " + pkt[IP].src
      # scapy's built in L2TP handling doesn't deal well with control messages so
      # we just grab the raw data from beyond the UDP header
      payload = pkt[UDP].build_payload()
      # Check what type of L2TP message arrived by chopping off the header and passing
      # the rest to getAVP
      packet_type = getAVP(word(0) + word(CONTROLMESSAGE), payload[12:])
      if(packet_type == SCCRQ):
        # If we get an SCCRQ, generate a StopCCN in response.
        print "Got an SCCRQ"
        client_ip = pkt[IP].src
        server_ip = pkt[IP].dst
        client_mac = pkt[Ether].src
        server_mac = pkt[Ether].dst
        tun_id = getAVP(word(0) + word(TUNNELID), payload[12:])
        print "Sending spoofed StopCCN from " + server_ip + "  to " + client_ip + "."
        sendp(Ether(src=server_mac, dst=client_mac)/IP(src=server_ip, dst=client_ip)/UDP(sport=1701, dport=1701)/Raw(load=genL2TP(CONTROL | L | S, tun_id, word(0), 0, 1, AVP(MANDATORY, 0, CONTROLMESSAGE, StopCCN) + AVP(MANDATORY, 0, ERRORMESSAGE, word(self.resultcode) + word(self.errorcode) + self.errormessage) + AVP(MANDATORY, 0, TUNNELID, word(12345)))), iface=self.interface)
        raise self.WAIT()
      elif(packet_type == SCCRP):
        print "is an SCCRP"
      elif(packet_type == StopCCN):
        print "is a StopCCN"
      else:
        print "is a ZLB or non-control message"