# Hotspot 2.0 filtering tests
# Copyright (c) 2015, Intel Deutschland GmbH
#
# This software may be distributed under the terms of the BSD license.
# See README for more details.
import hostapd
import hwsim_utils
import socket
import subprocess
import binascii
from utils import HwsimSkip, require_under_vm
import os
import time
from test_ap_hs20 import build_arp, build_na, hs20_ap_params
from test_ap_hs20 import interworking_select, interworking_connect
import struct
import logging
logger = logging.getLogger()
class IPAssign(object):
def __init__(self, iface, addr, ipv6=False):
self._iface = iface
self._addr = addr
self._cmd = ['ip']
if ipv6:
self._cmd.append('-6')
self._cmd.append('addr')
self._ipv6 = ipv6
def __enter__(self):
subprocess.call(self._cmd + ['add', self._addr, 'dev', self._iface])
if self._ipv6:
# wait for DAD to finish
while True:
o = subprocess.check_output(self._cmd + ['show', 'tentative', 'dev', self._iface]).decode()
if self._addr not in o:
break
time.sleep(0.1)
def __exit__(self, type, value, traceback):
subprocess.call(self._cmd + ['del', self._addr, 'dev', self._iface])
def hs20_filters_connect(dev, apdev, disable_dgaf=False, proxy_arp=False):
bssid = apdev[0]['bssid']
params = hs20_ap_params()
params['hessid'] = bssid
# Do not disable dgaf, to test that the station drops unicast IP packets
# encrypted with GTK.
params['disable_dgaf'] = '0'
params['proxy_arp'] = '1'
params['ap_isolate'] = '1'
params['bridge'] = 'ap-br0'
try:
hapd = hostapd.add_ap(apdev[0], params)
except:
# For now, do not report failures due to missing kernel support.
raise HwsimSkip("Could not start hostapd - assume proxyarp not supported in the kernel")
subprocess.call(['brctl', 'setfd', 'ap-br0', '0'])
subprocess.call(['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
dev[0].hs20_enable()
id = dev[0].add_cred_values({'realm': "example.com",
'username': "hs20-test",
'password': "password",
'ca_cert': "auth_serv/ca.pem",
'domain': "example.com",
'update_identifier': "1234"})
interworking_select(dev[0], bssid, "home", freq="2412")
interworking_connect(dev[0], bssid, "TTLS")
time.sleep(0.1)
return dev[0], hapd
def _test_ip4_gtk_drop(devs, apdevs, params, dst):
require_under_vm()
procfile = '/proc/sys/net/ipv4/conf/%s/drop_unicast_in_l2_multicast' % devs[0].ifname
if not os.path.exists(procfile):
raise HwsimSkip("kernel doesn't have capability")
[dev, hapd] = hs20_filters_connect(devs, apdevs)
with IPAssign(dev.ifname, '10.0.0.1/24'):
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.bind(("10.0.0.1", 12345))
s.settimeout(0.1)
pkt = dst
pkt += hapd.own_addr().replace(':', '')
pkt += '0800'
pkt += '45000020786840004011ae600a0000040a000001'
pkt += '30393039000c0000'
pkt += '61736466' # "asdf"
if "OK" not in hapd.request('DATA_TEST_FRAME ' + pkt):
raise Exception("DATA_TEST_FRAME failed")
try:
logger.info(s.recvfrom(1024))
logger.info("procfile=" + procfile + " val=" + open(procfile, 'r').read().rstrip())
raise Exception("erroneously received frame!")
except socket.timeout:
# this is the expected behaviour
pass
def test_ip4_gtk_drop_bcast(devs, apdevs, params):
"""Hotspot 2.0 frame filtering - IPv4 GTK drop broadcast"""
_test_ip4_gtk_drop(devs, apdevs, params, dst='ffffffffffff')
def test_ip4_gtk_drop_mcast(devs, apdevs, params):
"""Hotspot 2.0 frame filtering - IPv4 GTK drop multicast"""
_test_ip4_gtk_drop(devs, apdevs, params, dst='ff0000000000')
def _test_ip6_gtk_drop(devs, apdevs, params, dst):
require_under_vm()
dev = devs[0]
procfile = '/proc/sys/net/ipv6/conf/%s/drop_unicast_in_l2_multicast' % devs[0].ifname
if not os.path.exists(procfile):
raise HwsimSkip("kernel doesn't have capability")
[dev, hapd] = hs20_filters_connect(devs, apdevs)
with IPAssign(dev.ifname, 'fdaa::1/48', ipv6=True):
s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
s.bind(("fdaa::1", 12345))
s.settimeout(0.1)
pkt = dst
pkt += hapd.own_addr().replace(':', '')
pkt += '86dd'
pkt += '60000000000c1140fdaa0000000000000000000000000002fdaa0000000000000000000000000001'
pkt += '30393039000cde31'
pkt += '61736466' # "asdf"
if "OK" not in hapd.request('DATA_TEST_FRAME ' + pkt):
raise Exception("DATA_TEST_FRAME failed")
try:
logger.info(s.recvfrom(1024))
logger.info("procfile=" + procfile + " val=" + open(procfile, 'r').read().rstrip())
raise Exception("erroneously received frame!")
except socket.timeout:
# this is the expected behaviour
pass
def test_ip6_gtk_drop_bcast(devs, apdevs, params):
"""Hotspot 2.0 frame filtering - IPv6 GTK drop broadcast"""
_test_ip6_gtk_drop(devs, apdevs, params, dst='ffffffffffff')
def test_ip6_gtk_drop_mcast(devs, apdevs, params):
"""Hotspot 2.0 frame filtering - IPv6 GTK drop multicast"""
_test_ip6_gtk_drop(devs, apdevs, params, dst='ff0000000000')
def test_ip4_drop_gratuitous_arp(devs, apdevs, params):
"""Hotspot 2.0 frame filtering - IPv4 drop gratuitous ARP"""
require_under_vm()
procfile = '/proc/sys/net/ipv4/conf/%s/drop_gratuitous_arp' % devs[0].ifname
if not os.path.exists(procfile):
raise HwsimSkip("kernel doesn't have capability")
[dev, hapd] = hs20_filters_connect(devs, apdevs)
with IPAssign(dev.ifname, '10.0.0.2/24'):
# add an entry that can be updated by gratuitous ARP
subprocess.call(['ip', 'neigh', 'add', '10.0.0.1', 'lladdr', '02:00:00:00:00:ff', 'nud', 'reachable', 'dev', dev.ifname])
# wait for lock-time
time.sleep(1)
try:
ap_addr = hapd.own_addr()
cl_addr = dev.own_addr()
pkt = build_arp(cl_addr, ap_addr, 2, ap_addr, '10.0.0.1', ap_addr, '10.0.0.1')
pkt = binascii.hexlify(pkt).decode()
if "OK" not in hapd.request('DATA_TEST_FRAME ' + pkt):
raise Exception("DATA_TEST_FRAME failed")
# Wait for frames to be processed
time.sleep(0.1)
if hapd.own_addr() in subprocess.check_output(['ip', 'neigh', 'show']).decode():
raise Exception("gratuitous ARP frame updated erroneously")
finally:
subprocess.call(['ip', 'neigh', 'del', '10.0.0.1', 'dev', dev.ifname])
def test_ip6_drop_unsolicited_na(devs, apdevs, params):
"""Hotspot 2.0 frame filtering - IPv6 drop unsolicited NA"""
require_under_vm()
procfile = '/proc/sys/net/ipv6/conf/%s/drop_unsolicited_na' % devs[0].ifname
if not os.path.exists(procfile):
raise HwsimSkip("kernel doesn't have capability")
[dev, hapd] = hs20_filters_connect(devs, apdevs)
with IPAssign(dev.ifname, 'fdaa::1/48', ipv6=True):
# add an entry that can be updated by unsolicited NA
subprocess.call(['ip', '-6', 'neigh', 'add', 'fdaa::2', 'lladdr', '02:00:00:00:00:ff', 'nud', 'reachable', 'dev', dev.ifname])
try:
ap_addr = hapd.own_addr()
cl_addr = dev.own_addr()
pkt = build_na(ap_addr, 'fdaa::2', 'ff02::1', 'fdaa::2', flags=0x20,
opt=binascii.unhexlify('0201' + ap_addr.replace(':', '')))
pkt = binascii.hexlify(pkt).decode()
if "OK" not in hapd.request('DATA_TEST_FRAME ' + pkt):
raise Exception("DATA_TEST_FRAME failed")
# Wait for frames to be processed
time.sleep(0.1)
if hapd.own_addr() in subprocess.check_output(['ip', 'neigh', 'show']).decode():
raise Exception("unsolicited NA frame updated erroneously")
finally:
subprocess.call(['ip', '-6', 'neigh', 'del', 'fdaa::2', 'dev', dev.ifname])