extern/httpie-cli
1
0
Fork 1
mirror of https://github.com/httpie/cli.git synced 2025-03-13 06:18:33 +01:00
Code Issues Packages Projects Releases Wiki Activity

Move port and interface validation to the CLI layer

Browse Source
This commit is contained in:
Jakub Roztocil 2024-10-26 18:55:26 +02:00
parent 0eab08a655
commit 4cea2e80af
6 changed files with 201 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
httpie/cli/argtypes.py
View File

@ -259,7 +259,7 @@ PARSED_DEFAULT_FORMAT_OPTIONS = parse_format_options(
)
def response_charset_type(encoding: str) -> str:
def response_charset_arg_type(encoding: str) -> str:
try:
''.encode(encoding)
except LookupError:
@ -268,8 +268,17 @@ def response_charset_type(encoding: str) -> str:
return encoding
def response_mime_type(mime_type: str) -> str:
def response_mime_arg_type(mime_type: str) -> str:
if mime_type.count('/') != 1:
raise argparse.ArgumentTypeError(
f'{mime_type!r} doesn’t look like a mime type; use type/subtype')
return mime_type
def interface_arg_type(interface: str) -> str:
import ipaddress
try:
ipaddress.ip_interface(interface)
except ValueError as e:
raise argparse.ArgumentTypeError(str(e))
return interface

60
httpie/cli/definition.py
View File

