pylxd/client: inspect secret before trying to use it as a token

pylxd/client.py

@@ -11,6 +11,7 @@
 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 #    License for the specific language governing permissions and limitations
 #    under the License.
+import base64
 import json
 import os
 import re
@@ -349,6 +350,32 @@ def received_message(self, message):
         self.messages.append(json_message)
 
 
+# Helper function used by Client.authenticate()
+def _is_a_token(secret):
+    """Inspect the provided secret to determine if it is a trust token.
+
+    Try to base64 decode and parse the JSON to see if it contains a "secret" key.
+
+    :param secret: The secret to inspect
+    :type secret: str
+    :returns: True if the secret is a trust token
+
+    >>> _is_a_token("password")
+    False
+    >>> _is_a_token(base64.b64encode("password".encode("utf-8")))
+    False
+    >>> token = '{"client_name":"foo","fingerprint":"abcd","addresses":["192.0.2.1:8443"],"secret":"I-am-a-secret","expires_at":"0001-01-01T00:00:00Z","type":""}'
+    >>> _is_a_token(base64.b64encode(json.dumps(token).encode("utf-8")))
+    True
+    """
+    try:
+        b64 = base64.b64decode(secret)
+        token = json.loads(b64.decode("utf-8"))
+        return "secret" in token
+    except (TypeError, ValueError, json.JSONDecodeError, base64.binascii.Error):
+        return False
+
+
 class Client:
     """Client class for LXD REST API.
 
@@ -391,13 +418,13 @@ class Client:
         Use the name of the url part as attribute or item of an api object to
         create another api object appended with the new url part name, ie:
 
-            >>> api = Client().api
             # /
-            >>> response = api.get()
+            >> api = Client().api
+            >> response = api.get()
             # Check status code and response
-            >>> print response.status_code, response.json()
+            >> print(response.status_code, response.json())
             # /instances/test/
-            >>> print api.instances['test'].get().json()
+            >> print(api.instances['test'].get().json())
     """
 
     def __init__(
@@ -545,7 +572,14 @@ def authenticate(self, secret, use_token_auth=True):
             return
         cert = open(self.api.session.cert[0]).read().encode("utf-8")
 
-        if self.has_api_extension("explicit_trust_token") and use_token_auth:
+        # Quirk to handle 5.21 that supports explicit trust tokens as well as
+        # password auth. We need to ascertain if the provided secret is indeed a
+        # token before trying to use it as such.
+        secret_is_a_token = False
+        if use_token_auth and self.has_api_extension("explicit_trust_token"):
+            secret_is_a_token = _is_a_token(secret)
+
+        if secret_is_a_token:
             self.certificates.create(password="", cert_data=cert, secret=secret)
         else:
             self.certificates.create(password=secret, cert_data=cert)

pylxd/models/storage_pool.py

@@ -420,7 +420,7 @@ def create(cls, storage_pool, *args, **kwargs):
         The function signature 'hides' that the first parameter to the function
         is the definition.  The function should be called as:
 
-        >>> a_storage_pool.volumes.create(definition_dict, wait=<bool>)
+        >> a_storage_pool.volumes.create(definition_dict, wait=<bool>)
 
         where `definition_dict` is mandatory, and `wait` defaults to True,
         which makes the default a synchronous function call.

setup.cfg

@@ -77,7 +77,7 @@ setenv =
 deps =
     .[testing]
 commands =
-    pytest {posargs:pylxd}
+    pytest --doctest-modules {posargs:pylxd}
 
 [testenv:integration]
 passenv = LXD_*