diff --git a/src/kaggle/api/kaggle_api_extended.py b/src/kaggle/api/kaggle_api_extended.py index 7122f404..fbe5dc5d 100644 --- a/src/kaggle/api/kaggle_api_extended.py +++ b/src/kaggle/api/kaggle_api_extended.py @@ -1511,6 +1511,11 @@ def authenticate(self) -> None: print_auth_help() exit(1) + _ACCESS_TOKEN_PREFIX = "KGAT_" + _REFRESH_TOKEN_PREFIX = "KGRT_" + _ACCESS_TOKEN_RE = re.compile(r"^KGAT_[0-9a-fA-F]{32}$") + _LEGACY_API_KEY_RE = re.compile(r"^[0-9a-fA-F]{32}$") + def _authenticate_with_legacy_apikey(self) -> bool: """Authenticate the user with the Kaggle API using legacy API key. @@ -1521,6 +1526,36 @@ def _authenticate_with_legacy_apikey(self) -> bool: Returns: bool: True if legacy credentials are available. """ + key = self.config_values.get(self.CONFIG_NAME_KEY) + if isinstance(key, str) and key.strip().upper().startswith( + (self._ACCESS_TOKEN_PREFIX, self._REFRESH_TOKEN_PREFIX) + ): + source = "KAGGLE_KEY environment variable" if os.environ.get("KAGGLE_KEY") == key else self.config + print( + f"Error: Found an API token ({key.strip()[:5]}...) in {source}.\n" + f"API tokens starting with KGAT_ cannot be used as a legacy API key in KAGGLE_KEY or {self.config}.\n" + "To authenticate with a KGAT_ token, save the raw token to ~/.kaggle/access_token " + "or set the KAGGLE_API_TOKEN environment variable." + ) + self.config_values.pop(self.CONFIG_NAME_KEY, None) + return False + + for token_field in (self.CONFIG_NAME_TOKEN, "api_token", "apiToken"): + token_val = self.config_values.get(token_field) + if ( + isinstance(token_val, str) + and token_val.strip().upper().startswith((self._ACCESS_TOKEN_PREFIX, self._REFRESH_TOKEN_PREFIX)) + and os.environ.get("KAGGLE_API_TOKEN") != token_val + and os.environ.get("KAGGLE_TOKEN") != token_val + ): + print( + f"Error: Found an API token ({token_val.strip()[:5]}...) in {self.config}.\n" + f'{self.config} is for legacy JSON credentials ({{"username": "...", "key": "..."}}).\n' + "To authenticate with a KGAT_ token, save the raw token to ~/.kaggle/access_token " + "or set the KAGGLE_API_TOKEN environment variable." + ) + return False + if self.CONFIG_NAME_USER in self.config_values and self.CONFIG_NAME_KEY in self.config_values: self.config_values[self.CONFIG_NAME_AUTH_METHOD] = str(AuthMethod.LEGACY_API_KEY) self.logger.debug(f"Authenticated with legacy api key in: {self.config}") @@ -1562,11 +1597,74 @@ def _invoked_as_cli() -> bool: return True return prog.endswith("/kaggle/__main__.py") + def _validate_access_token_format(self, access_token: str, source: Optional[str]) -> bool: + """Validate that an access token does not have a legacy or malformed format.""" + source_label = "~/.kaggle/access_token" if source == "access_token" else (source or "access_token") + token = access_token.strip() + + if token.startswith("{"): + if self._ACCESS_TOKEN_PREFIX in token.upper(): + print( + f"Error: Found JSON content in {source_label}, which expects a plain-text API token.\n" + "Provide only the raw KGAT_... token string without JSON formatting." + ) + else: + print( + f"Error: Found legacy JSON credentials in {source_label}.\n" + 'Legacy credentials ({"username": "...", "key": "..."}) belong in ~/.kaggle/kaggle.json ' + "(or KAGGLE_USERNAME and KAGGLE_KEY).\n" + f"{source_label} expects a plain-text API token starting with KGAT_." + ) + return False + + if self._LEGACY_API_KEY_RE.match(token): + print( + f"Error: Found a legacy API key (32-character hex string without a KGAT_ prefix) in {source_label}.\n" + 'Legacy API keys must be paired with a username in ~/.kaggle/kaggle.json ({"username": "...", "key": "..."}) ' + "or via KAGGLE_USERNAME and KAGGLE_KEY.\n" + f"{source_label} expects an API access token starting with KGAT_." + ) + return False + + if token.upper().startswith(self._REFRESH_TOKEN_PREFIX): + print( + f"Error: Found an OAuth refresh token (KGRT_...) in {source_label}.\n" + f"{source_label} expects an API access token starting with KGAT_." + ) + return False + + if token.startswith(('"', "'")) or ("=" in token and self._ACCESS_TOKEN_PREFIX in token.upper()): + print( + f"Error: Malformed API access token in {source_label}.\n" + f"{source_label} must contain only the raw KGAT_... token string (without quotes or KEY=VALUE syntax)." + ) + return False + + if token.upper().startswith("KGAT") and not self._ACCESS_TOKEN_RE.match(token): + print( + f"Error: Malformed API access token in {source_label}.\n" + 'A Kaggle API token must start with "KGAT_" followed by 32 hexadecimal characters. ' + "Please verify that your token was not truncated or modified." + ) + return False + + if any(c.isspace() for c in token): + print( + f"Error: Malformed API access token in {source_label}.\n" + f"{source_label} must contain a single valid API token (e.g. KGAT_<32-hex-chars>)." + ) + return False + + return True + def _authenticate_with_access_token(self) -> bool: access_token, source = get_access_token_from_env() if not access_token: return False + if not self._validate_access_token_format(access_token, source): + return False + username = self._introspect_token(access_token) if not username: self.logger.debug(f'Ignoring invalid/expired access token in "{source}".') @@ -1614,7 +1712,7 @@ def _introspect_token(self, access_token: str) -> Optional[str]: return None return response.username except HTTPError as e: - if e.response.status_code in (400, 403, 404): + if e.response is not None and e.response.status_code in (400, 401, 403, 404): self.logger.debug("Access token invalid: %s", e) return None raise @@ -1701,8 +1799,31 @@ def read_config_file(self, config_data: Optional[Dict[str, str]] = None, quiet: ) with open(self.config) as f: - config_data = json.load(f) or {} - except: + raw_content = f.read().strip() + + if raw_content: + try: + loaded = json.loads(raw_content) + except json.JSONDecodeError: + loaded = None + + if isinstance(loaded, dict): + config_data = loaded + elif self._ACCESS_TOKEN_PREFIX in raw_content.upper(): + print( + f"Error: Found an API access token (KGAT_...) in {self.config}.\n" + f'{self.config} is for legacy JSON credentials ({{"username": "...", "key": "..."}}).\n' + "To authenticate with a KGAT_ token, save the raw token to ~/.kaggle/access_token " + "or set the KAGGLE_API_TOKEN environment variable." + ) + else: + print( + f"Error: {self.config} is not a valid JSON object.\n" + 'Expected legacy credentials in the format {"username": "", "key": ""}.\n' + "If you are using an API access token (KGAT_...), save it to ~/.kaggle/access_token " + "or set the KAGGLE_API_TOKEN environment variable." + ) + except Exception: pass else: diff --git a/tests/unit/test_authenticate.py b/tests/unit/test_authenticate.py index 3f160f3e..5a560af9 100644 --- a/tests/unit/test_authenticate.py +++ b/tests/unit/test_authenticate.py @@ -2,9 +2,15 @@ # python -m unittest tests.test_authenticate +from contextlib import redirect_stdout +import io import os +import tempfile import unittest -from unittest.mock import patch +from unittest.mock import MagicMock, patch + +from requests.exceptions import HTTPError +from requests.models import Response class TestAuthenticate(unittest.TestCase): @@ -125,6 +131,263 @@ def test_legacy_apikey_does_not_handle_anonymous_fallback(self): with patch("sys.argv", ["kaggle", "datasets", "download", "some/dataset"]): self.assertFalse(api._authenticate_with_legacy_apikey()) + def test_kgat_token_in_kaggle_key_env_var_is_rejected(self): + api = KaggleApi() + token = "KGAT_0123456789abcdef0123456789abcdef" + with patch.dict(os.environ, {"KAGGLE_USERNAME": "dinosaur", "KAGGLE_KEY": token}, clear=False): + api.config_values = {"username": "dinosaur", "key": token} + captured = io.StringIO() + with redirect_stdout(captured): + result = api._authenticate_with_legacy_apikey() + + self.assertFalse(result) + self.assertNotIn("key", api.config_values) + output = captured.getvalue() + self.assertIn("KAGGLE_KEY", output) + self.assertIn("KAGGLE_API_TOKEN", output) + self.assertIn("access_token", output) + + def test_kgat_token_in_kaggle_json_key_is_rejected(self): + api = KaggleApi() + token = "KGAT_0123456789abcdef0123456789abcdef" + with tempfile.TemporaryDirectory() as tmpdir: + api.config = os.path.join(tmpdir, "kaggle.json") + with open(api.config, "w", encoding="utf-8") as f: + f.write(f'{{"username": "dinosaur", "key": "{token}"}}') + os.chmod(api.config, 0o600) + + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("KAGGLE_KEY", None) + api._load_config() + captured = io.StringIO() + with redirect_stdout(captured): + result = api._authenticate_with_legacy_apikey() + + self.assertFalse(result) + self.assertNotIn("key", api.config_values) + output = captured.getvalue() + self.assertIn("kaggle.json", output) + self.assertIn("KAGGLE_API_TOKEN", output) + self.assertIn("access_token", output) + + def test_raw_kgat_token_in_kaggle_json_file_prints_error(self): + api = KaggleApi() + token = "KGAT_0123456789abcdef0123456789abcdef" + with tempfile.TemporaryDirectory() as tmpdir: + api.config = os.path.join(tmpdir, "kaggle.json") + with open(api.config, "w", encoding="utf-8") as f: + f.write(token) + os.chmod(api.config, 0o600) + + captured = io.StringIO() + with redirect_stdout(captured): + config_data = api.read_config_file(quiet=True) + + self.assertEqual(config_data, {}) + output = captured.getvalue() + self.assertIn("KGAT_", output) + self.assertIn("kaggle.json", output) + self.assertIn("access_token", output) + + def test_malformed_json_in_kaggle_json_file_prints_error(self): + api = KaggleApi() + with tempfile.TemporaryDirectory() as tmpdir: + api.config = os.path.join(tmpdir, "kaggle.json") + with open(api.config, "w", encoding="utf-8") as f: + f.write("not-valid-json") + os.chmod(api.config, 0o600) + + captured = io.StringIO() + with redirect_stdout(captured): + config_data = api.read_config_file(quiet=True) + + self.assertEqual(config_data, {}) + output = captured.getvalue() + self.assertIn("kaggle.json", output) + self.assertIn("JSON", output) + + @patch.object(KaggleApi, "_introspect_token") + @patch("kaggle.api.kaggle_api_extended.get_access_token_from_env") + def test_legacy_json_in_access_token_is_rejected(self, mock_get_token, mock_introspect): + mock_get_token.return_value = ( + '{"username": "dinosaur", "key": "0123456789abcdef0123456789abcdef"}', + "access_token", + ) + api = KaggleApi() + captured = io.StringIO() + with redirect_stdout(captured): + result = api._authenticate_with_access_token() + + self.assertFalse(result) + mock_introspect.assert_not_called() + output = captured.getvalue() + self.assertIn("kaggle.json", output) + self.assertIn("access_token", output) + + @patch.object(KaggleApi, "_introspect_token") + @patch("kaggle.api.kaggle_api_extended.get_access_token_from_env") + def test_legacy_hex_key_in_access_token_is_rejected(self, mock_get_token, mock_introspect): + mock_get_token.return_value = ("0123456789abcdef0123456789abcdef", "KAGGLE_API_TOKEN") + api = KaggleApi() + captured = io.StringIO() + with redirect_stdout(captured): + result = api._authenticate_with_access_token() + + self.assertFalse(result) + mock_introspect.assert_not_called() + output = captured.getvalue() + self.assertIn("legacy API key", output) + self.assertIn("KAGGLE_API_TOKEN", output) + self.assertIn("KGAT_", output) + + @patch.object(KaggleApi, "_introspect_token") + @patch("kaggle.api.kaggle_api_extended.get_access_token_from_env") + def test_malformed_kgat_token_in_access_token_is_rejected(self, mock_get_token, mock_introspect): + mock_get_token.return_value = ("KGAT_too_short", "KAGGLE_API_TOKEN") + api = KaggleApi() + captured = io.StringIO() + with redirect_stdout(captured): + result = api._authenticate_with_access_token() + + self.assertFalse(result) + mock_introspect.assert_not_called() + output = captured.getvalue() + self.assertIn("Malformed", output) + self.assertIn("KGAT_", output) + self.assertIn("32 hexadecimal", output) + + @patch.object(KaggleApi, "_introspect_token", return_value="dinosaur") + @patch("kaggle.api.kaggle_api_extended.get_access_token_from_env") + def test_valid_kgat_token_in_access_token_succeeds(self, mock_get_token, mock_introspect): + token = "KGAT_0123456789abcdef0123456789abcdef" + mock_get_token.return_value = (token, "KAGGLE_API_TOKEN") + api = KaggleApi() + api.config_values = {} + result = api._authenticate_with_access_token() + + self.assertTrue(result) + mock_introspect.assert_called_once_with(token) + self.assertEqual(api.config_values["token"], token) + self.assertEqual(api.config_values["username"], "dinosaur") + + @patch.object(KaggleApi, "build_kaggle_client") + def test_introspect_token_handles_401_without_raising(self, mock_build_client): + response = Response() + response.status_code = 401 + mock_kaggle = MagicMock() + mock_kaggle.security.oauth_client.introspect_token.side_effect = HTTPError(response=response) + mock_build_client.return_value.__enter__.return_value = mock_kaggle + mock_build_client.return_value.__exit__.return_value = False + + api = KaggleApi() + self.assertIsNone(api._introspect_token("KGAT_0123456789abcdef0123456789abcdef")) + + def test_kgat_token_in_kaggle_json_token_field_is_rejected(self): + api = KaggleApi() + token = "KGAT_0123456789abcdef0123456789abcdef" + with tempfile.TemporaryDirectory() as tmpdir: + api.config = os.path.join(tmpdir, "kaggle.json") + with open(api.config, "w", encoding="utf-8") as f: + f.write(f'{{"token": "{token}"}}') + os.chmod(api.config, 0o600) + + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("KAGGLE_KEY", None) + os.environ.pop("KAGGLE_API_TOKEN", None) + api._load_config() + captured = io.StringIO() + with redirect_stdout(captured): + result = api._authenticate_with_legacy_apikey() + + self.assertFalse(result) + output = captured.getvalue() + self.assertIn("kaggle.json", output) + self.assertIn("KAGGLE_API_TOKEN", output) + + @patch.object(KaggleApi, "_introspect_token") + @patch("kaggle.api.kaggle_api_extended.get_access_token_from_env") + def test_json_wrapped_kgat_token_in_access_token_is_rejected(self, mock_get_token, mock_introspect): + mock_get_token.return_value = ( + '{"token": "KGAT_0123456789abcdef0123456789abcdef"}', + "access_token", + ) + api = KaggleApi() + captured = io.StringIO() + with redirect_stdout(captured): + result = api._authenticate_with_access_token() + + self.assertFalse(result) + mock_introspect.assert_not_called() + output = captured.getvalue() + self.assertIn("JSON", output) + self.assertIn("KGAT_", output) + + @patch.object(KaggleApi, "_introspect_token") + @patch("kaggle.api.kaggle_api_extended.get_access_token_from_env") + def test_refresh_token_in_access_token_is_rejected(self, mock_get_token, mock_introspect): + mock_get_token.return_value = ("KGRT_0123456789abcdef0123456789abcdef", "KAGGLE_API_TOKEN") + api = KaggleApi() + captured = io.StringIO() + with redirect_stdout(captured): + result = api._authenticate_with_access_token() + + self.assertFalse(result) + mock_introspect.assert_not_called() + output = captured.getvalue() + self.assertIn("refresh token", output) + self.assertIn("KGAT_", output) + + @patch.object(KaggleApi, "_introspect_token") + @patch("kaggle.api.kaggle_api_extended.get_access_token_from_env") + def test_env_assignment_syntax_in_access_token_is_rejected(self, mock_get_token, mock_introspect): + mock_get_token.return_value = ( + "KAGGLE_API_TOKEN=KGAT_0123456789abcdef0123456789abcdef", + "access_token", + ) + api = KaggleApi() + captured = io.StringIO() + with redirect_stdout(captured): + result = api._authenticate_with_access_token() + + self.assertFalse(result) + mock_introspect.assert_not_called() + output = captured.getvalue() + self.assertIn("KEY=VALUE", output) + + @patch.object(KaggleApi, "_introspect_token", return_value="dinosaur") + @patch("kaggle.api.kaggle_api_extended.get_access_token_from_env") + def test_oauth_bearer_token_in_access_token_succeeds(self, mock_get_token, mock_introspect): + oauth_token = "CfDJ8N2_fake_encrypted_oauth_access_token_payload" + mock_get_token.return_value = (oauth_token, "KAGGLE_API_TOKEN") + api = KaggleApi() + api.config_values = {} + result = api._authenticate_with_access_token() + + self.assertTrue(result) + mock_introspect.assert_called_once_with(oauth_token) + self.assertEqual(api.config_values["token"], oauth_token) + self.assertEqual(api.config_values["username"], "dinosaur") + + def test_valid_kaggle_json_file_succeeds(self): + api = KaggleApi() + key = "0123456789abcdef0123456789abcdef" + with tempfile.TemporaryDirectory() as tmpdir: + api.config = os.path.join(tmpdir, "kaggle.json") + with open(api.config, "w", encoding="utf-8") as f: + f.write(f'{{"username": "dinosaur", "key": "{key}"}}') + os.chmod(api.config, 0o600) + + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("KAGGLE_USERNAME", None) + os.environ.pop("KAGGLE_KEY", None) + os.environ.pop("KAGGLE_API_TOKEN", None) + api._load_config() + result = api._authenticate_with_legacy_apikey() + + self.assertTrue(result) + self.assertEqual(api.config_values["username"], "dinosaur") + self.assertEqual(api.config_values["key"], key) + if __name__ == "__main__": unittest.main()