Merge pull request #11367 from paulozulato/feat-gcp-devices

feat(gcp-iot): port GCP IoT Core compatibility layer from e4.4

apps/emqx_authn/src/emqx_authn_enterprise.erl

@@ -9,7 +9,10 @@
 -if(?EMQX_RELEASE_EDITION == ee).
 
 providers() ->
-    [{{password_based, ldap}, emqx_ldap_authn}].
+    [
+        {{password_based, ldap}, emqx_ldap_authn},
+        {gcp_device, emqx_gcp_device_authn}
+    ].
 
 resource_provider() ->
     [emqx_ldap_authn].

apps/emqx_gcp_device/BSL.txt

@@ -0,0 +1,94 @@
+Business Source License 1.1
+
+Licensor:             Hangzhou EMQ Technologies Co., Ltd.
+Licensed Work:        EMQX Enterprise Edition
+                      The Licensed Work is (c) 2023
+                      Hangzhou EMQ Technologies Co., Ltd.
+Additional Use Grant: Students and educators are granted right to copy,
+                      modify, and create derivative work for research
+                      or education.
+Change Date:          2027-02-01
+Change License:       Apache License, Version 2.0
+
+For information about alternative licensing arrangements for the Software,
+please contact Licensor: https://www.emqx.com/en/contact
+
+Notice
+
+The Business Source License (this document, or the “License”) is not an Open
+Source license. However, the Licensed Work will eventually be made available
+under an Open Source License, as stated in this License.
+
+License text copyright (c) 2017 MariaDB Corporation Ab, All Rights Reserved.
+“Business Source License” is a trademark of MariaDB Corporation Ab.
+
+-----------------------------------------------------------------------------
+
+Business Source License 1.1
+
+Terms
+
+The Licensor hereby grants you the right to copy, modify, create derivative
+works, redistribute, and make non-production use of the Licensed Work. The
+Licensor may make an Additional Use Grant, above, permitting limited
+production use.
+
+Effective on the Change Date, or the fourth anniversary of the first publicly
+available distribution of a specific version of the Licensed Work under this
+License, whichever comes first, the Licensor hereby grants you rights under
+the terms of the Change License, and the rights granted in the paragraph
+above terminate.
+
+If your use of the Licensed Work does not comply with the requirements
+currently in effect as described in this License, you must purchase a
+commercial license from the Licensor, its affiliated entities, or authorized
+resellers, or you must refrain from using the Licensed Work.
+
+All copies of the original and modified Licensed Work, and derivative works
+of the Licensed Work, are subject to this License. This License applies
+separately for each version of the Licensed Work and the Change Date may vary
+for each version of the Licensed Work released by Licensor.
+
+You must conspicuously display this License on each original or modified copy
+of the Licensed Work. If you receive the Licensed Work in original or
+modified form from a third party, the terms and conditions set forth in this
+License apply to your use of that work.
+
+Any use of the Licensed Work in violation of this License will automatically
+terminate your rights under this License for the current and all other
+versions of the Licensed Work.
+
+This License does not grant you any right in any trademark or logo of
+Licensor or its affiliates (provided that you may use a trademark or logo of
+Licensor as expressly required by this License).
+
+TO THE EXTENT PERMITTED BY APPLICABLE LAW, THE LICENSED WORK IS PROVIDED ON
+AN “AS IS” BASIS. LICENSOR HEREBY DISCLAIMS ALL WARRANTIES AND CONDITIONS,
+EXPRESS OR IMPLIED, INCLUDING (WITHOUT LIMITATION) WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, NON-INFRINGEMENT, AND
+TITLE.
+
+MariaDB hereby grants you permission to use this License’s text to license
+your works, and to refer to it using the trademark “Business Source License”,
+as long as you comply with the Covenants of Licensor below.
+
+Covenants of Licensor
+
+In consideration of the right to use this License’s text and the “Business
+Source License” name and trademark, Licensor covenants to MariaDB, and to all
+other recipients of the licensed work to be provided by Licensor:
+
+1. To specify as the Change License the GPL Version 2.0 or any later version,
+   or a license that is compatible with GPL Version 2.0 or a later version,
+   where “compatible” means that software provided under the Change License can
+   be included in a program with software provided under GPL Version 2.0 or a
+   later version. Licensor may specify additional Change Licenses without
+   limitation.
+
+2. To either: (a) specify an additional grant of rights to use that does not
+   impose any additional restriction on the right granted in this License, as
+   the Additional Use Grant; or (b) insert the text “None”.
+
+3. To specify a Change Date.
+
+4. Not to modify this License in any other way.

apps/emqx_gcp_device/README.md

@@ -0,0 +1,7 @@
+# emqx_gcp_device
+
+An application for simplified migration from Google IoT Core.
+
+It implements import of IoT Core device config and authentication data,
+so that end devices can authenticate and obtain config as usual.
+

apps/emqx_gcp_device/rebar.config

@@ -0,0 +1,6 @@
+{erl_opts, [debug_info]}.
+{deps, [
+    {emqx, {path, "../emqx"}},
+    {emqx_utils, {path, "../emqx_utils"}},
+    {emqx_authn, {path, "../emqx_authn"}}
+]}.

apps/emqx_gcp_device/src/emqx_gcp_device.app.src

@@ -0,0 +1,15 @@
+{application, emqx_gcp_device, [
+    {description, "Application simplifying migration from GCP IoT Core"},
+    {vsn, "0.1.0"},
+    {registered, []},
+    {mod, {emqx_gcp_device_app, []}},
+    {applications, [
+        kernel,
+        stdlib,
+        emqx_authn
+    ]},
+    {env, []},
+    {modules, []},
+
+    {links, []}
+]}.

apps/emqx_gcp_device/src/emqx_gcp_device.appup.src

@@ -0,0 +1,9 @@
+%% -*- mode: erlang -*-
+{VSN,
+ [ {<<".*">>,
+    []}
+ ],
+ [ {<<".*">>,
+    []}
+ ]
+}.

apps/emqx_gcp_device/src/emqx_gcp_device.erl

