NXPY-218: Introduce the BasicAuth class

CHANGELOG.rst

@@ -11,10 +11,12 @@ Release date: ``2021-0x-xx``
 - `NXPY-214 <https://jira.nuxeo.com/browse/NXPY-214>`__: Add a code coverage GitHub Action on PRs
 - `NXPY-215 <https://jira.nuxeo.com/browse/NXPY-215>`__: Add support for the JSON Web Token authentication
 - `NXPY-217 <https://jira.nuxeo.com/browse/NXPY-217>`__: Restore Python 2.7 support
+- `NXPY-218 <https://jira.nuxeo.com/browse/NXPY-218>`__: Introduce the ``BasicAuth`` class
 
 Technical changes
 -----------------
 
+- Added nuxeo/auth/basic.py
 - Added nuxeo/auth/jwt.py
 - Added nuxeo/auth/oauth2.py
 - Added nuxeo/exceptions.py::\ ``OAuth2Error``

nuxeo/auth/__init__.py

@@ -1,9 +1,10 @@
 # coding: utf-8
 from __future__ import unicode_literals
 
+from .basic import BasicAuth
 from .jwt import JWTAuth
 from .oauth2 import OAuth2
 from .portal_sso import PortalSSOAuth
 from .token import TokenAuth
 
-__all__ = ("JWTAuth", "OAuth2", "PortalSSOAuth", "TokenAuth")
+__all__ = ("BasicAuth", "JWTAuth", "OAuth2", "PortalSSOAuth", "TokenAuth")

nuxeo/auth/basic.py

@@ -0,0 +1,55 @@
+# coding: utf-8
+from __future__ import unicode_literals
+import base64
+
+from ..compat import get_bytes, get_text
+from .base import AuthBase
+
+try:
+    from typing import TYPE_CHECKING
+
+    if TYPE_CHECKING:
+        from typing import Text
+        from requests import Request
+except ImportError:
+    pass
+
+
+class BasicAuth(AuthBase):
+    """ Attaches a simple Basic Authentication to the given Request object. """
+
+    __slots__ = ("username", "password", "_token_header")
+
+    AUTHORIZATION = get_bytes("Authorization")
+
+    def __init__(self, username, password):
+        # type: (Text, Text) -> None
+        self.username = username
+        self.password = password
+        self.set_token(password)
+
+    def set_token(self, token):
+        # type: (Text) -> None
+        """Apply the given *token*."""
+        self.password = token
+        self._token_header = "Basic " + get_text(
+            base64.b64encode(get_bytes(self.username + ":" + self.password))
+        )
+
+    def __eq__(self, other):
+        # type: (object) -> bool
+        return all(
+            [
+                self.username == getattr(other, "username", None),
+                self.password == getattr(other, "password", None),
+            ]
+        )
+
+    def __ne__(self, other):
+        # type: (object) -> bool
+        return not self == other
+
+    def __call__(self, r):
+        # type: (Request) -> Request
+        r.headers[self.AUTHORIZATION] = self._token_header
+        return r

nuxeo/auth/token.py

@@ -1,15 +1,18 @@
 # coding: utf-8
 from __future__ import unicode_literals
 
-from ..compat import get_bytes
+from ..constants import DEFAULT_APP_NAME
+from ..compat import get_bytes, text
 from .base import AuthBase
 
 try:
     from typing import TYPE_CHECKING
 
     if TYPE_CHECKING:
-        from typing import Text
+        from typing import Any, Dict, Optional, Text
         from requests import Request
+
+        Token = Dict[str, Any]
 except ImportError:
     pass
 
@@ -25,6 +28,47 @@ def __init__(self, token):
         # type: (Text) -> None
         self.token = token
 
+    def request_token(
+        self,
+        client,
+        device_id,  # type: Text
+        permission,  # type: Text
+        app_name=DEFAULT_APP_NAME,  # type: Text
+        device=None,  # type: Optional[Text]
+        revoke=False,  # type: bool
+        auth=None,
+    ):
+        # type: (...) -> Token
+        """
+        Request a token.
+
+        :param device_id: device identifier
+        :param permission: read/write permissions
+        :param app_name: application name
+        :param device: optional device description
+        :param revoke: revoke the token
+        """
+
+        params = {
+            "deviceId": device_id,
+            "applicationName": app_name,
+            "permission": permission,
+            "revoke": text(revoke).lower(),
+        }
+        if device:
+            params["deviceDescription"] = device
+
+        path = "authentication/token"
+        token = client.request("GET", path, params=params, auth=auth).text
+        token = "" if (revoke or "\n" in token) else token
+        self.set_token(token)
+        return token
+
+    def set_token(self, token):
+        # type: (Token) -> None
+        """Apply the given *token*."""
+        self.token = token
+
     def __eq__(self, other):
         # type: (object) -> bool
         return self.token == getattr(other, "token", None)

nuxeo/client.py

@@ -24,7 +24,7 @@
     users,
     workflows,
 )
