Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit dc517e36 authored by Treehugger Robot's avatar Treehugger Robot Committed by Automerger Merge Worker
Browse files

Merge "LeL2cap: Add security check for connection request" am: 03d12cf9 

Change-Id: Iee80c70e8e34ee6c32caf0c9da91371c9c57fcb9
parents b789d9fe 03d12cf9

system/gd/cert/py_l2cap.py

+6 −2
Original line number Diff line number Diff line
@@ -18,6 +18,7 @@ from google.protobuf import empty_pb2 as empty_proto 


from l2cap.classic import facade_pb2 as l2cap_facade_pb2

from l2cap.le import facade_pb2 as l2cap_le_facade_pb2

from l2cap.le.facade_pb2 import SecurityLevel

from bluetooth_packets_python3 import l2cap_packets

from cert.event_stream import FilteringEventStream

from cert.event_stream import EventStream, IEventStream
@@ -203,10 +204,13 @@ class PyLeL2cap(Closable): 
    def get_fixed_channel(self, cid=4):

        return PyLeL2capFixedChannel(self._device, cid, self._le_l2cap_stream)



    def register_coc(self, cert_address, psm=0x33):

    def register_coc(self,

                     cert_address,

                     psm=0x33,

                     security_level=SecurityLevel.NO_SECURITY):

        self._device.l2cap_le.SetDynamicChannel(

            l2cap_le_facade_pb2.SetEnableDynamicChannelRequest(

                psm=psm, enable=True))

                psm=psm, enable=True, security_level=security_level))

        return PyLeL2capDynamicChannel(self._device, cert_address, psm,

                                       self._le_l2cap_stream)

system/gd/l2cap/le/cert/cert_le_l2cap.py

+10 −0
Original line number Diff line number Diff line
@@ -158,6 +158,16 @@ class CertLeL2cap(Closable): 
        self._cid_to_cert_channels[scid] = channel

        return channel



    def open_channel_with_expected_result(

            self, psm=0x33,

            result=LeCreditBasedConnectionResponseResult.SUCCESS):

        self.control_channel.send(

            l2cap_packets.LeCreditBasedConnectionRequestBuilder(

                1, psm, 0x40, 1000, 100, 6))



        response = L2capMatchers.CreditBasedConnectionResponse(result)

        assertThat(self.control_channel).emits(response)



    def verify_and_respond_open_channel_from_remote(

            self,

            psm=0x33,

system/gd/l2cap/le/cert/le_l2cap_test.py

+83 −16
Original line number Diff line number Diff line
@@ -25,6 +25,7 @@ import bluetooth_packets_python3 as bt_packets 
from bluetooth_packets_python3 import hci_packets, l2cap_packets

from bluetooth_packets_python3.l2cap_packets import LeCreditBasedConnectionResponseResult

from l2cap.le.cert.cert_le_l2cap import CertLeL2cap

from l2cap.le.facade_pb2 import SecurityLevel



# Assemble a sample packet.

SAMPLE_PACKET = bt_packets.RawBuilder([0x19, 0x26, 0x08, 0x17])
@@ -41,12 +42,15 @@ class LeL2capTest(GdBaseTestClass): 
        self.dut_l2cap = PyLeL2cap(self.dut)

        self.cert_l2cap = CertLeL2cap(self.cert)

        self.dut_address = common.BluetoothAddressWithType(

            address = common.BluetoothAddress(address=bytes(b'0D:05:04:03:02:01')),

            address=common.BluetoothAddress(

                address=bytes(b'0D:05:04:03:02:01')),

            type=common.RANDOM_DEVICE_ADDRESS)

        self.cert_address = common.BluetoothAddressWithType(

            address = common.BluetoothAddress(address=bytes(b'55:11:FF:AA:33:22')),

            address=common.BluetoothAddress(

                address=bytes(b'55:11:FF:AA:33:22')),

            type=common.RANDOM_DEVICE_ADDRESS)

        self.cert_l2cap._device.hci_le_acl_manager.SetInitiatorAddress(self.cert_address)

        self.cert_l2cap._device.hci_le_acl_manager.SetInitiatorAddress(

            self.cert_address)



    def teardown_test(self):

        self.cert_l2cap.close()
@@ -68,9 +72,11 @@ class LeL2capTest(GdBaseTestClass): 
            event_type=le_advertising_facade.AdvertisingEventType.ADV_IND,

            address_type=common.RANDOM_DEVICE_ADDRESS,

            peer_address_type=common.PUBLIC_DEVICE_ADDRESS,

            peer_address=common.BluetoothAddress(address=bytes(b'00:00:00:00:00:00')),

            peer_address=common.BluetoothAddress(

                address=bytes(b'00:00:00:00:00:00')),

            channel_map=7,

            filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES)

            filter_policy=le_advertising_facade.AdvertisingFilterPolicy.

            ALL_DEVICES)

        request = le_advertising_facade.CreateAdvertiserRequest(config=config)

        create_response = self.dut.hci_le_advertising_manager.CreateAdvertiser(

            request)
