python-gitlab · nejch · Jan 19, 2022 · lmilbaum · Dec 19, 2022 · JohnVillalovos
diff --git a/docs/api-usage.rst b/docs/api-usage.rst
@@ -84,14 +84,32 @@ Note on password authentication
 
 GitLab has long removed password-based basic authentication. You can currently still use the
 `resource owner password credentials <https://docs.gitlab.com/ee/api/oauth2.html#resource-owner-password-credentials-flow>`_
-flow to obtain an OAuth token.
+flow and python-gitlab will obtain an OAuth token for you when instantiated.
 
 However, we do not recommend this as it will not work with 2FA enabled, and GitLab is removing
-ROPC-based flows without client IDs in a future release. We recommend you obtain tokens for
-automated workflows as linked above or obtain a session cookie from your browser.
+ROPC-based flows without client credentials in a future release. We recommend you obtain tokens for
+automated workflows.
 
-For a python example of password authentication using the ROPC-based OAuth2
-flow, see `this Ansible snippet <https://github.com/ansible-collections/community.general/blob/1c06e237c8100ac30d3941d5a3869a4428ba2974/plugins/module_utils/gitlab.py#L86-L92>`_.
+.. code-block:: python
+
+   import gitlab
+   from gitlab.oauth import PasswordCredentials
+
+   oauth_credentials = PasswordCredentials("username", "password")
+   gl = gitlab.Gitlab(oauth_credentials=oauth_credentials)
+
+   # Define a specific OAuth scope
+   oauth_credentials = PasswordCredentials("username", "password", scope="read_api")
+   gl = gitlab.Gitlab(oauth_credentials=oauth_credentials)
+
+   # Use with client credentials
+   oauth_credentials = PasswordCredentials(
+      "username",
+      "password",
+      client_id="your-client-id",
+      client_secret="your-client-secret",
+   )
+   gl = gitlab.Gitlab(oauth_credentials=oauth_credentials)
 
 Managers
 ========

diff --git a/docs/cli-usage.rst b/docs/cli-usage.rst
@@ -168,8 +168,7 @@ We recommend that you use `Credential helpers`_ to securely store your tokens.
        <https://docs.gitlab.com/ce/user/profile/personal_access_tokens.html>`__
        to learn how to obtain a token.
    * - ``oauth_token``
-     - An Oauth token for authentication. The Gitlab server must be configured
-       to support this authentication method.
+     - An Oauth token for authentication.
    * - ``job_token``
      - Your job token. See `the official documentation
        <https://docs.gitlab.com/ce/api/jobs.html#get-job-artifacts>`__

diff --git a/gitlab/client.py b/gitlab/client.py
@@ -12,7 +12,7 @@
 import gitlab.config
 import gitlab.const
 import gitlab.exceptions
