scrapy/tests/test_downloadermiddleware_redirect.py

from __future__ import annotations

import logging
from abc import ABC
from unittest.mock import MagicMock

import pytest

from scrapy.downloadermiddlewares.redirect import RedirectMiddleware
from scrapy.exceptions import NotConfigured
from scrapy.http import Request, Response
from scrapy.spidermiddlewares.referer import (
    POLICY_NO_REFERRER,
    POLICY_ORIGIN,
    POLICY_UNSAFE_URL,
    RefererMiddleware,
)
from scrapy.spiders import Spider
from scrapy.utils.misc import build_from_crawler
from scrapy.utils.spider import DefaultSpider
from scrapy.utils.test import get_crawler
from tests.utils.bases.redirect import TestRedirectBase
from tests.utils.redirect import REDIRECT_SCHEME_CASES, SCHEME_PARAMS


class TestRedirectMiddleware(TestRedirectBase):
    mwcls = RedirectMiddleware
    reason = 302

    def setup_method(self):
        crawler = get_crawler(DefaultSpider)
        crawler.spider = crawler._create_spider()
        self.mw = build_from_crawler(self.mwcls, crawler)

    def get_response(self, request, location, status=302):
        headers = {"Location": location}
        return Response(request.url, status=status, headers=headers)

    def test_redirect_307_308_preserve_method(self):
        def _test(method, status: int):
            url = f"http://www.example.com/{status}"
            url2 = "http://www.example.com/redirected"
            req = Request(url, method=method)
            rsp = Response(url, headers={"Location": url2}, status=status)

            req2 = self.mw.process_response(req, rsp)
            assert isinstance(req2, Request)
            assert req2.url == url2
            assert req2.method == method

            # response without Location header but with status code is 3XX should be ignored
            del rsp.headers["Location"]
            assert self.mw.process_response(req, rsp) is rsp

        _test("GET", status=307)
        _test("POST", status=307)
        _test("HEAD", status=307)

        _test("GET", status=308)
        _test("POST", status=308)
        _test("HEAD", status=308)

    @pytest.mark.parametrize("status", [301, 302, 303])
    def test_method_becomes_get(self, status):
        source_url = f"http://www.example.com/{status}"
        target_url = "http://www.example.com/redirected2"
        request = Request(
            source_url,
            method="POST",
            body="test",
            headers={"Content-Type": "text/plain", "Content-length": "4"},
        )
        response = Response(source_url, headers={"Location": target_url}, status=status)
        redirect_request = self.mw.process_response(request, response)
        assert isinstance(redirect_request, Request)
        assert redirect_request.url == target_url
        assert redirect_request.method == "GET"
        assert "Content-Type" not in redirect_request.headers
        assert "Content-Length" not in redirect_request.headers
        assert not redirect_request.body

    @pytest.mark.parametrize("status", [301, 302])
    @pytest.mark.parametrize("method", ["PUT", "DELETE"])
    def test_method_not_converted_on_301_302(self, status, method):
        url = f"http://www.example.com/{status}"
        url2 = "http://www.example.com/redirected"
        body = b"test-body"
        req = Request(
            url,
            method=method,
            body=body,
            headers={"Content-Type": "text/plain", "Content-Length": str(len(body))},
        )
        rsp = Response(url, headers={"Location": url2}, status=status)

        req2 = self.mw.process_response(req, rsp)
        assert isinstance(req2, Request)
        assert req2.url == url2
        assert req2.method == method
        assert req2.body == body
        assert req2.headers[b"Content-Type"] == b"text/plain"
        assert req2.headers[b"Content-Length"] == str(len(body)).encode()

    @pytest.mark.parametrize("method", ["PUT", "DELETE"])
    def test_method_converted_on_303(self, method):
        status = 303
        url = f"http://www.example.com/{status}"
        url2 = "http://www.example.com/redirected"
        req = Request(url, method=method)
        rsp = Response(url, headers={"Location": url2}, status=status)

        req2 = self.mw.process_response(req, rsp)
        assert isinstance(req2, Request)
        assert req2.url == url2
        assert req2.method == "GET"

    def test_get_method_body_preserved_on_303(self):
        status = 303
        url = f"http://www.example.com/{status}"
        url2 = "http://www.example.com/redirected"
        body = b"test-body"
        req = Request(
            url,
            method="GET",
            body=body,
            headers={"Content-Type": "text/plain", "Content-Length": str(len(body))},
        )
        rsp = Response(url, headers={"Location": url2}, status=status)

        req2 = self.mw.process_response(req, rsp)
        assert isinstance(req2, Request)
        assert req2.url == url2
        assert req2.method == "GET"
        assert req2.body == body
        assert req2.headers[b"Content-Type"] == b"text/plain"
        assert req2.headers[b"Content-Length"] == str(len(body)).encode()

    def test_redirect_strips_content_headers(self):
        url = "http://www.example.com/303"
        url2 = "http://www.example.com/redirected"
        headers = {
            "Content-Type": "application/json",
            "Content-Length": "100",
            "Content-Encoding": "gzip",
            "Content-Language": "en",
            "Content-Location": "http://www.example.com/original",
            "X-Custom": "foo",
        }
        req = Request(url, method="POST", headers=headers, body=b"foo")
        rsp = Response(url, headers={"Location": url2}, status=303)

        req2 = self.mw.process_response(req, rsp)
        assert isinstance(req2, Request)
        assert req2.url == url2
        assert req2.method == "GET"
        assert req2.body == b""
        assert "Content-Type" not in req2.headers
        assert "Content-Length" not in req2.headers
        assert "Content-Encoding" not in req2.headers
        assert "Content-Language" not in req2.headers
        assert "Content-Location" not in req2.headers
        assert req2.headers["X-Custom"] == b"foo"

    @pytest.mark.parametrize(
        (
            "referer",
            "source_url",
            "target_url",
            "policy_header",
            "expected_referer",
        ),
        [
            (
                None,
                "http://www.example.com/302",
                "http://www.example.com/redirected",
                None,
                b"http://www.example.com/302",
            ),
            (
                "http://example.com/old",
                "http://www.example.com/302",
                "http://www.example.com/redirected",
                None,
                b"http://www.example.com/302",
            ),
            (
                "https://example.com/old",
                "https://www.example.com/302",
                "http://www.example.com/redirected",
                None,
                None,
            ),
            (
                "http://example.com/old",
                "http://www.example.com/foo/bar",
                "http://www.example.com/redirected",
                "origin",
                b"http://www.example.com/",
            ),
        ],
    )
    def test_redirect_referer(
        self, referer, source_url, target_url, policy_header, expected_referer
    ):
        headers = {"Referer": referer} if referer else {}
        source_request = Request(source_url, headers=headers)
        resp_headers = {"Location": target_url}
        if policy_header:
            resp_headers["Referrer-Policy"] = policy_header
        response = Response(source_url, headers=resp_headers, status=302)
        crawler = get_crawler()
        referer_mw = build_from_crawler(RefererMiddleware, crawler)
        redirect_mw = build_from_crawler(self.mwcls, crawler)
        redirect_mw._referer_spider_middleware = referer_mw
        redirect_request = redirect_mw.process_response(source_request, response)
        if expected_referer:
            assert redirect_request.headers.get("Referer") == expected_referer
        else:
            assert "Referer" not in redirect_request.headers

    def test_redirect_strips_referer_no_middleware(self):
        source_url = "http://www.example.com/302"
        redirect_url = "http://www.example.com/redirected"
        source_request = Request(
            source_url, headers={"Referer": "http://example.com/old"}
        )
        response = Response(source_url, headers={"Location": redirect_url}, status=302)
        redirect_mw = build_from_crawler(self.mwcls, get_crawler())
        redirect_mw._referer_spider_middleware = None
        redirect_request = redirect_mw.process_response(source_request, response)
        assert "Referer" not in redirect_request.headers

    @pytest.mark.parametrize("status", [307, 308])
    def test_cross_origin_maintain_body(self, status):
        source_url = "https://example.com"
        target_url = "https://attacker.example"
        body = b"secret"
        request = Request(
            source_url,
            method="POST",
            body=body,
            headers={
                "Content-Type": "application/json",
                "Content-Length": str(len(body)),
            },
        )
        response1 = Response(
            source_url, headers={"Location": target_url}, status=status
        )
        redirect_request = self.mw.process_response(request, response1)
        assert isinstance(redirect_request, Request)
        assert redirect_request.url == target_url
        assert redirect_request.method == "POST"
        assert redirect_request.body == body
        assert redirect_request.headers[b"Content-Type"] == b"application/json"
        assert redirect_request.headers[b"Content-Length"] == str(len(body)).encode()

    def test_redirect_keeps_fragment(self):
        url = "http://www.example.com/301#frag"
        url2 = "http://www.example.com/redirected"
        req = Request(url)
        rsp = Response(url, headers={"Location": url2}, status=302)

        req2 = self.mw.process_response(req, rsp)
        assert isinstance(req2, Request)
        assert req2.url == "http://www.example.com/redirected#frag"

    def test_redirect_target_has_fragment(self):
        url = "http://www.example.com/302#original"
        url2 = "http://www.example.com/redirected#target"
        req = Request(url)
        rsp = Response(url, headers={"Location": url2}, status=302)

        req2 = self.mw.process_response(req, rsp)
        assert isinstance(req2, Request)
        assert req2.url == "http://www.example.com/redirected#target"

    def test_redirect_302_head(self):
        url = "http://www.example.com/302"
        url2 = "http://www.example.com/redirected2"
        req = Request(url, method="HEAD")
        rsp = Response(url, headers={"Location": url2}, status=302)

        req2 = self.mw.process_response(req, rsp)
        assert isinstance(req2, Request)
        assert req2.url == url2
        assert req2.method == "HEAD"

    def test_redirect_302_relative(self):
        url = "http://www.example.com/302"
        url2 = "///i8n.example2.com/302"
        url3 = "http://i8n.example2.com/302"
        req = Request(url, method="HEAD")
        rsp = Response(url, headers={"Location": url2}, status=302)

        req2 = self.mw.process_response(req, rsp)
        assert isinstance(req2, Request)
        assert req2.url == url3
        assert req2.method == "HEAD"

    def test_spider_handling(self):
        self.mw.crawler.spider.handle_httpstatus_list = [404, 301, 302]
        url = "http://www.example.com/301"
        url2 = "http://www.example.com/redirected"
        req = Request(url)
        rsp = Response(url, headers={"Location": url2}, status=301)
        r = self.mw.process_response(req, rsp)
        assert r is rsp

    def test_request_meta_handling(self):
        url = "http://www.example.com/301"
        url2 = "http://www.example.com/redirected"

        def _test_passthrough(req: Request) -> None:
            rsp = Response(url, headers={"Location": url2}, status=301, request=req)
            r = self.mw.process_response(req, rsp)
            assert r is rsp

        _test_passthrough(
            Request(url, meta={"handle_httpstatus_list": [404, 301, 302]})
        )
        _test_passthrough(Request(url, meta={"handle_httpstatus_all": True}))

    def test_latin1_location(self):
        req = Request("http://scrapytest.org/first")
        latin1_location = "/ação".encode("latin1")  # HTTP historically supports latin1
        resp = Response(
            "http://scrapytest.org/first",
            headers={"Location": latin1_location},
            status=302,
        )
        req_result = self.mw.process_response(req, resp)
        perc_encoded_utf8_url = "http://scrapytest.org/a%E7%E3o"
        assert perc_encoded_utf8_url == req_result.url

    def test_utf8_location(self):
        req = Request("http://scrapytest.org/first")
        utf8_location = "/ação".encode()  # header using UTF-8 encoding
        resp = Response(
            "http://scrapytest.org/first",
            headers={"Location": utf8_location},
            status=302,
        )
        req_result = self.mw.process_response(req, resp)
        perc_encoded_utf8_url = "http://scrapytest.org/a%C3%A7%C3%A3o"
        assert perc_encoded_utf8_url == req_result.url

    def test_no_location(self):
        request = Request("https://example.com")
        response = Response(request.url, status=302)
        assert self.mw.process_response(request, response) is response

    @pytest.mark.parametrize("status", [201, 300])
    def test_status_not_in_redirect_http_codes(self, status):
        """Status codes outside REDIRECT_HTTP_CODES are not followed, even if they
        report a Location header, as 201 (Created) or 300 (Multiple Choices) may
        do."""
        url = f"https://example.com/{status}"
        request = Request(url)
        response = Response(url, status=status, headers={"Location": "/target"})
        assert self.mw.process_response(request, response) is response

    def test_standard_status_not_in_redirect_http_codes(self):
        crawler = get_crawler(DefaultSpider, {"REDIRECT_HTTP_CODES": [301]})
        crawler.spider = crawler._create_spider()
        mw = RedirectMiddleware.from_crawler(crawler)
        url = "https://example.com/302"
        request = Request(url)
        response = Response(url, status=302, headers={"Location": "/redirected"})
        assert mw.process_response(request, response) is response

    def test_meta_http_codes(self):
        """The redirect_http_codes request metadata key overrides the
        REDIRECT_HTTP_CODES setting."""
        url = "https://example.com/201"
        request = Request(url, meta={"redirect_http_codes": [201]})
        response = Response(url, status=201, headers={"Location": "/created"})
        redirect_request = self.mw.process_response(request, response)
        assert isinstance(redirect_request, Request)
        assert redirect_request.url == "https://example.com/created"

    def test_meta_http_codes_excluding_standard_status(self):
        url = "https://example.com/302"
        request = Request(url, meta={"redirect_http_codes": [201]})
        response = Response(url, status=302, headers={"Location": "/redirected"})
        assert self.mw.process_response(request, response) is response