@@ -97,13 +103,16 @@ class LeL2capTest(GdBaseTestClass): 
            event_type=le_advertising_facade.AdvertisingEventType.ADV_IND,

            address_type=common.RANDOM_DEVICE_ADDRESS,

            peer_address_type=common.PUBLIC_DEVICE_ADDRESS,

            peer_address=common.BluetoothAddress(address=bytes(b'00:00:00:00:00:00')),

            peer_address=common.BluetoothAddress(

                address=bytes(b'00:00:00:00:00:00')),

            channel_map=7,

            filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES)

            filter_policy=le_advertising_facade.AdvertisingFilterPolicy.

            ALL_DEVICES)

        request = le_advertising_facade.CreateAdvertiserRequest(config=config)

        create_response = self.cert.hci_le_advertising_manager.CreateAdvertiser(

            request)

        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm)

        response_future = self.dut_l2cap.connect_coc_to_cert(

            self.cert_address, psm)

        self.cert_l2cap.wait_for_connection()

        # TODO: Currently we can only connect by using Dynamic channel API. Use fixed channel instead.

        cert_channel = self.cert_l2cap.verify_and_respond_open_channel_from_remote(
@@ -126,7 +135,8 @@ class LeL2capTest(GdBaseTestClass): 
        return (dut_channel, cert_channel)



    def _open_channel_from_dut(self, psm=0x33):

        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm)

        response_future = self.dut_l2cap.connect_coc_to_cert(

            self.cert_address, psm)

        cert_channel = self.cert_l2cap.verify_and_respond_open_channel_from_remote(

            psm)

        dut_channel = response_future.get_channel()
@@ -358,7 +368,8 @@ class LeL2capTest(GdBaseTestClass): 
        Verify that an IUT sending an LE Credit Based Connection Request to a legacy peer and receiving a Command Reject does not establish the channel.

        """

        self._setup_link_from_cert()

        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)

        response_future = self.dut_l2cap.connect_coc_to_cert(

            self.cert_address, psm=0x33)

        self.cert_l2cap.verify_and_reject_open_channel_from_remote(psm=0x33)

        assertThat(response_future.get_status()).isNotEqualTo(

            LeCreditBasedConnectionResponseResult.SUCCESS)
@@ -398,7 +409,8 @@ class LeL2capTest(GdBaseTestClass): 
        Verify that an IUT sending an LE Credit Based Connection Request on an unsupported LE_PSM will not establish a channel upon receiving an LE Credit Based Connection Response refusing the connection.

        """

        self._setup_link_from_cert()

        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)

        response_future = self.dut_l2cap.connect_coc_to_cert(

            self.cert_address, psm=0x33)

        self.cert_l2cap.verify_and_respond_open_channel_from_remote(

            psm=0x33,

            result=LeCreditBasedConnectionResponseResult.LE_PSM_NOT_SUPPORTED)
@@ -493,7 +505,8 @@ class LeL2capTest(GdBaseTestClass): 
        Verify that the IUT does not establish the channel upon receipt of an LE Credit Based Connection Response indicating the connection was refused with Result “0x0005 – Connection Refused – Insufficient Authentication".

        """

        self._setup_link_from_cert()

        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)

        response_future = self.dut_l2cap.connect_coc_to_cert(

            self.cert_address, psm=0x33)

        self.cert_l2cap.verify_and_respond_open_channel_from_remote(

            psm=0x33,

            result=LeCreditBasedConnectionResponseResult.
@@ -501,6 +514,23 @@ class LeL2capTest(GdBaseTestClass): 
        assertThat(response_future.get_status()).isEqualTo(

            LeCreditBasedConnectionResponseResult.INSUFFICIENT_AUTHENTICATION)



    @metadata(

        pts_test_id="L2CAP/LE/CFC/BV-11-C",

        pts_test_name="Security - Insufficient Authentication – Responder")

    def test_security_insufficient_authentication_responder(self):

        """

        Verify that an IUT refuses to create a connection upon reception of an LE Credit Based Connection

