Function test_upstream_proxy

test/mitmproxy/proxy/layers/http/test_http.py:814–939

Test that an upstream HTTP proxy is used.

(tctx, redirect, domain, scheme)

@pytest.mark.parametrize("domain", [b"example.com", b"xn--eckwd4c7c.xn--zckzah"])
@pytest.mark.parametrize("scheme", ["http", "https"])
def test_upstream_proxy(tctx, redirect, domain, scheme):
    """Test that an upstream HTTP proxy is used."""
    server = Placeholder(Server)
    server2 = Placeholder(Server)
    flow = Placeholder(HTTPFlow)
    tctx.client.proxy_mode = ProxyMode.parse("upstream:http://proxy:8080")
    playbook = Playbook(http.HttpLayer(tctx, HTTPMode.upstream), hooks=False)

    if scheme == "http":
        assert (
            playbook
            >> DataReceived(
                tctx.client,
                b"GET http://%s/ HTTP/1.1\r\nHost: %s\r\n\r\n" % (domain, domain),
            )
            << OpenConnection(server)
            >> reply(None)
            << SendData(
                server,
                b"GET http://%s/ HTTP/1.1\r\nHost: %s\r\n\r\n" % (domain, domain),
            )
        )

    else:
        assert (
            playbook
            >> DataReceived(
                tctx.client,
                b"CONNECT %s:443 HTTP/1.1\r\nHost: %s:443\r\n\r\n" % (domain, domain),
            )
            << SendData(tctx.client, b"HTTP/1.1 200 Connection established\r\n\r\n")
            >> DataReceived(tctx.client, b"GET / HTTP/1.1\r\nHost: %s\r\n\r\n" % domain)
            << layer.NextLayerHook(Placeholder())
            >> reply_next_layer(lambda ctx: http.HttpLayer(ctx, HTTPMode.transparent))
            << OpenConnection(server)
            >> reply(None)
            << SendData(
                server,
                b"CONNECT %s:443 HTTP/1.1\r\nHost: %s:443\r\n\r\n" % (domain, domain),
            )
            >> DataReceived(server, b"HTTP/1.1 200 Connection established\r\n\r\n")
            << SendData(server, b"GET / HTTP/1.1\r\nHost: %s\r\n\r\n" % domain)
        )

    playbook >> DataReceived(server, b"HTTP/1.1 418 OK\r\nContent-Length: 0\r\n\r\n")
    playbook << SendData(tctx.client, b"HTTP/1.1 418 OK\r\nContent-Length: 0\r\n\r\n")

    assert playbook
    assert server().address == ("proxy", 8080)

    if scheme == "http":
        playbook >> DataReceived(
            tctx.client,
            b"GET http://%s/two HTTP/1.1\r\nHost: %s\r\n\r\n" % (domain, domain),
        )
    else:
        playbook >> DataReceived(
            tctx.client, b"GET /two HTTP/1.1\r\nHost: %s\r\n\r\n" % domain