1#!/usr/bin/env python3
2#
3#   Copyright 2020 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an "AS IS" BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17import logging
18
19from bluetooth_packets_python3 import hci_packets
20from cert.captures import SecurityCaptures
21from cert.closable import Closable
22from cert.closable import safeClose
23from cert.event_stream import EventStream
24from cert.matchers import SecurityMatchers
25from cert.truth import assertThat
26from datetime import timedelta
27from facade import common_pb2 as common
28from google.protobuf import empty_pb2 as empty_proto
29from security.facade_pb2 import IoCapabilityMessage
30from security.facade_pb2 import AuthenticationRequirementsMessage
31from security.facade_pb2 import LeAuthRequirementsMessage
32from security.facade_pb2 import OobDataPresentMessage
33from security.facade_pb2 import UiCallbackMsg
34from security.facade_pb2 import UiCallbackType
35from security.facade_pb2 import HelperMsgType
36
37
38class PyLeSecurity(Closable):
39    """
40        Abstraction for security tasks and GRPC calls
41    """
42
43    _ui_event_stream = None
44    _bond_event_stream = None
45    _helper_event_stream = None
46
47    def __init__(self, device):
48        logging.info("DUT: Init")
49        self._device = device
50        self._device.wait_channel_ready()
51        self._ui_event_stream = EventStream(self._device.security.FetchUiEvents(empty_proto.Empty()))
52        self._bond_event_stream = EventStream(self._device.security.FetchBondEvents(empty_proto.Empty()))
53        self._helper_event_stream = EventStream(self._device.security.FetchHelperEvents(empty_proto.Empty()))
54
55    def get_ui_stream(self):
56        return self._ui_event_stream
57
58    def get_bond_stream(self):
59        return self._bond_event_stream
60
61    def wait_for_ui_event_passkey(self, timeout=timedelta(seconds=3)):
62        display_passkey_capture = SecurityCaptures.DisplayPasskey()
63        assertThat(self._ui_event_stream).emits(display_passkey_capture, timeout=timeout)
64        return display_passkey_capture.get()
65
66    def wait_device_disconnect(self, address):
67        assertThat(self._helper_event_stream).emits(
68            SecurityMatchers.HelperMsg(HelperMsgType.DEVICE_DISCONNECTED, address))
69
70    def SetLeAuthRequirements(self, *args, **kwargs):
71        return self._device.security.SetLeAuthRequirements(LeAuthRequirementsMessage(*args, **kwargs))
72
73    def close(self):
74        if self._ui_event_stream is not None:
75            safeClose(self._ui_event_stream)
76        else:
77            logging.info("DUT: UI Event Stream is None!")
78
79        if self._bond_event_stream is not None:
80            safeClose(self._bond_event_stream)
81        else:
82            logging.info("DUT: Bond Event Stream is None!")
83
84        if self._helper_event_stream is not None:
85            safeClose(self._helper_event_stream)
86        else:
87            logging.info("DUT: Helper Event Stream is None!")
88