-from .auth import TokenAuth
+from .auth import BasicAuth, TokenAuth
 from .compat import text
 from .constants import (
     CHUNK_SIZE,
@@ -109,7 +109,7 @@ def __init__(
         **kwargs  # type: Any
     ):
         # type: (...) -> None
-        self.auth = auth
+        self.auth = BasicAuth(*auth) if isinstance(auth, tuple) else auth
         self.host = host
         self.api_path = api_path
         self.chunk_size = chunk_size
@@ -282,6 +282,9 @@ def request(
         # to set `default` to `None`.
         default = kwargs.pop("default", object)
 
+        # Allow to pass a custom authentication class
+        auth = kwargs.pop("auth", None) or self.auth
+
         _kwargs = {k: v for k, v in kwargs.items() if k != "params"}
         logger.debug(
             (
@@ -299,7 +302,7 @@ def request(
         exc = None
         try:
             resp = self._session.request(
-                method, url, headers=headers, auth=self.auth, data=data, **kwargs
+                method, url, headers=headers, auth=auth, data=data, **kwargs
             )
             resp.raise_for_status()
         except Exception as exc:
@@ -340,29 +343,28 @@ def request_auth_token(
         # type: (...) -> Text
         """
         Request a token for the user.
+        It should only be used if you want to get a Nuxeo token from a Basic Auth.
 
         :param device_id: device identifier
         :param permission: read/write permissions
         :param app_name: application name
         :param device: optional device description
         :param revoke: revoke the token
         """
-
-        parameters = {
-            "deviceId": device_id,
-            "applicationName": app_name,
-            "permission": permission,
-            "revoke": text(revoke).lower(),
-        }
-        if device:
-            parameters["deviceDescription"] = device
-
-        path = "authentication/token"
-        token = self.request("GET", path, params=parameters).text
+        auth = TokenAuth("")
+        token = auth.request_token(
+            self,
+            device_id,
+            permission,
+            app_name=app_name,
+            device=device,
+            revoke=revoke,
+            auth=self.auth,
+        )
 
         # Use the (potentially re-newed) token from now on
         if not revoke:
-            self.auth = TokenAuth(token)
+            self.auth = auth
         return token
 
     def is_reachable(self):

tests/conftest.py

@@ -9,6 +9,7 @@
 import pytest
 from requests.cookies import RequestsCookieJar
 
+from nuxeo.auth import BasicAuth
 from nuxeo.client import Nuxeo
 from nuxeo.exceptions import HTTPError
 
@@ -75,7 +76,7 @@ def server(host):
     cookies.set("device", "python-client")
     server = Nuxeo(
         host=host,
-        auth=("Administrator", "Administrator"),
+        auth=BasicAuth("Administrator", "Administrator"),
         cookies=cookies,
     )
     server.client.set(schemas=["dublincore"])

tests/test_auth.py

@@ -4,7 +4,7 @@
 import pytest
 from requests import Request
 
-from nuxeo.auth import JWTAuth, OAuth2, PortalSSOAuth, TokenAuth
+from nuxeo.auth import BasicAuth, JWTAuth, OAuth2, PortalSSOAuth, TokenAuth
 from nuxeo.auth.utils import make_portal_sso_token
 from nuxeo.compat import text
 from nuxeo.exceptions import NuxeoError
@@ -13,6 +13,22 @@
 skip_logging = True
 
 
+def test_basic():
+    auth = BasicAuth("Alice", "her password")
+    req = Request("GET", "https://httpbin.org/get", auth=auth)
+    prepared = req.prepare()
+    assert prepared.headers[auth.AUTHORIZATION] == "Basic QWxpY2U6aGVyIHBhc3N3b3Jk"
+
+
+def test_basic_equality():
+    auth1 = BasicAuth("Alice", "a password")
+    auth2 = BasicAuth("Alice", "a new password")
+    auth3 = BasicAuth("Bob", "a password")
+    assert auth1 != auth2
+    assert auth1 != auth3
+    assert auth2 != auth3
+
+
 def test_jwt():
     auth = JWTAuth("<TOKEN>")
     req = Request("GET", "https://httpbin.org/get", auth=auth)

tests/test_users.py

@@ -3,6 +3,7 @@
 
 import pytest
 
+from nuxeo.auth import BasicAuth
 from nuxeo.client import Nuxeo
 from nuxeo.exceptions import BadQuery
 from nuxeo.users import User
@@ -83,7 +84,7 @@ def test_update_user(server, host):
         user = server.users.get("georges")
         assert user.properties["company"] == company
 
-        auth = ("georges", "Test")
+        auth = BasicAuth("georges", "Test")
         server2 = Nuxeo(host=host, auth=auth)
         assert server2.users.current_user()
 
@@ -93,6 +94,6 @@ def test_update_user_autoset_change_password(server, host):
         georges.change_password("Test2")
         georges.save()
 
-        auth = ("georges", "Test2")
+        auth = BasicAuth("georges", "Test2")
         server2 = Nuxeo(host=host, auth=auth)
         assert server2.users.current_user()