CVE-2022-24737 (Python, published 2022-03-07T23:15Z)
Description: value: HTTPie is a command-line HTTP client. HTTPie has the practical concept of sessions, which help users to persistently store some of the state that belongs to the outgoing requests and incoming responses on the disk for further usage. Before 3.1.0, HTTPie didn‘t distinguish between cookies and hosts they belonged. This behavior resulted in the exposure of some cookies when there are redirects originating from the actual host to a third party website. Users are advised to upgrade. There are no known workarounds.
Severity: MEDIUM (CVSS v3 base score: 6.5)
Weakness: CWE-200 \u2014 Exposure of Sensitive Information to an Unauthorized Actor
The product exposes sensitive information to an actor that is not explicitly authorized to have access to that information.
Repository: https://github.com/httpie/httpie
Files changed:
docs/README.md
httpie/client.py
httpie/config.py
httpie/manager/cli.py
httpie/manager/core.py
httpie/manager/tasks.py
httpie/sessions.py
httpie/utils.py
setup.py
tests/conftest.py
tests/fixtures/__init__.py
tests/fixtures/session_data/new/cookies_dict.json
tests/fixtures/session_data/new/cookies_dict_dev_version.json
tests/fixtures/session_data/new/cookies_dict_with_extras.json
tests/fixtures/session_data/new/empty_cookies_dict.json
tests/fixtures/session_data/new/empty_cookies_list.json
tests/fixtures/session_data/old/cookies_dict.json
tests/fixtures/session_data/old/cookies_dict_dev_version.json
tests/fixtures/session_data/old/cookies_dict_with_extras.json
tests/fixtures/session_data/old/empty_cookies_dict.json
tests/fixtures/session_data/old/empty_cookies_list.json
tests/test_cookie_on_redirects.py
tests/test_httpie_cli.py
tests/test_plugins_cli.py
tests/test_sessions.py
tests/utils/__init__.py
tests/utils/http_server.py
Commit message: Implement new style cookies


Vulnerable code:
            httpie_session.remove_cookies(
                # TODO: take path & domain into account?
                cookie['name'] for cookie in expired_cookies
            )
from typing import Union
        try:
            with self.path.open(encoding=UTF8) as f:
                try:
                    data = json.load(f)
                except ValueError as e:
                    raise ConfigFileError(
                        f'invalid {config_type} file: {e} [{self.path}]'
                    )
                self.update(data)
        except FileNotFoundError:
            pass
        except OSError as e:
            raise ConfigFileError(f'cannot read {config_type} file: {e}')

    def save(self):
        self['__meta__'] = {
            'httpie': __version__
        }
            obj=self,
            command_parser.add_argument(**argument)
from typing import Iterable, Optional, Union
from urllib.parse import urlsplit
from requests.cookies import RequestsCookieJar, create_cookie
    if os.path.sep in session_name:
        hostname = host or urlsplit(url).netloc.split('@')[-1]
        if not hostname:
            # HACK/FIXME: httpie-unixsocket's URLs have no hostname.
            hostname = 'localhost'

        # host:port => host_port
        hostname = hostname.replace(':', '_')
    session = Session(path)
    def __init__(self, path: Union[str, Path]):
        self['cookies'] = {}

                    self['cookies'][cookie_name] = {'value': morsel.value}
                del request_headers[name]
        jar = RequestsCookieJar()
        for name, cookie_dict in self['cookies'].items():
            jar.set_cookie(create_cookie(
                name, cookie_dict.pop('value'), **cookie_dict))
        jar.clear_expired_cookies()
        return jar
        # <https://docs.python.org/3/library/cookielib.html#cookie-objects>
        stored_attrs = ['value', 'path', 'secure', 'expires']
        self['cookies'] = {}
        for cookie in jar:
            self['cookies'][cookie.name] = {
                attname: getattr(cookie, attname)
                for attname in stored_attrs
            }

    def remove_cookies(self, names: Iterable[str]):
        for name in names:
            if name in self['cookies']:
                del self['cookies'][name]
from .utils import HTTPBIN_WITH_CHUNKED_SUPPORT_DOMAIN, HTTPBIN_WITH_CHUNKED_SUPPORT
from tests.utils import httpie


@pytest.mark.requires_installation
def test_plugins_cli_error_message_without_args():
    # No arguments
    result = httpie(no_debug=True)
    assert result.exit_status == ExitStatus.ERROR
    assert 'usage: ' in result.stderr
    assert 'specify one of these' in result.stderr
    assert 'please use the http/https commands:' in result.stderr


@pytest.mark.parametrize(
    'example', [
        'pie.dev/get',
        'DELETE localhost:8000/delete',
        'POST pie.dev/post header:value a=b header_2:value x:=1'
    ]
)
@pytest.mark.requires_installation
def test_plugins_cli_error_messages_with_example(example):
    result = httpie(*example.split(), no_debug=True)
    assert result.exit_status == ExitStatus.ERROR
    assert 'usage: ' in result.stderr
    assert f'http {example}' in result.stderr
    assert f'https {example}' in result.stderr


@pytest.mark.parametrize(
    'example', [
        'plugins unknown',
        'plugins unknown.com A:B c=d',
        'unknown.com UNPARSABLE????SYNTAX',
    ]
)
@pytest.mark.requires_installation
def test_plugins_cli_error_messages_invalid_example(example):
    result = httpie(*example.split(), no_debug=True)
    assert result.exit_status == ExitStatus.ERROR
    assert 'usage: ' in result.stderr
    assert f'http {example}' not in result.stderr
    assert f'https {example}' not in result.stderr
