vendors: token helper for SAP BTP, ABAP env service (#159)

Author: s7oev

CHANGELOG.md

@@ -6,6 +6,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
+### Fixed
+- Helper for obtaining a token for services coming from SAP BTP, ABAP environment - Stoyko Stoev
+
 ## [1.7.0]
 
 ### Added

docs/usage/vendors.rst

@@ -0,0 +1,36 @@
+Vendor-specific helpers
+==============
+
+This document presents the helpers found in ``pyodata.vendor`` 
+
+Get the service proxy client for an OData service published on SAP BTP, ABAP environment
+--------------------------------------------------------------------------
+
+Let's assume you have an ABAP environment service instance running on SAP Business Technology
+Platform. You have used this instance to provide an OData service by using, for example, the
+ABAP RESTful Application Programming Model. To connect to it, you need to provide several attributes 
+found in the JSON service key of the instance, as well as your username and password for SAP BTP.
+
+``pyodata.vendor.SAP`` offers a helper that takes the arguments described above, as well as an
+existing ``requests.Session`` object (or another one conforming to the same API), and adds the
+required token to the session object's authorization header.
+
+The following code demonstrates using the helper.
+
+.. code-block:: python
+  
+  import pyodata
+  from pyodata.vendor import SAP
+  import requests
+  import json
+
+  with open('key.txt', 'r') as f:
+      KEY = json.loads(f.read())
+
+  USER = "MyBtpUser"
+  PASSWORD = "MyBtpPassword"
+  SERVICE_URL = KEY["url"] + "/sap/opu/odata/sap/" + "ZMyBtpService"
+
+  session = SAP.add_btp_token_to_session(requests.Session(), KEY, USER, PASSWORD)
+  # do something more with session object if necessary (e.g. adding sap-client parameter, or CSRF token)
+  client = pyodata.Client(SERVICE_URL, session)

pyodata/vendor/SAP.py

@@ -24,6 +24,29 @@ def json_get(obj, member, typ, default=None):
     return value
 
 
+def add_btp_token_to_session(session, key, user, password):
+    """Using the provided credentials, the function tries to add the
+       necessary token for establishing a connection to an OData service
+       coming from SAP BTP, ABAP environment.
+
+       If any of the provided credentials are invalid, the server will
+       respond with 401, and the function will raise HttpError.
+    """
+    token_url = key['uaa']['url'] + f'/oauth/token?grant_type=password&username={user}&password={password}'
+    token_response = session.post(token_url, auth=(key['uaa']['clientid'], key['uaa']['clientsecret']))
+
+    if token_response.status_code != 200:
+        raise HttpError(
+            f'Token request failed, status code: {token_response.status_code}, body:\n{token_response.content}',
+            token_response)
+
+    token_response = json.loads(token_response.text)
+    token = token_response['id_token']
+
+    session.headers.update({'Authorization': f'Bearer {token}'})
+    return session
+
+
 class BusinessGatewayError(HttpError):
     """To display the right error message"""
 

tests/test_vendor_sap.py

@@ -5,6 +5,9 @@
 import pytest
 from pyodata.exceptions import PyODataException, HttpError
 from pyodata.vendor import SAP
+import responses
+import requests
+import json
 
 
 class MockResponse(NamedTuple):
@@ -183,3 +186,88 @@ def test_vendor_http_error(response_with_error):
     sap_error = HttpError('Another foo bar', response_with_error)
     assert isinstance(sap_error, SAP.BusinessGatewayError)
     assert str(sap_error) == 'Gateway Error'
+
+
+MOCK_AUTH_URL = "https://example.authentication.hana.ondemand.com"
+MOCK_BTP_USER = "[email protected]"
+MOCK_BTP_PASSWORD = "example_password"
+MOCK_KEY = {
+    "uaa": {
+        "url": MOCK_AUTH_URL,
+        "clientid": "example-client-id",
+        "clientsecret": "example-client-secret"
+    }
+}
+
+
+@responses.activate
+def test_add_btp_token_to_session_valid():
+    """Valid username, password and key return a session with set token"""
+
+    responses.add(
+        responses.POST,
+        MOCK_AUTH_URL + f'/oauth/token?grant_type=password&username={MOCK_BTP_USER}&password={MOCK_BTP_PASSWORD}',
+        headers={'Content-type': 'application/json'},
+        json={
+            'access_token': 'valid_access_token',
+            'token_type': 'bearer',
+            'id_token': 'valid_id_token',
+            'refresh_token': 'valid_refresh_token',
+            'expires_in': 43199,
+            'scope': 'openid uaa.user',
+            'jti': 'valid_jti'
+        },
+        status=200)
+
+    result = SAP.add_btp_token_to_session(requests.Session(), MOCK_KEY, MOCK_BTP_USER, MOCK_BTP_PASSWORD)
+    assert result.headers['Authorization'] == 'Bearer valid_id_token'
+
+
+@responses.activate
+def test_add_btp_token_to_session_invalid_user():
+    """Invalid username returns an HttpError"""
+
+    invalid_user = "[email protected]"
+
+    responses.add(
+        responses.POST,
+        MOCK_AUTH_URL + f'/oauth/token?grant_type=password&username={invalid_user}&password={MOCK_BTP_PASSWORD}',
+        headers={'Content-type': 'application/json'},
+        json={
+            'error': 'unauthorized',
+            'error_description': {
+                'error': 'invalid_grant',
+                'error_description': 'User authentication failed.'
+            }
+        },
+        status=401)
+
+    with pytest.raises(HttpError) as caught:
+        SAP.add_btp_token_to_session(requests.Session(), MOCK_KEY, invalid_user, MOCK_BTP_PASSWORD)
+
+    assert caught.value.response.status_code == 401
+    assert json.loads(caught.value.response.text)['error_description']['error'] == 'invalid_grant'
+
+
+@responses.activate
+def test_add_btp_token_to_session_invalid_clientid():
+    """Invalid clientid in key returns an HttpError"""
+
+    invalid_key = MOCK_KEY.copy()
+    invalid_key['uaa']['clientid'] = 'invalid-client-id'
+
+    responses.add(
+        responses.POST,
+        MOCK_AUTH_URL + f'/oauth/token?grant_type=password&username={MOCK_BTP_USER}&password={MOCK_BTP_PASSWORD}',
+        headers={'Content-type': 'application/json'},
+        json={
+            'error': 'unauthorized',
+            'error_description': 'Bad credentials'
+        },
+        status=401)
+
+    with pytest.raises(HttpError) as caught:
+        SAP.add_btp_token_to_session(requests.Session(), invalid_key, MOCK_BTP_USER, MOCK_BTP_PASSWORD)
+
+    assert caught.value.response.status_code == 401
+    assert json.loads(caught.value.response.text)['error_description'] == 'Bad credentials'