-from gitlab import _backends, utils
+from gitlab import _backends, oauth, utils
 
 REDIRECT_MSG = (
     "python-gitlab detected a {status_code} ({reason!r}) redirection. You must update "
@@ -41,8 +41,6 @@ class Gitlab:
             the value is a string, it is the path to a CA file used for
             certificate validation.
         timeout: Timeout to use for requests to the GitLab server.
-        http_username: Username for HTTP authentication
-        http_password: Password for HTTP authentication
         api_version: Gitlab API version to use (support for 4 only)
         pagination: Can be set to 'keyset' to use keyset pagination
         order_by: Set order_by globally
@@ -51,6 +49,7 @@ class Gitlab:
             or 52x responses. Defaults to False.
         keep_base_url: keep user-provided base URL for pagination if it
             differs from response headers
+        oauth_credentials: Password credentials for authenticating via OAuth ROPC flow
 
     Keyword Args:
         requests.Session session: HTTP Requests Session
@@ -64,8 +63,6 @@ def __init__(
         oauth_token: Optional[str] = None,
         job_token: Optional[str] = None,
         ssl_verify: Union[bool, str] = True,
-        http_username: Optional[str] = None,
-        http_password: Optional[str] = None,
         timeout: Optional[float] = None,
         api_version: str = "4",
         per_page: Optional[int] = None,
@@ -74,6 +71,8 @@ def __init__(
         user_agent: str = gitlab.const.USER_AGENT,
         retry_transient_errors: bool = False,
         keep_base_url: bool = False,
+        *,
+        oauth_credentials: Optional[oauth.PasswordCredentials] = None,
         **kwargs: Any,
     ) -> None:
    
6D40
     self._api_version = str(api_version)
@@ -92,11 +91,9 @@ def __init__(
         self.ssl_verify = ssl_verify
 
         self.private_token = private_token
-        self.http_username = http_username
-        self.http_password = http_password
         self.oauth_token = oauth_token
         self.job_token = job_token
-        self._set_auth_info()
+        self.oauth_credentials = oauth_credentials
 
         #: Create a session object for requests
         _backend: Type[_backends.DefaultBackend] = kwargs.pop(
@@ -105,6 +102,7 @@ def __init__(
         self._backend = _backend(**kwargs)
         self.session = self._backend.client
 
+        self._set_auth_info()
         self.per_page = per_page
         self.pagination = pagination
         self.order_by = order_by
@@ -271,8 +269,6 @@ def from_config(
             job_token=config.job_token,
             ssl_verify=config.ssl_verify,
             timeout=config.timeout,
-            http_username=config.http_username,
-            http_password=config.http_password,
             api_version=config.api_version,
             per_page=config.per_page,
             pagination=config.pagination,
@@ -471,41 +467,51 @@ def set_license(self, license: str, **kwargs: Any) -> Dict[str, Any]:
         return result
 
     def _set_auth_info(self) -> None:
-        tokens = [
-            token
-            for token in [self.private_token, self.oauth_token, self.job_token]
-            if token
+        auth_types = [
+            auth
+            for auth in [
+                self.private_token,
+                self.oauth_token,
+                self.oauth_credentials,
+                self.job_token,
+            ]
+            if auth
         ]
-        if len(tokens) > 1:
+        if len(auth_types) > 1:
             raise ValueError(
-                "Only one of private_token, oauth_token or job_token should "
-                "be defined"
-            )
-        if (self.http_username and not self.http_password) or (
-            not self.http_username and self.http_password
-        ):
-            raise ValueError("Both http_username and http_password should be defined")
-        if tokens and self.http_username:
-            raise ValueError(
-                "Only one of token authentications or http "
-                "authentication should be defined"
+                "Only one of private_token, oauth_token, oauth_credentials"
+                "or job_token should be defined"
             )
 
         self._auth: Optional[requests.auth.AuthBase] = None
         if self.private_token:
             self._auth = _backends.PrivateTokenAuth(self.private_token)
+            return
 
         if self.oauth_token:
             self._auth = _backends.OAuthTokenAuth(self.oauth_token)
+            return
+
+        if self.oauth_credentials:
+            post_data = {
+                "grant_type": self.oauth_credentials.grant_type,
+                "scope": self.oauth_credentials.scope,
+                "username": self.oauth_credentials.username,
+                "password": self.oauth_credentials.password,
+            }
+            response = self.http_post(
+                f"{self._base_url}/oauth/token", post_data=post_data
+            )
+            if isinstance(response, dict):
+                self.oauth_token = response["access_token"]
+            else:
+                self.oauth_token = response.json()["access_token"]
+            self._auth = self.oauth_credentials.basic_auth
+            return
 
         if self.job_token:
             self._auth = _backends.JobTokenAuth(self.job_token)
 
-        if self.http_username and self.http_password:
-            self._auth = requests.auth.HTTPBasicAuth(
-                self.http_username, self.http_password
-            )
-
     @staticmethod
     def enable_debug() -> None:
         import logging

diff --git a/gitlab/oauth.py b/gitlab/oauth.py
@@ -0,0 +1,33 @@
+import dataclasses
+from typing import Optional
+
+
+@dataclasses.dataclass
+class PasswordCredentials:
+    """
+    Resource owner password credentials modelled according to
+    https://docs.gitlab.com/ee/api/oauth2.html#resource-owner-password-credentials-flow
+    https://datatracker.ietf.org/doc/html/rfc6749#section-4-3.
+
+    If the GitLab server has disabled the ROPC flow without client credentials,
+    client_id and client_secret must be provided.
+    """
+
+    username: str
+    password: str
+    grant_type: str = "password"
+    scope: str = "api"
+    client_id: Optional[str] = None
+    client_secret: Optional[str] = None
+
+    def __post_init__(self) -> None:
+        basic_auth = (self.client_id, self.client_secret)
+
+        if not any(basic_auth):
+            self.basic_auth = None
+            return
+
+        if not all(basic_auth):
+            raise TypeError("Both client_id and client_secret must be defined")
+
+        self.basic_auth = basic_auth
diff --git a/tests/functional/api/test_gitlab.py b/tests/functional/api/test_gitlab.py
@@ -2,6 +2,7 @@
 import requests
 
 import gitlab
+from gitlab.oauth import PasswordCredentials
 
 
 @pytest.fixture(
@@ -22,6 +23,13 @@ def test_auth_from_config(gl, gitlab_config, temp_dir):
     assert isinstance(test_gitlab.user, gitlab.v4.objects.CurrentUser)
 
 
+def test_auth_with_ropc_flow(gl, temp_dir):
+    oauth_credentials = PasswordCredentials("root", "5iveL!fe")
+    test_gitlab = gitlab.Gitlab(gl.url, oauth_credentials=oauth_credentials)
+    test_gitlab.auth()
+    assert isinstance(test_gitlab.user, gitlab.v4.objects.CurrentUser)
+
+
 def test_no_custom_session(gl, temp_dir):
     """Test no custom session"""
     custom_session = requests.Session()

diff --git a/tests/unit/test_gitlab_auth.py b/tests/unit/test_gitlab_auth.py
@@ -8,6 +8,33 @@
 from gitlab import Gitlab
 from gitlab._backends import JobTokenAuth, OAuthTokenAuth, PrivateTokenAuth
 from gitlab.config import GitlabConfigParser
+from gitlab.oauth import PasswordCredentials
+
+
+# /oauth/token endpoint might be missing correct content-type header
+@pytest.fixture(params=["application/json", None])
+def resp_oauth_token(gl: Gitlab, request: pytest.FixtureRequest):
+    ropc_payload = {
+        "username": "foo",
+        "password": "bar",
+        "grant_type": "password",
+        "scope": "api",
+    }
+    ropc_response = {
+        "access_token": "test-token",
+        "token_type": "bearer",
+        "expires_in": 7200,
+    }
+    with responses.RequestsMock() as rsps:
+        rsps.add(
+            method=responses.POST,
+            url=f"{gl._base_url}/oauth/token",
+            status=201,
+            match=[responses.matchers.json_params_matcher(ropc_payload)],
+            json=ropc_response,
+            content_type=request.param,
+        )
+        yield rsps
 
 
 @pytest.fixture
@@ -91,24 +118,17 @@ def test_job_token_auth():
     assert "Authorization" not in p.headers
 
 
-def test_http_auth():
-    gl = Gitlab(
-        "http://localhost",
-        http_username="foo",
-        http_password="bar",
-        api_version="4",
-    )
+def test_oauth_resource_password_auth(resp_oauth_token):
+    oauth_credentials = PasswordCredentials("foo", "bar")
+    gl = Gitlab("http://localhost", oauth_credentials=oauth_credentials)
     p = PreparedRequest()
     p.prepare(url=gl.url, auth=gl._auth)
     assert gl.private_token is None
     assert gl.oauth_token is None
     assert gl.job_token is None
-    assert isinstance(gl._auth, requests.auth.HTTPBasicAuth)
     assert gl._auth.username == "foo"
     assert gl._auth.password == "bar"
     assert p.headers["Authorization"] == "Basic Zm9vOmJhcg=="
-    assert "PRIVATE-TOKEN" not in p.headers
-    assert "JOB-TOKEN" not in p.headers
 
 
 @responses.activate

diff --git a/tests/unit/test_oauth.py b/tests/unit/test_oauth.py
@@ -0,0 +1,27 @@
+import pytest
+
+from gitlab.oauth import PasswordCredentials
+
+
+def test_password_credentials_without_password_raises():
+    with pytest.raises(TypeError, match="missing 1 required positional argument"):
+        PasswordCredentials("username")
+
+
+def test_password_credentials_with_client_id_without_client_secret_raises():
+    with pytest.raises(TypeError, match="client_id and client_secret must be defined"):
+        PasswordCredentials(
+            "username",
+            "password",
+            client_id="abcdef123456",
+        )
+
+
+def test_password_credentials_with_client_credentials_sets_basic_auth():
+    credentials = PasswordCredentials(
+        "username",
+        "password",
+        client_id="abcdef123456",
+        client_secret="123456abcdef",
+    )
+    assert credentials.basic_auth == ("abcdef123456", "123456abcdef")