tests/test_request_headers.py
Rama master · réplica de lectura del repositorio original.
#!/usr/bin/env python
# encoding: UTF-8
"""
This file is part of Commix Project (https://commixproject.com).
Copyright (c) 2014-2026 Anastasios Stasinopoulos (@ancst).
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
For more see the file 'readme/COPYING' for copying permission.
"""
import os
import sys
import socket
import shutil
import tempfile
import unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# The modules below read the command line as they are imported, so keep pytest's own
# arguments away from them and hand the real ones back even if an import fails.
_COMMAND_LINE = list(sys.argv)
try:
sys.argv = ["commix.py"]
from src.utils import settings
from src.core.parse import cmdline as menu
from src.core.parse import request
finally:
sys.argv = _COMMAND_LINE
TARGET = "/review?x=1"
SETTINGS_STATE = ("RAW_HTTP_HEADERS", "HTTP_METHOD", "USER_DEFINED_POST_DATA",
"IGNORE_USER_DEFINED_POST_DATA", "MULTI_REQUEST_TARGETS", "SKIPPED_OUT_OF_SCOPE")
SOCKET_ATTEMPTS = []
"""
Fail closed: a parser test that reaches for the network is a broken test, not a slow one.
"""
def _no_transport(*args, **kwargs):
SOCKET_ATTEMPTS.append(args[:1])
raise RuntimeError("Request parsing is exercised offline; no transport may be opened.")
def _raw(*headers, **kwargs):
target = kwargs.get("target", TARGET)
return kwargs.get("method", "GET") + " " + target + " HTTP/1.1\n" + "".join(line + "\n" for line in headers) + "\n"
class _OfflineRequests(unittest.TestCase):
"""
A captured request file, parsed through the public entry point, with the network and the
command line state it touches put back the way they were found.
"""
def setUp(self):
self.saved_options = dict(vars(menu.options))
self.saved_command_line = dict(request._command_line_options)
self.saved_settings = {name: getattr(settings, name) for name in SETTINGS_STATE}
self.saved_sockets = (socket.socket, socket.create_connection, socket.getaddrinfo)
socket.socket, socket.create_connection, socket.getaddrinfo = _no_transport, _no_transport, _no_transport
# Nothing another test, an earlier suite or a real run left behind may decide what is expected here.
menu.options.scope = None
settings.SKIPPED_OUT_OF_SCOPE = []
request._command_line_options.clear()
self.directory = tempfile.mkdtemp(prefix="commix-request-")
self.path = os.path.join(self.directory, "captured.invalid.txt")
def tearDown(self):
socket.socket, socket.create_connection, socket.getaddrinfo = self.saved_sockets
for name in SETTINGS_STATE:
setattr(settings, name, self.saved_settings[name])
vars(menu.options).clear()
vars(menu.options).update(self.saved_options)
request._command_line_options.clear()
request._command_line_options.update(self.saved_command_line)
shutil.rmtree(self.directory, ignore_errors=True)
self.assertEqual([], SOCKET_ATTEMPTS, "a request parser test opened a transport")
def write(self, text):
with open(self.path, "w", encoding=settings.DEFAULT_CODEC) as handle:
handle.write(text)
return self.path
def targets(self, text):
try:
return request.parse_requests(self.write(text))
except SystemExit:
self.fail("a valid captured request was rejected: " + repr(text))
def rejected(self, text):
with self.assertRaises(SystemExit):
request.parse_requests(self.write(text))
class HeaderValue(_OfflineRequests):
"""
The optional whitespace a field line may carry, read off the value of one known header.
Guarded like the rest: a helper test that opens a transport is just as wrong as a parse one.
"""
def test_a_plain_space_is_not_part_of_the_value(self):
for line in ("Host: asset.invalid", "host: asset.invalid", "HOST: asset.invalid"):
self.assertEqual("asset.invalid", request._extract_header_value(settings.HOST, line))
def test_no_whitespace_after_the_colon_is_allowed(self):
for line in ("Host:asset.invalid", "Host:\tasset.invalid", "User-Agent:probe-a/1.0"):
expected = "asset.invalid" if "Host" in line else "probe-a/1.0"
name = settings.HOST if "Host" in line else settings.USER_AGENT
self.assertEqual(expected, request._extract_header_value(name, line))
def test_the_boundary_whitespace_of_the_value_is_dropped(self):
for line, expected in (("Host: asset.invalid", "asset.invalid"),
("Host: \t asset.invalid", "asset.invalid"),
("Host:\tasset.invalid \t", "asset.invalid"),
("Host: asset.invalid ", "asset.invalid"),
("Cookie: \ta=1; b=2\t", "a=1; b=2")):
self.assertEqual(expected, request._extract_header_value(line.split(":")[0], line))
def test_the_inside_of_the_value_is_left_alone(self):
for line, expected in (("Cookie: a=1; b=2", "a=1; b=2"),
("Referer: https://ref.invalid/p?q=a:b", "https://ref.invalid/p?q=a:b"),
("User-Agent: probe agent/1.0", "probe agent/1.0")):
self.assertEqual(expected, request._extract_header_value(line.split(":")[0], line))
def test_a_different_header_is_not_the_one_asked_for(self):
for name, line in ((settings.HOST, "X-Host: other.invalid"),
(settings.HOST, "Hosted-By: other.invalid"),
(settings.HOST, "User-Agent: probe-a/1.0")):
self.assertIsNone(request._extract_header_value(name, line))
class ParsedRequest(_OfflineRequests):
"""
The same field line, read through the public parse of a captured request file.
"""
def test_a_standard_request_is_parsed_as_before(self):
targets = self.targets(_raw("Host: asset.invalid", "User-Agent: probe-a/1.0",
"Cookie: a=1; b=2", "Referer: http://ref.invalid/p?q=a:b"))
self.assertEqual(1, len(targets))
self.assertEqual("asset.invalid", targets[0]["host"])
self.assertEqual("probe-a/1.0", targets[0]["agent"])
self.assertEqual("a=1; b=2", targets[0]["cookie"])
self.assertEqual("http://ref.invalid/p?q=a:b", targets[0]["referer"])
self.assertEqual("http://asset.invalid" + TARGET, targets[0]["url"])
def test_a_host_field_parses_whatever_whitespace_follows_its_colon(self):
for line in ("Host: asset.invalid",
"Host:asset.invalid",
"Host:\tasset.invalid",
"Host: asset.invalid",
"Host: \t asset.invalid",
"Host:\tasset.invalid \t",
"hOsT: asset.invalid"):
with self.subTest(line=line):
targets = self.targets(_raw(line))
self.assertEqual(1, len(targets))
self.assertEqual("asset.invalid", targets[0]["host"])
self.assertEqual("http://asset.invalid" + TARGET, targets[0]["url"])
def test_the_boundary_whitespace_of_a_known_field_does_not_survive(self):
targets = self.targets(_raw("Host: asset.invalid", "User-Agent:\tprobe agent/1.0 \t",
"Cookie:\ta=1; b=2 ", "Referer: https://ref.invalid/p?q=a:b "))
self.assertEqual("probe agent/1.0", targets[0]["agent"])
self.assertEqual("a=1; b=2", targets[0]["cookie"])
self.assertEqual("https://ref.invalid/p?q=a:b", targets[0]["referer"])
def test_a_host_field_without_a_value_is_rejected(self):
for line in ("Host:", "Host: ", "Host:\t", "Host: \t "):
with self.subTest(line=line):
self.rejected(_raw(line, "User-Agent: probe-a/1.0"))
def test_a_request_without_a_host_field_is_rejected(self):
self.rejected(_raw("User-Agent: probe-a/1.0", "Cookie: a=1"))
def test_a_field_that_only_ends_in_host_is_not_the_host(self):
for line in ("X-Host: other.invalid", "X-Forwarded-Host: other.invalid", "Hosted-By: other.invalid"):
with self.subTest(line=line):
self.rejected(_raw(line, "User-Agent: probe-a/1.0"))
def test_a_host_named_field_does_not_overrule_the_real_one(self):
targets = self.targets(_raw("Host: asset.invalid", "X-Forwarded-Host: other.invalid",
"X-Host: other.invalid"))
self.assertEqual("asset.invalid", targets[0]["host"])
self.assertEqual("http://asset.invalid" + TARGET, targets[0]["url"])
class LogfileTarget(_OfflineRequests):
"""
The same file, applied to the run's options the way '--requestfile' applies it.
"""
def test_a_captured_request_becomes_the_run_target(self):
self.write(_raw("Host:asset.invalid", "User-Agent:\tprobe-a/1.0", "Referer: https://ref.invalid/"))
menu.options.requestfile = self.path
menu.options.logfile = None
targets = request.logfile_parser()
self.assertEqual(1, len(targets))
self.assertEqual("asset.invalid", menu.options.host)
self.assertEqual("probe-a/1.0", menu.options.agent)
self.assertEqual("https://ref.invalid/", menu.options.referer)
self.assertEqual("https://asset.invalid" + TARGET, menu.options.url)
self.assertTrue(settings.RAW_HTTP_HEADERS.startswith("Host:"))
self.assertEqual(targets, settings.MULTI_REQUEST_TARGETS)
if __name__ == "__main__":
unittest.main()