diff --git a/src/pip/_internal/exceptions.py b/src/pip/_internal/exceptions.py index d6e9095fb55..a22e0c0aa33 100644 --- a/src/pip/_internal/exceptions.py +++ b/src/pip/_internal/exceptions.py @@ -31,6 +31,7 @@ from pip._internal.metadata import BaseDistribution from pip._internal.network.download import _FileDownload + from pip._internal.network.rfc9457 import ProblemDetails from pip._internal.req.req_install import InstallRequirement logger = logging.getLogger(__name__) @@ -335,6 +336,30 @@ def __str__(self) -> str: return str(self.error_msg) +class HTTPProblemDetailsError(NetworkConnectionError): + """HTTP error with RFC 9457 Problem Details.""" + + def __init__( + self, + problem_details: ProblemDetails, + response: Response, + ) -> None: + """ + Initialize HTTPProblemDetailsError with problem details. + + Args: + problem_details: Parsed RFC 9457 problem details + response: The HTTP response object + """ + self.problem_details = problem_details + error_msg = str(problem_details) + + super().__init__( + error_msg=error_msg, + response=response, + ) + + class InvalidWheelFilename(InstallationError): """Invalid wheel filename.""" diff --git a/src/pip/_internal/network/rfc9457.py b/src/pip/_internal/network/rfc9457.py new file mode 100644 index 00000000000..7ec38031cb9 --- /dev/null +++ b/src/pip/_internal/network/rfc9457.py @@ -0,0 +1,91 @@ +"""RFC 9457 - Problem Details for HTTP APIs + +This module provides support for RFC 9457 (Problem Details for HTTP APIs), +a standardized format for describing errors in HTTP APIs. + +Reference: https://www.rfc-editor.org/rfc/rfc9457 +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from typing import Any + +from pip._vendor.requests.models import Response + +logger = logging.getLogger(__name__) + +RFC9457_CONTENT_TYPE = "application/problem+json" + + +@dataclass +class ProblemDetails: + """Represents an RFC 9457 Problem Details object. + + This class encapsulates the core fields defined in RFC 9457: + - status: The HTTP status code + - title: A short, human-readable summary of the problem type + - detail: A human-readable explanation specific to this occurrence + """ + + status: int | None = None + title: str | None = None + detail: str | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ProblemDetails: + return cls( + status=data.get("status"), + title=data.get("title"), + detail=data.get("detail"), + ) + + @classmethod + def from_json(cls, json_str: str) -> ProblemDetails: + data = json.loads(json_str) + if not isinstance(data, dict): + raise ValueError("Problem details JSON must be an object") + return cls.from_dict(data) + + def __str__(self) -> str: + parts = [] + + if self.title: + parts.append(f"{self.title}") + if self.status: + parts.append(f"(Status: {self.status})") + if self.detail: + parts.append(f"\n{self.detail}") + + return " ".join(parts) if parts else "Unknown problem" + + +def is_problem_details_response(response: Response) -> bool: + content_type = response.headers.get("Content-Type", "") + return content_type.startswith(RFC9457_CONTENT_TYPE) + + +def parse_problem_details(response: Response) -> ProblemDetails | None: + if not is_problem_details_response(response): + return None + + try: + body = response.content + if not body: + logger.warning("Problem details response has empty body") + return None + + problem = ProblemDetails.from_json(body.decode("utf-8")) + + if problem.status is None: + problem.status = response.status_code + + logger.debug( + "Parsed problem details: status=%s, title=%s", problem.status, problem.title + ) + return problem + + except (json.JSONDecodeError, ValueError): + return None diff --git a/src/pip/_internal/network/utils.py b/src/pip/_internal/network/utils.py index 74d3111cff0..e6702d94f7c 100644 --- a/src/pip/_internal/network/utils.py +++ b/src/pip/_internal/network/utils.py @@ -2,7 +2,8 @@ from pip._vendor.requests.models import Response -from pip._internal.exceptions import NetworkConnectionError +from pip._internal.exceptions import HTTPProblemDetailsError, NetworkConnectionError +from pip._internal.network.rfc9457 import parse_problem_details # The following comments and HTTP headers were originally added by # Donald Stufft in git commit 22c562429a61bb77172039e480873fb239dd8c03. @@ -29,6 +30,14 @@ def raise_for_status(resp: Response) -> None: + problem_details = parse_problem_details(resp) + if problem_details: + raise HTTPProblemDetailsError( + problem_details=problem_details, + response=resp, + ) + + # Fallback to standard error handling for non-RFC 9457 responses http_error_msg = "" if isinstance(resp.reason, bytes): # We attempt to decode utf-8 first because some servers diff --git a/tests/unit/test_network_rfc9457.py b/tests/unit/test_network_rfc9457.py new file mode 100644 index 00000000000..627d36112b8 --- /dev/null +++ b/tests/unit/test_network_rfc9457.py @@ -0,0 +1,169 @@ +"""Tests for RFC 9457 (Problem Details for HTTP APIs) support.""" + +import json + +import pytest + +from pip._internal.exceptions import HTTPProblemDetailsError, NetworkConnectionError +from pip._internal.network.rfc9457 import ( + ProblemDetails, + is_problem_details_response, + parse_problem_details, +) +from pip._internal.network.utils import raise_for_status + +from tests.lib.requests_mocks import MockResponse + + +class TestProblemDetails: + def test_from_dict(self) -> None: + data = { + "status": 404, + "title": "Not Found", + "detail": "Resource not found", + } + + problem = ProblemDetails.from_dict(data) + assert problem.status == 404 + assert problem.title == "Not Found" + assert problem.detail == "Resource not found" + + def test_from_json(self) -> None: + json_str = json.dumps( + { + "status": 404, + "title": "Not Found", + "detail": "Resource not found", + } + ) + + problem = ProblemDetails.from_json(json_str) + assert problem.status == 404 + assert problem.title == "Not Found" + + def test_string_representation(self) -> None: + """Test string representation of ProblemDetails.""" + problem = ProblemDetails( + status=403, + title="Access Denied", + detail="Your API token does not have permission", + ) + + str_repr = str(problem) + assert "Access Denied" in str_repr + assert "403" in str_repr + assert "API token" in str_repr + + +class TestIsProblemDetailsResponse: + def test_detects_problem_json_content_type(self) -> None: + resp = MockResponse(b"") + resp.headers = {"Content-Type": "application/problem+json"} + + assert is_problem_details_response(resp) is True + + def test_detects_problem_json_with_charset(self) -> None: + resp = MockResponse(b"") + resp.headers = {"Content-Type": "application/problem+json; charset=utf-8"} + + assert is_problem_details_response(resp) is True + + def test_does_not_detect_regular_json(self) -> None: + resp = MockResponse(b"") + resp.headers = {"Content-Type": "application/json"} + + assert is_problem_details_response(resp) is False + + def test_does_not_detect_without_content_type(self) -> None: + resp = MockResponse(b"") + resp.headers = {} + + assert is_problem_details_response(resp) is False + + +class TestParseProblemDetails: + def test_parses_valid_problem_details(self) -> None: + problem_data = { + "status": 404, + "title": "Not Found", + "detail": "The package 'test-package' was not found", + } + resp = MockResponse(json.dumps(problem_data).encode()) + resp.headers = {"Content-Type": "application/problem+json"} + resp.status_code = 404 + + problem = parse_problem_details(resp) + assert problem is not None + assert problem.status == 404 + assert problem.title == "Not Found" + assert problem.detail is not None + assert "test-package" in problem.detail + + def test_returns_none_for_non_problem_details(self) -> None: + resp = MockResponse(b"Error") + resp.headers = {"Content-Type": "text/html"} + + problem = parse_problem_details(resp) + assert problem is None + + def test_handles_malformed_json(self) -> None: + resp = MockResponse(b"not valid json") + resp.headers = {"Content-Type": "application/problem+json"} + + problem = parse_problem_details(resp) + assert problem is None + + +@pytest.mark.parametrize( + "status_code, title, detail", + [ + (404, "Package Not Found", "The requested package does not exist"), + (500, "Internal Server Error", "An unexpected error occurred"), + (403, "Forbidden", "Access denied to this resource"), + ], +) +class TestRaiseForStatusWithProblemDetails: + def test_raises_http_problem_details_error( + self, status_code: int, title: str, detail: str + ) -> None: + problem_data = { + "status": status_code, + "title": title, + "detail": detail, + } + resp = MockResponse(json.dumps(problem_data).encode()) + resp.status_code = status_code + resp.headers = {"Content-Type": "application/problem+json"} + resp.url = "https://pypi.org/simple/some-package/" + + with pytest.raises(HTTPProblemDetailsError) as excinfo: + raise_for_status(resp) + + assert excinfo.value.problem_details.status == status_code + assert excinfo.value.problem_details.title == title + assert excinfo.value.response == resp + + +@pytest.mark.parametrize( + "status_code, error_type", + [ + (404, "Client Error"), + (500, "Server Error"), + (403, "Client Error"), + ], +) +class TestRaiseForStatusBackwardCompatibility: + def test_raises_network_connection_error( + self, status_code: int, error_type: str + ) -> None: + resp = MockResponse(b"Error") + resp.status_code = status_code + resp.headers = {"Content-Type": "text/html"} + resp.url = "https://pypi.org/simple/nonexistent-package/" + resp.reason = "Error" + + with pytest.raises(NetworkConnectionError) as excinfo: + raise_for_status(resp) + + assert f"{status_code} {error_type}" in str(excinfo.value) + assert excinfo.value.response == resp