@@ -0,0 +1,268 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_gcp_device).
+
+-include_lib("emqx_authn/include/emqx_authn.hrl").
+-include_lib("emqx/include/emqx.hrl").
+-include_lib("emqx/include/logger.hrl").
+-include_lib("stdlib/include/ms_transform.hrl").
+
+%% Management
+-export([put_device/1, get_device/1, remove_device/1]).
+%% Management: import
+-export([import_devices/1]).
+%% Authentication
+-export([get_device_actual_keys/1]).
+%% Internal API
+-export([create_table/0, clear_table/0, format_device/1]).
+
+-ifdef(TEST).
+-export([config_topic/1]).
+% to avoid test flakiness
+-define(ACTIVITY, sync_dirty).
+-else.
+-define(ACTIVITY, async_dirty).
+-endif.
+
+-type deviceid() :: binary().
+-type project() :: binary().
+-type location() :: binary().
+-type registry() :: binary().
+-type device_loc() :: {project(), location(), registry()}.
+-type key_type() :: binary().
+-type key() :: binary().
+-type expires_at() :: pos_integer().
+-type key_record() :: {key_type(), key(), expires_at()}.
+-type created_at() :: pos_integer().
+-type extra() :: map().
+
+-record(emqx_gcp_device, {
+    id :: deviceid(),
+    keys :: [key_record()],
+    device_loc :: device_loc(),
+    created_at :: created_at(),
+    extra :: extra()
+}).
+-type emqx_gcp_device() :: #emqx_gcp_device{}.
+
+-type formatted_key() ::
+    #{
+        key_type := key_type(),
+        key := key(),
+        expires_at := expires_at()
+    }.
+-type encoded_config() :: binary().
+-type formatted_device() ::
+    #{
+        deviceid := deviceid(),
+        keys := [formatted_key()],
+        config := encoded_config(),
+        project => project(),
+        location => location(),
+        registry => registry(),
+        created_at => created_at()
+    }.
+-export_type([formatted_device/0, deviceid/0, encoded_config/0]).
+
+-define(TAB, ?MODULE).
+
+-dialyzer({nowarn_function, perform_dirty/2}).
+
+%%--------------------------------------------------------------------
+%% API
+%%--------------------------------------------------------------------
+
+-spec put_device(formatted_device()) -> ok.
+put_device(FormattedDevice) ->
+    try
+        perform_dirty(?ACTIVITY, fun() -> put_device_no_transaction(FormattedDevice) end)
+    catch
+        _Error:Reason ->
+            ?SLOG(error, #{
+                msg => "Failed to put device",
+                device => FormattedDevice,
+                reason => Reason
+            }),
+            {error, Reason}
+    end.
+
+-spec get_device(deviceid()) -> {ok, formatted_device()} | not_found.
+get_device(DeviceId) ->
+    case ets:lookup(?TAB, DeviceId) of
+        [] ->
+            not_found;
+        [Device] ->
+            {ok, format_device(Device)}
+    end.
+
+-spec remove_device(deviceid()) -> ok.
+remove_device(DeviceId) ->
+    ok = mria:dirty_delete({?TAB, DeviceId}),
+    ok = put_config(DeviceId, <<>>).
+
+-spec get_device_actual_keys(deviceid()) -> [key()] | not_found.
+get_device_actual_keys(DeviceId) ->
+    try ets:lookup(?TAB, DeviceId) of
+        [] ->
+            not_found;
+        [Device] ->
+            actual_keys(Device)
+    catch
+        error:badarg ->
+            not_found
+    end.
+
+-spec import_devices([formatted_device()]) ->
+    {NumImported :: non_neg_integer(), NumError :: non_neg_integer()}.
+import_devices(FormattedDevices) when is_list(FormattedDevices) ->
+    perform_dirty(fun() -> lists:foldl(fun import_device/2, {0, 0}, FormattedDevices) end).
+
+%%--------------------------------------------------------------------
+%% Internal API
+%%--------------------------------------------------------------------
+
+-spec create_table() -> ok.
+create_table() ->
+    ok = mria:create_table(?TAB, [
+        {rlog_shard, ?AUTH_SHARD},
+        {type, ordered_set},
+        {storage, disc_copies},
+        {record_name, emqx_gcp_device},
+        {attributes, record_info(fields, emqx_gcp_device)},
+        {storage_properties, [{ets, [{read_concurrency, true}]}, {dets, [{auto_save, 10_000}]}]}
+    ]),
+    ok = mria:wait_for_tables([?TAB]).
+
+-spec clear_table() -> ok.
+clear_table() ->
+    {atomic, ok} = mria:clear_table(?TAB),
+    ok.
+
+%%--------------------------------------------------------------------
+%% Internal functions
+%%--------------------------------------------------------------------
+
+-spec perform_dirty(function()) -> term().
+perform_dirty(Fun) ->
+    perform_dirty(?ACTIVITY, Fun).
+
+-spec perform_dirty(async_dirty | sync_dirty, function()) -> term().
+perform_dirty(async_dirty, Fun) ->
+    mria:async_dirty(?AUTH_SHARD, Fun);
+perform_dirty(sync_dirty, Fun) ->
+    mria:sync_dirty(?AUTH_SHARD, Fun).
+
+-spec put_device_no_transaction(formatted_device()) -> ok.
+put_device_no_transaction(
+    #{
+        deviceid := DeviceId,
+        keys := Keys,
+        config := EncodedConfig
+    } = Device
+) ->
+    DeviceLoc =
+        list_to_tuple([maps:get(Key, Device, <<>>) || Key <- [project, location, registry]]),
+    ok = put_device_no_transaction(DeviceId, DeviceLoc, Keys),
+    ok = put_config(DeviceId, EncodedConfig).
+
+-spec put_device_no_transaction(deviceid(), device_loc(), [key()]) -> ok.
+put_device_no_transaction(DeviceId, DeviceLoc, Keys) ->
+    CreatedAt = erlang:system_time(second),
+    Extra = #{},
+    Device =
+        #emqx_gcp_device{
+            id = DeviceId,
+            keys = formatted_keys_to_records(Keys),
+            device_loc = DeviceLoc,
+            created_at = CreatedAt,
+            extra = Extra
+        },
+    mnesia:write(Device).
+
+-spec formatted_keys_to_records([formatted_key()]) -> [key_record()].
+formatted_keys_to_records(Keys) ->
+    lists:map(fun formatted_key_to_record/1, Keys).
+
+-spec formatted_key_to_record(formatted_key()) -> key_record().
+formatted_key_to_record(#{
+    key_type := KeyType,
+    key := Key,
+    expires_at := ExpiresAt
+}) ->
+    {KeyType, Key, ExpiresAt}.
+
+-spec format_device(emqx_gcp_device()) -> formatted_device().
+format_device(#emqx_gcp_device{
+    id = DeviceId,
+    device_loc = {Project, Location, Registry},
+    keys = Keys,
+    created_at = CreatedAt
+}) ->
+    #{
+        deviceid => DeviceId,
+        project => Project,
+        location => Location,
+        registry => Registry,
+        keys => lists:map(fun format_key/1, Keys),
+        created_at => CreatedAt,
+        config => base64:encode(get_device_config(DeviceId))
+    }.
+
+-spec format_key(key_record()) -> formatted_key().
+format_key({KeyType, Key, ExpiresAt}) ->
+    #{
+        key_type => KeyType,
+        key => Key,
+        expires_at => ExpiresAt
+    }.
+
+-spec put_config(deviceid(), encoded_config()) -> ok.
+put_config(DeviceId, EncodedConfig) ->
+    Config = base64:decode(EncodedConfig),
+    Topic = config_topic(DeviceId),
+    Message = emqx_message:make(DeviceId, 1, Topic, Config, #{retain => true}, #{}),
+    _ = emqx_broker:publish(Message),
+    ok.
+
+-spec get_device_config(deviceid()) -> emqx_types:payload().
+get_device_config(DeviceId) ->
+    Topic = config_topic(DeviceId),
+    get_retained_payload(Topic).
+
+-spec actual_keys(emqx_gcp_device()) -> [key()].
+actual_keys(#emqx_gcp_device{keys = Keys}) ->
+    Now = erlang:system_time(second),
+    [Key || {_KeyType, Key, ExpiresAt} <- Keys, ExpiresAt == 0 orelse ExpiresAt >= Now].
+
+-spec import_device(formatted_device(), {
+    NumImported :: non_neg_integer(), NumError :: non_neg_integer()
+}) -> {NumImported :: non_neg_integer(), NumError :: non_neg_integer()}.
+import_device(Device, {NumImported, NumError}) ->
+    try
+        ok = put_device_no_transaction(Device),
+        {NumImported + 1, NumError}
+    catch
+        Error:Reason:Stacktrace ->
+            ?SLOG(error, #{
+                msg => "Failed to import device",
+                exception => Error,
+                reason => Reason,
+                stacktrace => Stacktrace
+            }),
+            {NumImported, NumError + 1}
+    end.
+
+-spec get_retained_payload(binary()) -> emqx_types:payload().
+get_retained_payload(Topic) ->
+    case emqx_retainer:read_message(Topic) of
+        {ok, []} ->
+            <<>>;
+        {ok, [Message]} ->
+            Message#message.payload
+    end.
+
+-spec config_topic(deviceid()) -> binary().
+config_topic(DeviceId) ->
+    <<"/devices/", DeviceId/binary, "/config">>.

apps/emqx_gcp_device/src/emqx_gcp_device_api.erl

@@ -0,0 +1,456 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_gcp_device_api).
+
+-behaviour(minirest_api).
+
+-include_lib("hocon/include/hoconsc.hrl").
+-include_lib("stdlib/include/qlc.hrl").
+-include_lib("stdlib/include/ms_transform.hrl").
+-include_lib("emqx/include/logger.hrl").
+
+-define(TAGS, [<<"GCP Devices">>]).
+-define(TAB, emqx_gcp_device).
+-define(FORMAT_FUN, {emqx_gcp_device, format_device}).
+
+-export([import_devices/1]).
+-export([get_device/1, update_device/1, remove_device/1]).
+
+-export([
+    api_spec/0,
+    paths/0,
+    schema/1,
+    fields/1
+]).
+
+-export([
+    '/gcp_devices'/2,
+    '/gcp_devices/:deviceid'/2
+]).
+
+-type deviceid() :: emqx_gcp_device:deviceid().
+-type formatted_device() :: emqx_gcp_device:formatted_device().
+-type base64_encoded_config() :: emqx_gcp_device:encoded_config().
+-type imported_key() :: #{
+    binary() := binary() | non_neg_integer()
+    % #{
+    %     <<"key">> => binary(),
+    %     <<"key_type">> => binary(),
+    %     <<"expires_at">> => non_neg_integer()
+    % }.
+}.
+-type key_fields() :: key | key_type | expires_at.
+-type imported_device() :: #{
+    binary() := deviceid() | binary() | [imported_key()] | base64_encoded_config() | boolean()
+    % #{
+    %     <<"deviceid">> => deviceid(),
+    %     <<"project">> => binary(),
+    %     <<"location">> => binary(),
+    %     <<"registry">> => binary(),
+    %     <<"keys">> => [imported_key()],
+    %     <<"config">> => base64_encoded_config(),
+    %     <<"blocked">> => boolean(),
+    % }.
+}.
+-type device_fields() :: deviceid | project | location | registry | keys | config.
+-type checked_device_fields() :: device_fields() | key_fields().
+-type validated_device() :: #{checked_device_fields() := term()}.
+
+%%-------------------------------------------------------------------------------------------------
+%% `minirest' and `minirest_trails' API
+%%-------------------------------------------------------------------------------------------------
+
+api_spec() ->
+    emqx_dashboard_swagger:spec(?MODULE, #{check_schema => true}).
+
+paths() ->
+    [
+        "/gcp_devices",
+        "/gcp_devices/:deviceid"
+    ].
+
+schema("/gcp_devices") ->
+    #{
+        'operationId' => '/gcp_devices',
+        get => #{
+            description => ?DESC(gcp_devices_get),
+            tags => ?TAGS,
+            parameters => [
+                hoconsc:ref(emqx_dashboard_swagger, page),
+                hoconsc:ref(emqx_dashboard_swagger, limit)
+            ],
+            responses => #{
+                200 => [
+                    {data, hoconsc:mk(hoconsc:array(hoconsc:ref(gcp_device_all_info)), #{})},
+                    {meta, hoconsc:mk(hoconsc:ref(emqx_dashboard_swagger, meta), #{})}
+                ]
+            }
+        },
+        post => #{
+            description => ?DESC(gcp_devices_post),
+            tags => ?TAGS,
+            'requestBody' => hoconsc:mk(hoconsc:array(?R_REF(gcp_exported_device)), #{}),
+            responses =>
+                #{
+                    200 => hoconsc:ref(import_result),
+                    400 => emqx_dashboard_swagger:error_codes(
+                        ['BAD_REQUEST'],
+                        <<"Bad Request">>
+                    )
+                }
+        }
+    };
+schema("/gcp_devices/:deviceid") ->
+    #{
+        'operationId' => '/gcp_devices/:deviceid',
+        get =>
+            #{
+                description => ?DESC(gcp_device_get),
+                tags => ?TAGS,
+                parameters => [deviceid(#{in => path})],
+                responses =>
+                    #{
+                        200 => hoconsc:mk(
+                            hoconsc:ref(gcp_device_all_info),
+                            #{
+                                desc => ?DESC(gcp_device_all_info)
+                            }
+                        ),
+                        404 => emqx_dashboard_swagger:error_codes(
+                            ['NOT_FOUND'],
+                            ?DESC(gcp_device_response404)
+                        )
+                    }
+            },
+        put =>
+            #{
+                description => ?DESC(gcp_device_put),
+                tags => ?TAGS,
+                parameters => [deviceid(#{in => path})],
+                'requestBody' => hoconsc:ref(gcp_device),
+                responses =>
+                    #{
+                        200 => hoconsc:mk(
+                            hoconsc:ref(gcp_device_info),
+                            #{
+                                desc => ?DESC(gcp_device_info)
+                            }
+                        ),
+                        400 => emqx_dashboard_swagger:error_codes(
+                            ['BAD_REQUEST'],
+                            <<"Bad Request">>
+                        )
+                    }
+            },
+        delete => #{
+            description => ?DESC(gcp_device_delete),
+            tags => ?TAGS,
+            parameters => [deviceid(#{in => path})],
+            responses => #{
+                204 => <<"GCP device deleted">>
+            }
+        }
+    }.
+
+fields(gcp_device) ->
+    [
+        {registry,
+            hoconsc:mk(
+                binary(),
+                #{
+                    desc => ?DESC(registry),
+                    default => <<>>,
+                    example => <<"my-registry">>
+                }
+            )},
+        {project,
+            hoconsc:mk(
+                binary(),
+                #{
+                    desc => ?DESC(project),
+                    default => <<>>,
+                    example => <<"iot-export">>
+                }
+            )},
+        {location,
+            hoconsc:mk(
+                binary(),
+                #{
+                    desc => ?DESC(location),
+                    default => <<>>,
+                    example => <<"europe-west1">>
+                }
+            )},
+        {keys,
+            hoconsc:mk(
+                ?ARRAY(hoconsc:ref(key)),
+                #{
+                    desc => ?DESC(keys),
+                    default => []
+                }
+            )},
+        {config,
+            hoconsc:mk(
+                binary(),
+                #{
+                    desc => ?DESC(config),
+                    required => true,
+                    example => <<"bXktY29uZmln">>
+                }
+            )}
+    ];
+fields(gcp_device_info) ->
+    fields(deviceid) ++ fields(gcp_device);
+fields(gcp_device_all_info) ->
+    [
+        {created_at,
+            hoconsc:mk(
+                non_neg_integer(),
+                #{
+                    desc => ?DESC(created_at),
+                    required => true,
+                    example => 1690484400
+                }
+            )}
+    ] ++ fields(gcp_device_info);
+fields(gcp_exported_device) ->
+    [
+        {blocked,
+            hoconsc:mk(
+                boolean(),
+                #{
+                    desc => ?DESC(blocked),
+                    required => true,
+                    example => false
+                }
+            )}
+    ] ++ fields(deviceid) ++ fields(gcp_device);
+fields(import_result) ->
+    [
+        {errors,
+            hoconsc:mk(
+                non_neg_integer(),
+                #{
+                    desc => ?DESC(imported_counter_errors),
+                    required => true,
+                    example => 0
+                }
+            )},
+        {imported,
+            hoconsc:mk(
+                non_neg_integer(),
+                #{
+                    desc => ?DESC(imported_counter),
+                    required => true,
+                    example => 14
+                }
+            )}
+    ];
+fields(key) ->
+    [
+        {key,
+            hoconsc:mk(
+                binary(),
+                #{
+                    desc => ?DESC(key),
+                    required => true,
+                    example => <<"<DEVICE-PUBLIC-KEY>">>
+                }
+            )},
+        {key_type,
+            hoconsc:mk(
+                binary(),
+                #{
+                    desc => ?DESC(key_type),
+                    required => true,
+                    example => <<"ES256_PEM">>
+                }
+            )},
+        {expires_at,
+            hoconsc:mk(
+                non_neg_integer(),
+                #{
+                    desc => ?DESC(expires_at),
+                    required => true,
+                    example => 1706738400
+                }
+            )}
+    ];
+fields(deviceid) ->
+    [
+        deviceid()
+    ].
+
+'/gcp_devices'(get, #{query_string := Params}) ->
+    Response = emqx_mgmt_api:paginate(?TAB, Params, ?FORMAT_FUN),
+    {200, Response};
+'/gcp_devices'(post, #{body := Body}) ->
+    import_devices(Body).
+
+'/gcp_devices/:deviceid'(get, #{bindings := #{deviceid := DeviceId}}) ->
+    get_device(DeviceId);
+'/gcp_devices/:deviceid'(put, #{bindings := #{deviceid := DeviceId}, body := Body}) ->
+    update_device(maps:merge(Body, #{<<"deviceid">> => DeviceId}));
+'/gcp_devices/:deviceid'(delete, #{bindings := #{deviceid := DeviceId}}) ->
+    remove_device(DeviceId).
+
+%%------------------------------------------------------------------------------
+%% Handlers
+%%------------------------------------------------------------------------------
+
+-spec import_devices([imported_device()]) ->
+    {200, #{imported := non_neg_integer(), errors := non_neg_integer()}}
+    | {400, #{message := binary()}}.
+import_devices(Devices) ->
+    case validate_devices(Devices) of
+        {ok, FormattedDevices} ->
+            {NumImported, NumErrors} = emqx_gcp_device:import_devices(FormattedDevices),
+            {200, #{imported => NumImported, errors => NumErrors}};
+        {error, Reason} ->
+            {400, #{message => Reason}}
+    end.
+
+-spec get_device(deviceid()) -> {200, formatted_device()} | {404, 'NOT_FOUND', binary()}.
+get_device(DeviceId) ->
+    case emqx_gcp_device:get_device(DeviceId) of
+        {ok, Device} ->
+            {200, Device};
+        not_found ->
+            Message = list_to_binary(io_lib:format("device not found: ~s", [DeviceId])),
+            {404, 'NOT_FOUND', Message}
+    end.
+
+-spec update_device(imported_device()) -> {200, formatted_device()} | {400, binary()}.
+update_device(Device) ->
+    case validate_device(Device) of
+        {ok, ValidatedDevice} ->
+            ok = emqx_gcp_device:put_device(ValidatedDevice),
+            {200, ValidatedDevice};
+        {error, Reason} ->
+            {400, Reason}
+    end.
+
+-spec remove_device(deviceid()) -> {204}.
+remove_device(DeviceId) ->
+    ok = emqx_gcp_device:remove_device(DeviceId),
+    {204}.
+
+%%------------------------------------------------------------------------------
+%% Internal functions
+%%------------------------------------------------------------------------------
+
+-define(KEY_TYPES, [<<"RSA_PEM">>, <<"RSA_X509_PEM">>, <<"ES256_PEM">>, <<"ES256_X509_PEM">>]).
+
+-spec deviceid() -> tuple().
+deviceid() ->
+    deviceid(#{}).
+
+-spec deviceid(map()) -> tuple().
+deviceid(Override) ->
+    {deviceid,
+        hoconsc:mk(
+            binary(),
+            maps:merge(
+                #{
+                    desc => ?DESC(deviceid),
+                    required => true,
+                    example => <<"c2-ec-x509">>
+                },
+                Override
+            )
+        )}.
+
+-spec validate_devices([imported_device()]) -> {ok, [validated_device()]} | {error, binary()}.
+validate_devices(Devices) ->
+    validate_devices(Devices, []).
+
+-spec validate_devices([imported_device()], [validated_device()]) ->
+    {ok, [validated_device()]} | {error, binary()}.
+validate_devices([], Validated) ->
+    {ok, lists:reverse(Validated)};
+validate_devices([Device | Devices], Validated) ->
+    case validate_device(Device) of
+        {ok, ValidatedDevice} ->
+            validate_devices(Devices, [ValidatedDevice | Validated]);
+        {error, _} = Error ->
+            Error
+    end.
+
+-spec validate_device(imported_device()) -> {ok, validated_device()} | {error, binary()}.
+validate_device(Device) ->
+    validate([deviceid, project, location, registry, keys, config], Device).
+
+-spec validate([checked_device_fields()], imported_device()) ->
+    {ok, validated_device()} | {error, binary()}.
+validate(Fields, Device) ->
+    validate(Fields, Device, #{}).
+
+-spec validate([checked_device_fields()], imported_device(), validated_device()) ->
+    {ok, validated_device()} | {error, binary()}.
+validate([], _Device, Validated) ->
+    {ok, Validated};
+validate([key_type | Fields], #{<<"key_type">> := KeyType} = Device, Validated) ->
+    case lists:member(KeyType, ?KEY_TYPES) of
+        true ->
+            validate(Fields, Device, Validated#{key_type => KeyType});
+        false ->
+            {error, <<"invalid key_type">>}
+    end;
+validate([key | Fields], #{<<"key">> := Key} = Device, Validated) ->
+    validate(Fields, Device, Validated#{key => Key});
+validate([expires_at | Fields], #{<<"expires_at">> := Expire} = Device, Validated) when
+    is_integer(Expire)
+->
+    validate(Fields, Device, Validated#{expires_at => Expire});
+validate([expires_at | _Fields], #{<<"expires_at">> := _}, _Validated) ->
+    {error, <<"invalid expires_at">>};
+validate([expires_at | Fields], Device, Validated) ->
+    validate(Fields, Device, Validated#{expires_at => 0});
+validate([Field | Fields], Device, Validated) when Field =:= deviceid; Field =:= key ->
+    FieldBin = atom_to_binary(Field),
+    case maps:find(FieldBin, Device) of
+        {ok, Value} when is_binary(Value) ->
+            validate(Fields, Device, Validated#{Field => Value});
+        _ ->
+            {error, <<"invalid or missing field: ", FieldBin/binary>>}
+    end;
+validate([Field | Fields], Device, Validated) when
+    Field =:= project; Field =:= location; Field =:= registry; Field =:= config
+->
+    FieldBin = atom_to_binary(Field),
+    case maps:find(FieldBin, Device) of
+        {ok, Value} when is_binary(Value) ->
+            validate(Fields, Device, Validated#{Field => Value});
+        error ->
+            validate(Fields, Device, Validated#{Field => <<>>});
+        _ ->
+            {error, <<"invalid field: ", FieldBin/binary>>}
+    end;
+validate([keys | Fields], #{<<"keys">> := Keys} = Device, Validated) when is_list(Keys) ->
+    case validate_keys(Keys) of
+        {ok, ValidatedKeys} ->
+            validate(Fields, Device, Validated#{keys => ValidatedKeys});
+        {error, _} = Error ->
+            Error
+    end;
+validate([Field | _Fields], _Device, _Validated) ->
+    {error, <<"invalid or missing field: ", (atom_to_binary(Field))/binary>>}.
+
+-spec validate_keys([imported_key()]) ->
+    {ok, [validated_device()]} | {error, binary()}.
+validate_keys(Keys) ->
+    validate_keys(Keys, []).
+
+-spec validate_keys([imported_key()], [validated_device()]) ->
+    {ok, [validated_device()]} | {error, binary()}.
+validate_keys([], Validated) ->
+    {ok, lists:reverse(Validated)};
+validate_keys([Key | Keys], Validated) ->
+    case validate([key, key_type, expires_at], Key) of
+        {ok, ValidatedKey} ->
+            validate_keys(Keys, [ValidatedKey | Validated]);
+        {error, _} = Error ->
+            Error
+    end.

apps/emqx_gcp_device/src/emqx_gcp_device_app.erl

@@ -0,0 +1,21 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_gcp_device_app).
+
+-behaviour(application).
+
+-emqx_plugin(?MODULE).
+
+-export([
+    start/2,
+    stop/1
+]).
+
+start(_StartType, _StartArgs) ->
+    emqx_gcp_device:create_table(),
+    emqx_gcp_device_sup:start_link().
+
+stop(_State) ->
+    ok.

apps/emqx_gcp_device/src/emqx_gcp_device_authn.erl

@@ -0,0 +1,213 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_gcp_device_authn).
+
+-include_lib("emqx_authn/include/emqx_authn.hrl").
+-include_lib("emqx/include/logger.hrl").
+-include_lib("hocon/include/hoconsc.hrl").
+-include_lib("jose/include/jose_jwt.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+
+-behaviour(hocon_schema).
+
+-export([
+    namespace/0,
+    tags/0,
+    roots/0,
+    fields/1,
+    desc/1
+]).
+
+-export([
+    refs/0,
+    create/2,
+    update/2,
+    authenticate/2,
+    destroy/1
+]).
+
+%%------------------------------------------------------------------------------
+%% Hocon Schema
+%%------------------------------------------------------------------------------
+
+namespace() -> "authn".
+
+tags() ->
+    [<<"Authentication">>].
+
+%% used for config check when the schema module is resolved
+roots() ->
+    [{?CONF_NS, hoconsc:mk(hoconsc:ref(gcp_device))}].
+
+fields(gcp_device) ->
+    common_fields().
+
+desc(gcp_device) ->
+    ?DESC(emqx_gcp_device_api, gcp_device);
+desc(_) ->
+    undefined.
+
+%%------------------------------------------------------------------------------
+%% APIs
+%%------------------------------------------------------------------------------
+
+refs() ->
+    [
+        hoconsc:ref(?MODULE, gcp_device)
+    ].
+
+create(_AuthenticatorID, _Config) ->
+    {ok, #{}}.
+
+update(
+    _Config,
+    State
+) ->
+    {ok, State}.
+
+authenticate(#{auth_method := _}, _) ->
+    ignore;
+authenticate(Credential, _State) ->
+    check(Credential).
+
+destroy(_State) ->
+    emqx_gcp_device:clear_table(),
+    ok.
+
+%%--------------------------------------------------------------------
+%% Internal functions
+%%--------------------------------------------------------------------
+
+common_fields() ->
+    [
+        {mechanism, emqx_authn_schema:mechanism('gcp_device')}
+    ] ++ emqx_authn_schema:common_fields().
+
+% The check logic is the following:
+%% 1. If clientid is not GCP-like or password is not a JWT, the result is ignore
+%% 2. If clientid is GCP-like and password is a JWT, but expired, the result is password_error
+%% 3. If clientid is GCP-like and password is a valid and not expired JWT:
+%%  3.1 If there are no keys for the client, the result is ignore
+%%  3.2 If there are some keys for the client:
+%%   3.2.1 If there are no actual (not expired keys), the result is password_error
+%%   3.2.2 If there are some actual keys and one of them matches the JWT, the result is success
+%%   3.2.3 If there are some actual keys and none of them matches the JWT, the result is password_error
+check(#{password := Password} = ClientInfo) ->
+    case gcp_deviceid_from_clientid(ClientInfo) of
+        {ok, DeviceId} ->
+            case is_valid_jwt(Password) of
+                true ->
+                    check_jwt(ClientInfo, DeviceId);
+                {false, not_a_jwt} ->
+                    ?tp(authn_gcp_device_check, #{
+                        result => ignore, reason => "not a JWT", client => ClientInfo
+                    }),
+                    ?TRACE_AUTHN_PROVIDER(debug, "auth_ignored", #{
+                        reason => "not a JWT",
+                        client => ClientInfo
+                    }),
+                    ignore;
+                {false, expired} ->
+                    ?tp(authn_gcp_device_check, #{
+                        result => not_authorized, reason => "expired JWT", client => ClientInfo
+                    }),
+                    ?TRACE_AUTHN_PROVIDER(info, "auth_failed", #{
+                        reason => "expired JWT",
+                        client => ClientInfo
+                    }),
+                    {error, not_authorized}
+            end;
+        not_a_gcp_clientid ->
+            ?tp(authn_gcp_device_check, #{
+                result => ignore, reason => "not a GCP ClientId", client => ClientInfo
+            }),
+            ?TRACE_AUTHN_PROVIDER(debug, "auth_ignored", #{
+                reason => "not a GCP ClientId",
+                client => ClientInfo
+            }),
+            ignore
+    end.
+
+check_jwt(ClientInfo, DeviceId) ->
+    case emqx_gcp_device:get_device_actual_keys(DeviceId) of
+        not_found ->
+            ?tp(authn_gcp_device_check, #{
+                result => ignore, reason => "key not found", client => ClientInfo
+            }),
+            ?TRACE_AUTHN_PROVIDER(debug, "auth_ignored", #{
+                reason => "key not found",
+                client => ClientInfo
+            }),
+            ignore;
+        Keys ->
+            case any_key_matches(Keys, ClientInfo) of
+                true ->
+                    ?tp(authn_gcp_device_check, #{
+                        result => ok, reason => "auth success", client => ClientInfo
+                    }),
+                    ?TRACE_AUTHN_PROVIDER(debug, "auth_success", #{
+                        reason => "auth success",
+                        client => ClientInfo
+                    }),
+                    ok;
+                false ->
+                    ?tp(authn_gcp_device_check, #{
+                        result => {error, bad_username_or_password},
+                        reason => "no matching or valid keys",
+                        client => ClientInfo
+                    }),
+                    ?TRACE_AUTHN_PROVIDER(info, "auth_failed", #{
+                        reason => "no matching or valid keys",
+                        client => ClientInfo
+                    }),
+                    {error, bad_username_or_password}
+            end
+    end.
+
+any_key_matches(Keys, ClientInfo) ->
+    lists:any(fun(Key) -> key_matches(Key, ClientInfo) end, Keys).
+
+key_matches(KeyRaw, #{password := Jwt} = _ClientInfo) ->
+    Jwk = jose_jwk:from_pem(KeyRaw),
+    case jose_jws:verify(Jwk, Jwt) of
+        {true, _, _} ->
+            true;
+        {false, _, _} ->
+            false
+    end.
+
+gcp_deviceid_from_clientid(#{clientid := <<"projects/", RestClientId/binary>>}) ->
+    case binary:split(RestClientId, <<"/">>, [global]) of
+        [
+            _Project,
+            <<"locations">>,
+            _Location,
+            <<"registries">>,
+            _Registry,
+            <<"devices">>,
+            DeviceId
+        ] ->
+            {ok, DeviceId};
+        _ ->
+            not_a_gcp_clientid
+    end;
+gcp_deviceid_from_clientid(_ClientInfo) ->
+    not_a_gcp_clientid.
+
+is_valid_jwt(Password) ->
+    Now = erlang:system_time(second),
+    try jose_jwt:peek(Password) of
+        #jose_jwt{fields = #{<<"exp">> := Exp}} when is_integer(Exp) andalso Exp >= Now ->
+            true;
+        #jose_jwt{fields = #{<<"exp">> := _Exp}} ->
+            {false, expired};
+        #jose_jwt{} ->
+            true;
+        _ ->
+            {false, not_a_jwt}
+    catch
+        _:_ ->
+            {false, not_a_jwt}
+    end.

apps/emqx_gcp_device/src/emqx_gcp_device_sup.erl

@@ -0,0 +1,25 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_gcp_device_sup).
+
+-behaviour(supervisor).
+
+-export([start_link/0]).
+-export([init/1]).
+
+-define(SERVER, ?MODULE).
+
+start_link() ->
+    supervisor:start_link({local, ?SERVER}, ?MODULE, []).
+
+init([]) ->
+    SupFlags =
+        #{
+            strategy => one_for_all,
+            intensity => 0,
+            period => 1
+        },
+    ChildSpecs = [],
+    {ok, {SupFlags, ChildSpecs}}.

apps/emqx_gcp_device/test/data/keys/c1_ec_private.pem

@@ -0,0 +1,5 @@
+-----BEGIN EC PRIVATE KEY-----
+MHcCAQEEIGN8JyB8C3vW+SKTj5JcOeFdU9zM4mV35o+JumELI/w+oAoGCCqGSM49
+AwEHoUQDQgAE4tqkxsDZ1tZPhtLcCi5BdhT0idF5wqP9I2ITa7trw+n6YRsrqnbr
++sklCPN6tySLRrGT8IpFlLo0xJFRmuAyLw==
+-----END EC PRIVATE KEY-----

apps/emqx_gcp_device/test/data/keys/c1_ec_public.pem

@@ -0,0 +1,4 @@
+-----BEGIN PUBLIC KEY-----
+MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE4tqkxsDZ1tZPhtLcCi5BdhT0idF5
+wqP9I2ITa7trw+n6YRsrqnbr+sklCPN6tySLRrGT8IpFlLo0xJFRmuAyLw==
+-----END PUBLIC KEY-----

apps/emqx_gcp_device/test/data/keys/c2_ec_cert.pem

@@ -0,0 +1,8 @@
+-----BEGIN CERTIFICATE-----
+MIIBEjCBuAIJAPKVZoroXatKMAoGCCqGSM49BAMCMBExDzANBgNVBAMMBnVudXNl
+ZDAeFw0yMzA0MTIxMzQ2NTJaFw0yMzA1MTIxMzQ2NTJaMBExDzANBgNVBAMMBnVu
+dXNlZDBZMBMGByqGSM49AgEGCCqGSM49AwEHA0IABAugsuay/y2SpGEVDKfiVw9q
+VHGdZHvLXDqxj9XndUi6LEpA209ZfaC1eJ+mZiW3zBC94AdqVu+QLzS7rPT72jkw
+CgYIKoZIzj0EAwIDSQAwRgIhAMBp+1S5w0UJDuylI1TJS8vXjWOhgluUdZfFtxES
+E85SAiEAvKIAhjRhuIxanhqyv3HwOAL/zRAcv6iHsPMKYBt1dOs=
+-----END CERTIFICATE-----

apps/emqx_gcp_device/test/data/keys/c2_ec_private.pem

@@ -0,0 +1,5 @@
+-----BEGIN EC PRIVATE KEY-----
+MHcCAQEEIECpfvahaDpwOVSqQmf//F9nzK6W5m9BQklpx8DbAHscoAoGCCqGSM49
+AwEHoUQDQgAEC6Cy5rL/LZKkYRUMp+JXD2pUcZ1ke8tcOrGP1ed1SLosSkDbT1l9
+oLV4n6ZmJbfMEL3gB2pW75AvNLus9PvaOQ==
+-----END EC PRIVATE KEY-----

apps/emqx_gcp_device/test/data/keys/c3_rsa_private.pem

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDNzbsdfe3e0EDf
+VUT6QdWE+WgBUGm8JIdH4d8Q4j76oJSI76dTV0AgqxJa7iKlvAKXnuh61DCZ4GdH
+j1FrQHqzlTI6yw3D1qJFVQVl1Uo6J+hpchnLU1N8bq2CnZarx8vkrIDlKr4/MWR/
+IP1rFhALmNsX6cnvw4eDT/cRhCE3s28KKlODEJxb+ggYbq/Q3pXlbvH7uLElZss/
+JlgzS8RMg3QEjFvHVH58aTJRYCq81+95PI3nCrsxj33XkFIvIZSmuyr2fkMmaILj
+hElVq7Yx8feYpDGSfxz/YvSFnOorqzCPUR0T8ECID637ruY5Uwugas6p87Xpjd/t
+K/kPtdyPAgMBAAECggEAU3PcL05UOai61ZUPHme5vG0iFn5UEd3CGYzm1kLYBOs+
+r/R2Jl5X+6dDDypHVGtTpcXjQYNvncYYOzVLb7E60D1sm9ig4UvUi0a5pJyDt+dc
+3/1Lpl5ImUmMBE4AvfGLpVOqBMN7V8agmMh42oacxQcbuKutnhLsjXvMlQa+LYZT
++FQV8kQV8D4GgjmP2jl0/Y2M6BjKEK2Ih7qPvo46L439vk2JGF8N+NtGjCKy6Wra
+X9uFA3+RjsqcN6mPa77OEDmN9HjpSPraJowPlZR+xrJjbekIri/uyNWMZ6BCmkPx
+0kRkScUmZMfq+SIIdsMszp8P549nwmBNCgFgcOJTYQKBgQDt1ZZzA7r07lhF9T9W
+0bfzbg230v03LiPGHMsjerZfWCMMs+RgBkkgLPG4XyMKZNCUsj5Pt5WyVBXaaWY4
+LrE5kLdpIn/oRykaK1i+AGkXHhIWAlvqsWWg+R2sLCwaIiolGuc1b+ZERS+5VMrf
+c71t/i8OB22uCPrRShIIQqrGsQKBgQDdhdeQ8ZoumNFFcapN0I/sKNhuuvw1mtOI
+tduNkOyf68XCpM7yDe86DV8cPbFNHhGMZnhpSxu0yyHQLuL9Nwv9gAB66yIzvk+N
+iv+WTIqgIDQN26Ljz2q4hc9SpT8zLRLrDAIJBxAti37xZTs6sj6fjXwlEE8l2RRM
++FTECIonPwKBgFBZkXuH7hijkWUJJv3w2kG+k5ngCTYkO2fKAIMbCRQLFcRL3kLm
+vLvHE17jnVX8m08xLMYH0uYtbDie1S7z72HwV1aIlkfmCqfRryh5wQdTXG7dGyqe
+BiStJO4u+jNWCYEBps0x4cx8x1PIpsV5N606a7FEpzRdykb8zDzIMSPxAoGAGHLK
+HMwdaSEij5iA5D+tcrH7WRU3+q6QxBjWF2S0SN4boGTSFjLlgTGymopQhCNaanVw
+uqY4c5arr69NDAdEQoEbDHXg+3b4jrWVib/+2LdVJ2ZjLuNYcu8Jt6RXOk2yNdDI
+dLib13r60qeKhurfMHrMBccsBRBVRj1uFYifvr8CgYEAynbD898pShniuKii5c4i
+3RrzhK/V6XGLfOJzDtjZ/uRcv8nt42kdbU3z+M87GE6hXn0rm6AIgVQKtSoaUHWH
+oTVOtmdctkx8GmcdhSX5fs2wzVxvVsqyf1wjo6UG/90k9nxY+AjMU144ZpuRYuKQ
+pWtPdQWBlw58XRAHW8r9Zxs=
+-----END PRIVATE KEY-----

apps/emqx_gcp_device/test/data/keys/c3_rsa_public.pem

@@ -0,0 +1,9 @@
+-----BEGIN PUBLIC KEY-----
+MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAzc27HX3t3tBA31VE+kHV
+hPloAVBpvCSHR+HfEOI++qCUiO+nU1dAIKsSWu4ipbwCl57oetQwmeBnR49Ra0B6
+s5UyOssNw9aiRVUFZdVKOifoaXIZy1NTfG6tgp2Wq8fL5KyA5Sq+PzFkfyD9axYQ
+C5jbF+nJ78OHg0/3EYQhN7NvCipTgxCcW/oIGG6v0N6V5W7x+7ixJWbLPyZYM0vE
+TIN0BIxbx1R+fGkyUWAqvNfveTyN5wq7MY9915BSLyGUprsq9n5DJmiC44RJVau2
+MfH3mKQxkn8c/2L0hZzqK6swj1EdE/BAiA+t+67mOVMLoGrOqfO16Y3f7Sv5D7Xc
+jwIDAQAB
+-----END PUBLIC KEY-----

apps/emqx_gcp_device/test/data/keys/c4_rsa_cert.pem

@@ -0,0 +1,17 @@
+-----BEGIN CERTIFICATE-----
+MIICnjCCAYYCCQCh+b8WxXjihDANBgkqhkiG9w0BAQsFADARMQ8wDQYDVQQDDAZ1
+bnVzZWQwHhcNMjMwNDEyMTM0NjUyWhcNMjMwNTEyMTM0NjUyWjARMQ8wDQYDVQQD
+DAZ1bnVzZWQwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDOtvDuketC
+56nvrZw61UyP+MJikYbqqxIqIqwyih2KDCzlF6gTBI6vbFNwZx1b366VOfDhuj6j
+44+cN44AoVKtqSzpsDjdlIRClcBIv4k2ndXjr6yV1cJ9lrMB9vPbr8fiQOxr31Cf
+ZUk0OZPppdsC5iqYpUeOdrSttOgBRIaTohBUXMatICxhc+9gC5yj9mQJuwckx6fE
+b+gJ9JrZ1/0wSW1EZNfS9hlOhA0nRUnty5wyqrpxdX4UL/G86SFl7njW9S1PBuPe
+HK7AdHZ6C3FAMfqpnETiWV149k/DR4UQQ7a23QsbgVJOM/7R9IAyln9LARhF9Bpp
+y/W2HPpBn8JHAgMBAAEwDQYJKoZIhvcNAQELBQADggEBAGFl+G3yk/BfELjX1mT6
+4mrGlJq3I6vXLN4ICSTmI4YZQgMmudIHEd6o/cZHJq8HOOqQ5SfFhQI7tBXZpXSG
+dybOStl+GnfyIQFjsNzFXJEiaHoBPP1ccpZyCW/IBkXX39h9N/Pq0XB+xDurXpOD
+VE8nICTATe1Th11rs8j6qwFCkaoQwrzg+JWOKvFnRTPPDNg21fNRRTS+SE27asF2
+PhBWZOD4G2g6WD73SHUs+prR/q4foSVXt63Ih8uQIQJllRtpI4ZkpwSXDH9DUZSY
+WyFtYkD0EAV/FaRuALZQzxX7wda4xwBhvDL8Wua1WENTGZq7ssRHldAdFrz8NENC
+Hqk=
+-----END CERTIFICATE-----

apps/emqx_gcp_device/test/data/keys/c4_rsa_private.pem

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQDOtvDuketC56nv
+rZw61UyP+MJikYbqqxIqIqwyih2KDCzlF6gTBI6vbFNwZx1b366VOfDhuj6j44+c
+N44AoVKtqSzpsDjdlIRClcBIv4k2ndXjr6yV1cJ9lrMB9vPbr8fiQOxr31CfZUk0
+OZPppdsC5iqYpUeOdrSttOgBRIaTohBUXMatICxhc+9gC5yj9mQJuwckx6fEb+gJ
+9JrZ1/0wSW1EZNfS9hlOhA0nRUnty5wyqrpxdX4UL/G86SFl7njW9S1PBuPeHK7A
+dHZ6C3FAMfqpnETiWV149k/DR4UQQ7a23QsbgVJOM/7R9IAyln9LARhF9Bppy/W2
+HPpBn8JHAgMBAAECggEBALKiEM55Nq7Yd1fx1UJaNRFtTL3VOJvuPYI/+EKsbB5x
+qxJGQS4+D/e0St6lnQ9Z2wqFyY2nXp5N9jpvH72Xq1T7Dx7a9Ck3QJwxwLqdGjwi
+ZUWe+Ct7T9krs4GNIOrFmpwAss39azRzWLFS2GletEZrFIBYw99u4XADF0KRLyK/
+qno/gnYWqrhc0NVG3OR2n+AruJF+EElbOBCmzPzgVgNYinLXpvpttBtlRIS3XIPD
+UhdP9O33oTAyUGxRUcqbnwWLMPa3mQijT8fwIMvmeK94RYWsGz5r3+GQRC8ieeVy
+4MjJLSwGLk2apxiuEWQzCwjnda0T9OwuIzJM0uT1CyECgYEA5njmwMQOqBsFresn
+AdGLWlMKA2sM9sl/A+I6d/+B1NtAvpcq4UHQDfOSbthiOiU4/uIx/ZP4wmB7Smk/
+WB7NfuXZySpJTEWn0fwEaKcXIksqumQ2Lwom0QCV1m5nSnVdw+VLdWVIngqqG+Id
+c6Rh0F96KpT8MalyxR1TsgP0jRkCgYEA5Zxirqc9SYQm/jaBfjGW9teunY/zQj7m
+lCEUEp4aS9zfwcOS973sU80HXsfU1dsbQy15whvozqbTQMAYoaKz54DCebuPkW3I
+o4tY6oCuFEHlOiait0KnRPG8ZiHZKeO3TGcLajQWGssNLbbDFlhby8S9thlJ1+GT
+ldSW0AxhVl8CgYAy+zGIGJZpZzjVZPwG8fRScaX4ZZjDioT3NfbbDoEItctXnZbV
+pzo/q86LiIAJ/qvh7eVDA5V2YeND7Y4ejwnD9VI8pob6QTpDP+01vShn5Jq6CmrV
+8vftKaT7fwaIOPgZ2kHb4SC0HQXODzGWoBkm/8fFXZl/3szNf5RA/5D8GQKBgQDX
+Y9pWiF+/pQ6HDk5vOMmrCSyudaj2jdbzQgx4YoO8gpgMRhCKAkm9Wun9CWwoqP9s
+By7e3huIL4qghRMWHXCyTGEinMXS4K+Ea2WfpdKnAiGsaS3ex9HtpO7cyAfVed4q
+98cHe5D41V2pcnaTcZO7FPX56sMQlnVB6kkHJXXx9QKBgHQPmp1uT+MCYOd+HLqo
+b2tDxSukm/qe5MioiAKx4MhO8ZI/4BFDvlIEfcjWLCfvjXjZRIreYPys2idq8kX5
+Sb2n8ikw+YO79QfRuKmjtvXp/Ur+FROGIxb+/+OVzcZKF/An6p7oKmG4ACaBG6DP
+LOJcBiQ8TVXz9f0V7jRko1kK
+-----END PRIVATE KEY-----

apps/emqx_gcp_device/test/data/keys/c5_rsa_private.pem

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQDQPzrNyYYt1aNX
+ZLXAuq7i1qN5TEf+pKd3nnpng8pH3JeeXZhxAFlmHqB/1jlG4vOyL4Q3Tw27KtaO
+xw43XphGYlg4lofi8Q6O2zQRtaX96SX+5ofZVowHJMI5dkf2hojWGhiuv1gXIDEm
+WtI9zfoP4mzdZPKMXqav5UJk8zUytlP3seJz+8nXZT9hIOqToBg91kbt1tqVZHg0
+taKbuXsSfAeLYrAMtR7CgZfw79YOmR48PzJ2LHOqAwyCN7kQyWi114xITloptJJZ
+BYkg/Kk22bQWJylIxqGSwwbcbNnMYJ70Ia8NZbzb7qI/YDb+amxoJ7g8gDWyANeo
+bdSghU4zAgMBAAECggEAE4Q5gJvIZXdGLaSUnBFi3oN7Ip0Rij3oK//APP9O79ku
+pHrlFIIR3s40AIcVKx2N9T8axwwznzzuiscBABNvdfk1h2gkKBKraJwGjzpU6iz1
+kKQOS0IfMXQyd6wsJmCJZndfpNDt8ozjzlJorb4mF2MDDOSvDpS4TnfP9yIL9EqS
+pcHWsLQkqab5WjC7bwXvgFOIMwE32UhX/M6U3nAi8UuAanWVI2bXowywdK5f9HYU
+2TOw4TK+S773savQhczC7BAzBlNeOKguLQsO8St+4aLs/1k60qST/mcoFYgjhkXT
+iMMFrTp4kQNBfNto7LHOwLEXlT6rHGNMlYWJXzkvgQKBgQD//EDc3rMSudEKJxrU
+gzZ9D4ji+Rloa5lc4Qdg0Mxm2e2hrEgJqgPhBFO2v86t84NqtzQ+3Iu+j7o4Idor
+feEPx/74NztjQRDdU06kMGHHE6jNC1f+V0NmMgbvR26PqtZImI0FM10KGpPDjl/W
+t7w+D+XLBkjRysekkf1kYsX20wKBgQDQQkcSpevgzebGubp+cQ24mKCP8q4nyOul
+0vLK9iX05q2A4cWOQNlLVcxeV5uA2Y/aZUKMsjwcyF8xi/vDW4CzQej0fi0zQrxD
+hImhUzPDqejaRtG+qdj7u6IQN7QjWetLKbU9OPmzsZt0EZQu7B7S9ftkZzjK5Inv
+crbXPjlvIQKBgCt7MZlSyqAXqAZNdiU61HqRtPK41TQDct1v68zqKo4d3ltj5Cig
+FGCYV4/nLLgncN8jl2BGHgaUa1E1jtVsYFpJ4mlPGGtXlgHCMM162mDyWe3aS2wM
+bophXQQv4fvNTPCv2ORVQSyCLy88c9MJCpSQJrxBqQTZqOevVJdEn9O5AoGAIULk
+nQrY8G+SMx0ItxcRTPE7e6ITxJDnafWWB2pmx4VsIpBsf/rFea27VToCwQJ+YjAX
+/+abiTFLWttzm1Dq7jZRoXLhfzViYhox7Q0f0Fk7sljrONthp1rhWFu9LoQ2+ysv
+IhcOcm+kV1ZTZ2cYyTK2MuP1gxobGZ4lq5zpiWECgYB86qlXEAZP2YinZfuPETII
+RPPfTHESserJmikGxAxDk00yfWtoW8kJePKvNIPuDCe0NsMVp8PFaN08stD8Xj4k
+8gZTkasoH8kbcZXjUDRbNOM0oHWlIYLaRTfdknyh27HRbDHPukXJV/IxQGqahmBs
+K0Yh5NkZp9Rxn7iQtojCvQ==
+-----END PRIVATE KEY-----

apps/emqx_gcp_device/test/data/keys/c5_rsa_public.pem

@@ -0,0 +1,9 @@
+-----BEGIN PUBLIC KEY-----
+MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA0D86zcmGLdWjV2S1wLqu
+4tajeUxH/qSnd556Z4PKR9yXnl2YcQBZZh6gf9Y5RuLzsi+EN08NuyrWjscON16Y
+RmJYOJaH4vEOjts0EbWl/ekl/uaH2VaMByTCOXZH9oaI1hoYrr9YFyAxJlrSPc36
+D+Js3WTyjF6mr+VCZPM1MrZT97Hic/vJ12U/YSDqk6AYPdZG7dbalWR4NLWim7l7
+EnwHi2KwDLUewoGX8O/WDpkePD8ydixzqgMMgje5EMlotdeMSE5aKbSSWQWJIPyp
+Ntm0FicpSMahksMG3GzZzGCe9CGvDWW82+6iP2A2/mpsaCe4PIA1sgDXqG3UoIVO
+MwIDAQAB
+-----END PUBLIC KEY-----

apps/emqx_gcp_device/test/emqx_gcp_device_SUITE.erl

@@ -0,0 +1,390 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_gcp_device_SUITE).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+-include_lib("emqx_authn/include/emqx_authn.hrl").
+-include_lib("emqx/include/emqx.hrl").
+
+all() ->
+    emqx_common_test_helpers:all(?MODULE).
+
+init_per_suite(Config) ->
+    ok = emqx_common_test_helpers:start_apps([emqx_conf, emqx_authn, emqx_retainer, emqx_gcp_device]),
+    Config.
+
+end_per_suite(Config) ->
+    _ = emqx_common_test_helpers:stop_apps([emqx_authn, emqx_retainer, emqx_gcp_device]),
+    Config.
+
+init_per_testcase(_TestCase, Config) ->
+    {ok, _} = emqx_cluster_rpc:start_link(node(), emqx_cluster_rpc, 1000),
+    emqx_authn_test_lib:delete_authenticators(
+        [authentication],
+        ?GLOBAL
+    ),
+    clear_data(),
+    Config.
+
+end_per_testcase(_TestCase, Config) ->
+    clear_data(),
+    Config.
+
+%%--------------------------------------------------------------------
+%% Tests
+%%--------------------------------------------------------------------
+
+t_ignore_non_jwt(_Config) ->
+    ClientId = gcp_client_id(<<"clientid">>),
+    ClientInfo = client_info(ClientId, <<"non_jwt_password">>),
+    ?check_trace(
+        ?assertEqual(
+            ignore,
+            emqx_gcp_device_authn:authenticate(ClientInfo, #{})
+        ),
+        fun(Trace) ->
+            ?assertMatch(
+                [#{result := ignore, reason := "not a JWT"}],
+                ?of_kind(authn_gcp_device_check, Trace)
+            )
+        end
+    ),
+    ok.
+
+t_ignore_non_gcp_clientid(_Config) ->
+    % GCP Client pattern:
+    % projects/<project>/locations/<location>/registries/<registry>/devices/<deviceid>
+    NonGCPClientIdList = [
+        <<"non_gcp_clientid">>,
+        <<"projects/non_gcp_client">>,
+        <<"projects/proj/locations/non_gcp_client">>,
+        <<"projects/proj/locations/loc/registries/non_gcp_client">>,
+        <<"projects/proj/locations/loc/registries/reg/device/non_gcp_client">>
+    ],
+    [{_DeviceId, KeyType, PrivateKeyName, _PublicKey} | _] = keys(),
+    Payload = #{<<"exp">> => 0},
+    JWT = generate_jws(Payload, KeyType, PrivateKeyName),
+    lists:foreach(
+        fun(ClientId) ->
+            ClientInfo = client_info(ClientId, JWT),
+            ?check_trace(
+                ?assertEqual(
+                    ignore,
+                    emqx_gcp_device_authn:authenticate(ClientInfo, #{}),
+                    ClientId
+                ),
+                fun(Trace) ->
+                    ?assertMatch(
+                        [#{result := ignore, reason := "not a GCP ClientId"}],
+                        ?of_kind(authn_gcp_device_check, Trace),
+                        ClientId
+                    )
+                end
+            )
+        end,
+        NonGCPClientIdList
+    ),
+    ok.
+
+t_deny_expired_jwt(_Config) ->
+    lists:foreach(
+        fun({DeviceId, KeyType, PrivateKeyName, _PublicKey}) ->
+            ClientId = gcp_client_id(DeviceId),
+            Payload = #{<<"exp">> => 0},
+            JWT = generate_jws(Payload, KeyType, PrivateKeyName),
+            ClientInfo = client_info(ClientId, JWT),
+            ?check_trace(
+                ?assertMatch(
+                    {error, _},
+                    emqx_gcp_device_authn:authenticate(ClientInfo, #{}),
+                    DeviceId
+                ),
+                fun(Trace) ->
+                    ?assertMatch(
+                        [#{result := not_authorized, reason := "expired JWT"}],
+                        ?of_kind(authn_gcp_device_check, Trace),
+                        DeviceId
+                    )
+                end
+            )
+        end,
+        keys()
+    ),
+    ok.
+
+t_no_keys(_Config) ->
+    lists:foreach(
+        fun({DeviceId, KeyType, PrivateKeyName, _PublicKey}) ->
+            ClientId = gcp_client_id(DeviceId),
+            Payload = #{<<"exp">> => erlang:system_time(second) + 3600},
+            JWT = generate_jws(Payload, KeyType, PrivateKeyName),
+            ClientInfo = client_info(ClientId, JWT),
+            ?check_trace(
+                ?assertMatch(
+                    ignore,
+                    emqx_gcp_device_authn:authenticate(ClientInfo, #{}),
+                    DeviceId
+                ),
+                fun(Trace) ->
+                    ?assertMatch(
+                        [#{result := ignore, reason := "key not found"}],
+                        ?of_kind(authn_gcp_device_check, Trace),
+                        DeviceId
+                    )
+                end
+            )
+        end,
+        keys()
+    ),
+    ok.
+
+t_expired_keys(_Config) ->
+    lists:foreach(
+        fun({DeviceId, KeyType, PrivateKeyName, PublicKey}) ->
+            ClientId = gcp_client_id(DeviceId),
+            Device = #{
+                deviceid => DeviceId,
+                config => <<>>,
+                keys =>
+                    [
+                        #{
+                            key_type => KeyType,
+                            key => key_data(PublicKey),
+                            expires_at => erlang:system_time(second) - 3600
+                        }
+                    ]
+            },
+            ok = emqx_gcp_device:put_device(Device),
+            Payload = #{<<"exp">> => erlang:system_time(second) + 3600},
+            JWT = generate_jws(Payload, KeyType, PrivateKeyName),
+            ClientInfo = client_info(ClientId, JWT),
+            ?check_trace(
+                ?assertMatch(
+                    {error, _},
+                    emqx_gcp_device_authn:authenticate(ClientInfo, #{}),
+                    DeviceId
+                ),
+                fun(Trace) ->
+                    ?assertMatch(
+                        [
+                            #{
+                                result := {error, bad_username_or_password},
+                                reason := "no matching or valid keys"
+                            }
+                        ],
+                        ?of_kind(authn_gcp_device_check, Trace),
+                        DeviceId
+                    )
+                end
+            )
+        end,
+        keys()
+    ),
+    ok.
+
+t_valid_keys(_Config) ->
+    [
+        {DeviceId, KeyType0, PrivateKeyName0, PublicKey0},
+        {_DeviceId1, KeyType1, PrivateKeyName1, PublicKey1},
+        {_DeviceId2, KeyType2, PrivateKeyName2, _PublicKey}
+        | _
+    ] = keys(),
+    Device = #{
+        deviceid => DeviceId,
+        config => <<>>,
+        keys =>
+            [
+                #{
+                    key_type => KeyType0,
+                    key => key_data(PublicKey0),
+                    expires_at => erlang:system_time(second) + 3600
+                },
+                #{
+                    key_type => KeyType1,
+                    key => key_data(PublicKey1),
+                    expires_at => erlang:system_time(second) + 3600
+                }
+            ]
+    },
+    ok = emqx_gcp_device:put_device(Device),
+    Payload = #{<<"exp">> => erlang:system_time(second) + 3600},
+    JWT0 = generate_jws(Payload, KeyType0, PrivateKeyName0),
+    JWT1 = generate_jws(Payload, KeyType1, PrivateKeyName1),
+    JWT2 = generate_jws(Payload, KeyType2, PrivateKeyName2),
+    ClientId = gcp_client_id(DeviceId),
+    lists:foreach(
+        fun(JWT) ->
+            ?check_trace(
+                begin
+                    ClientInfo = client_info(ClientId, JWT),
+                    ?assertMatch(
+                        ok,
+                        emqx_gcp_device_authn:authenticate(ClientInfo, #{})
+                    )
+                end,
+                fun(Trace) ->
+                    ?assertMatch(
+                        [#{result := ok, reason := "auth success"}],
+                        ?of_kind(authn_gcp_device_check, Trace)
+                    )
+                end
+            )
+        end,
+        [JWT0, JWT1]
+    ),
+    ?check_trace(
+        begin
+            ClientInfo = client_info(ClientId, JWT2),
+            ?assertMatch(
+                {error, bad_username_or_password},
+                emqx_gcp_device_authn:authenticate(ClientInfo, #{})
+            )
+        end,
+        fun(Trace) ->
+            ?assertMatch(
+                [
+                    #{
+                        result := {error, bad_username_or_password},
+                        reason := "no matching or valid keys"
+                    }
+                ],
+                ?of_kind(authn_gcp_device_check, Trace)
+            )
+        end
+    ),
+    ok.
+
+t_all_key_types(_Config) ->
+    lists:foreach(
+        fun({DeviceId, KeyType, _PrivateKeyName, PublicKey}) ->
+            Device = #{
+                deviceid => DeviceId,
+                config => <<>>,
+                keys =>
+                    [
+                        #{
+                            key_type => KeyType,
+                            key => key_data(PublicKey),
+                            expires_at => 0
+                        }
+                    ]
+            },
+            ok = emqx_gcp_device:put_device(Device)
+        end,
+        keys()
+    ),
+    Payload = #{<<"exp">> => erlang:system_time(second) + 3600},
+    lists:foreach(
+        fun({DeviceId, KeyType, PrivateKeyName, _PublicKey}) ->
+            ClientId = gcp_client_id(DeviceId),
+            JWT = generate_jws(Payload, KeyType, PrivateKeyName),
+            ClientInfo = client_info(ClientId, JWT),
+            ?check_trace(
+                ?assertMatch(
+                    ok,
+                    emqx_gcp_device_authn:authenticate(ClientInfo, #{})
+                ),
+                fun(Trace) ->
+                    ?assertMatch(
+                        [#{result := ok, reason := "auth success"}],
+                        ?of_kind(authn_gcp_device_check, Trace)
+                    )
+                end
+            )
+        end,
+        keys()
+    ),
+    ok.
+
+t_config(_Config) ->
+    Device = #{
+        deviceid => <<"t">>,
+        config => base64:encode(<<"myconf">>),
+        keys => []
+    },
+    ok = emqx_gcp_device:put_device(Device),
+
+    {ok, Pid} = emqtt:start_link(),
+    {ok, _} = emqtt:connect(Pid),
+    {ok, _, _} = emqtt:subscribe(Pid, <<"/devices/t/config">>, 0),
+
+    receive
+        {publish, #{payload := <<"myconf">>}} ->
+            ok
+    after 1000 ->
+        ct:fail("No config received")
+    end,
+    emqtt:stop(Pid),
+    ok.
+
+t_wrong_device(_Config) ->
+    Device = #{wrong_field => wrong_value},
+    ?assertMatch(
+        {error, {function_clause, _}},
+        emqx_gcp_device:put_device(Device)
+    ),
+    ok.
+
+t_import_wrong_devices(_Config) ->
+    InvalidDevices = [
+        #{wrong_field => wrong_value},
+        #{another_wrong_field => another_wrong_value},
+        #{yet_another_wrong_field => yet_another_wrong_value}
+    ],
+    ValidDevices = [
+        #{
+            deviceid => gcp_client_id(<<"valid_device_1">>),
+            config => <<>>,
+            keys => []
+        },
+        #{
+            deviceid => gcp_client_id(<<"valid_device_2">>),
+            config => <<>>,
+            keys => []
+        }
+    ],
+    Devices = InvalidDevices ++ ValidDevices,
+    InvalidDevicesLength = length(InvalidDevices),
+    ValidDevicesLength = length(ValidDevices),
+    ?assertMatch(
+        {ValidDevicesLength, InvalidDevicesLength},
+        emqx_gcp_device:import_devices(Devices)
+    ),
+    ok.
+
+%%--------------------------------------------------------------------
+%% Helpers
+%%--------------------------------------------------------------------
+
+client_info(ClientId, Password) ->
+    emqx_gcp_device_test_helpers:client_info(ClientId, Password).
+
+device_loc(DeviceId) ->
+    {<<"iot-export">>, <<"europe-west1">>, <<"my-registry">>, DeviceId}.
+
+gcp_client_id(DeviceId) ->
+    emqx_gcp_device_test_helpers:client_id(DeviceId).
+
+keys() ->
+    emqx_gcp_device_test_helpers:keys().
+
+key_data(Filename) ->
+    emqx_gcp_device_test_helpers:key(Filename).
+
+generate_jws(Payload, KeyType, PrivateKeyName) ->
+    emqx_gcp_device_test_helpers:generate_jws(Payload, KeyType, PrivateKeyName).
+
+clear_data() ->
+    emqx_gcp_device_test_helpers:clear_data(),
+    emqx_authn_test_lib:delete_authenticators(
+        [authentication],
+        ?GLOBAL
+    ),
+    ok.

apps/emqx_gcp_device/test/emqx_gcp_device_api_SUITE.erl

@@ -0,0 +1,327 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_gcp_device_api_SUITE).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+-include_lib("emqx_authn/include/emqx_authn.hrl").
+-include_lib("emqx/include/emqx.hrl").
+
+-define(PATH, [authentication]).
+-define(BASE_CONF, <<
+    ""
+    "\n"
+    "retainer {\n"
+    "    enable = true\n"
+    "}"
+    ""
+>>).
+
+all() ->
+    emqx_common_test_helpers:all(?MODULE).
+
+init_per_suite(Config) ->
+    ok = emqx_config:init_load(emqx_retainer_schema, ?BASE_CONF),
+    ok = emqx_common_test_helpers:start_apps([emqx_gcp_device, emqx_authn, emqx_conf, emqx_retainer]),
+    emqx_dashboard_api_test_helpers:set_default_config(),
+    emqx_mgmt_api_test_util:init_suite(),
+    Config.
+
+end_per_suite(Config) ->
+    emqx_mgmt_api_test_util:end_suite(),
+    _ = emqx_common_test_helpers:stop_apps([emqx_authn, emqx_retainer, emqx_gcp_device]),
+    Config.
+
+init_per_testcase(_TestCase, Config) ->
+    {ok, _} = emqx_cluster_rpc:start_link(node(), emqx_cluster_rpc, 1000),
+    emqx_authn_test_lib:delete_authenticators(
+        [authentication],
+        ?GLOBAL
+    ),
+    clear_data(),
+    Config.
+
+end_per_testcase(_TestCase, Config) ->
+    clear_data(),
+    Config.
+
+%%--------------------------------------------------------------------
+%% Tests
+%%--------------------------------------------------------------------
+
+t_import(_Config) ->
+    ?assertMatch(
+        {ok, #{<<"errors">> := 0, <<"imported">> := 14}},
+        api(post, ["gcp_devices"], emqx_gcp_device_test_helpers:exported_data())
+    ),
+
+    InvalidData =
+        [
+            #{<<"deviceid">> => <<"device1">>, <<"device_numid">> => <<"device1">>},
+            #{<<"name">> => []}
+        ],
+    ?assertMatch({error, {_, 400, _}}, api(post, ["gcp_devices"], InvalidData)),
+
+    ?assertMatch(
+        {ok, #{<<"meta">> := #{<<"count">> := 14}}},
+        api(get, ["gcp_devices"])
+    ),
+
+    ?assertMatch(
+        {ok, #{
+            <<"meta">> :=
+                #{
+                    <<"count">> := 14,
+                    <<"page">> := 2,
+                    <<"limit">> := 3
+                }
+        }},
+        api(get, ["gcp_devices"], [{"limit", "3"}, {"page", "2"}])
+    ).
+
+t_device_crud_ok(_Config) ->
+    AuthConfig = raw_config(),
+    DeviceId = <<"my device">>,
+    DeviceIdReq = emqx_http_lib:uri_encode(DeviceId),
+    ConfigTopic = emqx_gcp_device:config_topic(DeviceId),
+    DeviceConfig = <<"myconfig">>,
+    EncodedConfig = base64:encode(DeviceConfig),
+    {ok, _} = emqx:update_config(?PATH, {create_authenticator, ?GLOBAL, AuthConfig}),
+
+    Payload = #{<<"exp">> => erlang:system_time(second) + 3600},
+    JWT = generate_jws(Payload, <<"ES256_PEM">>, "c1_ec_private.pem"),
+    ClientInfo = client_info(client_id(DeviceId), JWT),
+    ?assertMatch(
+        {error, _},
+        emqx_access_control:authenticate(ClientInfo)
+    ),
+    Device0 =
+        #{
+            <<"project">> => <<"iot-export">>,
+            <<"location">> => <<"europe-west1">>,
+            <<"registry">> => <<"my-registry">>,
+            <<"keys">> =>
+                [
+                    #{
+                        <<"key">> => emqx_gcp_device_test_helpers:key("c1_ec_public.pem"),
+                        <<"key_type">> => <<"ES256_PEM">>,
+                        <<"expires_at">> => 0
+                    },
+                    #{
+                        <<"key">> => emqx_gcp_device_test_helpers:key("c1_ec_public.pem"),
+                        <<"key_type">> => <<"ES256_PEM">>,
+                        <<"expires_at">> => 0
+                    }
+                ],
+            <<"config">> => EncodedConfig
+        },
+    ?assertMatch(
+        {ok, #{<<"deviceid">> := DeviceId}},
+        api(put, ["gcp_devices", DeviceIdReq], Device0)
+    ),
+    ?assertMatch(
+        {ok, _},
+        emqx_access_control:authenticate(ClientInfo)
+    ),
+
+    ?retry(
+        _Sleep = 100,
+        _Attempts = 10,
+        ?assertMatch(
+            {ok, [#message{payload = DeviceConfig}]},
+            emqx_retainer:read_message(ConfigTopic)
+        )
+    ),
+    ?assertMatch(
+        {ok, #{
+            <<"project">> := <<"iot-export">>,
+            <<"location">> := <<"europe-west1">>,
+            <<"registry">> := <<"my-registry">>,
+            <<"keys">> :=
+                [
+                    #{
+                        <<"key">> := _,
+                        <<"key_type">> := <<"ES256_PEM">>,
+                        <<"expires_at">> := 0
+                    },
+                    #{
+                        <<"key">> := _,
+                        <<"key_type">> := <<"ES256_PEM">>,
+                        <<"expires_at">> := 0
+                    }
+                ],
+            <<"config">> := EncodedConfig
+        }},
+        api(get, ["gcp_devices", DeviceIdReq])
+    ),
+
+    Device1 = maps:without([<<"project">>, <<"location">>, <<"registry">>], Device0),
+    ?assertMatch(
+        {ok, #{<<"deviceid">> := DeviceId}},
+        api(put, ["gcp_devices", DeviceIdReq], Device1)
+    ),
+
+    ?assertMatch(
+        {ok, #{
+            <<"project">> := <<>>,
+            <<"location">> := <<>>,
+            <<"registry">> := <<>>
+        }},
+        api(get, ["gcp_devices", DeviceIdReq])
+    ),
+    ?assertMatch({ok, {{_, 204, _}, _, _}}, api(delete, ["gcp_devices", DeviceIdReq])),
+
+    ?retry(
+        _Sleep = 100,
+        _Attempts = 10,
+        ?assertNotMatch(
+            {ok, [#message{payload = DeviceConfig}]},
+            emqx_retainer:read_message(ConfigTopic)
+        )
+    ),
+    ?assertMatch({error, {_, 404, _}}, api(get, ["gcp_devices", DeviceIdReq])).
+
+t_device_crud_nok(_Config) ->
+    DeviceId = <<"my device">>,
+    DeviceIdReq = emqx_http_lib:uri_encode(DeviceId),
+    Config = <<"myconfig">>,
+    EncodedConfig = base64:encode(Config),
+
+    BadDevices =
+        [
+            #{
+                <<"project">> => 5,
+                <<"keys">> => [],
+                <<"config">> => EncodedConfig
+            },
+            #{
+                <<"keys">> => <<"keys">>,
+                <<"config">> => EncodedConfig
+            },
+            #{
+                <<"keys">> => [<<"key">>],
+                <<"config">> => EncodedConfig
+            },
+            #{
+                <<"keys">> => [#{<<"key">> => <<"key">>}],
+                <<"config">> => EncodedConfig
+            },
+            #{
+                <<"keys">> => [#{<<"key_type">> => <<"ES256_PEM">>}],
+                <<"config">> => EncodedConfig
+            },
+            #{
+                <<"keys">> =>
+                    [
+                        #{
+                            <<"key">> => <<"key">>,
+                            <<"key_type">> => <<"ES256_PEM">>,
+                            <<"expires_at">> => <<"123">>
+                        }
+                    ],
+                <<"config">> => EncodedConfig
+            }
+        ],
+
+    lists:foreach(
+        fun(BadDevice) ->
+            ?assertMatch(
+                {error, {_, 400, _}},
+                api(put, ["gcp_devices", DeviceIdReq], BadDevice)
+            )
+        end,
+        BadDevices
+    ).
+
+%%--------------------------------------------------------------------
+%% Helpers
+%%--------------------------------------------------------------------
+
+assert_no_retained(ConfigTopic) ->
+    {ok, Pid} = emqtt:start_link(),
+    {ok, _} = emqtt:connect(Pid),
+    {ok, _, _} = emqtt:subscribe(Pid, ConfigTopic, 0),
+
+    receive
+        {publish, #{payload := Config}} ->
+            ct:fail("Unexpected config received: ~p", [Config])
+    after 100 ->
+        ok
+    end,
+
+    _ = emqtt:stop(Pid).
+
+api(get, Path) ->
+    api(get, Path, "");
+api(delete, Path) ->
+    api(delete, Path, []).
+
+api(get, Path, Query) ->
+    maybe_decode_response(
+        emqx_mgmt_api_test_util:request_api(
+            get,
+            emqx_mgmt_api_test_util:api_path(Path),
+            uri_string:compose_query(Query),
+            emqx_mgmt_api_test_util:auth_header_()
+        )
+    );
+api(delete, Path, Query) ->
+    emqx_mgmt_api_test_util:request_api(
+        delete,
+        emqx_mgmt_api_test_util:api_path(Path),
+        uri_string:compose_query(Query),
+        emqx_mgmt_api_test_util:auth_header_(),
+        [],
+        #{return_all => true}
+    );
+api(Method, Path, Data) when
+    Method =:= put orelse Method =:= post
+->
+    api(Method, Path, [], Data).
+
+api(Method, Path, Query, Data) when
+    Method =:= put orelse Method =:= post
+->
+    maybe_decode_response(
+        emqx_mgmt_api_test_util:request_api(
+            Method,
+            emqx_mgmt_api_test_util:api_path(Path),
+            uri_string:compose_query(Query),
+            emqx_mgmt_api_test_util:auth_header_(),
+            Data
+        )
+    ).
+
+maybe_decode_response({ok, ResponseBody}) ->
+    {ok, jiffy:decode(list_to_binary(ResponseBody), [return_maps])};
+maybe_decode_response({error, _} = Error) ->
+    Error.
+
+generate_jws(Payload, KeyType, PrivateKeyName) ->
+    emqx_gcp_device_test_helpers:generate_jws(Payload, KeyType, PrivateKeyName).
+
+client_info(ClientId, Password) ->
+    emqx_gcp_device_test_helpers:client_info(ClientId, Password).
+
+client_id(DeviceId) ->
+    emqx_gcp_device_test_helpers:client_id(DeviceId).
+
+raw_config() ->
+    #{
+        <<"mechanism">> => <<"gcp_device">>,
+        <<"enable">> => <<"true">>
+    }.
+
+clear_data() ->
+    emqx_gcp_device_test_helpers:clear_data(),
+    emqx_authn_test_lib:delete_authenticators(
+        [authentication],
+        ?GLOBAL
+    ),
+    ok.

apps/emqx_gcp_device/test/emqx_gcp_device_authn_SUITE.erl

@@ -0,0 +1,175 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_gcp_device_authn_SUITE).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+-include_lib("emqx_authn/include/emqx_authn.hrl").
+
+-define(PATH, [authentication]).
+-define(DEVICE_ID, <<"test-device">>).
+-define(PROJECT, <<"iot-export">>).
+-define(LOCATION, <<"europe-west1">>).
+-define(REGISTRY, <<"my-registry">>).
+
+all() ->
+    emqx_common_test_helpers:all(?MODULE).
+
+init_per_suite(Config0) ->
+    ok = snabbkaffe:start_trace(),
+    emqx_common_test_helpers:start_apps([emqx_conf, emqx_authn, emqx_gcp_device]),
+    ValidExpirationTime = erlang:system_time(second) + 3600,
+    ValidJWT = generate_jws(ValidExpirationTime),
+    ExpiredJWT = generate_jws(0),
+    ValidClient = generate_client(ValidExpirationTime),
+    ExpiredClient = generate_client(0),
+    [
+        {device_id, ?DEVICE_ID},
+        {client_id, client_id()},
+        {valid_jwt, ValidJWT},
+        {expired_jwt, ExpiredJWT},
+        {valid_client, ValidClient},
+        {expired_client, ExpiredClient}
+        | Config0
+    ].
+
+end_per_suite(_) ->
+    _ = emqx_common_test_helpers:stop_apps([emqx_authn, emqx_gcp_device]),
+    ok.
+
+init_per_testcase(_, Config) ->
+    {ok, _} = emqx_cluster_rpc:start_link(node(), emqx_cluster_rpc, 1000),
+    Config.
+
+end_per_testcase(_Case, Config) ->
+    emqx_authn_test_lib:delete_authenticators(
+        [authentication],
+        ?GLOBAL
+    ),
+    Config.
+
+%%------------------------------------------------------------------------------
+%% Tests
+%%------------------------------------------------------------------------------
+
+t_create(_Config) ->
+    AuthConfig = raw_config(),
+    {ok, _} = emqx:update_config(?PATH, {create_authenticator, ?GLOBAL, AuthConfig}),
+    ?assertMatch(
+        {ok, [#{provider := emqx_gcp_device_authn}]},
+        emqx_authentication:list_authenticators(?GLOBAL)
+    ).
+
+t_destroy(Config) ->
+    ClientId = ?config(client_id, Config),
+    JWT = ?config(valid_jwt, Config),
+    Credential = credential(ClientId, JWT),
+    Client = ?config(valid_client, Config),
+    AuthConfig = raw_config(),
+    {ok, _} = emqx:update_config(?PATH, {create_authenticator, ?GLOBAL, AuthConfig}),
+    ok = emqx_gcp_device:put_device(Client),
+    ?assertMatch(
+        {ok, _},
+        emqx_access_control:authenticate(Credential)
+    ),
+    emqx_authn_test_lib:delete_authenticators([authentication], ?GLOBAL),
+    ?assertMatch(
+        ignore,
+        emqx_gcp_device_authn:authenticate(Credential, #{})
+    ).
+
+t_expired_client(Config) ->
+    ClientId = ?config(client_id, Config),
+    JWT = ?config(expired_jwt, Config),
+    Credential = credential(ClientId, JWT),
+    Client = ?config(expired_client, Config),
+    AuthConfig = raw_config(),
+    {ok, _} = emqx:update_config(?PATH, {create_authenticator, ?GLOBAL, AuthConfig}),
+    ?assertMatch(
+        {ok, [#{provider := emqx_gcp_device_authn}]},
+        emqx_authentication:list_authenticators(?GLOBAL)
+    ),
+    ok = emqx_gcp_device:put_device(Client),
+    ?assertMatch(
+        {error, not_authorized},
+        emqx_access_control:authenticate(Credential)
+    ).
+
+%%------------------------------------------------------------------------------
+%% Helpers
+%%------------------------------------------------------------------------------
+
+raw_config() ->
+    #{
+        <<"mechanism">> => <<"gcp_device">>,
+        <<"enable">> => <<"true">>
+    }.
+
+generate_client(ExpirationTime) ->
+    generate_client(?DEVICE_ID, ExpirationTime).
+
+generate_client(ClientId, ExpirationTime) ->
+    #{
+        deviceid => ClientId,
+        project => ?PROJECT,
+        location => ?LOCATION,
+        registry => ?REGISTRY,
+        config => <<>>,
+        keys =>
+            [
+                #{
+                    key_type => <<"RSA_PEM">>,
+                    key => public_key(),
+                    expires_at => ExpirationTime
+                }
+            ]
+    }.
+
+client_id() ->
+    client_id(?DEVICE_ID).
+
+client_id(DeviceId) ->
+    <<"projects/", ?PROJECT/binary, "/locations/", ?LOCATION/binary, "/registries/",
+        ?REGISTRY/binary, "/devices/", DeviceId/binary>>.
+
+generate_jws(ExpirationTime) ->
+    Payload = #{<<"exp">> => ExpirationTime},
+    JWK = jose_jwk:from_pem_file(test_rsa_key(private)),
+    Header = #{<<"alg">> => <<"RS256">>, <<"typ">> => <<"JWT">>},
+    Signed = jose_jwt:sign(JWK, Header, Payload),
+    {_, JWS} = jose_jws:compact(Signed),
+    JWS.
+
+public_key() ->
+    {ok, Data} = file:read_file(test_rsa_key(public)),
+    Data.
+
+private_key() ->
+    {ok, Data} = file:read_file(test_rsa_key(private)),
+    Data.
+
+test_rsa_key(public) ->
+    data_file("public_key.pem");
+test_rsa_key(private) ->
+    data_file("private_key.pem").
+
+data_file(Name) ->
+    Dir = code:lib_dir(emqx_authn, test),
+    list_to_binary(filename:join([Dir, "data", Name])).
+
+credential(ClientId, JWT) ->
+    #{
+        listener => 'tcp:default',
+        protocol => mqtt,
+        clientid => ClientId,
+        password => JWT
+    }.
+
+check(Module, HoconConf) ->
+    emqx_hocon:check(Module, ["authentication= ", HoconConf]).

apps/emqx_gcp_device/test/emqx_gcp_device_test_helpers.erl

@@ -0,0 +1,66 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_gcp_device_test_helpers).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-define(KEYS, [
+    {<<"c1-ec">>, <<"ES256_PEM">>, <<"c1_ec_private.pem">>, <<"c1_ec_public.pem">>},
+    {<<"c2-ec-x509">>, <<"ES256_X509_PEM">>, <<"c2_ec_private.pem">>, <<"c2_ec_cert.pem">>},
+    {<<"c3-rsa">>, <<"RSA_PEM">>, <<"c3_rsa_private.pem">>, <<"c3_rsa_public.pem">>},
+    {<<"c4-rsa-x509">>, <<"RSA_X509_PEM">>, <<"c4_rsa_private.pem">>, <<"c4_rsa_cert.pem">>}
+]).
+
+exported_data() ->
+    FileName =
+        filename:join([code:lib_dir(emqx_gcp_device), "test", "data", "gcp-data.json"]),
+    {ok, Data} = file:read_file(FileName),
+    jiffy:decode(Data, [return_maps]).
+
+key(Name) ->
+    {ok, Data} = file:read_file(key_path(Name)),
+    Data.
+
+key_path(Name) ->
+    filename:join([code:lib_dir(emqx_gcp_device), "test", "data", "keys", Name]).
+
+clear_data() ->
+    {atomic, ok} = mria:clear_table(emqx_gcp_device),
+    ok = emqx_retainer:clean(),
+    ok.
+
+keys() ->
+    ?KEYS.
+
+client_id(DeviceId) ->
+    <<"projects/iot-export/locations/europe-west1/registries/my-registry/devices/",
+        DeviceId/binary>>.
+
+generate_jws(Payload, KeyType, PrivateKeyName) ->
+    JWK = jose_jwk:from_pem_file(
+        emqx_gcp_device_test_helpers:key_path(PrivateKeyName)
+    ),
+    Header = #{<<"alg">> => alg(KeyType), <<"typ">> => <<"JWT">>},
+    Signed = jose_jwt:sign(JWK, Header, Payload),
+    {_, JWS} = jose_jws:compact(Signed),
+    JWS.
+
+alg(<<"ES256_PEM">>) ->
+    <<"ES256">>;
+alg(<<"ES256_X509_PEM">>) ->
+    <<"ES256">>;
+alg(<<"RSA_PEM">>) ->
+    <<"RS256">>;
+alg(<<"RSA_X509_PEM">>) ->
+    <<"RS256">>.
+
+client_info(ClientId, JWT) ->
+    #{
+        listener => 'tcp:default',
+        protocol => mqtt,
+        clientid => ClientId,
+        password => JWT
+    }.

apps/emqx_machine/priv/reboot_lists.eterm

@@ -113,7 +113,8 @@
             emqx_eviction_agent,
             emqx_node_rebalance,
             emqx_ft,
-            emqx_ldap
+            emqx_ldap,
+            emqx_gcp_device
         ],
     %% must always be of type `load'
     ce_business_apps =>

apps/emqx_retainer/src/emqx_retainer.erl

@@ -40,6 +40,7 @@
     update_config/1,
     clean/0,
     delete/1,
+    read_message/1,
     page_read/3,
     post_config_update/5,
     stats_fun/0,
@@ -157,6 +158,9 @@ delete(Topic) ->
 retained_count() ->
     call(?FUNCTION_NAME).
 
+read_message(Topic) ->
+    call({?FUNCTION_NAME, Topic}).
+
 page_read(Topic, Page, Limit) ->
     call({?FUNCTION_NAME, Topic, Page, Limit}).
 
@@ -210,6 +214,10 @@ handle_call(clean, _, #{context := Context} = State) ->
 handle_call({delete, Topic}, _, #{context := Context} = State) ->
     delete_message(Context, Topic),
     {reply, ok, State};
+handle_call({read_message, Topic}, _, #{context := Context} = State) ->
+    Mod = get_backend_module(),
+    Result = Mod:read_message(Context, Topic),
+    {reply, Result, State};
 handle_call({page_read, Topic, Page, Limit}, _, #{context := Context} = State) ->
     Mod = get_backend_module(),
     Result = Mod:page_read(Context, Topic, Page, Limit),

apps/emqx_retainer/test/emqx_retainer_SUITE.erl

@@ -135,9 +135,17 @@ t_store_and_clean(_) ->
 
     {ok, List} = emqx_retainer:page_read(<<"retained">>, 1, 10),
     ?assertEqual(1, length(List)),
+    ?assertMatch(
+        {ok, [#message{payload = <<"this is a retained message">>}]},
+        emqx_retainer:read_message(<<"retained">>)
+    ),
 
     {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained">>, [{qos, 0}, {rh, 0}]),
     ?assertEqual(1, length(receive_messages(1))),
+    ?assertMatch(
+        {ok, [#message{payload = <<"this is a retained message">>}]},
+        emqx_retainer:read_message(<<"retained">>)
+    ),
 
     {ok, #{}, [0]} = emqtt:unsubscribe(C1, <<"retained">>),
 
@@ -145,10 +153,18 @@ t_store_and_clean(_) ->
     timer:sleep(100),
     {ok, #{}, [0]} = emqtt:subscribe(C1, <<"retained">>, [{qos, 0}, {rh, 0}]),
     ?assertEqual(0, length(receive_messages(1))),
+    ?assertMatch(
+        {ok, []},
+        emqx_retainer:read_message(<<"retained">>)
+    ),
 
     ok = emqx_retainer:clean(),
     {ok, List2} = emqx_retainer:page_read(<<"retained">>, 1, 10),
     ?assertEqual(0, length(List2)),
+    ?assertMatch(
+        {ok, []},
+        emqx_retainer:read_message(<<"retained">>)
+    ),
 
     ok = emqtt:disconnect(C1).
 

changes/ee/feat-11367.en.md

@@ -0,0 +1 @@
+Ported GCP IoT Hub authentication support.

mix.exs

@@ -220,7 +220,8 @@ defmodule EMQXUmbrella.MixProject do
       :emqx_enterprise,
       :emqx_bridge_kinesis,
       :emqx_bridge_azure_event_hub,
-      :emqx_ldap
+      :emqx_ldap,
+      :emqx_gcp_device
     ])
   end
 

rebar.config.erl

@@ -107,6 +107,7 @@ is_community_umbrella_app("apps/emqx_enterprise") -> false;
 is_community_umbrella_app("apps/emqx_bridge_kinesis") -> false;
 is_community_umbrella_app("apps/emqx_bridge_azure_event_hub") -> false;
 is_community_umbrella_app("apps/emqx_ldap") -> false;
+is_community_umbrella_app("apps/emqx_gcp_device") -> false;
 is_community_umbrella_app(_) -> true.
 
 is_jq_supported() ->

rel/i18n/emqx_gcp_device_api.hocon

@@ -0,0 +1,95 @@
+emqx_gcp_device_api {
+
+gcp_device.desc:
+"""Configuration of authenticator using GCP Device as authentication data source."""
+
+gcp_devices_get.desc:
+"""List all devices imported from GCP IoT Core"""
+gcp_devices_get.label:
+"""List all GCP devices"""
+
+gcp_devices_post.desc:
+"""Import authentication and config data for devices from GCP IoT Core"""
+gcp_devices_post.label:
+"""Import GCP devices"""
+
+gcp_device_get.desc:
+"""Get a device imported from GCP IoT Core"""
+gcp_device_get.label:
+"""Get GCP device"""
+
+gcp_device_put.desc:
+"""Update a device imported from GCP IoT Core"""
+gcp_device_put.label:
+"""Update GCP device"""
+
+gcp_device_delete.desc:
+"""Remove a device imported from GCP IoT Core"""
+gcp_device_delete.label:
+"""Remove GCP device"""
+
+project.desc:
+"""Cloud project identifier"""
+project.label:
+"""Project"""
+
+location.desc:
+"""Cloud region"""
+location.label:
+"""Region"""
+
+registry.desc:
+"""Device registry identifier"""
+registry.label:
+"""Registry"""
+
+deviceid.label:
+"""Device identifier"""
+deviceid.desc:
+"""Device identifier"""
+
+keys.desc:
+"""Public keys associated to GCP device"""
+keys.label:
+"""Public keys"""
+
+key.desc:
+"""Public key"""
+key.label:
+"""Public key"""
+
+key_type.desc:
+"""Public key type"""
+key_type.label:
+"""Public key type"""
+
+expires_at.desc:
+"""Public key expiration time"""
+expires_at.label:
+"""Expiration time"""
+
+created_at.desc:
+"""Time when GCP device was imported"""
+created_at.label:
+"""Creation time"""
+
+config.label:
+"""Device configuration"""
+config.desc:
+"""Configuration"""
+
+blocked.label:
+"""If device is blocked from communicating to GCP IoT Core"""
+blocked.desc:
+"""Blocked"""
+
+gcp_device_response404.desc:
+"""The GCP device was not found"""
+
+imported_counter.desc:
+"""Number of successfully imported GCP devices"""
+
+imported_counter_errors.desc:
+"""Number of GCP devices not imported due to some error"""
+
+}