Client: ClientSession will not send secure cookie to localhost on unsecure connection.

[DanielDewberry]

A ClientSession with an unsafe cookiejar will not send a secure cookie to a localhost endpoint, when that connection is over http (i,e. unsecure).

According to Mozilla documentation:

A cookie with the Secure attribute is sent to the server only with an encrypted request over the HTTPS protocol, never with unsecured HTTP (except on localhost), and therefore can't easily be accessed by a man-in-the-middle attacker. Insecure sites (with http: in the URL) can't set cookies with the Secure attribute

Note the (except on localhost) clause.

💡 To Reproduce
The following example sets up a test (TestIntegration::test_cookie_is_sent_server) which fails. Supplementary tests are included to demonstrate the setup operates as expected. Scroll to the end of the codeblock to see that particular test.

#! /usr/bin/env pytest
# -*- coding: utf-8 -*-

"""Test module."""

import base64
import http
from typing import cast

from aiohttp import ClientSession
from aiohttp import web
from aiohttp.cookiejar import CookieJar
from aiohttp.test_utils import TestClient

from aiohttp_session import get_session, setup
from aiohttp_session.cookie_storage import EncryptedCookieStorage

from cryptography import fernet

import pytest


fernet_key = fernet.Fernet.generate_key()


async def index_get_handler(request: web.Request) -> web.Response:
    session = await get_session(request)
    secret_key = session.get('session_id') or 'not set'
    key = 'key: {}'.format(secret_key)

    response = web.Response(
        status=http.HTTPStatus(200),
        headers={'Content-Type': 'text/plain; charset=UTF-8',
                 },
        body=key,
    )

    return response


async def login_post_handler(request: web.Request) -> web.Response:
    session = await get_session(request)
    session['session_id'] = 'some secret data'

    response = web.Response(
        status=http.HTTPStatus(303),
        headers={'Content-Type': 'text/plain; charset=UTF-8',
                 'Location': '/',
                 },
    )

    return response


@pytest.fixture
def application_instance() -> web.Application:
    """Generate the application instance."""
    application = web.Application()

    application.add_routes(
        [
            web.get('/', index_get_handler, name='index'),
            web.post('/login', login_post_handler, name='login'),
        ]
    )

    setup(
        app=application,
        storage=EncryptedCookieStorage(
            secret_key=base64.urlsafe_b64decode(fernet_key),
            cookie_name='session_data',
            secure=True,
        ),
    )

    return application


# Tests

class TestIndexHandlerGetMethod:
    async def test_index_handler_returns_default_key_text(
        self,
        aiohttp_client,
        loop,
        application_instance,
    ):
        """index handler returns default text when session cookie not present."""
        jar = CookieJar(unsafe=True)
        client: TestClient = await aiohttp_client(
            application_instance,
            cookie_jar=jar,
        )
        url = 'http://127.0.0.1:{}/'.format(client.port)

        async with client.session as session:
            cast(ClientSession, session)

            response = await session.get(
                url=url,
                allow_redirects=False,
            )
            body = await response.text()
        assert 'key: not set' == body


    async def test_index_page_returns_no_session_cookie(
        self,
        aiohttp_client,
        loop,
        application_instance,
    ):
        """index handler does not set a session cookie."""
        jar = CookieJar(unsafe=True)
        client: TestClient = await aiohttp_client(
            application_instance,
            cookie_jar=jar,
        )
        url = 'http://127.0.0.1:{}/'.format(client.port)

        async with client.session as session:
            cast(ClientSession, session)

            response = await session.get(
                url=url,
                allow_redirects=False,
            )

            assert response.cookies.get('session_data') is None


    async def test_index_page_returns_200_status(
        self,
        aiohttp_client,
        loop,
        application_instance,
    ):
        """index handler returns 200 status."""
        jar = CookieJar(unsafe=True)
        client: TestClient = await aiohttp_client(
            application_instance,
            cookie_jar=jar,
        )
        url = 'http://127.0.0.1:{}/'.format(client.port)

        async with client.session as session:
            cast(ClientSession, session)

            response = await session.get(
                url=url,
                allow_redirects=False,
            )

            assert response.status == 200