class NonStandardStatusMixin(ABC):
    """Tests of status codes added to REDIRECT_HTTP_CODES for which the HTTP
    standard does not define redirection handling, and which therefore get
    conservative handling."""

    status: int

    def setup_method(self):
        crawler = get_crawler(
            DefaultSpider,
            {"REDIRECT_HTTP_CODES": [self.status, 301, 302, 303, 307, 308]},
        )
        crawler.spider = crawler._create_spider()
        self.mw = RedirectMiddleware.from_crawler(crawler)
        self.url = f"https://example.com/{self.status}"

    def _response(
        self, body: bytes = b"", location: str | None = "/target"
    ) -> Response:
        headers = {"Location": location} if location else {}
        return Response(self.url, status=self.status, headers=headers, body=body)

    def test_empty_body(self):
        redirect_request = self.mw.process_response(Request(self.url), self._response())
        assert isinstance(redirect_request, Request)
        assert redirect_request.url == "https://example.com/target"
        assert redirect_request.method == "GET"
        assert redirect_request.meta["redirect_urls"] == [self.url]
        assert redirect_request.meta["redirect_reasons"] == [self.status]

    def test_non_empty_body(self):
        """A response that carries a resource in its body is not followed, doing so
        would discard that resource."""
        response = self._response(body=b"file content")
        assert self.mw.process_response(Request(self.url), response) is response

    def test_no_location(self):
        response = self._response(location=None)
        assert self.mw.process_response(Request(self.url), response) is response

    @pytest.mark.parametrize("method", ["POST", "PUT"])
    def test_method_becomes_get(self, method):
        request = Request(
            self.url,
            method=method,
            body=b"payload",
            headers={"Content-Type": "text/plain", "Content-Length": "7"},
        )
        redirect_request = self.mw.process_response(request, self._response())
        assert isinstance(redirect_request, Request)
        assert redirect_request.method == "GET"
        assert not redirect_request.body
        assert "Content-Type" not in redirect_request.headers
        assert "Content-Length" not in redirect_request.headers

    def test_head_method_preserved(self):
        request = Request(self.url, method="HEAD")
        redirect_request = self.mw.process_response(request, self._response())
        assert isinstance(redirect_request, Request)
        assert redirect_request.method == "HEAD"

    @pytest.mark.parametrize(
        "meta",
        [
            {"dont_redirect": True},
            {"handle_httpstatus_all": True},
        ],
    )
    def test_request_meta_handling(self, meta):
        request = Request(self.url, meta=meta)
        response = self._response()
        assert self.mw.process_response(request, response) is response

    def test_request_meta_status_handling(self):
        request = Request(self.url, meta={"handle_httpstatus_list": [self.status]})
        response = self._response()
        assert self.mw.process_response(request, response) is response

    def test_spider_handling(self):
        self.mw.crawler.spider.handle_httpstatus_list = [self.status]  # type: ignore[union-attr]
        response = self._response()
        assert self.mw.process_response(Request(self.url), response) is response


