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
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.
UsageUsage 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")
WARNING: No route found for IPv6 destination :: (no default route?)
Welcome to Scapy (2.0.1)
>>> lns = LNS()
>>> lns.resultcode = 1
>>> lns.errorcode = 6
>>> lns.errormessage = "Oh, no!"
Received L2TP packet from 220.127.116.11
Got an SCCRQ
Sending spoofed StopCCN from 172.16.0.20 to 18.104.22.168.
Sent 1 packets.
Received L2TP packet from 22.214.171.124
MANDATORY = 32768
HIDDEN = 16384
CONTROL = 32768
L = 16384
S = 2048
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'
# Generates a two byte representation of the provided number
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:
loc = loc + avp_len
interface = "eth1"
resultcode = 0
errorcode = 4
errormessage = "Internal error"
# Define possible states
# Since this is so simple we only need one state :)
# Define transitions
# Transitions from WAIT
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)
elif(packet_type == SCCRP):
print "is an SCCRP"
elif(packet_type == StopCCN):
print "is a StopCCN"
print "is a ZLB or non-control message"