# Copyright (c) 2014-present PlatformIO # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # pylint: disable=unused-argument import os import time import requests.adapters from requests.packages.urllib3.util.retry import Retry # pylint:disable=import-error from platformio import __pioaccount_api__, app from platformio.commands.account import exception from platformio.exception import InternetIsOffline class AccountClient(object): SUMMARY_CACHE_TTL = 60 * 60 * 24 * 7 def __init__( self, api_base_url=__pioaccount_api__, retries=3, ): if api_base_url.endswith("/"): api_base_url = api_base_url[:-1] self.api_base_url = api_base_url self._session = requests.Session() self._session.headers.update({"User-Agent": app.get_user_agent()}) retry = Retry( total=retries, read=retries, connect=retries, backoff_factor=2, method_whitelist=list(Retry.DEFAULT_METHOD_WHITELIST) + ["POST"], ) adapter = requests.adapters.HTTPAdapter(max_retries=retry) self._session.mount(api_base_url, adapter) @staticmethod def get_refresh_token(): try: return app.get_state_item("account").get("auth").get("refresh_token") except: # pylint:disable=bare-except raise exception.AccountNotAuthorized() @staticmethod def delete_local_session(): app.delete_state_item("account") @staticmethod def delete_local_state(key): account = app.get_state_item("account") if not account or key not in account: return del account[key] app.set_state_item("account", account) def login(self, username, password): try: self.fetch_authentication_token() except: # pylint:disable=bare-except pass else: raise exception.AccountAlreadyAuthorized( app.get_state_item("account", {}).get("email", "") ) result = self.send_request( "post", self.api_base_url + "/v1/login", data={"username": username, "password": password}, ) app.set_state_item("account", result) return result def login_with_code(self, client_id, code, redirect_uri): try: self.fetch_authentication_token() except: # pylint:disable=bare-except pass else: raise exception.AccountAlreadyAuthorized( app.get_state_item("account", {}).get("email", "") ) result = self.send_request( "post", self.api_base_url + "/v1/login/code", data={"client_id": client_id, "code": code, "redirect_uri": redirect_uri}, ) app.set_state_item("account", result) return result def logout(self): refresh_token = self.get_refresh_token() self.delete_local_session() try: self.send_request( "post", self.api_base_url + "/v1/logout", data={"refresh_token": refresh_token}, ) except exception.AccountError: pass return True def change_password(self, old_password, new_password): token = self.fetch_authentication_token() self.send_request( "post", self.api_base_url + "/v1/password", headers={"Authorization": "Bearer %s" % token}, data={"old_password": old_password, "new_password": new_password}, ) return True def registration( self, username, email, password, firstname, lastname ): # pylint:disable=too-many-arguments try: self.fetch_authentication_token() except: # pylint:disable=bare-except pass else: raise exception.AccountAlreadyAuthorized( app.get_state_item("account", {}).get("email", "") ) return self.send_request( "post", self.api_base_url + "/v1/registration", data={ "username": username, "email": email, "password": password, "firstname": firstname, "lastname": lastname, }, ) def auth_token(self, password, regenerate): token = self.fetch_authentication_token() result = self.send_request( "post", self.api_base_url + "/v1/token", headers={"Authorization": "Bearer %s" % token}, data={"password": password, "regenerate": 1 if regenerate else 0}, ) return result.get("auth_token") def forgot_password(self, username): return self.send_request( "post", self.api_base_url + "/v1/forgot", data={"username": username}, ) def get_profile(self): token = self.fetch_authentication_token() return self.send_request( "get", self.api_base_url + "/v1/profile", headers={"Authorization": "Bearer %s" % token}, ) def update_profile(self, profile, current_password): token = self.fetch_authentication_token() profile["current_password"] = current_password self.delete_local_state("summary") response = self.send_request( "put", self.api_base_url + "/v1/profile", headers={"Authorization": "Bearer %s" % token}, data=profile, ) return response def get_account_info(self, offline): account = app.get_state_item("account") if not account: raise exception.AccountNotAuthorized() if ( account.get("summary") and account["summary"].get("expire_at", 0) > time.time() ): return account["summary"] if offline: return { "profile": { "email": account.get("email"), "username": account.get("username"), } } token = self.fetch_authentication_token() result = self.send_request( "get", self.api_base_url + "/v1/summary", headers={"Authorization": "Bearer %s" % token}, ) account["summary"] = dict( profile=result.get("profile"), packages=result.get("packages"), subscriptions=result.get("subscriptions"), user_id=result.get("user_id"), expire_at=int(time.time()) + self.SUMMARY_CACHE_TTL, ) app.set_state_item("account", account) return result def fetch_authentication_token(self): if "PLATFORMIO_AUTH_TOKEN" in os.environ: return os.environ["PLATFORMIO_AUTH_TOKEN"] auth = app.get_state_item("account", {}).get("auth", {}) if auth.get("access_token") and auth.get("access_token_expire"): if auth.get("access_token_expire") > time.time(): return auth.get("access_token") if auth.get("refresh_token"): try: result = self.send_request( "post", self.api_base_url + "/v1/login", headers={ "Authorization": "Bearer %s" % auth.get("refresh_token") }, ) app.set_state_item("account", result) return result.get("auth").get("access_token") except exception.AccountError: self.delete_local_session() raise exception.AccountNotAuthorized() def send_request(self, method, url, headers=None, data=None): try: response = getattr(self._session, method)( url, headers=headers or {}, data=data or {} ) except requests.exceptions.ConnectionError: raise InternetIsOffline() return self.raise_error_from_response(response) def raise_error_from_response(self, response, expected_codes=(200, 201, 202)): if response.status_code in expected_codes: try: return response.json() except ValueError: pass try: message = response.json()["message"] except (KeyError, ValueError): message = response.text if "Authorization session has been expired" in message: self.delete_local_session() raise exception.AccountError(message)