Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions src/pip/_internal/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)
Expand Down Expand Up @@ -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."""

Expand Down
91 changes: 91 additions & 0 deletions src/pip/_internal/network/rfc9457.py
Original file line number Diff line number Diff line change
@@ -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
11 changes: 10 additions & 1 deletion src/pip/_internal/network/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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
Expand Down
169 changes: 169 additions & 0 deletions tests/unit/test_network_rfc9457.py
Original file line number Diff line number Diff line change
@@ -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"<html>Error</html>")
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"<html>Error</html>")
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