Fix #377: Allow basic auth-scheme (#384)

* Allow basic auth-scheme

* refactor: improve code readability and maintainability

* Revert to BearerTokenAuth class for bearer token

* 'make format'

* 'make format' with latest ruff

---------

Co-authored-by: Mathieu Leplatre <[email protected]>

Author: moses946

src/kinto_http/__init__.py

@@ -18,6 +18,7 @@
 
 __all__ = (
     "BrowserOAuth",
+    "TokenAuth",
     "BearerTokenAuth",
     "Endpoints",
     "Session",
@@ -31,11 +32,16 @@
 )
 
 
-class BearerTokenAuth(requests.auth.AuthBase):
+class TokenAuth(requests.auth.AuthBase):
     def __init__(self, token, type=None):
         self.token = token
         self.type = type or "Bearer"
 
     def __call__(self, r):
+        # Sets auth-scheme to either Bearer or Basic
         r.headers["Authorization"] = "{} {}".format(self.type, self.token)
         return r
+
+
+class BearerTokenAuth(TokenAuth):
+    pass

src/kinto_http/cli_utils.py

@@ -75,7 +75,7 @@ def add_parser_options(
     parser.add_argument(
         "-a",
         "--auth",
-        help="BasicAuth credentials: `token:my-secret` or " "Authorization header: `Bearer token`",
+        help="BasicAuth credentials: `token:my-secret` or Authorization header: `Bearer token`",
         type=str,
         default=default_auth,
         action=AuthAction,
@@ -97,8 +97,7 @@ def add_parser_options(
 
     parser.add_argument(
         "--retry-after",
-        help="Delay in seconds between retries when requests fail. "
-        "(default: provided by server)",
+        help="Delay in seconds between retries when requests fail. (default: provided by server)",
         type=int,
         default=default_retry_after,
     )

src/kinto_http/replication.py

@@ -42,23 +42,22 @@ def get_arguments():  # pragma: nocover
 
     parser.add_argument(
         "--origin-auth",
-        help="The origin authentication credentials. "
-        "Will use the same as the remote if omitted",
+        help="The origin authentication credentials. Will use the same as the remote if omitted",
         action=cli_utils.AuthAction,
         default=None,
     )
 
     parser.add_argument(
         "--origin-bucket",
         dest="origin_bucket",
-        help="The name of the origin bucket. " "Will use the same as the remote if omitted",
+        help="The name of the origin bucket. Will use the same as the remote if omitted",
         default=None,
     )
 
     parser.add_argument(
         "--origin-collection",
         dest="origin_collection",
-        help="The name of the origin collection. " "Will use the same as the remote if omitted",
+        help="The name of the origin collection. Will use the same as the remote if omitted",
         default=None,
     )
     cli_utils.set_parser_server_options(parser)

src/kinto_http/session.py

@@ -45,10 +45,13 @@ def create_session(server_url=None, auth=None, session=None, **kwargs):
             # eg, "Bearer ghruhgrwyhg"
             _type, token = auth.split(" ", 1)
             auth = kinto_http.BearerTokenAuth(token, type=_type)
+        elif "basic" in auth.lower():
+            _type, token = auth.split(" ", 1)
+            auth = kinto_http.TokenAuth(token, type=_type)
         elif auth:  # not empty
             raise ValueError(
                 "Unsupported `auth` parameter value. Must be a tuple() or string "
-                "in the form of `user:pass` or `Bearer xyz`"
+                "in the form of `user:pass` or `Bearer xyz` or `Basic xyz`"
             )
 
     if session is None:

tests/support.py

@@ -51,9 +51,9 @@ def get_user_id(server_url: str, credentials: Tuple[str, str]) -> str:
 
 def assert_option_strings(parser, *option_strings_list):
     for option_strings in option_strings_list:
-        assert any(
-            [action.option_strings == option_strings for action in parser._actions]
-        ), f"{option_strings} not found"
+        assert any([action.option_strings == option_strings for action in parser._actions]), (
+            f"{option_strings} not found"
+        )
 
 
 def build_response(data, headers=None):

tests/test_async_client.py

@@ -43,13 +43,13 @@ def test_client_is_represented_properly_with_bucket_and_collection(async_client_
     client = async_client_setup.clone(
         server_url=SERVER_URL, bucket="homebrewing", collection="recipes"
     )
-    expected_repr = f"<KintoAsyncClient {SERVER_URL}/" "buckets/homebrewing/collections/recipes>"
+    expected_repr = f"<KintoAsyncClient {SERVER_URL}/buckets/homebrewing/collections/recipes>"
     assert str(client) == expected_repr
 
 
 def test_client_is_represented_properly_with_bucket(async_client_setup: Client):
     client = async_client_setup.clone(server_url=SERVER_URL, bucket="homebrewing")
-    expected_repr = f"<KintoAsyncClient {SERVER_URL}/" "buckets/homebrewing>"
+    expected_repr = f"<KintoAsyncClient {SERVER_URL}/buckets/homebrewing>"
     assert str(client) == expected_repr
 
 
@@ -843,7 +843,7 @@ async def test_records_timestamp_is_cached_per_collection(record_async_setup: Cl
 async def test_pagination_is_followed(record_async_setup: Client):
     client = record_async_setup
     # Mock the calls to request.
-    link = "http://example.org/buckets/buck/collections/coll/records/" "?token=1234"
+    link = "http://example.org/buckets/buck/collections/coll/records/?token=1234"
 
     client.session.request.side_effect = [
         # First one returns a list of items with a pagination token.
@@ -871,7 +871,7 @@ async def test_pagination_is_followed(record_async_setup: Client):
 async def test_pagination_is_followed_generator(record_async_setup: Client):
     client = record_async_setup
     # Mock the calls to request.
-    link = "http://example.org/buckets/buck/collections/coll/records/" "?token=1234"
+    link = "http://example.org/buckets/buck/collections/coll/records/?token=1234"
 
     response = [
         # First one returns a list of items with a pagination token.
@@ -897,7 +897,7 @@ async def test_pagination_is_followed_generator(record_async_setup: Client):
 async def test_pagination_is_followed_for_number_of_pages(record_async_setup: Client):
     client = record_async_setup
     # Mock the calls to request.
-    link = "http://example.org/buckets/buck/collections/coll/records/" "?token=1234"
+    link = "http://example.org/buckets/buck/collections/coll/records/?token=1234"
 
     client.session.request.side_effect = [
         # First one returns a list of items with a pagination token.
@@ -923,7 +923,7 @@ async def test_pagination_is_followed_for_number_of_pages(record_async_setup: Cl
 async def test_pagination_is_not_followed_if_limit_is_specified(record_async_setup: Client):
     client = record_async_setup
     # Mock the calls to request.
-    link = "http://example.org/buckets/buck/collections/coll/records/" "?token=1234"
+    link = "http://example.org/buckets/buck/collections/coll/records/?token=1234"
 
     client.session.request.side_effect = [
         build_response(
@@ -938,7 +938,7 @@ async def test_pagination_is_not_followed_if_limit_is_specified(record_async_set
 
 async def test_pagination_supports_if_none_match(record_async_setup: Client):
     client = record_async_setup
-    link = "http://example.org/buckets/buck/collections/coll/records/" "?token=1234"
+    link = "http://example.org/buckets/buck/collections/coll/records/?token=1234"
 
     client.session.request.side_effect = [
         # First one returns a list of items with a pagination token.
@@ -964,7 +964,7 @@ async def test_pagination_supports_if_none_match(record_async_setup: Client):
 
 async def test_pagination_generator_if_none_match(record_async_setup: Client):
     client = record_async_setup
-    link = "http://example.org/buckets/buck/collections/coll/records/" "?token=1234"
+    link = "http://example.org/buckets/buck/collections/coll/records/?token=1234"
 
     response = [
         # First one returns a list of items with a pagination token.

tests/test_client.py

@@ -209,13 +209,13 @@ def test_batch_options_are_transmitted(client_setup: Client, mocker: MockerFixtu
 
 def test_client_is_represented_properly_with_bucket_and_collection(client_setup: Client):
     client = client_setup.clone(server_url=SERVER_URL, bucket="homebrewing", collection="recipes")
-    expected_repr = f"<KintoClient {SERVER_URL}/" "buckets/homebrewing/collections/recipes>"
+    expected_repr = f"<KintoClient {SERVER_URL}/buckets/homebrewing/collections/recipes>"
     assert str(client) == expected_repr
 
 
 def test_client_is_represented_properly_with_bucket(client_setup: Client):
     client = client_setup.clone(server_url=SERVER_URL, bucket="homebrewing")
-    expected_repr = f"<KintoClient {SERVER_URL}/" "buckets/homebrewing>"
+    expected_repr = f"<KintoClient {SERVER_URL}/buckets/homebrewing>"
     assert str(client) == expected_repr
 
 
@@ -1003,7 +1003,7 @@ def test_records_timestamp_is_cached_per_collection(record_setup: Client):
 def test_pagination_is_followed(record_setup: Client):
     client = record_setup
     # Mock the calls to request.
-    link = "http://example.org/buckets/buck/collections/coll/records/" "?token=1234"
+    link = "http://example.org/buckets/buck/collections/coll/records/?token=1234"
 
     client.session.request.side_effect = [
         # First one returns a list of items with a pagination token.
@@ -1031,7 +1031,7 @@ def test_pagination_is_followed(record_setup: Client):
 def test_pagination_is_followed_generator(record_setup: Client):
     client = record_setup
     # Mock the calls to request.
-    link = "http://example.org/buckets/buck/collections/coll/records/" "?token=1234"
+    link = "http://example.org/buckets/buck/collections/coll/records/?token=1234"
 
     response = [
         # First one returns a list of items with a pagination token.
@@ -1057,7 +1057,7 @@ def test_pagination_is_followed_generator(record_setup: Client):
 def test_pagination_is_followed_for_number_of_pages(record_setup: Client):
     client = record_setup
     # Mock the calls to request.
-    link = "http://example.org/buckets/buck/collections/coll/records/" "?token=1234"
+    link = "http://example.org/buckets/buck/collections/coll/records/?token=1234"
 
     client.session.request.side_effect = [
         # First one returns a list of items with a pagination token.
@@ -1083,7 +1083,7 @@ def test_pagination_is_followed_for_number_of_pages(record_setup: Client):
 def test_pagination_is_not_followed_if_limit_is_specified(record_setup: Client):
     client = record_setup
     # Mock the calls to request.
-    link = "http://example.org/buckets/buck/collections/coll/records/" "?token=1234"
+    link = "http://example.org/buckets/buck/collections/coll/records/?token=1234"
 
     client.session.request.side_effect = [
         build_response(
@@ -1098,7 +1098,7 @@ def test_pagination_is_not_followed_if_limit_is_specified(record_setup: Client):
 
 def test_pagination_supports_if_none_match(record_setup: Client):
     client = record_setup
-    link = "http://example.org/buckets/buck/collections/coll/records/" "?token=1234"
+    link = "http://example.org/buckets/buck/collections/coll/records/?token=1234"
 
     client.session.request.side_effect = [
         # First one returns a list of items with a pagination token.
@@ -1124,7 +1124,7 @@ def test_pagination_supports_if_none_match(record_setup: Client):
 
 def test_pagination_generator_if_none_match(record_setup: Client):
     client = record_setup
-    link = "http://example.org/buckets/buck/collections/coll/records/" "?token=1234"
+    link = "http://example.org/buckets/buck/collections/coll/records/?token=1234"
 
     response = [
         # First one returns a list of items with a pagination token.

tests/test_session.py

@@ -246,6 +246,13 @@ def test_auth_can_be_passed_as_colon_separate(session_setup: Tuple[MagicMock, Se
 
 
 def test_auth_can_be_passed_as_basic_header(session_setup: Tuple[MagicMock, Session]):
+    session = create_session(auth="Basic asdfghjkl;")
+    assert isinstance(session.auth, kinto_http.TokenAuth)
+    assert session.auth.type == "Basic"
+    assert session.auth.token == "asdfghjkl;"
+
+
+def test_auth_can_be_passed_as_bearer(session_setup: Tuple[MagicMock, Session]):
     session = create_session(auth="Bearer+OIDC abcdef")
     assert isinstance(session.auth, kinto_http.BearerTokenAuth)
     assert session.auth.type == "Bearer+OIDC"