reformat and pin pyjwt version

Author: Florian Sandel

certbot_dns_stackit/stackit.py

@@ -161,12 +161,12 @@ def _get_zone_id(self, domain: str) -> str:
         :param domain: The domain (zone dnsName) for which the zone ID is needed.
         :return: The ID of the zone.
         """
-        parts = domain.split('.')
+        parts = domain.split(".")
 
         # we are searching for the best matching zone. We can do that by iterating over the parts of the domain
         # from left to right.
         for i in range(len(parts)):
-            subdomain = '.'.join(parts[i:])
+            subdomain = ".".join(parts[i:])
             res = requests.get(
                 f"{self.base_url}/v1/projects/{self.project_id}/zones?dnsName[eq]={subdomain}&active[eq]=true",
                 headers=self.headers,
@@ -340,8 +340,8 @@ def _load_service_file(self, file_path: str) -> Optional[ServiceFileCredentials]
         :return: Service file credentials if the file is found and valid, None otherwise.
         """
         try:
-            with open(file_path, 'r') as file:
-                return json.load(file)['credentials']
+            with open(file_path, "r") as file:
+                return json.load(file)["credentials"]
         except FileNotFoundError:
             logging.error(f"File not found: {file_path}")
             return None
@@ -354,15 +354,17 @@ def _generate_jwt(self, credentials: ServiceFileCredentials) -> str:
         :return: A JWT token as a string.
         """
         payload = {
-            "iss": credentials['iss'],
-            "sub": credentials['sub'],
-            "aud": credentials['aud'],
+            "iss": credentials["iss"],
+            "sub": credentials["sub"],
+            "aud": credentials["aud"],
             "exp": int(time.time()) + 900,
             "iat": int(time.time()),
-            "jti": str(uuid.uuid4())
+            "jti": str(uuid.uuid4()),
         }
-        headers = {'kid': credentials['kid']}
-        return jwt.encode(payload, credentials['privateKey'], algorithm='RS512', headers=headers)
+        headers = {"kid": credentials["kid"]}
+        return jwt.encode(
+            payload, credentials["privateKey"], algorithm="RS512", headers=headers
+        )
 
     def _request_access_token(self, jwt_token: str) -> str:
         """
