// Copyright 2015 The Weave Authors. All rights reserved. // Use of this source code is governed by a BSD-style license that can be // found in the LICENSE file. #include "src/device_registration_info.h" #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "src/bind_lambda.h" #include "src/commands/cloud_command_proxy.h" #include "src/commands/schema_constants.h" #include "src/data_encoding.h" #include "src/http_constants.h" #include "src/json_error_codes.h" #include "src/notification/xmpp_channel.h" #include "src/privet/auth_manager.h" #include "src/string_utils.h" #include "src/utils.h" namespace weave { const char kErrorAlreayRegistered[] = "already_registered"; namespace { const int kPollingPeriodSeconds = 7; const int kBackupPollingPeriodMinutes = 30; namespace fetch_reason { const char kDeviceStart[] = "device_start"; // Initial queue fetch at startup. const char kRegularPull[] = "regular_pull"; // Regular fetch before XMPP is up. const char kNewCommand[] = "new_command"; // A new command is available. const char kJustInCase[] = "just_in_case"; // Backup fetch when XMPP is live. } // namespace fetch_reason using provider::HttpClient; inline void SetUnexpectedError(ErrorPtr* error) { Error::AddTo(error, FROM_HERE, "unexpected_response", "Unexpected GCD error"); } void ParseGCDError(const base::DictionaryValue* json, ErrorPtr* error) { const base::Value* list_value = nullptr; const base::ListValue* error_list = nullptr; if (!json->Get("error.errors", &list_value) || !list_value->GetAsList(&error_list)) { SetUnexpectedError(error); return; } for (size_t i = 0; i < error_list->GetSize(); i++) { const base::Value* error_value = nullptr; const base::DictionaryValue* error_object = nullptr; if (!error_list->Get(i, &error_value) || !error_value->GetAsDictionary(&error_object)) { SetUnexpectedError(error); continue; } std::string error_code, error_message; if (error_object->GetString("reason", &error_code) && error_object->GetString("message", &error_message)) { Error::AddTo(error, FROM_HERE, error_code, error_message); } else { SetUnexpectedError(error); } } } std::string AppendQueryParams(const std::string& url, const WebParamList& params) { CHECK_EQ(std::string::npos, url.find_first_of("?#")); if (params.empty()) return url; return url + '?' + WebParamsEncode(params); } std::string BuildURL(const std::string& url, const std::string& subpath, const WebParamList& params) { std::string result = url; if (!result.empty() && result.back() != '/' && !subpath.empty()) { CHECK_NE('/', subpath.front()); result += '/'; } result += subpath; return AppendQueryParams(result, params); } void IgnoreCloudErrorWithCallback(const base::Closure& cb, ErrorPtr) { cb.Run(); } void IgnoreCloudError(ErrorPtr) {} void IgnoreCloudResult(const base::DictionaryValue&, ErrorPtr error) {} void IgnoreCloudResultWithCallback(const DoneCallback& cb, const base::DictionaryValue&, ErrorPtr error) { cb.Run(std::move(error)); } class RequestSender final { public: RequestSender(HttpClient::Method method, const std::string& url, HttpClient* transport) : method_{method}, url_{url}, transport_{transport} {} void Send(const HttpClient::SendRequestCallback& callback) { static int debug_id = 0; ++debug_id; VLOG(1) << "Sending request. id:" << debug_id << " method:" << EnumToString(method_) << " url:" << url_; VLOG(2) << "Request data: " << data_; auto on_done = []( int debug_id, const HttpClient::SendRequestCallback& callback, std::unique_ptr response, ErrorPtr error) { if (error) { VLOG(1) << "Request failed, id=" << debug_id << ", reason: " << error->GetCode() << ", message: " << error->GetMessage(); return callback.Run({}, std::move(error)); } VLOG(1) << "Request succeeded. id:" << debug_id << " status:" << response->GetStatusCode(); VLOG(2) << "Response data: " << response->GetData(); callback.Run(std::move(response), nullptr); }; transport_->SendRequest(method_, url_, GetFullHeaders(), data_, base::Bind(on_done, debug_id, callback)); } void SetAccessToken(const std::string& access_token) { access_token_ = access_token; } void SetData(const std::string& data, const std::string& mime_type) { data_ = data; mime_type_ = mime_type; } void SetFormData( const std::vector>& data) { SetData(WebParamsEncode(data), http::kWwwFormUrlEncoded); } void SetJsonData(const base::Value& json) { std::string data; CHECK(base::JSONWriter::Write(json, &data)); SetData(data, http::kJsonUtf8); } private: HttpClient::Headers GetFullHeaders() const { HttpClient::Headers headers; if (!access_token_.empty()) headers.emplace_back(http::kAuthorization, "Bearer " + access_token_); if (!mime_type_.empty()) headers.emplace_back(http::kContentType, mime_type_); return headers; } HttpClient::Method method_; std::string url_; std::string data_; std::string mime_type_; std::string access_token_; HttpClient* transport_{nullptr}; DISALLOW_COPY_AND_ASSIGN(RequestSender); }; std::unique_ptr ParseJsonResponse( const HttpClient::Response& response, ErrorPtr* error) { // Make sure we have a correct content type. Do not try to parse // binary files, or HTML output. Limit to application/json and text/plain. std::string content_type = SplitAtFirst(response.GetContentType(), ";", true).first; if (content_type != http::kJson && content_type != http::kPlain) { return Error::AddTo( error, FROM_HERE, "non_json_content_type", "Unexpected content type: \'" + response.GetContentType() + "\'"); } const std::string& json = response.GetData(); std::string error_message; auto value = base::JSONReader::ReadAndReturnError(json, base::JSON_PARSE_RFC, nullptr, &error_message); if (!value) { Error::AddToPrintf(error, FROM_HERE, errors::json::kParseError, "Error '%s' occurred parsing JSON string '%s'", error_message.c_str(), json.c_str()); return std::unique_ptr(); } base::DictionaryValue* dict_value = nullptr; if (!value->GetAsDictionary(&dict_value)) { Error::AddToPrintf(error, FROM_HERE, errors::json::kObjectExpected, "Response is not a valid JSON object: '%s'", json.c_str()); return std::unique_ptr(); } else { // |value| is now owned by |dict_value|, so release the scoped_ptr now. base::IgnoreResult(value.release()); } return std::unique_ptr(dict_value); } bool IsSuccessful(const HttpClient::Response& response) { int code = response.GetStatusCode(); return code >= http::kContinue && code < http::kBadRequest; } } // anonymous namespace DeviceRegistrationInfo::DeviceRegistrationInfo( Config* config, ComponentManager* component_manager, provider::TaskRunner* task_runner, provider::HttpClient* http_client, provider::Network* network, privet::AuthManager* auth_manager) : http_client_{http_client}, task_runner_{task_runner}, config_{config}, component_manager_{component_manager}, network_{network}, auth_manager_{auth_manager} { cloud_backoff_policy_.reset(new BackoffEntry::Policy{}); cloud_backoff_policy_->num_errors_to_ignore = 0; cloud_backoff_policy_->initial_delay_ms = 1000; cloud_backoff_policy_->multiply_factor = 2.0; cloud_backoff_policy_->jitter_factor = 0.1; cloud_backoff_policy_->maximum_backoff_ms = 30000; cloud_backoff_policy_->entry_lifetime_ms = -1; cloud_backoff_policy_->always_use_initial_delay = false; cloud_backoff_entry_.reset(new BackoffEntry{cloud_backoff_policy_.get()}); oauth2_backoff_entry_.reset(new BackoffEntry{cloud_backoff_policy_.get()}); bool revoked = !GetSettings().cloud_id.empty() && !HaveRegistrationCredentials(); gcd_state_ = revoked ? GcdState::kInvalidCredentials : GcdState::kUnconfigured; component_manager_->AddTraitDefChangedCallback(base::Bind( &DeviceRegistrationInfo::OnTraitDefsChanged, weak_factory_.GetWeakPtr())); component_manager_->AddComponentTreeChangedCallback( base::Bind(&DeviceRegistrationInfo::OnComponentTreeChanged, weak_factory_.GetWeakPtr())); component_manager_->AddStateChangedCallback(base::Bind( &DeviceRegistrationInfo::OnStateChanged, weak_factory_.GetWeakPtr())); } DeviceRegistrationInfo::~DeviceRegistrationInfo() = default; std::string DeviceRegistrationInfo::GetServiceURL( const std::string& subpath, const WebParamList& params) const { return BuildURL(GetSettings().service_url, subpath, params); } std::string DeviceRegistrationInfo::GetDeviceURL( const std::string& subpath, const WebParamList& params) const { CHECK(!GetSettings().cloud_id.empty()) << "Must have a valid device ID"; return BuildURL(GetSettings().service_url, "devices/" + GetSettings().cloud_id + "/" + subpath, params); } std::string DeviceRegistrationInfo::GetOAuthURL( const std::string& subpath, const WebParamList& params) const { return BuildURL(GetSettings().oauth_url, subpath, params); } void DeviceRegistrationInfo::Start() { if (HaveRegistrationCredentials()) { StartNotificationChannel(); // Wait a significant amount of time for local daemons to publish their // state to Buffet before publishing it to the cloud. // TODO(wiley) We could do a lot of things here to either expose this // timeout as a configurable knob or allow local // daemons to signal that their state is up to date so that // we need not wait for them. ScheduleCloudConnection(base::TimeDelta::FromSeconds(5)); } } void DeviceRegistrationInfo::ScheduleCloudConnection( const base::TimeDelta& delay) { SetGcdState(GcdState::kConnecting); if (!task_runner_) return; // Assume we're in test task_runner_->PostDelayedTask( FROM_HERE, base::Bind(&DeviceRegistrationInfo::ConnectToCloud, AsWeakPtr(), nullptr), delay); } bool DeviceRegistrationInfo::HaveRegistrationCredentials() const { return !GetSettings().refresh_token.empty() && !GetSettings().cloud_id.empty() && !GetSettings().robot_account.empty(); } bool DeviceRegistrationInfo::VerifyRegistrationCredentials( ErrorPtr* error) const { const bool have_credentials = HaveRegistrationCredentials(); VLOG(2) << "Device registration record " << ((have_credentials) ? "found" : "not found."); if (!have_credentials) { return Error::AddTo(error, FROM_HERE, "device_not_registered", "No valid device registration record found"); } return true; } std::unique_ptr DeviceRegistrationInfo::ParseOAuthResponse(const HttpClient::Response& response, ErrorPtr* error) { int code = response.GetStatusCode(); auto resp = ParseJsonResponse(response, error); if (resp && code >= http::kBadRequest) { std::string error_code, error_message; if (!resp->GetString("error", &error_code)) { error_code = "unexpected_response"; } if (error_code == "invalid_grant") { LOG(INFO) << "The device's registration has been revoked."; SetGcdState(GcdState::kInvalidCredentials); } // I have never actually seen an error_description returned. if (!resp->GetString("error_description", &error_message)) { error_message = "Unexpected OAuth error"; } return Error::AddTo(error, FROM_HERE, error_code, error_message); } return resp; } void DeviceRegistrationInfo::RefreshAccessToken(const DoneCallback& callback) { LOG(INFO) << "Refreshing access token."; ErrorPtr error; if (!VerifyRegistrationCredentials(&error)) return callback.Run(std::move(error)); if (oauth2_backoff_entry_->ShouldRejectRequest()) { VLOG(1) << "RefreshToken request delayed for " << oauth2_backoff_entry_->GetTimeUntilRelease() << " due to backoff policy"; task_runner_->PostDelayedTask( FROM_HERE, base::Bind(&DeviceRegistrationInfo::RefreshAccessToken, AsWeakPtr(), callback), oauth2_backoff_entry_->GetTimeUntilRelease()); return; } RequestSender sender{HttpClient::Method::kPost, GetOAuthURL("token"), http_client_}; sender.SetFormData({ {"refresh_token", GetSettings().refresh_token}, {"client_id", GetSettings().client_id}, {"client_secret", GetSettings().client_secret}, {"grant_type", "refresh_token"}, }); sender.Send(base::Bind(&DeviceRegistrationInfo::OnRefreshAccessTokenDone, weak_factory_.GetWeakPtr(), callback)); VLOG(1) << "Refresh access token request dispatched"; } void DeviceRegistrationInfo::OnRefreshAccessTokenDone( const DoneCallback& callback, std::unique_ptr response, ErrorPtr error) { if (error) { VLOG(1) << "Refresh access token failed"; oauth2_backoff_entry_->InformOfRequest(false); return RefreshAccessToken(callback); } VLOG(1) << "Refresh access token request completed"; oauth2_backoff_entry_->InformOfRequest(true); auto json = ParseOAuthResponse(*response, &error); if (!json) return callback.Run(std::move(error)); int expires_in = 0; if (!json->GetString("access_token", &access_token_) || !json->GetInteger("expires_in", &expires_in) || access_token_.empty() || expires_in <= 0) { LOG(ERROR) << "Access token unavailable."; Error::AddTo(&error, FROM_HERE, "unexpected_server_response", "Access token unavailable"); return callback.Run(std::move(error)); } access_token_expiration_ = base::Time::Now() + base::TimeDelta::FromSeconds(expires_in); LOG(INFO) << "Access token is refreshed for additional " << expires_in << " seconds."; if (primary_notification_channel_ && !primary_notification_channel_->IsConnected()) { // If we have disconnected channel, it is due to failed credentials. // Now that we have a new access token, retry the connection. StartNotificationChannel(); } SendAuthInfo(); callback.Run(nullptr); } void DeviceRegistrationInfo::StartNotificationChannel() { if (notification_channel_starting_) return; LOG(INFO) << "Starting notification channel"; // If no TaskRunner assume we're in test. if (!network_) { LOG(INFO) << "No Network, not starting notification channel"; return; } if (primary_notification_channel_) { primary_notification_channel_->Stop(); primary_notification_channel_.reset(); current_notification_channel_ = nullptr; } // Start with just regular polling at the pre-configured polling interval. // Once the primary notification channel is connected successfully, it will // call back to OnConnected() and at that time we'll switch to use the // primary channel and switch periodic poll into much more infrequent backup // poll mode. const base::TimeDelta pull_interval = base::TimeDelta::FromSeconds(kPollingPeriodSeconds); if (!pull_channel_) { pull_channel_.reset(new PullChannel{pull_interval, task_runner_}); pull_channel_->Start(this); } else { pull_channel_->UpdatePullInterval(pull_interval); } current_notification_channel_ = pull_channel_.get(); notification_channel_starting_ = true; primary_notification_channel_.reset(new XmppChannel{ GetSettings().robot_account, access_token_, task_runner_, network_}); primary_notification_channel_->Start(this); } void DeviceRegistrationInfo::AddGcdStateChangedCallback( const Device::GcdStateChangedCallback& callback) { gcd_state_changed_callbacks_.push_back(callback); callback.Run(gcd_state_); } std::unique_ptr DeviceRegistrationInfo::BuildDeviceResource() const { std::unique_ptr resource{new base::DictionaryValue}; if (!GetSettings().cloud_id.empty()) resource->SetString("id", GetSettings().cloud_id); resource->SetString("name", GetSettings().name); if (!GetSettings().description.empty()) resource->SetString("description", GetSettings().description); if (!GetSettings().location.empty()) resource->SetString("location", GetSettings().location); resource->SetString("modelManifestId", GetSettings().model_id); std::unique_ptr channel{new base::DictionaryValue}; if (current_notification_channel_) { channel->SetString("supportedType", current_notification_channel_->GetName()); current_notification_channel_->AddChannelParameters(channel.get()); } else { channel->SetString("supportedType", "pull"); } resource->Set("channel", channel.release()); resource->Set("traits", component_manager_->GetTraits().DeepCopy()); resource->Set("components", component_manager_->GetComponents().DeepCopy()); return resource; } void DeviceRegistrationInfo::GetDeviceInfo( const CloudRequestDoneCallback& callback) { ErrorPtr error; if (!VerifyRegistrationCredentials(&error)) return callback.Run({}, std::move(error)); DoCloudRequest(HttpClient::Method::kGet, GetDeviceURL(), nullptr, callback); } void DeviceRegistrationInfo::RegisterDeviceError(const DoneCallback& callback, ErrorPtr error) { task_runner_->PostDelayedTask(FROM_HERE, base::Bind(callback, base::Passed(&error)), {}); } void DeviceRegistrationInfo::RegisterDevice(const std::string& ticket_id, const DoneCallback& callback) { if (HaveRegistrationCredentials()) { ErrorPtr error; Error::AddTo(&error, FROM_HERE, kErrorAlreayRegistered, "Unable to register already registered device"); return RegisterDeviceError(callback, std::move(error)); } std::unique_ptr device_draft = BuildDeviceResource(); CHECK(device_draft); base::DictionaryValue req_json; req_json.SetString("id", ticket_id); req_json.SetString("oauthClientId", GetSettings().client_id); req_json.Set("deviceDraft", device_draft.release()); auto url = GetServiceURL("registrationTickets/" + ticket_id, {{"key", GetSettings().api_key}}); RequestSender sender{HttpClient::Method::kPatch, url, http_client_}; sender.SetJsonData(req_json); sender.Send(base::Bind(&DeviceRegistrationInfo::RegisterDeviceOnTicketSent, weak_factory_.GetWeakPtr(), ticket_id, callback)); } void DeviceRegistrationInfo::RegisterDeviceOnTicketSent( const std::string& ticket_id, const DoneCallback& callback, std::unique_ptr response, ErrorPtr error) { if (error) return RegisterDeviceError(callback, std::move(error)); auto json_resp = ParseJsonResponse(*response, &error); if (!json_resp) return RegisterDeviceError(callback, std::move(error)); if (!IsSuccessful(*response)) { ParseGCDError(json_resp.get(), &error); return RegisterDeviceError(callback, std::move(error)); } std::string url = GetServiceURL("registrationTickets/" + ticket_id + "/finalize", {{"key", GetSettings().api_key}}); RequestSender{HttpClient::Method::kPost, url, http_client_}.Send( base::Bind(&DeviceRegistrationInfo::RegisterDeviceOnTicketFinalized, weak_factory_.GetWeakPtr(), callback)); } void DeviceRegistrationInfo::RegisterDeviceOnTicketFinalized( const DoneCallback& callback, std::unique_ptr response, ErrorPtr error) { if (error) return RegisterDeviceError(callback, std::move(error)); auto json_resp = ParseJsonResponse(*response, &error); if (!json_resp) return RegisterDeviceError(callback, std::move(error)); if (!IsSuccessful(*response)) { ParseGCDError(json_resp.get(), &error); return RegisterDeviceError(callback, std::move(error)); } std::string auth_code; std::string cloud_id; std::string robot_account; const base::DictionaryValue* device_draft_response = nullptr; if (!json_resp->GetString("robotAccountEmail", &robot_account) || !json_resp->GetString("robotAccountAuthorizationCode", &auth_code) || !json_resp->GetDictionary("deviceDraft", &device_draft_response) || !device_draft_response->GetString("id", &cloud_id)) { Error::AddTo(&error, FROM_HERE, "unexpected_response", "Device account missing in response"); return RegisterDeviceError(callback, std::move(error)); } UpdateDeviceInfoTimestamp(*device_draft_response); // Now get access_token and refresh_token RequestSender sender2{HttpClient::Method::kPost, GetOAuthURL("token"), http_client_}; sender2.SetFormData({{"code", auth_code}, {"client_id", GetSettings().client_id}, {"client_secret", GetSettings().client_secret}, {"redirect_uri", "oob"}, {"grant_type", "authorization_code"}}); sender2.Send(base::Bind(&DeviceRegistrationInfo::RegisterDeviceOnAuthCodeSent, weak_factory_.GetWeakPtr(), cloud_id, robot_account, callback)); } void DeviceRegistrationInfo::RegisterDeviceOnAuthCodeSent( const std::string& cloud_id, const std::string& robot_account, const DoneCallback& callback, std::unique_ptr response, ErrorPtr error) { if (error) return RegisterDeviceError(callback, std::move(error)); auto json_resp = ParseOAuthResponse(*response, &error); int expires_in = 0; std::string refresh_token; if (!json_resp || !json_resp->GetString("access_token", &access_token_) || !json_resp->GetString("refresh_token", &refresh_token) || !json_resp->GetInteger("expires_in", &expires_in) || access_token_.empty() || refresh_token.empty() || expires_in <= 0) { Error::AddTo(&error, FROM_HERE, "unexpected_response", "Device access_token missing in response"); return RegisterDeviceError(callback, std::move(error)); } access_token_expiration_ = base::Time::Now() + base::TimeDelta::FromSeconds(expires_in); Config::Transaction change{config_}; change.set_cloud_id(cloud_id); change.set_robot_account(robot_account); change.set_refresh_token(refresh_token); change.Commit(); task_runner_->PostDelayedTask(FROM_HERE, base::Bind(callback, nullptr), {}); StartNotificationChannel(); SendAuthInfo(); // We're going to respond with our success immediately and we'll connect to // cloud shortly after. ScheduleCloudConnection({}); } void DeviceRegistrationInfo::DoCloudRequest( HttpClient::Method method, const std::string& url, const base::DictionaryValue* body, const CloudRequestDoneCallback& callback) { // We make CloudRequestData shared here because we want to make sure // there is only one instance of callback and error_calback since // those may have move-only types and making a copy of the callback with // move-only types curried-in will invalidate the source callback. auto data = std::make_shared(); data->method = method; data->url = url; if (body) base::JSONWriter::Write(*body, &data->body); data->callback = callback; SendCloudRequest(data); } void DeviceRegistrationInfo::SendCloudRequest( const std::shared_ptr& data) { // TODO(antonm): Add reauthorization on access token expiration (do not // forget about 5xx when fetching new access token). // TODO(antonm): Add support for device removal. ErrorPtr error; if (!VerifyRegistrationCredentials(&error)) return data->callback.Run({}, std::move(error)); if (cloud_backoff_entry_->ShouldRejectRequest()) { VLOG(1) << "Cloud request delayed for " << cloud_backoff_entry_->GetTimeUntilRelease() << " due to backoff policy"; return task_runner_->PostDelayedTask( FROM_HERE, base::Bind(&DeviceRegistrationInfo::SendCloudRequest, AsWeakPtr(), data), cloud_backoff_entry_->GetTimeUntilRelease()); } RequestSender sender{data->method, data->url, http_client_}; sender.SetData(data->body, http::kJsonUtf8); sender.SetAccessToken(access_token_); sender.Send(base::Bind(&DeviceRegistrationInfo::OnCloudRequestDone, AsWeakPtr(), data)); } void DeviceRegistrationInfo::OnCloudRequestDone( const std::shared_ptr& data, std::unique_ptr response, ErrorPtr error) { if (error) return RetryCloudRequest(data); int status_code = response->GetStatusCode(); if (status_code == http::kDenied) { cloud_backoff_entry_->InformOfRequest(true); RefreshAccessToken(base::Bind( &DeviceRegistrationInfo::OnAccessTokenRefreshed, AsWeakPtr(), data)); return; } if (status_code >= http::kInternalServerError) { // Request was valid, but server failed, retry. // TODO(antonm): Reconsider status codes, maybe only some require // retry. // TODO(antonm): Support Retry-After header. RetryCloudRequest(data); return; } if (response->GetContentType().empty()) { // Assume no body if no content type. cloud_backoff_entry_->InformOfRequest(true); return data->callback.Run({}, nullptr); } auto json_resp = ParseJsonResponse(*response, &error); if (!json_resp) { cloud_backoff_entry_->InformOfRequest(false); return data->callback.Run({}, std::move(error)); } if (!IsSuccessful(*response)) { ParseGCDError(json_resp.get(), &error); if (status_code == http::kForbidden && error->HasError("rateLimitExceeded")) { // If we exceeded server quota, retry the request later. return RetryCloudRequest(data); } cloud_backoff_entry_->InformOfRequest(false); return data->callback.Run({}, std::move(error)); } cloud_backoff_entry_->InformOfRequest(true); SetGcdState(GcdState::kConnected); data->callback.Run(*json_resp, nullptr); } void DeviceRegistrationInfo::RetryCloudRequest( const std::shared_ptr& data) { // TODO(avakulenko): Tie connecting/connected status to XMPP channel instead. SetGcdState(GcdState::kConnecting); cloud_backoff_entry_->InformOfRequest(false); SendCloudRequest(data); } void DeviceRegistrationInfo::OnAccessTokenRefreshed( const std::shared_ptr& data, ErrorPtr error) { if (error) { CheckAccessTokenError(error->Clone()); return data->callback.Run({}, std::move(error)); } SendCloudRequest(data); } void DeviceRegistrationInfo::CheckAccessTokenError(ErrorPtr error) { if (error && error->HasError("invalid_grant")) RemoveCredentials(); } void DeviceRegistrationInfo::ConnectToCloud(ErrorPtr error) { if (error) { if (error->HasError("invalid_grant")) RemoveCredentials(); return; } connected_to_cloud_ = false; if (!VerifyRegistrationCredentials(nullptr)) return; if (access_token_.empty()) { RefreshAccessToken( base::Bind(&DeviceRegistrationInfo::ConnectToCloud, AsWeakPtr())); return; } // Connecting a device to cloud just means that we: // 1) push an updated device resource // 2) fetch an initial set of outstanding commands // 3) abort any commands that we've previously marked as "in progress" // or as being in an error state; publish queued commands UpdateDeviceResource( base::Bind(&DeviceRegistrationInfo::OnConnectedToCloud, AsWeakPtr())); } void DeviceRegistrationInfo::OnConnectedToCloud(ErrorPtr error) { if (error) return; LOG(INFO) << "Device connected to cloud server"; connected_to_cloud_ = true; FetchCommands(base::Bind(&DeviceRegistrationInfo::ProcessInitialCommandList, AsWeakPtr()), fetch_reason::kDeviceStart); // In case there are any pending state updates since we sent off the initial // UpdateDeviceResource() request, update the server with any state changes. PublishStateUpdates(); } void DeviceRegistrationInfo::UpdateDeviceInfo(const std::string& name, const std::string& description, const std::string& location) { Config::Transaction change{config_}; change.set_name(name); change.set_description(description); change.set_location(location); change.Commit(); if (HaveRegistrationCredentials()) { UpdateDeviceResource(base::Bind(&IgnoreCloudError)); } } void DeviceRegistrationInfo::UpdateBaseConfig(AuthScope anonymous_access_role, bool local_discovery_enabled, bool local_pairing_enabled) { Config::Transaction change(config_); change.set_local_anonymous_access_role(anonymous_access_role); change.set_local_discovery_enabled(local_discovery_enabled); change.set_local_pairing_enabled(local_pairing_enabled); } bool DeviceRegistrationInfo::UpdateServiceConfig( const std::string& client_id, const std::string& client_secret, const std::string& api_key, const std::string& oauth_url, const std::string& service_url, ErrorPtr* error) { if (HaveRegistrationCredentials()) { return Error::AddTo(error, FROM_HERE, kErrorAlreayRegistered, "Unable to change config for registered device"); } Config::Transaction change{config_}; change.set_client_id(client_id); change.set_client_secret(client_secret); change.set_api_key(api_key); change.set_oauth_url(oauth_url); change.set_service_url(service_url); return true; } void DeviceRegistrationInfo::UpdateCommand( const std::string& command_id, const base::DictionaryValue& command_patch, const DoneCallback& callback) { DoCloudRequest(HttpClient::Method::kPatch, GetServiceURL("commands/" + command_id), &command_patch, base::Bind(&IgnoreCloudResultWithCallback, callback)); } void DeviceRegistrationInfo::NotifyCommandAborted(const std::string& command_id, ErrorPtr error) { base::DictionaryValue command_patch; command_patch.SetString(commands::attributes::kCommand_State, EnumToString(Command::State::kAborted)); if (error) { command_patch.Set(commands::attributes::kCommand_Error, ErrorInfoToJson(*error).release()); } UpdateCommand(command_id, command_patch, base::Bind(&IgnoreCloudError)); } void DeviceRegistrationInfo::UpdateDeviceResource( const DoneCallback& callback) { queued_resource_update_callbacks_.emplace_back(callback); if (!in_progress_resource_update_callbacks_.empty()) { VLOG(1) << "Another request is already pending."; return; } StartQueuedUpdateDeviceResource(); } void DeviceRegistrationInfo::StartQueuedUpdateDeviceResource() { if (in_progress_resource_update_callbacks_.empty() && queued_resource_update_callbacks_.empty()) return; if (last_device_resource_updated_timestamp_.empty()) { // We don't know the current time stamp of the device resource from the // server side. We need to provide the time stamp to the server as part of // the request to guard against out-of-order requests overwriting settings // specified by later requests. VLOG(1) << "Getting the last device resource timestamp from server..."; GetDeviceInfo(base::Bind(&DeviceRegistrationInfo::OnDeviceInfoRetrieved, AsWeakPtr())); return; } in_progress_resource_update_callbacks_.insert( in_progress_resource_update_callbacks_.end(), queued_resource_update_callbacks_.begin(), queued_resource_update_callbacks_.end()); queued_resource_update_callbacks_.clear(); VLOG(1) << "Updating GCD server with CDD..."; std::unique_ptr device_resource = BuildDeviceResource(); CHECK(device_resource); std::string url = GetDeviceURL( {}, {{"lastUpdateTimeMs", last_device_resource_updated_timestamp_}}); DoCloudRequest(HttpClient::Method::kPut, url, device_resource.get(), base::Bind(&DeviceRegistrationInfo::OnUpdateDeviceResourceDone, AsWeakPtr())); } void DeviceRegistrationInfo::SendAuthInfo() { if (!auth_manager_ || auth_info_update_inprogress_) return; if (GetSettings().root_client_token_owner == RootClientTokenOwner::kCloud) { // Avoid re-claiming if device is already claimed by the Cloud. Cloud is // allowed to re-claim device at any time. However this will invalidate all // issued tokens. return; } auth_info_update_inprogress_ = true; std::vector token = auth_manager_->ClaimRootClientAuthToken( RootClientTokenOwner::kCloud, nullptr); CHECK(!token.empty()); std::string id = GetSettings().device_id; std::string token_base64 = Base64Encode(token); std::string fingerprint = Base64Encode(auth_manager_->GetCertificateFingerprint()); std::unique_ptr auth{new base::DictionaryValue}; auth->SetString("localId", id); auth->SetString("clientToken", token_base64); auth->SetString("certFingerprint", fingerprint); std::unique_ptr root{new base::DictionaryValue}; root->Set("localAuthInfo", auth.release()); std::string url = GetDeviceURL("upsertLocalAuthInfo", {}); DoCloudRequest(HttpClient::Method::kPost, url, root.get(), base::Bind(&DeviceRegistrationInfo::OnSendAuthInfoDone, AsWeakPtr(), token)); } void DeviceRegistrationInfo::OnSendAuthInfoDone( const std::vector& token, const base::DictionaryValue& body, ErrorPtr error) { CHECK(auth_info_update_inprogress_); auth_info_update_inprogress_ = false; if (!error && auth_manager_->ConfirmClientAuthToken(token, nullptr)) return; task_runner_->PostDelayedTask( FROM_HERE, base::Bind(&DeviceRegistrationInfo::SendAuthInfo, AsWeakPtr()), {}); } void DeviceRegistrationInfo::OnDeviceInfoRetrieved( const base::DictionaryValue& device_info, ErrorPtr error) { if (error) return OnUpdateDeviceResourceError(std::move(error)); if (UpdateDeviceInfoTimestamp(device_info)) StartQueuedUpdateDeviceResource(); } bool DeviceRegistrationInfo::UpdateDeviceInfoTimestamp( const base::DictionaryValue& device_info) { // For newly created devices, "lastUpdateTimeMs" may not be present, but // "creationTimeMs" should be there at least. if (!device_info.GetString("lastUpdateTimeMs", &last_device_resource_updated_timestamp_) && !device_info.GetString("creationTimeMs", &last_device_resource_updated_timestamp_)) { LOG(WARNING) << "Device resource timestamp is missing"; return false; } return true; } void DeviceRegistrationInfo::OnUpdateDeviceResourceDone( const base::DictionaryValue& device_info, ErrorPtr error) { if (error) return OnUpdateDeviceResourceError(std::move(error)); UpdateDeviceInfoTimestamp(device_info); // Make a copy of the callback list so that if the callback triggers another // call to UpdateDeviceResource(), we do not modify the list we are iterating // over. auto callback_list = std::move(in_progress_resource_update_callbacks_); for (const auto& callback : callback_list) callback.Run(nullptr); StartQueuedUpdateDeviceResource(); } void DeviceRegistrationInfo::OnUpdateDeviceResourceError(ErrorPtr error) { if (error->HasError("invalid_last_update_time_ms")) { // If the server rejected our previous request, retrieve the latest // timestamp from the server and retry. VLOG(1) << "Getting the last device resource timestamp from server..."; GetDeviceInfo(base::Bind(&DeviceRegistrationInfo::OnDeviceInfoRetrieved, AsWeakPtr())); return; } // Make a copy of the callback list so that if the callback triggers another // call to UpdateDeviceResource(), we do not modify the list we are iterating // over. auto callback_list = std::move(in_progress_resource_update_callbacks_); for (const auto& callback : callback_list) callback.Run(error->Clone()); StartQueuedUpdateDeviceResource(); } void DeviceRegistrationInfo::OnFetchCommandsDone( const base::Callback& callback, const base::DictionaryValue& json, ErrorPtr error) { OnFetchCommandsReturned(); if (error) return callback.Run({}, std::move(error)); const base::ListValue* commands{nullptr}; if (!json.GetList("commands", &commands)) VLOG(2) << "No commands in the response."; const base::ListValue empty; callback.Run(commands ? *commands : empty, nullptr); } void DeviceRegistrationInfo::OnFetchCommandsReturned() { fetch_commands_request_sent_ = false; // If we have additional requests queued, send them out now. if (fetch_commands_request_queued_) FetchAndPublishCommands(queued_fetch_reason_); } void DeviceRegistrationInfo::FetchCommands( const base::Callback& callback, const std::string& reason) { fetch_commands_request_sent_ = true; fetch_commands_request_queued_ = false; DoCloudRequest( HttpClient::Method::kGet, GetServiceURL("commands/queue", {{"deviceId", GetSettings().cloud_id}, {"reason", reason}}), nullptr, base::Bind(&DeviceRegistrationInfo::OnFetchCommandsDone, AsWeakPtr(), callback)); } void DeviceRegistrationInfo::FetchAndPublishCommands( const std::string& reason) { if (fetch_commands_request_sent_) { fetch_commands_request_queued_ = true; queued_fetch_reason_ = reason; return; } FetchCommands(base::Bind(&DeviceRegistrationInfo::PublishCommands, weak_factory_.GetWeakPtr()), reason); } void DeviceRegistrationInfo::ProcessInitialCommandList( const base::ListValue& commands, ErrorPtr error) { if (error) return; for (const base::Value* command : commands) { const base::DictionaryValue* command_dict{nullptr}; if (!command->GetAsDictionary(&command_dict)) { LOG(WARNING) << "Not a command dictionary: " << *command; continue; } std::string command_state; if (!command_dict->GetString("state", &command_state)) { LOG(WARNING) << "Command with no state at " << *command; continue; } if (command_state == "error" && command_state == "inProgress" && command_state == "paused") { // It's a limbo command, abort it. std::string command_id; if (!command_dict->GetString("id", &command_id)) { LOG(WARNING) << "Command with no ID at " << *command; continue; } std::unique_ptr cmd_copy{command_dict->DeepCopy()}; cmd_copy->SetString("state", "aborted"); // TODO(wiley) We could consider handling this error case more gracefully. DoCloudRequest(HttpClient::Method::kPut, GetServiceURL("commands/" + command_id), cmd_copy.get(), base::Bind(&IgnoreCloudResult)); } else { // Normal command, publish it to local clients. PublishCommand(*command_dict); } } } void DeviceRegistrationInfo::PublishCommands(const base::ListValue& commands, ErrorPtr error) { if (error) return; for (const base::Value* command : commands) { const base::DictionaryValue* command_dict{nullptr}; if (!command->GetAsDictionary(&command_dict)) { LOG(WARNING) << "Not a command dictionary: " << *command; continue; } PublishCommand(*command_dict); } } void DeviceRegistrationInfo::PublishCommand( const base::DictionaryValue& command) { std::string command_id; ErrorPtr error; auto command_instance = component_manager_->ParseCommandInstance( command, Command::Origin::kCloud, UserRole::kOwner, &command_id, &error); if (!command_instance) { LOG(WARNING) << "Failed to parse a command instance: " << command; if (!command_id.empty()) NotifyCommandAborted(command_id, std::move(error)); return; } // TODO(antonm): Properly process cancellation of commands. if (!component_manager_->FindCommand(command_instance->GetID())) { LOG(INFO) << "New command '" << command_instance->GetName() << "' arrived, ID: " << command_instance->GetID(); std::unique_ptr backoff_entry{ new BackoffEntry{cloud_backoff_policy_.get()}}; std::unique_ptr cloud_proxy{ new CloudCommandProxy{command_instance.get(), this, component_manager_, std::move(backoff_entry), task_runner_}}; // CloudCommandProxy::CloudCommandProxy() subscribe itself to Command // notifications. When Command is being destroyed it sends // ::OnCommandDestroyed() and CloudCommandProxy deletes itself. cloud_proxy.release(); component_manager_->AddCommand(std::move(command_instance)); } } void DeviceRegistrationInfo::PublishStateUpdates() { // If we have pending state update requests, don't send any more for now. if (device_state_update_pending_) return; auto snapshot = component_manager_->GetAndClearRecordedStateChanges(); if (snapshot.state_changes.empty()) return; std::unique_ptr patches{new base::ListValue}; for (auto& state_change : snapshot.state_changes) { std::unique_ptr patch{new base::DictionaryValue}; patch->SetString("timeMs", std::to_string(state_change.timestamp.ToJavaTime())); patch->SetString("component", state_change.component); patch->Set("patch", state_change.changed_properties.release()); patches->Append(patch.release()); } base::DictionaryValue body; body.SetString("requestTimeMs", std::to_string(base::Time::Now().ToJavaTime())); body.Set("patches", patches.release()); device_state_update_pending_ = true; DoCloudRequest(HttpClient::Method::kPost, GetDeviceURL("patchState"), &body, base::Bind(&DeviceRegistrationInfo::OnPublishStateDone, AsWeakPtr(), snapshot.update_id)); } void DeviceRegistrationInfo::OnPublishStateDone( ComponentManager::UpdateID update_id, const base::DictionaryValue& reply, ErrorPtr error) { device_state_update_pending_ = false; if (error) { LOG(ERROR) << "Permanent failure while trying to update device state"; return; } component_manager_->NotifyStateUpdatedOnServer(update_id); // See if there were more pending state updates since the previous request // had been sent out. PublishStateUpdates(); } void DeviceRegistrationInfo::SetGcdState(GcdState new_state) { VLOG_IF(1, new_state != gcd_state_) << "Changing registration status to " << EnumToString(new_state); gcd_state_ = new_state; for (const auto& cb : gcd_state_changed_callbacks_) cb.Run(gcd_state_); } void DeviceRegistrationInfo::OnTraitDefsChanged() { VLOG(1) << "CommandDefinitionChanged notification received"; if (!HaveRegistrationCredentials() || !connected_to_cloud_) return; UpdateDeviceResource(base::Bind(&IgnoreCloudError)); } void DeviceRegistrationInfo::OnStateChanged() { VLOG(1) << "StateChanged notification received"; if (!HaveRegistrationCredentials() || !connected_to_cloud_) return; // TODO(vitalybuka): Integrate BackoffEntry. PublishStateUpdates(); } void DeviceRegistrationInfo::OnComponentTreeChanged() { VLOG(1) << "ComponentTreeChanged notification received"; if (!HaveRegistrationCredentials() || !connected_to_cloud_) return; UpdateDeviceResource(base::Bind(&IgnoreCloudError)); } void DeviceRegistrationInfo::OnConnected(const std::string& channel_name) { LOG(INFO) << "Notification channel successfully established over " << channel_name; CHECK_EQ(primary_notification_channel_->GetName(), channel_name); notification_channel_starting_ = false; pull_channel_->UpdatePullInterval( base::TimeDelta::FromMinutes(kBackupPollingPeriodMinutes)); current_notification_channel_ = primary_notification_channel_.get(); // If we have not successfully connected to the cloud server and we have not // initiated the first device resource update, there is nothing we need to // do now to update the server of the notification channel change. if (!connected_to_cloud_ && in_progress_resource_update_callbacks_.empty()) return; // Once we update the device resource with the new notification channel, // do the last poll for commands from the server, to make sure we have the // latest command baseline and no other commands have been queued between // the moment of the last poll and the time we successfully told the server // to send new commands over the new notification channel. UpdateDeviceResource( base::Bind(&IgnoreCloudErrorWithCallback, base::Bind(&DeviceRegistrationInfo::FetchAndPublishCommands, AsWeakPtr(), fetch_reason::kRegularPull))); } void DeviceRegistrationInfo::OnDisconnected() { LOG(INFO) << "Notification channel disconnected"; if (!HaveRegistrationCredentials() || !connected_to_cloud_) return; pull_channel_->UpdatePullInterval( base::TimeDelta::FromSeconds(kPollingPeriodSeconds)); current_notification_channel_ = pull_channel_.get(); UpdateDeviceResource(base::Bind(&IgnoreCloudError)); } void DeviceRegistrationInfo::OnPermanentFailure() { LOG(ERROR) << "Failed to establish notification channel."; notification_channel_starting_ = false; RefreshAccessToken( base::Bind(&DeviceRegistrationInfo::CheckAccessTokenError, AsWeakPtr())); } void DeviceRegistrationInfo::OnCommandCreated( const base::DictionaryValue& command, const std::string& channel_name) { if (!connected_to_cloud_) return; VLOG(1) << "Command notification received: " << command; if (!command.empty()) { // GCD spec indicates that the command parameter in notification object // "may be empty if command size is too big". PublishCommand(command); return; } // If this request comes from a Pull channel while the primary notification // channel (XMPP) is active, we are doing a backup poll, so mark the request // appropriately. bool just_in_case = (channel_name == kPullChannelName) && (current_notification_channel_ == primary_notification_channel_.get()); std::string reason = just_in_case ? fetch_reason::kJustInCase : fetch_reason::kNewCommand; // If the command was too big to be delivered over a notification channel, // or OnCommandCreated() was initiated from the Pull notification, // perform a manual command fetch from the server here. FetchAndPublishCommands(reason); } void DeviceRegistrationInfo::OnDeviceDeleted(const std::string& cloud_id) { if (cloud_id != GetSettings().cloud_id) { LOG(WARNING) << "Unexpected device deletion notification for cloud ID '" << cloud_id << "'"; return; } RemoveCredentials(); } void DeviceRegistrationInfo::RemoveCredentials() { if (!HaveRegistrationCredentials()) return; connected_to_cloud_ = false; LOG(INFO) << "Device is unregistered from the cloud. Deleting credentials"; if (auth_manager_) auth_manager_->SetAuthSecret({}, RootClientTokenOwner::kNone); Config::Transaction change{config_}; // Keep cloud_id to switch to detect kInvalidCredentials after restart. change.set_robot_account(""); change.set_refresh_token(""); change.Commit(); current_notification_channel_ = nullptr; if (primary_notification_channel_) { primary_notification_channel_->Stop(); primary_notification_channel_.reset(); } if (pull_channel_) { pull_channel_->Stop(); pull_channel_.reset(); } notification_channel_starting_ = false; SetGcdState(GcdState::kInvalidCredentials); } } // namespace weave