Request which fails to satisfy authentication requirements.

        """

        self._setup_link_from_cert()

        psm = 0x33

        self.dut_l2cap.register_coc(

            self.cert_address, psm,

            SecurityLevel.AUTHENTICATED_PAIRING_WITH_ENCRYPTION)

        self.cert_l2cap.open_channel_with_expected_result(

            psm,

            LeCreditBasedConnectionResponseResult.INSUFFICIENT_AUTHENTICATION)



    @metadata(

        pts_test_id="L2CAP/LE/CFC/BV-12-C",

        pts_test_name="Security - Insufficient Authorization – Initiator")
@@ -509,7 +539,8 @@ class LeL2capTest(GdBaseTestClass): 
        Verify that the IUT does not establish the channel upon receipt of an LE Credit Based Connection Response indicating the connection was refused with Result “0x0006 – Connection Refused – Insufficient Authorization”.

        """

        self._setup_link_from_cert()

        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)

        response_future = self.dut_l2cap.connect_coc_to_cert(

            self.cert_address, psm=0x33)

        self.cert_l2cap.verify_and_respond_open_channel_from_remote(

            psm=0x33,

            result=LeCreditBasedConnectionResponseResult.
@@ -517,6 +548,22 @@ class LeL2capTest(GdBaseTestClass): 
        assertThat(response_future.get_status()).isEqualTo(

            LeCreditBasedConnectionResponseResult.INSUFFICIENT_AUTHORIZATION)



    @metadata(

        pts_test_id="L2CAP/LE/CFC/BV-13-C",

        pts_test_name="Security - Insufficient Authorization – Responder")

    def test_security_insufficient_authorization_responder(self):

        """

        Verify that an IUT refuses to create a connection upon reception of an LE Credit Based Connection

        Request which fails to satisfy authentication requirements.

        """

        self._setup_link_from_cert()

        psm = 0x33

        self.dut_l2cap.register_coc(self.cert_address, psm,

                                    SecurityLevel.AUTHORIZATION)

        self.cert_l2cap.open_channel_with_expected_result(

            psm,

            LeCreditBasedConnectionResponseResult.INSUFFICIENT_AUTHORIZATION)



    @metadata(

        pts_test_id="L2CAP/LE/CFC/BV-14-C",

        pts_test_name="Security - Insufficient Key Size – Initiator")
@@ -538,6 +585,23 @@ class LeL2capTest(GdBaseTestClass): 
            LeCreditBasedConnectionResponseResult.

            INSUFFICIENT_ENCRYPTION_KEY_SIZE)



    @metadata(

        pts_test_id="L2CAP/LE/CFC/BV-15-C",

        pts_test_name="Security - Insufficient Encryption Key Size – Responder")

    def test_security_insufficient_encryption_key_size_responder(self):

        """

        Verify that an IUT refuses to create a connection upon receipt of an LE Credit Based Connection

        Request which fails to satisfy Encryption Key Size requirements.

        """

        self._setup_link_from_cert()

        psm = 0x33

        self.dut_l2cap.register_coc(

            self.cert_address, psm,

            SecurityLevel.AUTHENTICATED_PAIRING_WITH_128_BIT_KEY)

        self.cert_l2cap.open_channel_with_expected_result(

            psm, LeCreditBasedConnectionResponseResult.

            INSUFFICIENT_ENCRYPTION_KEY_SIZE)



    @metadata(

        pts_test_id="L2CAP/LE/CFC/BV-16-C",

        pts_test_name=
@@ -569,7 +633,8 @@ class LeL2capTest(GdBaseTestClass): 
        Verify that an IUT sending an LE Credit Based Connection Request does not establish the channel upon receiving an LE Credit Based Connection Response refusing the connection with result "0x0009 – Connection refused – Invalid Source CID".

        """

        self._setup_link_from_cert()

        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)

        response_future = self.dut_l2cap.connect_coc_to_cert(

            self.cert_address, psm=0x33)

        self.cert_l2cap.verify_and_respond_open_channel_from_remote(

            psm=0x33,

            result=LeCreditBasedConnectionResponseResult.INVALID_SOURCE_CID)
@@ -587,7 +652,8 @@ class LeL2capTest(GdBaseTestClass): 
        Verify that an IUT sending an LE Credit Based Connection Request does not establish the channel upon receiving an LE Credit Based Connection Response refusing the connection with result "0x000A – Connection refused – Source CID already allocated".

        """

        self._setup_link_from_cert()

        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)

        response_future = self.dut_l2cap.connect_coc_to_cert(

            self.cert_address, psm=0x33)

        self.cert_l2cap.verify_and_respond_open_channel_from_remote(

            psm=0x33,

            result=LeCreditBasedConnectionResponseResult.
@@ -625,7 +691,8 @@ class LeL2capTest(GdBaseTestClass): 
        Verify that an IUT sending an LE Credit Based Connection Request does not establish the channel upon receiving an LE Credit Based Connection Response refusing the connection with result "0x000B – Connection refused – Unacceptable Parameters".

        """

        self._setup_link_from_cert()

        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)

        response_future = self.dut_l2cap.connect_coc_to_cert(

            self.cert_address, psm=0x33)

        self.cert_l2cap.verify_and_respond_open_channel_from_remote(

            psm=0x33,

            result=LeCreditBasedConnectionResponseResult.UNACCEPTABLE_PARAMETERS

system/gd/l2cap/le/dynamic_channel_manager.cc

+1 −0
Original line number Diff line number Diff line
@@ -47,6 +47,7 @@ bool DynamicChannelManager::RegisterService(Psm psm, DynamicChannelConfiguration 
                                            OnConnectionOpenCallback on_connection_open, os::Handler* handler) {

  internal::DynamicChannelServiceImpl::PendingRegistration pending_registration{

      .user_handler_ = handler,

      .security_policy_ = security_policy,

      .on_registration_complete_callback_ = std::move(on_registration_complete),

      .on_connection_open_callback_ = std::move(on_connection_open),

      .configuration_ = configuration_option,

system/gd/l2cap/le/facade.cc

+6 −4
Original line number Diff line number Diff line
@@ -81,8 +81,10 @@ class L2capLeModuleFacadeService : public L2capLeModuleFacade::Service { 
                                   const ::bluetooth::l2cap::le::SetEnableDynamicChannelRequest* request,

                                   ::google::protobuf::Empty* response) override {

    if (request->enable()) {

      dynamic_channel_helper_map_.emplace(request->psm(), std::make_unique<L2capDynamicChannelHelper>(

                                                              this, l2cap_layer_, facade_handler_, request->psm()));

      dynamic_channel_helper_map_.emplace(

          request->psm(),

          std::make_unique<L2capDynamicChannelHelper>(this, l2cap_layer_, facade_handler_, request->psm(),

                                                      static_cast<SecurityPolicy::Level>(request->security_level())));

      return ::grpc::Status::OK;

    } else {

      auto service_helper = dynamic_channel_helper_map_.find(request->psm());
@@ -111,11 +113,11 @@ class L2capLeModuleFacadeService : public L2capLeModuleFacade::Service { 
  class L2capDynamicChannelHelper {

   public:

    L2capDynamicChannelHelper(L2capLeModuleFacadeService* service, L2capLeModule* l2cap_layer, os::Handler* handler,

                              Psm psm)

                              Psm psm, SecurityPolicy::Level security_level)

        : facade_service_(service), l2cap_layer_(l2cap_layer), handler_(handler), psm_(psm) {

      dynamic_channel_manager_ = l2cap_layer_->GetDynamicChannelManager();

      dynamic_channel_manager_->RegisterService(

          psm, {}, {},

          psm, {}, security_level,

          common::BindOnce(&L2capDynamicChannelHelper::on_l2cap_service_registration_complete,

                           common::Unretained(this)),

          common::Bind(&L2capDynamicChannelHelper::on_connection_open, common::Unretained(this)), handler_);