from .utils import HTTP_OK, MockEnvironment, http, mk_config_dir
            "headers": {
                "cookie": "...",
                "zzz": "..."
                basic_auth_header = "Basic " + b64encode(self.raw_auth.encode()).strip().decode('latin1')
        assert updated_session['auth']['raw_auth'] == "user:password"
            ({'id': {'value': 123}}, 'id'),
            ({'id': {'value': 123}}, 'token')
    def test_removes_expired_cookies_from_session_obj(self, initial_cookie, expired_cookie, httpbin):
        session = Session(self.config_dir)
    env = kwargs.setdefault('env', MockEnvironment())
        response = StrCLIResponse(env.stdout.read())

Fixed code:
### Host-based Cookie Policy

Cookies in stored HTTPie sessions have a `domain` field which is binding them to the
specified hostname. For example, in the following session:

```json
{
    "cookies": [
        {
            "domain": "pie.dev",
            "name": "secret_cookie",
            "value": "value_1"
        },
        {
            "domain": "httpbin.org",
            "name": "secret_cookie",
            "value": "value_2"
        }
    ]
}
```

we will send `Cookie:secret_cookie=value_1` only when you are making a request against `pie.dev` (it
also includes the domains, like `api.pie.dev`), and `Cookie:secret_cookie=value_2` when you use `httpbin.org`.

```bash
$ http --session=./session.json pie.dev/cookies
```

```json
{
    "cookies": {
        "secret_cookie": "value_1"
    }
}
```

```bash
$ http --session=./session.json httpbin.org/cookies
```

```json
{
    "cookies": {
        "secret_cookie": "value_2"
    }
}
```

If you want to make a cookie domain unbound, you can simply set the `domain`
field to `null` by editing the session file directly:

```json
{
    "cookies": [
        {
            "domain": null,
            "expires": null,
            "name": "generic_cookie",
            "path": "/",
            "secure": false,
            "value": "generic_value"
        }
    ]
}
```

```bash
$ http --session=./session.json pie.dev/cookies
```

```json
{
    "cookies": {
        "generic_cookie": "generic_value"
    }
}
```

### Upgrading Sessions

In rare circumstances, HTTPie makes changes in it's session layout. For allowing a smoother transition of existing files
from the old layout to the new layout we offer 2 interfaces:

- `httpie cli sessions upgrade`
- `httpie cli sessions upgrade-all`


With `httpie cli sessions upgrade`, you can upgrade a single session with it's name (or it's path, if it is an
[anonymous session](#anonymous-sessions)) and the hostname it belongs to. For example:

([named session](#named-sessions))

```bash
$ httpie cli sessions upgrade pie.dev api_auth
Refactored 'api_auth' (for 'pie.dev') to the version 3.1.0.
```

([anonymous session](#anonymous-sessions))

```bash
$ httpie cli sessions upgrade pie.dev ./session.json
Refactored 'session' (for 'pie.dev') to the version 3.1.0.
```

If you want to upgrade every existing [named session](#named-sessions), you can use `httpie cli sessions upgrade-all` (be aware
that this won't upgrade [anonymous sessions](#anonymous-sessions)):

```bash
$ httpie cli sessions upgrade-all
Refactored 'api_auth' (for 'pie.dev') to the version 3.1.0.
Refactored 'login_cookies' (for 'httpie.io') to the version 3.1.0.
```

#### Additional Customizations

| Flag             | Description                                                                                                                                                                                                                  |
|------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `--bind-cookies` | Bind all the unbound cookies to the hostname that session belongs. By default, if the cookie is unbound (the `domain` attribute does not exist / set to an empty string) then it will still continue to be a generic cookie. |

These flags can be used to customize the defaults during an `upgrade` operation. They can
be used in both `sessions upgrade` and `sessions upgrade-all`.

### Security

#### Exposure of Cookies To The 3rd Party Hosts On Redirects

*Vulnerability Type*: [CWE-200](https://cwe.mitre.org/data/definitions/200.html)
*Severity Level*: LOW
*Affected Versions*: `<3.1.0`

The handling of [cookies](#cookies) was not compatible with the [RFC 6265](https://datatracker.ietf.org/doc/html/rfc6265)
on the point of handling the `Domain` attribute when they were saved into [session](#sessions) files. All cookies were shared
across all hosts during the runtime, including redirects to the 3rd party hosts.

This vulnerability has been fixed in [3.1.0](https://github.com/httpie/httpie/releases/tag/3.1.0) and the
[`httpie cli sessions upgrade`](#upgrading-sessions)/[`httpie cli sessions upgrade-all`]((#upgrading-sessions) commands
have been put in place in order to allow a smooth transition to the new session layout from the existing [session](#sessions)
files.

            env=env,
            httpie_session.remove_cookies(expired_cookies)
from typing import Any, Dict, Union
def read_raw_config(config_type: str, path: Path) -> Dict[str, Any]:
    try:
        with path.open(encoding=UTF8) as f:
            try:
                return json.load(f)
            except ValueError as e:
                raise ConfigFileError(
                    f'invalid {config_type} file: {e} [{path}]'
                )
    except FileNotFoundError:
        pass
    except OSError as e:
        raise ConfigFileError(f'cannot read {config_type} file: {e}')


    def pre_process_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Hook for processing the incoming config data."""
        return data

    def post_process_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Hook for processing the outgoing config data."""
        return data

        data = read_raw_config(config_type, self.path)
        if data is not None:
            data = self.pre_process_data(data)
            self.update(data)

    def save(self, *, bump_version: bool = False):
        self.setdefault('__meta__', {})
        if bump_version or 'httpie' not in self['__meta__']:
            self['__meta__']['httpie'] = __version__
            obj=self.post_process_data(self),
    @property
    def version(self):
        return self.get(
            '__meta__', {}
        ).get('httpie', __version__)

CLI_SESSION_UPGRADE_FLAGS = [
    {
        'variadic': ['--bind-cookies'],
        'action': 'store_true',
        'default': False,
        'help': 'Bind domainless cookies to the host that session belongs.'
    }
]

    'cli': {
        'help': 'Manage HTTPie for Terminal',
        'sessions': {
            'help': 'Manage HTTPie sessions',
            'upgrade': [
                'Upgrade the given HTTPie session with the latest '
                'layout. A list of changes between different session versions '
                'can be found in the official documentation.',
                {
                    'dest': 'hostname',
                    'metavar': 'HOSTNAME',
                    'help': 'The host this session belongs.'
                },
                {
                    'dest': 'session',
                    'metavar': 'SESSION_NAME_OR_PATH',
                    'help': 'The name or the path for the session that will be upgraded.'
                },
                *CLI_SESSION_UPGRADE_FLAGS
            ],
            'upgrade-all': [
                'Upgrade all named sessions with the latest layout. A list of '
                'changes between different session versions can be found in the official '
                'documentation.',
                *CLI_SESSION_UPGRADE_FLAGS
            ],
        }
    }
        properties = properties.copy()

            argument = argument.copy()
            variadic = argument.pop('variadic', [])
            command_parser.add_argument(*variadic, **argument)
from typing import Optional
from httpie.manager.tasks import CLI_TASKS
def dispatch_cli_task(env: Environment, action: Optional[str], args: argparse.Namespace) -> ExitStatus:
    if action is None:
        parser.error(missing_subcommand('cli'))

    return CLI_TASKS[action](env, args)


    elif args.action == 'cli':
        return dispatch_cli_task(env, args.cli_action, args)
import argparse
from typing import TypeVar, Callable, Tuple

from httpie.sessions import SESSIONS_DIR_NAME, Session, get_httpie_session
from httpie.status import ExitStatus
from httpie.context import Environment
from httpie.manager.cli import missing_subcommand, parser

T = TypeVar('T')

CLI_TASKS = {}


def task(name: str) -> Callable[[T], T]:
    def wrapper(func: T) -> T:
        CLI_TASKS[name] = func
        return func
    return wrapper


@task('sessions')
def cli_sessions(env: Environment, args: argparse.Namespace) -> ExitStatus:
    action = args.cli_sessions_action
    if action is None:
        parser.error(missing_subcommand('cli', 'sessions'))

    if action == 'upgrade':
        return cli_upgrade_session(env, args)
    elif action == 'upgrade-all':
        return cli_upgrade_all_sessions(env, args)
    else:
        raise ValueError(f'Unexpected action: {action}')


def is_version_greater(version_1: str, version_2: str) -> bool:
    # In an ideal scenerio, we would depend on `packaging` in order
    # to offer PEP 440 compatible parsing. But since it might not be
    # commonly available for outside packages, and since we are only
    # going to parse HTTPie's own version it should be fine to compare
    # this in a SemVer subset fashion.

    def split_version(version: str) -> Tuple[int, ...]:
        parts = []
        for part in version.split('.')[:3]:
            try:
                parts.append(int(part))
            except ValueError:
                break
        return tuple(parts)

    return split_version(version_1) > split_version(version_2)


def fix_cookie_layout(session: Session, hostname: str, args: argparse.Namespace) -> None:
    if not isinstance(session['cookies'], dict):
        return None

    session['cookies'] = [
        {
            'name': key,
            **value
        }
        for key, value in session['cookies'].items()
    ]
    for cookie in session.cookies:
        if cookie.domain == '':
            if args.bind_cookies:
                cookie.domain = hostname
            else:
                cookie._rest['is_explicit_none'] = True


FIXERS_TO_VERSIONS = {
    '3.1.0': fix_cookie_layout
}


def upgrade_session(env: Environment, args: argparse.Namespace, hostname: str, session_name: str):
    session = get_httpie_session(
        env=env,
        config_dir=env.config.directory,
        session_name=session_name,
        host=hostname,
        url=hostname,
        refactor_mode=True
    )

    session_name = session.path.stem
    if session.is_new():
        env.log_error(f'{session_name!r} (for {hostname!r}) does not exist.')
        return ExitStatus.ERROR

    fixers = [
        fixer
        for version, fixer in FIXERS_TO_VERSIONS.items()
        if is_version_greater(version, session.version)
    ]

    if len(fixers) == 0:
        env.stdout.write(f'{session_name!r} (for {hostname!r}) is already up-to-date.\n')
        return ExitStatus.SUCCESS

    for fixer in fixers:
        fixer(session, hostname, args)

    session.save(bump_version=True)
    env.stdout.write(f'Refactored {session_name!r} (for {hostname!r}) to the version {session.version}.\n')
    return ExitStatus.SUCCESS


def cli_upgrade_session(env: Environment, args: argparse.Namespace) -> ExitStatus:
    return upgrade_session(
        env,
        args=args,
        hostname=args.hostname,
        session_name=args.session
    )


def cli_upgrade_all_sessions(env: Environment, args: argparse.Namespace) -> ExitStatus:
    session_dir_path = env.config_dir / SESSIONS_DIR_NAME

    status = ExitStatus.SUCCESS
    for host_path in session_dir_path.iterdir():
        hostname = host_path.name
        for session_path in host_path.glob("*.json"):
            session_name = session_path.stem
            status |= upgrade_session(
                env,
                args=args,
                hostname=hostname,
                session_name=session_name
            )
    return status
from http.cookiejar import Cookie
from typing import Any, Dict, Optional, Union
from requests.cookies import RequestsCookieJar, remove_cookie_by_name
from .context import Environment
from .utils import url_as_host
# Cookie related options
KEPT_COOKIE_OPTIONS = ['name', 'expires', 'path', 'value', 'domain', 'secure']
DEFAULT_COOKIE_PATH = '/'

INSECURE_COOKIE_JAR_WARNING = '''\
Outdated layout detected for the current session. Please consider updating it,
in order to not get affected by potential security problems.

For fixing the current session:

    With binding all cookies to the current host (secure):
        $ httpie cli sessions upgrade --bind-cookies {hostname} {session_id}

    Without binding cookies (leaving them as is) (insecure):
        $ httpie cli sessions upgrade {hostname} {session_id}
'''

INSECURE_COOKIE_JAR_WARNING_FOR_NAMED_SESSIONS = '''\

For fixing all named sessions:

    With binding all cookies to the current host (secure):
        $ httpie cli sessions upgrade-all --bind-cookies

    Without binding cookies (leaving them as is) (insecure):
        $ httpie cli sessions upgrade-all

See https://pie.co/docs/security for more information.
'''


def is_anonymous_session(session_name: str) -> bool:
    return os.path.sep in session_name


def materialize_cookie(cookie: Cookie) -> Dict[str, Any]:
    materialized_cookie = {
        option: getattr(cookie, option)
        for option in KEPT_COOKIE_OPTIONS
    }

    if (
        cookie._rest.get('is_explicit_none')
        and materialized_cookie['domain'] == ''
    ):
        materialized_cookie['domain'] = None

    return materialized_cookie

    env: Environment,
    *,
    refactor_mode: bool = False
    bound_hostname = host or url_as_host(url)
    if not bound_hostname:
        # HACK/FIXME: httpie-unixsocket's URLs have no hostname.
        bound_hostname = 'localhost'

    # host:port => host_port
    hostname = bound_hostname.replace(':', '_')
    if is_anonymous_session(session_name):
        session_id = path
        session_id = session_name

    session = Session(
        path,
        env=env,
        session_id=session_id,
        bound_host=bound_hostname.split(':')[0],
        refactor_mode=refactor_mode
    )
    def __init__(
        self,
        path: Union[str, Path],
        env: Environment,
        bound_host: str,
        session_id: str,
        refactor_mode: bool = False,
    ):
        self['cookies'] = []
        self.env = env
        self.cookie_jar = RequestsCookieJar()
        self.session_id = session_id
        self.bound_host = bound_host
        self.refactor_mode = refactor_mode

    def pre_process_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
        cookies = data.get('cookies')
        if isinstance(cookies, dict):
            normalized_cookies = [
                {
                    'name': key,
                    **value
                }
                for key, value in cookies.items()
            ]
        elif isinstance(cookies, list):
            normalized_cookies = cookies
        else:
            normalized_cookies = []

        should_issue_warning = False
        for cookie in normalized_cookies:
            domain = cookie.get('domain', '')
            if domain == '' and isinstance(cookies, dict):
                should_issue_warning = True
            elif domain is None:
                # domain = None means explicitly lack of cookie, though
                # requests requires domain to be string so we'll cast it
                # manually.
                cookie['domain'] = ''
                cookie['rest'] = {'is_explicit_none': True}

            self.cookie_jar.set(**cookie)

        if should_issue_warning and not self.refactor_mode:
            warning = INSECURE_COOKIE_JAR_WARNING.format(hostname=self.bound_host, session_id=self.session_id)
            if not is_anonymous_session(self.session_id):
                warning += INSECURE_COOKIE_JAR_WARNING_FOR_NAMED_SESSIONS

            self.env.log_error(
                warning,
                level='warning'
            )

        return data

    def post_process_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
        cookies = data.get('cookies')
        # Save in the old-style fashion

        normalized_cookies = [
            materialize_cookie(cookie)
            for cookie in self.cookie_jar
        ]
        if isinstance(cookies, dict):
            data['cookies'] = {
                cookie.pop('name'): cookie
                for cookie in normalized_cookies
            }
        else:
            data['cookies'] = normalized_cookies

        return data
            original_value = value
                    if not morsel['path']:
                        morsel['path'] = DEFAULT_COOKIE_PATH
                    self.cookie_jar.set(cookie_name, morsel)

                all_cookie_headers = request_headers.getall(name)
                if len(all_cookie_headers) > 1:
                    all_cookie_headers.remove(original_value)
                else:
                    request_headers.popall(name)
        self.cookie_jar.clear_expired_cookies()
        return self.cookie_jar
        self.cookie_jar = jar

    def remove_cookies(self, cookies: Dict[str, str]):
        for cookie in cookies:
            remove_cookie_by_name(
                self.cookie_jar,
                cookie['name'],
                domain=cookie.get('domain', None),
                path=cookie.get('path', None)
            )
from urllib.parse import urlsplit


def url_as_host(url: str) -> str:
    return urlsplit(url).netloc.split('@')[-1]
    'pytest-lazy-fixture>=0.0.6',
from .utils import ( # noqa
    HTTPBIN_WITH_CHUNKED_SUPPORT_DOMAIN,
    HTTPBIN_WITH_CHUNKED_SUPPORT,
    mock_env
)
import json
from typing import Optional, Dict, Any
import httpie

SESSION_FILES_PATH = FIXTURES_ROOT / 'session_data'
SESSION_FILES_OLD = sorted((SESSION_FILES_PATH / 'old').glob('*.json'))
SESSION_FILES_NEW = sorted((SESSION_FILES_PATH / 'new').glob('*.json'))

SESSION_VARIABLES = {
    '__version__': httpie.__version__,
    '__host__': 'null',
}



def read_session_file(session_file: Path, *, extra_variables: Optional[Dict[str, str]] = None) -> Any:
    with open(session_file) as stream:
        data = stream.read()

    session_vars = {**SESSION_VARIABLES, **(extra_variables or {})}
    for variable, value in session_vars.items():
        data = data.replace(variable, value)

    return json.loads(data)
{
    "__meta__": {
        "about": "HTTPie session file",
        "help": "https://httpie.io/docs#sessions",
        "httpie": "__version__"
    },
    "auth": {
        "password": null,
        "type": null,
        "username": null
    },
    "cookies": [
        {
            "domain": __host__,
            "expires": null,
            "name": "baz",
            "path": "/",
            "secure": false,
            "value": "quux"
        },
        {
            "domain": __host__,
            "expires": null,
            "name": "foo",
            "path": "/",
            "secure": false,
            "value": "bar"
        }
    ],
    "headers": {}
}
{
    "__meta__": {
        "about": "HTTPie session file",
        "help": "https://httpie.io/docs#sessions",
        "httpie": "__version__"
    },
    "auth": {
        "password": null,
        "type": null,
        "username": null
    },
    "cookies": [
        {
            "domain": __host__,
            "expires": null,
            "name": "baz",
            "path": "/",
            "secure": false,
            "value": "quux"
        },
        {
            "domain": __host__,
            "expires": null,
            "name": "foo",
            "path": "/",
            "secure": false,
            "value": "bar"
        }
    ],
    "headers": {}
}
{
    "__meta__": {
        "about": "HTTPie session file",
        "help": "https://httpie.io/docs#sessions",
        "httpie": "__version__"
    },
    "auth": {
        "raw_auth": "foo:bar",
        "type": "basic"
    },
    "cookies": [
        {
            "domain": __host__,
            "expires": null,
            "name": "baz",
            "path": "/",
            "secure": false,
            "value": "quux"
        },
        {
            "domain": __host__,
            "expires": null,
            "name": "foo",
            "path": "/",
            "secure": false,
            "value": "bar"
        }
    ],
    "headers": {
        "X-Data": "value",
        "X-Foo": "bar"
    }
}
{
    "__meta__": {
        "about": "HTTPie session file",
        "help": "https://httpie.io/docs#sessions",
        "httpie": "__version__"
    },
    "auth": {
        "password": null,
        "type": null,
        "username": null
    },
    "cookies": [],
    "headers": {}
}
{
    "__meta__": {
        "about": "HTTPie session file",
        "help": "https://httpie.io/docs#sessions",
        "httpie": "__version__"
    },
    "auth": {
        "password": null,
        "type": null,
        "username": null
    },
    "cookies": [],
    "headers": {}
}
{
    "__meta__": {
        "about": "HTTPie session file",
        "help": "https://httpie.io/docs#sessions",
        "httpie": "3.0.2"
    },
    "auth": {
        "password": null,
        "type": null,
        "username": null
    },
    "cookies": {
        "baz": {
            "expires": null,
            "path": "/",
            "secure": false,
            "value": "quux"
        },
        "foo": {
            "expires": null,
            "path": "/",
            "secure": false,
            "value": "bar"
        }
    },
    "headers": {}
}
{
    "__meta__": {
        "about": "HTTPie session file",
        "help": "https://httpie.io/docs#sessions",
        "httpie": "2.7.0.dev0"
    },
    "auth": {
        "password": null,
        "type": null,
        "username": null
    },
    "cookies": {
        "baz": {
            "expires": null,
            "path": "/",
            "secure": false,
            "value": "quux"
        },
        "foo": {
            "expires": null,
            "path": "/",
            "secure": false,
            "value": "bar"
        }
    },
    "headers": {}
}
{
    "__meta__": {
        "about": "HTTPie session file",
        "help": "https://httpie.io/docs#sessions",
        "httpie": "3.0.2"
    },
    "auth": {
        "raw_auth": "foo:bar",
        "type": "basic"
    },
    "cookies": {
        "baz": {
            "expires": null,
            "path": "/",
            "secure": false,
            "value": "quux"
        },
        "foo": {
            "expires": null,
            "path": "/",
            "secure": false,
            "value": "bar"
        }
    },
    "headers": {
        "X-Data": "value",
        "X-Foo": "bar"
    }
}
{
    "__meta__": {
        "about": "HTTPie session file",
        "help": "https://httpie.io/docs#sessions",
        "httpie": "3.0.2"
    },
    "auth": {
        "password": null,
        "type": null,
        "username": null
    },
    "cookies": {},
    "headers": {}
}
{
    "__meta__": {
        "about": "HTTPie session file",
        "help": "https://httpie.io/docs#sessions",
        "httpie": "3.0.2"
    },
    "auth": {
        "password": null,
        "type": null,
        "username": null
    },
    "cookies": [],
    "headers": {}
}
import pytest
from .utils import http


@pytest.fixture
def remote_httpbin(httpbin_with_chunked_support):
    return httpbin_with_chunked_support


def _stringify(fixture):
    return fixture + ''


@pytest.mark.parametrize('instance', [
    pytest.lazy_fixture('httpbin'),
    pytest.lazy_fixture('remote_httpbin'),
])
def test_explicit_user_set_cookie(httpbin, instance):
    # User set cookies ARE NOT persisted within redirects
    # when there is no session, even on the same domain.

    r = http(
        '--follow',
        httpbin + '/redirect-to',
        f'url=={_stringify(instance)}/cookies',
        'Cookie:a=b'
    )
    assert r.json == {'cookies': {}}


@pytest.mark.parametrize('instance', [
    pytest.lazy_fixture('httpbin'),
    pytest.lazy_fixture('remote_httpbin'),
])
def test_explicit_user_set_cookie_in_session(tmp_path, httpbin, instance):
    # User set cookies ARE persisted within redirects
    # when there is A session, even on the same domain.

    r = http(
        '--follow',
        '--session',
        str(tmp_path / 'session.json'),
        httpbin + '/redirect-to',
        f'url=={_stringify(instance)}/cookies',
        'Cookie:a=b'
    )
    assert r.json == {'cookies': {'a': 'b'}}


@pytest.mark.parametrize('instance', [
    pytest.lazy_fixture('httpbin'),
    pytest.lazy_fixture('remote_httpbin'),
])
def test_saved_user_set_cookie_in_session(tmp_path, httpbin, instance):
    # User set cookies ARE persisted within redirects
    # when there is A session, even on the same domain.

    http(
        '--follow',
        '--session',
        str(tmp_path / 'session.json'),
        httpbin + '/get',
        'Cookie:a=b'
    )
    r = http(
        '--follow',
        '--session',
        str(tmp_path / 'session.json'),
        httpbin + '/redirect-to',
        f'url=={_stringify(instance)}/cookies',
    )
    assert r.json == {'cookies': {'a': 'b'}}


@pytest.mark.parametrize('instance', [
    pytest.lazy_fixture('httpbin'),
    pytest.lazy_fixture('remote_httpbin'),
])
@pytest.mark.parametrize('session', [True, False])
def test_explicit_user_set_headers(httpbin, tmp_path, instance, session):
    # User set headers ARE persisted within redirects
    # even on different domains domain with or without
    # an active session.
    session_args = []
    if session:
        session_args.extend([
            '--session',
            str(tmp_path / 'session.json')
        ])

    r = http(
        '--follow',
        *session_args,
        httpbin + '/redirect-to',
        f'url=={_stringify(instance)}/get',
        'X-Custom-Header:value'
    )
    assert 'X-Custom-Header' in r.json['headers']


@pytest.mark.parametrize('session', [True, False])
def test_server_set_cookie_on_redirect_same_domain(tmp_path, httpbin, session):
    # Server set cookies ARE persisted on the same domain
    # when they are forwarded.

    session_args = []
    if session:
        session_args.extend([
            '--session',
            str(tmp_path / 'session.json')
        ])

    r = http(
        '--follow',
        *session_args,
        httpbin + '/cookies/set/a/b',
    )
    assert r.json['cookies'] == {'a': 'b'}


@pytest.mark.parametrize('session', [True, False])
def test_server_set_cookie_on_redirect_different_domain(tmp_path, http_server, httpbin, session):
    # Server set cookies ARE persisted on different domains
    # when they are forwarded.

    session_args = []
    if session:
        session_args.extend([
            '--session',
            str(tmp_path / 'session.json')
        ])

    r = http(
        '--follow',
        *session_args,
        http_server + '/cookies/set-and-redirect',
        f"X-Redirect-To:{httpbin + '/cookies'}",
        'X-Cookies:a=b'
    )
    assert r.json['cookies'] == {'a': 'b'}


def test_saved_session_cookies_on_same_domain(tmp_path, httpbin):
    # Saved session cookies ARE persisted when making a new
    # request to the same domain.
    http(
        '--session',
        str(tmp_path / 'session.json'),
        httpbin + '/cookies/set/a/b'
    )
    r = http(
        '--session',
        str(tmp_path / 'session.json'),
        httpbin + '/cookies'
    )
    assert r.json == {'cookies': {'a': 'b'}}


def test_saved_session_cookies_on_different_domain(tmp_path, httpbin, remote_httpbin):
    # Saved session cookies ARE persisted when making a new
    # request to a different domain.
    http(
        '--session',
        str(tmp_path / 'session.json'),
        httpbin + '/cookies/set/a/b'
    )
    r = http(
        '--session',
        str(tmp_path / 'session.json'),
        remote_httpbin + '/cookies'
    )
    assert r.json == {'cookies': {}}


@pytest.mark.parametrize('initial_domain, first_request_domain, second_request_domain, expect_cookies', [
    (
        # Cookies are set by    Domain A
        # Initial domain is     Domain A
        # Redirected domain is  Domain A
        pytest.lazy_fixture('httpbin'),
        pytest.lazy_fixture('httpbin'),
        pytest.lazy_fixture('httpbin'),
        True,
    ),
    (
        # Cookies are set by    Domain A
        # Initial domain is     Domain B
        # Redirected domain is  Domain B
        pytest.lazy_fixture('httpbin'),
        pytest.lazy_fixture('remote_httpbin'),
        pytest.lazy_fixture('remote_httpbin'),
        False,
    ),
    (
        # Cookies are set by    Domain A
        # Initial domain is     Domain A
        # Redirected domain is  Domain B
        pytest.lazy_fixture('httpbin'),
        pytest.lazy_fixture('httpbin'),
        pytest.lazy_fixture('remote_httpbin'),
        False,
    ),
    (
        # Cookies are set by    Domain A
        # Initial domain is     Domain B
        # Redirected domain is  Domain A
        pytest.lazy_fixture('httpbin'),
        pytest.lazy_fixture('remote_httpbin'),
        pytest.lazy_fixture('httpbin'),
        True,
    ),
])
def test_saved_session_cookies_on_redirect(tmp_path, initial_domain, first_request_domain, second_request_domain, expect_cookies):
    http(
        '--session',
        str(tmp_path / 'session.json'),
        initial_domain + '/cookies/set/a/b'
    )
    r = http(
        '--session',
        str(tmp_path / 'session.json'),
        '--follow',
        first_request_domain + '/redirect-to',
        f'url=={_stringify(second_request_domain)}/cookies'
    )
    if expect_cookies:
        expected_data = {'cookies': {'a': 'b'}}
    else:
        expected_data = {'cookies': {}}
    assert r.json == expected_data


def test_saved_session_cookie_pool(tmp_path, httpbin, remote_httpbin):
    http(
        '--session',
        str(tmp_path / 'session.json'),
        httpbin + '/cookies/set/a/b'
    )
    http(
        '--session',
        str(tmp_path / 'session.json'),
        remote_httpbin + '/cookies/set/a/c'
    )
    http(
        '--session',
        str(tmp_path / 'session.json'),
        remote_httpbin + '/cookies/set/b/d'
    )

    response = http(
        '--session',
        str(tmp_path / 'session.json'),
        httpbin + '/cookies'
    )
    assert response.json['cookies'] == {'a': 'b'}

    response = http(
        '--session',
        str(tmp_path / 'session.json'),
        remote_httpbin + '/cookies'
    )
    assert response.json['cookies'] == {'a': 'c', 'b': 'd'}
import pytest
import shutil
import json
from httpie.sessions import SESSIONS_DIR_NAME
from httpie.status import ExitStatus
from tests.utils import DUMMY_HOST, httpie
from tests.fixtures import SESSION_FILES_PATH, SESSION_FILES_NEW, SESSION_FILES_OLD, read_session_file


OLD_SESSION_FILES_PATH = SESSION_FILES_PATH / 'old'


@pytest.mark.requires_installation
def test_plugins_cli_error_message_without_args():
    # No arguments
    result = httpie(no_debug=True)
    assert result.exit_status == ExitStatus.ERROR
    assert 'usage: ' in result.stderr
    assert 'specify one of these' in result.stderr
    assert 'please use the http/https commands:' in result.stderr


@pytest.mark.parametrize(
    'example',
    [
        'pie.dev/get',
        'DELETE localhost:8000/delete',
        'POST pie.dev/post header:value a=b header_2:value x:=1',
    ],
)
@pytest.mark.requires_installation
def test_plugins_cli_error_messages_with_example(example):
    result = httpie(*example.split(), no_debug=True)
    assert result.exit_status == ExitStatus.ERROR
    assert 'usage: ' in result.stderr
    assert f'http {example}' in result.stderr
    assert f'https {example}' in result.stderr


@pytest.mark.parametrize(
    'example',
    [
        'cli',
        'plugins',
        'cli foo',
        'plugins unknown',
        'plugins unknown.com A:B c=d',
        'unknown.com UNPARSABLE????SYNTAX',
    ],
)
@pytest.mark.requires_installation
def test_plugins_cli_error_messages_invalid_example(example):
    result = httpie(*example.split(), no_debug=True)
    assert result.exit_status == ExitStatus.ERROR
    assert 'usage: ' in result.stderr
    assert f'http {example}' not in result.stderr
    assert f'https {example}' not in result.stderr


HTTPIE_CLI_SESSIONS_UPGRADE_OPTIONS = [
    (
        # Default settings
        [],
        {'__host__': json.dumps(None)},
    ),
    (
        # When --bind-cookies is applied, the __host__ becomes DUMMY_URL.
        ['--bind-cookies'],
        {'__host__': json.dumps(DUMMY_HOST)},
    ),
]


@pytest.mark.parametrize(
    'old_session_file, new_session_file', zip(SESSION_FILES_OLD, SESSION_FILES_NEW)
)
@pytest.mark.parametrize(
    'extra_args, extra_variables',
    HTTPIE_CLI_SESSIONS_UPGRADE_OPTIONS,
)
def test_httpie_sessions_upgrade(tmp_path, old_session_file, new_session_file, extra_args, extra_variables):
    session_path = tmp_path / 'session.json'
    shutil.copyfile(old_session_file, session_path)

    result = httpie(
        'cli', 'sessions', 'upgrade', *extra_args, DUMMY_HOST, str(session_path)
    )
    assert result.exit_status == ExitStatus.SUCCESS
    assert read_session_file(session_path) == read_session_file(
        new_session_file, extra_variables=extra_variables
    )


def test_httpie_sessions_upgrade_on_non_existent_file(tmp_path):
    session_path = tmp_path / 'session.json'
    result = httpie('cli', 'sessions', 'upgrade', DUMMY_HOST, str(session_path))
    assert result.exit_status == ExitStatus.ERROR
    assert 'does not exist' in result.stderr


@pytest.mark.parametrize(
    'extra_args, extra_variables',
    HTTPIE_CLI_SESSIONS_UPGRADE_OPTIONS,
)
def test_httpie_sessions_upgrade_all(tmp_path, mock_env, extra_args, extra_variables):
    mock_env._create_temp_config_dir = False
    mock_env.config_dir = tmp_path / "config"

    session_dir = mock_env.config_dir / SESSIONS_DIR_NAME / DUMMY_HOST
    session_dir.mkdir(parents=True)
    for original_session_file in SESSION_FILES_OLD:
        shutil.copy(original_session_file, session_dir)

    result = httpie(
        'cli', 'sessions', 'upgrade-all', *extra_args, env=mock_env
    )
    assert result.exit_status == ExitStatus.SUCCESS

    for refactored_session_file, expected_session_file in zip(
        sorted(session_dir.glob("*.json")),
        SESSION_FILES_NEW
    ):
        assert read_session_file(refactored_session_file) == read_session_file(
            expected_session_file, extra_variables=extra_variables
        )
from contextlib import contextmanager
from pathlib import Path
from typing import Iterator
from httpie.context import Environment
from .utils import DUMMY_HOST, HTTP_OK, MockEnvironment, http, mk_config_dir
            'headers': {
                'cookie': '...',
                'zzz': '...'
                basic_auth_header = 'Basic ' + b64encode(self.raw_auth.encode()).strip().decode('latin1')
        assert updated_session['auth']['raw_auth'] == 'user:password'
            ({'id': {'value': 123}}, {'name': 'id'}),
            ({'id': {'value': 123}}, {'name': 'token'})
    def test_removes_expired_cookies_from_session_obj(self, initial_cookie, expired_cookie, httpbin, mock_env):
        session = Session(self.config_dir, env=mock_env, session_id=None, bound_host=None)


@pytest.fixture
def basic_session(httpbin, tmp_path):
    session_path = tmp_path / 'session.json'
    http(
        '--session', str(session_path),
        httpbin + '/get'
    )
    return session_path


@contextmanager
def open_session(path: Path, env: Environment, read_only: bool = False) -> Iterator[Session]:
    session = Session(path, env, session_id='test', bound_host=DUMMY_HOST)
    session.load()
    yield session
    if not read_only:
        session.save()


@contextmanager
def open_raw_session(path: Path, read_only: bool = False) -> None:
    with open(path) as stream:
        raw_session = json.load(stream)

    yield raw_session

    if not read_only:
        with open(path, 'w') as stream:
            json.dump(raw_session, stream)


def read_stderr(env: Environment) -> bytes:
    env.stderr.seek(0)
    stderr_data = env.stderr.read()
    if isinstance(stderr_data, str):
        return stderr_data.encode()
    else:
        return stderr_data


def test_old_session_version_saved_as_is(basic_session, mock_env):
    with open_session(basic_session, mock_env) as session:
        session['__meta__'] = {'httpie': '0.0.1'}

    with open_session(basic_session, mock_env, read_only=True) as session:
        assert session['__meta__']['httpie'] == '0.0.1'


def test_old_session_cookie_layout_warning(basic_session, mock_env):
    with open_session(basic_session, mock_env) as session:
        # Use the old layout & set a cookie
        session['cookies'] = {}
        session.cookies.set('foo', 'bar')

    assert read_stderr(mock_env) == b''

    with open_session(basic_session, mock_env, read_only=True) as session:
        assert b'Outdated layout detected' in read_stderr(mock_env)


@pytest.mark.parametrize('cookies, expect_warning', [
    # Old-style cookie format
    (
        # Without 'domain' set
        {'foo': {'value': 'bar'}},
        True
    ),
    (
        # With 'domain' set to empty string
        {'foo': {'value': 'bar', 'domain': ''}},
        True
    ),
    (
        # With 'domain' set to null
        {'foo': {'value': 'bar', 'domain': None}},
        False,
    ),
    (
        # With 'domain' set to a URL
        {'foo': {'value': 'bar', 'domain': DUMMY_HOST}},
        False,
    ),
    # New style cookie format
    (
        # Without 'domain' set
        [{'name': 'foo', 'value': 'bar'}],
        False
    ),
    (
        # With 'domain' set to empty string
        [{'name': 'foo', 'value': 'bar', 'domain': ''}],
        False
    ),
    (
        # With 'domain' set to null
        [{'name': 'foo', 'value': 'bar', 'domain': None}],
        False,
    ),
    (
        # With 'domain' set to a URL
        [{'name': 'foo', 'value': 'bar', 'domain': DUMMY_HOST}],
        False,
    ),
])
def test_cookie_security_warnings_on_raw_cookies(basic_session, mock_env, cookies, expect_warning):
    with open_raw_session(basic_session) as raw_session:
        raw_session['cookies'] = cookies

    with open_session(basic_session, mock_env, read_only=True):
        warning = b'Outdated layout detected'
        stderr = read_stderr(mock_env)

        if expect_warning:
            assert warning in stderr
        else:
            assert warning not in stderr


def test_old_session_cookie_layout_loading(basic_session, httpbin, mock_env):
    with open_session(basic_session, mock_env) as session:
        # Use the old layout & set a cookie
        session['cookies'] = {}
        session.cookies.set('foo', 'bar')

    response = http(
        '--session', str(basic_session),
        httpbin + '/cookies'
    )
    assert response.json['cookies'] == {'foo': 'bar'}


@pytest.mark.parametrize('layout_type', [
    dict, list
])
def test_session_cookie_layout_preservance(basic_session, mock_env, layout_type):
    with open_session(basic_session, mock_env) as session:
        session['cookies'] = layout_type()
        session.cookies.set('foo', 'bar')
        session.save()

    with open_session(basic_session, mock_env, read_only=True) as session:
        assert isinstance(session['cookies'], layout_type)


@pytest.mark.parametrize('layout_type', [
    dict, list
])
def test_session_cookie_layout_preservance_on_new_cookies(basic_session, httpbin, mock_env, layout_type):
    with open_session(basic_session, mock_env) as session:
        session['cookies'] = layout_type()
        session.cookies.set('foo', 'bar')
        session.save()

    http(
        '--session', str(basic_session),
        httpbin + '/cookies/set/baz/quux'
    )

    with open_session(basic_session, mock_env, read_only=True) as session:
        assert isinstance(session['cookies'], layout_type)
import pytest
from contextlib import suppress
from httpie.utils import url_as_host
DUMMY_HOST = url_as_host(DUMMY_URL)
@pytest.fixture
def mock_env() -> MockEnvironment:
    env = MockEnvironment(stdout_mode='')
    yield env
    env.cleanup()


    env = kwargs.setdefault('env', MockEnvironment(stdout_mode=''))
        output = env.stdout.read()
        if isinstance(output, bytes):
            with suppress(UnicodeDecodeError):
                output = output.decode()

        if isinstance(output, bytes):
            response = BytesCLIResponse(output)
        else:
            response = StrCLIResponse(output)

@TestHandler.handler('GET', '/cookies/set-and-redirect')
def set_cookie_and_redirect(handler):
    handler.send_response(302)

    redirect_to = handler.headers.get('X-Redirect-To', '/headers')
    handler.send_header('Location', redirect_to)

    raw_cookies = handler.headers.get('X-Cookies', 'a=b')
    for cookie in raw_cookies.split(', '):
        handler.send_header('Set-Cookie', cookie)
    handler.end_headers()