class TestRedirect201(NonStandardStatusMixin):
    """201 (Created) reports the created resource through Location."""

    status = 201


class TestRedirect300(NonStandardStatusMixin):
    """300 (Multiple Choices) may report a preferred choice through Location. It
    is a 3xx status code, but the standard defines no redirection handling for
    it, so it gets the same conservative handling as 201."""

    status = 300


@pytest.mark.parametrize(SCHEME_PARAMS, REDIRECT_SCHEME_CASES)
def test_redirect_schemes(url, location, target):
    crawler = get_crawler(Spider)
    mw = build_from_crawler(RedirectMiddleware, crawler)
    request = Request(url)
    response = Response(url, headers={"Location": location}, status=301)
    redirect = mw.process_response(request, response)
    if target is None:
        assert redirect == response
    else:
        assert isinstance(redirect, Request)
        assert redirect.url == target


@pytest.mark.parametrize(
    ("policy", "source_url", "target_url", "expected_referrer"),
    [
        # The policy header affects the outcome.
        # (without it, the https → http switch would drop the referer)
        (
            POLICY_UNSAFE_URL,
            "https://a.example/1",
            "http://a.example/2",
            b"https://a.example/1",
        ),
        # The policy header can get the Referer header removed.
        (
            POLICY_NO_REFERRER,
            "http://a.example/1",
            "http://a.example/2",
            None,
        ),
        # The policy header can get the Referer header edited (path stripped).
        (
            POLICY_ORIGIN,
            "http://a.example/1",
            "http://a.example/2",
            b"http://a.example/",
        ),
    ],
)
def test_response_referrer_policy(policy, source_url, target_url, expected_referrer):
    crawler = get_crawler()
    referrer_mw = build_from_crawler(RefererMiddleware, crawler)
    redirect_mw = build_from_crawler(RedirectMiddleware, crawler)
    redirect_mw._referer_spider_middleware = referrer_mw
    source_request = Request(source_url)
    extra_headers = {}
    if policy:
        extra_headers["Referrer-Policy"] = policy
    response_redirect = Response(
        source_request.url,
        status=301,
        headers={"Location": target_url, **extra_headers},
    )
    target_request = redirect_mw.process_response(source_request, response_redirect)
    assert isinstance(target_request, Request)

    assert target_request.headers.get("Referer") == expected_referrer