@@ -372,13 +374,17 @@ def _request_access_token(self, jwt_token: str) -> str:
         :return: An access token if the request is successful, None otherwise.
         """
         data = {
-            'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
-            'assertion': jwt_token
+            "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
+            "assertion": jwt_token,
         }
         try:
-            response = requests.post('https://service-account.api.stackit.cloud/token', data=data, headers={'Content-Type': 'application/x-www-form-urlencoded'})
+            response = requests.post(
+                "https://service-account.api.stackit.cloud/token",
+                data=data,
+                headers={"Content-Type": "application/x-www-form-urlencoded"},
+            )
             response.raise_for_status()
-            return response.json().get('access_token')
+            return response.json().get("access_token")
         except requests.exceptions.RequestException as e:
             raise errors.PluginError(f"Failed to request access token: {e}")
 

certbot_dns_stackit/test_stackit.py

@@ -220,20 +220,24 @@ def setUp(self):
 
     @patch.object(Authenticator, "conf")
     @patch.object(Authenticator, "_configure_credentials")
-    def test_setup_credentials_with_service_account(self, mock_configure_credentials, mock_conf):
+    def test_setup_credentials_with_service_account(
+        self, mock_configure_credentials, mock_conf
+    ):
         # Simulate `service_account` being set
-        mock_conf.return_value = 'service_account_value'
+        mock_conf.return_value = "service_account_value"
 
         self.authenticator._setup_credentials()
 
         # Assert _configure_credentials was not called
         mock_configure_credentials.assert_not_called()
         # Assert service_account is set correctly
-        self.assertEqual(self.authenticator.service_account, 'service_account_value')
+        self.assertEqual(self.authenticator.service_account, "service_account_value")
 
     @patch.object(Authenticator, "conf")
     @patch.object(Authenticator, "_configure_credentials")
-    def test_setup_credentials_without_service_account(self, mock_configure_credentials, mock_conf):
+    def test_setup_credentials_without_service_account(
+        self, mock_configure_credentials, mock_conf
+    ):
         # Simulate `service_account` not being set
         mock_conf.return_value = None
         mock_creds = Mock()
@@ -283,7 +287,11 @@ def test_cleanup(self, mock_get_client):
             "test_domain", "validation_name_test", "validation_test"
         )
 
-    @patch("builtins.open", new_callable=mock_open, read_data='{"credentials": {"iss": "test_iss", "sub": "test_sub", "aud": "test_aud", "kid": "test_kid", "privateKey": "test_private_key"}}')
+    @patch(
+        "builtins.open",
+        new_callable=mock_open,
+        read_data='{"credentials": {"iss": "test_iss", "sub": "test_sub", "aud": "test_aud", "kid": "test_kid", "privateKey": "test_private_key"}}',
+    )
     @patch("json.load", lambda x: json.loads(x.read()))
     def test_load_service_file(self, mock_load_service_file):
         expected_credentials = {
@@ -307,70 +315,85 @@ def test_load_service_file_not_found(self, mock_log, mock_file):
     @patch("jwt.encode")
     def test_generate_jwt(self, mock_jwt_encode):
         credentials = {
-            'iss': 'issuer',
-            'sub': 'subject',
-            'aud': 'audience',
-            'kid': 'key_id',
-            'privateKey': 'private_key'
+            "iss": "issuer",
+            "sub": "subject",
+            "aud": "audience",
+            "kid": "key_id",
+            "privateKey": "private_key",
         }
         self.authenticator._generate_jwt(credentials)
         mock_jwt_encode.assert_called()
 
     def test_generate_jwt_fail(self):
         credentials = {
-            'iss': 'issuer',
-            'sub': 'subject',
-            'aud': 'audience',
-            'kid': 'key_id',
-            'privateKey': 'not_a_valid_key'
+            "iss": "issuer",
+            "sub": "subject",
+            "aud": "audience",
+            "kid": "key_id",
+            "privateKey": "not_a_valid_key",
         }
         with self.assertRaises(jwt.exceptions.InvalidKeyError):
             token = self.authenticator._generate_jwt(credentials)
             self.assertIsNone(token)
 
-    @patch('requests.post')
+    @patch("requests.post")
     def test_request_access_token_success(self, mock_post):
         mock_response = mock_post.return_value
-        mock_response.raise_for_status = lambda: None  # Mock raise_for_status to do nothing
-        mock_response.json.return_value = {'access_token': 'mocked_access_token'}
+        mock_response.raise_for_status = (
+            lambda: None
+        )  # Mock raise_for_status to do nothing
+        mock_response.json.return_value = {"access_token": "mocked_access_token"}
 
-        result = self.authenticator._request_access_token('jwt_token_example')
+        result = self.authenticator._request_access_token("jwt_token_example")
 
         # Assertions
         mock_post.assert_called_once_with(
-            'https://service-account.api.stackit.cloud/token',
-            data={'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer', 'assertion': 'jwt_token_example'},
-            headers={'Content-Type': 'application/x-www-form-urlencoded'}
+            "https://service-account.api.stackit.cloud/token",
+            data={
+                "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
+                "assertion": "jwt_token_example",
+            },
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
         )
-        self.assertEqual(result, 'mocked_access_token')
+        self.assertEqual(result, "mocked_access_token")
 
-    @patch('requests.post')
+    @patch("requests.post")
     def test_request_access_token_failure_raises_http_error(self, mock_post):
         mock_response = Response()
         mock_response.status_code = 403
         mock_post.return_value = mock_response
         mock_response.raise_for_status = lambda: (_ for _ in ()).throw(HTTPError())
 
         with self.assertRaises(errors.PluginError):
-            self.authenticator._request_access_token('jwt_token_example')
+            self.authenticator._request_access_token("jwt_token_example")
 
         mock_post.assert_called_once()
 
-    @patch("builtins.open", new_callable=mock_open, read_data='{"credentials": {"iss": "test_iss", "sub": "test_sub", "aud": "test_aud", "kid": "test_kid", "privateKey": "test_private_key"}}')
-    @patch.object(Authenticator, '_request_access_token')
-    @patch.object(Authenticator, '_generate_jwt')
-    @patch.object(Authenticator, '_load_service_file')
-    def test_generate_jwt_token_success(self, mock_load_service_file, mock_generate_jwt, mock_request_access_token, mock_open):
-        mock_load_service_file.return_value = {'dummy': 'credentials'}
-        mock_generate_jwt.return_value = 'jwt_token_example'
-        mock_request_access_token.return_value = 'access_token_example'
-
-        result = self.authenticator._generate_jwt_token('path/to/service/file')
-
-        self.assertEqual(result, 'access_token_example')
-        mock_load_service_file.assert_called_once_with('path/to/service/file')
-        mock_generate_jwt.assert_called_once_with({'dummy': 'credentials'})
-        mock_request_access_token.assert_called_once_with('jwt_token_example')
+    @patch(
+        "builtins.open",
+        new_callable=mock_open,
+        read_data='{"credentials": {"iss": "test_iss", "sub": "test_sub", "aud": "test_aud", "kid": "test_kid", "privateKey": "test_private_key"}}',
+    )
+    @patch.object(Authenticator, "_request_access_token")
+    @patch.object(Authenticator, "_generate_jwt")
+    @patch.object(Authenticator, "_load_service_file")
+    def test_generate_jwt_token_success(
+        self,
+        mock_load_service_file,
+        mock_generate_jwt,
+        mock_request_access_token,
+        mock_open,
+    ):
+        mock_load_service_file.return_value = {"dummy": "credentials"}
+        mock_generate_jwt.return_value = "jwt_token_example"
+        mock_request_access_token.return_value = "access_token_example"
+
+        result = self.authenticator._generate_jwt_token("path/to/service/file")
+
+        self.assertEqual(result, "access_token_example")
+        mock_load_service_file.assert_called_once_with("path/to/service/file")
+        mock_generate_jwt.assert_called_once_with({"dummy": "credentials"})
+        mock_request_access_token.assert_called_once_with("jwt_token_example")
 
 
 if __name__ == "__main__":

setup.cfg

@@ -51,7 +51,7 @@ install_requires =
     black
     click==8.1.7
     coverage
-    PyJWT
+    PyJWT==2.9.0
 
 [options.entry_points]
 certbot.plugins =

setup.py

@@ -20,7 +20,7 @@
     "black",
     "click==8.1.7",
     "coverage",
-    "PyJWT"
+    "PyJWT==2.9.0"
 ]
 
 # read the contents of your README file