Handle invalid creds keyword formats. Closes #909.

src/falconpy/__init__.py

@@ -59,7 +59,8 @@
     InvalidBaseURL,
     PayloadValidationError,
     NoAuthenticationMechanism,
-    InvalidIndex
+    InvalidIndex,
+    InvalidCredentialFormat
     )
 from ._result import (
     Result,
@@ -169,7 +170,7 @@
     "InterfaceConfiguration", "RequestBehavior", "RequestConnection", "RequestMeta",
     "RequestPayloads", "RequestValidator", "PayloadValidationError", "MIN_TOKEN_RENEW_WINDOW",
     "MAX_TOKEN_RENEW_WINDOW", "GLOBAL_API_MAX_RETURN", "MOCK_OPERATIONS",
-    "NoAuthenticationMechanism", "InvalidIndex", "version"
+    "NoAuthenticationMechanism", "InvalidIndex", "version", "InvalidCredentialFormat"
     ]
 """
 This is free and unencumbered software released into the public domain.

src/falconpy/_auth_object/_falcon_interface.py

@@ -40,6 +40,11 @@
 """
 import time
 import warnings
+from json import loads
+try:
+    from simplejson import JSONDecodeError
+except ImportError:
+    from json.decoder import JSONDecodeError
 from logging import Logger, getLogger
 from typing import Dict, Optional, Union
 from ._base_falcon_auth import BaseFalconAuth
@@ -55,7 +60,7 @@
     login_payloads,
     logout_payloads
     )
-from .._error import InvalidCredentials, NoAuthenticationMechanism
+from .._error import InvalidCredentials, NoAuthenticationMechanism, InvalidCredentialFormat
 
 
 # pylint: disable=R0902
@@ -89,7 +94,7 @@ class FalconInterface(BaseFalconAuth):
     #
     # The default constructor for all authentication objects. Ingests provided credentials
     # and sets the necessary class attributes based upon the authentication detail received.
-    # pylint: disable=R0913,R0914
+    # pylint: disable=R0912,R0913,R0914
     def __init__(self,
                  access_token: Optional[Union[str, bool]] = False,
                  base_url: Optional[str] = "https://api.crowdstrike.com",
@@ -134,7 +139,16 @@ def __init__(self,
         elif not creds:
             creds = {}
         # Credential Authentication (also powers Direct Authentication).
-        self.creds: Dict[str, str] = creds
+        if isinstance(creds, str):
+            try:
+                # Try and clean up any attempts to provide the dictionary as a string
+                self.creds: Dict[str, str] = loads(creds.replace("'", "\""))
+            except (TypeError, JSONDecodeError) as bad_cred_format:
+                raise InvalidCredentialFormat from bad_cred_format
+        elif isinstance(creds, dict):
+            self.creds: Dict[str, str] = creds
+        else:
+            raise InvalidCredentialFormat
         # Set the token renewal window, ignored when using Legacy Authentication.
         self.renew_window: int = max(min(renew_window, MAX_TOKEN_RENEW_WINDOW),
                                      MIN_TOKEN_RENEW_WINDOW

src/falconpy/_error/__init__.py

@@ -49,7 +49,8 @@
     InvalidBaseURL,
     PayloadValidationError,
     FeatureNotSupportedByPythonVersion,
-    InvalidIndex
+    InvalidIndex,
+    InvalidCredentialFormat
     )
 from ._warnings import (
     SDKWarning,
@@ -64,5 +65,5 @@
            "InvalidCredentials", "APIError", "NoContentWarning", "CannotRevokeToken",
            "FunctionalityNotImplemented", "InvalidBaseURL", "PayloadValidationError",
            "NoAuthenticationMechanism", "FeatureNotSupportedByPythonVersion",
-           "InvalidIndex"
+           "InvalidIndex", "InvalidCredentialFormat"
            ]

src/falconpy/_error/_exceptions.py

@@ -174,3 +174,9 @@ class InvalidIndex(SDKError):
     """Item ID was not found in the list or string."""
 
     _message = "Item not found. Check your index and try again."
+
+
+class InvalidCredentialFormat(SDKError):
+    """Credentials dictionary was provided as a datatype that will not convert to a dictionary."""
+
+    _message = "Invalid credential format. This keyword must be provided as a dictionary."

tests/test_authentications.py

@@ -18,7 +18,7 @@
 # Import our sibling src folder into the path
 sys.path.append(os.path.abspath('src'))
 # Classes to test - manually imported from sibling folder
-from falconpy import ZeroTrustAssessment, CloudConnectAWS, OAuth2, APIHarness, version
+from falconpy import ZeroTrustAssessment, CloudConnectAWS, OAuth2, APIHarness, version, InvalidCredentialFormat, Hosts
 from falconpy._util import confirm_base_region
 from falconpy._version import _TITLE, _VERSION
 
@@ -267,5 +267,29 @@ def test_version_compare_exact_match(self):
         assert bool(version(version()))  # Should be a while before we hit that...
 
     def test_legacy_token_lookup(self):
-        test_object = ZeroTrustAssessment(auth_object=auth.authorization)
+        test_object = Hosts(auth_object=auth.authorization)
         assert bool(test_object.token)
+
+    @pytest.mark.skipif(auth.authorization.base_url == "https://api.laggar.gcw.crowdstrike.com",
+                        reason="Test unsupported in GovCloud"
+                        )
+    def test_string_credentials_dictionary(self):
+        key = auth.config["falcon_client_id"]
+        the_other_bit = auth.config["falcon_client_secret"]
+        test_string = "{" + f"'client_id': '{key}', 'client_secret': '{the_other_bit}'" + "}"
+        test_object = Hosts(creds=test_string)
+        test_object.login()
+        assert bool(test_object.authenticated())
+
+    def test_bad_credentials_dictionary(self):
+        _success = False
+        test = "Bob"
+        try:
+            test_object = Hosts(creds=test)
+        except InvalidCredentialFormat:
+            test = 2
+            try:
+                test_object = Hosts(creds=test)
+            except InvalidCredentialFormat:
+                _success = True
+        assert _success