@ -5,30 +5,50 @@ import textwrap
from argparse import FileType
from httpie import __doc__, __version__
from httpie.cli.argtypes import (KeyValueArgType, SessionNameValidator,
SSLCredentials, readable_file_arg,
response_charset_type, response_mime_type)
from httpie.cli.constants import (BASE_OUTPUT_OPTIONS, DEFAULT_FORMAT_OPTIONS,
OUT_REQ_BODY, OUT_REQ_HEAD, OUT_RESP_BODY,
OUT_RESP_HEAD, OUT_RESP_META, OUTPUT_OPTIONS,
OUTPUT_OPTIONS_DEFAULT, PRETTY_MAP,
PRETTY_STDOUT_TTY_ONLY,
SEPARATOR_GROUP_ALL_ITEMS, SEPARATOR_PROXY,
SORTED_FORMAT_OPTIONS_STRING,
UNSORTED_FORMAT_OPTIONS_STRING, RequestType)
from httpie.cli.options import ParserSpec, Qualifiers, to_argparse
from httpie.output.formatters.colors import (AUTO_STYLE, DEFAULT_STYLE, BUNDLED_STYLES,
get_available_styles)
from httpie.output.formatters.colors import (
AUTO_STYLE,
BUNDLED_STYLES,
DEFAULT_STYLE,
get_available_styles,
)
from httpie.plugins.builtin import BuiltinAuthPlugin
from httpie.plugins.registry import plugin_manager
from httpie.ssl_ import AVAILABLE_SSL_VERSION_ARG_MAPPING, DEFAULT_SSL_CIPHERS_STRING
from .argtypes import (
KeyValueArgType,
SSLCredentials,
SessionNameValidator,
interface_arg_type,
readable_file_arg,
response_charset_arg_type,
response_mime_arg_type,
)
from .constants import (
BASE_OUTPUT_OPTIONS,
DEFAULT_FORMAT_OPTIONS,
OUTPUT_OPTIONS,
OUTPUT_OPTIONS_DEFAULT,
OUT_REQ_BODY,
OUT_REQ_HEAD,
OUT_RESP_BODY,
OUT_RESP_HEAD,
OUT_RESP_META,
PRETTY_MAP,
PRETTY_STDOUT_TTY_ONLY,
RequestType,
SEPARATOR_GROUP_ALL_ITEMS,
SEPARATOR_PROXY,
SORTED_FORMAT_OPTIONS_STRING,
UNSORTED_FORMAT_OPTIONS_STRING,
)
from .options import ParserSpec, Qualifiers, to_argparse
from .ports import local_port_arg_type
# Man pages are static (built when making a release).
# We use this check to not include generated, system-specific information there (e.g., default --ciphers).
IS_MAN_PAGE = bool(os.environ.get('HTTPIE_BUILDING_MAN_PAGES'))
options = ParserSpec(
'http',
description=f'{__doc__.strip()} <https://httpie.io>',
@ -349,7 +369,7 @@ output_processing.add_argument(
output_processing.add_argument(
'--response-charset',
metavar='ENCODING',
type=response_charset_type,
type=response_charset_arg_type,
short_help='Override the response encoding for terminal display purposes.',
help="""
Override the response encoding for terminal display purposes, e.g.:
@ -362,7 +382,7 @@ output_processing.add_argument(
output_processing.add_argument(
'--response-mime',
metavar='MIME_TYPE',
type=response_mime_type,
type=response_mime_arg_type,
short_help='Override the response mime type for coloring and formatting for the terminal.',
help="""
Override the response mime type for coloring and formatting for the terminal, e.g.:
@ -894,12 +914,14 @@ network.add_argument(
)
network.add_argument(
"--interface",
default=None,
type=interface_arg_type,
default='0.0.0.0',
short_help="Bind to a specific network interface.",
)
network.add_argument(
"--local-port",
default=None,
type=local_port_arg_type,
default=0,
short_help="Set the local port to be used for the outgoing request.",
help="""
It can be either a port range (e.g. "11221-14555") or a single port.

50
httpie/cli/ports.py Normal file
View File

@ -0,0 +1,50 @@
import argparse
from random import randint
from typing import Tuple
MIN_PORT = 0
MAX_PORT = 65535
OUTSIDE_VALID_PORT_RANGE_ERROR = f'outside valid port range {MIN_PORT}-{MAX_PORT}'
def local_port_arg_type(port: str) -> int:
port = parse_local_port_arg(port)
if isinstance(port, tuple):
port = randint(*port)
return port
def parse_local_port_arg(port: str) -> int | Tuple[int, int]:
if '-' in port[1:]: # Don’t treat negative port as range.
return _clean_port_range(port)
return _clean_port(port)
def _clean_port_range(port_range: str) -> Tuple[int, int]:
"""
We allow two digits separated by a hyphen to represent a port range.
The parsing is done so that even negative numbers get parsed correctly, allowing us to
give a more specific outside-range error message.
"""
sep_pos = port_range.find('-', 1)
start, end = port_range[:sep_pos], port_range[sep_pos + 1:]
start = _clean_port(start)
end = _clean_port(end)
if start > end:
raise argparse.ArgumentTypeError(f'{port_range!r} is not a valid port range')
return start, end
def _clean_port(port: str) -> int:
try:
port = int(port)
except ValueError:
raise argparse.ArgumentTypeError(f'{port!r} is not a number')
if not (MIN_PORT <= port <= MAX_PORT):
raise argparse.ArgumentTypeError(
f'{port!r} is {OUTSIDE_VALID_PORT_RANGE_ERROR}'
)
return port

43
httpie/client.py
View File

@ -5,7 +5,6 @@ import json
import sys
import typing
from pathlib import Path
from random import randint
from time import monotonic
from typing import Any, Dict, Callable, Iterable
from urllib.parse import urlparse, urlunparse
@ -68,48 +67,10 @@ def collect_messages(
)
send_kwargs = make_send_kwargs(args)
send_kwargs_mergeable_from_env = make_send_kwargs_mergeable_from_env(args)
source_address = None
if args.interface:
# automatically raises ValueError upon invalid IP
ipaddress.ip_address(args.interface)
source_address = (args.interface, 0)
if args.local_port:
if '-' not in args.local_port:
try:
parsed_port = int(args.local_port)
except ValueError:
raise ValueError(f'"{args.local_port}" is not a valid port number.')
source_address = (args.interface or "0.0.0.0", parsed_port)
else:
if args.local_port.count('-') != 1:
raise ValueError(f'"{args.local_port}" is not a valid port range. i.e. we accept value like "25441-65540".')
try:
min_port, max_port = args.local_port.split('-', 1)
except ValueError:
raise ValueError(f'The port range you gave in input "{args.local_port}" is not a valid range.')
if min_port == "":
raise ValueError("Negative port number are all invalid values.")
if max_port == "":
raise ValueError('Port range requires both start and end ports to be specified. e.g. "25441-65540".')
try:
min_port, max_port = int(min_port), int(max_port)
except ValueError:
raise ValueError(f'Either "{min_port}" or/and "{max_port}" is an invalid port number.')
source_address = (args.interface or "0.0.0.0", randint(int(min_port), int(max_port)))
parsed_url = parse_url(args.url)
resolver = args.resolver or None
# we want to make sure every ".localhost" host resolve to loopback
# We want to make sure every ".localhost" host resolve to loopback
if parsed_url.host and parsed_url.host.endswith(".localhost"):
ensure_resolver = f"in-memory://default/?hosts={parsed_url.host}:127.0.0.1&hosts={parsed_url.host}:[::1]"
@ -157,7 +118,7 @@ def collect_messages(
resolver=resolver,
disable_ipv6=args.ipv4,
disable_ipv4=args.ipv6,
source_address=source_address,
source_address=(args.interface, args.local_port),
quic_cache=env.config.quic_file,
happy_eyeballs=args.happy_eyeballs,
)

6
tests/test_binary.py
View File

@ -32,9 +32,9 @@ class TestBinaryRequestData:
class TestBinaryResponseData:
"""local httpbin crash due to an unfixed bug.
See https://github.com/psf/httpbin/pull/41
It is merged but not yet released."""
# Local httpbin crashes due to an unfixed bug — it is merged but not yet released.
# <https://github.com/psf/httpbin/pull/41>
# TODO: switch to the local `httpbin` fixture when the fix is released.
def test_binary_suppresses_when_terminal(self, remote_httpbin):
r = http('GET', remote_httpbin + '/bytes/1024?seed=1')

187
tests/test_network.py
View File

@ -1,121 +1,122 @@
import argparse
import pytest
from httpie.cli.ports import (
MAX_PORT,
MIN_PORT,
OUTSIDE_VALID_PORT_RANGE_ERROR,
local_port_arg_type,
parse_local_port_arg,
)
from .utils import HTTP_OK, http
def test_ensure_interface_parameter(httpbin):
"""We ensure that HTTPie properly wire interface by passing an interface that
does not exist. thus, we expect an error."""
def test_non_existent_interface_arg(httpbin):
"""We ensure that HTTPie properly wire interface by passing an interface that does not exist. thus, we expect an error."""
r = http(
"--interface=1.1.1.1",
httpbin + "/get",
'--interface=1.1.1.1',
httpbin + '/get',
tolerate_error_exit_status=True
)
assert r.exit_status != 0
assert "assign requested address" in r.stderr or "The requested address is not valid in its context" in r.stderr
assert (
'assign requested address' in r.stderr
or 'The requested address is not valid in its context' in r.stderr
)
def test_ensure_local_port_parameter(httpbin):
"""We ensure that HTTPie properly wire local-port by passing a port that
does not exist. thus, we expect an error."""
@pytest.mark.parametrize(['local_port_arg', 'expected_output'], [
# Single ports — valid
('0', 0),
('-0', 0),
(str(MAX_PORT), MAX_PORT),
('8000', 8000),
# Single ports — invalid
(f'{MIN_PORT - 1}', OUTSIDE_VALID_PORT_RANGE_ERROR),
(f'{MAX_PORT + 1}', OUTSIDE_VALID_PORT_RANGE_ERROR),
('-', 'not a number'),
('AAA', 'not a number'),
(' ', 'not a number'),
# Port ranges — valid
(f'{MIN_PORT}-{MAX_PORT}', (MIN_PORT, MAX_PORT)),
('3000-8000', (3000, 8000)),
('-0-8000', (0, 8000)),
('0-0', (0, 0)),
# Port ranges — invalid
(f'2-1', 'not a valid port range'),
(f'2-', 'not a number'),
(f'2-A', 'not a number'),
(f'A-A', 'not a number'),
(f'A-2', 'not a number'),
(f'-10-1', OUTSIDE_VALID_PORT_RANGE_ERROR),
(f'1--1', OUTSIDE_VALID_PORT_RANGE_ERROR),
(f'-10--1', OUTSIDE_VALID_PORT_RANGE_ERROR),
(f'1-{MAX_PORT + 1}', OUTSIDE_VALID_PORT_RANGE_ERROR),
])
def test_parse_local_port_arg(local_port_arg, expected_output):
expected_error = expected_output if isinstance(expected_output, str) else None
if not expected_error:
assert parse_local_port_arg(local_port_arg) == expected_output
else:
with pytest.raises(argparse.ArgumentTypeError, match=expected_error):
parse_local_port_arg(local_port_arg)
def test_local_port_arg_type():
assert local_port_arg_type('1') == 1
assert local_port_arg_type('1-1') == 1
assert local_port_arg_type('1-3') in {1, 2, 3}
def test_invoke_with_out_of_range_local_port_arg(httpbin):
# An addition to the unittest tests
r = http(
"--local-port=70000",
httpbin + "/get",
'--local-port=70000',
httpbin + '/get',
tolerate_error_exit_status=True
)
assert r.exit_status != 0
assert "port must be 0-65535" in r.stderr
assert OUTSIDE_VALID_PORT_RANGE_ERROR in r.stderr
def test_ensure_interface_and_port_parameters(httpbin):
@pytest.mark.parametrize('interface_arg', [
'',
'-',
'10.25.a.u',
'abc',
'localhost',
])
def test_invalid_interface_arg(httpbin, interface_arg):
r = http(
"--interface=0.0.0.0", # it's valid, setting 0.0.0.0 means "take the default" here.
"--local-port=0", # this will automatically pick a free port in range 1024-65535
httpbin + "/get",
)
assert r.exit_status == 0
assert HTTP_OK in r
def test_invalid_interface_given(httpbin):
r = http(
"--interface=10.25.a.u", # invalid IP
httpbin + "/get",
'--interface',
interface_arg,
httpbin + '/get',
tolerate_error_exit_status=True,
)
assert "'10.25.a.u' does not appear to be an IPv4 or IPv6 address" in r.stderr
r = http(
"--interface=abc", # invalid IP
httpbin + "/get",
tolerate_error_exit_status=True,
)
assert "'abc' does not appear to be an IPv4 or IPv6 address" in r.stderr
def test_invalid_local_port_given(httpbin):
r = http(
"--local-port=127.0.0.1", # invalid port
httpbin + "/get",
tolerate_error_exit_status=True,
)
assert '"127.0.0.1" is not a valid port number.' in r.stderr
r = http(
"--local-port=a8", # invalid port
httpbin + "/get",
tolerate_error_exit_status=True,
)
assert '"a8" is not a valid port number.' in r.stderr
r = http(
"--local-port=-8", # invalid port
httpbin + "/get",
tolerate_error_exit_status=True,
)
assert 'Negative port number are all invalid values.' in r.stderr
r = http(
"--local-port=a-8", # invalid port range
httpbin + "/get",
tolerate_error_exit_status=True,
)
assert 'Either "a" or/and "8" is an invalid port number.' in r.stderr
r = http(
"--local-port=5555-", # invalid port range
httpbin + "/get",
tolerate_error_exit_status=True,
)
assert 'Port range requires both start and end ports to be specified.' in r.stderr
assert f"'{interface_arg}' does not appear to be an IPv4 or IPv6" in r.stderr
def test_force_ipv6_on_unsupported_system(remote_httpbin):
from httpie.compat import urllib3
orig_has_ipv6 = urllib3.util.connection.HAS_IPV6
urllib3.util.connection.HAS_IPV6 = False
r = http(
"-6", # invalid port
remote_httpbin + "/get",
tolerate_error_exit_status=True,
)
urllib3.util.connection.HAS_IPV6 = True
try:
r = http(
"-6", # invalid port
remote_httpbin + "/get",
tolerate_error_exit_status=True,
)
finally:
urllib3.util.connection.HAS_IPV6 = orig_has_ipv6
assert 'Unable to force IPv6 because your system lack IPv6 support.' in r.stderr
def test_force_both_ipv6_and_ipv4(remote_httpbin):
r = http(
"-6", # force IPv6
"-4", # force IPv4
remote_httpbin + "/get",
'-6', # force IPv6
'-4', # force IPv4
remote_httpbin + '/get',
tolerate_error_exit_status=True,
)
@ -124,9 +125,9 @@ def test_force_both_ipv6_and_ipv4(remote_httpbin):
def test_happy_eyeballs(remote_httpbin_secure):
r = http(
"--heb", # this will automatically and concurrently try IPv6 and IPv4 endpoints
"--verify=no",
remote_httpbin_secure + "/get",
'--heb', # this will automatically and concurrently try IPv6 and IPv4 endpoints
'--verify=no',
remote_httpbin_secure + '/get',
)
assert r.exit_status == 0