fixed compute

Author: eduardomourey

gbmapi/auth.py

@@ -1,20 +1,22 @@
 import hashlib
 import time
 from base64 import urlsafe_b64encode
+
 import pyotp
-import requests
 
-from .exceptions import ResponseError
-from .utils import generate_secure_base64_string, STANDARD_HEADERS, load_jwt
+from .base import GBMApiBase
+from .utils import generate_secure_base64_string, load_jwt
 
 CLIENT_ID = "7c464570619a417080b300076e163289"
 
 
-class GBMAuth:
-    AUTH_URL = "https://auth.gbm.com"
+class GBMAuth(GBMApiBase):
+    API_URL = "https://auth.gbm.com"
     ORIGIN = 'https://app.gbm.com'
 
-    def __init__(self, user, password, secret, device, latitude, longitude, device_mac_address, client_id=CLIENT_ID):
+    def __init__(self, user, password, secret, device, latitude, longitude, device_mac_address,
+                 client_id=CLIENT_ID):
+
         self.client_id = client_id
         self.user = user
         self.password = password
@@ -25,6 +27,8 @@ def __init__(self, user, password, secret, device, latitude, longitude, device_m
         self.device_mac_address = device_mac_address
         self._credentials = None
 
+        super().__init__(self)
+
     @property
     def credentials(self):
         return self._credentials
@@ -33,64 +37,31 @@ def credentials(self):
     def credentials(self, value):
         self._credentials = value
 
-    def check_credentials(self):
+    def load_credentials(self):
         cred = self.credentials
         if cred is None:
-            self.login()
-            cred = self.credentials
+            cred = self.login()
+            self.credentials = cred
         else:
             _, b, _ = load_jwt(cred['accessToken'])
             if b['exp'] + 60 <= time.time():
-                self.refresh()
-                cred = self.credentials
+                cred = self.refresh()
+                self.credentials = cred
 
         return cred
 
     def access_token(self):
-        cred = self.check_credentials()
+        cred = self.load_credentials()
         return cred['tokenType'] + ' ' + cred['accessToken']
 
     def identity_token(self):
-        cred = self.check_credentials()
+        cred = self.load_credentials()
         return cred['identityToken']
 
-    def _request(self, method, path, headers=None, json=None, check_success=True):
-        resp = requests.request(
-            method,
-            self.AUTH_URL + path,
-            headers={
-                **STANDARD_HEADERS,
-                'origin': self.ORIGIN,
-                **(headers if headers else {})
-            },
-            json=json
-        )
-        if resp.status_code != 200:
-            if resp.status_code == 401:
-                self.credentials = None
-            raise ResponseError(resp)
-
-        resp = resp.json()
-        if check_success:
-            assert resp['code'] == 0, resp
-            assert resp['id'] == "Success", resp
-            assert resp['message'] == "Exitoso", resp
-        return resp
-
-    def _auth_request(self, method, path, json=None):
-        return self._request(
-            method,
-            path,
-            headers={
-                'authorization': self.access_token()
-            },
-            json=json
-        )
-
     def client(self):
         resp = self._request(
-            method="GET",
             path=f"/api/v1/clients/{self.client_id}",
+            authenticate=False
         )
         return resp
 
@@ -100,7 +71,7 @@ def login(self):
             'device-latitude': self.latitude,
             'device-longitude': self.longitude,
             'device-mac-address': self.device_mac_address,
-            'origin': GBMAuth.AUTH_URL
+            'origin': GBMAuth.API_URL
         }
 
         # Create a code verifier and challenge
@@ -110,8 +81,8 @@ def login(self):
         code_challenge = urlsafe_b64encode(m.digest()).decode().rstrip("=")
 
         resp = self._request(
-            method="POST",
             path="/api/v1/session/user",
+            authenticate=False,
             headers=device_headers,
             json={
                 "clientId": self.client_id,
@@ -130,8 +101,8 @@ def login(self):
         otp = totp.now()
 
         resp = self._request(
-            method="POST",
             path="/api/v1/session/user/challenge",
+            authenticate=False,
             headers=device_headers,
             json={
                 "challengeType": resp['challengeType'],
@@ -143,68 +114,65 @@ def login(self):
                 "responseType": "code",
                 "codeChallenge": code_challenge,
                 "codeChallengeMethod": "SHA256"
-            }
+            },
+            check_success=True
         )
 
         resp = self._request(
-            method="POST",
             path="/api/v1/session/token",
+            authenticate=False,
             json={
                 "clientId": self.client_id,
                 "codeVerifier": code_verifier,
                 "code": resp['authorizationCode']
-            }
+            },
+            check_success=True
         )
-        self.credentials = resp
+        return resp
 
     def refresh(self):
-        cred = self.credentials
-        refresh_token = cred['refreshToken']
+        refresh_token = self.credentials['refreshToken']
 
         resp = self._request(
-            method="POST",
             path="/api/v1/session/user/refresh",
+            authenticate=False,
             json={
                 "clientId": self.client_id,
                 "refreshToken": refresh_token
             }
         )
         resp['refreshToken'] = refresh_token
-        self.credentials = resp
+        return resp
 
     # After Authenticated
     def logout(self):
         # Clear Session Token
-        self._auth_request(
-            "DELETE",
+        self._request(
             path=f"/api/v1/session/user?client_id={self.client_id}",
+            method="DELETE"
         )
         self.credentials = None
 
     def security_settings(self):
         """Returns the security settings of the user"""
-        return self._auth_request(
-            "GET",
+        return self._request(
             path="/api/v1/security-settings"
         )
 
     def login_history(self):
         """Returns the login history of the user"""
-        return self._auth_request(
-            "GET",
+        return self._request(
             path="/api/v1/security-settings/login-history"
         )
 
     def challenge(self):
         """Returns information about the challenge."""
-        return self._auth_request(
-            method="GET",
+        return self._request(
             path="/api/v1/challenge"
         )
 
     def token(self):
         """Returns information about the token."""
-        return self._auth_request(
-            method="GET",
+        return self._request(
             path="/api/v1/token"
         )

gbmapi/base.py

@@ -1,6 +1,5 @@
 import requests
 
-from . import GBMAuth
 from .exceptions import ResponseError
 from .utils import STANDARD_HEADERS
 
@@ -9,30 +8,28 @@ class GBMApiBase:
     API_URL = None  # "https://api.gbm.com"
     ORIGIN = None  # "https://homebroker.gbm.com"
 
-    def __init__(self, auth: GBMAuth):
+    def __init__(self, auth):
         self.auth = auth
 
-    def _request(self, path, headers=None, json=None):
+    def _request(self, path, method=None, headers=None, json=None, check_success=False, authenticate=True):
         resp = requests.request(
-            "GET" if json is None else "POST",
+            method if method else ("GET" if json is None else "POST"),
             self.API_URL + path,
             headers={
                 **STANDARD_HEADERS,
-                'authorization': self.auth.access_token(),
+                **({'authorization': self.auth.access_token()} if authenticate else {}),
                 'origin': self.ORIGIN,
                 **(headers if headers else {})
             },
             json=json
         )
 
         if resp.status_code != 200:
-            if resp.status_code == 401:
-                self.auth.credentials = None
-
             raise ResponseError(resp)
 
         resp = resp.json()
-        # assert resp['code'] == 0, resp
-        # assert resp['id'] == "Success", resp
-        # assert resp['message'] == "Exitoso", resp
+        if check_success:
+            assert resp['code'] == 0, resp
+            assert resp['id'] == "Success", resp
+            assert resp['message'] == "Exitoso", resp
         return resp