scrapy/tests/test_downloadermiddleware_redirect.py

import logging
from unittest.mock import MagicMock

import pytest

from scrapy.downloadermiddlewares.redirect import RedirectMiddleware
from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning
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 _assert_passthrough(
        self, req: Request, mw: RedirectMiddleware | None = None
    ) -> None:
        url2 = "http://www.example.com/redirected"
        rsp = Response(req.url, headers={"Location": url2}, status=301, request=req)
        assert (mw or self.mw).process_response(req, rsp) is rsp

    def _assert_redirected(
        self, req: Request, mw: RedirectMiddleware | None = None
    ) -> None:
        url2 = "http://www.example.com/redirected"
        rsp = Response(req.url, headers={"Location": url2}, status=301, request=req)
        assert isinstance((mw or self.mw).process_response(req, rsp), Request)

    def test_setting_handling(self):
        url = "http://www.example.com/301"
        crawler = get_crawler(DefaultSpider, {"HANDLE_HTTP_CODES": [404, 301, 302]})
        crawler.spider = crawler._create_spider()
        mw = self.mwcls.from_crawler(crawler)
        self._assert_passthrough(Request(url), mw)

    def test_setting_handling_all(self):
        url = "http://www.example.com/301"
        crawler = get_crawler(DefaultSpider, {"HANDLE_HTTP_CODES": True})
        crawler.spider = crawler._create_spider()
        mw = self.mwcls.from_crawler(crawler)
        self._assert_passthrough(Request(url), mw)

    def test_setting_handling_unrelated_code(self):
        url = "http://www.example.com/301"
        crawler = get_crawler(DefaultSpider, {"HANDLE_HTTP_CODES": [404]})
        crawler.spider = crawler._create_spider()
        mw = self.mwcls.from_crawler(crawler)
        self._assert_redirected(Request(url), mw)

    def test_httperror_settings_ignored(self):
        """The deprecated HTTPERROR_* settings never applied to redirects."""
        url = "http://www.example.com/301"
        crawler = get_crawler(DefaultSpider, {"HTTPERROR_ALLOW_ALL": True})
        crawler.spider = crawler._create_spider()
        mw = self.mwcls.from_crawler(crawler)
        self._assert_redirected(Request(url), mw)

    def test_spider_handling(self):
        mw = self._spider_attribute_mw([404, 301, 302])
        self._assert_passthrough(Request("http://www.example.com/301"), mw)

    def test_request_meta_handling(self):
        url = "http://www.example.com/301"
        self._assert_passthrough(Request(url, meta={"handle_http_codes": [301]}))
        self._assert_passthrough(Request(url, meta={"handle_http_codes": True}))
        self._assert_redirected(Request(url, meta={"handle_http_codes": [404]}))
        self._assert_redirected(Request(url, meta={"handle_http_codes": False}))

    def test_request_meta_handling_deprecated(self):
        url = "http://www.example.com/301"
        with pytest.warns(ScrapyDeprecationWarning):
            self._assert_passthrough(
                Request(url, meta={"handle_httpstatus_list": [404, 301, 302]})
            )
        with pytest.warns(ScrapyDeprecationWarning):
            self._assert_passthrough(Request(url, meta={"handle_httpstatus_all": True}))
        # Already warned about above, hence no warning expected here.
        self._assert_redirected(Request(url, meta={"handle_httpstatus_list": [404]}))

    def _spider_attribute_mw(self, codes: list[int]) -> RedirectMiddleware:
        class HandlingSpider(DefaultSpider):
            handle_httpstatus_list = codes

        crawler = get_crawler(HandlingSpider)
        crawler.spider = crawler._create_spider()
        mw = self.mwcls.from_crawler(crawler)
        with pytest.warns(ScrapyDeprecationWarning):
            mw.spider_opened(crawler.spider)
        return mw

    @pytest.mark.parametrize(
        "meta",
        [
            {"handle_httpstatus_list": [404]},
            {"handle_httpstatus_all": False},
        ],
    )
    def test_deprecated_meta_combines_with_spider_attribute(self, meta):
        """This middleware used to combine the deprecated meta key with the
        deprecated spider attribute, instead of overriding it."""
        mw = self._spider_attribute_mw([301])
        with pytest.warns(ScrapyDeprecationWarning):
            self._assert_passthrough(
                Request("http://www.example.com/301", meta=meta), mw
            )

    def test_meta_overrides_spider_attribute(self):
        mw = self._spider_attribute_mw([301])
        self._assert_redirected(
            Request("http://www.example.com/301", meta={"handle_http_codes": [404]}), mw
        )

    def test_request_meta_overrides_setting(self):
        url = "http://www.example.com/301"
        crawler = get_crawler(DefaultSpider, {"HANDLE_HTTP_CODES": True})
        crawler.spider = crawler._create_spider()
        mw = self.mwcls.from_crawler(crawler)
        self._assert_redirected(Request(url, meta={"handle_http_codes": False}), mw)

    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(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)