add new coinbase key type

models/config/coinbase_parser.py

@@ -6,6 +6,7 @@
 
 from .default_parser import is_currency_valid, default_config_parse, merge_config_and_args
 from models.exchange.Granularity import Granularity
+from utils.PyCryptoBot import validate_ec_private_key as _validate_ec_private_key
 
 
 def is_market_valid(market) -> bool:
@@ -67,7 +68,7 @@ def parser(app, coinbase_config, args={}):
                 try:
                     with open(app.api_key_file, "r") as f:
                         key = f.readline().strip()
-                        secret = f.readline().strip()
+                        secret = f.readline().replace('\\n', os.linesep).strip()
                     coinbase_config["api_key"] = key
                     coinbase_config["api_secret"] = secret
                 except Exception:
@@ -78,15 +79,14 @@ def parser(app, coinbase_config, args={}):
 
         if "api_key" in coinbase_config and "api_secret" in coinbase_config and "api_url" in coinbase_config:
             # validates the api key is syntactically correct
-            p = re.compile(r"^[A-z0-9]{16,16}$")
+            p = re.compile(r"^organizations/[0-9a-fA-F-]{36,36}/apiKeys/[0-9a-fA-F-]{36,36}$")
             if not p.match(coinbase_config["api_key"]):
                 raise TypeError("Coinbase API key is invalid")
 
             app.api_key = coinbase_config["api_key"]
 
             # validates the api secret is syntactically correct
-            p = re.compile(r"^[A-z0-9]{32,32}$")
-            if not p.match(coinbase_config["api_secret"]):
+            if _validate_ec_private_key(coinbase_config["api_secret"]):
                 raise TypeError("Coinbase API secret is invalid")
 
             app.api_secret = coinbase_config["api_secret"]

models/exchange/coinbase/api.py

@@ -17,6 +17,8 @@
 from websocket import create_connection, WebSocketConnectionClosedException
 from models.exchange.Granularity import Granularity
 from views.PyCryptoBot import RichText
+from coinbase import jwt_generator
+from utils.PyCryptoBot import validate_ec_private_key as _validate_ec_private_key
 
 MARGIN_ADJUSTMENT = 0.0025
 DEFAULT_MAKER_FEE_RATE = 0.004
@@ -67,13 +69,12 @@ def __init__(self, api_key="", api_secret="", api_url="https://api.coinbase.com"
             api_url = api_url + "/"
 
         # validates the api key is syntactically correct
-        p = re.compile(r"^[a-zA-Z0-9]{16}$")
+        p = re.compile(r"^organizations/[0-9a-fA-F-]{36,36}/apiKeys/[0-9a-fA-F-]{36,36}$")
         if not p.match(api_key):
             self.handle_init_error("Coinbase API key is invalid", app=app)
 
         # validates the api secret is syntactically correct
-        p = re.compile(r"^[a-zA-Z0-9]{32}$")
-        if not p.match(api_secret):
+        if _validate_ec_private_key(api_secret):
             self.handle_init_error("Coinbase API secret is invalid", app=app)
 
         # app
@@ -93,22 +94,23 @@ def handle_init_error(self, err: str, app: object = None) -> None:
 
     def __call__(self, request) -> Request:
         """Signs the request"""
+
+        api_key = self._api_key
+        api_secret = self._api_secret
 
-        timestamp = str(int(time.time()))
-        body = (request.body or b"").decode()
-        url = request.path_url.split("?")[0]
-        message = f"{timestamp}{request.method}{url}{body}"
-        signature = hmac.new(self._api_secret.encode("utf-8"), message.encode("utf-8"), digestmod=hashlib.sha256).hexdigest()
+        request_method = request.method
+        request_path = request.path_url.split("?")[0]
 
+        jwt_uri = jwt_generator.format_jwt_uri(request_method, request_path)
+        jwt_token = jwt_generator.build_rest_jwt(jwt_uri, api_key, api_secret)
+
         request.headers.update(
             {
-                "CB-ACCESS-SIGN": signature,
-                "CB-ACCESS-TIMESTAMP": timestamp,
-                "CB-ACCESS-KEY": self._api_key,
-                "Content-Type": "application/json",
+                "Authorization": f"Bearer {jwt_token}",
+                "Content-Type": "application/json"
             }
         )
-
+        
         return request
 
     # wallet:accounts:read

requirements.txt

@@ -1,7 +1,7 @@
 urllib3>=1.26.5
 pandas==1.5.1
 pandas-ta
-numpy
+numpy<2.0.0
 requests==2.31.0
 matplotlib==3.3.3
 mock==4.0.3
@@ -21,3 +21,4 @@ dash-bootstrap-components
 codespell
 rich
 regex
+coinbase-advanced-py

utils/PyCryptoBot.py

@@ -2,7 +2,8 @@
 
 import math
 from typing import Union
-
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.backends import default_backend
 
 def truncate(f: Union[int, float], n: Union[int, float]) -> str:
     """
@@ -42,3 +43,18 @@ def compare(val1, val2, label="", precision=2):
             return f"{truncate(val1, precision)} = {truncate(val2, precision)}"
         else:
             return f"{label}: {truncate(val1, precision)} = {truncate(val2, precision)}"
+
+
+def validate_ec_private_key(key_pem):
+    try:
+        # Load the private key
+        private_key = serialization.load_pem_private_key(
+            key_pem.encode(),  # Convert to bytes if it's a string
+            password=None,
+            backend=default_backend()
+        )
+        # If no exception was raised, the key is valid
+        return False
+    except Exception as e:
+        # Handle the exception for invalid keys
+        raise ValueError (f"Invalid EC Private Key: {str(e)}")