diff --git a/README.rst b/README.rst
index 6e16b17..9124c93 100644
--- a/README.rst
+++ b/README.rst
@@ -1,5 +1,5 @@
.. image:: https://github.com/ovh/python-ovh/raw/master/docs/img/logo.png
- :alt: Python & OVH APIs
+ :alt: Python & OVHcloud APIs
:target: https://pypi.python.org/pypi/ovh
Lightweight wrapper around OVHcloud's APIs. Handles all the hard work including
@@ -73,9 +73,9 @@ To interact with the APIs, the SDK needs to identify itself using an
``application_key`` and an ``application_secret``. To get them, you need
to register your application. Depending the API you plan to use, visit:
-- `OVH Europe `_
-- `OVH US `_
-- `OVH North-America `_
+- `OVHcloud Europe `_
+- `OVHcloud US `_
+- `OVHcloud North-America `_
- `So you Start Europe `_
- `So you Start North America `_
- `Kimsufi Europe `_
@@ -104,12 +104,15 @@ it looks like:
; uncomment following line when writing a script application
; with a single consumer key.
;consumer_key=my_consumer_key
+ ; uncomment to unable oauth2 authentication
+ ;client_id=my_client_id
+ ;client_secret=my_client_secret
Depending on the API you want to use, you may set the ``endpoint`` to:
-* ``ovh-eu`` for OVH Europe API
-* ``ovh-us`` for OVH US API
-* ``ovh-ca`` for OVH North-America API
+* ``ovh-eu`` for OVHcloud Europe API
+* ``ovh-us`` for OVHcloud US API
+* ``ovh-ca`` for OVHcloud North-America API
* ``soyoustart-eu`` for So you Start Europe API
* ``soyoustart-ca`` for So you Start North America API
* ``kimsufi-eu`` for Kimsufi Europe API
@@ -120,8 +123,21 @@ See Configuration_ for more information on available configuration mechanisms.
.. note:: When using a versioning system, make sure to add ``ovh.conf`` to ignored
files. It contains confidential/security-sensitive information!
-3. Authorize your application to access a customer account
-**********************************************************
+3. Authorize your application to access a customer account using OAuth2
+***********************************************************************
+
+``python-ovh`` supports two forms of authentication:
+* OAuth2, using scopped service accounts, and compatible with OVHcloud IAM
+* application key & application secret & consumer key (covered in the next chapter)
+
+For OAuth2, first, you need to generate a pair of valid ``client_id`` and ``client_secret``: you
+can proceed by [following this documentation](https://help.ovhcloud.com/csm/en-manage-service-account?id=kb_article_view&sysparm_article=KB0059343)
+
+Once you have retrieved your ``client_id`` and ``client_secret``, you can create and edit
+a configuration file that will be used by ``python-ovh``.
+
+4. Authorize your application to access a customer account using custom OVHcloud authentication
+***********************************************************************************************
To allow your application to access a customer account using the API on your
behalf, you need a **consumer key (CK)**.
@@ -164,7 +180,7 @@ Install a new mail redirection
------------------------------
e-mail redirections may be freely configured on domains and DNS zones hosted by
-OVH to an arbitrary destination e-mail using API call
+OVHcloud to an arbitrary destination e-mail using API call
``POST /email/domain/{domain}/redirection``.
For this call, the api specifies that the source address shall be given under the
@@ -195,7 +211,7 @@ is only supported with reserved keywords.
Grab bill list
--------------
-Let's say you want to integrate OVH bills into your own billing system, you
+Let's say you want to integrate OVHcloud bills into your own billing system, you
could just script around the ``/me/bills`` endpoints and even get the details
of each bill lines using ``/me/bill/{billId}/details/{billDetailId}``.
@@ -359,7 +375,7 @@ You have 3 ways to provide configuration to the client:
Embed the configuration in the code
-----------------------------------
-The straightforward way to use OVH's API keys is to embed them directly in the
+The straightforward way to use OVHcloud's API keys is to embed them directly in the
application code. While this is very convenient, it lacks of elegance and
flexibility.
@@ -547,8 +563,8 @@ build HTML documentation:
Supported APIs
==============
-OVH Europe
-----------
+OVHcloud Europe
+---------------
- **Documentation**: https://eu.api.ovh.com/
- **Community support**: api-subscribe@ml.ovh.net
@@ -556,16 +572,16 @@ OVH Europe
- **Create application credentials**: https://eu.api.ovh.com/createApp/
- **Create script credentials** (all keys at once): https://eu.api.ovh.com/createToken/
-OVH US
-----------
+OVHcloud US
+-----------
- **Documentation**: https://api.us.ovhcloud.com/
- **Console**: https://api.us.ovhcloud.com/console/
- **Create application credentials**: https://api.us.ovhcloud.com/createApp/
- **Create script credentials** (all keys at once): https://api.us.ovhcloud.com/createToken/
-OVH North America
------------------
+OVHcloud North America
+---------------------
- **Documentation**: https://ca.api.ovh.com/
- **Community support**: api-subscribe@ml.ovh.net
diff --git a/ovh/client.py b/ovh/client.py
index 92f5947..789bbc7 100644
--- a/ovh/client.py
+++ b/ovh/client.py
@@ -49,6 +49,7 @@
BadParametersError,
Forbidden,
HTTPError,
+ InvalidConfiguration,
InvalidCredential,
InvalidKey,
InvalidRegion,
@@ -60,8 +61,9 @@
ResourceExpiredError,
ResourceNotFoundError,
)
+from .oauth2 import OAuth2
-#: Mapping between OVH API region names and corresponding endpoints
+# Mapping between OVH API region names and corresponding endpoints
ENDPOINTS = {
"ovh-eu": "https://eu.api.ovh.com/1.0",
"ovh-us": "https://api.us.ovhcloud.com/1.0",
@@ -72,9 +74,16 @@
"soyoustart-ca": "https://ca.api.soyoustart.com/1.0",
}
-#: Default timeout for each request. 180 seconds connect, 180 seconds read.
+# Default timeout for each request. 180 seconds connect, 180 seconds read.
TIMEOUT = 180
+# OAuth2 token provider URLs
+OAUTH2_TOKEN_URLS = {
+ "ovh-eu": "https://www.ovh.com/auth/oauth2/token",
+ "ovh-ca": "https://ca.ovh.com/auth/oauth2/token",
+ "ovh-us": "https://us.ovhcloud.com/auth/oauth2/token",
+}
+
class Client:
"""
@@ -116,18 +125,24 @@ def __init__(
consumer_key=None,
timeout=TIMEOUT,
config_file=None,
+ client_id=None,
+ client_secret=None,
):
"""
Creates a new Client. No credential check is done at this point.
- The ``application_key`` identifies your application while
- ``application_secret`` authenticates it. On the other hand, the
- ``consumer_key`` uniquely identifies your application's end user without
- requiring his personal password.
+ When using OAuth2 authentication, ``client_id`` and ``client_secret``
+ will be used to initiate a 2-Legged OAuth2 flow.
+
+ When using the OVHcloud authentication method, the ``application_key``
+ identifies your application while ``application_secret`` authenticates
+ it. On the other hand, the ``consumer_key`` uniquely identifies your
+ application's end user without requiring his personal password.
- If any of ``endpoint``, ``application_key``, ``application_secret``
- or ``consumer_key`` is not provided, this client will attempt to locate
- from them from environment, ~/.ovh.cfg or /etc/ovh.cfg.
+ If any of ``endpoint``, ``application_key``, ``application_secret``,
+ ``consumer_key``, ``client_id`` or ``client_secret`` is not provided,
+ this client will attempt to locate from them from environment,
+ ``~/.ovh.cfg`` or ``/etc/ovh.cfg``.
See :py:mod:`ovh.config` for more information on supported
configuration mechanisms.
@@ -139,9 +154,11 @@ def __init__(
180 seconds for connection and 180 seconds for read.
:param str endpoint: API endpoint to use. Valid values in ``ENDPOINTS``
- :param str application_key: Application key as provided by OVH
- :param str application_secret: Application secret key as provided by OVH
+ :param str application_key: Application key as provided by OVHcloud
+ :param str application_secret: Application secret key as provided by OVHcloud
:param str consumer_key: uniquely identifies
+ :param str client_id: OAuth2 client ID
+ :param str client_secret: OAuth2 client secret
:param tuple timeout: Connection and read timeout for each request
:param float timeout: Same timeout for both connection and read
:raises InvalidRegion: if ``endpoint`` can't be found in ``ENDPOINTS``.
@@ -175,6 +192,50 @@ def __init__(
consumer_key = configuration.get(endpoint, "consumer_key")
self._consumer_key = consumer_key
+ # load OAuth2 data
+ if client_id is None:
+ client_id = configuration.get(endpoint, "client_id")
+ self._client_id = client_id
+
+ if client_secret is None:
+ client_secret = configuration.get(endpoint, "client_secret")
+ self._client_secret = client_secret
+
+ # configuration validation
+ if bool(self._client_id) is not bool(self._client_secret):
+ raise InvalidConfiguration("Invalid OAuth2 config, both client_id and client_secret must be given")
+
+ if bool(self._application_key) is not bool(self._application_secret):
+ raise InvalidConfiguration(
+ "Invalid authentication config, both application_key and application_secret must be given"
+ )
+
+ if self._client_id is not None and self._application_key is not None:
+ raise InvalidConfiguration(
+ "Can't use both application_key/application_secret and OAuth2 client_id/client_secret"
+ )
+ if self._client_id is None and self._application_key is None:
+ raise InvalidConfiguration(
+ "Missing authentication information, you need to provide at least an application_key/application_secret"
+ " or a client_id/client_secret"
+ )
+ if self._client_id and endpoint not in OAUTH2_TOKEN_URLS:
+ raise InvalidConfiguration(
+ "OAuth2 authentication is not compatible with endpoint "
+ + endpoint
+ + " (it can only be used with ovh-eu, ovh-ca and ovh-us)"
+ )
+
+ # when in OAuth2 mode, instantiate the oauthlib client
+ if self._client_id:
+ self._oauth2 = OAuth2(
+ client_id=self._client_id,
+ client_secret=self._client_secret,
+ token_url=OAUTH2_TOKEN_URLS[endpoint],
+ )
+ else:
+ self._oauth2 = None
+
# lazy load time delta
self._time_delta = None
@@ -524,7 +585,6 @@ def raw_call(self, method, path, data=None, need_auth=True, headers=None):
if headers is None:
headers = {}
- headers["X-Ovh-Application"] = self._application_key
# include payload
if data is not None:
@@ -533,6 +593,9 @@ def raw_call(self, method, path, data=None, need_auth=True, headers=None):
# sign request. Never sign 'time' or will recurse infinitely
if need_auth:
+ if self._oauth2:
+ return self._oauth2.session.request(method, target, headers=headers, data=body, timeout=self._timeout)
+
if not self._application_secret:
raise InvalidKey("Invalid ApplicationSecret '%s'" % self._application_secret)
@@ -551,4 +614,5 @@ def raw_call(self, method, path, data=None, need_auth=True, headers=None):
headers["X-Ovh-Timestamp"] = now
headers["X-Ovh-Signature"] = "$1$" + signature.hexdigest()
+ headers["X-Ovh-Application"] = self._application_key
return self._session.request(method, target, headers=headers, data=body, timeout=self._timeout)
diff --git a/ovh/config.py b/ovh/config.py
index 6e7d15a..4e604fc 100644
--- a/ovh/config.py
+++ b/ovh/config.py
@@ -49,6 +49,8 @@
application_key=my_app_key
application_secret=my_application_secret
consumer_key=my_consumer_key
+ client_id=my_client_id
+ client_secret=my_client_secret
The client will successively attempt to locate this configuration file in
diff --git a/ovh/exceptions.py b/ovh/exceptions.py
index 6109d74..d622af3 100644
--- a/ovh/exceptions.py
+++ b/ovh/exceptions.py
@@ -59,6 +59,10 @@ class InvalidCredential(APIError):
"""Raised when trying to sign request with invalid consumer key"""
+class InvalidConfiguration(APIError):
+ """Raised when trying to load an invalid configuration into a client"""
+
+
class InvalidResponse(APIError):
"""Raised when api response is not valid json"""
@@ -101,3 +105,7 @@ class Forbidden(APIError):
class ResourceExpiredError(APIError):
"""Raised when requested resource expired."""
+
+
+class OAuth2FailureError(APIError):
+ """Raised when the OAuth2 workflow fails"""
diff --git a/ovh/oauth2.py b/ovh/oauth2.py
new file mode 100644
index 0000000..3663a05
--- /dev/null
+++ b/ovh/oauth2.py
@@ -0,0 +1,119 @@
+# Copyright (c) 2013-2024, OVH SAS.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+# * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+# * Redistributions in binary form must reproduce the above copyright
+# notice, this list of conditions and the following disclaimer in the
+# documentation and/or other materials provided with the distribution.
+# * Neither the name of OVH SAS nor the
+# names of its contributors may be used to endorse or promote products
+# derived from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY OVH SAS AND CONTRIBUTORS ````AS IS'' AND ANY
+# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL OVH SAS AND CONTRIBUTORS BE LIABLE FOR ANY
+# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+"""
+Thanks to https://github.com/requests/requests-oauthlib/issues/260
+"""
+
+from oauthlib.oauth2 import BackendApplicationClient, MissingTokenError, OAuth2Error, TokenExpiredError
+from requests_oauthlib import OAuth2Session
+
+from .exceptions import OAuth2FailureError
+
+
+class RefreshOAuth2Session(OAuth2Session):
+ _error = None
+
+ def __init__(self, token_url, **kwargs):
+ self.token_url = token_url
+ super().__init__(**kwargs)
+ self.register_compliance_hook("access_token_response", self.save_error)
+ self.register_compliance_hook("refresh_token_response", self.save_error)
+
+ # This hijacks the
+ def save_error(self, resp):
+ if 200 <= resp.status_code <= 299:
+ self._error = "Received invalid body: " + resp.text
+ if resp.status_code >= 400:
+ self._error = "Token creation failed with status_code={}, body={}".format(resp.status_code, resp.text)
+ return resp
+
+ def fetch_token(self, *args, **kwargs):
+ try:
+ return super().fetch_token(*args, **kwargs)
+ except MissingTokenError as e:
+ desc = "OAuth2 failure: " + e.description
+ if self._error:
+ desc += " " + self._error
+
+ raise OAuth2FailureError(desc) from e
+ except OAuth2Error as e:
+ raise OAuth2FailureError("OAuth2 failure: " + str(e)) from e
+
+ def request(self, *args, **kwargs):
+ try:
+ return super().request(*args, **kwargs)
+ except TokenExpiredError:
+ self.token = self.fetch_token(token_url=self.token_url, **self.auto_refresh_kwargs)
+ self.token_updater(self.token)
+ return super().request(*args, **kwargs)
+
+
+class OAuth2:
+ _session = None
+ _token = None
+
+ def __init__(self, client_id, client_secret, token_url):
+ self.client_id = client_id
+ self.client_secret = client_secret
+ self.token_url = token_url
+
+ def token_updater(self, token):
+ self._token = token
+
+ @property
+ def session(self):
+ if self._session is None:
+ self._session = RefreshOAuth2Session(
+ token_url=self.token_url,
+ client=BackendApplicationClient(
+ client_id=self.client_id,
+ scope=["all"],
+ ),
+ token=self.token,
+ token_updater=self.token_updater,
+ auto_refresh_kwargs={
+ "client_id": self.client_id,
+ "client_secret": self.client_secret,
+ },
+ )
+ return self._session
+
+ @property
+ def token(self):
+ if self._token is None:
+ self._token = RefreshOAuth2Session(
+ token_url=self.token_url,
+ client=BackendApplicationClient(
+ client_id=self.client_id,
+ scope=["all"],
+ ),
+ ).fetch_token(
+ token_url=self.token_url,
+ client_id=self.client_id,
+ client_secret=self.client_secret,
+ )
+ return self._token
diff --git a/setup.cfg b/setup.cfg
index 625976f..6c6ea3b 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -35,7 +35,8 @@ setup_requires =
setuptools>=30.3.0
# requests: we need ssl+pooling fix from https://docs.python-requests.org/en/latest/community/updates/#id40
install_requires =
- requests>=2.11.0
+ requests>=2.31.0
+ requests-oauthlib>=2.0.0
include_package_data = True
[options.packages.find]
diff --git a/tests/data/user_both.ini b/tests/data/user_both.ini
new file mode 100644
index 0000000..b0fa43a
--- /dev/null
+++ b/tests/data/user_both.ini
@@ -0,0 +1,5 @@
+[ovh-eu]
+application_key=user
+application_secret=user
+client_id=foo
+client_secret=bar
\ No newline at end of file
diff --git a/tests/data/user_oauth2.ini b/tests/data/user_oauth2.ini
new file mode 100644
index 0000000..0501976
--- /dev/null
+++ b/tests/data/user_oauth2.ini
@@ -0,0 +1,3 @@
+[ovh-eu]
+client_id=foo
+client_secret=bar
\ No newline at end of file
diff --git a/tests/data/user_oauth2_incompatible.ini b/tests/data/user_oauth2_incompatible.ini
new file mode 100644
index 0000000..2bfcebb
--- /dev/null
+++ b/tests/data/user_oauth2_incompatible.ini
@@ -0,0 +1,3 @@
+[kimsufi-eu]
+client_id=foo
+client_secret=bar
\ No newline at end of file
diff --git a/tests/test_client.py b/tests/test_client.py
index efce8e1..070ecbb 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -24,6 +24,7 @@
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+import time
from unittest import mock
import pytest
@@ -41,6 +42,7 @@
NetworkError,
NotCredential,
NotGrantedCall,
+ OAuth2FailureError,
ResourceConflictError,
ResourceExpiredError,
ResourceNotFoundError,
@@ -57,7 +59,7 @@ class TestClient:
@mock.patch("time.time", return_value=1457018875.467238)
@mock.patch.object(Client, "call", return_value=1457018881)
def test_time_delta(self, m_call, m_time):
- api = Client("ovh-eu")
+ api = Client("ovh-eu", MockApplicationKey, MockApplicationSecret)
assert api._time_delta is None
assert m_call.called is False
assert m_time.called is False
@@ -76,7 +78,7 @@ def test_time_delta(self, m_call, m_time):
@mock.patch.object(Client, "call", return_value={"consumerKey": "CK"})
def test_request_consumerkey(self, m_call):
- api = Client("ovh-eu")
+ api = Client("ovh-eu", MockApplicationKey, MockApplicationSecret)
ret = api.request_consumerkey([{"method": "GET", "path": "/"}], "https://example.com", ["127.0.0.1/32"])
m_call.assert_called_once_with(
@@ -92,14 +94,14 @@ def test_request_consumerkey(self, m_call):
assert ret == {"consumerKey": "CK"}
def test_new_consumer_key_request(self):
- api = Client("ovh-eu")
+ api = Client("ovh-eu", MockApplicationKey, MockApplicationSecret)
ck = api.new_consumer_key_request()
assert ck._client == api
# test wrappers
def test__canonicalize_kwargs(self):
- api = Client("ovh-eu")
+ api = Client("ovh-eu", MockApplicationKey, MockApplicationSecret)
assert api._canonicalize_kwargs({}) == {}
assert api._canonicalize_kwargs({"from": "value"}) == {"from": "value"}
assert api._canonicalize_kwargs({"_to": "value"}) == {"_to": "value"}
@@ -107,7 +109,7 @@ def test__canonicalize_kwargs(self):
@mock.patch.object(Client, "call")
def test_query_string(self, m_call):
- api = Client("ovh-eu")
+ api = Client("ovh-eu", MockApplicationKey, MockApplicationSecret)
for method, call in (("GET", api.get), ("DELETE", api.delete)):
m_call.reset_mock()
@@ -128,7 +130,7 @@ def test_query_string(self, m_call):
@mock.patch.object(Client, "call")
def test_body(self, m_call):
- api = Client("ovh-eu")
+ api = Client("ovh-eu", MockApplicationKey, MockApplicationSecret)
for method, call in (("POST", api.post), ("PUT", api.put)):
m_call.reset_mock()
@@ -202,7 +204,7 @@ def test_call_query_id(self, m_req):
m_res.status_code = 99
m_res.headers = {"X-OVH-QUERYID": "FR.test1"}
- api = Client("ovh-eu", application_key=MockApplicationKey)
+ api = Client("ovh-eu", MockApplicationKey, MockApplicationSecret)
with pytest.raises(APIError) as e:
api.call("GET", "/unit/test", None, False)
assert e.value.query_id == "FR.test1"
@@ -211,7 +213,7 @@ def test_call_query_id(self, m_req):
def test_call_errors(self, m_req):
m_res = m_req.return_value
- api = Client("ovh-eu", application_key=MockApplicationKey)
+ api = Client("ovh-eu", MockApplicationKey, MockApplicationSecret)
# request fails, somehow
m_req.side_effect = requests.RequestException
@@ -246,16 +248,13 @@ def test_call_errors(self, m_req):
api.call("GET", "/unauth", None, False)
# errors
- api = Client("ovh-eu", MockApplicationKey, None, MockConsumerKey)
- with pytest.raises(InvalidKey):
- api.call("GET", "/unit/test", None, True)
api = Client("ovh-eu", MockApplicationKey, MockApplicationSecret, None)
with pytest.raises(InvalidKey):
api.call("GET", "/unit/test", None, True)
@mock.patch("ovh.client.Session.request", return_value="Let's assume requests will return this")
def test_raw_call_with_headers(self, m_req):
- api = Client("ovh-eu", MockApplicationKey)
+ api = Client("ovh-eu", MockApplicationKey, MockApplicationSecret)
r = api.raw_call("GET", "/unit/path", None, False, headers={"Custom-Header": "1"})
assert r == "Let's assume requests will return this"
assert m_req.call_args_list == [
@@ -274,7 +273,7 @@ def test_raw_call_with_headers(self, m_req):
# Perform real API tests.
def test_endpoints(self):
for endpoint in ENDPOINTS.keys():
- auth_time = Client(endpoint).get("/auth/time", _need_auth=False)
+ auth_time = Client(endpoint, MockApplicationKey, MockApplicationSecret).get("/auth/time", _need_auth=False)
assert auth_time > 0
@mock.patch("time.time", return_value=1457018875.467238)
@@ -308,3 +307,104 @@ def _h(prefix):
mock.call("GET", "https://eu.api.ovh.com/v1/call", headers=_h("v1"), data="", timeout=180),
mock.call("GET", "https://eu.api.ovh.com/v2/call", headers=_h("v2"), data="", timeout=180),
]
+
+ @mock.patch("ovh.client.Session.request")
+ def test_oauth2(self, m_req):
+ def resp(*args, **kwargs):
+ if args[0] == "POST" and args[1] == "https://www.ovh.com/auth/oauth2/token":
+ resp = mock.Mock()
+ resp.status_code = 200
+ resp.text = """{
+ "access_token":"MTQ0NjJkZmQ5OTM2NDE1ZTZjNGZmZjI3",
+ "token_type":"Bearer",
+ "expires_in":3,
+ "scope":"all"
+}"""
+ return resp
+
+ if args[0] == "GET" and args[1] == "https://eu.api.ovh.com/1.0/call":
+ resp = mock.Mock()
+ resp.status_code = 200
+ resp.text = "{}"
+ return resp
+
+ raise NotImplementedError("FIXME")
+
+ m_req.side_effect = resp
+
+ call_oauth = mock.call(
+ "POST",
+ "https://www.ovh.com/auth/oauth2/token",
+ headers={"Accept": "application/json", "Content-Type": "application/x-www-form-urlencoded"},
+ data={"grant_type": "client_credentials", "scope": "all"},
+ files=None,
+ timeout=None,
+ auth=mock.ANY,
+ verify=None,
+ proxies=None,
+ cert=None,
+ )
+ call_api = mock.call(
+ "GET",
+ "https://eu.api.ovh.com/1.0/call",
+ headers={"Authorization": "Bearer MTQ0NjJkZmQ5OTM2NDE1ZTZjNGZmZjI3"},
+ data="",
+ files=None,
+ timeout=180,
+ )
+
+ # First call triggers the fetch of a token, then the real call
+ api = Client("ovh-eu", client_id="oauth2_id", client_secret="oauth2_secret")
+ api.call("GET", "/call", None, True)
+ assert m_req.call_args_list == [call_oauth, call_api]
+
+ # Calling the API again does not trigger the fetch of a new token
+ api.call("GET", "/call", None, True)
+ assert m_req.call_args_list == [call_oauth, call_api, call_api]
+
+ # The fetched token had an `expires_in` set to 3, sleep more than that, which makes us fetch a now token
+ time.sleep(4)
+ api.call("GET", "/call", None, True)
+ assert m_req.call_args_list == [call_oauth, call_api, call_api, call_oauth, call_api]
+
+ @mock.patch("ovh.client.Session.request")
+ def test_oauth2_503(self, m_req):
+ m_res = m_req.return_value
+ m_res.status_code = 503
+ m_res.text = "
test
"
+
+ api = Client("ovh-eu", client_id="oauth2_id", client_secret="oauth2_secret")
+
+ with pytest.raises(OAuth2FailureError) as e:
+ api.call("GET", "/call", None, True)
+ assert str(e.value) == (
+ "OAuth2 failure: Missing access token parameter. Token creation failed with status_code=503, "
+ "body=test
"
+ )
+
+ @mock.patch("ovh.client.Session.request")
+ def test_oauth2_bad_json(self, m_req):
+ m_res = m_req.return_value
+ m_res.status_code = 200
+ m_res.text = "test
"
+
+ api = Client("ovh-eu", client_id="oauth2_id", client_secret="oauth2_secret")
+
+ with pytest.raises(OAuth2FailureError) as e:
+ api.call("GET", "/call", None, True)
+ assert str(e.value) == (
+ "OAuth2 failure: Missing access token parameter. Received invalid body: "
+ "test
"
+ )
+
+ @mock.patch("ovh.client.Session.request")
+ def test_oauth2_unknown_client(self, m_req):
+ m_res = m_req.return_value
+ m_res.status_code = 200
+ m_res.text = '{"error":"invalid_client", "error_description":"ovhcloud oauth2 client does not exists"}'
+
+ api = Client("ovh-eu", client_id="oauth2_id", client_secret="oauth2_secret")
+
+ with pytest.raises(OAuth2FailureError) as e:
+ api.call("GET", "/call", None, True)
+ assert str(e.value) == "OAuth2 failure: (invalid_client) ovhcloud oauth2 client does not exists"
diff --git a/tests/test_config.py b/tests/test_config.py
index bfb6ebc..81801f6 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -24,6 +24,7 @@
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+from configparser import MissingSectionHeaderError
import os
from pathlib import Path
from unittest.mock import patch
@@ -31,11 +32,16 @@
import pytest
import ovh
+from ovh.exceptions import InvalidConfiguration, InvalidRegion
TEST_DATA = str(Path(__file__).resolve().parent / "data")
systemConf = TEST_DATA + "/system.ini"
userPartialConf = TEST_DATA + "/userPartial.ini"
userConf = TEST_DATA + "/user.ini"
+userOAuth2Conf = TEST_DATA + "/user_oauth2.ini"
+userOAuth2InvalidConf = TEST_DATA + "/user_oauth2_invalid.ini"
+userOAuth2IncompatibleConfig = TEST_DATA + "/user_oauth2_incompatible.ini"
+userBothConf = TEST_DATA + "/user_both.ini"
localPartialConf = TEST_DATA + "/localPartial.ini"
doesNotExistConf = TEST_DATA + "/doesNotExist.ini"
invalidINIConf = TEST_DATA + "/invalid.ini"
@@ -76,24 +82,58 @@ def test_config_from_only_one_file(self):
@patch("ovh.config.CONFIG_PATH", [doesNotExistConf])
def test_config_from_non_existing_file(self):
- client = ovh.Client(endpoint="ovh-eu")
- assert client._application_key is None
- assert client._application_secret is None
- assert client._consumer_key is None
+ with pytest.raises(InvalidConfiguration) as e:
+ ovh.Client(endpoint="ovh-eu")
+
+ assert str(e.value) == (
+ "Missing authentication information, you need to provide at least an "
+ "application_key/application_secret or a client_id/client_secret"
+ )
@patch("ovh.config.CONFIG_PATH", [invalidINIConf])
def test_config_from_invalid_ini_file(self):
- from configparser import MissingSectionHeaderError
-
with pytest.raises(MissingSectionHeaderError):
ovh.Client(endpoint="ovh-eu")
@patch("ovh.config.CONFIG_PATH", [errorConf])
def test_config_from_invalid_file(self):
+ with pytest.raises(InvalidConfiguration) as e:
+ ovh.Client(endpoint="ovh-eu")
+
+ assert str(e.value) == (
+ "Missing authentication information, you need to provide at least an "
+ "application_key/application_secret or a client_id/client_secret"
+ )
+
+ @patch("ovh.config.CONFIG_PATH", [userOAuth2Conf])
+ def test_config_oauth2(self):
client = ovh.Client(endpoint="ovh-eu")
- assert client._application_key is None
- assert client._application_secret is None
- assert client._consumer_key is None
+ assert client._client_id == "foo"
+ assert client._client_secret == "bar"
+
+ @patch("ovh.config.CONFIG_PATH", [userBothConf])
+ def test_config_invalid_both(self):
+ with pytest.raises(InvalidConfiguration) as e:
+ ovh.Client(endpoint="ovh-eu")
+
+ assert str(e.value) == "Can't use both application_key/application_secret and OAuth2 client_id/client_secret"
+
+ @patch("ovh.config.CONFIG_PATH", [userOAuth2InvalidConf])
+ def test_config_invalid_oauth2(self):
+ with pytest.raises(InvalidConfiguration) as e:
+ ovh.Client(endpoint="ovh-eu")
+
+ assert str(e.value) == "Invalid OAuth2 config, both client_id and client_secret must be given"
+
+ @patch("ovh.config.CONFIG_PATH", [userOAuth2IncompatibleConfig])
+ def test_config_incompatible_oauth2(self):
+ with pytest.raises(InvalidConfiguration) as e:
+ ovh.Client(endpoint="kimsufi-eu")
+
+ assert str(e.value) == (
+ "OAuth2 authentication is not compatible with endpoint kimsufi-eu "
+ + "(it can only be used with ovh-eu, ovh-ca and ovh-us)"
+ )
@patch("ovh.config.CONFIG_PATH", [userConf])
@patch.dict(
@@ -121,7 +161,5 @@ def test_config_from_args(self):
assert client._consumer_key == "param"
def test_invalid_endpoint(self):
- from ovh.exceptions import InvalidRegion
-
with pytest.raises(InvalidRegion):
ovh.Client(endpoint="not_existing")