def test_no_warning_when_referer_middleware_present(caplog):
    crawler = get_crawler()
    crawler.get_spider_middleware = MagicMock(  # type: ignore[method-assign]
        return_value=MagicMock()
    )
    mw = build_from_crawler(RedirectMiddleware, crawler)
    caplog.clear()
    with caplog.at_level(logging.WARNING):
        mw._engine_started()
    assert not [
        record
        for record in caplog.records
        if record.name == "scrapy.downloadermiddlewares.redirect"
    ]


def test_warning_redirect_middleware(caplog):
    crawler = get_crawler()
    crawler.get_spider_middleware = MagicMock(  # type: ignore[method-assign]
        return_value=None
    )
    mw = build_from_crawler(RedirectMiddleware, crawler)
    with caplog.at_level(logging.WARNING):
        mw._engine_started()
    assert (
        "scrapy.downloadermiddlewares.redirect.RedirectMiddleware found no "
        "scrapy.spidermiddlewares.referer.RefererMiddleware"
    ) in caplog.text
    assert (
        "enable scrapy.spidermiddlewares.referer.RefererMiddleware (or a subclass)"
        in caplog.text
    )
    assert (
        "replace scrapy.downloadermiddlewares.redirect.RedirectMiddleware "
        "with a subclass that overrides the handle_referer() method"
    ) in caplog.text


def test_warning_subclass(caplog):
    class MyRedirectMiddleware(RedirectMiddleware):
        pass

    crawler = get_crawler()
    crawler.get_spider_middleware = MagicMock(  # type: ignore[method-assign]
        return_value=None
    )
    mw = build_from_crawler(MyRedirectMiddleware, crawler)
    with caplog.at_level(logging.WARNING):
        mw._engine_started()
    assert (
        "test_warning_subclass.<locals>.MyRedirectMiddleware found no "
        "scrapy.spidermiddlewares.referer.RefererMiddleware"
    ) in caplog.text
    assert (
        "enable scrapy.spidermiddlewares.referer.RefererMiddleware (or a subclass)"
        in caplog.text
    )
    assert "edit " in caplog.text
    assert "test_warning_subclass.<locals>.MyRedirectMiddleware" in caplog.text
    assert (
        "(if defined in your code base) to override the handle_referer() method"
    ) in caplog.text


def test_not_configured():
    crawler = get_crawler(DefaultSpider, {"REDIRECT_ENABLED": False})
    with pytest.raises(NotConfigured):
        build_from_crawler(RedirectMiddleware, crawler)