class TestLoginHandlerPostMethod:
    async def test_handler_sets_cookie(
        self,
        aiohttp_client,
        loop,
        application_instance,
    ):
        """login handler sets session cookie."""
        jar = CookieJar(unsafe=True)
        client: TestClient = await aiohttp_client(
            application_instance,
            cookie_jar=jar,
        )
        url = 'http://127.0.0.1:{}/login'.format(client.port)

        async with client.session as session:
            cast(ClientSession, session)

            response = await session.post(
                url=url,
                allow_redirects=False,
            )
            assert response.cookies.get('session_data') is not None

    async def test_handler_redirects_to_index(
        self,
        aiohttp_client,
        loop,
        application_instance,
    ):
        """login handler redirects to the index page."""
        jar = CookieJar(unsafe=True)
        client: TestClient = await aiohttp_client(
            application_instance,
            cookie_jar=jar,
        )
        url = 'http://127.0.0.1:{}/login'.format(client.port)

        async with client.session as session:
            cast(ClientSession, session)

            response = await session.post(
                url=url,
                allow_redirects=False,
            )

            assert response.status == 303
            assert response.headers.get('location') == '/'


class TestIntegration:

    async def test_cookie_is_sent_server(
            self,
            aiohttp_client,
            loop,
            application_instance,
        ):
        """A session cookie sent to index handler returns the session secret in response body.

        This test fails as the cookie storate filters away the secure cookied when the request
        asked to send cookies over an unsecure connection."""
        jar = CookieJar(unsafe=True)
        client: TestClient = await aiohttp_client(
            application_instance,
            cookie_jar=jar,
        )

        async with client.session as session:
            cast(ClientSession, session)

            assert session.cookie_jar is jar

            login_post_url = 'http://127.0.0.1:{}/login'.format(client.port)
            response = await session.post(
                url=login_post_url,
            )

            assert response.status == 200
            body = await response.text()
            assert body is not None
            assert body == 'key: some secret data'

Requirements:

aiohttp           == 3.7.4
aiohttp-session   == 2.9.0
cryptography      == 3.4.6
multidict         == 5.1.0
packaging         == 20.9
py                == 1.10.0
setuptools        == 53.0.0
yarl              == 1.6.3
pytest            == 6.2.2
pytest-aiohttp    ==  0.3.0

💡 Expected behavior
Secure cookies should be sent to localhost/ 127.0.0.1 on unsecure connection (http)

📋 Your version of the Python

python 3.8.5

📋 Your version of the aiohttp/yarl/multidict distributions

aiohttp: 3.7.4
multidict: 5.1.0
yarl: 1.6.3

📋 Additional context
This is a client issue.

Proposed Solution:

change this line of aiohttp/cookiejar.py
from

if is_not_secure and cookie["secure"]:

to

if is_not_secure and cookie["secure"] and domain not in ['localhost', '127.0.0.1']:

or provide a way to influence the is_not_secure variable in order to proceed to sending the cookie, whether the connection is secure or not. The latter would allow for the ipv4 and ipv6 loopbacks to be affected without exceptional cases (as was the case in the former proposal), and would also provide a way for hosts in /etc/hosts to be included in the unsecured requests.

Further rationale:

This interpretation shows the RFC section that states:

The Secure attribute limits the scope of the cookie to "secure"
channels (where "secure" is defined by the user agent). When a
cookie has the Secure attribute, the user agent will include the
cookie in an HTTP request only if the request is transmitted over a
secure channel (typically HTTP over Transport Layer Security (TLS)
[RFC2818]).
Although seemingly useful for protecting cookies from active network
attackers, the Secure attribute protects only the cookie's
confidentiality

I, the user agent, consider the sending and receipt of a request on a single machine, not across a network, to be secure. Particularly when performing unit testing.

[Marco-Kaulea]

We are running into the same issue during testing. Another solution would be to allow secure cookies over an insecure connection when the unsafe parameter is set. Or split out that check into a class method, so it is easier to replace that part by creating a TestCookieJar.

[webknjaz]

@DanielDewberry could you send a pull request with those failing tests? Let's incorporate them first and then talk about the possible fix.

