kgess-mirror/python-ovh
1
0
Fork 0
mirror of https://github.com/ovh/python-ovh.git synced 2026-01-12 06:52:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat: handle Client Credential OAuth2 authentication method

Browse source 
Signed-off-by: Adrien Barreau <adrien.barreau@ovhcloud.com>
This commit is contained in:
Adrien Barreau 2024-06-26 17:03:47 +00:00
parent 3f40c7d573
commit d1b0c7867e
12 changed files with 423 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

52
README.rst
View file

@ -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 <https://eu.api.ovh.com/createApp/>`_
- `OVH US <https://api.us.ovhcloud.com/createApp/>`_
- `OVH North-America <https://ca.api.ovh.com/createApp/>`_
- `OVHcloud Europe <https://eu.api.ovh.com/createApp/>`_
- `OVHcloud US <https://api.us.ovhcloud.com/createApp/>`_
- `OVHcloud North-America <https://ca.api.ovh.com/createApp/>`_
- `So you Start Europe <https://eu.api.soyoustart.com/createApp/>`_
- `So you Start North America <https://ca.api.soyoustart.com/createApp/>`_
- `Kimsufi Europe <https://eu.api.kimsufi.com/createApp/>`_
@ -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 enable 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

88
ovh/client.py
View file

@ -49,6 +49,7 @@ from .exceptions import (
BadParametersError,
Forbidden,
HTTPError,
InvalidConfiguration,
InvalidCredential,
InvalidKey,
InvalidRegion,
@ -60,8 +61,9 @@ from .exceptions import (
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 @@ ENDPOINTS = {
"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 @@ class Client:
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 Client Credential OAuth2 flow.
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.
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``,
``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 @@ class Client:
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 @@ class Client:
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 @@ class Client:
if headers is None:
headers = {}
headers["X-Ovh-Application"] = self._application_key
# include payload
if data is not None:
@ -533,6 +593,9 @@ class Client:
# 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 @@ class Client:
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)

2
ovh/config.py
View file

@ -49,6 +49,8 @@ provided, it will look for a configuration file of the form:
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

8
ovh/exceptions.py
View file

@ -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"""

125
ovh/oauth2.py Normal file
View file

@ -0,0 +1,125 @@
# 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 for the base used in this file.
"""
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)
# This hijacks the hook mechanism to save details about the last token creation failure.
# For now, there is no easy other way to access to these details;
# see https://github.com/requests/requests-oauthlib/pull/441
self.register_compliance_hook("access_token_response", self.save_error)
self.register_compliance_hook("refresh_token_response", self.save_error)
# See __init__, used as compliance hooks
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
# Wraps OAuth2Session.fetch_token to enrich returned exception messages, wrapped in an unique class
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
# Wraps OAuth2Session.request to handle TokenExpiredError by fetching a new token and retrying
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

3
setup.cfg
View file

@ -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]

5
tests/data/user_both.ini Normal file
View file

@ -0,0 +1,5 @@
[ovh-eu]
application_key=user
application_secret=user
client_id=foo
client_secret=bar

3
tests/data/user_oauth2.ini Normal file
View file

@ -0,0 +1,3 @@
[ovh-eu]
client_id=foo
client_secret=bar

3
tests/data/user_oauth2_incompatible.ini Normal file
View file

@ -0,0 +1,3 @@
[kimsufi-eu]
client_id=foo
client_secret=bar

3
tests/data/user_oauth2_invalid.ini Normal file
View file

@ -0,0 +1,3 @@
[ovh-eu]
client_id=foo
client_secret=

126
tests/test_client.py
View file

@ -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 @@ from ovh.exceptions import (
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 @@ class TestClient:
@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 @@ class TestClient:
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 @@ class TestClient:
@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 @@ class TestClient:
@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 @@ class TestClient:
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 @@ class TestClient:
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 @@ class TestClient:
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 @@ class TestClient:
# 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 @@ class TestClient:
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 = "<html><body><p>test</p></body></html>"
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=<html><body><p>test</p></body></html>"
)
@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 = "<html><body><p>test</p></body></html>"
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: "
"<html><body><p>test</p></body></html>"
)
@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"

60
tests/test_config.py
View file

@ -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 @@ from unittest.mock import patch
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 @@ class TestConfig:
@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 @@ class TestConfig:
assert client._consumer_key == "param"
def test_invalid_endpoint(self):
from ovh.exceptions import InvalidRegion
with pytest.raises(InvalidRegion):
ovh.Client(endpoint="not_existing")