Intermediate changes

commit_hash:39273c986cf85d42ec97d1388aaaf324a02a04bc

6 files changed, 406 insertions, 0 deletions

diff --git a/contrib/python/oauthlib/tests/oauth2/rfc8628/endpoints/__init__.py b/contrib/python/oauthlib/tests/oauth2/rfc8628/endpoints/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
--- /dev/null
+++ b/contrib/python/oauthlib/tests/oauth2/rfc8628/endpoints/__init__.py
diff --git a/contrib/python/oauthlib/tests/oauth2/rfc8628/endpoints/test_device_application_server.py b/contrib/python/oauthlib/tests/oauth2/rfc8628/endpoints/test_device_application_server.py
new file mode 100644
index 00000000000..f0436754a5e
--- /dev/null
+++ b/contrib/python/oauthlib/tests/oauth2/rfc8628/endpoints/test_device_application_server.py
@@ -0,0 +1,26 @@
+import json
+from unittest import TestCase, mock
+
+from oauthlib.common import Request, urlencode
+from oauthlib.oauth2.rfc6749 import errors
+from oauthlib.oauth2.rfc8628.endpoints.pre_configured import DeviceApplicationServer
+from oauthlib.oauth2.rfc8628.request_validator import RequestValidator
+
+
+def test_server_set_up_device_endpoint_instance_attributes_correctly():
+    """
+    Simple test that just instantiates DeviceApplicationServer
+    and asserts the important attributes are present
+    """
+    validator = mock.MagicMock(spec=RequestValidator)
+    validator.get_default_redirect_uri.return_value = None
+    validator.get_code_challenge.return_value = None
+
+    verification_uri = "test.com/device"
+    verification_uri_complete = "test.com/device?user_code=123"
+    device = DeviceApplicationServer(validator,  verification_uri=verification_uri, verification_uri_complete=verification_uri_complete)
+    device_vars = vars(device)
+    assert device_vars["_verification_uri_complete"] == "test.com/device?user_code=123"
+    assert device_vars["_verification_uri"] == "test.com/device"
+    assert device_vars["_expires_in"] == 1800
+    assert device_vars["_interval"] == 5
diff --git a/contrib/python/oauthlib/tests/oauth2/rfc8628/endpoints/test_error_responses.py b/contrib/python/oauthlib/tests/oauth2/rfc8628/endpoints/test_error_responses.py
new file mode 100644
index 00000000000..c799cc81af2
--- /dev/null
+++ b/contrib/python/oauthlib/tests/oauth2/rfc8628/endpoints/test_error_responses.py
@@ -0,0 +1,95 @@
+import json
+from unittest import TestCase, mock
+
+from oauthlib.common import Request, urlencode
+from oauthlib.oauth2.rfc6749 import errors
+from oauthlib.oauth2.rfc8628.endpoints.pre_configured import DeviceApplicationServer
+from oauthlib.oauth2.rfc8628.request_validator import RequestValidator
+
+
+class ErrorResponseTest(TestCase):
+    def set_client(self, request):
+        request.client = mock.MagicMock()
+        request.client.client_id = "mocked"
+        return True
+
+    def build_request(self, uri="https://example.com/device_authorize", client_id="foo"):
+        body = ""
+        if client_id:
+            body = f"client_id={client_id}"
+        return Request(
+            uri,
+            http_method="POST",
+            body=body,
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
+        )
+
+    def assert_request_raises(self, error, request):
+        """Test that the request fails similarly on the validation and response endpoint."""
+        self.assertRaises(
+            error,
+            self.device.validate_device_authorization_request,
+            request,
+        )
+        self.assertRaises(
+            error,
+            self.device.create_device_authorization_response,
+            uri=request.uri,
+            http_method=request.http_method,
+            body=request.body,
+            headers=request.headers,
+        )
+
+    def setUp(self):
+        self.validator = mock.MagicMock(spec=RequestValidator)
+        self.validator.get_default_redirect_uri.return_value = None
+        self.validator.get_code_challenge.return_value = None
+        self.device = DeviceApplicationServer(self.validator, "https://example.com/verify")
+
+    def test_missing_client_id(self):
+        # Device code grant
+        request = self.build_request(client_id=None)
+        self.assert_request_raises(errors.MissingClientIdError, request)
+
+    def test_empty_client_id(self):
+        # Device code grant
+        self.assertRaises(
+            errors.MissingClientIdError,
+            self.device.create_device_authorization_response,
+            "https://i.l/",
+            "POST",
+            "client_id=",
+            {"Content-Type": "application/x-www-form-urlencoded"},
+        )
+
+    def test_invalid_client_id(self):
+        request = self.build_request(client_id="foo")
+        # Device code grant
+        self.validator.validate_client_id.return_value = False
+        self.assert_request_raises(errors.InvalidClientIdError, request)
+
+    def test_duplicate_client_id(self):
+        request = self.build_request()
+        request.body = "client_id=foo&client_id=bar"
+        # Device code grant
+        self.validator.validate_client_id.return_value = False
+        self.assert_request_raises(errors.InvalidRequestFatalError, request)
+
+    def test_unauthenticated_confidential_client(self):
+        self.validator.client_authentication_required.return_value = True
+        self.validator.authenticate_client.return_value = False
+        request = self.build_request()
+        self.assert_request_raises(errors.InvalidClientError, request)
+
+    def test_unauthenticated_public_client(self):
+        self.validator.client_authentication_required.return_value = False
+        self.validator.authenticate_client_id.return_value = False
+        request = self.build_request()
+        self.assert_request_raises(errors.InvalidClientError, request)
+
+    def test_duplicate_scope_parameter(self):
+        request = self.build_request()
+        request.body = "client_id=foo&scope=foo&scope=bar"
+        # Device code grant
+        self.validator.validate_client_id.return_value = False
+        self.assert_request_raises(errors.InvalidRequestFatalError, request)
diff --git a/contrib/python/oauthlib/tests/oauth2/rfc8628/grant_types/__init__.py b/contrib/python/oauthlib/tests/oauth2/rfc8628/grant_types/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
--- /dev/null
+++ b/contrib/python/oauthlib/tests/oauth2/rfc8628/grant_types/__init__.py
diff --git a/contrib/python/oauthlib/tests/oauth2/rfc8628/grant_types/test_device_code.py b/contrib/python/oauthlib/tests/oauth2/rfc8628/grant_types/test_device_code.py
new file mode 100644
index 00000000000..da0592f7426
--- /dev/null
+++ b/contrib/python/oauthlib/tests/oauth2/rfc8628/grant_types/test_device_code.py
@@ -0,0 +1,172 @@
+import json
+from unittest import mock
+import pytest
+
+from oauthlib import common
+
+from oauthlib.oauth2.rfc8628.grant_types import DeviceCodeGrant
+from oauthlib.oauth2.rfc6749.tokens import BearerToken
+
+def create_request(body: str = "") -> common.Request:
+    request = common.Request("http://a.b/path", body=body or None)
+    request.scopes = ("hello", "world")
+    request.expires_in = 1800
+    request.client = "batman"
+    request.client_id = "abcdef"
+    request.code = "1234"
+    request.response_type = "code"
+    request.grant_type = "urn:ietf:params:oauth:grant-type:device_code"
+    request.redirect_uri = "https://a.b/"
+    return request
+
+
+def create_device_code_grant(mock_validator: mock.MagicMock) -> DeviceCodeGrant:
+    return DeviceCodeGrant(request_validator=mock_validator)
+
+
+def test_custom_auth_validators_unsupported():
+    custom_validator = mock.Mock()
+    validator = mock.MagicMock()
+
+    expected = (
+        "DeviceCodeGrant does not "
+        "support authorization validators. Use token validators instead."
+    )
+    with pytest.raises(ValueError, match=expected):
+        DeviceCodeGrant(validator, pre_auth=[custom_validator])
+
+    with pytest.raises(ValueError, match=expected):
+        DeviceCodeGrant(validator, post_auth=[custom_validator])
+
+    expected = "'tuple' object has no attribute 'append'"
+    auth = DeviceCodeGrant(validator)
+    with pytest.raises(AttributeError, match=expected):
+        auth.custom_validators.pre_auth.append(custom_validator)
+
+
+def test_custom_pre_and_post_token_validators():
+    client = mock.MagicMock()
+
+    validator = mock.MagicMock()
+    pre_token_validator = mock.Mock()
+    post_token_validator = mock.Mock()
+
+    request: common.Request = create_request()
+    request.client = client
+
+    auth = DeviceCodeGrant(validator)
+
+    auth.custom_validators.pre_token.append(pre_token_validator)
+    auth.custom_validators.post_token.append(post_token_validator)
+
+    bearer = BearerToken(validator)
+    auth.create_token_response(request, bearer)
+
+    pre_token_validator.assert_called()
+    post_token_validator.assert_called()
+
+
+def test_create_token_response():
+    validator = mock.MagicMock()
+    request: common.Request = create_request()
+    request.client = mock.Mock()
+
+    auth = DeviceCodeGrant(validator)
+
+    bearer = BearerToken(validator)
+
+    headers, body, status_code = auth.create_token_response(request, bearer)
+    token = json.loads(body)
+
+    assert headers == {
+        "Content-Type": "application/json",
+        "Cache-Control": "no-store",
+        "Pragma": "no-cache",
+    }
+
+    # when a custom token generator callable isn't used
+    # the random generator is used as default for the access token
+    assert token == {
+        "access_token": mock.ANY,
+        "expires_in": 3600,
+        "token_type": "Bearer",
+        "scope": "hello world",
+        "refresh_token": mock.ANY,
+    }
+
+    assert status_code == 200
+
+    validator.save_token.assert_called_once()
+
+
+def test_invalid_client_error():
+    validator = mock.MagicMock()
+    request: common.Request = create_request()
+    request.client = mock.Mock()
+
+    auth = DeviceCodeGrant(validator)
+    bearer = BearerToken(validator)
+
+    validator.authenticate_client.return_value = False
+
+    headers, body, status_code = auth.create_token_response(request, bearer)
+    body = json.loads(body)
+
+    assert headers == {
+        "Content-Type": "application/json",
+        "Cache-Control": "no-store",
+        "Pragma": "no-cache",
+        "WWW-Authenticate": 'Bearer error="invalid_client"',
+    }
+    assert body == {"error": "invalid_client"}
+    assert status_code == 401
+
+    validator.save_token.assert_not_called()
+
+
+def test_invalid_grant_type_error():
+    validator = mock.MagicMock()
+    request: common.Request = create_request()
+    request.client = mock.Mock()
+
+    request.grant_type = "not_device_code"
+
+    auth = DeviceCodeGrant(validator)
+    bearer = BearerToken(validator)
+
+    headers, body, status_code = auth.create_token_response(request, bearer)
+    body = json.loads(body)
+
+    assert headers == {
+        "Content-Type": "application/json",
+        "Cache-Control": "no-store",
+        "Pragma": "no-cache",
+    }
+    assert body == {"error": "unsupported_grant_type"}
+    assert status_code == 400
+
+    validator.save_token.assert_not_called()
+
+
+def test_duplicate_params_error():
+    validator = mock.MagicMock()
+    request: common.Request = create_request(
+        "client_id=123&scope=openid&scope=openid"
+    )
+    request.client = mock.Mock()
+
+    auth = DeviceCodeGrant(validator)
+    bearer = BearerToken(validator)
+
+    headers, body, status_code = auth.create_token_response(request, bearer)
+    body = json.loads(body)
+
+    assert headers == {
+        "Content-Type": "application/json",
+        "Cache-Control": "no-store",
+        "Pragma": "no-cache",
+    }
+    assert body == {"error": "invalid_request", "error_description": "Duplicate scope parameter."}
+    assert status_code == 400
+
+    validator.save_token.assert_not_called()
diff --git a/contrib/python/oauthlib/tests/oauth2/rfc8628/test_server.py b/contrib/python/oauthlib/tests/oauth2/rfc8628/test_server.py
new file mode 100644
index 00000000000..52025032849
--- /dev/null
+++ b/contrib/python/oauthlib/tests/oauth2/rfc8628/test_server.py
@@ -0,0 +1,113 @@
+import json
+from unittest import mock
+
+from oauthlib.oauth2.rfc8628.endpoints import DeviceAuthorizationEndpoint
+from oauthlib.oauth2.rfc8628.request_validator import RequestValidator
+
+from tests.unittest import TestCase
+
+
+class DeviceAuthorizationEndpointTest(TestCase):
+    def _configure_endpoint(
+        self, interval=None, verification_uri_complete=None, user_code_generator=None
+    ):
+        self.endpoint = DeviceAuthorizationEndpoint(
+            request_validator=mock.MagicMock(spec=RequestValidator),
+            verification_uri=self.verification_uri,
+            interval=interval,
+            verification_uri_complete=verification_uri_complete,
+            user_code_generator=user_code_generator,
+        )
+
+    def setUp(self):
+        self.request_validator = mock.MagicMock(spec=RequestValidator)
+        self.verification_uri = "http://i.b/l/verify"
+        self.uri = "http://i.b/l"
+        self.http_method = "POST"
+        self.body = "client_id=abc"
+        self.headers = {"Content-Type": "application/x-www-form-urlencoded"}
+        self._configure_endpoint()
+
+    def response_payload(self):
+        return self.uri, self.http_method, self.body, self.headers
+
+    @mock.patch("oauthlib.oauth2.rfc8628.endpoints.device_authorization.generate_token")
+    def test_device_authorization_grant(self, generate_token):
+        generate_token.side_effect = ["abc", "def"]
+        _, body, status_code = self.endpoint.create_device_authorization_response(
+            *self.response_payload()
+        )
+        expected_payload = {
+            "verification_uri": "http://i.b/l/verify",
+            "user_code": "abc",
+            "device_code": "def",
+            "expires_in": 1800,
+        }
+        self.assertEqual(200, status_code)
+        self.assertEqual(body, expected_payload)
+
+    @mock.patch(
+        "oauthlib.oauth2.rfc8628.endpoints.device_authorization.generate_token",
+        lambda: "abc",
+    )
+    def test_device_authorization_grant_interval(self):
+        self._configure_endpoint(interval=5)
+        _, body, _ = self.endpoint.create_device_authorization_response(*self.response_payload())
+        self.assertEqual(5, body["interval"])
+
+    @mock.patch(
+        "oauthlib.oauth2.rfc8628.endpoints.device_authorization.generate_token",
+        lambda: "abc",
+    )
+    def test_device_authorization_grant_interval_with_zero(self):
+        self._configure_endpoint(interval=0)
+        _, body, _ = self.endpoint.create_device_authorization_response(*self.response_payload())
+        self.assertEqual(0, body["interval"])
+
+    @mock.patch(
+        "oauthlib.oauth2.rfc8628.endpoints.device_authorization.generate_token",
+        lambda: "abc",
+    )
+    def test_device_authorization_grant_verify_url_complete_string(self):
+        self._configure_endpoint(verification_uri_complete="http://i.l/v?user_code={user_code}")
+        _, body, _ = self.endpoint.create_device_authorization_response(*self.response_payload())
+        self.assertEqual(
+            "http://i.l/v?user_code=abc",
+            body["verification_uri_complete"],
+        )
+
+    @mock.patch(
+        "oauthlib.oauth2.rfc8628.endpoints.device_authorization.generate_token",
+        lambda: "abc",
+    )
+    def test_device_authorization_grant_verify_url_complete_callable(self):
+        self._configure_endpoint(verification_uri_complete=lambda u: f"http://i.l/v?user_code={u}")
+        _, body, _ = self.endpoint.create_device_authorization_response(*self.response_payload())
+        self.assertEqual(
+            "http://i.l/v?user_code=abc",
+            body["verification_uri_complete"],
+        )
+
+    @mock.patch(
+        "oauthlib.oauth2.rfc8628.endpoints.device_authorization.generate_token",
+        lambda: "abc",
+    )
+    def test_device_authorization_grant_user_gode_generator(self):
+        def user_code():
+            """
+            A friendly user code the device can display and the user
+            can type in. It's up to the device how
+            this code should be displayed. e.g 123-456
+            """
+            return "123456"
+
+        self._configure_endpoint(
+            verification_uri_complete=lambda u: f"http://i.l/v?user_code={u}",
+            user_code_generator=user_code,
+        )
+
+        _, body, _ = self.endpoint.create_device_authorization_response(*self.response_payload())
+        self.assertEqual(
+            "http://i.l/v?user_code=123456",
+            body["verification_uri_complete"],
+        )