@Marco-Kaulea this may be a good solution, I need to think about it. We could talk about this in the PR with tests that would demonstrate how such interfaces could be used. I expect that we can follow TDD in this case and merge in failing (but agreed on) tests marked as xfail. And then, somebody could implement what those tests expect.

[DanielDewberry]

@webknjaz Would you like the PR to a new or existing branch?

[webknjaz]

Just a standalone PR with failing tests against master would be fine. We can then mark the tests as xfail per https://pganssle-talks.github.io/xfail-lightning/ and merge them (that would be followed by a backport PR to the 3.8 branch but you needn't worry about this right now).

Once the tests are in, it'll be clearly documented what behavior is expected going forward as code. The next step would be to produce a separate PR making those tests pass by effectively changing the behavior.

Of course, these changes could be a single PR but it's usually easier to agree upon and merge smaller chunks rather than a big set of changes because in big PRs there's usually some subset of patches that could be merged in right away but they would still be stuck in that PR until all other things are fixed which causes longer review/reiteration times.

[DanielDewberry]

@webknjaz @Marco-Kaulea I have submitted a PR as requested.
The test that I have added does not resemble the source code previously submitted on this thread.

It is now:

• far simpler
• not dependent upon external dependencies outside those already used in the aiohttp test suite
• focused directly on the CookieJar, not the ClientSession which is where the behaviour lies

Here is the revised test:

#! /usr/bin/env python3
# -*- coding: utf-8 -*-

from http.cookies import SimpleCookie

from aiohttp.cookiejar import CookieJar

import pytest

from yarl import URL


async def test_secure_cookie_not_filtered_from_unsafe_cookiejar_when_given_unsecured_endpoint() -> None:
    """Secure SimpleCookie should not be filtered from unsafe CookieJar when given an unsecured endpoint.

    There are times when sending a secure cookie to an unsecured endpoint is desireable.  Such an
    occasion is during testing. RFC 6265 section-4.1.2.5 states that this behaviour is a decision
    based on the trust of a network by the user agent.
    """
    endpoint = 'http://127.0.0.1/'

    secure_cookie = SimpleCookie(
        "cookie-key=cookie-value; HttpOnly; Path=/; Secure",
    )

    jar = CookieJar(unsafe=True)

    # Confirm the jar is empty
    assert len(jar) == 0

    jar.update_cookies(
        secure_cookie,
        URL(endpoint),
    )

    # Confirm the jar contains the cookie
    assert len(jar) == 1

    filtered_cookies = jar.filter_cookies(request_url=endpoint)

    # Confirm the filtered results contain the cookie
    assert len(filtered_cookies) == 1

[Dreamsorcerer]

Based on that, it looks like you want some way to indicate you are on a safe/secure network, and when that value is set, the secure cookie would not get filtered out.

[Dreamsorcerer]

Or rather, a way to indicate the destination is within the same secure network.

[DanielDewberry]

@Dreamsorcerer The decision to send a secure cookie to the connected host should be made by the user agent, independent of which network they are on. CookieJar/ filter_cookies should provide an option to include secure cookies from unsecured hosts in the results.

The CookieJar interface could be extended like so:

class CookieJar(AbstractCookieJar):
    def __init__(
        self, *, 
        unsafe: bool = False, 
        quote_cookie: bool = True
        include_secure_cookie_over_unsecured_connection: bool = False,   # new kwarg        
    ) -> None:
    self._include_secure_cookie_over_unsecured_connection = include_secure_cookie_over_unsecured_connection

Then filter_cookies would include this in the security check:

    def filter_cookies(
        self, request_url: URL = URL()
    ) -> Union["BaseCookie[str]", "SimpleCookie[str]"]:
    
    # lines leading up to 253
            if is_not_secure and \
               cookie["secure"] and \
               not self._include_secure_cookie_over_unsecured_connection:

(I like verbose names but appreciate that's a long one).

One suggestion was to add a TestCookieJar class with the desired functionality, but this would mean the CookieJar would continue to reject cookies during user testing on the LAN and the issue I raised would persist.

[derlih]

IMO it would be more flexible to use a predicate instead of boolean flag.
The first check in the loop inside filter_cookies would be against this predicate.