libeap/tests/hwsim/test_p2p_invitation.py

   1 # P2P invitation test cases
   2 # Copyright (c) 2013, Jouni Malinen <j@w1.fi>
   3 #
   4 # This software may be distributed under the terms of the BSD license.
   5 # See README for more details.
   6
   7 import logging
   8 logger = logging.getLogger()
   9
  10 import hwsim_utils
  11
  12 def test_p2p_go_invite(dev):
  13     """P2P GO inviting a client to join"""
  14     addr0 = dev[0].p2p_dev_addr()
  15     addr1 = dev[1].p2p_dev_addr()
  16
  17     logger.info("Generate BSS table entry for old group")
  18     # this adds more coverage to testing by forcing the GO to be found with an
  19     # older entry in the BSS table and with that entry having a different
  20     # operating channel.
  21     dev[0].p2p_start_go(freq=2422)
  22     dev[1].scan()
  23     dev[0].remove_group()
  24
  25     logger.info("Discover peer")
  26     dev[1].p2p_listen()
  27     if not dev[0].discover_peer(addr1, social=True):
  28         raise Exception("Peer " + addr1 + " not found")
  29
  30     logger.info("Start GO on non-social channel")
  31     res = dev[0].p2p_start_go(freq=2417)
  32     logger.debug("res: " + str(res))
  33
  34     logger.info("Invite peer to join the group")
  35     dev[0].global_request("P2P_INVITE group=" + dev[0].group_ifname + " peer=" + addr1)
  36     ev = dev[1].wait_global_event(["P2P-INVITATION-RECEIVED"], timeout=10)
  37     if ev is None:
  38         raise Exception("Timeout on invitation on peer")
  39     ev = dev[0].wait_global_event(["P2P-INVITATION-RESULT"], timeout=10)
  40     if ev is None:
  41         raise Exception("Timeout on invitation on GO")
  42     if "status=1" not in ev:
  43         raise Exception("Unexpected invitation result")
  44
  45     logger.info("Join the group")
  46     pin = dev[1].wps_read_pin()
  47     dev[0].p2p_go_authorize_client(pin)
  48     dev[1].p2p_connect_group(addr0, pin, timeout=60)
  49     logger.info("Client connected")
  50     hwsim_utils.test_connectivity_p2p(dev[0], dev[1])
  51
  52     logger.info("Terminate group")
  53     dev[0].remove_group()
  54     dev[1].wait_go_ending_session()
  55
  56 def test_p2p_go_invite_auth(dev):
  57     """P2P GO inviting a client to join (authorized invitation)"""
  58     addr0 = dev[0].p2p_dev_addr()
  59     addr1 = dev[1].p2p_dev_addr()
  60
  61     logger.info("Generate BSS table entry for old group")
  62     # this adds more coverage to testing by forcing the GO to be found with an
  63     # older entry in the BSS table and with that entry having a different
  64     # operating channel.
  65     dev[0].p2p_start_go(freq=2432)
  66     dev[1].scan()
  67     dev[0].remove_group()
  68     dev[0].dump_monitor()
  69     dev[1].dump_monitor()
  70
  71     logger.info("Discover peer")
  72     dev[1].p2p_listen()
  73     if not dev[0].discover_peer(addr1, social=True):
  74         raise Exception("Peer " + addr1 + " not found")
  75     dev[0].p2p_listen()
  76     if not dev[1].discover_peer(addr0, social=True):
  77         raise Exception("Peer " + addr0 + " not found")
  78     dev[1].p2p_listen()
  79
  80     logger.info("Authorize invitation")
  81     pin = dev[1].wps_read_pin()
  82     dev[1].global_request("P2P_CONNECT " + addr0 + " " + pin + " join auth")
  83
  84     logger.info("Start GO on non-social channel")
  85     res = dev[0].p2p_start_go(freq=2427)
  86     logger.debug("res: " + str(res))
  87
  88     logger.info("Invite peer to join the group")
  89     dev[0].p2p_go_authorize_client(pin)
  90     dev[0].global_request("P2P_INVITE group=" + dev[0].group_ifname + " peer=" + addr1)
  91     ev = dev[1].wait_global_event(["P2P-INVITATION-RECEIVED",
  92                                    "P2P-GROUP-STARTED"], timeout=20)
  93     if ev is None:
  94         raise Exception("Timeout on invitation on peer")
  95     if "P2P-INVITATION-RECEIVED" in ev:
  96         raise Exception("Unexpected request to accept pre-authorized invitaton")
  97     dev[1].group_form_result(ev)
  98     dev[0].dump_monitor()
  99
 100     logger.info("Client connected")
 101     hwsim_utils.test_connectivity_p2p(dev[0], dev[1])
 102
 103     logger.info("Terminate group")
 104     dev[0].remove_group()
 105     dev[1].wait_go_ending_session()
 106
 107 def test_p2p_go_invite_unknown(dev):
 108     """P2P GO inviting a client that has not discovered the GO"""
 109     try:
 110         addr0 = dev[0].p2p_dev_addr()
 111         addr1 = dev[1].p2p_dev_addr()
 112
 113         dev[1].p2p_listen()
 114         if not dev[0].discover_peer(addr1, social=True):
 115             raise Exception("Peer " + addr1 + " not found")
 116         dev[1].global_request("P2P_FLUSH")
 117         dev[1].p2p_listen()
 118
 119         dev[0].p2p_start_go(freq=2412)
 120
 121         logger.info("Invite peer to join the group")
 122         # Prevent peer entry from being added for testing coverage
 123         if "OK" not in dev[1].global_request("P2P_SET peer_filter 00:11:22:33:44:55"):
 124             raise Exception("Failed to set peer_filter")
 125         dev[0].p2p_go_authorize_client("12345670")
 126         dev[0].global_request("P2P_INVITE group=" + dev[0].group_ifname + " peer=" + addr1)
 127         ev = dev[1].wait_global_event(["P2P-INVITATION-RECEIVED"], timeout=15)
 128         if ev is None:
 129             raise Exception("Invitation Request not received")
 130         ev = dev[0].wait_global_event(["P2P-INVITATION-RESULT"], timeout=15)
 131         if ev is None:
 132             raise Exception("Invitation Response not received")
 133         if "status=1" not in ev:
 134             raise Exception("Unexpected invitation result: " + ev)
 135     finally:
 136         dev[1].global_request("P2P_SET peer_filter 00:00:00:00:00:00")
 137
 138 def test_p2p_cli_invite(dev):
 139     """P2P Client inviting a device to join"""
 140     addr0 = dev[0].p2p_dev_addr()
 141     addr1 = dev[1].p2p_dev_addr()
 142     addr2 = dev[2].p2p_dev_addr()
 143
 144     dev[0].p2p_start_go(freq=2412)
 145     pin = dev[1].wps_read_pin()
 146     dev[0].p2p_go_authorize_client(pin)
 147     dev[1].p2p_connect_group(addr0, pin, timeout=60)
 148
 149     dev[2].p2p_listen()
 150     if not dev[1].discover_peer(addr2, social=True):
 151         raise Exception("Peer " + addr2 + " not found")
 152
 153     if "OK" not in dev[1].global_request("P2P_INVITE group=" + dev[1].group_ifname + " peer=" + addr2):
 154         raise Exception("Unexpected failure of P2P_INVITE to known peer")
 155     ev = dev[2].wait_global_event(["P2P-INVITATION-RECEIVED"], timeout=10)
 156     if ev is None:
 157         raise Exception("Timeout on invitation invited peer")
 158     if "sa=" + addr1 not in ev:
 159         raise Exception("Incorrect source address")
 160     if "go_dev_addr=" + addr0 not in ev:
 161         raise Exception("Incorrect GO address")
 162     ev = dev[1].wait_global_event(["P2P-INVITATION-RESULT"], timeout=10)
 163     if ev is None:
 164         raise Exception("Timeout on invitation on inviting client")
 165     if "status=1" not in ev:
 166         raise Exception("Unexpected invitation result")
 167
 168     pin = dev[2].wps_read_pin()
 169     dev[0].p2p_go_authorize_client(pin)
 170     dev[2].p2p_connect_group(addr0, pin, timeout=60)
 171
 172     if "FAIL" not in dev[1].global_request("P2P_INVITE group=" + dev[1].group_ifname + " peer=00:11:22:33:44:55"):
 173         raise Exception("Unexpected success of P2P_INVITE to unknown peer")
 174
 175     dev[0].remove_group()
 176     dev[1].wait_go_ending_session()
 177     dev[2].wait_go_ending_session()
 178
 179 def test_p2p_invite_invalid(dev):
 180     """Invalid parameters to P2P_INVITE"""
 181     id = dev[0].add_network()
 182     for cmd in [ "foo=bar",
 183                  "persistent=123 peer=foo",
 184                  "persistent=123",
 185                  "persistent=%d" % id,
 186                  "group=foo",
 187                  "group=foo peer=foo",
 188                  "group=foo peer=00:11:22:33:44:55 go_dev_addr=foo" ]:
 189         if "FAIL" not in dev[0].request("P2P_INVITE " + cmd):
 190             raise Exception("Invalid P2P_INVITE accepted: " + cmd)