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