Adds a backend for BitBucket

skanct · skanct · commit 4f35ab115b4e · 2019-08-11T14:40:20.000+02:00
diff --git a/example/plugins/backends/bitbucket_backend.yaml.example b/example/plugins/backends/bitbucket_backend.yaml.example
@@ -0,0 +1,30 @@
+module: satosa.backends.bitbucket.BitBucketBackend
+name: bitbucket
+config:
+  authz_page: bitbucket/auth/callback
+  base_url:  <base_url>
+  client_config:
+    client_id: 
+  client_secret:
+  scope: ["account", "email"]
+  response_type: code
+  allow_signup: false
+  server_info: {
+    authorization_endpoint: 'https://bitbucket.org/site/oauth2/authorize',
+    token_endpoint: 'https://bitbucket.org/site/oauth2/access_token',
+    user_endpoint: 'https://api.bitbucket.org/2.0/user'
+  }
+  entity_info:
+    organization:
+      display_name:
+      - ["BitBucket", "en"]
+      name:
+      - ["BitBucket", "en"]
+      url:
+      - ["https://www.bitbucket.com/", "en"]
+    ui_info:
+      description:
+      - ["Login to a service using your BitBucket credentials", "en"]
+      display_name:
+      - ["BitBucket", "en"]
+
diff --git a/src/satosa/backends/bitbucket.py b/src/satosa/backends/bitbucket.py
@@ -0,0 +1,85 @@
+"""
+OAuth backend for BitBucket
+"""
+import json
+import logging
+import requests
+
+from oic.utils.authn.authn_context import UNSPECIFIED
+from oic.oauth2.consumer import stateID
+
+from satosa.backends.oauth import _OAuthBackend
+from satosa.internal import AuthenticationInformation
+
+logger = logging.getLogger(__name__)
+
+
+class BitBucketBackend(_OAuthBackend):
+    """BitBucket OAuth 2.0 backend"""
+
+    logprefix = "BitBucket Backend:"
+
+    def __init__(self, outgoing, internal_attributes, config, base_url, name):
+        """BitBucket backend constructor
+        :param outgoing: Callback should be called by the module after the
+            authorization in the backend is done.
+        :param internal_attributes: Mapping dictionary between SATOSA internal
+            attribute names and the names returned by underlying IdP's/OP's as
+            well as what attributes the calling SP's and RP's expects namevice.
+        :param config: configuration parameters for the module.
+        :param base_url: base url of the service
+        :param name: name of the plugin
+        :type outgoing:
+            (satosa.context.Context, satosa.internal.InternalData) ->
+            satosa.response.Response
+        :type internal_attributes: dict[string, dict[str, str | list[str]]]
+        :type config: dict[str, dict[str, str] | list[str] | str]
+        :type base_url: str
+        :type name: str
+        """
+        config.setdefault('response_type', 'code')
+        config['verify_accesstoken_state'] = False
+        super().__init__(outgoing, internal_attributes, config, base_url,
+                         name, 'bitbucket', 'account_id')
+
+    def get_request_args(self, get_state=stateID):
+        request_args = super().get_request_args(get_state=get_state)
+
+        client_id = self.config["client_config"]["client_id"]
+        extra_args = {
+            arg_name: arg_val
+            for arg_name in ["auth_type", "scope"]
+            for arg_val in [self.config.get(arg_name, [])]
+            if arg_val
+        }
+        extra_args.update({"client_id": client_id})
+        request_args.update(extra_args)
+        return request_args
+
+    def auth_info(self, request):
+        return AuthenticationInformation(
+            UNSPECIFIED, None,
+            self.config['server_info']['authorization_endpoint'])
+
+    def user_information(self, access_token):
+        url = self.config['server_info']['user_endpoint']
+        email_url = "{}/emails".format(url)
+        headers = {'Authorization': 'Bearer {}'.format(access_token)}
+        resp = requests.get(url, headers=headers)
+        data = json.loads(resp.text)
+        if 'email' in self.config['scope']:
+            resp = requests.get(email_url, headers=headers)
+            emails = json.loads(resp.text)
+            data.update({
+                'email': [e for e in [d.get('email')
+                                      for d in emails.get('values')
+                                      if d.get('is_primary')
+                                      ]
+                          ],
+                'email_confirmed': [e for e in [d.get('email')
+                                                for d in emails.get('values')
+                                                if d.get('is_confirmed')
+                                                ]
+                                    ]
+                })
+        return data
diff --git a/tests/satosa/backends/test_bitbucket.py b/tests/satosa/backends/test_bitbucket.py
@@ -0,0 +1,194 @@
+import json
+from unittest.mock import Mock
+from urllib.parse import urlparse, parse_qsl
+
+import pytest
+import responses
+
+from saml2.saml import NAMEID_FORMAT_TRANSIENT
+
+from satosa.backends.bitbucket import BitBucketBackend
+from satosa.internal import InternalData
+
+BB_USER_RESPONSE = {
+    "account_id": "bb_id",
+    "is_staff": False,
+    "username": "bb_username",
+    "nickname": "bb_username",
+    "display_name": "bb_first_name bb_last_name",
+    "has_2fa_enabled": False,
+    "created_on": "2019-10-12T09:14:00+0000"
+}
+BB_USER_EMAIL_RESPONSE = {
+    "values": [
+       {
+            "email": "bb_username@example.com",
+            "is_confirmed": True,
+            "is_primary": True
+        },
+       {
+            "email": "bb_username_1@example.com",
+            "is_confirmed": True,
+            "is_primary": False
+        },
+       {
+            "email": "bb_username_2@example.com",
+            "is_confirmed": False,
+            "is_primary": False
+        }
+    ]
+}
+BASE_URL = "https://client.example.com"
+AUTHZ_PAGE = 'bitbucket'
+CLIENT_ID = "bitbucket_client_id"
+BB_CONFIG = {
+    'server_info': {
+        'authorization_endpoint':
+            'https://bitbucket.org/site/oauth2/authorize',
+        'token_endpoint': 'https://bitbucket.org/site/oauth2/access_token',
+        'user_endpoint': 'https://api.bitbucket.org/2.0/user'
+    },
+    'client_secret': 'bitbucket_secret',
+    'base_url': BASE_URL,
+    'state_encryption_  key': 'state_encryption_key',
+    'encryption_key': 'encryption_key',
+    'authz_page': AUTHZ_PAGE,
+    'client_config': {'client_id': CLIENT_ID},
+    'scope': ["account", "email"]
+
+}
+BB_RESPONSE_CODE = "the_bb_code"
+
+INTERNAL_ATTRIBUTES = {
+    'attributes': {
+        'mail': {'bitbucket': ['email']},
+        'subject-id': {'bitbucket': ['account_id']},
+        'displayname': {'bitbucket': ['display_name']},
+        'name': {'bitbucket': ['display_name']},
+    }
+}
+
+mock_get_state = Mock(return_value="abcdef")
+
+
+class TestBitBucketBackend(object):
+    @pytest.fixture(autouse=True)
+    def create_backend(self):
+        self.bb_backend = BitBucketBackend(Mock(), INTERNAL_ATTRIBUTES,
+                                           BB_CONFIG, "base_url", "bitbucket")
+
+    @pytest.fixture
+    def incoming_authn_response(self, context):
+        context.path = 'bitbucket/sso/redirect'
+        state_data = dict(state=mock_get_state.return_value)
+        context.state[self.bb_backend.name] = state_data
+        context.request = {
+            "code": BB_RESPONSE_CODE,
+            "state": mock_get_state.return_value
+        }
+
+        return context
+
+    def setup_bitbucket_response(self):
+        _user_endpoint = BB_CONFIG['server_info']['user_endpoint']
+        responses.add(responses.GET,
+                      _user_endpoint,
+                      body=json.dumps(BB_USER_RESPONSE),
+                      status=200,
+                      content_type='application/json')
+
+        responses.add(responses.GET,
+                      '{}/emails'.format(_user_endpoint),
+                      body=json.dumps(BB_USER_EMAIL_RESPONSE),
+                      status=200,
+                      content_type='application/json')
+
+    def assert_expected_attributes(self):
+        expected_attributes = {
+            "subject-id": [BB_USER_RESPONSE["account_id"]],
+            "name": [BB_USER_RESPONSE["display_name"]],
+            "displayname": [BB_USER_RESPONSE["display_name"]],
+            "mail": [BB_USER_EMAIL_RESPONSE["values"][0]["email"]],
+        }
+
+        context, internal_resp = self.bb_backend \
+            .auth_callback_func \
+            .call_args[0]
+        assert internal_resp.attributes == expected_attributes
+
+    def assert_token_request(self, request_args, state, **kwargs):
+        assert request_args["code"] == BB_RESPONSE_CODE
+        assert request_args["redirect_uri"] == "%s/%s" % (BASE_URL, AUTHZ_PAGE)
+        assert request_args["state"] == mock_get_state.return_value
+        assert state == mock_get_state.return_value
+
+    def test_register_endpoints(self):
+        url_map = self.bb_backend.register_endpoints()
+        expected_url_map = [('^bitbucket$', self.bb_backend._authn_response)]
+        assert url_map == expected_url_map
+
+    def test_start_auth(self, context):
+        context.path = 'bitbucket/sso/redirect'
+        internal_request = InternalData(
+            subject_type=NAMEID_FORMAT_TRANSIENT, requester='test_requester'
+        )
+
+        resp = self.bb_backend.start_auth(context,
+                                          internal_request,
+                                          mock_get_state)
+        login_url = resp.message
+        assert login_url.startswith(
+                BB_CONFIG["server_info"]["authorization_endpoint"])
+        expected_params = {
+            "client_id": CLIENT_ID,
+            "state": mock_get_state.return_value,
+            "response_type": "code",
+            "scope": " ".join(BB_CONFIG["scope"]),
+            "redirect_uri": "%s/%s" % (BASE_URL, AUTHZ_PAGE)
+        }
+        actual_params = dict(parse_qsl(urlparse(login_url).query))
+        assert actual_params == expected_params
+
+    @responses.activate
+    def test_authn_response(self, incoming_authn_response):
+        self.setup_bitbucket_response()
+
+        mock_do_access_token_request = Mock(
+                return_value={"access_token": "bb access token"})
+        self.bb_backend.consumer.do_access_token_request = \
+            mock_do_access_token_request
+
+        self.bb_backend._authn_response(incoming_authn_response)
+        assert self.bb_backend.name not in incoming_authn_response.state
+
+        self.assert_expected_attributes()
+        self.assert_token_request(**mock_do_access_token_request.call_args[1])
+
+    @responses.activate
+    def test_entire_flow(self, context):
+        """
+        Tests start of authentication (incoming auth req) and receiving auth
+        response.
+        """
+        responses.add(responses.POST,
+                      BB_CONFIG["server_info"]["token_endpoint"],
+                      body=json.dumps({"access_token": "qwerty",
+                                       "token_type": "bearer",
+                                       "expires_in": 9999999999999}),
+                      status=200,
+                      content_type='application/json')
+        self.setup_bitbucket_response()
+
+        context.path = 'bitbucket/sso/redirect'
+        internal_request = InternalData(
+            subject_type=NAMEID_FORMAT_TRANSIENT, requester='test_requester'
+        )
+
+        self.bb_backend.start_auth(context, internal_request, mock_get_state)
+        context.request = {
+            "code": BB_RESPONSE_CODE,
+            "state": mock_get_state.return_value
+        }
+        self.bb_backend._authn_response(context)
+        assert self.bb_backend.name not in context.state
+        self.assert_expected_attributes()
